Add durable workflow orchestration foundation#78
Open
Waishnav wants to merge 3 commits into
Open
Conversation
Co-Authored-By: Claude <noreply@anthropic.com>
Co-Authored-By: Claude <noreply@anthropic.com>
Co-Authored-By: Claude <noreply@anthropic.com>
📝 WalkthroughWalkthroughChangesDurable workflow execution
Estimated code review effort: 5 (Critical) | ~120 minutes Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/db/migrations.ts`:
- Around line 184-264: Make migration 4 fail rather than silently accepting
incompatible existing workflow objects: in the workflow schema creation block of
src/db/migrations.ts (lines 184-264), replace the IF NOT EXISTS table/index
creation or validate every existing object’s columns, keys, and foreign keys
before recording the migration. In src/workflows/migrations.test.ts (lines
172-208), add coverage for a legacy workflow_events table containing
workflow_run_id and sequence but missing required event/payload columns, and
assert the migration fails and rolls back.
In `@src/workflows/store.ts`:
- Around line 264-273: The claim lease checks in src/workflows/store.ts lines
264-273 and 339-344 must revoke expired-token authority. In the renewal logic
around getNodeRow, require claim_expires_at > now alongside the running status
and token match; in the running-node transition logic, require the same
unexpired-lease condition in both validation and the update predicate, so
expired claims are rejected and reclamation requires a fresh token.
In `@src/workflows/workflows.test.ts`:
- Around line 538-567: Update the worker cleanup in the workflow test’s
running/try/finally flow to terminate every spawned child process before
removing coordinationDir. Retain references to the spawned processes alongside
readyPath and result, and in finally request termination and await worker
completion as needed before rmSync deletes the barrier directory, including
timeout and setup-failure paths.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 393765c3-8b13-4bb6-9a8c-3ec409627766
📒 Files selected for processing (9)
package.jsonsrc/db/migrations.tssrc/db/schema.tssrc/oauth-store.test.tssrc/workflows/migrations.test.tssrc/workflows/orchestrator.tssrc/workflows/store.tssrc/workflows/types.tssrc/workflows/workflows.test.ts
Comment on lines
+184
to
+264
| create table if not exists workflow_runs ( | ||
| id text primary key, | ||
| definition_version integer not null, | ||
| status text not null, | ||
| definition_json text not null, | ||
| input_json text not null, | ||
| policy_json text not null, | ||
| idempotency_key text unique, | ||
| request_hash text not null, | ||
| result_json text, | ||
| error_json text, | ||
| cancellation_requested_at text, | ||
| event_sequence integer not null default 0, | ||
| created_at text not null, | ||
| updated_at text not null, | ||
| started_at text, | ||
| completed_at text | ||
| ); | ||
|
|
||
| create index if not exists workflow_runs_status_idx | ||
| on workflow_runs(status, created_at); | ||
|
|
||
| create table if not exists workflow_nodes ( | ||
| id text primary key, | ||
| workflow_run_id text not null, | ||
| node_key text not null, | ||
| node_type text not null, | ||
| status text not null, | ||
| definition_json text not null, | ||
| attempt integer not null default 0, | ||
| claim_token text, | ||
| claimed_at text, | ||
| claim_expires_at text, | ||
| result_json text, | ||
| error_json text, | ||
| created_at text not null, | ||
| updated_at text not null, | ||
| completed_at text, | ||
| foreign key (workflow_run_id) references workflow_runs(id) on delete cascade | ||
| ); | ||
|
|
||
| create unique index if not exists workflow_nodes_run_key_idx | ||
| on workflow_nodes(workflow_run_id, node_key); | ||
|
|
||
| create unique index if not exists workflow_nodes_run_id_idx | ||
| on workflow_nodes(workflow_run_id, id); | ||
|
|
||
| create index if not exists workflow_nodes_status_idx | ||
| on workflow_nodes(workflow_run_id, status, created_at); | ||
|
|
||
| create table if not exists workflow_edges ( | ||
| workflow_run_id text not null, | ||
| from_node_id text not null, | ||
| to_node_id text not null, | ||
| primary key (workflow_run_id, from_node_id, to_node_id), | ||
| foreign key (workflow_run_id) references workflow_runs(id) on delete cascade, | ||
| foreign key (workflow_run_id, from_node_id) | ||
| references workflow_nodes(workflow_run_id, id) on delete cascade, | ||
| foreign key (workflow_run_id, to_node_id) | ||
| references workflow_nodes(workflow_run_id, id) on delete cascade | ||
| ); | ||
|
|
||
| create index if not exists workflow_edges_to_node_idx | ||
| on workflow_edges(workflow_run_id, to_node_id); | ||
|
|
||
| create table if not exists workflow_events ( | ||
| workflow_run_id text not null, | ||
| sequence integer not null, | ||
| event_type text not null, | ||
| node_id text, | ||
| payload_json text not null, | ||
| created_at text not null, | ||
| primary key (workflow_run_id, sequence), | ||
| foreign key (workflow_run_id) references workflow_runs(id) on delete cascade, | ||
| foreign key (workflow_run_id, node_id) | ||
| references workflow_nodes(workflow_run_id, id) | ||
| ); | ||
|
|
||
| create index if not exists workflow_events_cursor_idx | ||
| on workflow_events(workflow_run_id, sequence); | ||
| `); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Fail migration 4 when workflow objects already exist.
IF NOT EXISTS can silently accept an incompatible legacy table that happens to contain the indexed columns, record version 4, and leave runtime queries failing on missing columns or constraints.
src/db/migrations.ts#L184-L264: use plainCREATE TABLE/CREATE INDEX, or explicitly verify every existing object's columns, keys, and foreign keys before recording the migration.src/workflows/migrations.test.ts#L172-L208: cover a legacyworkflow_eventstable containingworkflow_run_idandsequencebut missing required payload/event columns; migration must still fail and roll back.
📍 Affects 2 files
src/db/migrations.ts#L184-L264(this comment)src/workflows/migrations.test.ts#L172-L208
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/db/migrations.ts` around lines 184 - 264, Make migration 4 fail rather
than silently accepting incompatible existing workflow objects: in the workflow
schema creation block of src/db/migrations.ts (lines 184-264), replace the IF
NOT EXISTS table/index creation or validate every existing object’s columns,
keys, and foreign keys before recording the migration. In
src/workflows/migrations.test.ts (lines 172-208), add coverage for a legacy
workflow_events table containing workflow_run_id and sequence but missing
required event/payload columns, and assert the migration fails and rolls back.
Comment on lines
+264
to
+273
| if (row.status === "running" && row.claim_token === claim.claimToken) { | ||
| this.database.sqlite | ||
| .prepare( | ||
| `update workflow_nodes | ||
| set claim_expires_at = ?, updated_at = ? | ||
| where id = ? and status = 'running' and claim_token = ?`, | ||
| ) | ||
| .run(expiresAt, now, row.id, claim.claimToken); | ||
| return this.getNodeRow(claim.workflowId, claim.nodeKey); | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Revoke node authority when its claim lease expires.
A matching token can currently renew or complete a node after claim_expires_at. A delayed worker can therefore beat reclamation despite no longer holding a valid lease.
src/workflows/store.ts#L264-L273: renew only whileclaim_expires_at > now; reject an expired matching token and require a fresh token for reclamation.src/workflows/store.ts#L339-L344: require both token equality and an unexpired lease for running-node transitions, preferably in the update predicate as well.
📍 Affects 1 file
src/workflows/store.ts#L264-L273(this comment)src/workflows/store.ts#L339-L344
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/workflows/store.ts` around lines 264 - 273, The claim lease checks in
src/workflows/store.ts lines 264-273 and 339-344 must revoke expired-token
authority. In the renewal logic around getNodeRow, require claim_expires_at >
now alongside the running status and token match; in the running-node transition
logic, require the same unexpired-lease condition in both validation and the
update predicate, so expired claims are rejected and reclamation requires a
fresh token.
Comment on lines
+538
to
+567
| const running = workers.map((worker, index) => { | ||
| const readyPath = join(coordinationDir, `ready-${index}`); | ||
| const spec: WorkerSpec = { ...worker, stateDir, readyPath, barrierPath }; | ||
| const child = spawn(process.execPath, ["--import", "tsx", fileURLToPath(import.meta.url), JSON.stringify(spec)], { | ||
| stdio: ["ignore", "pipe", "pipe"], | ||
| }); | ||
| let stdout = ""; | ||
| let stderr = ""; | ||
| child.stdout.setEncoding("utf8").on("data", (chunk) => (stdout += chunk)); | ||
| child.stderr.setEncoding("utf8").on("data", (chunk) => (stderr += chunk)); | ||
| const result = new Promise<Record<string, unknown>>((resolve, reject) => { | ||
| child.on("error", reject); | ||
| child.on("exit", (code) => { | ||
| if (code !== 0) reject(new Error(`Workflow worker exited ${code}: ${stderr}`)); | ||
| else resolve(JSON.parse(stdout) as Record<string, unknown>); | ||
| }); | ||
| }); | ||
| return { readyPath, result }; | ||
| }); | ||
| try { | ||
| const deadline = performance.now() + 10_000; | ||
| while (running.some(({ readyPath }) => !existsSync(readyPath))) { | ||
| if (performance.now() >= deadline) throw new Error("Timed out waiting for workflow workers"); | ||
| await delay(5); | ||
| } | ||
| writeFileSync(barrierPath, "start"); | ||
| return await Promise.all(running.map(({ result }) => result)); | ||
| } finally { | ||
| rmSync(coordinationDir, { recursive: true, force: true }); | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Terminate worker subprocesses before removing their barrier directory.
If readiness times out or setup throws, workers continue polling the deleted barrierPath forever. Their open process and pipe handles can hang the test runner.
Proposed cleanup
- return { readyPath, result };
+ return { readyPath, child, result };
...
} finally {
+ for (const { child } of running) {
+ if (child.exitCode === null && child.signalCode === null) {
+ child.kill();
+ }
+ }
+ await Promise.allSettled(running.map(({ result }) => result));
rmSync(coordinationDir, { recursive: true, force: true });
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| const running = workers.map((worker, index) => { | |
| const readyPath = join(coordinationDir, `ready-${index}`); | |
| const spec: WorkerSpec = { ...worker, stateDir, readyPath, barrierPath }; | |
| const child = spawn(process.execPath, ["--import", "tsx", fileURLToPath(import.meta.url), JSON.stringify(spec)], { | |
| stdio: ["ignore", "pipe", "pipe"], | |
| }); | |
| let stdout = ""; | |
| let stderr = ""; | |
| child.stdout.setEncoding("utf8").on("data", (chunk) => (stdout += chunk)); | |
| child.stderr.setEncoding("utf8").on("data", (chunk) => (stderr += chunk)); | |
| const result = new Promise<Record<string, unknown>>((resolve, reject) => { | |
| child.on("error", reject); | |
| child.on("exit", (code) => { | |
| if (code !== 0) reject(new Error(`Workflow worker exited ${code}: ${stderr}`)); | |
| else resolve(JSON.parse(stdout) as Record<string, unknown>); | |
| }); | |
| }); | |
| return { readyPath, result }; | |
| }); | |
| try { | |
| const deadline = performance.now() + 10_000; | |
| while (running.some(({ readyPath }) => !existsSync(readyPath))) { | |
| if (performance.now() >= deadline) throw new Error("Timed out waiting for workflow workers"); | |
| await delay(5); | |
| } | |
| writeFileSync(barrierPath, "start"); | |
| return await Promise.all(running.map(({ result }) => result)); | |
| } finally { | |
| rmSync(coordinationDir, { recursive: true, force: true }); | |
| } | |
| const running = workers.map((worker, index) => { | |
| const readyPath = join(coordinationDir, `ready-${index}`); | |
| const spec: WorkerSpec = { ...worker, stateDir, readyPath, barrierPath }; | |
| const child = spawn(process.execPath, ["--import", "tsx", fileURLToPath(import.meta.url), JSON.stringify(spec)], { | |
| stdio: ["ignore", "pipe", "pipe"], | |
| }); | |
| let stdout = ""; | |
| let stderr = ""; | |
| child.stdout.setEncoding("utf8").on("data", (chunk) => (stdout += chunk)); | |
| child.stderr.setEncoding("utf8").on("data", (chunk) => (stderr += chunk)); | |
| const result = new Promise<Record<string, unknown>>((resolve, reject) => { | |
| child.on("error", reject); | |
| child.on("exit", (code) => { | |
| if (code !== 0) reject(new Error(`Workflow worker exited ${code}: ${stderr}`)); | |
| else resolve(JSON.parse(stdout) as Record<string, unknown>); | |
| }); | |
| }); | |
| return { readyPath, child, result }; | |
| }); | |
| try { | |
| const deadline = performance.now() + 10_000; | |
| while (running.some(({ readyPath }) => !existsSync(readyPath))) { | |
| if (performance.now() >= deadline) throw new Error("Timed out waiting for workflow workers"); | |
| await delay(5); | |
| } | |
| writeFileSync(barrierPath, "start"); | |
| return await Promise.all(running.map(({ result }) => result)); | |
| } finally { | |
| for (const { child } of running) { | |
| if (child.exitCode === null && child.signalCode === null) { | |
| child.kill(); | |
| } | |
| } | |
| await Promise.allSettled(running.map(({ result }) => result)); | |
| rmSync(coordinationDir, { recursive: true, force: true }); | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/workflows/workflows.test.ts` around lines 538 - 567, Update the worker
cleanup in the workflow test’s running/try/finally flow to terminate every
spawned child process before removing coordinationDir. Retain references to the
spawned processes alongside readyPath and result, and in finally request
termination and await worker completion as needed before rmSync deletes the
barrier directory, including timeout and setup-failure paths.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Stack
This is PR 1 of 5 for dynamic workflow orchestration.
Validation
🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Bug Fixes
Tests