feat: worker + postgres submodules and user guides - #33
Merged
Conversation
|
Warning Rate limit exceeded
Your organization is not enrolled in usage-based pricing. Contact your admin to enable usage-based pricing to continue reviews beyond the rate limit, or try again in 3 minutes and 39 seconds. ⌛ 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: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (11)
📝 WalkthroughWalkthroughAdds an experimental, queue-backed Worker runtime plus Postgres and SQLite persistence stores (queue, checkpointer, events, triggers, webhooks, credits, step progress, activity log), RFC and review for dynamic steps, README cross-links and extensive docs, an in-memory memstore and tests, and a small test refactor in Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant Worker
participant QueueStore
participant Handler
participant Heartbeat as Heartbeat Loop
participant Reaper as Reaper Loop
Client->>QueueStore: Enqueue(NewRun)
QueueStore-->>Client: ok
rect rgba(100, 150, 200, 0.5)
Note over Worker: Claim loop
loop
Worker->>QueueStore: ClaimQueued(workerID)
alt claim found
QueueStore-->>Worker: Claim
Worker->>Handler: Handle(ctx, claim)
else no queued runs
QueueStore-->>Worker: nil
Worker->>Worker: wait (Notify / poll)
end
end
end
rect rgba(150, 100, 200, 0.5)
Note over Heartbeat,QueueStore: Concurrent heartbeat & execution
par
loop while running
Heartbeat->>QueueStore: Heartbeat(Lease)
QueueStore-->>Heartbeat: ok
end
and
Handler->>Handler: execute workflow
Handler-->>Worker: Outcome
end
end
Worker->>QueueStore: Complete(Lease, Outcome)
QueueStore-->>Worker: ok
rect rgba(200, 150, 100, 0.5)
Note over Reaper: Reaper loop
loop periodically
Reaper->>QueueStore: DeadLetterStale(...)
QueueStore-->>Reaper: failedIDs
Reaper->>QueueStore: ReclaimStale(...)
QueueStore-->>Reaper: reclaimedCount
alt reclaimedCount > 0
Reaper->>Worker: Notify()
end
end
end
sequenceDiagram
participant Worker
participant QueueStore
participant EventStore
participant TriggerStore
participant WebhookStore
participant CreditStore
Worker->>QueueStore: Complete(lease, outcome)
QueueStore-->>Worker: ok
rect rgba(100, 200, 150, 0.5)
Note over Worker: afterComplete hooks
Worker->>EventStore: AppendEvent(...)
EventStore-->>Worker: ok
alt outcome == Completed
Worker->>TriggerStore: InsertTriggers(childRuns)
TriggerStore-->>Worker: ok
end
Worker->>WebhookStore: EnqueueWebhook(delivery)
WebhookStore-->>Worker: ok
alt outcome == Failed && creditsDebited
Worker->>CreditStore: Refund(...)
CreditStore-->>Worker: ok
end
end
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
Introduces github.com/deepnoodle-ai/workflow/worker (queue-backed runner with claim loop, heartbeat lease, reaper, and panic recovery) and github.com/deepnoodle-ai/workflow/postgres (a pgx-backed Store that satisfies QueueStore, Checkpointer, StepProgressStore, and ActivityLogger). Both ship as their own Go modules so the root module stays stdlib-only. Adds docs/worker.md and docs/postgres.md as user guides and links them from the README. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
… sqlite module Extend the worker with opt-in lifecycle subsystems — event streaming, transactional outbox triggers, credit ledger, and durable webhooks — plus their Postgres implementations and a new sqlite module providing a full QueueStore/Checkpointer/StepProgressStore/ActivityLogger stack backed by database/sql with no driver dependency. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Rework emission model to buffer-then-commit-on-success, add failed-resume semantics, narrow v1 scope to activity steps only, specify crash-recovery bypass, branch-name validation, precise commit point with rollback, and fix checkpoint example to match real Step wire shape. Add consolidated review document. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
myzie
force-pushed
the
worker-postgres-docs
branch
from
April 12, 2026 14:19
bce4566 to
1ab3e00
Compare
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 17
Note
Due to the large number of review comments, Critical, Major severity comments were prioritized as inline comments.
🟡 Minor comments (14)
docs/dynamic_steps_overlay_rfc.md-318-321 (1)
318-321:⚠️ Potential issue | 🟡 MinorUse “commit time” wording here to match the chosen model.
On Line 319, “rewrites … at staging time” conflicts with the earlier buffer-then-commit-on-success semantics. This should say rewrite at commit time.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@docs/dynamic_steps_overlay_rfc.md` around lines 318 - 321, Update the second bullet in the list (the one that currently reads "engine rewrites them to execution-unique names at staging time") to use "commit time" instead of "staging time" so it matches the buffer-then-commit-on-success semantics; ensure any other occurrences in the same bullet list or nearby bullets consistently use "commit time" (the bullets referencing rewriting intra-plan references and leaving references to static workflow steps unchanged should remain the same).docs/dynamic_steps_overlay_rfc_review.md-5-6 (1)
5-6:⚠️ Potential issue | 🟡 MinorStatus and checklist currently communicate different states.
Line 5 says revisions are still needed, while Line 219–234 shows all revision items complete. Please align these so document state is unambiguous.
Also applies to: 219-234
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@docs/dynamic_steps_overlay_rfc_review.md` around lines 5 - 6, The document shows a contradictory state: the status header string "Status: Review complete, revisions needed before implementation" conflicts with the later "Revision checklist" section where all items are marked complete; update either the header or the checklist so they match. Locate the header line containing "Status: Review complete, revisions needed before implementation" and either change it to reflect that revisions are complete (e.g., "Status: Review complete, revisions completed") or mark the checklist items in the "Revision checklist" section (the items currently shown as complete in lines containing the checklist entries) as still pending; ensure the final phrasing is consistent and unambiguous across both the status header and the "Revision checklist" section.README.md-125-131 (1)
125-131:⚠️ Potential issue | 🟡 MinorDependency statement is inconsistent with the earlier README text.
On Line 130–131, “root module stays stdlib-only” conflicts with Line 15, which states
github.com/deepnoodle-ai/expris an external dependency. Consider wording like: “stays stdlib + expr-only.”🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@README.md` around lines 125 - 131, Update the README wording to resolve the contradiction between the earlier dependency mention and the later claim that the "root module stays stdlib-only": change the phrase "root module stays stdlib-only" (near the sentence describing the worker/ and postgres/ submodules) to something like "root module stays stdlib + github.com/deepnoodle-ai/expr-only" or "stdlib + expr-only" so it aligns with the earlier reference to github.com/deepnoodle-ai/expr; ensure you update the sentence that currently reads "root module stays stdlib-only" to the new wording and keep references to `docs/worker.md`, `docs/postgres.md`, `worker/`, and `postgres/` intact.worker/handler.go-16-27 (1)
16-27:⚠️ Potential issue | 🟡 MinorDocument interrupted-run mapping explicitly as
StatusFailed.Please add explicit guidance that context-canceled/interrupted executions must map to failed outcomes, even if
SetFinished()wasn’t called, to keep handler implementations consistent.As per coding guidelines,
Classify interrupted executions (context canceled mid-run) as failed, even if SetFinished() was never called via buildResult.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@worker/handler.go` around lines 16 - 27, Update the handler comments to explicitly state that executions interrupted by context cancellation (e.g., due to timeout, parent cancel, or lease loss) must be classified as failed outcomes by using buildResult to produce StatusFailed even if SetFinished() was never called; mention the Handle method and buildResult/SetFinished symbols so implementers know to convert context-canceled runs to StatusFailed in the Outcome mapping.postgres/events.go-15-23 (1)
15-23:⚠️ Potential issue | 🟡 MinorGuard
AppendEventagainst nil input.
eventis dereferenced immediately; a nil call path will panic instead of returning a store error.Proposed fix
func (s *Store) AppendEvent(ctx context.Context, event *worker.Event) error { + if event == nil { + return fmt.Errorf("postgres: nil event") + } var payload []byte if event.Payload != nil {🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@postgres/events.go` around lines 15 - 23, The AppendEvent function dereferences the event parameter (and event.Payload) without checking for nil, which can cause a panic; add a guard at the start of Store.AppendEvent to return a descriptive error (e.g., fmt.Errorf("postgres: nil event")) when event == nil, and ensure you still handle event.Payload nil as currently implemented; reference the AppendEvent method on type Store and the worker.Event value to locate where to add the nil check.sqlite/events.go-14-22 (1)
14-22:⚠️ Potential issue | 🟡 MinorAdd nil input validation in
AppendEvent.A nil
eventwill panic on field access; return a clear error instead.Proposed fix
func (s *Store) AppendEvent(ctx context.Context, event *worker.Event) error { + if event == nil { + return fmt.Errorf("sqlite: nil event") + } var payload []byte if event.Payload != nil {🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@sqlite/events.go` around lines 14 - 22, Add a nil-input guard at the start of Store.AppendEvent: check if the incoming event parameter is nil and return a clear error (e.g., fmt.Errorf with "sqlite: append event: nil event") instead of allowing a panic when accessing event.Payload; update the AppendEvent function to perform this validation before any field access or json.Marshal calls.sqlite/checkpointer.go-37-40 (1)
37-40:⚠️ Potential issue | 🟡 MinorHandle
RowsAffected()errors before lease-loss classification.If
RowsAffectedfails, returningErrLeaseLostis misleading and can hide a real DB/driver issue.Proposed fix
- n, _ := result.RowsAffected() + n, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("sqlite: save checkpoint %s rows affected: %w", checkpoint.ExecutionID, err) + } if n == 0 { return worker.ErrLeaseLost }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@sqlite/checkpointer.go` around lines 37 - 40, The code currently ignores the error from result.RowsAffected() and returns worker.ErrLeaseLost on any failure; change the block to check the error returned by result.RowsAffected() first (e.g., n, err := result.RowsAffected()), and if err != nil return that error (or wrap it) so DB/driver errors surface; only when err == nil and n == 0 should you return worker.ErrLeaseLost. Update the code paths using result.RowsAffected() to follow this pattern (referencing result.RowsAffected and worker.ErrLeaseLost).docs/postgres.md-73-75 (1)
73-75:⚠️ Potential issue | 🟡 MinorDocumentation states three tables but schema creates seven.
The text says "
Migratecreates three tables" butpostgres/schema.sqlactually creates seven tables:workflow_runs,workflow_step_progress,workflow_activity_log,workflow_events,workflow_triggers,workflow_credit_ledger, andworkflow_webhooks.📝 Suggested fix
Either update the count and document all tables, or clarify which tables are "core" vs "optional subsystem" tables:
-`Migrate` creates three tables. +`Migrate` creates seven tables. The core tables are documented below; +the remaining tables (`workflow_events`, `workflow_triggers`, +`workflow_credit_ledger`, `workflow_webhooks`) support optional +subsystems documented in [`docs/worker.md`](./worker.md).🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@docs/postgres.md` around lines 73 - 75, The docs claim "`Migrate` creates three tables" but postgres/schema.sql actually creates seven tables (workflow_runs, workflow_step_progress, workflow_activity_log, workflow_events, workflow_triggers, workflow_credit_ledger, workflow_webhooks); update the docs: either change the count to seven and list/describe each table, or explain which three are "core" (name them, e.g., workflow_runs, workflow_step_progress, workflow_activity_log) and mark the others as optional/subsystem tables (workflow_events, workflow_triggers, workflow_credit_ledger, workflow_webhooks) so the documentation matches the schema.postgres/webhooks.go-65-67 (1)
65-67:⚠️ Potential issue | 🟡 MinorUnnecessary
pgx.ErrNoRowscheck and missing error context.
rows.Err()returns iteration errors fromrows.Next(), notpgx.ErrNoRows(which comes fromQueryRow.Scan). The check is harmless but misleading. Additionally, the error on line 66 is returned without the"postgres:"prefix used consistently elsewhere.🔧 Proposed fix
} - if err := rows.Err(); err != nil && err != pgx.ErrNoRows { - return nil, err + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("postgres: list pending webhooks: %w", err) } return out, nil🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@postgres/webhooks.go` around lines 65 - 67, In the rows iteration cleanup block where rows.Err() is checked (the rows.Err() check in postgres/webhooks.go), remove the unnecessary comparison to pgx.ErrNoRows (rows.Err() will never return pgx.ErrNoRows) and return the iteration error with the consistent "postgres:" prefix (e.g., wrap or prefix the returned error with "postgres:") so it matches other error messages in this package.sqlite/webhooks.go-13-27 (1)
13-27:⚠️ Potential issue | 🟡 MinorGenerated ID not assigned back to the delivery struct.
Same issue as the Postgres implementation: when
delivery.IDis empty, a new ID is generated but not written back to the struct.🔧 Proposed fix
func (s *Store) EnqueueWebhook(ctx context.Context, delivery *worker.WebhookDelivery) error { id := delivery.ID if id == "" { id = generateID("whk_") + delivery.ID = id } _, err := s.db.ExecContext(ctx, `🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@sqlite/webhooks.go` around lines 13 - 27, In EnqueueWebhook, when a new ID is generated for an empty delivery.ID, assign that generated id back to the delivery struct before executing the INSERT so the caller sees the persisted ID; locate the EnqueueWebhook function and set delivery.ID = id (or otherwise update the delivery.ID field) prior to calling s.db.ExecContext with the generated id.sqlite/store.go-86-90 (1)
86-90:⚠️ Potential issue | 🟡 MinorHandle
rand.Readerror to avoid predictable IDs.Ignoring the error from
crypto/rand.Readcould produce non-random or partially-filled IDs if the system entropy source fails. While rare, this could lead to ID collisions or predictability issues.🛡️ Proposed fix
func generateID(prefix string) string { var b [8]byte - _, _ = rand.Read(b[:]) + if _, err := rand.Read(b[:]); err != nil { + panic("crypto/rand unavailable: " + err.Error()) + } return prefix + hex.EncodeToString(b[:]) }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@sqlite/store.go` around lines 86 - 90, The generateID function currently ignores the error from crypto/rand.Read which can produce weak IDs; update generateID to check the return error from rand.Read (in the generateID function) and handle it deterministically (e.g., return an error up the stack by changing the signature to generateID(prefix string) (string, error) or, if callers cannot change, fail fast by logging and exiting/panicking) instead of proceeding to hex.EncodeToString on possibly-uninitialized bytes; ensure callers are updated to handle the new (string, error) signature if you choose that approach.sqlite/triggers.go-76-78 (1)
76-78:⚠️ Potential issue | 🟡 MinorSilently ignoring JSON unmarshal error could mask data corruption.
If
childSpeccontains invalid JSON, the error is discarded andt.ChildSpecremains zero-valued. This could cause subtle downstream bugs when the trigger is processed.🔧 Proposed fix
if childSpec != "" { - _ = json.Unmarshal([]byte(childSpec), &t.ChildSpec) + if err := json.Unmarshal([]byte(childSpec), &t.ChildSpec); err != nil { + return nil, fmt.Errorf("sqlite: unmarshal trigger child spec: %w", err) + } }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@sqlite/triggers.go` around lines 76 - 78, The json.Unmarshal call on childSpec is currently ignoring errors: when childSpec != "" the code calls json.Unmarshal([]byte(childSpec), &t.ChildSpec) without checking the returned error; change this to capture the error (err := json.Unmarshal(...)) and handle it appropriately (return the error up the call chain or log it with context and skip/mark the trigger) so corrupted JSON isn't silently ignored—update the logic surrounding childSpec and t.ChildSpec to propagate or surface the unmarshal error.postgres/webhooks.go-14-28 (1)
14-28:⚠️ Potential issue | 🟡 MinorGenerated ID not assigned back to the delivery struct.
When
delivery.IDis empty, a new ID is generated but not written back to the struct. The caller has no way to retrieve the assigned ID after enqueue.🔧 Proposed fix
func (s *Store) EnqueueWebhook(ctx context.Context, delivery *worker.WebhookDelivery) error { id := delivery.ID if id == "" { id = generateID("whk_") + delivery.ID = id } _, err := s.pool.Exec(ctx, `🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@postgres/webhooks.go` around lines 14 - 28, In EnqueueWebhook, when you generate a new ID with generateID("whk_") because delivery.ID is empty, assign that generated value back to the delivery struct (delivery.ID = id) before executing the INSERT so the caller can read the assigned ID; update the code path in Store.EnqueueWebhook that sets id to ensure delivery.ID is updated when using worker.WebhookDelivery.postgres/store_test.go-104-110 (1)
104-110:⚠️ Potential issue | 🟡 MinorReplace direct sentinel comparisons with
errors.Is.These checks use direct equality (
err != worker.ErrLeaseLost,err != workflow.ErrNoCheckpoint) which violates the error handling guideline. Sentinels must be checked viaerrors.Is()to remain compatible with wrapped implementations that callers should treat as equivalent. Replace all occurrences at lines 104, 109, 170, 198, and 206.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@postgres/store_test.go` around lines 104 - 110, Replace direct sentinel error equality checks with errors.Is checks: where the test compares the return from store.Heartbeat (and other functions) against worker.ErrLeaseLost or workflow.ErrNoCheckpoint, change the assertions to use errors.Is(err, worker.ErrLeaseLost) and errors.Is(err, workflow.ErrNoCheckpoint). For example, for the Heartbeat calls that pass `wrong` and `badAttempt` (worker.Lease with claim.ID), assert with errors.Is(...) and fail the test when that returns false; apply the same replacement to the other occurrences referenced (lines called out around 170, 198, 206) so all sentinel comparisons use errors.Is.
🧹 Nitpick comments (9)
worker/events.go (1)
30-33: Consider adding a bounded page size toListEvents.On Line 30–33,
afterSeqgives cursoring, but without alimitcontract this can still return very large batches for long-lived runs. A max page size in the interface (or explicit documented cap) would reduce memory and query pressure.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@worker/events.go` around lines 30 - 33, The ListEvents(ctx context.Context, runID string, afterSeq int64) ([]*Event, error) API exposes cursoring via afterSeq but lacks a page size cap; update the interface to accept a limit parameter (e.g., ListEvents(ctx, runID, afterSeq, limit int) or document/enforce a hard max) and update all implementations to honor and validate that limit (apply a sensible max cap, return at most limit events, and return a clear error for invalid limits) so callers cannot receive arbitrarily large batches of *Event.worker/triggers.go (1)
38-38: Definelimitsemantics for non-positive values.On Line 38, please document/enforce behavior for
limit <= 0(e.g., reject or clamp). That avoids backend-specific unbounded reads in pending-trigger polling loops.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@worker/triggers.go` at line 38, The ListPendingTriggers(ctx context.Context, limit int) signature lacks defined semantics for limit <= 0; update the API and implementations to either reject non-positive limits (return a clear error like ErrInvalidLimit) or clamp them to a safe default/max (e.g., defaultLimit or maxLimit) and document this behavior in the function comment. Modify the ListPendingTriggers interface doc comment in triggers.go and enforce the chosen policy at the start of each implementation of ListPendingTriggers (validate the limit parameter and either return the error or replace limit with the clamp value) so polling loops cannot trigger backend-specific unbounded reads.sqlite/schema.sql (1)
64-77: Add acreated_atindex for event cleanup scalability.
CleanupEventsdeletes bycreated_at; without an index this becomes a full scan on large event tables.Proposed fix
CREATE INDEX IF NOT EXISTS workflow_events_run ON workflow_events (run_id, seq); + +CREATE INDEX IF NOT EXISTS workflow_events_created_at + ON workflow_events (created_at);🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@sqlite/schema.sql` around lines 64 - 77, The cleanup path (CleanupEvents) deletes rows by created_at but there is no index, causing full table scans; add an index on the workflow_events.created_at column to improve delete scalability by adding a statement like CREATE INDEX IF NOT EXISTS workflow_events_created_at ON workflow_events (created_at) to the schema (alongside the existing workflow_events and workflow_events_run index) so CleanupEvents can use the index for efficient deletes.worker/worker_test.go (1)
79-79: Avoid discarding enqueue errors in test setup.Ignoring these errors can hide setup failures and make downstream assertions noisy.
Proposed fix
- _ = store.Enqueue(context.Background(), worker.NewRun{ID: "crash"}) + if err := store.Enqueue(context.Background(), worker.NewRun{ID: "crash"}); err != nil { + t.Fatal(err) + } ... - _ = store.Enqueue(context.Background(), worker.NewRun{ID: "stale"}) + if err := store.Enqueue(context.Background(), worker.NewRun{ID: "stale"}); err != nil { + t.Fatal(err) + }Also applies to: 122-122
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@worker/worker_test.go` at line 79, The test currently discards errors from store.Enqueue calls (e.g., store.Enqueue(context.Background(), worker.NewRun{ID: "crash"})), so update each enqueue in the test setup to capture the returned error and fail the test on error (for example, assign err := store.Enqueue(...); then call t.Fatalf or require.NoError(t, err) with a clear message). Apply this change to the occurrences that discard the error (including the instance using worker.NewRun{ID: "crash"} and the second occurrence noted around the later enqueue) so setup failures are surfaced instead of hidden.docs/worker.md (1)
104-106: Example uses hardcoded WorkerID which may confuse readers.The example hardcodes
WorkerID: "w"in the lease, but the configuration reference shows thatWorkerIDis typically auto-generated asworker-<host>-<rand>. Consider showing how to obtain the actual worker ID from the worker instance for clarity.📝 Suggested improvement
- lease := worker.Lease{RunID: c.ID, WorkerID: "w", Attempt: c.Attempt} + // workerID is typically obtained from the Worker instance or passed into the handler factory + lease := worker.Lease{RunID: c.ID, WorkerID: workerID, Attempt: c.Attempt}Alternatively, update the handler factory signature to accept the worker ID:
func handleRun(pgStore *postgres.Store, reg *workflow.ActivityRegistry, workerID string) worker.HandlerFunc {🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@docs/worker.md` around lines 104 - 106, The example currently hardcodes WorkerID: "w" in the worker.Lease construction; change it to use the actual worker ID from the runtime or handler context instead. Either retrieve the worker ID from the worker instance (e.g., use the worker object's ID accessor when creating lease := worker.Lease{RunID: c.ID, WorkerID: worker.ID(), Attempt: c.Attempt}) or update the handler factory signature (handleRun) to accept a workerID parameter (e.g., func handleRun(..., workerID string) worker.HandlerFunc) and pass that workerID into the Lease construction; ensure references to RunID, WorkerID, Attempt and the worker.HandlerFunc/handleRun symbols are updated accordingly.sqlite/triggers.go (1)
42-42: Wrap commit error for consistency.Other errors in this file use the
"sqlite: ..."prefix for context. The commit error should follow the same pattern.♻️ Proposed fix
- return tx.Commit() + if err := tx.Commit(); err != nil { + return fmt.Errorf("sqlite: commit trigger insert tx: %w", err) + } + return nil🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@sqlite/triggers.go` at line 42, The tx.Commit() return in sqlite/triggers.go should wrap the error with the same "sqlite: ..." prefix used elsewhere; replace the bare return tx.Commit() with error handling that captures err from tx.Commit() and returns fmt.Errorf("sqlite: commit triggers: %w", err) (or a similar "sqlite: ..." message consistent with other errors), ensuring you import fmt if needed and keep the tx variable and commit call unchanged.worker/webhooks.go (1)
16-16: Consider using typed status constants for consistency.The
Statusfield uses raw strings ("pending","delivered","failed"), whileworker/triggers.godefines typedTriggerStatusconstants (TriggerPending,TriggerProcessing, etc.). Using a similar pattern here would improve type safety and consistency across the codebase.♻️ Suggested pattern
type WebhookStatus string const ( WebhookPending WebhookStatus = "pending" WebhookDelivered WebhookStatus = "delivered" WebhookFailed WebhookStatus = "failed" ) type WebhookDelivery struct { // ... Status WebhookStatus // ... }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@worker/webhooks.go` at line 16, The Status field on WebhookDelivery currently uses raw strings; introduce a new typed status alias (e.g., type WebhookStatus string) and define constants (WebhookPending, WebhookDelivered, WebhookFailed) and change the Status field in the WebhookDelivery struct to use WebhookStatus instead of string; update any constructors, serializers, DB mappings, and comparisons that reference WebhookDelivery.Status to use the new WebhookStatus constants (search for WebhookDelivery and Status to locate usages).worker/memstore/memstore.go (1)
178-195: Mirror the running-state fence on completion here too.This in-memory store has the same edge case as the Postgres store:
claimedBy == ""is also the unclaimed sentinel, so a stale lease can still satisfy this check afterReclaimStale/DeadLetterStaleif an empty worker ID ever slips through. RequiringStatusRunninghere keeps the test store aligned with the durable store and makes the fence more robust.Possible hardening
row, ok := s.runs[lease.RunID] if !ok { return worker.ErrLeaseLost } - if row.claimedBy != lease.WorkerID || row.attempt != lease.Attempt { + if row.status != worker.StatusRunning || + row.claimedBy != lease.WorkerID || + row.attempt != lease.Attempt { return worker.ErrLeaseLost }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@worker/memstore/memstore.go` around lines 178 - 195, In Store.Complete, add the same running-state fence used in the Postgres store by checking that the run is currently running before accepting a completion: in the Complete method (type Store, function Complete) after validating the run exists and matching lease.WorkerID/lease.Attempt (row, s.runs, row.claimedBy, row.attempt), also require row.status == worker.StatusRunning (or return worker.ErrLeaseLost) so an unclaimed sentinel (claimedBy == "") cannot be used to complete a stale lease; keep the rest of the logic (setting row.status, row.result, row.errorMessage, row.completedAt) unchanged.postgres/queue.go (1)
15-27: Use shared sentinels for enqueue contract errors.The empty-ID path here, and the duplicate-ID path once
Execfails, both surface as opaque strings. That makes it hard for callers to branch witherrors.Isand keeps the Postgres and memstore implementations inconsistent at theQueueStoreboundary. Please promote these to shared sentinels and return them from both stores.As per coding guidelines "Use error sentinels with
errors.Isand structured errors viaWorkflowErrorfor error handling".🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@postgres/queue.go` around lines 15 - 27, Create shared sentinel errors (e.g., var ErrEmptyRunID = errors.New("empty run id") and var ErrDuplicateRunID = errors.New("duplicate run id")) in the queue package used by both Postgres and memstore implementations, then update Store.Enqueue: return ErrEmptyRunID when run.ID == "" instead of the opaque fmt error, and when s.pool.Exec returns an error detect Postgres unique-violation (pgconn.PgError with Code "23505") and return ErrDuplicateRunID (wrap it if you need context, e.g., fmt.Errorf("%w: %v", ErrDuplicateRunID, err) or via your WorkflowError wrapper) so callers can use errors.Is to branch; ensure memstore returns the same sentinels for its empty-ID and duplicate-ID paths as well so QueueStore behavior is consistent.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@postgres/checkpointer.go`:
- Around line 86-89: DeleteCheckpoint currently clears checkpoint by id only and
must be fenced like SaveCheckpoint to avoid stale-writer overwrite: modify
leasedCheckpointer.DeleteCheckpoint to include the same WHERE clauses for
claimed_by and attempt (e.g. "UPDATE workflow_runs SET checkpoint = NULL WHERE
id = $1 AND claimed_by = $2 AND attempt = $3"), pass executionID, c.claimedBy
(or the equivalent claimed_by field) and c.attempt as arguments to
c.store.pool.Exec, inspect the command tag / rows affected and if zero rows were
updated return worker.ErrLeaseLost instead of nil; mirror the parameter ordering
and error handling used in SaveCheckpoint.
- Around line 40-46: SaveCheckpoint currently uses checkpoint.ExecutionID in the
WHERE clause; change it to use the leased run id (c.lease.RunID), validate that
checkpoint.ExecutionID matches c.lease.RunID and reject mismatches, and keep the
existing (claimed_by = c.lease.WorkerID, attempt = c.lease.Attempt) fencing; for
DeleteCheckpoint add the same lease fencing to the UPDATE/DELETE (include
claimed_by and attempt conditions using c.lease.WorkerID and c.lease.Attempt),
check the affected row count and return worker.ErrLeaseLost when zero rows are
updated so stale workers cannot modify another run's checkpoint.
In `@postgres/events.go`:
- Around line 67-69: The code silently ignores json.Unmarshal errors when
decoding payload into e.Payload (payload, e.Payload, json.Unmarshal); change the
blank-assignment to capture the error (err := json.Unmarshal(payload,
&e.Payload)), and propagate or handle it instead of discarding—e.g., return or
wrap the error from the surrounding function (or log and return a wrapped error)
so callers see corrupted/invalid payloads rather than receiving partially
decoded events.
In `@postgres/queue.go`:
- Around line 115-123: The Complete fence currently only checks claimed_by and
attempt, allowing stale/empty workerID collisions; update the WHERE clause in
the postgres Complete function that contains the shown UPDATE (the method named
Complete in queue.go) to also require status = StatusRunning (or the same status
parameter used by Heartbeat) so the UPDATE only succeeds if the run is still
running, and mirror this same status check in the Complete method in
worker/memstore/memstore.go so both stores enforce the same fence semantics.
In `@postgres/step_progress.go`:
- Around line 21-22: Replace the plain fmt.Errorf wrapping in the "marshal
progress detail" error branches with a sentinel-compatible WorkflowError: define
a package sentinel error (e.g. ErrMarshalProgressDetail) and return a
WorkflowError that wraps the underlying err while exposing that sentinel so
callers can use errors.Is; update both occurrences (the marshal progress detail
branch and the similar branch around line 53) to return the
WorkflowError-wrapped error instead of fmt.Errorf.
In `@postgres/store.go`:
- Around line 34-49: Change New to validate the provided pool and fail fast:
update New signature to return (*Store, error), check if pool == nil and return
a clear error (e.g., fmt.Errorf("nil pgxpool.Pool")) instead of constructing a
Store that will panic later in Migrate; keep applying opts when non-nil but
ignore/skip any nil logger option values, and update callers to handle the new
error return. Ensure references to New, Store, pool, opts, and Migrate remain
consistent.
- Line 9: The postgres submodule currently imports
"github.com/jackc/pgx/v5/pgxpool" in postgres/store.go which violates the
single-module constraint in CLAUDE.md; either remove the pgx dependency by
deleting that import and refactoring any usages in postgres/store.go and related
types to use only stdlib-compatible abstractions (e.g., redesign to use
interfaces over *pgxpool.Pool and implement a stdlib-only backend, update
postgres/go.mod to drop pgx), or treat postgres as an independent module by
documenting this exception in CLAUDE.md and updating the repository
constraint/pattern to allow the new dependency; locate references to pgx (import
and any functions/methods using pgxpool) to refactor or justify in the doc.
In `@postgres/triggers.go`:
- Around line 99-106: MarkTriggerProcessing must perform a compare-and-swap to
actually claim a pending trigger: update the SQL in MarkTriggerProcessing to
include the current expected status in the WHERE clause (e.g., "WHERE id = $2
AND status = $3") and pass the expected value (worker.TriggerPending) along with
the new value (string(worker.TriggerProcessing)); after Exec, inspect the
result's RowsAffected() and if it is 0 return a sentinel/error indicating the
trigger was already claimed (treat as non-retriable/already-claimed) instead of
success; leave existing error wrapping for Exec failures.
- Around line 81-83: The current code silently ignores JSON decode errors for
the childSpec variable and leaves t.ChildSpec zero-valued; modify the block
around json.Unmarshal(childSpec, &t.ChildSpec) so that if Unmarshal returns an
error you wrap and return that error (including context like the trigger id or
row identifier) instead of discarding it, ensuring the caller (processTriggers)
sees a clear failure; update the function that contains this logic to propagate
the wrapped error up the call stack rather than swallowing it.
In `@sqlite/checkpointer.go`:
- Around line 63-66: The code currently only rejects checkpoints with
cp.SchemaVersion greater than workflow.CheckpointSchemaVersion; also add a guard
that rejects cp.SchemaVersion below the minimum supported schema version by
returning an error (e.g., "sqlite: checkpoint schema v%d is older than minimum
supported v%d"). Add a check comparing cp.SchemaVersion to the workflow
package's minimum supported constant (for example
workflow.MinSupportedCheckpointSchemaVersion or similar existing symbol)
alongside the existing upper-bound check so both too-new and too-old schema
versions fail in the function that performs this validation.
In `@sqlite/events.go`:
- Around line 68-70: ListEvents currently swallows JSON unmarshal errors for
payload fields (the json.Unmarshal call that decodes payload.String into
e.Payload), which hides corrupt rows; change ListEvents to check and surface the
error instead of ignoring it: capture the error returned by
json.Unmarshal([]byte(payload.String), &e.Payload), return or wrap that error
from ListEvents (or record which event ID failed) so callers are aware of decode
failures, and update any callers or tests expecting successful decode
accordingly. Ensure you reference the payload variable and e.Payload in the
error message or wrapper to make debugging straightforward.
In `@sqlite/step_progress.go`:
- Around line 17-18: Replace the plain fmt.Errorf returns for the
marshal/unmarshal cases (the "sqlite: marshal progress detail: %w" occurrence
and the similar error at the other location) with a structured WorkflowError
that uses a sentinel error so callers can use errors.Is; define or reuse a
sentinel like ErrProgressMarshal and return the WorkflowError via the project's
WorkflowError constructor (e.g., NewWorkflowError or the existing constructor
used elsewhere) including the sentinel, a short context message ("marshal
progress detail") and the original err as the cause so upstream code can
classify failures with errors.Is and inspect the underlying error.
In `@worker/subsystems.go`:
- Around line 29-37: debitCredits currently treats debit failures as best-effort
which allows afterComplete to unconditionally call Refund and create negative
standalone ledger entries when the debit never landed; update debitCredits (or
add a new helper) to return a boolean or error indicating whether a debit was
recorded (e.g., debitCredits -> (bool, error) or DebitResult), then modify
afterComplete to only call cfg.CreditStore.Refund when that debit-recorded flag
is true; alternatively remove the eager Refund call in afterComplete and rely
solely on ListUnrefunded-based reconciliation (remove or short-circuit the
Refund path). Ensure references to Claim.ID/OrgID/CreditCost and methods
CreditStore.Debit and CreditStore.Refund are used to gate the refund logic.
- Around line 176-198: The processWebhooks loop lacks a store-level claim step
so multiple workers can deliver the same webhook; update the flow to atomically
claim a webhook before calling Deliver by adding or using a store method (e.g.,
WebhookStore.ClaimPendingWebhook(ctx, id) or
WebhookStore.ListAndClaimPending(ctx, limit)) so only the claiming worker
proceeds to call WebhookDeliverer.Deliver; if the claim fails or returns false,
skip that webhook; keep subsequent transitions (MarkWebhookDelivered,
IncrementWebhookAttempts, MarkWebhookFailed) unchanged but ensure they operate
on claimed webhooks and that failed claims are handled by skipping delivery.
In `@worker/worker.go`:
- Around line 341-353: The outcome produced by w.safeHandle(runCtx, claim) must
be normalized to a failed terminal state when the runCtx was canceled mid-run;
before calling w.store.Complete(finalizeCtx, lease, outcome) check runCtx.Err()
and, if non-nil, replace or modify the outcome returned by safeHandle so its
status is set to StatusFailed (and include the context error in the outcome
message/reason) to ensure interrupted executions are classified as failed even
if SetFinished/buildResult never ran; do this normalization immediately after
computing outcome and before the Finalize/Complete call (referencing outcome,
runCtx, w.safeHandle, and w.store.Complete).
- Around line 188-196: If cfg.WebhookStore is provided but cfg.WebhookDeliverer
is nil, reject construction so callbacks don't queue forever: in the Worker
constructor (the code that sets defaults in worker.go where
cfg.Logger/Clock/WorkerID are initialized) add a validation that returns an
error when cfg.WebhookStore != nil && cfg.WebhookDeliverer == nil, referencing
WebhookStore, WebhookDeliverer and the afterComplete/Run/webhookLoop behavior;
update any callers/tests that construct Worker to handle the error.
---
Minor comments:
In `@docs/dynamic_steps_overlay_rfc_review.md`:
- Around line 5-6: The document shows a contradictory state: the status header
string "Status: Review complete, revisions needed before implementation"
conflicts with the later "Revision checklist" section where all items are marked
complete; update either the header or the checklist so they match. Locate the
header line containing "Status: Review complete, revisions needed before
implementation" and either change it to reflect that revisions are complete
(e.g., "Status: Review complete, revisions completed") or mark the checklist
items in the "Revision checklist" section (the items currently shown as complete
in lines containing the checklist entries) as still pending; ensure the final
phrasing is consistent and unambiguous across both the status header and the
"Revision checklist" section.
In `@docs/dynamic_steps_overlay_rfc.md`:
- Around line 318-321: Update the second bullet in the list (the one that
currently reads "engine rewrites them to execution-unique names at staging
time") to use "commit time" instead of "staging time" so it matches the
buffer-then-commit-on-success semantics; ensure any other occurrences in the
same bullet list or nearby bullets consistently use "commit time" (the bullets
referencing rewriting intra-plan references and leaving references to static
workflow steps unchanged should remain the same).
In `@docs/postgres.md`:
- Around line 73-75: The docs claim "`Migrate` creates three tables" but
postgres/schema.sql actually creates seven tables (workflow_runs,
workflow_step_progress, workflow_activity_log, workflow_events,
workflow_triggers, workflow_credit_ledger, workflow_webhooks); update the docs:
either change the count to seven and list/describe each table, or explain which
three are "core" (name them, e.g., workflow_runs, workflow_step_progress,
workflow_activity_log) and mark the others as optional/subsystem tables
(workflow_events, workflow_triggers, workflow_credit_ledger, workflow_webhooks)
so the documentation matches the schema.
In `@postgres/events.go`:
- Around line 15-23: The AppendEvent function dereferences the event parameter
(and event.Payload) without checking for nil, which can cause a panic; add a
guard at the start of Store.AppendEvent to return a descriptive error (e.g.,
fmt.Errorf("postgres: nil event")) when event == nil, and ensure you still
handle event.Payload nil as currently implemented; reference the AppendEvent
method on type Store and the worker.Event value to locate where to add the nil
check.
In `@postgres/store_test.go`:
- Around line 104-110: Replace direct sentinel error equality checks with
errors.Is checks: where the test compares the return from store.Heartbeat (and
other functions) against worker.ErrLeaseLost or workflow.ErrNoCheckpoint, change
the assertions to use errors.Is(err, worker.ErrLeaseLost) and errors.Is(err,
workflow.ErrNoCheckpoint). For example, for the Heartbeat calls that pass
`wrong` and `badAttempt` (worker.Lease with claim.ID), assert with
errors.Is(...) and fail the test when that returns false; apply the same
replacement to the other occurrences referenced (lines called out around 170,
198, 206) so all sentinel comparisons use errors.Is.
In `@postgres/webhooks.go`:
- Around line 65-67: In the rows iteration cleanup block where rows.Err() is
checked (the rows.Err() check in postgres/webhooks.go), remove the unnecessary
comparison to pgx.ErrNoRows (rows.Err() will never return pgx.ErrNoRows) and
return the iteration error with the consistent "postgres:" prefix (e.g., wrap or
prefix the returned error with "postgres:") so it matches other error messages
in this package.
- Around line 14-28: In EnqueueWebhook, when you generate a new ID with
generateID("whk_") because delivery.ID is empty, assign that generated value
back to the delivery struct (delivery.ID = id) before executing the INSERT so
the caller can read the assigned ID; update the code path in
Store.EnqueueWebhook that sets id to ensure delivery.ID is updated when using
worker.WebhookDelivery.
In `@README.md`:
- Around line 125-131: Update the README wording to resolve the contradiction
between the earlier dependency mention and the later claim that the "root module
stays stdlib-only": change the phrase "root module stays stdlib-only" (near the
sentence describing the worker/ and postgres/ submodules) to something like
"root module stays stdlib + github.com/deepnoodle-ai/expr-only" or "stdlib +
expr-only" so it aligns with the earlier reference to
github.com/deepnoodle-ai/expr; ensure you update the sentence that currently
reads "root module stays stdlib-only" to the new wording and keep references to
`docs/worker.md`, `docs/postgres.md`, `worker/`, and `postgres/` intact.
In `@sqlite/checkpointer.go`:
- Around line 37-40: The code currently ignores the error from
result.RowsAffected() and returns worker.ErrLeaseLost on any failure; change the
block to check the error returned by result.RowsAffected() first (e.g., n, err
:= result.RowsAffected()), and if err != nil return that error (or wrap it) so
DB/driver errors surface; only when err == nil and n == 0 should you return
worker.ErrLeaseLost. Update the code paths using result.RowsAffected() to follow
this pattern (referencing result.RowsAffected and worker.ErrLeaseLost).
In `@sqlite/events.go`:
- Around line 14-22: Add a nil-input guard at the start of Store.AppendEvent:
check if the incoming event parameter is nil and return a clear error (e.g.,
fmt.Errorf with "sqlite: append event: nil event") instead of allowing a panic
when accessing event.Payload; update the AppendEvent function to perform this
validation before any field access or json.Marshal calls.
In `@sqlite/store.go`:
- Around line 86-90: The generateID function currently ignores the error from
crypto/rand.Read which can produce weak IDs; update generateID to check the
return error from rand.Read (in the generateID function) and handle it
deterministically (e.g., return an error up the stack by changing the signature
to generateID(prefix string) (string, error) or, if callers cannot change, fail
fast by logging and exiting/panicking) instead of proceeding to
hex.EncodeToString on possibly-uninitialized bytes; ensure callers are updated
to handle the new (string, error) signature if you choose that approach.
In `@sqlite/triggers.go`:
- Around line 76-78: The json.Unmarshal call on childSpec is currently ignoring
errors: when childSpec != "" the code calls json.Unmarshal([]byte(childSpec),
&t.ChildSpec) without checking the returned error; change this to capture the
error (err := json.Unmarshal(...)) and handle it appropriately (return the error
up the call chain or log it with context and skip/mark the trigger) so corrupted
JSON isn't silently ignored—update the logic surrounding childSpec and
t.ChildSpec to propagate or surface the unmarshal error.
In `@sqlite/webhooks.go`:
- Around line 13-27: In EnqueueWebhook, when a new ID is generated for an empty
delivery.ID, assign that generated id back to the delivery struct before
executing the INSERT so the caller sees the persisted ID; locate the
EnqueueWebhook function and set delivery.ID = id (or otherwise update the
delivery.ID field) prior to calling s.db.ExecContext with the generated id.
In `@worker/handler.go`:
- Around line 16-27: Update the handler comments to explicitly state that
executions interrupted by context cancellation (e.g., due to timeout, parent
cancel, or lease loss) must be classified as failed outcomes by using
buildResult to produce StatusFailed even if SetFinished() was never called;
mention the Handle method and buildResult/SetFinished symbols so implementers
know to convert context-canceled runs to StatusFailed in the Outcome mapping.
---
Nitpick comments:
In `@docs/worker.md`:
- Around line 104-106: The example currently hardcodes WorkerID: "w" in the
worker.Lease construction; change it to use the actual worker ID from the
runtime or handler context instead. Either retrieve the worker ID from the
worker instance (e.g., use the worker object's ID accessor when creating lease
:= worker.Lease{RunID: c.ID, WorkerID: worker.ID(), Attempt: c.Attempt}) or
update the handler factory signature (handleRun) to accept a workerID parameter
(e.g., func handleRun(..., workerID string) worker.HandlerFunc) and pass that
workerID into the Lease construction; ensure references to RunID, WorkerID,
Attempt and the worker.HandlerFunc/handleRun symbols are updated accordingly.
In `@postgres/queue.go`:
- Around line 15-27: Create shared sentinel errors (e.g., var ErrEmptyRunID =
errors.New("empty run id") and var ErrDuplicateRunID = errors.New("duplicate run
id")) in the queue package used by both Postgres and memstore implementations,
then update Store.Enqueue: return ErrEmptyRunID when run.ID == "" instead of the
opaque fmt error, and when s.pool.Exec returns an error detect Postgres
unique-violation (pgconn.PgError with Code "23505") and return ErrDuplicateRunID
(wrap it if you need context, e.g., fmt.Errorf("%w: %v", ErrDuplicateRunID, err)
or via your WorkflowError wrapper) so callers can use errors.Is to branch;
ensure memstore returns the same sentinels for its empty-ID and duplicate-ID
paths as well so QueueStore behavior is consistent.
In `@sqlite/schema.sql`:
- Around line 64-77: The cleanup path (CleanupEvents) deletes rows by created_at
but there is no index, causing full table scans; add an index on the
workflow_events.created_at column to improve delete scalability by adding a
statement like CREATE INDEX IF NOT EXISTS workflow_events_created_at ON
workflow_events (created_at) to the schema (alongside the existing
workflow_events and workflow_events_run index) so CleanupEvents can use the
index for efficient deletes.
In `@sqlite/triggers.go`:
- Line 42: The tx.Commit() return in sqlite/triggers.go should wrap the error
with the same "sqlite: ..." prefix used elsewhere; replace the bare return
tx.Commit() with error handling that captures err from tx.Commit() and returns
fmt.Errorf("sqlite: commit triggers: %w", err) (or a similar "sqlite: ..."
message consistent with other errors), ensuring you import fmt if needed and
keep the tx variable and commit call unchanged.
In `@worker/events.go`:
- Around line 30-33: The ListEvents(ctx context.Context, runID string, afterSeq
int64) ([]*Event, error) API exposes cursoring via afterSeq but lacks a page
size cap; update the interface to accept a limit parameter (e.g.,
ListEvents(ctx, runID, afterSeq, limit int) or document/enforce a hard max) and
update all implementations to honor and validate that limit (apply a sensible
max cap, return at most limit events, and return a clear error for invalid
limits) so callers cannot receive arbitrarily large batches of *Event.
In `@worker/memstore/memstore.go`:
- Around line 178-195: In Store.Complete, add the same running-state fence used
in the Postgres store by checking that the run is currently running before
accepting a completion: in the Complete method (type Store, function Complete)
after validating the run exists and matching lease.WorkerID/lease.Attempt (row,
s.runs, row.claimedBy, row.attempt), also require row.status ==
worker.StatusRunning (or return worker.ErrLeaseLost) so an unclaimed sentinel
(claimedBy == "") cannot be used to complete a stale lease; keep the rest of the
logic (setting row.status, row.result, row.errorMessage, row.completedAt)
unchanged.
In `@worker/triggers.go`:
- Line 38: The ListPendingTriggers(ctx context.Context, limit int) signature
lacks defined semantics for limit <= 0; update the API and implementations to
either reject non-positive limits (return a clear error like ErrInvalidLimit) or
clamp them to a safe default/max (e.g., defaultLimit or maxLimit) and document
this behavior in the function comment. Modify the ListPendingTriggers interface
doc comment in triggers.go and enforce the chosen policy at the start of each
implementation of ListPendingTriggers (validate the limit parameter and either
return the error or replace limit with the clamp value) so polling loops cannot
trigger backend-specific unbounded reads.
In `@worker/webhooks.go`:
- Line 16: The Status field on WebhookDelivery currently uses raw strings;
introduce a new typed status alias (e.g., type WebhookStatus string) and define
constants (WebhookPending, WebhookDelivered, WebhookFailed) and change the
Status field in the WebhookDelivery struct to use WebhookStatus instead of
string; update any constructors, serializers, DB mappings, and comparisons that
reference WebhookDelivery.Status to use the new WebhookStatus constants (search
for WebhookDelivery and Status to locate usages).
In `@worker/worker_test.go`:
- Line 79: The test currently discards errors from store.Enqueue calls (e.g.,
store.Enqueue(context.Background(), worker.NewRun{ID: "crash"})), so update each
enqueue in the test setup to capture the returned error and fail the test on
error (for example, assign err := store.Enqueue(...); then call t.Fatalf or
require.NoError(t, err) with a clear message). Apply this change to the
occurrences that discard the error (including the instance using
worker.NewRun{ID: "crash"} and the second occurrence noted around the later
enqueue) so setup failures are surfaced instead of hidden.
🪄 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: defaults
Review profile: CHILL
Plan: Pro
Run ID: bf40e27c-832c-463b-bbeb-cb95d353a858
⛔ Files ignored due to path filters (2)
postgres/go.sumis excluded by!**/*.sumsqlite/go.sumis excluded by!**/*.sum
📒 Files selected for processing (43)
README.mdbranch_join_test.godocs/dynamic_steps_overlay_rfc.mddocs/dynamic_steps_overlay_rfc_review.mddocs/postgres.mddocs/worker.mdpostgres/activity_logger.gopostgres/checkpointer.gopostgres/credits.gopostgres/doc.gopostgres/events.gopostgres/go.modpostgres/queue.gopostgres/schema.sqlpostgres/step_progress.gopostgres/store.gopostgres/store_test.gopostgres/triggers.gopostgres/webhooks.gosqlite/activity_logger.gosqlite/checkpointer.gosqlite/credits.gosqlite/doc.gosqlite/events.gosqlite/go.modsqlite/queue.gosqlite/schema.sqlsqlite/step_progress.gosqlite/store.gosqlite/triggers.gosqlite/webhooks.goworker/credits.goworker/doc.goworker/events.goworker/go.modworker/handler.goworker/memstore/memstore.goworker/queue_store.goworker/subsystems.goworker/triggers.goworker/webhooks.goworker/worker.goworker/worker_test.go
Comment on lines
+115
to
+123
| tag, err := s.pool.Exec(ctx, ` | ||
| UPDATE workflow_runs | ||
| SET status = $1, | ||
| result = $2, | ||
| error_message = $3, | ||
| completed_at = CASE WHEN $1 IN ($6, $7) THEN NOW() ELSE completed_at END | ||
| WHERE id = $4 | ||
| AND claimed_by = $5 | ||
| AND attempt = $8 |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Verify that WorkerID is validated as non-empty before it reaches QueueStore methods.
rg -n -C3 '\bWorkerID\b' worker --type=go
rg -n -C3 'ClaimQueued\s*\(' worker --type=go
rg -n -C3 'Validate\s*\(|validate\s*\(' worker --type=goRepository: deepnoodle-ai/workflow
Length of output: 7225
🏁 Script executed:
# First, read the full Complete method in postgres/queue.go to see the WHERE clause
cat -n postgres/queue.go | head -150Repository: deepnoodle-ai/workflow
Length of output: 5386
🏁 Script executed:
# Also examine ReclaimStale to understand the reaper behavior
rg -n -A 15 'func.*ReclaimStale' postgres --type=goRepository: deepnoodle-ai/workflow
Length of output: 911
🏁 Script executed:
# Check the memstore Complete implementation for comparison
rg -n -B 5 -A 15 'func.*\(\s*s\s*\*\s*Store\s*\)\s*Complete' worker/memstore/memstore.goRepository: deepnoodle-ai/workflow
Length of output: 780
Add status = running check to Complete fence to match Heartbeat pattern and prevent collision on empty/matching workerIDs.
The postgres Complete method (lines 121-123) checks only (claimed_by, attempt), unlike Heartbeat which requires status = StatusRunning. This allows a stale claim to bypass the fence if ReclaimStale resets claimed_by to the same value. No upstream validation prevents empty workerID at the QueueStore layer, so the fence must account for potential collisions. Apply the suggested hardening to postgres and mirror it in worker/memstore/memstore.go Complete method.
Hardening for postgres/queue.go
WHERE id = $4
AND claimed_by = $5
- AND attempt = $8
+ AND attempt = $8
+ AND status = $9
`,
string(outcome.Status),
outcome.Result,
outcome.ErrorMessage,
lease.RunID,
lease.WorkerID,
string(worker.StatusCompleted),
string(worker.StatusFailed),
lease.Attempt,
+ string(worker.StatusRunning),
)Also apply the same fence check to worker/memstore/memstore.go Complete method.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@postgres/queue.go` around lines 115 - 123, The Complete fence currently only
checks claimed_by and attempt, allowing stale/empty workerID collisions; update
the WHERE clause in the postgres Complete function that contains the shown
UPDATE (the method named Complete in queue.go) to also require status =
StatusRunning (or the same status parameter used by Heartbeat) so the UPDATE
only succeeds if the run is still running, and mirror this same status check in
the Complete method in worker/memstore/memstore.go so both stores enforce the
same fence semantics.
Comment on lines
+21
to
+22
| return fmt.Errorf("postgres: marshal progress detail: %w", err) | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
Use WorkflowError/sentinel-compatible errors here.
These branches currently return plain wrapped errors, which makes classification less consistent with the repo’s error contract. Please switch to structured WorkflowError + sentinel-compatible wrapping so callers can reliably use errors.Is.
As per coding guidelines, **/*.go: Use error sentinels with errors.Is and structured errors via WorkflowError for error handling.
Also applies to: 53-53
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@postgres/step_progress.go` around lines 21 - 22, Replace the plain fmt.Errorf
wrapping in the "marshal progress detail" error branches with a
sentinel-compatible WorkflowError: define a package sentinel error (e.g.
ErrMarshalProgressDetail) and return a WorkflowError that wraps the underlying
err while exposing that sentinel so callers can use errors.Is; update both
occurrences (the marshal progress detail branch and the similar branch around
line 53) to return the WorkflowError-wrapped error instead of fmt.Errorf.
Comment on lines
+17
to
+18
| return fmt.Errorf("sqlite: marshal progress detail: %w", err) | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
Align SQLite progress errors with WorkflowError conventions.
Please return structured errors (with sentinel compatibility) instead of plain wrapped errors so upstream handling can classify failures consistently.
As per coding guidelines, **/*.go: Use error sentinels with errors.Is and structured errors via WorkflowError for error handling.
Also applies to: 49-49
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@sqlite/step_progress.go` around lines 17 - 18, Replace the plain fmt.Errorf
returns for the marshal/unmarshal cases (the "sqlite: marshal progress detail:
%w" occurrence and the similar error at the other location) with a structured
WorkflowError that uses a sentinel error so callers can use errors.Is; define or
reuse a sentinel like ErrProgressMarshal and return the WorkflowError via the
project's WorkflowError constructor (e.g., NewWorkflowError or the existing
constructor used elsewhere) including the sentinel, a short context message
("marshal progress detail") and the original err as the cause so upstream code
can classify failures with errors.Is and inspect the underlying error.
Nest unstable submodules to signal API is not yet finalized: - worker/ → experimental/worker/ - postgres/ → experimental/store/postgres/ - sqlite/ → experimental/store/sqlite/ Postgres and SQLite grouped under store/ as sibling persistence implementations. All module paths, imports, replace directives, and documentation updated. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
There was a problem hiding this comment.
Actionable comments posted: 19
♻️ Duplicate comments (10)
experimental/store/postgres/checkpointer.go (1)
40-46:⚠️ Potential issue | 🔴 CriticalCheckpoint writes/deletes are still not fully lease-fenced.
SaveCheckpointshould bind toc.lease.RunID(and reject mismatchedcheckpoint.ExecutionID), whileDeleteCheckpointalso needs(claimed_by, attempt)fencing andErrLeaseLostwhen no row matches.Also applies to: 86-89
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@experimental/store/postgres/checkpointer.go` around lines 40 - 46, SaveCheckpoint currently updates by id and uses checkpoint.ExecutionID but must also bind to the lease RunID: validate that checkpoint.ExecutionID equals c.lease.RunID and reject/mismatch early, then include c.lease.RunID in the WHERE clause (or replace id check with run_id = c.lease.RunID) so the write is lease-fenced; similarly, in DeleteCheckpoint modify the Exec query to include claimed_by = c.lease.WorkerID and attempt = c.lease.Attempt in the WHERE clause and return ErrLeaseLost when the Exec reports 0 rows affected (no matching row), ensuring both SaveCheckpoint and DeleteCheckpoint use c.lease.RunID / (claimed_by, attempt) fencing and return ErrLeaseLost on no-match.experimental/store/sqlite/step_progress.go (1)
17-18:⚠️ Potential issue | 🟠 MajorUse sentinel-compatible
WorkflowErrorfor step-progress marshal/upsert failures.These branches still return plain wrapped errors, so callers cannot reliably classify with
errors.Is.As per coding guidelines,
**/*.go: Use error sentinels witherrors.Isand structured errors viaWorkflowErrorfor error handling.Also applies to: 49-49
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@experimental/store/sqlite/step_progress.go` around lines 17 - 18, Replace the plain fmt.Errorf wraps in step_progress.go (e.g., the "sqlite: marshal progress detail: %w" return and the similar branch at line ~49) with a sentinel-compatible WorkflowError: define or use a sentinel like ErrStepProgressMarshal and return a WorkflowError that wraps the underlying err (so callers can use errors.Is(…, ErrStepProgressMarshal)); ensure you use the existing WorkflowError constructor/helper used across the project (or create NewWorkflowError/WrapWorkflowError) and include the original error as the cause and a descriptive message matching the current text.experimental/store/postgres/step_progress.go (1)
21-22:⚠️ Potential issue | 🟠 MajorSwitch these error returns to sentinel-compatible
WorkflowError.These branches still use plain wrapped errors and don’t expose stable sentinels for
errors.Is.As per coding guidelines,
**/*.go: Use error sentinels witherrors.Isand structured errors viaWorkflowErrorfor error handling.Also applies to: 53-53
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@experimental/store/postgres/step_progress.go` around lines 21 - 22, The plain wrapped errors using fmt.Errorf("postgres: marshal progress detail: %w", err) (and the similar return at the other occurrence) must be replaced with a sentinel-compatible WorkflowError: declare a package-level sentinel (e.g. ErrMarshalProgressDetail) and return a WorkflowError that wraps the original err and exposes that sentinel so callers can use errors.Is; for example, replace the fmt.Errorf return sites with a call that constructs/returns a WorkflowError (e.g. NewWorkflowError(ErrMarshalProgressDetail, err, "postgres: marshal progress detail") or &WorkflowError{Sentinel: ErrMarshalProgressDetail, Err: err, Message: "postgres: marshal progress detail"}) so the sentinel is stable for errors.Is checks.experimental/store/postgres/events.go (1)
67-69:⚠️ Potential issue | 🟠 MajorDo not silently drop payload decode failures.
If a stored payload is corrupted, this currently returns an event with missing data instead of surfacing the persistence error.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@experimental/store/postgres/events.go` around lines 67 - 69, The json.Unmarshal call that decodes payload into e.Payload is currently ignoring errors, which masks corrupted stored payloads; capture the error (err := json.Unmarshal(payload, &e.Payload)), and propagate it instead of discarding it — either return the error (or wrap it with context using fmt.Errorf) from the containing function in events.go or add an error return to that function so callers can handle persistence/decoding failures; update any call sites to propagate the error upward.experimental/store/sqlite/events.go (1)
68-69:⚠️ Potential issue | 🟠 MajorDon't drop corrupt event payloads in
ListEvents.If a row contains invalid JSON, this silently returns an event with an empty payload instead of surfacing storage corruption to the caller.
Proposed fix
e.CreatedAt = parseTime(createdAt) if payload.Valid && payload.String != "" { - _ = json.Unmarshal([]byte(payload.String), &e.Payload) + if err := json.Unmarshal([]byte(payload.String), &e.Payload); err != nil { + return nil, fmt.Errorf("sqlite: unmarshal event payload seq=%d run_id=%s: %w", e.Seq, e.RunID, err) + } } out = append(out, &e) }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@experimental/store/sqlite/events.go` around lines 68 - 69, ListEvents currently ignores JSON unmarshal errors for the scanned payload (the payload variable / payload.String) and leaves e.Payload empty, hiding storage corruption; change the code in ListEvents so that when json.Unmarshal([]byte(payload.String), &e.Payload) returns an error you propagate that error (or wrap it with context identifying the event id) back to the caller instead of discarding it, ensuring callers receive a non-nil error for invalid JSON payloads and the event is not silently returned with an empty payload.experimental/worker/worker.go (2)
185-196:⚠️ Potential issue | 🟠 MajorReject
WebhookStorewithout a deliverer.
ConfigdocumentsWebhookDelivereras required whenWebhookStoreis set, butNewaccepts that combination andRunsilently skips the webhook loop for it. That turns a wiring mistake into a no-op instead of a fast startup failure.🛠️ Proposed fix
if cfg.WebhookMaxAttempts <= 0 { cfg.WebhookMaxAttempts = DefaultWebhookMaxAttempts } + if cfg.WebhookStore != nil && cfg.WebhookDeliverer == nil { + return nil, errors.New("worker: Config.WebhookDeliverer is required when Config.WebhookStore is set") + } if cfg.Logger == nil { cfg.Logger = slog.New(slog.NewTextHandler(io.Discard, nil)) }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@experimental/worker/worker.go` around lines 185 - 196, The constructor New should validate that when Config.WebhookStore is non-nil, Config.WebhookDeliverer is also provided; currently New accepts WebhookStore without a deliverer and Run silently no-ops. Update New to check if cfg.WebhookStore != nil && cfg.WebhookDeliverer == nil and return an error (or panic if constructors in this codebase use panics) with a clear message about the missing WebhookDeliverer, referencing the Config fields and the New function so the wiring error fails fast instead of being skipped at Run time.
341-353:⚠️ Potential issue | 🔴 CriticalNormalize canceled runs to
StatusFailedbeforeComplete.If
runCtxis canceled by timeout, shutdown, or lease loss and the handler returns a zero-value or dormant outcome, this persists the wrong state. Convert canceled executions toStatusFailedbefore the detached finalize write. As per coding guidelines, "Classify interrupted executions (context canceled mid-run) as failed, even ifSetFinished()was never called viabuildResult".🛠️ Proposed fix
outcome := w.safeHandle(runCtx, claim) + if err := runCtx.Err(); err != nil && outcome.Status != StatusFailed { + outcome = Outcome{ + Status: StatusFailed, + ErrorMessage: err.Error(), + } + } stopHeartbeat() hbWG.Wait()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@experimental/worker/worker.go` around lines 341 - 353, The handler may return a zero/dormant outcome when runCtx was canceled, so before calling w.store.Complete(finalizeCtx, lease, outcome) normalize canceled executions to StatusFailed: inspect runCtx.Err() (or errors.Is(runCtx.Err(), context.Canceled) / context.DeadlineExceeded) after safeHandle returns and, if canceled and the returned outcome is zero/unfinished, set the outcome's status to StatusFailed (or call the same helper used by buildResult to mark failed) so the detached finalize write via w.store.Complete records a failed run rather than a blank/dormant one; update code around outcome := w.safeHandle(runCtx, claim) and the Complete call to perform this normalization.experimental/store/postgres/triggers.go (2)
81-83:⚠️ Potential issue | 🟠 MajorDon't swallow malformed
child_spec.A bad row currently deserializes to the zero value and then fails later as an opaque enqueue problem. Return a wrapped error here so the corrupt trigger is surfaced immediately.
🛠️ Proposed fix
if len(childSpec) > 0 { - _ = json.Unmarshal(childSpec, &t.ChildSpec) + if err := json.Unmarshal(childSpec, &t.ChildSpec); err != nil { + return nil, fmt.Errorf("postgres: unmarshal trigger child spec for %s: %w", t.ID, err) + } }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@experimental/store/postgres/triggers.go` around lines 81 - 83, The current code swallows JSON unmarshal errors for childSpec (json.Unmarshal(childSpec, &t.ChildSpec)), causing malformed child_spec to become a zero value and surface later; change this to check the error returned by json.Unmarshal and return a wrapped error (including context like the trigger ID/name and the raw error) from the containing function so corrupt triggers fail fast—ensure you reference the same variables (childSpec, t.ChildSpec) and use the function's error return path rather than assigning to _.
99-106:⚠️ Potential issue | 🟠 MajorClaim the trigger with a compare-and-swap.
This update can overwrite an already-claimed row, so two workers can both process the same trigger. Guard it with
AND status = 'pending'and treatRowsAffected() == 0as an “already claimed” sentinel so callers can ignore it witherrors.Is. As per coding guidelines, "Use error sentinels witherrors.Isand structured errors viaWorkflowErrorfor error handling".🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@experimental/store/postgres/triggers.go` around lines 99 - 106, The MarkTriggerProcessing function must perform a compare-and-swap to avoid double-claiming: change the UPDATE to include "AND status = 'pending'", use the Exec result (e.g., res := s.pool.Exec(...)) and check res.RowsAffected(); if RowsAffected() == 0 return a sentinel error (create ErrTriggerAlreadyClaimed) so callers can use errors.Is, and wrap other DB errors as before; ensure the sentinel is also represented via the project's WorkflowError pattern (e.g., return a WorkflowError wrapping ErrTriggerAlreadyClaimed) so structured error handling remains supported.experimental/store/sqlite/queue.go (1)
29-78:⚠️ Potential issue | 🔴 Critical
ClaimQueuedis still racy under SQLite.
BeginTx(..., &sql.TxOptions{})is a deferred transaction, notBEGIN IMMEDIATE, and the claim update still filters only byid. Two workers can read the same queued row before either write commits and both return a claim unless the DSN is forcing immediate tx locks elsewhere.🛠️ Proposed fix
_, err = tx.ExecContext(ctx, ` UPDATE workflow_runs SET status = ?, claimed_by = ?, heartbeat_at = ?, started_at = COALESCE(started_at, ?), attempt = ? - WHERE id = ? - `, string(worker.StatusRunning), workerID, now, now, newAttempt, id) + WHERE id = ? AND status = ? + `, string(worker.StatusRunning), workerID, now, now, newAttempt, id, string(worker.StatusQueued)) if err != nil { return nil, fmt.Errorf("sqlite: claim update: %w", err) } + if n, _ := result.RowsAffected(); n == 0 { + return nil, nil + }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@experimental/store/sqlite/queue.go` around lines 29 - 78, ClaimQueued is racy because BeginTx with default options does not use BEGIN IMMEDIATE and the UPDATE only filters by id; two workers can select the same row before either commits. Fix by starting an immediate lock before reading (execute "BEGIN IMMEDIATE" on the connection/tx in ClaimQueued immediately after obtaining tx) and make the UPDATE more defensive by adding the original status/claimed_by constraints (e.g., WHERE id = ? AND status = ? AND (claimed_by IS NULL OR claimed_by = '')) and check the exec result's RowsAffected to ensure only the winning transaction produced a change; if RowsAffected == 0, rollback and return nil to avoid duplicate claims.
🧹 Nitpick comments (6)
experimental/worker/queue_store.go (1)
136-143: Tighten concurrency-contract wording for interface scope.The contract text references
SaveCheckpoint, but that method is outsideQueueStore. Consider moving that note to checkpointer docs or rewording to avoid implementer confusion.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@experimental/worker/queue_store.go` around lines 136 - 143, The concurrency contract on QueueStore wrongly mentions SaveCheckpoint (which is outside QueueStore) and may confuse implementers; update the text to either remove SaveCheckpoint from the QueueStore contract or change the wording to refer generically to "checkpointing operations" and point implementers to the Checkpointer interface docs, and/or move the specific fencing sentence about SaveCheckpoint into the Checkpointer documentation; ensure references to ClaimQueued, Heartbeat, Complete, ReclaimStale, and DeadLetterStale remain unchanged and that fencing behavior is described only for methods actually declared on QueueStore or for the Checkpointer interface as appropriate.experimental/store/postgres/activity_logger.go (1)
103-105: Drop the impossiblepgx.ErrNoRowsbranch onrows.Err().
rows.Err()is only for iteration failures, so this special-case is misleading and it also returns the raw error withoutGetActivityHistorycontext. Wrap any non-nilrows.Err()directly instead.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@experimental/store/postgres/activity_logger.go` around lines 103 - 105, The rows.Err() check in GetActivityHistory incorrectly special-cases pgx.ErrNoRows; since rows.Err() only reports iteration failures this branch is misleading—remove the "&& err != pgx.ErrNoRows" condition and instead, if rows.Err() is non-nil, return a wrapped error that adds context (e.g., "GetActivityHistory: rows iteration failed") rather than returning the raw error. Ensure the change is applied in activity_logger.go where rows.Err() is inspected so callers receive contextualized errors.experimental/store/sqlite/checkpointer.go (1)
50-53: Useerrors.Isfor the no-row sentinel.This path should use
errors.Is(err, sql.ErrNoRows)instead oferr == sql.ErrNoRowsso it stays correct if the error gets wrapped later. As per coding guidelines "Use error sentinels witherrors.Isand structured errors viaWorkflowErrorfor error handling".🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@experimental/store/sqlite/checkpointer.go` around lines 50 - 53, The code currently compares the DB error with sql.ErrNoRows using ==; change that to use errors.Is(err, sql.ErrNoRows) so wrapped errors are detected correctly. In the function in checkpointer.go where you return workflow.ErrNoCheckpoint (the branch checking sql.ErrNoRows), replace the equality check with errors.Is and ensure the errors package is imported. Keep the existing return of workflow.ErrNoCheckpoint when errors.Is(err, sql.ErrNoRows) is true.experimental/store/postgres/events.go (1)
72-74: Simplify therows.Err()handling.
rows.Err()reports iteration failures; it should not be special-cased againstpgx.ErrNoRows. Returning the raw error here also loses thelist eventscontext.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@experimental/store/postgres/events.go` around lines 72 - 74, The rows.Err() check should not special-case pgx.ErrNoRows and must preserve context; replace the current if block that compares rows.Err() to pgx.ErrNoRows with a single check that if err := rows.Err(); err != nil { return nil, fmt.Errorf("list events: %w", err) } (or use errors.Wrap) so iteration errors are returned with the "list events" context and no special-casing of pgx.ErrNoRows.experimental/store/postgres/store_test.go (1)
104-110: Useerrors.Isfor sentinel assertions.These tests currently depend on exact error identity. If the store starts wrapping
worker.ErrLeaseLostorworkflow.ErrNoCheckpointwith context, the behavior stays correct but the tests will fail. As per coding guidelines "Use error sentinels witherrors.Isand structured errors viaWorkflowErrorfor error handling".Proposed fix
import ( "context" + "errors" "os" "testing" "time" @@ - if err := store.Heartbeat(ctx, wrong); err != worker.ErrLeaseLost { + if err := store.Heartbeat(ctx, wrong); !errors.Is(err, worker.ErrLeaseLost) { t.Fatalf("expected ErrLeaseLost, got %v", err) } @@ - if err := store.Heartbeat(ctx, badAttempt); err != worker.ErrLeaseLost { + if err := store.Heartbeat(ctx, badAttempt); !errors.Is(err, worker.ErrLeaseLost) { t.Fatalf("expected ErrLeaseLost, got %v", err) } @@ - if _, err := cp.LoadCheckpoint(ctx, claim.ID); err != workflow.ErrNoCheckpoint { + if _, err := cp.LoadCheckpoint(ctx, claim.ID); !errors.Is(err, workflow.ErrNoCheckpoint) { t.Fatalf("expected ErrNoCheckpoint, got %v", err) } @@ - if err := bogus.SaveCheckpoint(ctx, original); err != worker.ErrLeaseLost { + if err := bogus.SaveCheckpoint(ctx, original); !errors.Is(err, worker.ErrLeaseLost) { t.Fatalf("expected ErrLeaseLost, got %v", err) } @@ - if _, err := cp.LoadCheckpoint(ctx, claim.ID); err != workflow.ErrNoCheckpoint { + if _, err := cp.LoadCheckpoint(ctx, claim.ID); !errors.Is(err, workflow.ErrNoCheckpoint) { t.Fatalf("expected ErrNoCheckpoint after delete, got %v", err) }Also applies to: 170-171, 198-206
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@experimental/store/postgres/store_test.go` around lines 104 - 110, The tests compare errors by identity which will break if errors are wrapped; update the assertions in store_test.go to use errors.Is instead of direct equality checks for sentinel errors (e.g. replace comparisons of the result of store.Heartbeat(ctx, wrong) and store.Heartbeat(ctx, badAttempt) against worker.ErrLeaseLost, and similar checks against workflow.ErrNoCheckpoint) so they correctly detect wrapped errors; import the standard errors package and change the t.Fatalf checks to call errors.Is(err, worker.ErrLeaseLost) or errors.Is(err, workflow.ErrNoCheckpoint) accordingly, referencing the existing variables wrong, badAttempt, claim.ID and the store.Heartbeat calls.experimental/store/sqlite/schema.sql (1)
64-76: Add an index for event-retention scans.
CleanupEventsprunes bycreated_at, butworkflow_eventsis only indexed by(run_id, seq). Once this table grows, retention will degrade into a full scan and compete with append/list traffic.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@experimental/store/sqlite/schema.sql` around lines 64 - 76, workflow_events retention queries (CleanupEvents) prune by created_at but the table only has index ON workflow_events (run_id, seq), causing full scans; add an index on created_at (e.g., CREATE INDEX IF NOT EXISTS workflow_events_created_at ON workflow_events (created_at); or include seq as second column if helpful) to accelerate retention scans and avoid competing with append/list traffic—update the schema definition around the workflow_events table to create this new index.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@docs/postgres.md`:
- Around line 73-121: The Schema section is outdated: update the `Migrate`
documentation to list all tables the migration now creates (not just the
original three). Add `workflow_events`, `workflow_triggers`,
`workflow_credit_ledger`, and `workflow_webhooks` to the table list, describe
each table's purpose and key columns/indexes (e.g., event storage and indexing,
trigger definitions, credit ledger entries, webhook deliveries), update the
opening sentence that currently says “creates three tables” to reflect the
current set, and adjust the Indexes/notes to include any new indexes or
retention/permission implications so operators see the full migration footprint.
- Around line 24-26: The fenced code blocks containing the commands like "go get
github.com/deepnoodle-ai/workflow/experimental/store/postgres" should include a
language tag to satisfy MD040; add "sh" after the opening ``` for those blocks
(the block with that go get snippet and the other occurrence around lines
244-247) so the fences become ```sh and the linter stops flagging them.
In `@docs/worker.md`:
- Around line 31-33: The markdown fences that currently contain the go command
and the nearby plain-text example need language tags to satisfy MD040: add "sh"
to the fenced block containing the line "go get
github.com/deepnoodle-ai/workflow/experimental/worker" and add "text" (or
another appropriate language) to the other fenced block around the
human-readable/example content (the block at the other example around lines
43-47) so both fenced code blocks include a language identifier.
In `@experimental/store/postgres/credits.go`:
- Around line 11-33: Add input validation to Store.Debit and Store.Refund to
reject non-positive amounts: check the amount parameter and return an error if
amount <= 0 (e.g., fmt.Errorf("invalid amount: must be > 0")). This ensures
Debit and Refund only accept positive credit amounts and prevents callers from
passing signed values that flip ledger semantics; update both Debit and Refund
functions at their entry points before calling s.pool.Exec.
In `@experimental/store/postgres/events.go`:
- Around line 15-24: AppendEvent currently dereferences the event parameter and
will panic if called with nil; add an explicit nil-check at the start of func (s
*Store) AppendEvent(ctx context.Context, event *worker.Event) and return a
descriptive error (e.g., fmt.Errorf("postgres: AppendEvent: nil event")) instead
of allowing a panic, ensuring the code that follows (the payload marshalling and
s.pool.QueryRow usage) only runs when event is non-nil.
In `@experimental/store/postgres/queue.go`:
- Around line 146-150: The UPDATE in ClaimQueued currently clears started_at
when reclaiming runs; change the UPDATE so it does not null out started_at
(i.e., remove started_at = NULL from the SET or set started_at = started_at) so
existing started_at is preserved on reclaim. Locate the SQL UPDATE in the
ClaimQueued code path (the block that sets status, claimed_by, heartbeat_at,
started_at) and modify it to stop overwriting started_at to NULL while keeping
the existing COALESCE(started_at, ...) behavior on initial claims.
In `@experimental/store/postgres/store_test.go`:
- Around line 20-23: Update the example test command in the file header to point
at the new package path: replace the old "go test ./postgres/..." reference with
"go test ./experimental/store/postgres/..." (keeping the WORKFLOW_PG_DSN env var
example intact) so readers run tests against the moved package; locate the
comment block at the top of experimental/store/postgres/store_test.go and adjust
the command accordingly.
In `@experimental/store/postgres/webhooks.go`:
- Around line 14-28: EnqueueWebhook currently generates a new ID per retry and
never persists it back, causing duplicates; fix by generating a stable ID when
delivery.ID is empty, assign it back to delivery.ID (so caller sees the chosen
id), and make the DB insert idempotent by using the workflow_webhooks
primary/unique key on id with an INSERT ... ON CONFLICT (id) DO NOTHING (or DO
UPDATE as appropriate) in the s.pool.Exec call; locate and update the
EnqueueWebhook function, the generateID usage and the s.pool.Exec SQL to
implement these changes.
In `@experimental/store/sqlite/checkpointer.go`:
- Around line 71-77: DeleteCheckpoint in leasedCheckpointer currently updates
checkpoint by executionID only, allowing stale workers to delete a newer
worker's checkpoint; modify DeleteCheckpoint to include the same lease fence
used by SaveCheckpoint by adding claimed_by and attempt to the WHERE clause when
calling store.db.ExecContext (use the checkpointer's claim fields), inspect the
Exec result's RowsAffected and if zero return worker.ErrLeaseLost, otherwise
return nil, and keep the existing wrapped fmt.Errorf on database errors.
In `@experimental/store/sqlite/credits.go`:
- Around line 12-34: Both Debit and Refund should reject zero or negative
amounts at the store boundary: in the methods Store.Debit and Store.Refund
validate that amount > 0 at the start and return a clear error (e.g.
fmt.Errorf("invalid amount: must be > 0")) if not. This prevents Debit from
creating refund-like rows or Refund from creating positive charges; keep the
existing INSERT logic and unique constraints (workflow_credit_ledger, run_id,
reason, generateID("crd_")) unchanged.
In `@experimental/store/sqlite/queue.go`:
- Around line 148-150: The UPDATE in experimental/store/sqlite/queue.go
currently clears started_at when reclaiming runs, which erases the original
first-claim timestamp and breaks COALESCE(started_at, ?) used in ClaimQueued;
modify the SQL so it does not set started_at = NULL (remove that assignment from
the query variable) and leave started_at untouched when reclaiming (only reset
status, claimed_by, heartbeat_at, and attempt-related fields), ensuring
ClaimQueued's COALESCE(started_at, ?) still returns the original started_at when
present.
In `@experimental/store/sqlite/triggers.go`:
- Around line 76-78: The code currently calls json.Unmarshal([]byte(childSpec),
&t.ChildSpec) and ignores the error, which swallows corrupt child_spec JSON;
change this to unmarshal into a temporary variable (e.g., var cs
worker.TriggerChildSpec or similar), check the error returned by json.Unmarshal,
and if non-nil return or propagate the error (or log and fail fast) instead of
assigning an empty t.ChildSpec; reference the existing childSpec variable and
t.ChildSpec so the fix is applied where the current if childSpec != "" block
calls json.Unmarshal.
- Around line 85-88: The MarkTriggerProcessing function unconditionally updates
workflow_triggers to processing which allows two pollers to claim the same
trigger; change the SQL in MarkTriggerProcessing to include "AND status = ?"
(use worker.TriggerPending) so the UPDATE only succeeds if the trigger was
pending, then inspect the result's RowsAffected and treat 0 as a lost claim by
returning a sentinel error (e.g. ErrLostClaim) wrapped in a WorkflowError;
ensure you call ExecContext, check result.RowsAffected(), and use errors.Is to
allow callers to detect the lost-claim sentinel.
In `@experimental/store/sqlite/webhooks.go`:
- Around line 18-23: The INSERT is persisting an empty string for created_at
because formatTime(delivery.CreatedAt) returns "" for zero time; change the
logic in the s.db.ExecContext call that builds VALUES so that when
delivery.CreatedAt.IsZero() you pass nil (or a sql.NullTime with Valid=false) or
omit the created_at column so the DB default is used instead of the empty
string; update the call site where formatTime is used (the ExecContext
invocation in this file) to compute a createdAtArg variable based on
delivery.CreatedAt and pass that variable into ExecContext.
In `@experimental/worker/memstore/memstore.go`:
- Around line 185-194: The Complete handler currently only checks lease
ownership and attempt but allows any state rewrite; update it to require the
current row.status equals worker.StatusRunning and reject outcomes with invalid
statuses. Concretely, in the Complete path (where row.claimedBy, lease.WorkerID,
row.attempt, lease.Attempt are checked) first verify row.status ==
worker.StatusRunning and return an appropriate error if not, then validate
outcome.Status is one of worker.StatusCompleted or worker.StatusFailed (reject
other values, including StatusRunning/other non-terminal states). Only after
those checks should you set row.status, copy outcome.Result into row.result, set
row.errorMessage, and if outcome.Status is worker.StatusCompleted or
worker.StatusFailed set row.completedAt = s.now().
In `@experimental/worker/triggers.go`:
- Around line 37-42: The current ListPendingTriggers + MarkTriggerProcessing
race allows duplicate/stuck processing; replace or augment this with an atomic
claim API: add a new method (e.g., ClaimPendingTriggers(ctx context.Context,
workerID string, limit int, leaseDuration time.Duration) ([]Trigger, error))
that atomically selects pending triggers and marks them as processing with
owner/claimed_at/lease_expires metadata, and modify or deprecate
MarkTriggerProcessing(ctx, id string) to require ownerID and leaseDuration (or
implement ReclaimStaleTriggers(ctx context.Context, now time.Time, limit int) to
move expired leases back to pending); update InsertTriggers,
MarkTriggerCompleted, IncrementTriggerAttempts, and MarkTriggerFailed to respect
ownerID/lease semantics so only the claim holder can complete/fail, and ensure
the storage implementation uses a single transactional/conditional update to
prevent races.
In `@experimental/worker/webhooks.go`:
- Around line 23-31: The current WebhookStore contract is racy because
ListPendingWebhooks and subsequent delivery updates allow multiple workers to
deliver the same webhook; add an atomic claim/lease operation to the interface
(for example a ClaimPendingWebhooks(ctx context.Context, limit int)
([]*WebhookDelivery, error) that atomically transitions rows to a "processing"
state, or a ClaimWebhook(ctx context.Context, id string, owner string,
leaseUntil time.Time) error plus a ReclaimExpiredClaims/ReleaseClaim method) and
update callers to use this claim before calling Deliver; reference the existing
WebhookStore interface and methods ListPendingWebhooks, MarkWebhookDelivered,
IncrementWebhookAttempts, and MarkWebhookFailed when implementing and testing
the new claim/lease semantics so ownership is fenced before outbound HTTP calls.
In `@experimental/worker/worker_test.go`:
- Around line 68-70: The test currently drains the done channel with "<-done"
and ignores the returned error, which can hide non-cancellation failures;
instead, receive the error into a variable (e.g., err := <-done) after calling
cancel() and assert that the error indicates cancellation (for example,
require.True(t, errors.Is(err, context.Canceled)) or require.Equal(t,
context.Canceled) depending on test helpers). Update both places (the current
block and the similar block around lines 107-109) to use the returned error from
done and assert it matches context.Canceled, following the pattern used in
TestWorker_ReaperReclaimsStale.
---
Duplicate comments:
In `@experimental/store/postgres/checkpointer.go`:
- Around line 40-46: SaveCheckpoint currently updates by id and uses
checkpoint.ExecutionID but must also bind to the lease RunID: validate that
checkpoint.ExecutionID equals c.lease.RunID and reject/mismatch early, then
include c.lease.RunID in the WHERE clause (or replace id check with run_id =
c.lease.RunID) so the write is lease-fenced; similarly, in DeleteCheckpoint
modify the Exec query to include claimed_by = c.lease.WorkerID and attempt =
c.lease.Attempt in the WHERE clause and return ErrLeaseLost when the Exec
reports 0 rows affected (no matching row), ensuring both SaveCheckpoint and
DeleteCheckpoint use c.lease.RunID / (claimed_by, attempt) fencing and return
ErrLeaseLost on no-match.
In `@experimental/store/postgres/events.go`:
- Around line 67-69: The json.Unmarshal call that decodes payload into e.Payload
is currently ignoring errors, which masks corrupted stored payloads; capture the
error (err := json.Unmarshal(payload, &e.Payload)), and propagate it instead of
discarding it — either return the error (or wrap it with context using
fmt.Errorf) from the containing function in events.go or add an error return to
that function so callers can handle persistence/decoding failures; update any
call sites to propagate the error upward.
In `@experimental/store/postgres/step_progress.go`:
- Around line 21-22: The plain wrapped errors using fmt.Errorf("postgres:
marshal progress detail: %w", err) (and the similar return at the other
occurrence) must be replaced with a sentinel-compatible WorkflowError: declare a
package-level sentinel (e.g. ErrMarshalProgressDetail) and return a
WorkflowError that wraps the original err and exposes that sentinel so callers
can use errors.Is; for example, replace the fmt.Errorf return sites with a call
that constructs/returns a WorkflowError (e.g.
NewWorkflowError(ErrMarshalProgressDetail, err, "postgres: marshal progress
detail") or &WorkflowError{Sentinel: ErrMarshalProgressDetail, Err: err,
Message: "postgres: marshal progress detail"}) so the sentinel is stable for
errors.Is checks.
In `@experimental/store/postgres/triggers.go`:
- Around line 81-83: The current code swallows JSON unmarshal errors for
childSpec (json.Unmarshal(childSpec, &t.ChildSpec)), causing malformed
child_spec to become a zero value and surface later; change this to check the
error returned by json.Unmarshal and return a wrapped error (including context
like the trigger ID/name and the raw error) from the containing function so
corrupt triggers fail fast—ensure you reference the same variables (childSpec,
t.ChildSpec) and use the function's error return path rather than assigning to
_.
- Around line 99-106: The MarkTriggerProcessing function must perform a
compare-and-swap to avoid double-claiming: change the UPDATE to include "AND
status = 'pending'", use the Exec result (e.g., res := s.pool.Exec(...)) and
check res.RowsAffected(); if RowsAffected() == 0 return a sentinel error (create
ErrTriggerAlreadyClaimed) so callers can use errors.Is, and wrap other DB errors
as before; ensure the sentinel is also represented via the project's
WorkflowError pattern (e.g., return a WorkflowError wrapping
ErrTriggerAlreadyClaimed) so structured error handling remains supported.
In `@experimental/store/sqlite/events.go`:
- Around line 68-69: ListEvents currently ignores JSON unmarshal errors for the
scanned payload (the payload variable / payload.String) and leaves e.Payload
empty, hiding storage corruption; change the code in ListEvents so that when
json.Unmarshal([]byte(payload.String), &e.Payload) returns an error you
propagate that error (or wrap it with context identifying the event id) back to
the caller instead of discarding it, ensuring callers receive a non-nil error
for invalid JSON payloads and the event is not silently returned with an empty
payload.
In `@experimental/store/sqlite/queue.go`:
- Around line 29-78: ClaimQueued is racy because BeginTx with default options
does not use BEGIN IMMEDIATE and the UPDATE only filters by id; two workers can
select the same row before either commits. Fix by starting an immediate lock
before reading (execute "BEGIN IMMEDIATE" on the connection/tx in ClaimQueued
immediately after obtaining tx) and make the UPDATE more defensive by adding the
original status/claimed_by constraints (e.g., WHERE id = ? AND status = ? AND
(claimed_by IS NULL OR claimed_by = '')) and check the exec result's
RowsAffected to ensure only the winning transaction produced a change; if
RowsAffected == 0, rollback and return nil to avoid duplicate claims.
In `@experimental/store/sqlite/step_progress.go`:
- Around line 17-18: Replace the plain fmt.Errorf wraps in step_progress.go
(e.g., the "sqlite: marshal progress detail: %w" return and the similar branch
at line ~49) with a sentinel-compatible WorkflowError: define or use a sentinel
like ErrStepProgressMarshal and return a WorkflowError that wraps the underlying
err (so callers can use errors.Is(…, ErrStepProgressMarshal)); ensure you use
the existing WorkflowError constructor/helper used across the project (or create
NewWorkflowError/WrapWorkflowError) and include the original error as the cause
and a descriptive message matching the current text.
In `@experimental/worker/worker.go`:
- Around line 185-196: The constructor New should validate that when
Config.WebhookStore is non-nil, Config.WebhookDeliverer is also provided;
currently New accepts WebhookStore without a deliverer and Run silently no-ops.
Update New to check if cfg.WebhookStore != nil && cfg.WebhookDeliverer == nil
and return an error (or panic if constructors in this codebase use panics) with
a clear message about the missing WebhookDeliverer, referencing the Config
fields and the New function so the wiring error fails fast instead of being
skipped at Run time.
- Around line 341-353: The handler may return a zero/dormant outcome when runCtx
was canceled, so before calling w.store.Complete(finalizeCtx, lease, outcome)
normalize canceled executions to StatusFailed: inspect runCtx.Err() (or
errors.Is(runCtx.Err(), context.Canceled) / context.DeadlineExceeded) after
safeHandle returns and, if canceled and the returned outcome is zero/unfinished,
set the outcome's status to StatusFailed (or call the same helper used by
buildResult to mark failed) so the detached finalize write via w.store.Complete
records a failed run rather than a blank/dormant one; update code around outcome
:= w.safeHandle(runCtx, claim) and the Complete call to perform this
normalization.
---
Nitpick comments:
In `@experimental/store/postgres/activity_logger.go`:
- Around line 103-105: The rows.Err() check in GetActivityHistory incorrectly
special-cases pgx.ErrNoRows; since rows.Err() only reports iteration failures
this branch is misleading—remove the "&& err != pgx.ErrNoRows" condition and
instead, if rows.Err() is non-nil, return a wrapped error that adds context
(e.g., "GetActivityHistory: rows iteration failed") rather than returning the
raw error. Ensure the change is applied in activity_logger.go where rows.Err()
is inspected so callers receive contextualized errors.
In `@experimental/store/postgres/events.go`:
- Around line 72-74: The rows.Err() check should not special-case pgx.ErrNoRows
and must preserve context; replace the current if block that compares rows.Err()
to pgx.ErrNoRows with a single check that if err := rows.Err(); err != nil {
return nil, fmt.Errorf("list events: %w", err) } (or use errors.Wrap) so
iteration errors are returned with the "list events" context and no
special-casing of pgx.ErrNoRows.
In `@experimental/store/postgres/store_test.go`:
- Around line 104-110: The tests compare errors by identity which will break if
errors are wrapped; update the assertions in store_test.go to use errors.Is
instead of direct equality checks for sentinel errors (e.g. replace comparisons
of the result of store.Heartbeat(ctx, wrong) and store.Heartbeat(ctx,
badAttempt) against worker.ErrLeaseLost, and similar checks against
workflow.ErrNoCheckpoint) so they correctly detect wrapped errors; import the
standard errors package and change the t.Fatalf checks to call errors.Is(err,
worker.ErrLeaseLost) or errors.Is(err, workflow.ErrNoCheckpoint) accordingly,
referencing the existing variables wrong, badAttempt, claim.ID and the
store.Heartbeat calls.
In `@experimental/store/sqlite/checkpointer.go`:
- Around line 50-53: The code currently compares the DB error with sql.ErrNoRows
using ==; change that to use errors.Is(err, sql.ErrNoRows) so wrapped errors are
detected correctly. In the function in checkpointer.go where you return
workflow.ErrNoCheckpoint (the branch checking sql.ErrNoRows), replace the
equality check with errors.Is and ensure the errors package is imported. Keep
the existing return of workflow.ErrNoCheckpoint when errors.Is(err,
sql.ErrNoRows) is true.
In `@experimental/store/sqlite/schema.sql`:
- Around line 64-76: workflow_events retention queries (CleanupEvents) prune by
created_at but the table only has index ON workflow_events (run_id, seq),
causing full scans; add an index on created_at (e.g., CREATE INDEX IF NOT EXISTS
workflow_events_created_at ON workflow_events (created_at); or include seq as
second column if helpful) to accelerate retention scans and avoid competing with
append/list traffic—update the schema definition around the workflow_events
table to create this new index.
In `@experimental/worker/queue_store.go`:
- Around line 136-143: The concurrency contract on QueueStore wrongly mentions
SaveCheckpoint (which is outside QueueStore) and may confuse implementers;
update the text to either remove SaveCheckpoint from the QueueStore contract or
change the wording to refer generically to "checkpointing operations" and point
implementers to the Checkpointer interface docs, and/or move the specific
fencing sentence about SaveCheckpoint into the Checkpointer documentation;
ensure references to ClaimQueued, Heartbeat, Complete, ReclaimStale, and
DeadLetterStale remain unchanged and that fencing behavior is described only for
methods actually declared on QueueStore or for the Checkpointer interface as
appropriate.
🪄 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: defaults
Review profile: CHILL
Plan: Pro
Run ID: 22a21d68-87ea-4d86-837e-16dd2d952567
⛔ Files ignored due to path filters (2)
experimental/store/postgres/go.sumis excluded by!**/*.sumexperimental/store/sqlite/go.sumis excluded by!**/*.sum
📒 Files selected for processing (40)
README.mddocs/postgres.mddocs/worker.mdexperimental/store/postgres/activity_logger.goexperimental/store/postgres/checkpointer.goexperimental/store/postgres/credits.goexperimental/store/postgres/doc.goexperimental/store/postgres/events.goexperimental/store/postgres/go.modexperimental/store/postgres/queue.goexperimental/store/postgres/schema.sqlexperimental/store/postgres/step_progress.goexperimental/store/postgres/store.goexperimental/store/postgres/store_test.goexperimental/store/postgres/triggers.goexperimental/store/postgres/webhooks.goexperimental/store/sqlite/activity_logger.goexperimental/store/sqlite/checkpointer.goexperimental/store/sqlite/credits.goexperimental/store/sqlite/doc.goexperimental/store/sqlite/events.goexperimental/store/sqlite/go.modexperimental/store/sqlite/queue.goexperimental/store/sqlite/schema.sqlexperimental/store/sqlite/step_progress.goexperimental/store/sqlite/store.goexperimental/store/sqlite/triggers.goexperimental/store/sqlite/webhooks.goexperimental/worker/credits.goexperimental/worker/doc.goexperimental/worker/events.goexperimental/worker/go.modexperimental/worker/handler.goexperimental/worker/memstore/memstore.goexperimental/worker/queue_store.goexperimental/worker/subsystems.goexperimental/worker/triggers.goexperimental/worker/webhooks.goexperimental/worker/worker.goexperimental/worker/worker_test.go
✅ Files skipped from review due to trivial changes (8)
- README.md
- experimental/worker/go.mod
- experimental/store/postgres/go.mod
- experimental/store/postgres/doc.go
- experimental/store/sqlite/doc.go
- experimental/worker/doc.go
- experimental/store/sqlite/go.mod
- experimental/worker/subsystems.go
Comment on lines
+24
to
+26
| ``` | ||
| go get github.com/deepnoodle-ai/workflow/experimental/store/postgres | ||
| ``` |
There was a problem hiding this comment.
Add languages to these fenced code blocks.
Both fences currently trip MD040. sh would be enough here.
Also applies to: 244-247
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@docs/postgres.md` around lines 24 - 26, The fenced code blocks containing the
commands like "go get
github.com/deepnoodle-ai/workflow/experimental/store/postgres" should include a
language tag to satisfy MD040; add "sh" after the opening ``` for those blocks
(the block with that go get snippet and the other occurrence around lines
244-247) so the fences become ```sh and the linter stops flagging them.
Comment on lines
+73
to
+121
| ## Schema | ||
|
|
||
| `Migrate` creates three tables. | ||
|
|
||
| ### `workflow_runs` | ||
|
|
||
| The durable queue and state table. One row per run, including the | ||
| checkpoint blob. | ||
|
|
||
| | Column | Type | Notes | | ||
| | --------------- | ------------- | ---------------------------------------------------------------- | | ||
| | `id` | `TEXT` PK | The stable run identifier, also used as the workflow `ExecutionID`. | | ||
| | `spec` | `BYTEA` | Opaque payload from `Enqueue`. The worker never inspects it. | | ||
| | `status` | `TEXT` | One of the `worker.Status` values. | | ||
| | `attempt` | `INTEGER` | 0 for queued, increments on each claim. | | ||
| | `claimed_by` | `TEXT` | `WorkerID` of the current leaseholder, or `''`. | | ||
| | `heartbeat_at` | `TIMESTAMPTZ` | Refreshed by the worker's heartbeat goroutine. | | ||
| | `checkpoint` | `BYTEA` | JSON-encoded `workflow.Checkpoint`. | | ||
| | `result` | `BYTEA` | Opaque terminal/dormant payload from `Outcome.Result`. | | ||
| | `error_message` | `TEXT` | Failure reason for `StatusFailed`. | | ||
| | `created_at` | `TIMESTAMPTZ` | `NOW()` at enqueue time. | | ||
| | `started_at` | `TIMESTAMPTZ` | First claim timestamp. | | ||
| | `completed_at` | `TIMESTAMPTZ` | Set when the run reaches a terminal status. | | ||
|
|
||
| Indexes: | ||
|
|
||
| - `(status, created_at)` — claim loop ordering. | ||
| - `(status, heartbeat_at)` — reaper scans. | ||
|
|
||
| ### `workflow_step_progress` | ||
|
|
||
| One row per `(execution_id, step_name, branch_id)`; the latest | ||
| status update wins via `ON CONFLICT ... DO UPDATE`. A step that runs | ||
| on two branches produces two rows. | ||
|
|
||
| Use it to power a UI that watches workflow progress — the row stores | ||
| `status`, `activity`, `attempt`, `started_at`, `finished_at`, | ||
| `error`, and a JSONB `detail` blob for whatever extra context the | ||
| engine emits. | ||
|
|
||
| ### `workflow_activity_log` | ||
|
|
||
| Append-only log of every activity invocation with its parameters, | ||
| result, error, start time, and duration. Keyed by a stable | ||
| `entry.ID`. Good for audit trails, replay analysis, and | ||
| post-mortem debugging. | ||
|
|
||
| Indexed by `(execution_id, start_time)` so pulling the full history | ||
| for one run is a single range scan. |
There was a problem hiding this comment.
Schema section is stale.
Migrate no longer creates only three tables. The embedded schema now also adds workflow_events, workflow_triggers, workflow_credit_ledger, and workflow_webhooks, so this section under-documents the migration footprint and will surprise operators managing permissions, backups, and pruning.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@docs/postgres.md` around lines 73 - 121, The Schema section is outdated:
update the `Migrate` documentation to list all tables the migration now creates
(not just the original three). Add `workflow_events`, `workflow_triggers`,
`workflow_credit_ledger`, and `workflow_webhooks` to the table list, describe
each table's purpose and key columns/indexes (e.g., event storage and indexing,
trigger definitions, credit ledger entries, webhook deliveries), update the
opening sentence that currently says “creates three tables” to reflect the
current set, and adjust the Indexes/notes to include any new indexes or
retention/permission implications so operators see the full migration footprint.
Comment on lines
+31
to
+33
| ``` | ||
| go get github.com/deepnoodle-ai/workflow/experimental/worker | ||
| ``` |
There was a problem hiding this comment.
Add languages to these fenced code blocks.
These two fences currently violate MD040; sh and text would satisfy the linter.
Also applies to: 43-47
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@docs/worker.md` around lines 31 - 33, The markdown fences that currently
contain the go command and the nearby plain-text example need language tags to
satisfy MD040: add "sh" to the fenced block containing the line "go get
github.com/deepnoodle-ai/workflow/experimental/worker" and add "text" (or
another appropriate language) to the other fenced block around the
human-readable/example content (the block at the other example around lines
43-47) so both fenced code blocks include a language identifier.
Comment on lines
+11
to
+33
| func (s *Store) Debit(ctx context.Context, orgID, runID, workflowType string, amount int) error { | ||
| _, err := s.pool.Exec(ctx, ` | ||
| INSERT INTO workflow_credit_ledger (id, org_id, run_id, workflow_type, amount, reason) | ||
| VALUES ($1, $2, $3, $4, $5, 'debit') | ||
| ON CONFLICT (run_id, reason) DO NOTHING | ||
| `, generateID("crd_"), orgID, runID, workflowType, amount) | ||
| if err != nil { | ||
| return fmt.Errorf("postgres: debit credits: %w", err) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // Refund implements worker.CreditStore. Idempotent per (run_id, "refund"). | ||
| func (s *Store) Refund(ctx context.Context, orgID, runID, workflowType string, amount int) error { | ||
| _, err := s.pool.Exec(ctx, ` | ||
| INSERT INTO workflow_credit_ledger (id, org_id, run_id, workflow_type, amount, reason) | ||
| VALUES ($1, $2, $3, $4, $5, 'refund') | ||
| ON CONFLICT (run_id, reason) DO NOTHING | ||
| `, generateID("crd_"), orgID, runID, workflowType, -amount) | ||
| if err != nil { | ||
| return fmt.Errorf("postgres: refund credits: %w", err) | ||
| } | ||
| return nil |
There was a problem hiding this comment.
Reject zero or negative credit amounts here as well.
The current API accepts signed values, so Debit(..., -5) decreases the balance and Refund(..., -5) increases it. That is an easy way to corrupt credit accounting from a caller bug.
Adding an amount > 0 guard in both methods keeps the ledger semantics stable.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@experimental/store/postgres/credits.go` around lines 11 - 33, Add input
validation to Store.Debit and Store.Refund to reject non-positive amounts: check
the amount parameter and return an error if amount <= 0 (e.g.,
fmt.Errorf("invalid amount: must be > 0")). This ensures Debit and Refund only
accept positive credit amounts and prevents callers from passing signed values
that flip ledger semantics; update both Debit and Refund functions at their
entry points before calling s.pool.Exec.
Comment on lines
+15
to
+24
| func (s *Store) AppendEvent(ctx context.Context, event *worker.Event) error { | ||
| var payload []byte | ||
| if event.Payload != nil { | ||
| b, err := json.Marshal(event.Payload) | ||
| if err != nil { | ||
| return fmt.Errorf("postgres: marshal event payload: %w", err) | ||
| } | ||
| payload = b | ||
| } | ||
| err := s.pool.QueryRow(ctx, ` |
There was a problem hiding this comment.
Reject nil events before dereferencing.
AppendEvent(nil) panics at Line 17. This is an exported store method, so it should fail with a normal error instead of crashing the caller.
Proposed fix
func (s *Store) AppendEvent(ctx context.Context, event *worker.Event) error {
+ if event == nil {
+ return fmt.Errorf("postgres: nil event")
+ }
var payload []byte
if event.Payload != nil {📝 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
| func (s *Store) AppendEvent(ctx context.Context, event *worker.Event) error { | |
| var payload []byte | |
| if event.Payload != nil { | |
| b, err := json.Marshal(event.Payload) | |
| if err != nil { | |
| return fmt.Errorf("postgres: marshal event payload: %w", err) | |
| } | |
| payload = b | |
| } | |
| err := s.pool.QueryRow(ctx, ` | |
| func (s *Store) AppendEvent(ctx context.Context, event *worker.Event) error { | |
| if event == nil { | |
| return fmt.Errorf("postgres: nil event") | |
| } | |
| var payload []byte | |
| if event.Payload != nil { | |
| b, err := json.Marshal(event.Payload) | |
| if err != nil { | |
| return fmt.Errorf("postgres: marshal event payload: %w", err) | |
| } | |
| payload = b | |
| } | |
| err := s.pool.QueryRow(ctx, ` |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@experimental/store/postgres/events.go` around lines 15 - 24, AppendEvent
currently dereferences the event parameter and will panic if called with nil;
add an explicit nil-check at the start of func (s *Store) AppendEvent(ctx
context.Context, event *worker.Event) and return a descriptive error (e.g.,
fmt.Errorf("postgres: AppendEvent: nil event")) instead of allowing a panic,
ensuring the code that follows (the payload marshalling and s.pool.QueryRow
usage) only runs when event is non-nil.
Comment on lines
+185
to
+194
| if row.claimedBy != lease.WorkerID || row.attempt != lease.Attempt { | ||
| return worker.ErrLeaseLost | ||
| } | ||
| row.status = outcome.Status | ||
| row.result = append([]byte(nil), outcome.Result...) | ||
| row.errorMessage = outcome.ErrorMessage | ||
| if outcome.Status == worker.StatusCompleted || | ||
| outcome.Status == worker.StatusFailed { | ||
| row.completedAt = s.now() | ||
| } |
There was a problem hiding this comment.
Complete should require StatusRunning and reject invalid outcome statuses.
Right now, lease match alone is enough to rewrite state. That allows post-terminal mutation and invalid statuses to be persisted.
Focused hardening patch
func (s *Store) Complete(_ context.Context, lease worker.Lease, outcome worker.Outcome) error {
s.mu.Lock()
defer s.mu.Unlock()
row, ok := s.runs[lease.RunID]
if !ok {
return worker.ErrLeaseLost
}
- if row.claimedBy != lease.WorkerID || row.attempt != lease.Attempt {
+ if row.status != worker.StatusRunning ||
+ row.claimedBy != lease.WorkerID ||
+ row.attempt != lease.Attempt {
return worker.ErrLeaseLost
}
+ switch outcome.Status {
+ case worker.StatusCompleted, worker.StatusFailed, worker.StatusSuspended, worker.StatusReview:
+ default:
+ return fmt.Errorf("memstore: invalid outcome status %q", outcome.Status)
+ }
row.status = outcome.Status
row.result = append([]byte(nil), outcome.Result...)
row.errorMessage = outcome.ErrorMessage🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@experimental/worker/memstore/memstore.go` around lines 185 - 194, The
Complete handler currently only checks lease ownership and attempt but allows
any state rewrite; update it to require the current row.status equals
worker.StatusRunning and reject outcomes with invalid statuses. Concretely, in
the Complete path (where row.claimedBy, lease.WorkerID, row.attempt,
lease.Attempt are checked) first verify row.status == worker.StatusRunning and
return an appropriate error if not, then validate outcome.Status is one of
worker.StatusCompleted or worker.StatusFailed (reject other values, including
StatusRunning/other non-terminal states). Only after those checks should you set
row.status, copy outcome.Result into row.result, set row.errorMessage, and if
outcome.Status is worker.StatusCompleted or worker.StatusFailed set
row.completedAt = s.now().
Comment on lines
+37
to
+42
| InsertTriggers(ctx context.Context, triggers []Trigger) error | ||
| ListPendingTriggers(ctx context.Context, limit int) ([]Trigger, error) | ||
| MarkTriggerProcessing(ctx context.Context, id string) error | ||
| MarkTriggerCompleted(ctx context.Context, id string, childRunID string) error | ||
| IncrementTriggerAttempts(ctx context.Context, id string, errMsg string) error | ||
| MarkTriggerFailed(ctx context.Context, id string, errMsg string) error |
There was a problem hiding this comment.
Add a fenced/atomic trigger-claim API to prevent duplicate or stuck processing.
The current ListPendingTriggers + MarkTriggerProcessing(id) contract is not enough to guarantee single-consumer processing. Two workers can race, and crash windows can leave triggers stranded in processing without safe reclamation.
Possible contract direction
type TriggerStore interface {
- ListPendingTriggers(ctx context.Context, limit int) ([]Trigger, error)
- MarkTriggerProcessing(ctx context.Context, id string) error
+ ClaimPendingTriggers(ctx context.Context, workerID string, limit int) ([]Trigger, error) // atomic claim
+ RequeueStaleProcessing(ctx context.Context, staleBefore time.Time) (int, error) // recovery path
MarkTriggerCompleted(ctx context.Context, id string, childRunID string) error
IncrementTriggerAttempts(ctx context.Context, id string, errMsg string) error
MarkTriggerFailed(ctx context.Context, id string, errMsg string) error
}🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@experimental/worker/triggers.go` around lines 37 - 42, The current
ListPendingTriggers + MarkTriggerProcessing race allows duplicate/stuck
processing; replace or augment this with an atomic claim API: add a new method
(e.g., ClaimPendingTriggers(ctx context.Context, workerID string, limit int,
leaseDuration time.Duration) ([]Trigger, error)) that atomically selects pending
triggers and marks them as processing with owner/claimed_at/lease_expires
metadata, and modify or deprecate MarkTriggerProcessing(ctx, id string) to
require ownerID and leaseDuration (or implement ReclaimStaleTriggers(ctx
context.Context, now time.Time, limit int) to move expired leases back to
pending); update InsertTriggers, MarkTriggerCompleted, IncrementTriggerAttempts,
and MarkTriggerFailed to respect ownerID/lease semantics so only the claim
holder can complete/fail, and ensure the storage implementation uses a single
transactional/conditional update to prevent races.
Comment on lines
+23
to
+31
| // WebhookStore persists and manages webhook delivery state. Method | ||
| // names are prefixed to avoid collisions when a single store struct | ||
| // implements multiple interfaces. | ||
| type WebhookStore interface { | ||
| EnqueueWebhook(ctx context.Context, delivery *WebhookDelivery) error | ||
| ListPendingWebhooks(ctx context.Context, limit int) ([]*WebhookDelivery, error) | ||
| MarkWebhookDelivered(ctx context.Context, id string) error | ||
| IncrementWebhookAttempts(ctx context.Context, id string, lastError string) error | ||
| MarkWebhookFailed(ctx context.Context, id string, errMsg string) error |
There was a problem hiding this comment.
Webhook polling needs an atomic claim path.
ListPendingWebhooks plus post-delivery updates is not enough for multi-worker setups: two pollers can read the same pending row and both call Deliver(...) before either one marks it delivered or failed.
Please add an exclusive claim/lease or processing transition to the WebhookStore contract so delivery ownership is fenced before the outbound HTTP call.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@experimental/worker/webhooks.go` around lines 23 - 31, The current
WebhookStore contract is racy because ListPendingWebhooks and subsequent
delivery updates allow multiple workers to deliver the same webhook; add an
atomic claim/lease operation to the interface (for example a
ClaimPendingWebhooks(ctx context.Context, limit int) ([]*WebhookDelivery, error)
that atomically transitions rows to a "processing" state, or a ClaimWebhook(ctx
context.Context, id string, owner string, leaseUntil time.Time) error plus a
ReclaimExpiredClaims/ReleaseClaim method) and update callers to use this claim
before calling Deliver; reference the existing WebhookStore interface and
methods ListPendingWebhooks, MarkWebhookDelivered, IncrementWebhookAttempts, and
MarkWebhookFailed when implementing and testing the new claim/lease semantics so
ownership is fenced before outbound HTTP calls.
Comment on lines
+68
to
+70
| cancel() | ||
| <-done | ||
| return |
There was a problem hiding this comment.
Assert the worker exits cleanly after cancellation.
Both tests drain done but ignore the returned error. That means a non-cancellation worker failure can still pass the test unnoticed. The TestWorker_ReaperReclaimsStale pattern is safer here too.
Also applies to: 107-109
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@experimental/worker/worker_test.go` around lines 68 - 70, The test currently
drains the done channel with "<-done" and ignores the returned error, which can
hide non-cancellation failures; instead, receive the error into a variable
(e.g., err := <-done) after calling cancel() and assert that the error indicates
cancellation (for example, require.True(t, errors.Is(err, context.Canceled)) or
require.Equal(t, context.Canceled) depending on test helpers). Update both
places (the current block and the similar block around lines 107-109) to use the
returned error from done and assert it matches context.Canceled, following the
pattern used in TestWorker_ReaperReclaimsStale.
CRITICAL fixes: - Checkpointer: use lease.RunID instead of checkpoint.ExecutionID in SaveCheckpoint WHERE clause; add execution ID mismatch guard - Checkpointer: fence DeleteCheckpoint with (claimed_by, attempt) - SQLite ClaimQueued: replace SELECT+UPDATE tx with atomic UPDATE...RETURNING subquery to eliminate TOCTOU race - Worker: force StatusFailed when handler returns zero outcome on a canceled context (timeout, shutdown, lease loss) MAJOR fixes: - Surface JSON unmarshal errors in ListEvents and ListPendingTriggers instead of silently discarding them (postgres and sqlite) - Validate workerID is non-empty in store-level ClaimQueued - Panic on nil pool/db in store constructors for fail-fast behavior - MarkTriggerProcessing: add AND status='pending' CAS to prevent multiple workers processing the same trigger - Add MarkWebhookProcessing with CAS to WebhookStore interface and implementations; use it in processWebhooks before delivery - Reject WebhookStore without WebhookDeliverer at worker construction - Track debit success; only refund credits when debit actually landed - Reject checkpoint schema versions below 1 in both store checkpointers Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
There was a problem hiding this comment.
Actionable comments posted: 5
♻️ Duplicate comments (4)
experimental/store/sqlite/triggers.go (1)
89-100:⚠️ Potential issue | 🟡 MinorUse a sentinel/
WorkflowErrorfor the lost-claim path.The CAS itself is correct now, but Line 100 still turns the expected “someone else claimed it” case into an opaque formatted error. Returning a sentinel here would let callers handle contention with
errors.Isinstead of logging every race as a generic failure.As per coding guidelines, "Use error sentinels with
errors.Isand structured errors viaWorkflowErrorfor error handling".🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@experimental/store/sqlite/triggers.go` around lines 89 - 100, In Store.MarkTriggerProcessing, replace the opaque fmt.Errorf returned when RowsAffected() == 0 with a sentinel/WorkflowError so callers can detect lost-claim races via errors.Is; specifically, when n == 0 in MarkTriggerProcessing return a predefined sentinel like ErrTriggerAlreadyClaimed (or construct a WorkflowError with a distinct Kind) instead of fmt.Errorf, and ensure the sentinel is exported from the package so callers can use errors.Is to check for the contention case.experimental/store/sqlite/queue.go (1)
137-139:⚠️ Potential issue | 🟡 MinorKeep the original
started_atwhen reclaiming stale runs.Line 138 still clears
started_at, which erases the first-claim timestamp and makes reclaimed runs look newer than they are.Suggested fix
query := `UPDATE workflow_runs - SET status = ?, claimed_by = '', heartbeat_at = NULL, started_at = NULL + SET status = ?, claimed_by = '', heartbeat_at = NULL WHERE status = ? AND heartbeat_at < ? AND attempt < ?`🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@experimental/store/sqlite/queue.go` around lines 137 - 139, The UPDATE SQL in variable `query` currently resets `started_at` to NULL when reclaiming stale runs, which erases the original start time; update the SQL in `experimental/store/sqlite/queue.go` (the `query` string used to reclaim stale runs) to stop setting `started_at = NULL` so the original `started_at` value is preserved, leaving the rest of the clause (status, claimed_by, heartbeat_at) unchanged; ensure any code that relies on detecting "not started" still uses an appropriate check (e.g., NULL vs non-NULL) elsewhere if needed.experimental/store/sqlite/webhooks.go (1)
18-23:⚠️ Potential issue | 🟡 MinorDon't persist an empty
created_at.If
delivery.CreatedAtis zero, Line 23 stores""instead of a real timestamp. That breaks chronological ordering and cleanup semantics for pending deliveries.Suggested fix
func (s *Store) EnqueueWebhook(ctx context.Context, delivery *worker.WebhookDelivery) error { id := delivery.ID if id == "" { id = generateID("whk_") } + createdAt := delivery.CreatedAt + if createdAt.IsZero() { + createdAt = time.Now().UTC() + } _, err := s.db.ExecContext(ctx, ` INSERT INTO workflow_webhooks ( id, run_id, url, event_type, payload, status, created_at ) VALUES (?,?,?,?,?,?,?) `, id, delivery.RunID, delivery.URL, delivery.EventType, - delivery.Payload, "pending", formatTime(delivery.CreatedAt)) + delivery.Payload, "pending", formatTime(createdAt))🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@experimental/store/sqlite/webhooks.go` around lines 18 - 23, The insert currently writes formatTime(delivery.CreatedAt) which becomes an empty string when delivery.CreatedAt is zero; change the code that calls s.db.ExecContext so created_at is not an empty string: detect delivery.CreatedAt.IsZero() and instead pass a proper timestamp (e.g., formatTime(time.Now())) or pass nil/NULL if the DB column is nullable; update the call site around s.db.ExecContext and/or replace formatTime(delivery.CreatedAt) with a helper (e.g., formatTimeOrNow or nilWhenZero) so created_at is always a valid timestamp or NULL rather than "".experimental/store/postgres/triggers.go (1)
103-114:⚠️ Potential issue | 🟡 MinorReturn a sentinel for “already claimed” instead of a formatted error.
Line 113 still hides the expected CAS miss behind a string error. That makes it hard for the worker to use
errors.Isand treat contention differently from real Postgres failures.As per coding guidelines, "Use error sentinels with
errors.Isand structured errors viaWorkflowErrorfor error handling".🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@experimental/store/postgres/triggers.go` around lines 103 - 114, Replace the formatted error returned when the CAS update finds no rows in Store.MarkTriggerProcessing with a sentinel error (e.g., declare var ErrTriggerAlreadyClaimed = errors.New("trigger already claimed")) and return that sentinel instead of fmt.Errorf("postgres: trigger %s already claimed"); update callers to use errors.Is to detect this condition, and if your codebase uses WorkflowError for structured errors, wrap or convert the sentinel into a WorkflowError consistently (but do not embed the trigger id into a plain formatted error).
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@experimental/store/postgres/triggers.go`:
- Around line 16-19: The generateID function currently ignores errors from
crypto/rand.Read which can produce duplicate IDs; change generateID to return
(string, error), check and return the read error (or a wrapped sentinel) instead
of silently using partial/zero bytes, and update all callers (e.g., places that
call generateID in triggers.go) to handle and propagate the error appropriately
so ID generation failures are handled with structured error returns.
In `@experimental/store/sqlite/checkpointer.go`:
- Around line 77-86: In leasedCheckpointer.DeleteCheckpoint, the SQL call and
error message mistakenly use c.lease.RunID instead of the method parameter
executionID; update the ExecContext call in DeleteCheckpoint to pass executionID
as the id placeholder (keep claimed_by and attempt using c.lease.WorkerID and
c.lease.Attempt) and change the fmt.Errorf context to reference the executionID
so the function fails fast when callers supply the wrong ID.
In `@experimental/store/sqlite/webhooks.go`:
- Around line 71-83: Replace the plain formatted error on the CAS miss in
MarkWebhookProcessing with a sentinel error so callers can use errors.Is:
declare a package-level var ErrWebhookAlreadyClaimed = errors.New("webhook
already claimed") and when result.RowsAffected() == 0 return that sentinel
(optionally wrapped with fmt.Errorf("sqlite: %w", ErrWebhookAlreadyClaimed) or
converted into a WorkflowError that embeds the sentinel), ensuring callers like
processWebhooks can detect the contention via errors.Is( err,
ErrWebhookAlreadyClaimed ).
In `@experimental/worker/subsystems.go`:
- Around line 195-199: Wrap each webhook delivery call in its own short timeout
by creating a per-attempt context (e.g., ctxAttempt, cancel :=
context.WithTimeout(ctx, timeout)) immediately before calling
w.cfg.WebhookDeliverer.Deliver(ctxAttempt, d.URL, d.Payload), ensure you call
cancel() after the Deliver returns (do not use defer across loop iterations),
and use the result to decide whether to call
w.cfg.WebhookStore.IncrementWebhookAttempts (you can pass the original ctx or
ctxAttempt). Update the code around w.cfg.WebhookDeliverer.Deliver and the
handling of err so that a hung Deliver cannot block the loop and the per-attempt
timeout error is recorded/propagated to IncrementWebhookAttempts.
---
Duplicate comments:
In `@experimental/store/postgres/triggers.go`:
- Around line 103-114: Replace the formatted error returned when the CAS update
finds no rows in Store.MarkTriggerProcessing with a sentinel error (e.g.,
declare var ErrTriggerAlreadyClaimed = errors.New("trigger already claimed"))
and return that sentinel instead of fmt.Errorf("postgres: trigger %s already
claimed"); update callers to use errors.Is to detect this condition, and if your
codebase uses WorkflowError for structured errors, wrap or convert the sentinel
into a WorkflowError consistently (but do not embed the trigger id into a plain
formatted error).
In `@experimental/store/sqlite/queue.go`:
- Around line 137-139: The UPDATE SQL in variable `query` currently resets
`started_at` to NULL when reclaiming stale runs, which erases the original start
time; update the SQL in `experimental/store/sqlite/queue.go` (the `query` string
used to reclaim stale runs) to stop setting `started_at = NULL` so the original
`started_at` value is preserved, leaving the rest of the clause (status,
claimed_by, heartbeat_at) unchanged; ensure any code that relies on detecting
"not started" still uses an appropriate check (e.g., NULL vs non-NULL) elsewhere
if needed.
In `@experimental/store/sqlite/triggers.go`:
- Around line 89-100: In Store.MarkTriggerProcessing, replace the opaque
fmt.Errorf returned when RowsAffected() == 0 with a sentinel/WorkflowError so
callers can detect lost-claim races via errors.Is; specifically, when n == 0 in
MarkTriggerProcessing return a predefined sentinel like ErrTriggerAlreadyClaimed
(or construct a WorkflowError with a distinct Kind) instead of fmt.Errorf, and
ensure the sentinel is exported from the package so callers can use errors.Is to
check for the contention case.
In `@experimental/store/sqlite/webhooks.go`:
- Around line 18-23: The insert currently writes formatTime(delivery.CreatedAt)
which becomes an empty string when delivery.CreatedAt is zero; change the code
that calls s.db.ExecContext so created_at is not an empty string: detect
delivery.CreatedAt.IsZero() and instead pass a proper timestamp (e.g.,
formatTime(time.Now())) or pass nil/NULL if the DB column is nullable; update
the call site around s.db.ExecContext and/or replace
formatTime(delivery.CreatedAt) with a helper (e.g., formatTimeOrNow or
nilWhenZero) so created_at is always a valid timestamp or NULL rather than "".
🪄 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: defaults
Review profile: CHILL
Plan: Pro
Run ID: b0e20399-3f5d-4367-adf7-cbe120921215
📒 Files selected for processing (15)
experimental/store/postgres/checkpointer.goexperimental/store/postgres/events.goexperimental/store/postgres/queue.goexperimental/store/postgres/store.goexperimental/store/postgres/triggers.goexperimental/store/postgres/webhooks.goexperimental/store/sqlite/checkpointer.goexperimental/store/sqlite/events.goexperimental/store/sqlite/queue.goexperimental/store/sqlite/store.goexperimental/store/sqlite/triggers.goexperimental/store/sqlite/webhooks.goexperimental/worker/subsystems.goexperimental/worker/webhooks.goexperimental/worker/worker.go
✅ Files skipped from review due to trivial changes (2)
- experimental/store/postgres/events.go
- experimental/store/postgres/store.go
🚧 Files skipped from review as they are similar to previous changes (3)
- experimental/store/postgres/checkpointer.go
- experimental/worker/webhooks.go
- experimental/store/postgres/webhooks.go
…als, branching, testing, runner, and expressions Fills the main documentation gaps: activity creation and registration, checkpointing and resume, durable signals/sleep/pause, branching and joining, the workflowtest package, the Runner production entry point, and the expression/template engine. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…packages - generateID now returns (string, error) instead of silently ignoring crypto/rand.Read failures; all callers propagate the error - sqlite DeleteCheckpoint uses executionID parameter instead of c.lease.RunID - CAS miss in MarkTriggerProcessing/MarkWebhookProcessing returns sentinel errors (ErrTriggerAlreadyClaimed, ErrWebhookAlreadyClaimed) in both postgres and sqlite stores for errors.Is detection - Webhook delivery wrapped in per-attempt 30s context.WithTimeout - sqlite ReclaimStale preserves started_at instead of resetting to NULL - sqlite EnqueueWebhook defaults to time.Now() when CreatedAt is zero Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This was referenced Apr 12, 2026
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
github.com/deepnoodle-ai/workflow/worker— a queue-backed runner with a claim loop, heartbeat lease, reaper, panic recovery, and detached finalization. Consumers implement aHandlerthat turns aClaim.Specinto aworkflow.Execution; the worker owns everything else.github.com/deepnoodle-ai/workflow/postgres— a singleStore(pgx v5) that satisfiesworker.QueueStore,workflow.Checkpointer(viaNewCheckpointer(lease), lease-fenced on(claimed_by, attempt)),workflow.StepProgressStore, andworkflow.ActivityLogger. One embeddedschema.sql, idempotentMigrate.replacedirectives so the root module stays stdlib + expr only.docs/worker.mdanddocs/postgres.mdas user-guide style documentation covering lifecycle, configuration, lease fencing, suspension handling, the reaper, theQueueStorecontract, schema, and testing. Links both from the README's "Going to production" section.Test plan
go build ./...clean in root,worker/, andpostgres/go test ./worker/...passes (unit tests cover happy path, panic-to-failed, reaper reclaim, config validation)WORKFLOW_PG_DSN=... go test ./postgres/...against a throwaway Postgres (integration tests gated on the env var)🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Documentation
Tests