feat: coalesce trigger with concurrencyKey pending limit (#143)#148
Merged
Conversation
Step-by-step plan derived from the RFC. Will be deleted after implementation. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Step 1 of coalesce trigger implementation. Type and event definitions only — no runtime behavior changes. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Step 2 of coalesce trigger implementation. Adds a partial unique index to enforce at most one pending run per (job_name, concurrency_key). Updates existing concurrency tests to avoid creating multiple pending runs with the same key (which now violates the constraint). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…erResult (#143) Step 3 of coalesce trigger implementation. Breaking changes: - storage.enqueue()/enqueueMany() return { run, disposition } - trigger() returns TriggerResult (TypedRun & { disposition }) - batchTrigger() returns TriggerResult[] - triggerAndWait() result includes disposition - TriggerResponse includes disposition - run:trigger event no longer emitted for idempotent hits Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…#143) Step 4 of coalesce trigger implementation: - parseUniqueViolation() distinguishes idempotency vs concurrency conflicts - enqueue() catches INSERT conflict, supports coalesce: 'skip' (returns existing pending run) or throws ConflictError - enqueueInTx() refactor: accepts optional transaction for batch atomicity - enqueueMany() uses single transaction with per-item SAVEPOINTs - trigger()/batchTrigger() validate coalesce option and emit run:coalesced - findPendingByConcurrencyKey() shared query with deterministic ordering Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Step 5: expired leases with an existing pending run for the same concurrencyKey are now failed instead of reset to pending (which would violate the partial unique index). Remaining leases are reset per-row with SAVEPOINT to handle concurrent trigger() races. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Step 6: Wire run:coalesced event through the full stack: - durably-react types.ts: add run:coalesced to DurablyEvent union - use-runs.ts: refresh runs list on run:coalesced - use-job-subscription.ts: followLatest reacts to run:coalesced - server.ts: forward run:coalesced via SSE stream Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
The SSE-based client hooks were missing run:coalesced handling: - client/use-runs.ts: refresh list on run:coalesced events - client/use-job.ts: followLatest reacts to run:coalesced Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Step 7: Tests covering concurrencyKey pending limit, coalesce skip behavior, disposition values, events, batchTrigger, triggerAndWait, validation, and releaseExpiredLeases interaction. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Step 8: Update llms.md with: - trigger() returns TriggerResult with disposition - concurrencyKey enforces max 1 pending - coalesce: 'skip' option - run:coalesced event - Updated type definitions Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Step 9: PLAN.md served its purpose — all 9 steps implemented. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…pers (#143) Deduplicate coalesce validation and disposition-based event emission between trigger() and batchTrigger(). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
|
The latest updates on your projects. Learn more about Vercel for GitHub.
1 Skipped Deployment
|
Contributor
|
Warning Rate limit exceeded
⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (2)
📝 WalkthroughWalkthroughトリガーに Changes
Sequence DiagramsequenceDiagram
participant Client
participant Server
participant Database
participant EventEmitter
participant ReactHook
Client->>Server: POST /trigger (input, concurrencyKey, coalesce:'skip')
Server->>Database: INSERT run (status = pending)
alt Unique constraint violation (pending exists)
Database-->>Server: constraint error
Server->>Database: SELECT existing pending run
Database-->>Server: existing run
Server->>EventEmitter: emit('run:coalesced', { runId, jobName, skippedInput, skippedLabels })
Server-->>Client: { run, disposition: 'coalesced' }
else Insert succeeded
Database-->>Server: created run
Server->>EventEmitter: emit('run:trigger', { runId, jobName })
Server-->>Client: { run, disposition: 'created' }
end
EventEmitter->>ReactHook: run:coalesced / run:trigger
ReactHook->>ReactHook: if followLatest -> switch currentRunId
ReactHook->>ReactHook: refresh runs list / UI
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 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 |
- releaseExpiredLeases Phase 2: wrap in transaction (PostgreSQL requires SAVEPOINTs inside transaction blocks) - Add SAVEPOINT name reuse safety comment in enqueueInTx - Add NULL = NULL behavior comment in Phase 1 - Add idempotent no-event comment in emitDispositionEvent - Test: retrigger with pending same-key run throws ConflictError - Test: batch with mixed coalesce/non-coalesce for different keys - Test: idempotencyKey takes priority over concurrencyKey conflict Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
packages/durably/docs/llms.md (1)
289-316:⚠️ Potential issue | 🟡 Minor
run:triggerの idempotent 非発火も明記しておきたいです今回の breaking change にはイベント発火条件の変更も入っているので、イベント一覧か
run:triggerの説明に「idempotency hit では emit されない」を 1 行足しておくと、利用者と LLM の期待値ずれを防げます。 As per coding guidelines, "Updatepackages/durably/docs/llms.mdwhenever API changes are made to keep LLM documentation in sync".🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/durably/docs/llms.md` around lines 289 - 316, Add a one-line note to the Events section next to the run:trigger example clarifying that the 'run:trigger' event is not emitted on an idempotency hit (i.e., when a run is skipped due to idempotency); update the text around the durably.on('run:trigger', ...) sample to include that explicit statement so users and LLMs know idempotent skips do not fire 'run:trigger'.website/public/llms.txt (1)
1-1451:⚠️ Potential issue | 🟠 Major生成ファイルの直接編集を避け、生成コマンド経由にしてください。
website/public/llms.txtは生成物なので、手編集ではなくpackages/*/docs/llms.md側を更新してから再生成した成果物のみをコミットする形に揃えてください。As per coding guidelines, "Never edit generated files directly; regenerate
website/public/llms.txtusingpnpm --filter durably-website generate:llmswhenpackages/*/docs/llms.mdchanges".🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@website/public/llms.txt` around lines 1 - 1451, The committed change edited the generated LLM documentation artifact (llms.txt) directly; instead restore/revert that file and update the canonical source markdown (llms.md under the packages/docs source) then run the docs generation command (pnpm --filter durably-website generate:llms) to regenerate the artifact and commit the generated output; ensure future edits are made to the source llms.md and not the generated llms.txt.
🧹 Nitpick comments (1)
packages/durably/tests/shared/coalesce.shared.ts (1)
192-208:idempotencyKey優先の組み合わせ回帰テストも入れておきたいです仕様上は
idempotencyKeyがcoalesceより優先なので、{ idempotencyKey, concurrencyKey, coalesce: 'skip' }を同時に渡してidempotentを返すケースを 1 本追加しておくと、storage / job の分岐順序を将来壊しにくいです。🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/durably/tests/shared/coalesce.shared.ts` around lines 192 - 208, Add a regression test that verifies idempotencyKey takes precedence over coalesce: add an it block similar to the existing tests that calls d.jobs.job.trigger twice where the first call uses { idempotencyKey: 'idem-1', concurrencyKey: 'key-1' } and the second call uses { idempotencyKey: 'idem-1', concurrencyKey: 'key-1', coalesce: 'skip' }, then assert the second result's disposition is 'idempotent' (use the same pattern as the other tests calling d.jobs.job.trigger and expect(second.disposition).toBe('idempotent')).
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@packages/durably-react/src/hooks/use-job-subscription.ts`:
- Around line 136-143: The handler for the durably.on('run:coalesced') event
currently dispatches dispatch({ type: 'switch_to_run', runId: event.runId })
which causes the reducer to set status: 'leased'; change the handler so that
when event.jobName === jobName and followLatest is true it dispatches an action
that indicates a coalesced pending run (for example a distinct action type like
'switch_to_pending_run' or the existing 'switch_to_run' but with a payload flag
such as { runId: event.runId, status: 'pending' }), and update the reducer to
handle that action and set status: 'pending' instead of 'leased' so coalesced
runs render as pending rather than leased.
In `@packages/durably/src/storage.ts`:
- Around line 297-323: parseUniqueViolation currently returns null for many
non-unique-violation errors causing insertLabelRows and releaseExpiredLeases
catch blocks to treat connection errors and other failures as
pending-concurrency conflicts; update parseUniqueViolation to detect
driver/SQL-state specific unique-violation signals (e.g. Postgres err.code /
SQLSTATE '23505' and any driver-specific constraint property) and only return
'idempotency' or 'pending_concurrency' when you can confidently identify a
UNIQUE violation (fall back to null otherwise); then update the callers
(insertLabelRows and the Phase 2 catch in releaseExpiredLeases that rely on
findPendingByConcurrencyKey) to check the explicit unique-violation result
instead of treating any null/unknown as a concurrency conflict, and keep
original errors (or rethrow) for connection/unknown failures so retries/logging
can surface real issues.
- Around line 762-801: Phase 2 is using SAVEPOINTs outside a transaction and
unguarded updates that can overwrite concurrent state; wrap the per-row
reset/fail logic in an explicit transaction (use db.transaction or tx passed
into the block) instead of raw sql`SAVEPOINT` calls, remove raw savepoint SQL,
and run the update(s) against tx; when updating durably_runs for a reset or
fail, include the original guards in the WHERE (e.g. where id = row.id AND
status = 'leased' AND lease_expires_at <= now and, if available,
lease_generation = <value>) so you only change rows still leased/expired, and in
the catch only handle unique-violation errors by performing the "failed" update
(still with the guarded WHERE) while rethrowing any other DB errors.
---
Outside diff comments:
In `@packages/durably/docs/llms.md`:
- Around line 289-316: Add a one-line note to the Events section next to the
run:trigger example clarifying that the 'run:trigger' event is not emitted on an
idempotency hit (i.e., when a run is skipped due to idempotency); update the
text around the durably.on('run:trigger', ...) sample to include that explicit
statement so users and LLMs know idempotent skips do not fire 'run:trigger'.
In `@website/public/llms.txt`:
- Around line 1-1451: The committed change edited the generated LLM
documentation artifact (llms.txt) directly; instead restore/revert that file and
update the canonical source markdown (llms.md under the packages/docs source)
then run the docs generation command (pnpm --filter durably-website
generate:llms) to regenerate the artifact and commit the generated output;
ensure future edits are made to the source llms.md and not the generated
llms.txt.
---
Nitpick comments:
In `@packages/durably/tests/shared/coalesce.shared.ts`:
- Around line 192-208: Add a regression test that verifies idempotencyKey takes
precedence over coalesce: add an it block similar to the existing tests that
calls d.jobs.job.trigger twice where the first call uses { idempotencyKey:
'idem-1', concurrencyKey: 'key-1' } and the second call uses { idempotencyKey:
'idem-1', concurrencyKey: 'key-1', coalesce: 'skip' }, then assert the second
result's disposition is 'idempotent' (use the same pattern as the other tests
calling d.jobs.job.trigger and expect(second.disposition).toBe('idempotent')).
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 46eb190d-3f1e-4e11-9cd8-aa14295b7250
📒 Files selected for processing (25)
packages/durably-react/src/client/use-job.tspackages/durably-react/src/client/use-runs.tspackages/durably-react/src/hooks/use-job-subscription.tspackages/durably-react/src/hooks/use-runs.tspackages/durably-react/src/types.tspackages/durably/docs/llms.mdpackages/durably/src/durably.tspackages/durably/src/events.tspackages/durably/src/index.tspackages/durably/src/job.tspackages/durably/src/migrations.tspackages/durably/src/server.tspackages/durably/src/storage.tspackages/durably/tests/browser/coalesce.test.tspackages/durably/tests/browser/db-stress.test.tspackages/durably/tests/node/coalesce.test.tspackages/durably/tests/node/migration-consolidated.test.tspackages/durably/tests/shared/coalesce.shared.tspackages/durably/tests/shared/concurrency.shared.tspackages/durably/tests/shared/db-concurrency.shared.tspackages/durably/tests/shared/db-semantics.shared.tspackages/durably/tests/shared/db-stress.shared.tspackages/durably/tests/shared/durably.shared.tspackages/durably/tests/shared/storage.shared.tswebsite/public/llms.txt
- api/events.md: add run:coalesced event documentation - api/define-job.md: trigger/batchTrigger return TriggerResult, TriggerOptions.coalesce, triggerAndWait disposition - api/http-handler.md: coalesce in request, disposition in response - guide/concepts.md: coalesce option in trigger options example - guide/error-handling.md: ConflictError handling pattern Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
- use-job-subscription.ts: switch_to_run accepts optional status, run:coalesced dispatches with status 'pending' (not 'leased') - storage.ts: add isUniqueViolation() to distinguish unique constraint violations from connection/other errors before entering conflict handling - storage.ts: releaseExpiredLeases Phase 2 adds status/lease_expires_at WHERE guards to prevent overwriting concurrent state changes, catch block only handles unique violations (rethrows others) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
- llms.md: note that run:trigger is not emitted on idempotent hits - llms.txt: regenerated from source (not edited directly) - tests: add idempotencyKey precedence over coalesce regression test Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Contributor
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
packages/durably/src/job.ts (1)
338-349:⚠️ Potential issue | 🟠 Major検証済み labels をそのまま使ってください。
ここでは
validateJobInputOrThrow(labelsSchema, ...)の戻り値を捨てているので、schema の strip/default/coerce 結果がstorage.enqueue*()とrun:coalescedのskippedLabelsに反映されません。labelsSchemaでは通る値が後段のvalidateLabels()で落ちたり、余計なラベルが保存されるので、parse 済み値を保持して使い回した方が安全です。💡 修正例
- if (labelsSchema && options?.labels) { - validateJobInputOrThrow(labelsSchema, options.labels, 'labels') - } + const validatedLabels = + labelsSchema && options?.labels + ? validateJobInputOrThrow(labelsSchema, options.labels, 'labels') + : options?.labels const { run, disposition } = await storage.enqueue({ jobName: jobDef.name, input: validatedInput, idempotencyKey: options?.idempotencyKey, concurrencyKey: options?.concurrencyKey, - labels: options?.labels, + labels: validatedLabels, coalesce: options?.coalesce, }) emitDispositionEvent( disposition, run, validatedInput, - options?.labels as Record<string, string>, + validatedLabels as Record<string, string>, ) @@ - if (labelsSchema && opts?.labels) { - validateJobInputOrThrow( - labelsSchema, - opts.labels, - `labels at index ${i}`, - ) - } + const validatedLabels = + labelsSchema && opts?.labels + ? validateJobInputOrThrow( + labelsSchema, + opts.labels, + `labels at index ${i}`, + ) + : opts?.labels validated.push({ input: validatedInput, - options: opts, + options: { ...opts, labels: validatedLabels }, })Also applies to: 352-357, 502-512, 517-523, 527-534
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/durably/src/job.ts` around lines 338 - 349, The code currently calls validateJobInputOrThrow(labelsSchema, options.labels, 'labels') and discards its return value, so schema-driven parsing/stripping/coercion isn’t applied when passing labels to storage.enqueue or when building run.coalesced.skippedLabels; change the call to capture the parsed labels (e.g., const validatedLabels = validateJobInputOrThrow(...)) and use validatedLabels wherever options.labels is forwarded (notably in storage.enqueue and any places that set run.coalesced.skippedLabels or call validateLabels). Update all similar spots that validate labels (the blocks around the storage.enqueue call and the other listed ranges) to use the parsed/returned value instead of the original options.labels.
♻️ Duplicate comments (2)
packages/durably/src/storage.ts (2)
479-499:⚠️ Potential issue | 🟠 Major未分類エラーを pending 競合に畳み込まないでください。
violation === nullは「pending 競合」ではなく「ここでは分類できなかった失敗」です。今の分岐だとconcurrencyKeyがあるだけで INSERT / label 書き込み / 接続系の障害までConflictErrorかcoalescedに化けるので、pending 競合扱いは明示的に識別できたケースだけに絞るべきです。🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/durably/src/storage.ts` around lines 479 - 499, The code currently treats violation === null as a pending concurrency conflict which incorrectly collapses unknown/unclassified errors into pending conflicts; in the block that checks for pending concurrency (using parseUniqueViolation -> violation and input.concurrencyKey), remove the "|| violation === null" condition so only violation === 'pending_concurrency' is handled as a concurrency conflict, and ensure the null/unknown branch falls through (or is rethrown/handled elsewhere) instead of being converted to a ConflictError/coalesced result.
769-809:⚠️ Potential issue | 🔴 CriticalPhase 2 の更新は再ガードしないと別ワーカーの状態を上書きします。
remainingを読んだ後の reset/fail update がidだけなので、その間にcompleteRun()/cancelRun()/renewLease()/ 別のreleaseExpiredLeases()が進むと terminal / renewed 状態をpending/failedに戻せてしまいます。ここは少なくともstatus = 'leased'とlease_expires_at <= now（できれば元のlease_generationも）で再ガードし、0 件更新はcountに含めず、unique 以外の例外は再送出してください。💡 修正例
- const remaining = await db + const remaining = await db .selectFrom('durably_runs') - .select('id') + .select(['id', 'lease_generation']) @@ - await trx + const reset = await trx .updateTable('durably_runs') .set({ status: 'pending', lease_owner: null, lease_expires_at: null, updated_at: now, }) .where('id', '=', row.id) - .execute() - count++ - } catch { + .where('status', '=', 'leased') + .where('lease_expires_at', 'is not', null) + .where('lease_expires_at', '<=', now) + .where('lease_generation', '=', row.lease_generation) + .executeTakeFirst() + count += Number(reset.numUpdatedRows) + } catch (err) { + if (parseUniqueViolation(err) !== 'pending_concurrency') throw err await sql`ROLLBACK TO SAVEPOINT sp_release`.execute(trx) - await trx + const failed = await trx .updateTable('durably_runs') .set({ status: 'failed', error: 'Lease expired; pending run already exists', lease_owner: null, lease_expires_at: null, completed_at: now, updated_at: now, }) .where('id', '=', row.id) - .execute() - count++ + .where('status', '=', 'leased') + .where('lease_expires_at', 'is not', null) + .where('lease_expires_at', '<=', now) + .where('lease_generation', '=', row.lease_generation) + .executeTakeFirst() + count += Number(failed.numUpdatedRows) }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/durably/src/storage.ts` around lines 769 - 809, The phase-2 updates currently only filter by id and can overwrite concurrent state changes; modify the transactional update in the loop that sets status back to 'pending' (the trx.updateTable('durably_runs') call inside the try after SAVEPOINT sp_release) to re-guard the update by adding conditions: status = 'leased' AND lease_expires_at <= now (and include lease_generation if available) so it only updates when the lease still matches; after the update check the affected row count and only increment count when rowsAffected > 0 (do not count 0-updates). In the catch block (around ROLLBACK TO SAVEPOINT sp_release and the subsequent fail update) only handle unique-violation errors specially (perform the fail update and increment count); for any other exception rethrow it. Reference the loop over remaining, the SAVEPOINT sp_release/ROLLBACK/RELEASE calls, the two trx.updateTable('durably_runs') calls, and the count variable when making these changes.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@website/api/define-job.md`:
- Around line 82-83:
「coalesce」オプションだけが有効に見える説明になっているので、オプション表の「coalesce?」行に制約を追記して、(1) coalesce は
concurrencyKey が指定されている場合にのみ有効であること、(2) idempotencyKey がヒットした場合の disposition は
"coalesced" ではなく "idempotent" になることを明記してください；該当する記述（options 表の coalesce
行と同様の箇所、及び disposition の説明があるブロック）を更新して、coalesce と concurrencyKey／idempotencyKey
の関係と期待される disposition の分岐を短い一文で明確に示してください。
In `@website/api/http-handler.md`:
- Around line 132-140: 説明に「coalesce」の前提条件と返却値の意味を明確化してください: coalesce が有効になるのは
concurrencyKey が指定されている場合のみで、idempotencyKey が一致する場合は coalesced ではなく disposition
に "idempotent" を返すこと、また disposition !== "created" の場合の runId は既存の run
を指すことを追記してください。さらに SSE の挙動について run:coalesced イベントが発生することと、coalesced / idempotent
のケースでは run:trigger が発火しないことを明記して、クライアントが分岐処理を誤らないようにしてください（参照: coalesce,
concurrencyKey, idempotencyKey, disposition, runId, SSE events
run:coalesced/run:trigger）。
---
Outside diff comments:
In `@packages/durably/src/job.ts`:
- Around line 338-349: The code currently calls
validateJobInputOrThrow(labelsSchema, options.labels, 'labels') and discards its
return value, so schema-driven parsing/stripping/coercion isn’t applied when
passing labels to storage.enqueue or when building run.coalesced.skippedLabels;
change the call to capture the parsed labels (e.g., const validatedLabels =
validateJobInputOrThrow(...)) and use validatedLabels wherever options.labels is
forwarded (notably in storage.enqueue and any places that set
run.coalesced.skippedLabels or call validateLabels). Update all similar spots
that validate labels (the blocks around the storage.enqueue call and the other
listed ranges) to use the parsed/returned value instead of the original
options.labels.
---
Duplicate comments:
In `@packages/durably/src/storage.ts`:
- Around line 479-499: The code currently treats violation === null as a pending
concurrency conflict which incorrectly collapses unknown/unclassified errors
into pending conflicts; in the block that checks for pending concurrency (using
parseUniqueViolation -> violation and input.concurrencyKey), remove the "||
violation === null" condition so only violation === 'pending_concurrency' is
handled as a concurrency conflict, and ensure the null/unknown branch falls
through (or is rethrown/handled elsewhere) instead of being converted to a
ConflictError/coalesced result.
- Around line 769-809: The phase-2 updates currently only filter by id and can
overwrite concurrent state changes; modify the transactional update in the loop
that sets status back to 'pending' (the trx.updateTable('durably_runs') call
inside the try after SAVEPOINT sp_release) to re-guard the update by adding
conditions: status = 'leased' AND lease_expires_at <= now (and include
lease_generation if available) so it only updates when the lease still matches;
after the update check the affected row count and only increment count when
rowsAffected > 0 (do not count 0-updates). In the catch block (around ROLLBACK
TO SAVEPOINT sp_release and the subsequent fail update) only handle
unique-violation errors specially (perform the fail update and increment count);
for any other exception rethrow it. Reference the loop over remaining, the
SAVEPOINT sp_release/ROLLBACK/RELEASE calls, the two
trx.updateTable('durably_runs') calls, and the count variable when making these
changes.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 25e47986-6c0f-4368-952d-fbed622140a7
📒 Files selected for processing (8)
packages/durably/src/job.tspackages/durably/src/storage.tspackages/durably/tests/shared/coalesce.shared.tswebsite/api/define-job.mdwebsite/api/events.mdwebsite/api/http-handler.mdwebsite/guide/concepts.mdwebsite/guide/error-handling.md
✅ Files skipped from review due to trivial changes (3)
- website/guide/error-handling.md
- website/guide/concepts.md
- website/api/events.md
🚧 Files skipped from review as they are similar to previous changes (1)
- packages/durably/tests/shared/coalesce.shared.ts
- job.ts: use validated labels (parseJobInputOrThrow result) instead of raw options.labels for storage.enqueue and event emission - storage.ts: add comment explaining violation === null safety (after isUniqueViolation gate, only idempotency and pending_concurrency constraints exist on this table) - define-job.md: note coalesce requires concurrencyKey, idempotency takes priority - http-handler.md: document coalesce prerequisites, disposition meaning, SSE behavior for coalesced/idempotent triggers Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Owner
Author
CodeRabbit Review Round 2 — All addressed in 65522dcOutside diff: validated labels (job.ts:338-349)Fixed. Duplicate: violation === null (storage.ts:479-499)This is safe after the Duplicate: Phase 2 re-guard (storage.ts:769-809)Already fixed in 856ecb1 — WHERE clauses include |
Phase 2 count was inflated when concurrent workers already handled the row (WHERE guards cause 0-row updates). Now checks numUpdatedRows. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Reuses shared coalesce test suite with PostgreSQL dialect. Excluded from `pnpm test` by default (*.postgres.test.ts pattern), run with `pnpm --filter @coji/durably test:node:postgres`. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
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
coalesce, a second pending trigger throwsConflictErrorcoalesce: 'skip'— opt-in graceful handling: returns the existing pending run instead of creating a new onetrigger()returnsTriggerResult— extendsTypedRunwithdisposition: 'created' | 'idempotent' | 'coalesced'. Existing code reading run properties works unchangedrun:coalescedevent — new event for observability, carriesskippedInputandskippedLabelsreleaseExpiredLeases()2-phase — handles partial unique index safely when expired leases conflict with existing pending runstrigger()/batchTrigger()return type,TriggerResponseincludesdisposition,run:triggerno longer emitted on idempotent hitsBreaking changes
trigger()returnsTypedRunTriggerResult(TypedRun & { disposition })triggerAndWait()returns{ id, output }{ id, output, disposition }batchTrigger()returnsTypedRun[]TriggerResult[]TriggerResponsehasrunIdrunId+dispositionconcurrencyKeyallows unlimited pendingrun:triggeremitted on idempotent hitsCloses #143
Test plan
pnpm validate)🤖 Generated with Claude Code
Summary by CodeRabbit
新機能
ドキュメント
テスト
マイグレーション