feat: richer NewRun/Claim, HandlerContext, Run API, configurable schema - #38
Merged
Conversation
…rable schema Implements the library-side priorities from the v0.0.3 leverage report: - Phase 1.3: NewRun and Claim now carry ProjectID, ParentRunID, InitiatedBy, and Metadata. Postgres and sqlite schemas gain nullable project_id, parent_run_id, metadata (jsonb/text) columns with supporting indexes. - Phase 1.4: worker generates IDs for trigger and webhook outbox rows via Config.IDGenerator (crypto/rand 16-hex default). - Phase 2.1: DeadLetterStale returns DeadLetteredRun records so the worker can refund credits inline after reaping stale runs. - Phase 3: ListFailedWithCredits moves to QueueStore and is used by reconcileCredits to refund runs missing a ledger refund entry. - Phase 4.2: postgres.Store accepts WithSchema(name); defaults to "public". schema.sql is templated; Migrate runs CREATE SCHEMA IF NOT EXISTS and uses schema-qualified identifiers in every query. Schema names are validated as simple SQL identifiers. - Phase 5: Handler.Handle receives *HandlerContext carrying the Claim plus optional pre-fenced Checkpointer, StepProgressStore, ActivityLogger, and SignalStore. Config.Stores (HandlerStores factory) and Config.SignalStore wire them in. - Phase 6: postgres read-side Run API — GetRun, ListRuns (keyset pagination with metadata JSONB containment), CountRuns, DeleteRun (refuses running runs). Integration-tested against Postgres 16 for metadata round-trip, dead-letter return shape, ListFailedWithCredits filtering, the Run read API, and a non-default schema. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
📝 WalkthroughWalkthroughThis PR refactors the experimental worker and store subsystems to wire per-claim storage dependencies via a new Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant Worker
participant HandlerStores
participant Handler
participant Store as Store (Claim-scoped)
Client->>Worker: Start()
Worker->>Worker: ClaimQueued()
activate Worker
Worker->>HandlerStores: NewCheckpointer(claim)
activate HandlerStores
HandlerStores-->>Worker: Checkpointer
deactivate HandlerStores
Worker->>HandlerStores: NewStepProgressStore(claim)
HandlerStores-->>Worker: StepProgressStore
Worker->>HandlerStores: NewActivityLogger(claim)
HandlerStores-->>Worker: ActivityLogger
Worker->>Worker: Create HandlerContext<br/>(claim + dependencies)
Worker->>Handler: Handle(ctx, HandlerContext)
activate Handler
Handler->>Store: Use Checkpointer<br/>from context
Handler->>Store: Use StepProgressStore<br/>from context
Handler->>Store: Use ActivityLogger<br/>from context
Handler-->>Worker: Outcome
deactivate Handler
deactivate Worker
Worker->>Worker: Complete() or<br/>DeadLetterStale()
alt Refund Dead-Lettered
Worker->>Store: Refund(orgID, runID,<br/>workflowType, cost)
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 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 docstrings
🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 7
🧹 Nitpick comments (2)
experimental/store/postgres/integration_test.go (1)
183-187: Consider simplifying the claim drain loop.The loop assigns
claimbut only uses_ = claimand_ = id. The intent appears to be draining all queued runs to get them into running state. Consider making the intent clearer:Optional simplification
for _, id := range []string{"needs-refund", "already-refunded", "no-cost"} { - claim, _ := store.ClaimQueued(ctx, "w") - _ = claim - _ = id + _, _ = store.ClaimQueued(ctx, "w") }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@experimental/store/postgres/integration_test.go` around lines 183 - 187, The loop currently calls store.ClaimQueued(ctx, "w") but ignores both the returned claim and the loop variable id; to express the intent of draining queued runs, replace the for-range with a drain loop that repeatedly calls store.ClaimQueued until it returns nil or an error (e.g., for { claim, err := store.ClaimQueued(ctx, "w"); if err != nil || claim == nil { break } } ), removing the unused variables (claim, id) and making the intent explicit; reference the store.ClaimQueued call to locate the code to change.experimental/store/postgres/runs.go (1)
102-111: Extract the shared run projection.
GetRun,ListRuns, andscanRunare tightly coupled by column order. A future column insert/reorder here will break both query paths at runtime. A shared projection constant/helper would make this much harder to drift.Also applies to: 145-155, 306-345
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@experimental/store/postgres/runs.go` around lines 102 - 111, GetRun, ListRuns and scanRun are currently coupled to the same literal column order in the SELECT string; extract that projection into a single shared constant or helper (e.g. RunProjection or buildRunProjection()) and use it everywhere instead of repeating the formatted column list in the query construction (including the query in runs.go that uses s.t("workflow_runs") and the other SELECTs around lines 145-155 and 306-345), and update scanRun to rely on the same projection ordering or a named-column scan helper so adding/reordering columns only requires one change.
🤖 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/runs.go`:
- Around line 221-232: The DeleteRun implementation currently does a
check-then-act using checkQuery/checkArgs and then runs delQuery which creates a
race and drops the org scope; update DeleteRun to perform an atomic operation
inside a transaction: either SELECT status FROM <table> WHERE id=$1 AND
org_id=$X FOR UPDATE and then conditionally DELETE the row, or better replace
both steps with a single scoped DELETE ... RETURNING status (or a CTE) and check
the returned status to decide whether to return ErrCannotDeleteRunning or
ErrRunNotFound; ensure you always include the same org predicate and use
s.pool.BeginTx/Tx.Exec or Tx.QueryRow to run the atomic operation, and keep
existing error returns (ErrRunNotFound, ErrCannotDeleteRunning) and logging
semantics.
In `@experimental/store/postgres/schema.sql`:
- Around line 34-42: Extend the migration to relax initiated_by and convert
legacy empty-string sentinels to NULLs: add ALTER TABLE
{{.Schema}}.workflow_runs ALTER COLUMN initiated_by DROP NOT NULL and ALTER
COLUMN initiated_by DROP DEFAULT, and run UPDATE {{.Schema}}.workflow_runs SET
initiated_by = NULL WHERE initiated_by = ''; also rewrite existing org_id empty
strings to real NULLs by running UPDATE {{.Schema}}.workflow_runs SET org_id =
NULL WHERE org_id = ''; this ensures nullableString(run.InitiatedBy) in
experimental/store/postgres/queue.go works and that new filters using "org_id IS
NULL" will match migrated rows.
In `@experimental/store/postgres/store_test.go`:
- Around line 45-53: The cleanup loop in openTestStore is missing the
"workflow_checkpoints" table so stale checkpoint rows can make
TestStore_CheckpointerRoundTrip and LoadCheckpoint behave as if a checkpoint
exists; update the slice of table names in openTestStore (the for _, tbl :=
range []string{...} block) to include "workflow_checkpoints" so the test
truncates that table as well and starts from a clean state before running
TestStore_CheckpointerRoundTrip.
In `@experimental/store/sqlite/schema.sql`:
- Around line 14-21: The current migration approach re-running schema.sql won't
alter existing workflow_runs columns; implement a migration function (e.g.,
migrateWorkflowRuns called from sqlite.Store.Migrate) that detects the existing
workflow_runs column set and either ALTERs to add missing columns (project_id,
parent_run_id, org_id, initiated_by, metadata) or, when org_id/initiated_by are
present but defined as NOT NULL DEFAULT '' (legacy shape), rebuilds the table:
create a temporary table with the new schema (matching schema.sql),
copy/transform data from the old table (mapping NULLs/empty strings
appropriately), drop the old workflow_runs table, and rename the temp table to
workflow_runs; ensure this runs before queue.go starts writing new columns so
writes that pass NULLs or new columns won't fail.
In `@experimental/worker/credits.go`:
- Around line 9-12: The change removed the exported CreditStore method
ListUnrefunded which breaks existing implementations; restore compatibility by
keeping ListUnrefunded on the CreditStore interface as a deprecated method
(forwarding to the new QueueStore/ListFailedWithCredits behavior internally) or
add an additive side interface (e.g., CreditStoreUnrefunded) that extends
CreditStore with ListUnrefunded so existing callers and mocks still compile
while new code uses QueueStore.ListFailedWithCredits; update implementations to
implement the deprecated method by delegating to
QueueStore.ListFailedWithCredits and mark it as deprecated in comments.
In `@experimental/worker/memstore/memstore.go`:
- Around line 296-308: The current loop applies the limit while iterating an
unordered map, causing nondeterministic results; change the logic so you first
collect all matching worker.FailedRun entries into out (ignoring limit), then
call slices.SortFunc(out, ...) to sort deterministically by completed_at
ascending (use the FailedRun.CompletedAt field), and only after sorting apply
the limit (truncate out to at most limit) before returning; update the
comparator in slices.SortFunc to compare CompletedAt (and fall back to ID for
ties) so behavior matches the SQLite/Postgres ORDER BY completed_at ASC then
LIMIT semantics.
In `@experimental/worker/worker_test.go`:
- Around line 24-25: The public handler signature was changed to accept
*worker.HandlerContext, breaking existing handlers; revert the exported
HandlerFunc type back to the original func(context.Context, *worker.Claim)
worker.Outcome and implement an adapter that supplies a HandlerContext
additively (e.g., detect an optional interface like HandlerWithContext{
HandleWithContext(ctx context.Context, hc *worker.HandlerContext) worker.Outcome
} or wrap HandlerFunc into an internal func that builds HandlerContext and calls
the original Claim-based handler). Update the internal dispatch/path that
currently constructs a HandlerContext to call the adapter so existing handlers
that accept *worker.Claim continue working while new handlers can opt into the
richer HandlerContext via the optional side interface or wrapper.
---
Nitpick comments:
In `@experimental/store/postgres/integration_test.go`:
- Around line 183-187: The loop currently calls store.ClaimQueued(ctx, "w") but
ignores both the returned claim and the loop variable id; to express the intent
of draining queued runs, replace the for-range with a drain loop that repeatedly
calls store.ClaimQueued until it returns nil or an error (e.g., for { claim, err
:= store.ClaimQueued(ctx, "w"); if err != nil || claim == nil { break } } ),
removing the unused variables (claim, id) and making the intent explicit;
reference the store.ClaimQueued call to locate the code to change.
In `@experimental/store/postgres/runs.go`:
- Around line 102-111: GetRun, ListRuns and scanRun are currently coupled to the
same literal column order in the SELECT string; extract that projection into a
single shared constant or helper (e.g. RunProjection or buildRunProjection())
and use it everywhere instead of repeating the formatted column list in the
query construction (including the query in runs.go that uses
s.t("workflow_runs") and the other SELECTs around lines 145-155 and 306-345),
and update scanRun to rely on the same projection ordering or a named-column
scan helper so adding/reordering columns only requires one change.
🪄 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: 316ba57c-f6ed-4de1-b50a-96f6b7d00d8d
📒 Files selected for processing (27)
CLAUDE.mdexperimental/store/postgres/activity_logger.goexperimental/store/postgres/checkpointer.goexperimental/store/postgres/credits.goexperimental/store/postgres/events.goexperimental/store/postgres/integration_test.goexperimental/store/postgres/queue.goexperimental/store/postgres/runs.goexperimental/store/postgres/schema.sqlexperimental/store/postgres/schema_test.goexperimental/store/postgres/step_progress.goexperimental/store/postgres/store.goexperimental/store/postgres/store_test.goexperimental/store/postgres/triggers.goexperimental/store/postgres/webhooks.goexperimental/store/sqlite/credits.goexperimental/store/sqlite/queue.goexperimental/store/sqlite/schema.sqlexperimental/store/sqlite/store.goexperimental/worker/credits.goexperimental/worker/go.modexperimental/worker/handler.goexperimental/worker/memstore/memstore.goexperimental/worker/queue_store.goexperimental/worker/subsystems.goexperimental/worker/worker.goexperimental/worker/worker_test.go
💤 Files with no reviewable changes (1)
- experimental/store/sqlite/credits.go
Comment on lines
+221
to
+232
| var status string | ||
| if err := s.pool.QueryRow(ctx, checkQuery, checkArgs...).Scan(&status); err != nil { | ||
| if errors.Is(err, pgx.ErrNoRows) { | ||
| return ErrRunNotFound | ||
| } | ||
| return fmt.Errorf("postgres: delete run %s: check: %w", id, err) | ||
| } | ||
| if status == string(worker.StatusRunning) { | ||
| return ErrCannotDeleteRunning | ||
| } | ||
| delQuery := fmt.Sprintf(`DELETE FROM %s WHERE id = $1`, s.t("workflow_runs")) | ||
| if _, err := s.pool.Exec(ctx, delQuery, id); err != nil { |
There was a problem hiding this comment.
Make DeleteRun atomic and keep the org scope in the DELETE.
This is a check-then-act race: the row can transition to running after Line 222 but before Line 232 and still be deleted. The final DELETE also drops the orgID predicate used by the status check, so the DB is no longer enforcing the same scope you just validated. Please collapse this into one transactional operation (SELECT ... FOR UPDATE + scoped DELETE) or a single scoped DELETE ... RETURNING/CTE.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@experimental/store/postgres/runs.go` around lines 221 - 232, The DeleteRun
implementation currently does a check-then-act using checkQuery/checkArgs and
then runs delQuery which creates a race and drops the org scope; update
DeleteRun to perform an atomic operation inside a transaction: either SELECT
status FROM <table> WHERE id=$1 AND org_id=$X FOR UPDATE and then conditionally
DELETE the row, or better replace both steps with a single scoped DELETE ...
RETURNING status (or a CTE) and check the returned status to decide whether to
return ErrCannotDeleteRunning or ErrRunNotFound; ensure you always include the
same org predicate and use s.pool.BeginTx/Tx.Exec or Tx.QueryRow to run the
atomic operation, and keep existing error returns (ErrRunNotFound,
ErrCannotDeleteRunning) and logging semantics.
Comment on lines
+34
to
+42
| -- Upgrade path: tables existing from v0.0.3 had org_id NOT NULL | ||
| -- with a '' default and lacked project_id / parent_run_id / metadata. | ||
| -- Drop the not-null so empty-string sentinels can become real NULLs, | ||
| -- and add the new columns. | ||
| ALTER TABLE {{.Schema}}.workflow_runs ALTER COLUMN org_id DROP NOT NULL; | ||
| ALTER TABLE {{.Schema}}.workflow_runs ALTER COLUMN org_id DROP DEFAULT; | ||
| ALTER TABLE {{.Schema}}.workflow_runs ADD COLUMN IF NOT EXISTS project_id TEXT; | ||
| ALTER TABLE {{.Schema}}.workflow_runs ADD COLUMN IF NOT EXISTS parent_run_id TEXT; | ||
| ALTER TABLE {{.Schema}}.workflow_runs ADD COLUMN IF NOT EXISTS metadata JSONB; |
There was a problem hiding this comment.
The Postgres upgrade path is incomplete for legacy NULL semantics.
These ALTERs only fix org_id and add the new columns. Existing v0.0.3 tables still keep the old initiated_by NOT NULL DEFAULT '' shape, so experimental/store/postgres/queue.go can now fail when it writes nullableString(run.InitiatedBy). Also, legacy org_id = '' rows are never rewritten to NULL, but the new run filters use org_id IS NULL for the empty-org scope, so those rows will be skipped after upgrade. Please extend the migration to relax initiated_by too and normalize the old empty-string sentinels to NULL.
Suggested migration additions
ALTER TABLE {{.Schema}}.workflow_runs ALTER COLUMN org_id DROP NOT NULL;
ALTER TABLE {{.Schema}}.workflow_runs ALTER COLUMN org_id DROP DEFAULT;
+ALTER TABLE {{.Schema}}.workflow_runs ALTER COLUMN initiated_by DROP NOT NULL;
+ALTER TABLE {{.Schema}}.workflow_runs ALTER COLUMN initiated_by DROP DEFAULT;
+
+UPDATE {{.Schema}}.workflow_runs
+SET org_id = NULL
+WHERE org_id = '';
+
+UPDATE {{.Schema}}.workflow_runs
+SET initiated_by = NULL
+WHERE initiated_by = '';
+
ALTER TABLE {{.Schema}}.workflow_runs ADD COLUMN IF NOT EXISTS project_id TEXT;
ALTER TABLE {{.Schema}}.workflow_runs ADD COLUMN IF NOT EXISTS parent_run_id TEXT;
ALTER TABLE {{.Schema}}.workflow_runs ADD COLUMN IF NOT EXISTS metadata JSONB;📝 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
| -- Upgrade path: tables existing from v0.0.3 had org_id NOT NULL | |
| -- with a '' default and lacked project_id / parent_run_id / metadata. | |
| -- Drop the not-null so empty-string sentinels can become real NULLs, | |
| -- and add the new columns. | |
| ALTER TABLE {{.Schema}}.workflow_runs ALTER COLUMN org_id DROP NOT NULL; | |
| ALTER TABLE {{.Schema}}.workflow_runs ALTER COLUMN org_id DROP DEFAULT; | |
| ALTER TABLE {{.Schema}}.workflow_runs ADD COLUMN IF NOT EXISTS project_id TEXT; | |
| ALTER TABLE {{.Schema}}.workflow_runs ADD COLUMN IF NOT EXISTS parent_run_id TEXT; | |
| ALTER TABLE {{.Schema}}.workflow_runs ADD COLUMN IF NOT EXISTS metadata JSONB; | |
| -- Upgrade path: tables existing from v0.0.3 had org_id NOT NULL | |
| -- with a '' default and lacked project_id / parent_run_id / metadata. | |
| -- Drop the not-null so empty-string sentinels can become real NULLs, | |
| -- and add the new columns. | |
| ALTER TABLE {{.Schema}}.workflow_runs ALTER COLUMN org_id DROP NOT NULL; | |
| ALTER TABLE {{.Schema}}.workflow_runs ALTER COLUMN org_id DROP DEFAULT; | |
| ALTER TABLE {{.Schema}}.workflow_runs ALTER COLUMN initiated_by DROP NOT NULL; | |
| ALTER TABLE {{.Schema}}.workflow_runs ALTER COLUMN initiated_by DROP DEFAULT; | |
| UPDATE {{.Schema}}.workflow_runs | |
| SET org_id = NULL | |
| WHERE org_id = ''; | |
| UPDATE {{.Schema}}.workflow_runs | |
| SET initiated_by = NULL | |
| WHERE initiated_by = ''; | |
| ALTER TABLE {{.Schema}}.workflow_runs ADD COLUMN IF NOT EXISTS project_id TEXT; | |
| ALTER TABLE {{.Schema}}.workflow_runs ADD COLUMN IF NOT EXISTS parent_run_id TEXT; | |
| ALTER TABLE {{.Schema}}.workflow_runs ADD COLUMN IF NOT EXISTS metadata JSONB; |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@experimental/store/postgres/schema.sql` around lines 34 - 42, Extend the
migration to relax initiated_by and convert legacy empty-string sentinels to
NULLs: add ALTER TABLE {{.Schema}}.workflow_runs ALTER COLUMN initiated_by DROP
NOT NULL and ALTER COLUMN initiated_by DROP DEFAULT, and run UPDATE
{{.Schema}}.workflow_runs SET initiated_by = NULL WHERE initiated_by = ''; also
rewrite existing org_id empty strings to real NULLs by running UPDATE
{{.Schema}}.workflow_runs SET org_id = NULL WHERE org_id = ''; this ensures
nullableString(run.InitiatedBy) in experimental/store/postgres/queue.go works
and that new filters using "org_id IS NULL" will match migrated rows.
Comment on lines
+45
to
+53
| for _, tbl := range []string{ | ||
| "workflow_activity_log", | ||
| "workflow_step_progress", | ||
| "workflow_credit_ledger", | ||
| "workflow_triggers", | ||
| "workflow_webhooks", | ||
| "workflow_events", | ||
| "workflow_runs", | ||
| } { |
There was a problem hiding this comment.
Include workflow_checkpoints in test cleanup.
openTestStore now truncates several tables but still leaves checkpoint rows behind. TestStore_CheckpointerRoundTrip later expects LoadCheckpoint to return workflow.ErrNoCheckpoint before any save, so stale rows from a prior test run can make this helper lie about starting from a clean state.
💡 Proposed fix
for _, tbl := range []string{
"workflow_activity_log",
"workflow_step_progress",
+ "workflow_checkpoints",
"workflow_credit_ledger",
"workflow_triggers",
"workflow_webhooks",
"workflow_events",
"workflow_runs",
} {🤖 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 45 - 53, The cleanup
loop in openTestStore is missing the "workflow_checkpoints" table so stale
checkpoint rows can make TestStore_CheckpointerRoundTrip and LoadCheckpoint
behave as if a checkpoint exists; update the slice of table names in
openTestStore (the for _, tbl := range []string{...} block) to include
"workflow_checkpoints" so the test truncates that table as well and starts from
a clean state before running TestStore_CheckpointerRoundTrip.
Comment on lines
+14
to
+21
| org_id TEXT, | ||
| project_id TEXT, | ||
| parent_run_id TEXT, | ||
| workflow_type TEXT NOT NULL DEFAULT '', | ||
| initiated_by TEXT, | ||
| credit_cost INTEGER NOT NULL DEFAULT 0, | ||
| callback_url TEXT NOT NULL DEFAULT '', | ||
| metadata TEXT |
There was a problem hiding this comment.
Existing SQLite databases will not migrate to this shape.
sqlite.Store.Migrate() still just re-executes schema.sql, so these CREATE TABLE IF NOT EXISTS changes never retrofit an existing workflow_runs table. After upgrade, experimental/store/sqlite/queue.go already writes project_id, parent_run_id, and metadata, and it passes NULL for empty org_id / initiated_by; legacy databases will still be missing those columns and may still reject the NULL writes. Please add an explicit SQLite migration/rebuild path before relying on this new schema.
Possible direction
func (s *Store) Migrate(ctx context.Context) error {
+ if err := migrateWorkflowRuns(ctx, s.db); err != nil {
+ return err
+ }
_, err := s.db.ExecContext(ctx, schemaSQL)
return err
}migrateWorkflowRuns should detect legacy workflow_runs, add missing columns, and rebuild the table when org_id / initiated_by still use the old NOT NULL DEFAULT '' shape.
🤖 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 14 - 21, The current
migration approach re-running schema.sql won't alter existing workflow_runs
columns; implement a migration function (e.g., migrateWorkflowRuns called from
sqlite.Store.Migrate) that detects the existing workflow_runs column set and
either ALTERs to add missing columns (project_id, parent_run_id, org_id,
initiated_by, metadata) or, when org_id/initiated_by are present but defined as
NOT NULL DEFAULT '' (legacy shape), rebuilds the table: create a temporary table
with the new schema (matching schema.sql), copy/transform data from the old
table (mapping NULLs/empty strings appropriately), drop the old workflow_runs
table, and rename the temp table to workflow_runs; ensure this runs before
queue.go starts writing new columns so writes that pass NULLs or new columns
won't fail.
Comment on lines
+9
to
+12
| // | ||
| // CreditStore is pure ledger: listing which failed runs still need | ||
| // a refund is a QueueStore concern (ListFailedWithCredits) because | ||
| // it joins the ledger against run status. |
There was a problem hiding this comment.
Preserve CreditStore compatibility.
Moving refund discovery to QueueStore is fine, but removing ListUnrefunded from the exported CreditStore still breaks existing store implementations, mocks, and any callers compiled against the old contract. Please keep the old method as a deprecated compatibility path or introduce an additive side interface instead of changing the exported surface in place.
Based on learnings: "Interfaces should be small (one method when possible) and never modified once exported — use optional side interfaces instead."
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@experimental/worker/credits.go` around lines 9 - 12, The change removed the
exported CreditStore method ListUnrefunded which breaks existing
implementations; restore compatibility by keeping ListUnrefunded on the
CreditStore interface as a deprecated method (forwarding to the new
QueueStore/ListFailedWithCredits behavior internally) or add an additive side
interface (e.g., CreditStoreUnrefunded) that extends CreditStore with
ListUnrefunded so existing callers and mocks still compile while new code uses
QueueStore.ListFailedWithCredits; update implementations to implement the
deprecated method by delegating to QueueStore.ListFailedWithCredits and mark it
as deprecated in comments.
Comment on lines
+296
to
+308
| if limit > 0 && len(out) >= limit { | ||
| break | ||
| } | ||
| } | ||
| slices.SortFunc(out, func(a, b worker.FailedRun) int { | ||
| if a.ID < b.ID { | ||
| return -1 | ||
| } | ||
| if a.ID > b.ID { | ||
| return 1 | ||
| } | ||
| return 0 | ||
| }) |
There was a problem hiding this comment.
Limit applied before sort yields non-deterministic results.
The limit is checked inside the loop (which iterates over an unordered map), causing the function to stop collecting entries before sorting. This differs from the SQLite and Postgres implementations that use ORDER BY completed_at ASC before LIMIT, ensuring deterministic pagination.
Consider collecting all matching runs first, then sorting, then applying the limit:
Proposed fix
func (s *Store) ListFailedWithCredits(_ context.Context, limit int) ([]worker.FailedRun, error) {
s.mu.Lock()
defer s.mu.Unlock()
var out []worker.FailedRun
for _, row := range s.runs {
if row.status != worker.StatusFailed {
continue
}
if row.creditCost <= 0 {
continue
}
out = append(out, worker.FailedRun{
ID: row.id,
OrgID: row.orgID,
WorkflowType: row.workflowType,
CreditCost: row.creditCost,
})
- if limit > 0 && len(out) >= limit {
- break
- }
}
slices.SortFunc(out, func(a, b worker.FailedRun) int {
if a.ID < b.ID {
return -1
}
if a.ID > b.ID {
return 1
}
return 0
})
+ if limit > 0 && len(out) > limit {
+ out = out[:limit]
+ }
return out, 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
| if limit > 0 && len(out) >= limit { | |
| break | |
| } | |
| } | |
| slices.SortFunc(out, func(a, b worker.FailedRun) int { | |
| if a.ID < b.ID { | |
| return -1 | |
| } | |
| if a.ID > b.ID { | |
| return 1 | |
| } | |
| return 0 | |
| }) | |
| func (s *Store) ListFailedWithCredits(_ context.Context, limit int) ([]worker.FailedRun, error) { | |
| s.mu.Lock() | |
| defer s.mu.Unlock() | |
| var out []worker.FailedRun | |
| for _, row := range s.runs { | |
| if row.status != worker.StatusFailed { | |
| continue | |
| } | |
| if row.creditCost <= 0 { | |
| continue | |
| } | |
| out = append(out, worker.FailedRun{ | |
| ID: row.id, | |
| OrgID: row.orgID, | |
| WorkflowType: row.workflowType, | |
| CreditCost: row.creditCost, | |
| }) | |
| } | |
| slices.SortFunc(out, func(a, b worker.FailedRun) int { | |
| if a.ID < b.ID { | |
| return -1 | |
| } | |
| if a.ID > b.ID { | |
| return 1 | |
| } | |
| return 0 | |
| }) | |
| if limit > 0 && len(out) > limit { | |
| out = out[:limit] | |
| } | |
| return out, nil | |
| } |
🤖 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 296 - 308, The current
loop applies the limit while iterating an unordered map, causing
nondeterministic results; change the logic so you first collect all matching
worker.FailedRun entries into out (ignoring limit), then call
slices.SortFunc(out, ...) to sort deterministically by completed_at ascending
(use the FailedRun.CompletedAt field), and only after sorting apply the limit
(truncate out to at most limit) before returning; update the comparator in
slices.SortFunc to compare CompletedAt (and fall back to ID for ties) so
behavior matches the SQLite/Postgres ORDER BY completed_at ASC then LIMIT
semantics.
Comment on lines
+24
to
+25
| handler := worker.HandlerFunc(func(_ context.Context, hc *worker.HandlerContext) worker.Outcome { | ||
| handled <- hc.Claim |
There was a problem hiding this comment.
Avoid a source-breaking handler signature change.
These updates confirm the public handler callback no longer accepts *worker.Claim. That forces every existing handler implementation to change just to access the same claim data. Please keep the old callback shape and surface HandlerContext additively, e.g. via an optional side interface or wrapper, rather than replacing the exported contract.
Based on learnings: "Interfaces should be small (one method when possible) and never modified once exported — use optional side interfaces instead."
Also applies to: 83-83, 134-136, 179-184
🤖 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 24 - 25, The public handler
signature was changed to accept *worker.HandlerContext, breaking existing
handlers; revert the exported HandlerFunc type back to the original
func(context.Context, *worker.Claim) worker.Outcome and implement an adapter
that supplies a HandlerContext additively (e.g., detect an optional interface
like HandlerWithContext{ HandleWithContext(ctx context.Context, hc
*worker.HandlerContext) worker.Outcome } or wrap HandlerFunc into an internal
func that builds HandlerContext and calls the original Claim-based handler).
Update the internal dispatch/path that currently constructs a HandlerContext to
call the adapter so existing handlers that accept *worker.Claim continue working
while new handlers can opt into the richer HandlerContext via the optional side
interface or wrapper.
myzie
added a commit
that referenced
this pull request
Apr 13, 2026
Addresses the critical review on #38: - schema.sql: add UPDATE workflow_runs SET org_id/initiated_by = NULL WHERE = '' so single-tenant rows migrated from v0.0.3 stay visible to the new read API (GetRun, ListRuns), and drop NOT NULL on initiated_by alongside org_id. Expand the upgrade-path comment to name v0.0.3 -> v0.0.4 explicitly. - DeleteRun: collapse into one atomic DELETE ... WHERE status <> 'running' RETURNING status; a post-delete probe only runs to pick between ErrRunNotFound and ErrCannotDeleteRunning for the error message, not to gate the destructive path. - experimental/worker/runquery: new stdlib-only package holding Run, RunFilter, RunCursor, ErrRunNotFound, ErrCannotDeleteRunning, and the Store interface. Postgres now aliases the runquery types and satisfies runquery.Store at compile time so dashboards can import the neutral package instead of experimental/store/postgres. - sqlite ReclaimStale: also clear started_at, matching Postgres. - sqlite store.go: document that SQLite has no schema namespacing; consumers who need coexistence should hand the library a dedicated *sql.DB. - HandlerContext/HandlerStores: doc comment clarifying only Checkpointer is lease-fenced. ProgressStore and ActivityLogger accept *Claim for symmetry and typically ignore it; SignalStore is shared across claims and intentionally lives on Config, not the factory. - Rename QueueStore.ListFailedWithCredits -> ListRefundPending in worker, memstore, postgres, sqlite, and the integration test. - Add TestStore_UpgradeFromV003: seeds a v0.0.3-shaped table with a populated row, runs Migrate, and asserts the row is still findable via the new read API. Catches the very regression the review flagged. - Add TestStore_ClaimQueuedConcurrent: 8 goroutines race to claim a 20-run queue; every run must be claimed exactly once. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
3 tasks
myzie
added a commit
that referenced
this pull request
Apr 13, 2026
Addresses the critical review on #38: - schema.sql: add UPDATE workflow_runs SET org_id/initiated_by = NULL WHERE = '' so single-tenant rows migrated from v0.0.3 stay visible to the new read API (GetRun, ListRuns), and drop NOT NULL on initiated_by alongside org_id. Expand the upgrade-path comment to name v0.0.3 -> v0.0.4 explicitly. - DeleteRun: collapse into one atomic DELETE ... WHERE status <> 'running' RETURNING status; a post-delete probe only runs to pick between ErrRunNotFound and ErrCannotDeleteRunning for the error message, not to gate the destructive path. - experimental/worker/runquery: new stdlib-only package holding Run, RunFilter, RunCursor, ErrRunNotFound, ErrCannotDeleteRunning, and the Store interface. Postgres now aliases the runquery types and satisfies runquery.Store at compile time so dashboards can import the neutral package instead of experimental/store/postgres. - sqlite ReclaimStale: also clear started_at, matching Postgres. - sqlite store.go: document that SQLite has no schema namespacing; consumers who need coexistence should hand the library a dedicated *sql.DB. - HandlerContext/HandlerStores: doc comment clarifying only Checkpointer is lease-fenced. ProgressStore and ActivityLogger accept *Claim for symmetry and typically ignore it; SignalStore is shared across claims and intentionally lives on Config, not the factory. - Rename QueueStore.ListFailedWithCredits -> ListRefundPending in worker, memstore, postgres, sqlite, and the integration test. - Add TestStore_UpgradeFromV003: seeds a v0.0.3-shaped table with a populated row, runs Migrate, and asserts the row is still findable via the new read API. Catches the very regression the review flagged. - Add TestStore_ClaimQueuedConcurrent: 8 goroutines race to claim a 20-run queue; every run must be claimed exactly once. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
myzie
added a commit
that referenced
this pull request
Apr 13, 2026
…n loop - memstore.ListRefundPending: collect every match, sort by completed_at ASC NULLS LAST then ID, then apply the limit. The previous loop applied the limit while iterating an unordered map, then sorted the truncated set, so the returned page was nondeterministic. - postgres/runs.go: extract the GetRun/ListRuns column projection into a single runProjection constant kept in lock-step with scanRun. Adding or reordering a column now touches one place, not two. - postgres/integration_test.go: replace the for-range-with-blanks queue drain with an explicit "claim until empty" loop. Same effect, the intent is on the page. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
myzie
added a commit
that referenced
this pull request
Apr 13, 2026
…39) * fix: PR #38 follow-ups — upgrade path, DeleteRun race, runquery lift Addresses the critical review on #38: - schema.sql: add UPDATE workflow_runs SET org_id/initiated_by = NULL WHERE = '' so single-tenant rows migrated from v0.0.3 stay visible to the new read API (GetRun, ListRuns), and drop NOT NULL on initiated_by alongside org_id. Expand the upgrade-path comment to name v0.0.3 -> v0.0.4 explicitly. - DeleteRun: collapse into one atomic DELETE ... WHERE status <> 'running' RETURNING status; a post-delete probe only runs to pick between ErrRunNotFound and ErrCannotDeleteRunning for the error message, not to gate the destructive path. - experimental/worker/runquery: new stdlib-only package holding Run, RunFilter, RunCursor, ErrRunNotFound, ErrCannotDeleteRunning, and the Store interface. Postgres now aliases the runquery types and satisfies runquery.Store at compile time so dashboards can import the neutral package instead of experimental/store/postgres. - sqlite ReclaimStale: also clear started_at, matching Postgres. - sqlite store.go: document that SQLite has no schema namespacing; consumers who need coexistence should hand the library a dedicated *sql.DB. - HandlerContext/HandlerStores: doc comment clarifying only Checkpointer is lease-fenced. ProgressStore and ActivityLogger accept *Claim for symmetry and typically ignore it; SignalStore is shared across claims and intentionally lives on Config, not the factory. - Rename QueueStore.ListFailedWithCredits -> ListRefundPending in worker, memstore, postgres, sqlite, and the integration test. - Add TestStore_UpgradeFromV003: seeds a v0.0.3-shaped table with a populated row, runs Migrate, and asserts the row is still findable via the new read API. Catches the very regression the review flagged. - Add TestStore_ClaimQueuedConcurrent: 8 goroutines race to claim a 20-run queue; every run must be claimed exactly once. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * fix: address PR #38 review nits — memstore sort, run projection, drain loop - memstore.ListRefundPending: collect every match, sort by completed_at ASC NULLS LAST then ID, then apply the limit. The previous loop applied the limit while iterating an unordered map, then sorted the truncated set, so the returned page was nondeterministic. - postgres/runs.go: extract the GetRun/ListRuns column projection into a single runProjection constant kept in lock-step with scanRun. Adding or reordering a column now touches one place, not two. - postgres/integration_test.go: replace the for-range-with-blanks queue drain with an explicit "claim until empty" loop. Same effect, the intent is on the page. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2 tasks
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
Implements the library-side priorities from the v0.0.3 leverage report. Builds on #37 (Claim carries WorkerID + fencing).
NewRunandClaimcarryProjectID,ParentRunID,InitiatedBy, andMetadata map[string]string. Schemas gain nullable columns and supporting indexes.Config.IDGenerator.DeadLetterStalereturns[]DeadLetteredRunso the worker can refund credits inline after reaping.ListFailedWithCreditsmoves toQueueStore; worker reconciles refunds usingCreditStore.HasRefund.postgres.StoreacceptsWithSchema(name), defaulting to"public".schema.sqlis templated;MigraterunsCREATE SCHEMA IF NOT EXISTSand uses schema-qualified identifiers in every query. Schema names validated as simple SQL identifiers (letters, digits, underscore).Handler.Handlereceives*HandlerContextcarrying theClaimplus optional pre-fencedCheckpointer,StepProgressStore,ActivityLogger, andSignalStore. Wired viaConfig.Stores(HandlerStoresfactory) andConfig.SignalStore.GetRun,ListRuns(keyset pagination with metadata JSONB containment filter),CountRuns,DeleteRun(refuses running runs).Test plan
make test-allpasses (root module, worker, postgres, sqlite submodules)validateIdentifierandWithSchemadefaulting topublicDeadLetterStalereturnsDeadLetteredRunwith OrgID/WorkflowType/CreditCostListFailedWithCreditsexcludes already-refunded and zero-cost runsGetRunorg scoping,CountRuns/ListRunsfiltering, metadata JSONB containment, keyset paginationDeleteRunrefusesStatusRunning, returnsErrRunNotFoundfor missing rowsWithSchema("wf_alt")creates tables in the alt schema and enqueue/claim/complete works end-to-end🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Improvements