From 1cfd60eb16e3c0f6a433b30623cd819dbfb23738 Mon Sep 17 00:00:00 2001 From: Curtis Myzie Date: Sun, 12 Apr 2026 21:51:21 -0400 Subject: [PATCH 1/2] =?UTF-8?q?fix:=20PR=20#38=20follow-ups=20=E2=80=94=20?= =?UTF-8?q?upgrade=20path,=20DeleteRun=20race,=20runquery=20lift?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- .../store/postgres/integration_test.go | 156 +++++++++++++++- experimental/store/postgres/queue.go | 4 +- experimental/store/postgres/runs.go | 168 ++++++++---------- experimental/store/postgres/schema.sql | 20 ++- experimental/store/sqlite/queue.go | 10 +- experimental/store/sqlite/store.go | 19 ++ experimental/worker/credits.go | 2 +- experimental/worker/handler.go | 37 +++- experimental/worker/memstore/memstore.go | 4 +- experimental/worker/queue_store.go | 6 +- experimental/worker/runquery/runquery.go | 122 +++++++++++++ experimental/worker/subsystems.go | 2 +- 12 files changed, 435 insertions(+), 115 deletions(-) create mode 100644 experimental/worker/runquery/runquery.go diff --git a/experimental/store/postgres/integration_test.go b/experimental/store/postgres/integration_test.go index 457e6fd..0cf18c9 100644 --- a/experimental/store/postgres/integration_test.go +++ b/experimental/store/postgres/integration_test.go @@ -4,6 +4,8 @@ import ( "context" "errors" "os" + "strconv" + "sync" "testing" "time" @@ -11,6 +13,7 @@ import ( "github.com/deepnoodle-ai/workflow/experimental/store/postgres" "github.com/deepnoodle-ai/workflow/experimental/worker" + "github.com/deepnoodle-ai/workflow/experimental/worker/runquery" ) func TestStore_CustomSchema(t *testing.T) { @@ -152,7 +155,7 @@ func TestStore_DeadLetterStaleReturnsRunMetadata(t *testing.T) { } } -func TestStore_ListFailedWithCredits(t *testing.T) { +func TestStore_ListRefundPending(t *testing.T) { store, pool := openTestStore(t) ctx := context.Background() @@ -206,7 +209,7 @@ func TestStore_ListFailedWithCredits(t *testing.T) { t.Fatalf("refund: %v", err) } - failed, err := store.ListFailedWithCredits(ctx, 50) + failed, err := store.ListRefundPending(ctx, 50) if err != nil { t.Fatalf("list: %v", err) } @@ -289,3 +292,152 @@ func TestStore_RunReadAPI(t *testing.T) { t.Fatalf("expected ErrRunNotFound, got %v", err) } } + +// TestStore_UpgradeFromV003 seeds a v0.0.3-shaped workflow_runs table +// (org_id NOT NULL DEFAULT '', initiated_by NOT NULL DEFAULT '', no +// project_id/parent_run_id/metadata columns) with a populated row, +// then runs Migrate and confirms the new read API can still find the +// row. This catches the regression the review flagged: without the +// UPDATE ... SET org_id = NULL WHERE org_id = '' step, pre-existing +// single-tenant rows become invisible to GetRun / ListRuns. +func TestStore_UpgradeFromV003(t *testing.T) { + dsn := os.Getenv(dsnEnv) + if dsn == "" { + t.Skipf("set %s to run postgres store tests", dsnEnv) + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + pool, err := pgxpool.New(ctx, dsn) + if err != nil { + t.Fatalf("connect: %v", err) + } + t.Cleanup(func() { pool.Close() }) + + // Fresh schema so the v0.0.3 DDL below is what Migrate sees. + if _, err := pool.Exec(ctx, `DROP SCHEMA IF EXISTS wf_upgrade CASCADE`); err != nil { + t.Fatalf("drop schema: %v", err) + } + if _, err := pool.Exec(ctx, `CREATE SCHEMA wf_upgrade`); err != nil { + t.Fatalf("create schema: %v", err) + } + t.Cleanup(func() { + _, _ = pool.Exec(context.Background(), `DROP SCHEMA IF EXISTS wf_upgrade CASCADE`) + }) + + // v0.0.3 workflow_runs shape: org_id / initiated_by NOT NULL '', + // no project_id / parent_run_id / metadata columns. + if _, err := pool.Exec(ctx, ` + CREATE TABLE wf_upgrade.workflow_runs ( + id TEXT PRIMARY KEY, + spec BYTEA NOT NULL, + status TEXT NOT NULL, + attempt INTEGER NOT NULL DEFAULT 0, + claimed_by TEXT NOT NULL DEFAULT '', + heartbeat_at TIMESTAMPTZ, + checkpoint BYTEA, + result BYTEA, + error_message TEXT NOT NULL DEFAULT '', + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + started_at TIMESTAMPTZ, + completed_at TIMESTAMPTZ, + org_id TEXT NOT NULL DEFAULT '', + workflow_type TEXT NOT NULL DEFAULT '', + initiated_by TEXT NOT NULL DEFAULT '', + credit_cost INTEGER NOT NULL DEFAULT 0, + callback_url TEXT NOT NULL DEFAULT '' + ) + `); err != nil { + t.Fatalf("seed v0.0.3 table: %v", err) + } + // Insert a populated single-tenant row with the old sentinels. + if _, err := pool.Exec(ctx, ` + INSERT INTO wf_upgrade.workflow_runs (id, spec, status, workflow_type) + VALUES ('legacy-run', $1, 'completed', 'demo') + `, []byte(`{}`)); err != nil { + t.Fatalf("seed row: %v", err) + } + + // Run the upgrade. + store := postgres.New(pool, postgres.WithSchema("wf_upgrade")) + if err := store.Migrate(ctx); err != nil { + t.Fatalf("migrate: %v", err) + } + + // The row should still be findable via the single-tenant API + // (orgID == "" matches org_id IS NULL). If the UPDATE step is + // missing from schema.sql, this GetRun returns ErrRunNotFound. + got, err := store.GetRun(ctx, "", "legacy-run") + if err != nil { + t.Fatalf("GetRun after upgrade: %v", err) + } + if got.ID != "legacy-run" || got.WorkflowType != "demo" { + t.Fatalf("upgraded row mismatch: %+v", got) + } + + // And ListRuns in single-tenant mode should see it too. + list, _, err := store.ListRuns(ctx, "", runquery.RunFilter{}) + if err != nil { + t.Fatalf("ListRuns after upgrade: %v", err) + } + if len(list) != 1 || list[0].ID != "legacy-run" { + t.Fatalf("ListRuns after upgrade: got %+v", list) + } + + // Re-running Migrate must be a no-op (all statements idempotent). + if err := store.Migrate(ctx); err != nil { + t.Fatalf("re-migrate: %v", err) + } +} + +// TestStore_ClaimQueuedConcurrent spins up several goroutines claiming +// a fixed-size pool of queued runs. Every run must be claimed exactly +// once across all workers — no duplicates, no drops. +func TestStore_ClaimQueuedConcurrent(t *testing.T) { + store, _ := openTestStore(t) + ctx := context.Background() + + const numRuns = 20 + for i := 0; i < numRuns; i++ { + id := "concurrent-" + strconv.Itoa(i) + if err := store.Enqueue(ctx, worker.NewRun{ID: id, Spec: []byte(`{}`)}); err != nil { + t.Fatalf("enqueue %s: %v", id, err) + } + } + + const numWorkers = 8 + var ( + mu sync.Mutex + claims = map[string]string{} // runID -> workerID + wg sync.WaitGroup + ) + for w := 0; w < numWorkers; w++ { + wg.Add(1) + workerID := "w-" + strconv.Itoa(w) + go func(workerID string) { + defer wg.Done() + for { + claim, err := store.ClaimQueued(ctx, workerID) + if err != nil { + t.Errorf("claim: %v", err) + return + } + if claim == nil { + return + } + mu.Lock() + if prev, dup := claims[claim.ID]; dup { + t.Errorf("duplicate claim for %s: %s and %s", claim.ID, prev, workerID) + } + claims[claim.ID] = workerID + mu.Unlock() + } + }(workerID) + } + wg.Wait() + + if len(claims) != numRuns { + t.Fatalf("claimed %d runs, want %d", len(claims), numRuns) + } +} + diff --git a/experimental/store/postgres/queue.go b/experimental/store/postgres/queue.go index 4c06d8d..775cb2f 100644 --- a/experimental/store/postgres/queue.go +++ b/experimental/store/postgres/queue.go @@ -257,10 +257,10 @@ func (s *Store) DeadLetterStale(ctx context.Context, staleBefore time.Time, maxA return out, nil } -// ListFailedWithCredits implements worker.QueueStore by joining +// ListRefundPending implements worker.QueueStore by joining // workflow_runs against the credit ledger: runs in StatusFailed with // a matching debit but no matching refund. -func (s *Store) ListFailedWithCredits(ctx context.Context, limit int) ([]worker.FailedRun, error) { +func (s *Store) ListRefundPending(ctx context.Context, limit int) ([]worker.FailedRun, error) { if limit <= 0 { limit = 50 } diff --git a/experimental/store/postgres/runs.go b/experimental/store/postgres/runs.go index 6185630..ac9d82e 100644 --- a/experimental/store/postgres/runs.go +++ b/experimental/store/postgres/runs.go @@ -10,84 +10,35 @@ import ( "github.com/jackc/pgx/v5" "github.com/deepnoodle-ai/workflow/experimental/worker" + "github.com/deepnoodle-ai/workflow/experimental/worker/runquery" ) -// ErrRunNotFound is returned by GetRun and DeleteRun when no row -// matches the requested ID (within the given org scope). -var ErrRunNotFound = errors.New("postgres: run not found") +// ErrRunNotFound is an alias for runquery.ErrRunNotFound so existing +// callers comparing against postgres.ErrRunNotFound keep working. +// New code should use runquery.ErrRunNotFound directly. +var ErrRunNotFound = runquery.ErrRunNotFound -// ErrCannotDeleteRunning is returned by DeleteRun when asked to -// delete a run whose status is StatusRunning. Callers must cancel -// or wait for the run to complete before deleting it. -var ErrCannotDeleteRunning = errors.New("postgres: cannot delete a running run") +// ErrCannotDeleteRunning is an alias for +// runquery.ErrCannotDeleteRunning. +var ErrCannotDeleteRunning = runquery.ErrCannotDeleteRunning -// Run is a read-side view of a workflow_runs row. Returned from -// GetRun and ListRuns for operational dashboards and run management. -type Run struct { - ID string - OrgID string - ProjectID string - ParentRunID string - WorkflowType string - Status worker.Status - CreditCost int - InitiatedBy string - CallbackURL string - Spec []byte - Result []byte - ErrorMessage string - Metadata map[string]string - Attempt int - ClaimedBy string - HeartbeatAt *time.Time - CreatedAt time.Time - StartedAt *time.Time - CompletedAt *time.Time -} - -// RunFilter narrows a ListRuns or CountRuns query. Zero-value fields -// are ignored. Filter values are ANDed together. -type RunFilter struct { - // WorkflowType matches exactly when non-empty. - WorkflowType string - - // Status matches exactly when non-empty. - Status worker.Status - - // ProjectID matches exactly when non-empty. Empty means "any - // project, including NULL." - ProjectID string - - // ParentRunID matches exactly when non-empty. - ParentRunID string +// Run aliases runquery.Run so callers can write postgres.Run +// during the transition. New code should use runquery.Run. +type Run = runquery.Run - // InitiatedBy matches exactly when non-empty. - InitiatedBy string +// RunFilter aliases runquery.RunFilter. +type RunFilter = runquery.RunFilter - // Metadata is evaluated with JSONB containment (@>). Empty map - // matches all rows. - Metadata map[string]string +// RunCursor aliases runquery.RunCursor. +type RunCursor = runquery.RunCursor - // Cursor is the pagination anchor. Nil means "start from the - // newest row." - Cursor *RunCursor - - // Limit caps the number of rows returned. Zero means 50. - Limit int -} - -// RunCursor is an opaque keyset pagination anchor. Consumers -// typically base64-encode it for API transport and decode back to -// this shape on the next call. -type RunCursor struct { - CreatedAt time.Time - ID string -} +// Compile-time check that *Store implements the read-side contract. +var _ runquery.Store = (*Store)(nil) // GetRun returns a single run by ID, scoped to orgID. An empty // orgID matches rows with NULL org_id (single-tenant). Returns -// ErrRunNotFound when no matching row exists. -func (s *Store) GetRun(ctx context.Context, orgID, id string) (*Run, error) { +// runquery.ErrRunNotFound when no matching row exists. +func (s *Store) GetRun(ctx context.Context, orgID, id string) (*runquery.Run, error) { if id == "" { return nil, fmt.Errorf("postgres: GetRun: id is required") } @@ -113,7 +64,7 @@ func (s *Store) GetRun(ctx context.Context, orgID, id string) (*Run, error) { r, err := scanRun(row) if err != nil { if errors.Is(err, pgx.ErrNoRows) { - return nil, ErrRunNotFound + return nil, runquery.ErrRunNotFound } return nil, fmt.Errorf("postgres: get run %s: %w", id, err) } @@ -126,7 +77,7 @@ func (s *Store) GetRun(ctx context.Context, orgID, id string) (*Run, error) { // // orgID == "" lists runs with NULL org_id (single-tenant). Pass a // real org ID for scoped B2B listings. -func (s *Store) ListRuns(ctx context.Context, orgID string, filter RunFilter) ([]*Run, *RunCursor, error) { +func (s *Store) ListRuns(ctx context.Context, orgID string, filter runquery.RunFilter) ([]*runquery.Run, *runquery.RunCursor, error) { limit := filter.Limit if limit <= 0 { limit = 50 @@ -160,7 +111,7 @@ func (s *Store) ListRuns(ctx context.Context, orgID string, filter RunFilter) ([ } defer rows.Close() - out := make([]*Run, 0, limit) + out := make([]*runquery.Run, 0, limit) for rows.Next() { r, err := scanRun(rows) if err != nil { @@ -172,10 +123,10 @@ func (s *Store) ListRuns(ctx context.Context, orgID string, filter RunFilter) ([ return nil, nil, err } - var cursor *RunCursor + var cursor *runquery.RunCursor if len(out) > limit { last := out[limit-1] - cursor = &RunCursor{CreatedAt: last.CreatedAt, ID: last.ID} + cursor = &runquery.RunCursor{CreatedAt: last.CreatedAt, ID: last.ID} out = out[:limit] } return out, cursor, nil @@ -184,7 +135,7 @@ func (s *Store) ListRuns(ctx context.Context, orgID string, filter RunFilter) ([ // CountRuns returns the total number of rows matching filter. The // cursor field on filter is ignored: counts are over the entire // filtered set, not a single page. -func (s *Store) CountRuns(ctx context.Context, orgID string, filter RunFilter) (int, error) { +func (s *Store) CountRuns(ctx context.Context, orgID string, filter runquery.RunFilter) (int, error) { // Counts ignore the cursor. filter.Cursor = nil filter.Limit = 0 @@ -203,42 +154,75 @@ func (s *Store) CountRuns(ctx context.Context, orgID string, filter RunFilter) ( // DeleteRun removes a run row by ID. Running runs cannot be // deleted; the caller must cancel or wait for the run first. +// +// Implemented as a single DELETE ... RETURNING status so the check +// and the delete happen atomically. If RETURNING yields no row, we +// fall back to a cheap existence probe to distinguish "running" +// from "not found." func (s *Store) DeleteRun(ctx context.Context, orgID, id string) error { if id == "" { return fmt.Errorf("postgres: DeleteRun: id is required") } var ( - checkQuery string - checkArgs []any + delQuery string + delArgs []any ) if orgID == "" { - checkQuery = fmt.Sprintf(`SELECT status FROM %s WHERE id = $1 AND org_id IS NULL`, s.t("workflow_runs")) - checkArgs = []any{id} + delQuery = fmt.Sprintf(` + DELETE FROM %s + WHERE id = $1 AND org_id IS NULL AND status <> $2 + RETURNING status + `, s.t("workflow_runs")) + delArgs = []any{id, string(worker.StatusRunning)} } else { - checkQuery = fmt.Sprintf(`SELECT status FROM %s WHERE id = $1 AND org_id = $2`, s.t("workflow_runs")) - checkArgs = []any{id, orgID} + delQuery = fmt.Sprintf(` + DELETE FROM %s + WHERE id = $1 AND org_id = $2 AND status <> $3 + RETURNING status + `, s.t("workflow_runs")) + delArgs = []any{id, orgID, string(worker.StatusRunning)} + } + var deletedStatus string + err := s.pool.QueryRow(ctx, delQuery, delArgs...).Scan(&deletedStatus) + if err == nil { + return nil + } + if !errors.Is(err, pgx.ErrNoRows) { + return fmt.Errorf("postgres: delete run %s: %w", id, err) + } + // DELETE affected zero rows: either the run is missing or it is + // running. Probe once to pick the right sentinel. + var ( + probeQuery string + probeArgs []any + ) + if orgID == "" { + probeQuery = fmt.Sprintf(`SELECT status FROM %s WHERE id = $1 AND org_id IS NULL`, s.t("workflow_runs")) + probeArgs = []any{id} + } else { + probeQuery = fmt.Sprintf(`SELECT status FROM %s WHERE id = $1 AND org_id = $2`, s.t("workflow_runs")) + probeArgs = []any{id, orgID} } var status string - if err := s.pool.QueryRow(ctx, checkQuery, checkArgs...).Scan(&status); err != nil { + if err := s.pool.QueryRow(ctx, probeQuery, probeArgs...).Scan(&status); err != nil { if errors.Is(err, pgx.ErrNoRows) { - return ErrRunNotFound + return runquery.ErrRunNotFound } - return fmt.Errorf("postgres: delete run %s: check: %w", id, err) + return fmt.Errorf("postgres: delete run %s: probe: %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 { - return fmt.Errorf("postgres: delete run %s: %w", id, err) + return runquery.ErrCannotDeleteRunning } - return nil + // Race: row existed and was not running when we probed, but the + // DELETE missed it. It must have transitioned out from under us; + // treat as not found. + return runquery.ErrRunNotFound } // runFilterConds builds the WHERE clause for ListRuns / CountRuns. // Returns the list of conditions and the corresponding positional // args, starting at $1. -func (s *Store) runFilterConds(orgID string, f RunFilter) ([]string, []any) { +func (s *Store) runFilterConds(orgID string, f runquery.RunFilter) ([]string, []any) { var ( conds []string args []any @@ -303,9 +287,9 @@ type rowScanner interface { Scan(dest ...any) error } -func scanRun(r rowScanner) (*Run, error) { +func scanRun(r rowScanner) (*runquery.Run, error) { var ( - run Run + run runquery.Run metadataRaw []byte heartbeatAt *time.Time startedAt *time.Time diff --git a/experimental/store/postgres/schema.sql b/experimental/store/postgres/schema.sql index 8e71fe3..20b0b04 100644 --- a/experimental/store/postgres/schema.sql +++ b/experimental/store/postgres/schema.sql @@ -31,16 +31,28 @@ CREATE TABLE IF NOT EXISTS {{.Schema}}.workflow_runs ( 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. +-- v0.0.3 -> v0.0.4 upgrade path. Existing tables from v0.0.3 had +-- org_id TEXT NOT NULL DEFAULT '', initiated_by TEXT NOT NULL DEFAULT '', +-- and lacked project_id / parent_run_id / metadata. This block drops +-- the NOT NULLs so empty-string sentinels can become real NULLs, +-- adds the new columns, then rewrites any pre-existing '' sentinels +-- to NULL. Every statement is idempotent, so Migrate() is safe to +-- call repeatedly. 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; 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; +-- Rewrite '' sentinels from v0.0.3 to NULL so the new read API +-- (GetRun, ListRuns, DeleteRun) can find them via "org_id IS NULL" +-- and the partial indexes (WHERE org_id IS NOT NULL) cover them +-- correctly. New rows never insert '' for these columns. +UPDATE {{.Schema}}.workflow_runs SET org_id = NULL WHERE org_id = ''; +UPDATE {{.Schema}}.workflow_runs SET initiated_by = NULL WHERE initiated_by = ''; + -- Claim loop orders queued runs by created_at. CREATE INDEX IF NOT EXISTS workflow_runs_status_created ON {{.Schema}}.workflow_runs (status, created_at); diff --git a/experimental/store/sqlite/queue.go b/experimental/store/sqlite/queue.go index f4720d9..836868d 100644 --- a/experimental/store/sqlite/queue.go +++ b/experimental/store/sqlite/queue.go @@ -169,10 +169,12 @@ func (s *Store) Complete(ctx context.Context, claim *worker.Claim, outcome worke return nil } -// ReclaimStale implements worker.QueueStore. +// ReclaimStale implements worker.QueueStore. Clears started_at so a +// reclaimed run has no stale start timestamp — it will be re-stamped +// on the next claim. func (s *Store) ReclaimStale(ctx context.Context, staleBefore time.Time, maxAttempts int, excludeIDs []string) (int, error) { query := `UPDATE workflow_runs - SET status = ?, claimed_by = '', heartbeat_at = NULL + SET status = ?, claimed_by = '', heartbeat_at = NULL, started_at = NULL WHERE status = ? AND heartbeat_at < ? AND attempt < ?` args := []any{ string(worker.StatusQueued), @@ -236,8 +238,8 @@ func (s *Store) DeadLetterStale(ctx context.Context, staleBefore time.Time, maxA return out, rows.Err() } -// ListFailedWithCredits implements worker.QueueStore. -func (s *Store) ListFailedWithCredits(ctx context.Context, limit int) ([]worker.FailedRun, error) { +// ListRefundPending implements worker.QueueStore. +func (s *Store) ListRefundPending(ctx context.Context, limit int) ([]worker.FailedRun, error) { if limit <= 0 { limit = 50 } diff --git a/experimental/store/sqlite/store.go b/experimental/store/sqlite/store.go index 2c556d6..3bfd58f 100644 --- a/experimental/store/sqlite/store.go +++ b/experimental/store/sqlite/store.go @@ -22,6 +22,25 @@ const timeFormat = "2006-01-02T15:04:05.000+00:00" // Store is a SQLite-backed implementation of the worker QueueStore // and the workflow engine's persistence interfaces. +// +// # Coexistence with consumer tables +// +// Unlike the postgres store, sqlite has no schema namespacing +// escape hatch — SQLite does not support schemas, and the library +// tables (workflow_runs, workflow_step_progress, workflow_events, +// etc.) are always unqualified. A consumer that uses SQLite for +// its own persistence and needs those same table names for its +// own data (or is worried about a future name collision) should +// hand the library a **dedicated *sql.DB** pointing at a separate +// .db file: +// +// libDB, _ := sql.Open("sqlite", "library_workflow.db") +// store := sqlite.New(libDB) +// +// appDB, _ := sql.Open("sqlite", "app.db") // your own tables +// +// Alternatively, use `ATTACH DATABASE 'app.db' AS app` on the +// library DB and scope every consumer query with `app.`. type Store struct { db *sql.DB logger *slog.Logger diff --git a/experimental/worker/credits.go b/experimental/worker/credits.go index b58e133..3478e26 100644 --- a/experimental/worker/credits.go +++ b/experimental/worker/credits.go @@ -8,7 +8,7 @@ import "context" // double-refund. // // CreditStore is pure ledger: listing which failed runs still need -// a refund is a QueueStore concern (ListFailedWithCredits) because +// a refund is a QueueStore concern (ListRefundPending) because // it joins the ledger against run status. type CreditStore interface { // Debit records a credit charge for a run. Idempotent: calling diff --git a/experimental/worker/handler.go b/experimental/worker/handler.go index 8d469fd..133bc48 100644 --- a/experimental/worker/handler.go +++ b/experimental/worker/handler.go @@ -15,17 +15,27 @@ import ( // worker's Config.Stores factory is set and returns a non-nil // value for that concern. A Handler backed by an in-memory engine // can ignore the store fields entirely. +// +// Lease fencing is **only** applied to Checkpointer. The other +// store fields are either append-only (ProgressStore, +// ActivityLogger) or globally shared (SignalStore), so they do +// not need to fence on (WorkerID, Attempt). Concrete stores may +// accept a *Claim in their factory method for symmetry and +// ignore it — that is expected, not a bug. type HandlerContext struct { // Claim is the run being executed. Claim *Claim // Checkpointer is a lease-fenced workflow.Checkpointer scoped // to this claim. Writes that fail the (WorkerID, Attempt) fence - // return worker.ErrLeaseLost. + // return worker.ErrLeaseLost. This is the only store in the + // bundle that enforces lease fencing. Checkpointer workflow.Checkpointer - // ProgressStore is a lease-fenced workflow.StepProgressStore - // scoped to this claim. + // ProgressStore is a workflow.StepProgressStore scoped to this + // claim. Step progress writes are derived observability data + // and are not fenced — the "latest update wins" semantics mean + // a stale writer cannot corrupt durable state. ProgressStore workflow.StepProgressStore // ActivityLogger is a workflow.ActivityLogger scoped to this @@ -33,7 +43,7 @@ type HandlerContext struct { ActivityLogger workflow.ActivityLogger // SignalStore is a workflow.SignalStore for signal/wait - // coordination. Shared across runs; not fenced. + // coordination. Shared across runs and claims; not fenced. SignalStore workflow.SignalStore } @@ -45,9 +55,28 @@ type HandlerContext struct { // Each method returns nil when the factory does not support that // concern; the worker propagates the nil into HandlerContext so // the Handler can check for availability. +// +// Of the three factory methods, **only NewCheckpointer is expected +// to return a lease-fenced store**. NewStepProgressStore and +// NewActivityLogger take a *Claim for API symmetry, but the +// returned implementations typically ignore the claim and return +// the shared store unchanged: step progress is write-through +// observability and activity logs are append-only, so neither +// needs fencing. SignalStore is not part of this factory at all — +// it is shared across claims and lives on Config directly. type HandlerStores interface { + // NewCheckpointer returns a lease-fenced checkpointer whose + // writes must return worker.ErrLeaseLost when the claim's + // (WorkerID, Attempt) pair no longer owns the run. NewCheckpointer(claim *Claim) workflow.Checkpointer + + // NewStepProgressStore returns a step progress store for the + // claim. The claim is usually ignored; lease fencing is not + // required because writes are idempotent replacements. NewStepProgressStore(claim *Claim) workflow.StepProgressStore + + // NewActivityLogger returns an activity logger for the claim. + // The claim is usually ignored; activity logs are append-only. NewActivityLogger(claim *Claim) workflow.ActivityLogger } diff --git a/experimental/worker/memstore/memstore.go b/experimental/worker/memstore/memstore.go index 55e4977..616181a 100644 --- a/experimental/worker/memstore/memstore.go +++ b/experimental/worker/memstore/memstore.go @@ -272,11 +272,11 @@ func (s *Store) DeadLetterStale(_ context.Context, staleBefore time.Time, maxAtt return out, nil } -// ListFailedWithCredits implements worker.QueueStore. The in-memory +// ListRefundPending implements worker.QueueStore. The in-memory // store has no credit ledger, so we return every failed run with a // non-zero CreditCost — the caller is expected to check HasRefund // via the CreditStore before issuing a refund. -func (s *Store) ListFailedWithCredits(_ context.Context, limit int) ([]worker.FailedRun, error) { +func (s *Store) ListRefundPending(_ context.Context, limit int) ([]worker.FailedRun, error) { s.mu.Lock() defer s.mu.Unlock() var out []worker.FailedRun diff --git a/experimental/worker/queue_store.go b/experimental/worker/queue_store.go index c7e1b3e..4423736 100644 --- a/experimental/worker/queue_store.go +++ b/experimental/worker/queue_store.go @@ -181,7 +181,7 @@ type DeadLetteredRun struct { } // FailedRun is a credit-tracking failed run returned by -// ListFailedWithCredits. Used by the reconcile loop as a backstop +// ListRefundPending. Used by the reconcile loop as a backstop // for DeadLetterStale's inline refund. type FailedRun struct { ID string @@ -244,11 +244,11 @@ type QueueStore interface { // this to emit observability events and refund credits inline. DeadLetterStale(ctx context.Context, staleBefore time.Time, maxAttempts int, excludeIDs []string) ([]DeadLetteredRun, error) - // ListFailedWithCredits returns failed runs that were debited + // ListRefundPending returns failed runs that were debited // but have not yet been refunded. Used by the credit reconcile // loop as a backstop for DeadLetterStale's inline refund. // // Implementations that do not track credits can return an empty // slice — the reconcile loop will do nothing. - ListFailedWithCredits(ctx context.Context, limit int) ([]FailedRun, error) + ListRefundPending(ctx context.Context, limit int) ([]FailedRun, error) } diff --git a/experimental/worker/runquery/runquery.go b/experimental/worker/runquery/runquery.go new file mode 100644 index 0000000..940d335 --- /dev/null +++ b/experimental/worker/runquery/runquery.go @@ -0,0 +1,122 @@ +// Package runquery defines the backend-neutral read API for +// workflow runs. Operational UIs and dashboards consume this +// package and any QueueStore implementation that satisfies +// runquery.Store — avoiding a direct import of a concrete store +// package such as experimental/store/postgres. +// +// The types here are intentionally minimal and stdlib-only: a +// row shape (Run), a filter (RunFilter), a keyset cursor +// (RunCursor), two sentinel errors, and the Store interface +// itself. Concrete store packages implement runquery.Store on +// their *Store type. +package runquery + +import ( + "context" + "errors" + "time" + + "github.com/deepnoodle-ai/workflow/experimental/worker" +) + +// ErrRunNotFound is returned by Store.GetRun and Store.DeleteRun +// when no row matches the requested ID (within the given org +// scope). +var ErrRunNotFound = errors.New("runquery: run not found") + +// ErrCannotDeleteRunning is returned by Store.DeleteRun when +// asked to delete a run whose status is worker.StatusRunning. +// Callers must cancel or wait for the run to complete before +// deleting it. +var ErrCannotDeleteRunning = errors.New("runquery: cannot delete a running run") + +// Run is a read-side view of a workflow_runs row, suitable for +// operational dashboards and run management surfaces. It mirrors +// the queue-side worker.NewRun / worker.Claim but also exposes +// the lifecycle columns (status, attempt, timestamps) that only +// exist on persisted runs. +type Run struct { + ID string + OrgID string + ProjectID string + ParentRunID string + WorkflowType string + Status worker.Status + CreditCost int + InitiatedBy string + CallbackURL string + Spec []byte + Result []byte + ErrorMessage string + Metadata map[string]string + Attempt int + ClaimedBy string + HeartbeatAt *time.Time + CreatedAt time.Time + StartedAt *time.Time + CompletedAt *time.Time +} + +// RunFilter narrows a ListRuns or CountRuns query. Zero-value +// fields are ignored. Filter values are ANDed together. +type RunFilter struct { + // WorkflowType matches exactly when non-empty. + WorkflowType string + + // Status matches exactly when non-empty. + Status worker.Status + + // ProjectID matches exactly when non-empty. + ProjectID string + + // ParentRunID matches exactly when non-empty. + ParentRunID string + + // InitiatedBy matches exactly when non-empty. + InitiatedBy string + + // Metadata is evaluated with map containment: every key in + // the filter must be present with an equal value on the row. + // An empty map matches all rows. + Metadata map[string]string + + // Cursor is the pagination anchor. Nil means "start from the + // newest row." + Cursor *RunCursor + + // Limit caps the number of rows returned. Zero means the + // store's default page size. + Limit int +} + +// RunCursor is an opaque keyset pagination anchor. Consumers +// typically base64-encode it for API transport and decode back +// to this shape on the next call. +type RunCursor struct { + CreatedAt time.Time + ID string +} + +// Store is the backend-neutral read API for workflow runs. +// Implemented by concrete store packages (e.g., postgres). +// +// The orgID parameter scopes multi-tenant queries. Empty string +// matches rows with NULL org_id (single-tenant deployments); +// non-empty matches rows with that exact org_id. +type Store interface { + // GetRun returns a single run by ID. + GetRun(ctx context.Context, orgID, id string) (*Run, error) + + // ListRuns returns a page of runs matching filter, ordered + // newest-first, along with a cursor for the next page. The + // returned cursor is nil when no more rows exist. + ListRuns(ctx context.Context, orgID string, filter RunFilter) ([]*Run, *RunCursor, error) + + // CountRuns returns the total number of rows matching filter. + // The cursor field on filter is ignored. + CountRuns(ctx context.Context, orgID string, filter RunFilter) (int, error) + + // DeleteRun removes a run row by ID. Running runs cannot be + // deleted; ErrCannotDeleteRunning is returned in that case. + DeleteRun(ctx context.Context, orgID, id string) error +} diff --git a/experimental/worker/subsystems.go b/experimental/worker/subsystems.go index 3e19d70..3ace3ec 100644 --- a/experimental/worker/subsystems.go +++ b/experimental/worker/subsystems.go @@ -223,7 +223,7 @@ func (w *Worker) reconcileLoop(ctx context.Context) { } func (w *Worker) reconcileCredits(ctx context.Context) { - failed, err := w.store.ListFailedWithCredits(ctx, 50) + failed, err := w.store.ListRefundPending(ctx, 50) if err != nil { w.cfg.Logger.Error("list failed with credits", "error", err) return From bc68f29afdd5d64c2fdf6a0e339ffa4caa52fb7d Mon Sep 17 00:00:00 2001 From: Curtis Myzie Date: Sun, 12 Apr 2026 21:57:22 -0400 Subject: [PATCH 2/2] =?UTF-8?q?fix:=20address=20PR=20#38=20review=20nits?= =?UTF-8?q?=20=E2=80=94=20memstore=20sort,=20run=20projection,=20drain=20l?= =?UTF-8?q?oop?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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) --- .../store/postgres/integration_test.go | 11 ++-- experimental/store/postgres/runs.go | 36 +++++-------- experimental/worker/memstore/memstore.go | 53 ++++++++++++++----- 3 files changed, 62 insertions(+), 38 deletions(-) diff --git a/experimental/store/postgres/integration_test.go b/experimental/store/postgres/integration_test.go index 0cf18c9..3739546 100644 --- a/experimental/store/postgres/integration_test.go +++ b/experimental/store/postgres/integration_test.go @@ -183,10 +183,13 @@ func TestStore_ListRefundPending(t *testing.T) { WorkflowType: "free", }) - for _, id := range []string{"needs-refund", "already-refunded", "no-cost"} { - claim, _ := store.ClaimQueued(ctx, "w") - _ = claim - _ = id + // Drain the queue so every enqueued run is in StatusRunning before + // we force them to failed below. + for { + claim, err := store.ClaimQueued(ctx, "w") + if err != nil || claim == nil { + break + } } // Force all three to failed. diff --git a/experimental/store/postgres/runs.go b/experimental/store/postgres/runs.go index ac9d82e..ec13861 100644 --- a/experimental/store/postgres/runs.go +++ b/experimental/store/postgres/runs.go @@ -35,6 +35,16 @@ type RunCursor = runquery.RunCursor // Compile-time check that *Store implements the read-side contract. var _ runquery.Store = (*Store)(nil) +// runProjection is the column list used by GetRun and ListRuns. It +// must stay in lock-step with the Scan order in scanRun — adding or +// reordering a column requires touching both. +const runProjection = `id, + COALESCE(org_id, ''), COALESCE(project_id, ''), COALESCE(parent_run_id, ''), + workflow_type, status, credit_cost, COALESCE(initiated_by, ''), callback_url, + spec, result, error_message, metadata, + attempt, claimed_by, + heartbeat_at, created_at, started_at, completed_at` + // GetRun returns a single run by ID, scoped to orgID. An empty // orgID matches rows with NULL org_id (single-tenant). Returns // runquery.ErrRunNotFound when no matching row exists. @@ -50,16 +60,8 @@ func (s *Store) GetRun(ctx context.Context, orgID, id string) (*runquery.Run, er where = "id = $1 AND org_id = $2" args = append(args, orgID) } - query := fmt.Sprintf(` - SELECT id, - COALESCE(org_id, ''), COALESCE(project_id, ''), COALESCE(parent_run_id, ''), - workflow_type, status, credit_cost, COALESCE(initiated_by, ''), callback_url, - spec, result, error_message, metadata, - attempt, claimed_by, - heartbeat_at, created_at, started_at, completed_at - FROM %s - WHERE %s - `, s.t("workflow_runs"), where) + query := fmt.Sprintf(`SELECT %s FROM %s WHERE %s`, + runProjection, s.t("workflow_runs"), where) row := s.pool.QueryRow(ctx, query, args...) r, err := scanRun(row) if err != nil { @@ -92,18 +94,8 @@ func (s *Store) ListRuns(ctx context.Context, orgID string, filter runquery.RunF // Fetch limit+1 so we can decide whether there is a next page // without a second query. args = append(args, limit+1) - query := fmt.Sprintf(` - SELECT id, - COALESCE(org_id, ''), COALESCE(project_id, ''), COALESCE(parent_run_id, ''), - workflow_type, status, credit_cost, COALESCE(initiated_by, ''), callback_url, - spec, result, error_message, metadata, - attempt, claimed_by, - heartbeat_at, created_at, started_at, completed_at - FROM %s - %s - ORDER BY created_at DESC, id DESC - LIMIT $%d - `, s.t("workflow_runs"), where, len(args)) + query := fmt.Sprintf(`SELECT %s FROM %s %s ORDER BY created_at DESC, id DESC LIMIT $%d`, + runProjection, s.t("workflow_runs"), where, len(args)) rows, err := s.pool.Query(ctx, query, args...) if err != nil { diff --git a/experimental/worker/memstore/memstore.go b/experimental/worker/memstore/memstore.go index 616181a..b39aaf3 100644 --- a/experimental/worker/memstore/memstore.go +++ b/experimental/worker/memstore/memstore.go @@ -279,7 +279,11 @@ func (s *Store) DeadLetterStale(_ context.Context, staleBefore time.Time, maxAtt func (s *Store) ListRefundPending(_ context.Context, limit int) ([]worker.FailedRun, error) { s.mu.Lock() defer s.mu.Unlock() - var out []worker.FailedRun + type candidate struct { + run worker.FailedRun + completedAt time.Time + } + var matches []candidate for _, row := range s.runs { if row.status != worker.StatusFailed { continue @@ -287,25 +291,50 @@ func (s *Store) ListRefundPending(_ context.Context, limit int) ([]worker.Failed if row.creditCost <= 0 { continue } - out = append(out, worker.FailedRun{ - ID: row.id, - OrgID: row.orgID, - WorkflowType: row.workflowType, - CreditCost: row.creditCost, + matches = append(matches, candidate{ + run: worker.FailedRun{ + ID: row.id, + OrgID: row.orgID, + WorkflowType: row.workflowType, + CreditCost: row.creditCost, + }, + completedAt: row.completedAt, }) - if limit > 0 && len(out) >= limit { - break - } } - slices.SortFunc(out, func(a, b worker.FailedRun) int { - if a.ID < b.ID { + // Match the postgres/sqlite ORDER BY completed_at ASC NULLS LAST, + // then id for tie-breaking. Sort the full set first, then apply + // the limit — applying the limit during map iteration would yield + // nondeterministic results. + slices.SortFunc(matches, func(a, b candidate) int { + aZero, bZero := a.completedAt.IsZero(), b.completedAt.IsZero() + switch { + case aZero && !bZero: + return 1 + case !aZero && bZero: + return -1 + case !aZero && !bZero: + if a.completedAt.Before(b.completedAt) { + return -1 + } + if a.completedAt.After(b.completedAt) { + return 1 + } + } + if a.run.ID < b.run.ID { return -1 } - if a.ID > b.ID { + if a.run.ID > b.run.ID { return 1 } return 0 }) + if limit > 0 && len(matches) > limit { + matches = matches[:limit] + } + out := make([]worker.FailedRun, len(matches)) + for i, m := range matches { + out[i] = m.run + } return out, nil }