From 3cd056ab5d285ed659bdf20c8a6f9fc8a26bac8c Mon Sep 17 00:00:00 2001 From: Michael McQuade Date: Mon, 19 Oct 2026 09:06:42 -0500 Subject: [PATCH 01/19] fix(driver): find running jobs by primary key whatever the statistics say --- driver/internal/pgsql/jobs.go | 23 ++++++++++++++++++----- 1 file changed, 18 insertions(+), 5 deletions(-) diff --git a/driver/internal/pgsql/jobs.go b/driver/internal/pgsql/jobs.go index 992b999..d9fe954 100644 --- a/driver/internal/pgsql/jobs.go +++ b/driver/internal/pgsql/jobs.go @@ -575,6 +575,15 @@ func claimRows(ctx context.Context, conn Conn, query string, args ...any) ([]*dr // and workflow dependents. Error timestamps come from database time and // awaited jobs announce themselves on the done channel. An attempt's log // is appended to the job's, which keep the latest $13 attempts. +// +// The running state is compared as text, which no partial index's +// predicate matches, so the results always find their rows by primary key. +// Compared as the enum, it lets the planner scan hopper_jobs_running +// instead, which statistics taken while few jobs ran (right after a backlog +// was inserted, say) make look like a page or two: the statement then +// compared every running row with every result, and slowed down as +// unfinalized results piled up as running rows. runningSQL is the same +// check for statements that find one running job by ID. var jobFinalizeHeadSQL = ` WITH RECURSIVE r AS ( SELECT * FROM unnest( @@ -596,20 +605,24 @@ retry AS ( THEN r.snooze_reason END, wake_requested_at = NULL FROM r - WHERE j.id = r.id AND j.state = 'running' AND j.attempted_by = r.attempted_by AND j.fence = r.fence + WHERE j.id = r.id AND j.state::text = 'running' AND j.attempted_by = r.attempted_by AND j.fence = r.fence AND r.state IN ('retryable', 'scheduled') RETURNING j.id ), finished AS ( DELETE FROM {{schema}}.hopper_jobs j USING r - WHERE j.id = r.id AND j.state = 'running' AND j.attempted_by = r.attempted_by AND j.fence = r.fence + WHERE j.id = r.id AND j.state::text = 'running' AND j.attempted_by = r.attempted_by AND j.fence = r.fence AND r.state IN ('completed', 'canceled', 'discarded') RETURNING ` + finishedColumns(`CASE WHEN r.error IS NULL THEN j.errors ELSE j.errors || jsonb_set(r.error, '{at}', to_jsonb(now())) END`, appendLogSQL, "r.output", "r.archive", "r.state", "$9") + ` ), ` +// runningSQL is the running check of jobFinalizeHeadSQL for a statement +// whose hopper_jobs row needs no alias. +const runningSQL = "state::text = 'running'" + var jobFinalizeSQL = jobFinalizeHeadSQL + finalizeTailSQL("$9", "$10") + ` SELECT id FROM retry UNION ALL SELECT id FROM finished` @@ -761,7 +774,7 @@ SET progress = jsonb_build_object( 'step', CASE WHEN $5 THEN coalesce(progress->>'step', '') ELSE $4::text END, 'at', statement_timestamp()) || CASE WHEN $6::jsonb IS NULL THEN '{}'::jsonb ELSE jsonb_build_object('cursor', $6::jsonb) END -WHERE id = $1 AND state = 'running' AND attempted_by = $2 AND fence = $3 AND attempt = $7%s +WHERE id = $1 AND ` + runningSQL + ` AND attempted_by = $2 AND fence = $3 AND attempt = $7%s RETURNING progress::text` var ( @@ -769,7 +782,7 @@ var ( jobRecordProgressNoWaitQuery = fmt.Sprintf(jobRecordProgressSQL, ` WITH locked AS ( SELECT id FROM {{schema}}.hopper_jobs - WHERE id = $1 AND state = 'running' AND attempted_by = $2 AND fence = $3 AND attempt = $7 + WHERE id = $1 AND `+runningSQL+` AND attempted_by = $2 AND fence = $3 AND attempt = $7 FOR UPDATE NOWAIT )`, " AND id IN (SELECT id FROM locked)") ) @@ -1502,7 +1515,7 @@ const jobLockStateSQL = "SELECT state FROM {{schema}}.hopper_jobs WHERE id = $1 var ( jobCancelRunningSQL = fmt.Sprintf(` UPDATE {{schema}}.hopper_jobs SET cancel_requested_at = coalesce(cancel_requested_at, now()) -WHERE id = $1 AND state = 'running' +WHERE id = $1 AND `+runningSQL+` RETURNING %s`, JobColumns("")) // jobCancelWaitingSQL finalizes a waiting job as canceled, with // everything that entails for its batch and its dependents. From 3b9e3ae7bea333198a28e59d41add4930b82dc71 Mon Sep 17 00:00:00 2001 From: Michael McQuade Date: Mon, 19 Oct 2026 09:06:42 -0500 Subject: [PATCH 02/19] fix(driver): probe indexes for the finalize tail's and promotion's lookups --- driver/internal/pgsql/workflow.go | 27 ++++++++++++++++++++------- 1 file changed, 20 insertions(+), 7 deletions(-) diff --git a/driver/internal/pgsql/workflow.go b/driver/internal/pgsql/workflow.go index 592f6a0..6195087 100644 --- a/driver/internal/pgsql/workflow.go +++ b/driver/internal/pgsql/workflow.go @@ -112,6 +112,16 @@ const plainFinalizeTailSQL = `done AS (SELECT * FROM finished), // are announced on the done channel, and promoted jobs and callbacks on // the insert channel. Jobs without a batch or dependents cost index probes // on small tables and nothing else. +// +// Every lookup keyed by the statement's own rows (the cascade, done, the +// batches counted and those completed) is a = ANY (ARRAY(...)), which the +// planner takes for ten keys and so probes an index for. Joined to the CTEs +// instead, it guessed hundreds of rows for them, and with no statistics on +// hopper_job_deps a probe looked like a few hundred rows more, so it scanned +// hopper_job_deps, hopper_jobs and hopper_batches whole on every finalize, +// at a cost that grew with the workflows in flight. The recursive step of +// the cascade cannot take that form, and relies on the leader keeping +// hopper_job_deps analyzed (JobsMaintain). func finalizeTailSQL(doneChannelParam, insertChannelParam string) string { return `cascade AS ( SELECT d.job_id FROM {{schema}}.hopper_job_deps d JOIN finished f ON d.depends_on = f.id @@ -121,8 +131,8 @@ func finalizeTailSQL(doneChannelParam, insertChannelParam string) string { WHERE d.on_failure = 'cancel' ), cascaded AS ( - DELETE FROM {{schema}}.hopper_jobs j USING (SELECT DISTINCT job_id FROM cascade) c - WHERE j.id = c.job_id AND j.state = 'pending' + DELETE FROM {{schema}}.hopper_jobs j + WHERE j.id = ANY (ARRAY(SELECT job_id FROM cascade)) AND j.state = 'pending' RETURNING ` + finishedColumns(appendErrorSQL("hopper: dependency failed"), "j.logs", "NULL::jsonb", "true", "'canceled'::text", doneChannelParam) + ` ), done AS (SELECT * FROM finished UNION ALL SELECT * FROM cascaded), @@ -134,7 +144,7 @@ batch_counts AS ( ready AS ( SELECT b.id, b.failed + p.failed + c.f AS failed FROM batch_counts c JOIN {{schema}}.hopper_batches b ON b.id = c.batch_id, ` + batchProgressSQL + ` - WHERE b.completed_at IS NULL AND b.pending - p.done - c.n <= 0 + WHERE b.id = ANY (ARRAY(SELECT batch_id FROM batch_counts)) AND b.completed_at IS NULL AND b.pending - p.done - c.n <= 0 ), batches AS ( UPDATE {{schema}}.hopper_batches b SET completed_at = now(), pending = 0, failed = r.failed @@ -147,14 +157,14 @@ progress AS ( RETURNING batch_id ), progress_cleared AS ( - DELETE FROM {{schema}}.hopper_batch_progress p WHERE p.batch_id IN (SELECT id FROM batches) RETURNING p.batch_id + DELETE FROM {{schema}}.hopper_batch_progress p WHERE p.batch_id = ANY (ARRAY(SELECT id FROM batches)) RETURNING p.batch_id ), ` + batchCallbacksSQL(insertChannelParam) + `, promoted AS ( UPDATE {{schema}}.hopper_jobs j SET state = CASE WHEN j.scheduled_at > now() THEN 'scheduled' ELSE 'available' END::{{schema}}.hopper_job_state WHERE j.state = 'pending' - AND j.id IN (SELECT d.job_id FROM {{schema}}.hopper_job_deps d JOIN done ON d.depends_on = done.id) + AND j.id = ANY (ARRAY(SELECT d.job_id FROM {{schema}}.hopper_job_deps d WHERE d.depends_on = ANY (ARRAY(SELECT id FROM done)))) AND j.id NOT IN (SELECT id FROM cascaded) AND NOT EXISTS ( SELECT 1 FROM {{schema}}.hopper_job_deps d JOIN {{schema}}.hopper_jobs x ON x.id = d.depends_on @@ -162,7 +172,7 @@ promoted AS ( RETURNING pg_notify(` + insertChannelParam + `, j.queue) ), edges AS ( - DELETE FROM {{schema}}.hopper_job_deps d WHERE d.job_id IN (SELECT id FROM done) RETURNING d.job_id + DELETE FROM {{schema}}.hopper_job_deps d WHERE d.job_id = ANY (ARRAY(SELECT id FROM done)) RETURNING d.job_id )` } @@ -170,12 +180,15 @@ edges AS ( // candidates selects none of whose dependencies is still live, skipping rows // a finalize holds: that finalize decides them. It returns how many it // promoted, how many candidates there were, and the greatest candidate ID. +// The candidates are looked up as finalizeTailSQL looks up its own rows: +// joined instead, a sweep's LIMIT $2, which a generic plan takes for a +// tenth of the edges, had it scan hopper_jobs whole for every chunk. func workflowPromoteSQL(candidates string) string { return ` WITH candidates AS (` + candidates + `), ready AS ( SELECT j.id FROM {{schema}}.hopper_jobs j - WHERE j.id IN (SELECT id FROM candidates) + WHERE j.id = ANY (ARRAY(SELECT id FROM candidates)) AND j.state = 'pending' AND NOT EXISTS ( SELECT 1 FROM {{schema}}.hopper_job_deps d JOIN {{schema}}.hopper_jobs x ON x.id = d.depends_on From 3b0de87783e3cde1265b0ce7452234d9d112b8d1 Mon Sep 17 00:00:00 2001 From: Michael McQuade Date: Mon, 19 Oct 2026 09:06:42 -0500 Subject: [PATCH 03/19] fix(driver): keep hopper_job_deps analyzed with the live tables --- driver/driver.go | 7 ++++--- driver/internal/pgsql/cluster.go | 10 ++++++++-- 2 files changed, 12 insertions(+), 5 deletions(-) diff --git a/driver/driver.go b/driver/driver.go index 837c8f9..d1c15ab 100644 --- a/driver/driver.go +++ b/driver/driver.go @@ -330,9 +330,10 @@ type Executor interface { // that have left, so pickup latency climbs between vacuums; autovacuum // looks only every autovacuum_naptime (a minute by default). On Postgres // the live tags table (schema v17), whose index a listing by tag walks, - // is kept the same way, and each table is kept whether or not the - // other's maintenance fails. Engines without such maintenance return - // zero values. + // and the workflow dependency table, whose statistics the finalize + // statement's plan rests on, are kept the same way, and each table is + // kept whether or not another's maintenance fails. Engines without such + // maintenance return zero values. JobsMaintain(ctx context.Context) (JobsMaintainResult, error) } diff --git a/driver/internal/pgsql/cluster.go b/driver/internal/pgsql/cluster.go index 6779d2d..6c94b6b 100644 --- a/driver/internal/pgsql/cluster.go +++ b/driver/internal/pgsql/cluster.go @@ -399,7 +399,11 @@ const analyzeEvery = 30 * time.Second // // The live tags table churns with tagged jobs, and a listing by tag walks // its index, so it is kept the same way, except that its analyze keeps the -// server's statistics target. Each table is kept whether or not the other's +// server's statistics target. So does the dependency table, which churns +// with workflows: until it has statistics, the planner takes each lookup of +// a job's edges for a two-hundredth of the table, and scans the table +// instead, in finalizes and in the sweep that promotes dependents (see +// finalizeTailSQL). Each table is kept whether or not another's // maintenance fails, such as a vacuum of a large hopper_jobs that runs past // the role's statement_timeout on every pass. func (e *Executor) JobsMaintain(ctx context.Context) (driver.JobsMaintainResult, error) { @@ -424,13 +428,15 @@ func (e *Executor) JobsMaintain(ctx context.Context) (driver.JobsMaintainResult, // their analyze uses, or zero for the server's. The tags table keeps the // server's target: a listing by several tags (rarestTagSQL) tells common // tags apart by its most common values, of which a target of 10 would keep -// only ten. +// only ten. So does the dependency table, whose estimates rest on how many +// distinct jobs its sample holds, which a larger sample counts better. var liveTables = []struct { name string statisticsTarget int }{ {"hopper_jobs", 10}, {"hopper_job_tags", 0}, + {"hopper_job_deps", 0}, } func (e *Executor) maintainLiveTable(ctx context.Context, table string, statisticsTarget int) (vacuumed, analyzed bool, err error) { From 32b908edb95d38b425dab366785f02b6f0c399ae Mon Sep 17 00:00:00 2001 From: Michael McQuade Date: Mon, 19 Oct 2026 09:06:42 -0500 Subject: [PATCH 04/19] test(driver): finalize plans under stale statistics, workflow chaos, dependency table maintenance --- driver/hopperpgx/postgres_test.go | 149 +++++++++++++++++++++++ drivertest/cluster.go | 43 +++++++ drivertest/drivertest.go | 2 + drivertest/workflow.go | 190 ++++++++++++++++++++++++++++++ 4 files changed, 384 insertions(+) diff --git a/driver/hopperpgx/postgres_test.go b/driver/hopperpgx/postgres_test.go index fca11aa..63ab391 100644 --- a/driver/hopperpgx/postgres_test.go +++ b/driver/hopperpgx/postgres_test.go @@ -1201,3 +1201,152 @@ func TestRescueCandidatesWithoutAttemptTime(t *testing.T) { t.Fatalf("candidates = %+v, %v; want the row, not stuck", got, err) } } + +// TestFinalizePlansIgnoreStaleStatistics explains, with generic and custom +// plans, the statements that finalize jobs, follow a flush up, and find a +// running job by ID, under the statistics a backlog of workflows has once +// it starts running: the leader analyzed it as it was inserted, when +// nothing ran, and thousands of its jobs have been claimed since. The +// results' rows are found by primary key, not by a scan of +// hopper_jobs_running, which those statistics say is all but empty, and no +// statement scans hopper_jobs, hopper_job_deps or hopper_batches whole, +// which would cost every finalize more with each workflow in flight. +func TestFinalizePlansIgnoreStaleStatistics(t *testing.T) { + t.Parallel() + ctx, d, exec := setup(t) + pool := d.Pool() + exec1 := func(q string) { + t.Helper() + if _, err := pool.Exec(ctx, q); err != nil { + t.Fatalf("%s: %v", q, err) + } + } + for _, table := range []string{"hopper_jobs", "hopper_job_deps", "hopper_batches", "hopper_batch_progress"} { + exec1("ALTER TABLE " + table + " SET (autovacuum_enabled = false)") + } + // Five thousand workflows of ten steps: one, eight after it, one after + // those. + exec1(`CREATE TEMP TABLE steps AS + SELECT w, s, hopper_uuidv7() AS id, b.id AS batch_id + FROM (SELECT w, hopper_uuidv7() AS id FROM generate_series(1, 5000) w) b, generate_series(0, 9) s ORDER BY w, s`) + exec1(`INSERT INTO hopper_batches (id, pending, total, name) SELECT DISTINCT batch_id, 10, 10, 'w' FROM steps`) + exec1(`INSERT INTO hopper_jobs (id, kind, state, args, batch_id) + SELECT id, 'step', (CASE WHEN s = 0 THEN 'available' ELSE 'pending' END)::hopper_job_state, '{}', batch_id FROM steps ORDER BY w, s`) + exec1(`INSERT INTO hopper_job_deps (job_id, depends_on) + SELECT m.id, r.id FROM steps m JOIN steps r ON r.w = m.w AND r.s = 0 WHERE m.s BETWEEN 1 AND 8 + UNION ALL SELECT l.id, m.id FROM steps l JOIN steps m ON m.w = l.w AND m.s BETWEEN 1 AND 8 WHERE l.s = 9`) + analyzed := func() bool { + var n int + if err := pool.QueryRow(ctx, `SELECT count(*) FROM pg_stat_user_tables + WHERE relid IN ('hopper_jobs'::regclass, 'hopper_job_deps'::regclass) AND last_analyze IS NOT NULL`).Scan(&n); err != nil { + t.Fatal(err) + } + return n == 2 + } + // Postgres 14's statistics collector reports the inserts a moment later. + for deadline := time.Now().Add(10 * time.Second); !analyzed(); time.Sleep(100 * time.Millisecond) { + if time.Now().After(deadline) { + t.Fatal("the leader's maintenance did not analyze hopper_jobs and hopper_job_deps") + } + if _, err := exec.JobsMaintain(ctx); err != nil { + t.Fatal(err) + } + } + clientID := register(t, ctx, exec) + var running []*driver.JobRow + for len(running) < 2000 { + res, err := exec.JobClaim(ctx, driver.JobClaimParams{Queue: "default", ClientID: clientID, Limit: 500}) + if err != nil || len(res.Jobs) == 0 { + t.Fatalf("claim = %d jobs, %v", len(res.Jobs), err) + } + running = append(running, res.Jobs...) + } + results := func(jobs []*driver.JobRow) []driver.JobFinalize { + out := make([]driver.JobFinalize, len(jobs)) + for i, j := range jobs { + out[i] = driver.JobFinalize{ID: j.ID, AttemptedBy: clientID, Fence: j.Fence, State: driver.JobStateCompleted, Archive: true} + } + return out + } + + log := &statementLog{} + traced := tracedDriver(t, pool, log).Executor() + finished, err := traced.JobFinalizeMany(ctx, driver.JobFinalizeParams{Jobs: results(running[:100])}) + if err != nil || len(finished) != 100 { + t.Fatalf("finalize = %d, %v", len(finished), err) + } + if _, err := traced.JobFinalizeMany(ctx, driver.JobFinalizeParams{Jobs: results(running[100:200]), NoBatches: true}); err != nil { + t.Fatal(err) + } + if _, err := traced.WorkflowPromoteReady(ctx, driver.WorkflowPromoteParams{DependsOn: finished}); err != nil { + t.Fatal(err) + } + if _, err := traced.WorkflowPromoteReady(ctx, driver.WorkflowPromoteParams{Limit: 1000}); err != nil { + t.Fatal(err) + } + j := running[200] + if _, err := traced.JobRecordProgress(ctx, driver.JobRecordProgressParams{ID: j.ID, AttemptedBy: clientID, Fence: j.Fence, Attempt: j.Attempt, Step: "s"}); err != nil { + t.Fatal(err) + } + if _, err := traced.JobRecordProgress(ctx, driver.JobRecordProgressParams{ID: j.ID, AttemptedBy: clientID, Fence: j.Fence, Attempt: j.Attempt, Step: "s", NoWait: true}); err != nil { + t.Fatal(err) + } + if _, err := traced.JobCancel(ctx, running[201].ID); err != nil { + t.Fatal(err) + } + + conn, err := pool.Acquire(ctx) + if err != nil { + t.Fatal(err) + } + defer conn.Release() + explain := func(mode, query string, args []any) string { + t.Helper() + if _, err := conn.Exec(ctx, "SET plan_cache_mode = "+mode); err != nil { + t.Fatal(err) + } + if _, err := conn.Exec(ctx, "PREPARE explained AS "+query); err != nil { + t.Fatal(err) + } + defer conn.Exec(ctx, "DEALLOCATE explained") //nolint:errcheck // the connection's last use + var types []string + if err := conn.QueryRow(ctx, "SELECT parameter_types::text[] FROM pg_prepared_statements WHERE name = 'explained'").Scan(&types); err != nil { + t.Fatal(err) + } + literals := make([]string, len(args)) + for i, arg := range args { + if err := conn.QueryRow(ctx, "SELECT quote_nullable($1::"+types[i]+")", arg).Scan(&literals[i]); err != nil { + t.Fatalf("argument %d: %v", i+1, err) + } + literals[i] += "::" + types[i] + } + var plan string + if err := conn.QueryRow(ctx, "EXPLAIN (FORMAT JSON) EXECUTE explained("+strings.Join(literals, ", ")+")").Scan(&plan); err != nil { + t.Fatal(err) + } + return plan + } + log.mu.Lock() + statements, args := log.sql, log.args + log.mu.Unlock() + scan := regexp.MustCompile(`"Node Type": "Seq Scan",[^}]*?"Relation Name": "(hopper_jobs|hopper_job_deps|hopper_batches)"`) + explained := 0 + for i, q := range statements { + if !strings.Contains(q, "hopper_jobs") || strings.HasPrefix(strings.TrimSpace(q), "SELECT state FROM") { + continue + } + explained++ + for _, mode := range []string{"force_generic_plan", "force_custom_plan"} { + plan := explain(mode, q, args[i]) + if strings.Contains(plan, `"Index Name": "hopper_jobs_running"`) { + t.Errorf("%s plan scans hopper_jobs_running:\n%s\n%s", mode, q, plan) + } + if m := scan.FindStringSubmatch(plan); m != nil { + t.Errorf("%s plan scans %s whole:\n%s\n%s", mode, m[1], q, plan) + } + } + } + if explained != 7 { + t.Errorf("explained %d statements, want 7", explained) + } +} diff --git a/drivertest/cluster.go b/drivertest/cluster.go index 6a4f35e..e5434e7 100644 --- a/drivertest/cluster.go +++ b/drivertest/cluster.go @@ -1110,6 +1110,49 @@ func testMaintain[TTx any](t *testing.T, f Fixture[TTx]) { } } +// testDepsTableMaintained checks that the leader's live table maintenance +// analyzes the dependency table once a burst of workflows has filled it: +// until the table has statistics, the planner takes each lookup of a job's +// edges for a two-hundredth of the table, and finalizes and the promotion +// sweep scan it instead. +func testDepsTableMaintained[TTx any](t *testing.T, f Fixture[TTx]) { + ctx := context.Background() + d := f.NewDriver(t) + exec := d.Executor() + f.Exec(ctx, t, d, "ALTER TABLE hopper_job_deps SET (autovacuum_enabled = false)") + deps := make([]driver.JobDependency, 1200) + for i := range deps { + deps[i] = driver.JobDependency{Job: i + 1, DependsOn: 0} + } + if _, err := exec.WorkflowInsert(ctx, driver.WorkflowInsertParams{Name: "fan-out", Jobs: params("step", len(deps)+1), Deps: deps}); err != nil { + t.Fatal(err) + } + stat := func(q string) string { + return f.QueryText(ctx, t, d, "SELECT ("+q+")::text FROM pg_stat_user_tables WHERE relid = 'hopper_job_deps'::regclass") + } + // Postgres 14's statistics collector reports the inserts a moment later. + for deadline := time.Now().Add(10 * time.Second); stat("n_mod_since_analyze >= 1200") != "true"; time.Sleep(100 * time.Millisecond) { + if time.Now().After(deadline) { + t.Fatal("the dependency table's inserts were not counted") + } + } + if stat("last_analyze IS NULL") != "true" { + t.Fatal("dependency table analyzed before maintenance") + } + res := waitMaintain(ctx, t, exec, func(r driver.JobsMaintainResult) bool { return r.Analyzed }) + if !res.Analyzed { + t.Fatalf("maintenance after 1200 edges = %+v", res) + } + for deadline := time.Now().Add(10 * time.Second); stat("last_analyze IS NOT NULL") != "true"; time.Sleep(100 * time.Millisecond) { + if time.Now().After(deadline) { + t.Fatal("dependency table not analyzed") + } + if _, err := exec.JobsMaintain(ctx); err != nil { + t.Fatal(err) + } + } +} + // analyzeEmpty gives the engine statistics that say the live table is // empty, as an idle installation has, through the Fixture when it can. func analyzeEmpty(ctx context.Context, t *testing.T, exec driver.Executor) { diff --git a/drivertest/drivertest.go b/drivertest/drivertest.go index be2551b..baa7f43 100644 --- a/drivertest/drivertest.go +++ b/drivertest/drivertest.go @@ -144,8 +144,10 @@ func Run[TTx any](t *testing.T, f Fixture[TTx]) { {"BatchesConcurrent", testBatchesConcurrent[TTx]}, {"Workflows", testWorkflows[TTx]}, {"WorkflowFanInConcurrent", testWorkflowFanInConcurrent[TTx]}, + {"WorkflowsConcurrent", testWorkflowsConcurrent[TTx]}, {"Streams", testStreams[TTx]}, {"Maintain", testMaintain[TTx]}, + {"DepsTableMaintained", testDepsTableMaintained[TTx]}, } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { diff --git a/drivertest/workflow.go b/drivertest/workflow.go index 281b4fd..ffbbf88 100644 --- a/drivertest/workflow.go +++ b/drivertest/workflow.go @@ -4,7 +4,10 @@ import ( "context" "encoding/json" "errors" + "fmt" + "strconv" "strings" + "sync" "testing" "time" @@ -322,3 +325,190 @@ func testWorkflowFanInConcurrent[TTx any](t *testing.T, f Fixture[TTx]) { } } } + +// testWorkflowsConcurrent works a hundred workflows (a step, four after it, +// one after those) with eight workers at once, each finalizing what it +// claimed in one statement and following up as a client does, while the +// leader's sweeps run and every statement's results are sent a second time. +// In one workflow in five, a middle step fails. Every job is finalized +// once, no step runs before its dependencies have finished, the step after +// a failed one is canceled, and every workflow completes once, with its +// failures counted. +func testWorkflowsConcurrent[TTx any](t *testing.T, f Fixture[TTx]) { + ctx := context.Background() + d := f.NewDriver(t) + exec := d.Executor() + const workflows, middles = 100, 4 + cb := &driver.JobInsertParams{Kind: "done", Queue: "callbacks", Priority: 2, MaxAttempts: 3, Args: json.RawMessage(`{}`)} + for w := range workflows { + jobs := params("root", 1) + mids := params("mid", middles) + if w%5 == 0 { + mids[0].Kind = "fail" + } + jobs = append(append(jobs, mids...), params("last", 1)...) + var deps []driver.JobDependency + for m := 1; m <= middles; m++ { + deps = append(deps, driver.JobDependency{Job: m, DependsOn: 0}, driver.JobDependency{Job: middles + 1, DependsOn: m}) + } + if _, err := exec.WorkflowInsert(ctx, driver.WorkflowInsertParams{Name: "chaos", Jobs: jobs, Deps: deps, OnComplete: cb}); err != nil { + t.Fatal(err) + } + } + + wctx, cancel := context.WithTimeout(ctx, time.Minute) + defer cancel() + var ( + mu sync.Mutex + applied = map[driver.JobID]int{} + errs = make(chan error, 16) + replay = make(chan []driver.JobFinalize, 256) + wg sync.WaitGroup + ) + record := func(ids []driver.JobID) { + mu.Lock() + defer mu.Unlock() + for _, id := range ids { + applied[id]++ + } + } + fail := func(err error) { + if wctx.Err() == nil { + select { + case errs <- err: + default: + } + cancel() + } + } + // A failed statement is sent again whole, as the finalizer does; a + // deadlock between two finalizes completing batches fails one of them. + finalizeRetrying := func(fin []driver.JobFinalize) []driver.JobID { + for wctx.Err() == nil { + ids, err := exec.JobFinalizeMany(wctx, driver.JobFinalizeParams{Jobs: fin}) + if err == nil { + return ids + } + time.Sleep(time.Millisecond) + } + return nil + } + // Small claims split a workflow's middle steps across statements. + for w := range 8 { + clientID := register(ctx, t, exec) + wg.Go(func() { + for wctx.Err() == nil { + res, err := exec.JobClaim(wctx, driver.JobClaimParams{Queue: "default", ClientID: clientID, Limit: 1 + w%3}) + if err != nil { + fail(fmt.Errorf("claim: %w", err)) + return + } + if len(res.Jobs) == 0 { + time.Sleep(2 * time.Millisecond) + continue + } + fin := make([]driver.JobFinalize, len(res.Jobs)) + batches := make([]driver.JobID, len(res.Jobs)) + for i, j := range res.Jobs { + fin[i] = driver.JobFinalize{ID: j.ID, AttemptedBy: clientID, Fence: j.Fence, State: driver.JobStateCompleted, Archive: true} + if j.Kind == "fail" { + fin[i].State = driver.JobStateDiscarded + } + batches[i] = j.BatchID + } + ids := finalizeRetrying(fin) + record(ids) + select { + case replay <- fin: + default: + } + // The follow-up may time out on a batch row; the sweeps + // below settle what it leaves. + _, _ = exec.BatchSettle(wctx, driver.BatchSettleParams{IDs: batches}) + _, _ = exec.WorkflowPromoteReady(wctx, driver.WorkflowPromoteParams{DependsOn: ids}) + } + }) + } + wg.Go(func() { + for wctx.Err() == nil { + select { + case fin := <-replay: + record(finalizeRetrying(fin)) + case <-wctx.Done(): + } + } + }) + wg.Go(func() { + var settleAfter, promoteAfter driver.JobID + for wctx.Err() == nil { + s, err := exec.BatchSettle(wctx, driver.BatchSettleParams{After: settleAfter, Limit: 16}) + if err == nil { + settleAfter = s.Last + if s.Folded < 16 { + settleAfter = driver.JobID{} + } + } + p, err := exec.WorkflowPromoteReady(wctx, driver.WorkflowPromoteParams{After: promoteAfter, Limit: 16}) + if err == nil { + promoteAfter = p.Last + if p.Scanned < 16 { + promoteAfter = driver.JobID{} + } + } + time.Sleep(5 * time.Millisecond) + } + }) + for wctx.Err() == nil && f.QueryText(ctx, t, d, `SELECT ((SELECT count(*) FROM hopper_jobs WHERE queue = 'default') = 0 + AND (SELECT count(*) FROM hopper_batches WHERE completed_at IS NULL) = 0)::text`) != "true" { + time.Sleep(20 * time.Millisecond) + } + timedOut := wctx.Err() != nil + cancel() + wg.Wait() + close(errs) + for err := range errs { + t.Fatal(err) + } + if timedOut { + t.Fatalf("not finished in a minute: %s", f.QueryText(ctx, t, d, + "SELECT coalesce(string_agg(state || ' ' || n, ', '), 'none') FROM (SELECT state::text, count(*) AS n FROM hopper_jobs GROUP BY state) s")) + } + + failing := workflows / 5 + want := workflows*(middles+2) - failing + twice := 0 + for _, n := range applied { + if n != 1 { + twice++ + } + } + if len(applied) != want || twice != 0 { + t.Errorf("results applied for %d jobs, %d of them more than once; want %d jobs, once each", len(applied), twice, want) + } + checks := []struct{ what, query, want string }{ + { + "history rows, distinct jobs", `SELECT count(*) || ' ' || count(DISTINCT id) FROM hopper_job_history WHERE queue = 'default'`, + fmt.Sprintf("%d %d", workflows*(middles+2), workflows*(middles+2)), + }, + {"steps run before a dependency finished", `SELECT count(*) FROM hopper_job_history s + JOIN hopper_job_history dep ON dep.batch_id = s.batch_id + AND (s.kind IN ('mid', 'fail') AND dep.kind = 'root' OR s.kind = 'last' AND dep.kind IN ('mid', 'fail')) + WHERE s.attempted_at IS NOT NULL AND s.attempted_at < dep.finalized_at`, "0"}, + { + "last steps by state", `SELECT string_agg(state || ' ' || n, ', ' ORDER BY state) FROM ( + SELECT state::text, count(*) AS n FROM hopper_job_history WHERE kind = 'last' GROUP BY state) s`, + fmt.Sprintf("canceled %d, completed %d", failing, workflows-failing), + }, + {"canceled last steps that ran", `SELECT count(*) FROM hopper_job_history WHERE kind = 'last' AND state = 'canceled' AND attempted_at IS NOT NULL`, "0"}, + {"workflows whose failures miscount", `SELECT count(*) FROM hopper_batches b + WHERE b.failed <> (SELECT count(*) FROM hopper_job_history h WHERE h.batch_id = b.id AND h.state <> 'completed')`, "0"}, + {"failures", `SELECT sum(failed) FROM hopper_batches`, strconv.Itoa(2 * failing)}, + {"completion callbacks", `SELECT count(*) FROM hopper_jobs WHERE queue = 'callbacks'`, strconv.Itoa(workflows)}, + {"edges left", `SELECT count(*) FROM hopper_job_deps`, "0"}, + } + for _, c := range checks { + if got := f.QueryText(ctx, t, d, "SELECT ("+c.query+")::text"); got != c.want { + t.Errorf("%s = %s, want %s", c.what, got, c.want) + } + } +} From 890fb347dd21d1e7e2cb90bcee9d5b080758b4e8 Mon Sep 17 00:00:00 2001 From: Michael McQuade Date: Mon, 19 Oct 2026 09:06:42 -0500 Subject: [PATCH 05/19] docs: finalize plan stability and dependency table maintenance in PLAN --- docs/PLAN.md | 27 ++++++++++++++++++++++----- 1 file changed, 22 insertions(+), 5 deletions(-) diff --git a/docs/PLAN.md b/docs/PLAN.md index 93cfdfd..1e3c009 100644 --- a/docs/PLAN.md +++ b/docs/PLAN.md @@ -967,7 +967,20 @@ RETURNING j.*; the update to it with a sequential scan and a hash, so every claim scanned the table (a 94% loss under `force_generic_plan`), and in the automatic mode the planner rejected that plan and replanned every execution instead (−27% on the CI runner). Rows locked - beyond the count are released when the statement ends. + beyond the count are released when the statement ends. The finalize statement finds + its results' rows by primary key whatever the statistics say, comparing the running + state as text, which no partial index's predicate matches: compared as the enum, + statistics taken while few jobs ran (an analyze just after a backlog was inserted) + made a scan of all of `hopper_jobs_running` look like a page, and the statement + compared every running row with every result, 0.6 s a statement at 40,000 running + rows, which piled up as the finalizes fell behind, so a workflow drain of 3 s took + 20–48 s. The statements that look up one running job by ID check it the same way. + The finalize tail and the dependents' promotion look up rows keyed by the + statement's own rows (the finished jobs, their batches, the candidates) as + `= ANY (ARRAY(...))`, which the planner takes for ten keys and so probes an index + for: joined to the CTEs, it guessed hundreds of rows for them, and scanned + `hopper_job_deps`, `hopper_jobs` and `hopper_batches` whole on every finalize, at a + cost that grew with the workflows in flight. - **Live-table maintenance.** Every claim walks the claim index past the entries of jobs that have run and left, which only a vacuum removes, so pickup latency climbs between vacuums: at 30,000 jobs/s on the reference hardware, from 10 ms to seconds @@ -980,10 +993,14 @@ RETURNING j.*; on the live table for autovacuum too; what matters is their number, about five hundred index pages, not their share of the table, so a short burst never triggers one) and re-analyzes it, with a small sample, - only when the row count the planner holds is ten times off the live count: an analyze - in the middle of a burst cost a CPU-starved server a fifth of its throughput, so it is - not done on churn alone. It never waits for a lock. Schema v6 also sets the table's - autovacuum thresholds low with no cost delay. + at once when the row count the planner holds is ten times off the live count, and + otherwise once a tenth of the table has changed but at most every thirty seconds: an + analyze invalidates every connection's cached plans, and one in the middle of a burst + cost a CPU-starved server a fifth of its throughput. The live tags table and + `hopper_job_deps` are kept the same way, with the server's sample size: until the + dependency table has statistics, the planner takes each lookup of a job's edges for a + two-hundredth of the table, and scans it instead. It never waits for a lock. Schema v6 + also sets the table's autovacuum thresholds low with no cost delay. - **Batching.** A producer claims only when at least `min(MaxWorkers/4, free slots)` slots are free or a short cooldown (20ms by default) has passed. Busy queues are From 73fb09d005ae9e436247e71bb6da6ffae9fa8d1b Mon Sep 17 00:00:00 2001 From: Michael McQuade Date: Mon, 19 Oct 2026 09:11:24 -0500 Subject: [PATCH 06/19] test(drivertest): check a claimed step's dependencies are gone, not timestamps --- drivertest/workflow.go | 22 ++++++++++++++-------- 1 file changed, 14 insertions(+), 8 deletions(-) diff --git a/drivertest/workflow.go b/drivertest/workflow.go index ffbbf88..75e757f 100644 --- a/drivertest/workflow.go +++ b/drivertest/workflow.go @@ -331,7 +331,7 @@ func testWorkflowFanInConcurrent[TTx any](t *testing.T, f Fixture[TTx]) { // claimed in one statement and following up as a client does, while the // leader's sweeps run and every statement's results are sent a second time. // In one workflow in five, a middle step fails. Every job is finalized -// once, no step runs before its dependencies have finished, the step after +// once, no step is claimed while a dependency is live, the step after // a failed one is canceled, and every workflow completes once, with its // failures counted. func testWorkflowsConcurrent[TTx any](t *testing.T, f Fixture[TTx]) { @@ -409,15 +409,25 @@ func testWorkflowsConcurrent[TTx any](t *testing.T, f Fixture[TTx]) { } fin := make([]driver.JobFinalize, len(res.Jobs)) batches := make([]driver.JobID, len(res.Jobs)) + ids := make([]string, len(res.Jobs)) for i, j := range res.Jobs { fin[i] = driver.JobFinalize{ID: j.ID, AttemptedBy: clientID, Fence: j.Fence, State: driver.JobStateCompleted, Archive: true} if j.Kind == "fail" { fin[i].State = driver.JobStateDiscarded } batches[i] = j.BatchID + ids[i] = j.ID.String() } - ids := finalizeRetrying(fin) - record(ids) + // A dependency leaves the live table when its finalize + // commits, and never returns: one still there ran after + // its dependent was claimed. + if n := f.QueryText(ctx, t, d, `SELECT count(*)::text FROM hopper_job_deps e + JOIN hopper_jobs x ON x.id = e.depends_on WHERE e.job_id = ANY($1::uuid[])`, "{"+strings.Join(ids, ",")+"}"); n != "0" { + fail(fmt.Errorf("claimed %d jobs with %s dependencies still live", len(ids), n)) + return + } + finished := finalizeRetrying(fin) + record(finished) select { case replay <- fin: default: @@ -425,7 +435,7 @@ func testWorkflowsConcurrent[TTx any](t *testing.T, f Fixture[TTx]) { // The follow-up may time out on a batch row; the sweeps // below settle what it leaves. _, _ = exec.BatchSettle(wctx, driver.BatchSettleParams{IDs: batches}) - _, _ = exec.WorkflowPromoteReady(wctx, driver.WorkflowPromoteParams{DependsOn: ids}) + _, _ = exec.WorkflowPromoteReady(wctx, driver.WorkflowPromoteParams{DependsOn: finished}) } }) } @@ -490,10 +500,6 @@ func testWorkflowsConcurrent[TTx any](t *testing.T, f Fixture[TTx]) { "history rows, distinct jobs", `SELECT count(*) || ' ' || count(DISTINCT id) FROM hopper_job_history WHERE queue = 'default'`, fmt.Sprintf("%d %d", workflows*(middles+2), workflows*(middles+2)), }, - {"steps run before a dependency finished", `SELECT count(*) FROM hopper_job_history s - JOIN hopper_job_history dep ON dep.batch_id = s.batch_id - AND (s.kind IN ('mid', 'fail') AND dep.kind = 'root' OR s.kind = 'last' AND dep.kind IN ('mid', 'fail')) - WHERE s.attempted_at IS NOT NULL AND s.attempted_at < dep.finalized_at`, "0"}, { "last steps by state", `SELECT string_agg(state || ' ' || n, ', ' ORDER BY state) FROM ( SELECT state::text, count(*) AS n FROM hopper_job_history WHERE kind = 'last' GROUP BY state) s`, From 110bbd1463284fb7f18953357991bfe37ac25445 Mon Sep 17 00:00:00 2001 From: Michael McQuade Date: Mon, 19 Oct 2026 09:20:55 -0500 Subject: [PATCH 07/19] ci(perf): gate the workflow scenario --- .github/workflows/perf.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/perf.yml b/.github/workflows/perf.yml index 44596b3..a0cb929 100644 --- a/.github/workflows/perf.yml +++ b/.github/workflows/perf.yml @@ -49,7 +49,7 @@ jobs: --health-interval 2s --health-timeout 5s --health-retries 30 env: DATABASE_URL: postgres://hopper:hopper@localhost:5432/hopper_bench?sslmode=disable - SCENARIOS: throughput,insert,copy + SCENARIOS: throughput,insert,copy,workflow ROUNDS: "6" JOBS: "300000" steps: From b462b8df927ecda762ff37667c61eced75ee64f3 Mon Sep 17 00:00:00 2001 From: Michael McQuade Date: Mon, 19 Oct 2026 09:53:47 -0500 Subject: [PATCH 08/19] test(driver): keep the chaos test's finalize replies, plan test setup on any connection --- driver/hopperpgx/postgres_test.go | 20 +++++++++++++++++--- drivertest/cluster.go | 7 +++---- drivertest/workflow.go | 15 ++++++++++++--- 3 files changed, 32 insertions(+), 10 deletions(-) diff --git a/driver/hopperpgx/postgres_test.go b/driver/hopperpgx/postgres_test.go index 63ab391..15a2d93 100644 --- a/driver/hopperpgx/postgres_test.go +++ b/driver/hopperpgx/postgres_test.go @@ -1225,8 +1225,9 @@ func TestFinalizePlansIgnoreStaleStatistics(t *testing.T) { exec1("ALTER TABLE " + table + " SET (autovacuum_enabled = false)") } // Five thousand workflows of ten steps: one, eight after it, one after - // those. - exec1(`CREATE TEMP TABLE steps AS + // those. The steps table is in the test's schema, not a temporary one, + // which only the pool's connection that created it would see. + exec1(`CREATE UNLOGGED TABLE steps AS SELECT w, s, hopper_uuidv7() AS id, b.id AS batch_id FROM (SELECT w, hopper_uuidv7() AS id FROM generate_series(1, 5000) w) b, generate_series(0, 9) s ORDER BY w, s`) exec1(`INSERT INTO hopper_batches (id, pending, total, name) SELECT DISTINCT batch_id, 10, 10, 'w' FROM steps`) @@ -1275,7 +1276,20 @@ func TestFinalizePlansIgnoreStaleStatistics(t *testing.T) { if err != nil || len(finished) != 100 { t.Fatalf("finalize = %d, %v", len(finished), err) } - if _, err := traced.JobFinalizeMany(ctx, driver.JobFinalizeParams{Jobs: results(running[100:200]), NoBatches: true}); err != nil { + // Jobs outside workflows, first in the claim order, for the statement + // that finalizes jobs without batches. + plain := insertParams("plain", 100) + for i := range plain { + plain[i].Priority = 1 + } + if _, err := exec.JobInsertMany(ctx, plain, driver.JobInsertOpts{}); err != nil { + t.Fatal(err) + } + claimed, err := exec.JobClaim(ctx, driver.JobClaimParams{Queue: "default", ClientID: clientID, Limit: 100}) + if err != nil || len(claimed.Jobs) != 100 || claimed.Jobs[0].Kind != "plain" || claimed.Jobs[99].Kind != "plain" { + t.Fatalf("claim of the jobs outside workflows = %d jobs, %v", len(claimed.Jobs), err) + } + if _, err := traced.JobFinalizeMany(ctx, driver.JobFinalizeParams{Jobs: results(claimed.Jobs), NoBatches: true}); err != nil { t.Fatal(err) } if _, err := traced.WorkflowPromoteReady(ctx, driver.WorkflowPromoteParams{DependsOn: finished}); err != nil { diff --git a/drivertest/cluster.go b/drivertest/cluster.go index e5434e7..e0d3643 100644 --- a/drivertest/cluster.go +++ b/drivertest/cluster.go @@ -1139,10 +1139,9 @@ func testDepsTableMaintained[TTx any](t *testing.T, f Fixture[TTx]) { if stat("last_analyze IS NULL") != "true" { t.Fatal("dependency table analyzed before maintenance") } - res := waitMaintain(ctx, t, exec, func(r driver.JobsMaintainResult) bool { return r.Analyzed }) - if !res.Analyzed { - t.Fatalf("maintenance after 1200 edges = %+v", res) - } + // The result's Analyzed would also report the analyze of hopper_jobs, + // which the workflow's 1201 jobs earn, so the table's own statistics + // are what tell. for deadline := time.Now().Add(10 * time.Second); stat("last_analyze IS NOT NULL") != "true"; time.Sleep(100 * time.Millisecond) { if time.Now().After(deadline) { t.Fatal("dependency table not analyzed") diff --git a/drivertest/workflow.go b/drivertest/workflow.go index 75e757f..da8b520 100644 --- a/drivertest/workflow.go +++ b/drivertest/workflow.go @@ -361,6 +361,7 @@ func testWorkflowsConcurrent[TTx any](t *testing.T, f Fixture[TTx]) { var ( mu sync.Mutex applied = map[driver.JobID]int{} + lastErr error errs = make(chan error, 16) replay = make(chan []driver.JobFinalize, 256) wg sync.WaitGroup @@ -383,12 +384,20 @@ func testWorkflowsConcurrent[TTx any](t *testing.T, f Fixture[TTx]) { } // A failed statement is sent again whole, as the finalizer does; a // deadlock between two finalizes completing batches fails one of them. + // The statement runs on a context the end of the test does not cancel: + // canceled while its reply was on the way, a finalize the database + // applied would count as never applied. finalizeRetrying := func(fin []driver.JobFinalize) []driver.JobID { for wctx.Err() == nil { - ids, err := exec.JobFinalizeMany(wctx, driver.JobFinalizeParams{Jobs: fin}) + sctx, scancel := context.WithTimeout(ctx, 10*time.Second) + ids, err := exec.JobFinalizeMany(sctx, driver.JobFinalizeParams{Jobs: fin}) + scancel() if err == nil { return ids } + mu.Lock() + lastErr = err + mu.Unlock() time.Sleep(time.Millisecond) } return nil @@ -430,7 +439,7 @@ func testWorkflowsConcurrent[TTx any](t *testing.T, f Fixture[TTx]) { record(finished) select { case replay <- fin: - default: + case <-wctx.Done(): } // The follow-up may time out on a batch row; the sweeps // below settle what it leaves. @@ -480,7 +489,7 @@ func testWorkflowsConcurrent[TTx any](t *testing.T, f Fixture[TTx]) { t.Fatal(err) } if timedOut { - t.Fatalf("not finished in a minute: %s", f.QueryText(ctx, t, d, + t.Fatalf("not finished in a minute (last finalize error: %v): %s", lastErr, f.QueryText(ctx, t, d, "SELECT coalesce(string_agg(state || ' ' || n, ', '), 'none') FROM (SELECT state::text, count(*) AS n FROM hopper_jobs GROUP BY state) s")) } From db220f5b94fdb502b8f744a424d89ed3d454ce4b Mon Sep 17 00:00:00 2001 From: Michael McQuade Date: Mon, 19 Oct 2026 09:53:47 -0500 Subject: [PATCH 09/19] fix(driver): analyze hopper_job_deps until it has statistics, read all counters at once --- driver/internal/pgsql/cluster.go | 101 +++++++++++++++++++++---------- 1 file changed, 68 insertions(+), 33 deletions(-) diff --git a/driver/internal/pgsql/cluster.go b/driver/internal/pgsql/cluster.go index 6c94b6b..8eda116 100644 --- a/driver/internal/pgsql/cluster.go +++ b/driver/internal/pgsql/cluster.go @@ -399,11 +399,13 @@ const analyzeEvery = 30 * time.Second // // The live tags table churns with tagged jobs, and a listing by tag walks // its index, so it is kept the same way, except that its analyze keeps the -// server's statistics target. So does the dependency table, which churns -// with workflows: until it has statistics, the planner takes each lookup of +// server's statistics target. So is the dependency table, which churns +// with workflows, except that it is analyzed on churn only until it has +// statistics (see liveTable): until then, the planner takes each lookup of // a job's edges for a two-hundredth of the table, and scans the table // instead, in finalizes and in the sweep that promotes dependents (see -// finalizeTailSQL). Each table is kept whether or not another's +// finalizeTailSQL). The counters of all three are read in one statement. +// Each table is kept whether or not another's // maintenance fails, such as a vacuum of a large hopper_jobs that runs past // the role's statement_timeout on every pass. func (e *Executor) JobsMaintain(ctx context.Context) (driver.JobsMaintainResult, error) { @@ -411,9 +413,39 @@ func (e *Executor) JobsMaintain(ctx context.Context) (driver.JobsMaintainResult, if e.InTx { return res, errors.New("hopper: live table maintenance must run on the pool, not in a transaction") } + names := make([]string, len(liveTables)) + for i, t := range liveTables { + names[i] = e.Schema.Table(t.name) + } + rows, err := e.Conn.Query(ctx, ` + SELECT n, s.n_dead_tup, s.n_live_tup, s.n_mod_since_analyze, c.reltuples, + coalesce(extract(epoch FROM now() - greatest(s.last_analyze, s.last_autoanalyze)), 1e9) + FROM unnest($1::text[]) n + JOIN pg_stat_user_tables s ON s.relid = to_regclass(n) JOIN pg_class c ON c.oid = s.relid`, textArray(names)) + if err != nil { + return res, fmt.Errorf("hopper: live table statistics: %w", err) + } + stats := map[string]liveTableStats{} + for rows.Next() { + var name string + var st liveTableStats + if err := rows.Scan(&name, &st.dead, &st.live, &st.modified, &st.believed, &st.sinceAnalyze); err != nil { + rows.Close() + return res, fmt.Errorf("hopper: live table statistics: %w", err) + } + stats[name] = st + } + rows.Close() + if err := rows.Err(); err != nil { + return res, fmt.Errorf("hopper: live table statistics: %w", err) + } var errs []error - for _, t := range liveTables { - vacuumed, analyzed, err := e.maintainLiveTable(ctx, t.name, t.statisticsTarget) + for i, t := range liveTables { + st, ok := stats[names[i]] + if !ok { + continue + } + vacuumed, analyzed, err := e.maintainLiveTable(ctx, t, names[i], st) if err != nil { errs = append(errs, err) continue @@ -424,63 +456,66 @@ func (e *Executor) JobsMaintain(ctx context.Context) (driver.JobsMaintainResult, return res, errors.Join(errs...) } +// liveTableStats are a live table's activity counters and the row count +// the planner holds. +type liveTableStats struct { + dead, live, modified int64 + believed, sinceAnalyze float64 +} + // liveTables are the tables JobsMaintain keeps, with the statistics target // their analyze uses, or zero for the server's. The tags table keeps the // server's target: a listing by several tags (rarestTagSQL) tells common // tags apart by its most common values, of which a target of 10 would keep // only ten. So does the dependency table, whose estimates rest on how many // distinct jobs its sample holds, which a larger sample counts better. -var liveTables = []struct { +var liveTables = []liveTable{ + {"hopper_jobs", 10, true}, + {"hopper_job_tags", 0, true}, + {"hopper_job_deps", 0, false}, +} + +// liveTable is a table JobsMaintain keeps. A table without churn is +// analyzed on churn only until it has statistics, and then only when its +// size is off: what the planner reads from the dependency table, how many +// edges a job has and how many jobs depend on one, does not change as +// workflows come and go, and each analyze has every connection replan the +// finalize statement. +type liveTable struct { name string statisticsTarget int -}{ - {"hopper_jobs", 10}, - {"hopper_job_tags", 0}, - {"hopper_job_deps", 0}, + churn bool } -func (e *Executor) maintainLiveTable(ctx context.Context, table string, statisticsTarget int) (vacuumed, analyzed bool, err error) { - var dead, live, modified int64 - var believed, sinceAnalyze float64 - err = e.Conn.QueryRow(ctx, ` - SELECT s.n_dead_tup, s.n_live_tup, s.n_mod_since_analyze, c.reltuples, - coalesce(extract(epoch FROM now() - greatest(s.last_analyze, s.last_autoanalyze)), 1e9) - FROM pg_stat_user_tables s JOIN pg_class c ON c.oid = s.relid - WHERE s.relid = to_regclass($1)`, e.Schema.Table(table)).Scan(&dead, &live, &modified, &believed, &sinceAnalyze) - if errors.Is(err, ErrNoRows) { - return false, false, nil - } - if err != nil { - return false, false, fmt.Errorf("hopper: %s statistics: %w", table, err) - } - if dead >= vacuumDeadRows { +func (e *Executor) maintainLiveTable(ctx context.Context, t liveTable, table string, st liveTableStats) (vacuumed, analyzed bool, err error) { + if st.dead >= vacuumDeadRows { // TRUNCATE false: a vacuum otherwise ends by trying, for up to five // seconds, to take an exclusive lock to give back empty trailing // pages, and whenever it briefly gets one every claim and finalize // waits behind it. A churning live table reuses those pages within // seconds anyway. - if _, err := e.Conn.Exec(ctx, "VACUUM (SKIP_LOCKED, TRUNCATE false) "+e.Schema.Table(table)); err != nil { - return false, false, fmt.Errorf("hopper: vacuum %s: %w", table, err) + if _, err := e.Conn.Exec(ctx, "VACUUM (SKIP_LOCKED, TRUNCATE false) "+table); err != nil { + return false, false, fmt.Errorf("hopper: vacuum %s: %w", t.name, err) } return true, false, nil } floor := func(v float64) float64 { return max(v, 100) } - sizeOff := believed >= 0 && (float64(live) >= 10*floor(believed) || believed >= 10*floor(float64(live))) - churned := modified >= 1000 && modified*10 >= live && sinceAnalyze >= analyzeEvery.Seconds() + sizeOff := st.believed >= 0 && (float64(st.live) >= 10*floor(st.believed) || st.believed >= 10*floor(float64(st.live))) + churned := st.modified >= 1000 && st.modified*10 >= st.live && st.sinceAnalyze >= analyzeEvery.Seconds() && (t.churn || st.believed < 0) if !sizeOff && !churned { return false, false, nil } err = e.withTx(ctx, func(tx Tx) error { - if statisticsTarget > 0 { - if _, err := tx.Exec(ctx, "SET LOCAL default_statistics_target = "+strconv.Itoa(statisticsTarget)); err != nil { + if t.statisticsTarget > 0 { + if _, err := tx.Exec(ctx, "SET LOCAL default_statistics_target = "+strconv.Itoa(t.statisticsTarget)); err != nil { return err } } - _, err := tx.Exec(ctx, "ANALYZE (SKIP_LOCKED) "+e.Schema.Table(table)) + _, err := tx.Exec(ctx, "ANALYZE (SKIP_LOCKED) "+table) return err }) if err != nil { - return false, false, fmt.Errorf("hopper: analyze %s: %w", table, err) + return false, false, fmt.Errorf("hopper: analyze %s: %w", t.name, err) } return false, true, nil } From b1b8734335802bf33656fd3223c56d38fea87525 Mon Sep 17 00:00:00 2001 From: Michael McQuade Date: Mon, 19 Oct 2026 09:53:47 -0500 Subject: [PATCH 10/19] refactor(driver): one spelling of the running check; cascade and deps analyze docs --- docs/PLAN.md | 9 ++++++--- driver/internal/pgsql/jobs.go | 8 ++++---- driver/internal/pgsql/workflow.go | 5 +++-- 3 files changed, 13 insertions(+), 9 deletions(-) diff --git a/docs/PLAN.md b/docs/PLAN.md index 1e3c009..7e73128 100644 --- a/docs/PLAN.md +++ b/docs/PLAN.md @@ -997,9 +997,12 @@ RETURNING j.*; otherwise once a tenth of the table has changed but at most every thirty seconds: an analyze invalidates every connection's cached plans, and one in the middle of a burst cost a CPU-starved server a fifth of its throughput. The live tags table and - `hopper_job_deps` are kept the same way, with the server's sample size: until the - dependency table has statistics, the planner takes each lookup of a job's edges for a - two-hundredth of the table, and scans it instead. It never waits for a lock. Schema v6 + `hopper_job_deps` are kept the same way, with the server's sample size, except that + the dependency table is analyzed on churn only until it has statistics, and then only + when its row count is off: until it has them, the planner takes each lookup of a job's + edges for a two-hundredth of the table, and scans it instead, while what it reads from + them (edges per job, dependents per job) does not change as workflows come and go, and + each analyze has every connection replan the finalize statement. It never waits for a lock. Schema v6 also sets the table's autovacuum thresholds low with no cost delay. - **Batching.** A producer claims only when at least `min(MaxWorkers/4, free slots)` diff --git a/driver/internal/pgsql/jobs.go b/driver/internal/pgsql/jobs.go index d9fe954..281ccf9 100644 --- a/driver/internal/pgsql/jobs.go +++ b/driver/internal/pgsql/jobs.go @@ -605,22 +605,22 @@ retry AS ( THEN r.snooze_reason END, wake_requested_at = NULL FROM r - WHERE j.id = r.id AND j.state::text = 'running' AND j.attempted_by = r.attempted_by AND j.fence = r.fence + WHERE j.id = r.id AND j.` + runningSQL + ` AND j.attempted_by = r.attempted_by AND j.fence = r.fence AND r.state IN ('retryable', 'scheduled') RETURNING j.id ), finished AS ( DELETE FROM {{schema}}.hopper_jobs j USING r - WHERE j.id = r.id AND j.state::text = 'running' AND j.attempted_by = r.attempted_by AND j.fence = r.fence + WHERE j.id = r.id AND j.` + runningSQL + ` AND j.attempted_by = r.attempted_by AND j.fence = r.fence AND r.state IN ('completed', 'canceled', 'discarded') RETURNING ` + finishedColumns(`CASE WHEN r.error IS NULL THEN j.errors ELSE j.errors || jsonb_set(r.error, '{at}', to_jsonb(now())) END`, appendLogSQL, "r.output", "r.archive", "r.state", "$9") + ` ), ` -// runningSQL is the running check of jobFinalizeHeadSQL for a statement -// whose hopper_jobs row needs no alias. +// runningSQL is the running check of jobFinalizeHeadSQL, which prefixes it +// with its alias, for the statements that find one running job by ID. const runningSQL = "state::text = 'running'" var jobFinalizeSQL = jobFinalizeHeadSQL + finalizeTailSQL("$9", "$10") + ` diff --git a/driver/internal/pgsql/workflow.go b/driver/internal/pgsql/workflow.go index 6195087..04645ed 100644 --- a/driver/internal/pgsql/workflow.go +++ b/driver/internal/pgsql/workflow.go @@ -119,8 +119,9 @@ const plainFinalizeTailSQL = `done AS (SELECT * FROM finished), // instead, it guessed hundreds of rows for them, and with no statistics on // hopper_job_deps a probe looked like a few hundred rows more, so it scanned // hopper_job_deps, hopper_jobs and hopper_batches whole on every finalize, -// at a cost that grew with the workflows in flight. The recursive step of -// the cascade cannot take that form, and relies on the leader keeping +// at a cost that grew with the workflows in flight. The cascade itself +// still joins, its first step to finished and its recursive step to its +// own rows, which cannot take that form, and relies on the leader keeping // hopper_job_deps analyzed (JobsMaintain). func finalizeTailSQL(doneChannelParam, insertChannelParam string) string { return `cascade AS ( From e8e2948879deddc311ad503420c617bb92081762 Mon Sep 17 00:00:00 2001 From: Michael McQuade Date: Mon, 19 Oct 2026 09:53:47 -0500 Subject: [PATCH 11/19] ci(perf): drain 100,000 jobs in the workflow scenario --- .github/workflows/perf.yml | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/.github/workflows/perf.yml b/.github/workflows/perf.yml index a0cb929..eef697a 100644 --- a/.github/workflows/perf.yml +++ b/.github/workflows/perf.yml @@ -8,7 +8,9 @@ name: Performance gate # side's runs, averaged, is what is compared. Each run drains 300,000 jobs # (about ten seconds on a runner): a shorter drain measures start-up # effects, such as the leader's first statistics refresh, rather than the -# hot paths. The reference numbers in +# hot paths. The workflow scenario, whose jobs each pass through the +# finalize tail and dependency promotion, drains 100,000 (ten thousand +# workflows), about as long. The reference numbers in # docs/PLAN.md §8.2 come from the release hardware, not from here. on: pull_request: @@ -52,6 +54,7 @@ jobs: SCENARIOS: throughput,insert,copy,workflow ROUNDS: "6" JOBS: "300000" + WORKFLOW_JOBS: "100000" steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 with: @@ -72,8 +75,10 @@ jobs: run: | for i in $(seq "$ROUNDS"); do for s in ${SCENARIOS//,/ }; do - "$RUNNER_TEMP/bench-base" -database-url "$DATABASE_URL" -scenarios "$s" -jobs "$JOBS" -schema "base_$i" >> base.jsonl - "$RUNNER_TEMP/bench-head" -database-url "$DATABASE_URL" -scenarios "$s" -jobs "$JOBS" -schema "head_$i" >> head.jsonl + jobs=$JOBS + if [ "$s" = workflow ]; then jobs=$WORKFLOW_JOBS; fi + "$RUNNER_TEMP/bench-base" -database-url "$DATABASE_URL" -scenarios "$s" -jobs "$jobs" -schema "base_$i" >> base.jsonl + "$RUNNER_TEMP/bench-head" -database-url "$DATABASE_URL" -scenarios "$s" -jobs "$jobs" -schema "head_$i" >> head.jsonl done done echo "== base"; cat base.jsonl From 467bbf949e96e8491ba8caafc357e59af9e73e26 Mon Sep 17 00:00:00 2001 From: Michael McQuade Date: Mon, 19 Oct 2026 10:37:35 -0500 Subject: [PATCH 12/19] fix(driver): analyze hopper_job_deps until it was analyzed once, even after a vacuum --- docs/PLAN.md | 25 +++++++++----- driver/internal/pgsql/cluster.go | 59 ++++++++++++++++---------------- drivertest/cluster.go | 12 ++++--- drivertest/workflow.go | 5 +++ 4 files changed, 60 insertions(+), 41 deletions(-) diff --git a/docs/PLAN.md b/docs/PLAN.md index 7e73128..598327c 100644 --- a/docs/PLAN.md +++ b/docs/PLAN.md @@ -975,12 +975,19 @@ RETURNING j.*; compared every running row with every result, 0.6 s a statement at 40,000 running rows, which piled up as the finalizes fell behind, so a workflow drain of 3 s took 20–48 s. The statements that look up one running job by ID check it the same way. + The cast stays because Postgres (14–18) uses a partial index only for a condition on + the identical expression, and no index but the primary key serves a lookup by ID; + matching by primary key with the state checked after, `= ANY ('{running}')` or a + scalar subquery all still matched `hopper_jobs_running` in EXPLAIN. The finalize tail and the dependents' promotion look up rows keyed by the - statement's own rows (the finished jobs, their batches, the candidates) as + statement's own rows (the finished jobs, the completed batches, the candidates) as `= ANY (ARRAY(...))`, which the planner takes for ten keys and so probes an index for: joined to the CTEs, it guessed hundreds of rows for them, and scanned `hopper_job_deps`, `hopper_jobs` and `hopper_batches` whole on every finalize, at a - cost that grew with the workflows in flight. + cost that grew with the workflows in flight. The counted batches are looked up one by + one by primary key, in a `LATERAL` subquery fenced with `OFFSET 0`: as `= ANY` the + planner rescanned the counts once per batch row, quadratic in the batches a statement + touches. - **Live-table maintenance.** Every claim walks the claim index past the entries of jobs that have run and left, which only a vacuum removes, so pickup latency climbs between vacuums: at 30,000 jobs/s on the reference hardware, from 10 ms to seconds @@ -998,12 +1005,14 @@ RETURNING j.*; analyze invalidates every connection's cached plans, and one in the middle of a burst cost a CPU-starved server a fifth of its throughput. The live tags table and `hopper_job_deps` are kept the same way, with the server's sample size, except that - the dependency table is analyzed on churn only until it has statistics, and then only - when its row count is off: until it has them, the planner takes each lookup of a job's - edges for a two-hundredth of the table, and scans it instead, while what it reads from - them (edges per job, dependents per job) does not change as workflows come and go, and - each analyze has every connection replan the finalize statement. It never waits for a lock. Schema v6 - also sets the table's autovacuum thresholds low with no cost delay. + the dependency table is analyzed on churn only until it has been analyzed once (a + vacuum gives the planner its row count but no statistics), and then only when its row + count is off: until it has statistics, the planner takes each lookup of a job's edges + for a two-hundredth of the table, and scans it instead, while what it reads from them + (edges per job, dependents per job) does not change as workflows come and go, and each + analyze has every connection replan the finalize statement. The maintenance never + waits for a lock. Schema v6 also sets `hopper_jobs`' autovacuum thresholds low with no + cost delay. - **Batching.** A producer claims only when at least `min(MaxWorkers/4, free slots)` slots are free or a short cooldown (20ms by default) has passed. Busy queues are diff --git a/driver/internal/pgsql/cluster.go b/driver/internal/pgsql/cluster.go index 8eda116..7e5b51c 100644 --- a/driver/internal/pgsql/cluster.go +++ b/driver/internal/pgsql/cluster.go @@ -387,9 +387,9 @@ const analyzeEvery = 30 * time.Second // // ANALYZE keeps two things current that the planner needs: the row count, // which a plan cached while the table was empty gets wrong by orders of -// magnitude, and the mix of states, which decides how the finalize joins -// its parameter list to the running rows (with a stale mix that says -// almost nothing is running, it probes the list once per running row). +// magnitude, and the mix of states, from which the planner costs scans of +// the partial indexes by state. (The finalize does not depend on it: it +// compares the running state as text; see jobFinalizeHeadSQL.) // So it runs at once when the planner's row count is ten times off, and // otherwise once a tenth of the table has changed but no more than every // thirty seconds, since an analyze invalidates every connection's cached @@ -417,27 +417,18 @@ func (e *Executor) JobsMaintain(ctx context.Context) (driver.JobsMaintainResult, for i, t := range liveTables { names[i] = e.Schema.Table(t.name) } - rows, err := e.Conn.Query(ctx, ` + all, err := scanLiveTableStats(e.Conn.Query(ctx, ` SELECT n, s.n_dead_tup, s.n_live_tup, s.n_mod_since_analyze, c.reltuples, - coalesce(extract(epoch FROM now() - greatest(s.last_analyze, s.last_autoanalyze)), 1e9) + coalesce(extract(epoch FROM now() - greatest(s.last_analyze, s.last_autoanalyze)), 1e9), + greatest(s.last_analyze, s.last_autoanalyze) IS NOT NULL FROM unnest($1::text[]) n - JOIN pg_stat_user_tables s ON s.relid = to_regclass(n) JOIN pg_class c ON c.oid = s.relid`, textArray(names)) + JOIN pg_stat_user_tables s ON s.relid = to_regclass(n) JOIN pg_class c ON c.oid = s.relid`, textArray(names))) if err != nil { return res, fmt.Errorf("hopper: live table statistics: %w", err) } - stats := map[string]liveTableStats{} - for rows.Next() { - var name string - var st liveTableStats - if err := rows.Scan(&name, &st.dead, &st.live, &st.modified, &st.believed, &st.sinceAnalyze); err != nil { - rows.Close() - return res, fmt.Errorf("hopper: live table statistics: %w", err) - } - stats[name] = st - } - rows.Close() - if err := rows.Err(); err != nil { - return res, fmt.Errorf("hopper: live table statistics: %w", err) + stats := make(map[string]liveTableStats, len(all)) + for _, st := range all { + stats[st.table] = st } var errs []error for i, t := range liveTables { @@ -456,13 +447,23 @@ func (e *Executor) JobsMaintain(ctx context.Context) (driver.JobsMaintainResult, return res, errors.Join(errs...) } -// liveTableStats are a live table's activity counters and the row count -// the planner holds. +// liveTableStats are a live table's activity counters, the row count the +// planner holds, and whether it was ever analyzed. A vacuum, or an index +// built over existing rows, also sets the row count, so a count that is +// known does not mean the table has column statistics. type liveTableStats struct { + table string dead, live, modified int64 believed, sinceAnalyze float64 + everAnalyzed bool } +var scanLiveTableStats = collect(func(r Row) (liveTableStats, error) { + var st liveTableStats + err := r.Scan(&st.table, &st.dead, &st.live, &st.modified, &st.believed, &st.sinceAnalyze, &st.everAnalyzed) + return st, err +}) + // liveTables are the tables JobsMaintain keeps, with the statistics target // their analyze uses, or zero for the server's. The tags table keeps the // server's target: a listing by several tags (rarestTagSQL) tells common @@ -475,16 +476,16 @@ var liveTables = []liveTable{ {"hopper_job_deps", 0, false}, } -// liveTable is a table JobsMaintain keeps. A table without churn is -// analyzed on churn only until it has statistics, and then only when its -// size is off: what the planner reads from the dependency table, how many -// edges a job has and how many jobs depend on one, does not change as -// workflows come and go, and each analyze has every connection replan the -// finalize statement. +// liveTable is a table JobsMaintain keeps. One without analyzeOnChurn is +// analyzed on churn only until it has been analyzed once, and then only +// when its size is off: what the planner reads from the dependency table, +// how many edges a job has and how many jobs depend on one, does not +// change as workflows come and go, and each analyze has every connection +// replan the finalize statement. type liveTable struct { name string statisticsTarget int - churn bool + analyzeOnChurn bool } func (e *Executor) maintainLiveTable(ctx context.Context, t liveTable, table string, st liveTableStats) (vacuumed, analyzed bool, err error) { @@ -501,7 +502,7 @@ func (e *Executor) maintainLiveTable(ctx context.Context, t liveTable, table str } floor := func(v float64) float64 { return max(v, 100) } sizeOff := st.believed >= 0 && (float64(st.live) >= 10*floor(st.believed) || st.believed >= 10*floor(float64(st.live))) - churned := st.modified >= 1000 && st.modified*10 >= st.live && st.sinceAnalyze >= analyzeEvery.Seconds() && (t.churn || st.believed < 0) + churned := st.modified >= 1000 && st.modified*10 >= st.live && st.sinceAnalyze >= analyzeEvery.Seconds() && (t.analyzeOnChurn || !st.everAnalyzed) if !sizeOff && !churned { return false, false, nil } diff --git a/drivertest/cluster.go b/drivertest/cluster.go index e0d3643..f5f6846 100644 --- a/drivertest/cluster.go +++ b/drivertest/cluster.go @@ -1111,10 +1111,11 @@ func testMaintain[TTx any](t *testing.T, f Fixture[TTx]) { } // testDepsTableMaintained checks that the leader's live table maintenance -// analyzes the dependency table once a burst of workflows has filled it: -// until the table has statistics, the planner takes each lookup of a job's -// edges for a two-hundredth of the table, and finalizes and the promotion -// sweep scan it instead. +// analyzes the dependency table once a burst of workflows has filled it, +// even when a vacuum has already given the planner its row count: until +// the table has statistics, the planner takes each lookup of a job's edges +// for a two-hundredth of the table, and finalizes and the promotion sweep +// scan it instead. func testDepsTableMaintained[TTx any](t *testing.T, f Fixture[TTx]) { ctx := context.Background() d := f.NewDriver(t) @@ -1139,6 +1140,9 @@ func testDepsTableMaintained[TTx any](t *testing.T, f Fixture[TTx]) { if stat("last_analyze IS NULL") != "true" { t.Fatal("dependency table analyzed before maintenance") } + // A vacuum gives the planner the table's row count but no column + // statistics: the table still needs its first analyze. + f.Exec(ctx, t, d, "VACUUM hopper_job_deps") // The result's Analyzed would also report the analyze of hopper_jobs, // which the workflow's 1201 jobs earn, so the table's own statistics // are what tell. diff --git a/drivertest/workflow.go b/drivertest/workflow.go index da8b520..714cf5b 100644 --- a/drivertest/workflow.go +++ b/drivertest/workflow.go @@ -441,6 +441,11 @@ func testWorkflowsConcurrent[TTx any](t *testing.T, f Fixture[TTx]) { case replay <- fin: case <-wctx.Done(): } + // As the client does, a flush that finished nothing is not + // followed up: a nil DependsOn is the leader's sweep. + if len(finished) == 0 { + continue + } // The follow-up may time out on a batch row; the sweeps // below settle what it leaves. _, _ = exec.BatchSettle(wctx, driver.BatchSettleParams{IDs: batches}) From d0ab64537ffcd094d760fd4574b394156eea40ae Mon Sep 17 00:00:00 2001 From: Michael McQuade Date: Mon, 19 Oct 2026 10:37:35 -0500 Subject: [PATCH 13/19] fix(driver): look counted batches up by primary key, one by one --- driver/internal/pgsql/workflow.go | 30 ++++++++++++++++++------------ 1 file changed, 18 insertions(+), 12 deletions(-) diff --git a/driver/internal/pgsql/workflow.go b/driver/internal/pgsql/workflow.go index 04645ed..504b85a 100644 --- a/driver/internal/pgsql/workflow.go +++ b/driver/internal/pgsql/workflow.go @@ -113,16 +113,20 @@ const plainFinalizeTailSQL = `done AS (SELECT * FROM finished), // the insert channel. Jobs without a batch or dependents cost index probes // on small tables and nothing else. // -// Every lookup keyed by the statement's own rows (the cascade, done, the -// batches counted and those completed) is a = ANY (ARRAY(...)), which the -// planner takes for ten keys and so probes an index for. Joined to the CTEs -// instead, it guessed hundreds of rows for them, and with no statistics on -// hopper_job_deps a probe looked like a few hundred rows more, so it scanned -// hopper_job_deps, hopper_jobs and hopper_batches whole on every finalize, -// at a cost that grew with the workflows in flight. The cascade itself -// still joins, its first step to finished and its recursive step to its -// own rows, which cannot take that form, and relies on the leader keeping -// hopper_job_deps analyzed (JobsMaintain). +// The lookups keyed by the statement's own rows (the cascade's jobs, done, +// the batches completed) are = ANY (ARRAY(...)), which the planner takes +// for ten keys and so probes an index for. Joined to the CTEs instead, it +// guessed hundreds of rows for them, the cascade's alone guessed at 180, +// and with no statistics on hopper_job_deps a probe looked like a few +// hundred rows more, so it scanned hopper_job_deps, hopper_jobs and +// hopper_batches whole on every finalize, at a cost that grew with the +// workflows in flight. ready looks each counted batch up by primary key +// in a LATERAL subquery that OFFSET 0 keeps from being flattened into a +// join: joined, a hopper_batches without statistics was scanned whole, and +// with a = ANY the planner rescanned batch_counts for every batch row. +// The cascade still joins too, its first step to finished and its +// recursive step to its own rows, which cannot take that form, and relies +// on the leader keeping hopper_job_deps analyzed (JobsMaintain). func finalizeTailSQL(doneChannelParam, insertChannelParam string) string { return `cascade AS ( SELECT d.job_id FROM {{schema}}.hopper_job_deps d JOIN finished f ON d.depends_on = f.id @@ -144,8 +148,10 @@ batch_counts AS ( ), ready AS ( SELECT b.id, b.failed + p.failed + c.f AS failed - FROM batch_counts c JOIN {{schema}}.hopper_batches b ON b.id = c.batch_id, ` + batchProgressSQL + ` - WHERE b.id = ANY (ARRAY(SELECT batch_id FROM batch_counts)) AND b.completed_at IS NULL AND b.pending - p.done - c.n <= 0 + FROM batch_counts c, + LATERAL (SELECT b.id, b.failed, b.pending FROM {{schema}}.hopper_batches b + WHERE b.id = c.batch_id AND b.completed_at IS NULL OFFSET 0) b, ` + batchProgressSQL + ` + WHERE b.pending - p.done - c.n <= 0 ), batches AS ( UPDATE {{schema}}.hopper_batches b SET completed_at = now(), pending = 0, failed = r.failed From e1811c649e3671e0387bf5cd3fc67c1f555e9513 Mon Sep 17 00:00:00 2001 From: Michael McQuade Date: Mon, 19 Oct 2026 10:37:35 -0500 Subject: [PATCH 14/19] refactor(driver): build the running check from a column name --- driver/internal/pgsql/jobs.go | 21 +++++++++++---------- 1 file changed, 11 insertions(+), 10 deletions(-) diff --git a/driver/internal/pgsql/jobs.go b/driver/internal/pgsql/jobs.go index 281ccf9..f1e7000 100644 --- a/driver/internal/pgsql/jobs.go +++ b/driver/internal/pgsql/jobs.go @@ -582,7 +582,7 @@ func claimRows(ctx context.Context, conn Conn, query string, args ...any) ([]*dr // instead, which statistics taken while few jobs ran (right after a backlog // was inserted, say) make look like a page or two: the statement then // compared every running row with every result, and slowed down as -// unfinalized results piled up as running rows. runningSQL is the same +// unfinalized results piled up as running rows. isRunning is the same // check for statements that find one running job by ID. var jobFinalizeHeadSQL = ` WITH RECURSIVE r AS ( @@ -605,23 +605,24 @@ retry AS ( THEN r.snooze_reason END, wake_requested_at = NULL FROM r - WHERE j.id = r.id AND j.` + runningSQL + ` AND j.attempted_by = r.attempted_by AND j.fence = r.fence + WHERE j.id = r.id AND ` + isRunning("j.state") + ` AND j.attempted_by = r.attempted_by AND j.fence = r.fence AND r.state IN ('retryable', 'scheduled') RETURNING j.id ), finished AS ( DELETE FROM {{schema}}.hopper_jobs j USING r - WHERE j.id = r.id AND j.` + runningSQL + ` AND j.attempted_by = r.attempted_by AND j.fence = r.fence + WHERE j.id = r.id AND ` + isRunning("j.state") + ` AND j.attempted_by = r.attempted_by AND j.fence = r.fence AND r.state IN ('completed', 'canceled', 'discarded') RETURNING ` + finishedColumns(`CASE WHEN r.error IS NULL THEN j.errors ELSE j.errors || jsonb_set(r.error, '{at}', to_jsonb(now())) END`, appendLogSQL, "r.output", "r.archive", "r.state", "$9") + ` ), ` -// runningSQL is the running check of jobFinalizeHeadSQL, which prefixes it -// with its alias, for the statements that find one running job by ID. -const runningSQL = "state::text = 'running'" +// isRunning checks that the hopper_jobs state column named by column is +// running, as jobFinalizeHeadSQL explains, for the statements that find +// running jobs by ID. +func isRunning(column string) string { return column + "::text = 'running'" } var jobFinalizeSQL = jobFinalizeHeadSQL + finalizeTailSQL("$9", "$10") + ` SELECT id FROM retry UNION ALL SELECT id FROM finished` @@ -768,13 +769,13 @@ LIMIT 1`, JobColumns(""), historyColumns("")) // // With NoWait, the row is locked first with NOWAIT, and the update applies // the fence again to the row as locked. -const jobRecordProgressSQL = `%s +var jobRecordProgressSQL = `%s UPDATE {{schema}}.hopper_jobs SET progress = jsonb_build_object( 'step', CASE WHEN $5 THEN coalesce(progress->>'step', '') ELSE $4::text END, 'at', statement_timestamp()) || CASE WHEN $6::jsonb IS NULL THEN '{}'::jsonb ELSE jsonb_build_object('cursor', $6::jsonb) END -WHERE id = $1 AND ` + runningSQL + ` AND attempted_by = $2 AND fence = $3 AND attempt = $7%s +WHERE id = $1 AND ` + isRunning("state") + ` AND attempted_by = $2 AND fence = $3 AND attempt = $7%s RETURNING progress::text` var ( @@ -782,7 +783,7 @@ var ( jobRecordProgressNoWaitQuery = fmt.Sprintf(jobRecordProgressSQL, ` WITH locked AS ( SELECT id FROM {{schema}}.hopper_jobs - WHERE id = $1 AND `+runningSQL+` AND attempted_by = $2 AND fence = $3 AND attempt = $7 + WHERE id = $1 AND `+isRunning("state")+` AND attempted_by = $2 AND fence = $3 AND attempt = $7 FOR UPDATE NOWAIT )`, " AND id IN (SELECT id FROM locked)") ) @@ -1515,7 +1516,7 @@ const jobLockStateSQL = "SELECT state FROM {{schema}}.hopper_jobs WHERE id = $1 var ( jobCancelRunningSQL = fmt.Sprintf(` UPDATE {{schema}}.hopper_jobs SET cancel_requested_at = coalesce(cancel_requested_at, now()) -WHERE id = $1 AND `+runningSQL+` +WHERE id = $1 AND `+isRunning("state")+` RETURNING %s`, JobColumns("")) // jobCancelWaitingSQL finalizes a waiting job as canceled, with // everything that entails for its batch and its dependents. From 110c2aa3d419dfda5cbc9dc290de9abe1444de98 Mon Sep 17 00:00:00 2001 From: Michael McQuade Date: Mon, 19 Oct 2026 11:39:48 -0500 Subject: [PATCH 15/19] fix(driver): read live table counters by index, refresh dependency statistics every ten minutes --- driver/internal/pgsql/cluster.go | 59 ++++++++++++++++++-------------- 1 file changed, 34 insertions(+), 25 deletions(-) diff --git a/driver/internal/pgsql/cluster.go b/driver/internal/pgsql/cluster.go index 7e5b51c..2882aaa 100644 --- a/driver/internal/pgsql/cluster.go +++ b/driver/internal/pgsql/cluster.go @@ -400,12 +400,15 @@ const analyzeEvery = 30 * time.Second // The live tags table churns with tagged jobs, and a listing by tag walks // its index, so it is kept the same way, except that its analyze keeps the // server's statistics target. So is the dependency table, which churns -// with workflows, except that it is analyzed on churn only until it has -// statistics (see liveTable): until then, the planner takes each lookup of +// with workflows, except that churn re-analyzes it at most every +// analyzeDepsEvery: until it has statistics, the planner takes each lookup of // a job's edges for a two-hundredth of the table, and scans the table // instead, in finalizes and in the sweep that promotes dependents (see -// finalizeTailSQL). The counters of all three are read in one statement. -// Each table is kept whether or not another's +// finalizeTailSQL). The counters of all three are read in one statement, +// each table's in a LATERAL subquery that OFFSET 0 keeps from being +// flattened: joined, the view's GROUP BY kept the table's OID from reaching +// pg_class, so every pass summed the counters of every relation in the +// database. Each table is kept whether or not another's // maintenance fails, such as a vacuum of a large hopper_jobs that runs past // the role's statement_timeout on every pass. func (e *Executor) JobsMaintain(ctx context.Context) (driver.JobsMaintainResult, error) { @@ -418,11 +421,11 @@ func (e *Executor) JobsMaintain(ctx context.Context) (driver.JobsMaintainResult, names[i] = e.Schema.Table(t.name) } all, err := scanLiveTableStats(e.Conn.Query(ctx, ` - SELECT n, s.n_dead_tup, s.n_live_tup, s.n_mod_since_analyze, c.reltuples, - coalesce(extract(epoch FROM now() - greatest(s.last_analyze, s.last_autoanalyze)), 1e9), - greatest(s.last_analyze, s.last_autoanalyze) IS NOT NULL - FROM unnest($1::text[]) n - JOIN pg_stat_user_tables s ON s.relid = to_regclass(n) JOIN pg_class c ON c.oid = s.relid`, textArray(names))) + SELECT n, s.n_dead_tup, s.n_live_tup, s.n_mod_since_analyze, s.reltuples, + coalesce(extract(epoch FROM now() - greatest(s.last_analyze, s.last_autoanalyze)), 1e9) + FROM unnest($1::text[]) n, LATERAL ( + SELECT s.*, c.reltuples FROM pg_stat_user_tables s JOIN pg_class c ON c.oid = s.relid + WHERE s.relid = to_regclass(n) OFFSET 0) s`, textArray(names))) if err != nil { return res, fmt.Errorf("hopper: live table statistics: %w", err) } @@ -448,19 +451,18 @@ func (e *Executor) JobsMaintain(ctx context.Context) (driver.JobsMaintainResult, } // liveTableStats are a live table's activity counters, the row count the -// planner holds, and whether it was ever analyzed. A vacuum, or an index -// built over existing rows, also sets the row count, so a count that is -// known does not mean the table has column statistics. +// planner holds, and the time since its last analyze (1e9 s if never: a +// vacuum, or an index built over existing rows, also sets the row count, +// so a count that is known does not mean the table has statistics). type liveTableStats struct { table string dead, live, modified int64 believed, sinceAnalyze float64 - everAnalyzed bool } var scanLiveTableStats = collect(func(r Row) (liveTableStats, error) { var st liveTableStats - err := r.Scan(&st.table, &st.dead, &st.live, &st.modified, &st.believed, &st.sinceAnalyze, &st.everAnalyzed) + err := r.Scan(&st.table, &st.dead, &st.live, &st.modified, &st.believed, &st.sinceAnalyze) return st, err }) @@ -471,21 +473,28 @@ var scanLiveTableStats = collect(func(r Row) (liveTableStats, error) { // only ten. So does the dependency table, whose estimates rest on how many // distinct jobs its sample holds, which a larger sample counts better. var liveTables = []liveTable{ - {"hopper_jobs", 10, true}, - {"hopper_job_tags", 0, true}, - {"hopper_job_deps", 0, false}, + {"hopper_jobs", 10, analyzeEvery}, + {"hopper_job_tags", 0, analyzeEvery}, + {"hopper_job_deps", 0, analyzeDepsEvery}, } -// liveTable is a table JobsMaintain keeps. One without analyzeOnChurn is -// analyzed on churn only until it has been analyzed once, and then only -// when its size is off: what the planner reads from the dependency table, -// how many edges a job has and how many jobs depend on one, does not -// change as workflows come and go, and each analyze has every connection -// replan the finalize statement. +// analyzeDepsEvery is the least time between the dependency table's +// statistics refreshes driven by churn. What the planner reads from it, +// how many edges a job has and how many jobs depend on one, changes only +// as the shape of the workflows does, and each analyze has every +// connection replan the finalize statement; but a table first analyzed +// empty, or holding one large fan-out, has the planner scan it on every +// lookup, so its statistics are still refreshed now and then. A table +// never analyzed is analyzed at once. +const analyzeDepsEvery = 10 * time.Minute + +// liveTable is a table JobsMaintain keeps, with its analyze's statistics +// target (zero for the server's) and the least time between analyzes +// driven by churn. type liveTable struct { name string statisticsTarget int - analyzeOnChurn bool + analyzeEvery time.Duration } func (e *Executor) maintainLiveTable(ctx context.Context, t liveTable, table string, st liveTableStats) (vacuumed, analyzed bool, err error) { @@ -502,7 +511,7 @@ func (e *Executor) maintainLiveTable(ctx context.Context, t liveTable, table str } floor := func(v float64) float64 { return max(v, 100) } sizeOff := st.believed >= 0 && (float64(st.live) >= 10*floor(st.believed) || st.believed >= 10*floor(float64(st.live))) - churned := st.modified >= 1000 && st.modified*10 >= st.live && st.sinceAnalyze >= analyzeEvery.Seconds() && (t.analyzeOnChurn || !st.everAnalyzed) + churned := st.modified >= 1000 && st.modified*10 >= st.live && st.sinceAnalyze >= t.analyzeEvery.Seconds() if !sizeOff && !churned { return false, false, nil } From 9039c969400a4803e3320c55ac539b899ba490ec Mon Sep 17 00:00:00 2001 From: Michael McQuade Date: Mon, 19 Oct 2026 11:39:48 -0500 Subject: [PATCH 16/19] docs: finalize SQL in PLAN checks the running state as text, dependency analyze cadence --- docs/PLAN.md | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/docs/PLAN.md b/docs/PLAN.md index 598327c..5818c26 100644 --- a/docs/PLAN.md +++ b/docs/PLAN.md @@ -1005,12 +1005,12 @@ RETURNING j.*; analyze invalidates every connection's cached plans, and one in the middle of a burst cost a CPU-starved server a fifth of its throughput. The live tags table and `hopper_job_deps` are kept the same way, with the server's sample size, except that - the dependency table is analyzed on churn only until it has been analyzed once (a - vacuum gives the planner its row count but no statistics), and then only when its row - count is off: until it has statistics, the planner takes each lookup of a job's edges - for a two-hundredth of the table, and scans it instead, while what it reads from them - (edges per job, dependents per job) does not change as workflows come and go, and each - analyze has every connection replan the finalize statement. The maintenance never + churn re-analyzes the dependency table at most every ten minutes (at once if it was + never analyzed; a vacuum gives the planner its row count but no statistics): until it + has statistics, the planner takes each lookup of a job's edges for a two-hundredth of + the table, and scans it instead, while what it reads from them (edges per job, + dependents per job) changes only with the shape of the workflows, and each analyze has + every connection replan the finalize statement. The maintenance never waits for a lock. Schema v6 also sets `hopper_jobs`' autovacuum thresholds low with no cost delay. @@ -1082,12 +1082,12 @@ retry AS ( snooze_reason = CASE WHEN r.snooze AND NOT coalesce(j.wake_requested_at >= j.attempted_at, false) THEN r.snooze_reason END, wake_requested_at = NULL FROM r WHERE j.id = r.id AND r.state IN ('retryable','scheduled') - AND j.state = 'running' AND j.attempted_by = r.attempted_by AND j.fence = r.fence + AND j.state::text = 'running' AND j.attempted_by = r.attempted_by AND j.fence = r.fence RETURNING j.id), done AS ( DELETE FROM hopper_jobs j USING r WHERE j.id = r.id AND r.state IN ('completed','canceled','discarded') - AND j.state = 'running' AND j.attempted_by = r.attempted_by AND j.fence = r.fence + AND j.state::text = 'running' AND j.attempted_by = r.attempted_by AND j.fence = r.fence RETURNING j.*, r.state AS final_state, r.error, r.output, r.archive), archived AS (INSERT INTO hopper_job_history SELECT … , now() FROM done WHERE archive) SELECT id FROM retry UNION ALL SELECT id FROM done; From dcfb9de69956716ffc0ce42d8b10ec9feb615a87 Mon Sep 17 00:00:00 2001 From: Michael McQuade Date: Mon, 19 Oct 2026 15:17:32 -0500 Subject: [PATCH 17/19] test(driver): keep the plan test's scan match inside one node, check the dependency analyze cap --- driver/hopperpgx/postgres_test.go | 5 ++++- drivertest/cluster.go | 19 +++++++++++++++++++ 2 files changed, 23 insertions(+), 1 deletion(-) diff --git a/driver/hopperpgx/postgres_test.go b/driver/hopperpgx/postgres_test.go index 15a2d93..0ab5a8b 100644 --- a/driver/hopperpgx/postgres_test.go +++ b/driver/hopperpgx/postgres_test.go @@ -1343,7 +1343,10 @@ func TestFinalizePlansIgnoreStaleStatistics(t *testing.T) { log.mu.Lock() statements, args := log.sql, log.args log.mu.Unlock() - scan := regexp.MustCompile(`"Node Type": "Seq Scan",[^}]*?"Relation Name": "(hopper_jobs|hopper_job_deps|hopper_batches)"`) + // A node's own relation comes before its child plans, so the match + // stays inside the node: a scan of another table whose subplan probes + // one of these is not a scan of it. + scan := regexp.MustCompile(`"Node Type": "Seq Scan",[^{}]*?"Relation Name": "(hopper_jobs|hopper_job_deps|hopper_batches)"`) explained := 0 for i, q := range statements { if !strings.Contains(q, "hopper_jobs") || strings.HasPrefix(strings.TrimSpace(q), "SELECT state FROM") { diff --git a/drivertest/cluster.go b/drivertest/cluster.go index f5f6846..c11a82a 100644 --- a/drivertest/cluster.go +++ b/drivertest/cluster.go @@ -1154,6 +1154,25 @@ func testDepsTableMaintained[TTx any](t *testing.T, f Fixture[TTx]) { t.Fatal(err) } } + // Churn alone re-analyzes it only every ten minutes. + analyzedAt := stat("last_analyze") + if _, err := exec.WorkflowInsert(ctx, driver.WorkflowInsertParams{Name: "fan-out", Jobs: params("step", len(deps)+1), Deps: deps}); err != nil { + t.Fatal(err) + } + for deadline := time.Now().Add(10 * time.Second); stat("n_mod_since_analyze >= 1200") != "true"; time.Sleep(100 * time.Millisecond) { + if time.Now().After(deadline) { + t.Fatal("the second workflow's edges were not counted") + } + } + for range 3 { + if _, err := exec.JobsMaintain(ctx); err != nil { + t.Fatal(err) + } + } + time.Sleep(time.Second) + if got := stat("last_analyze"); got != analyzedAt { + t.Errorf("dependency table analyzed again on churn within ten minutes: %s, then %s", analyzedAt, got) + } } // analyzeEmpty gives the engine statistics that say the live table is From 051c24dc02dded8c9867841abac3c96a1afee2c2 Mon Sep 17 00:00:00 2001 From: Michael McQuade Date: Mon, 19 Oct 2026 15:17:32 -0500 Subject: [PATCH 18/19] perf(driver): read only the live table counters JobsMaintain uses --- driver/internal/pgsql/cluster.go | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/driver/internal/pgsql/cluster.go b/driver/internal/pgsql/cluster.go index 2882aaa..9a7847d 100644 --- a/driver/internal/pgsql/cluster.go +++ b/driver/internal/pgsql/cluster.go @@ -424,7 +424,8 @@ func (e *Executor) JobsMaintain(ctx context.Context) (driver.JobsMaintainResult, SELECT n, s.n_dead_tup, s.n_live_tup, s.n_mod_since_analyze, s.reltuples, coalesce(extract(epoch FROM now() - greatest(s.last_analyze, s.last_autoanalyze)), 1e9) FROM unnest($1::text[]) n, LATERAL ( - SELECT s.*, c.reltuples FROM pg_stat_user_tables s JOIN pg_class c ON c.oid = s.relid + SELECT s.n_dead_tup, s.n_live_tup, s.n_mod_since_analyze, s.last_analyze, s.last_autoanalyze, c.reltuples + FROM pg_stat_user_tables s JOIN pg_class c ON c.oid = s.relid WHERE s.relid = to_regclass(n) OFFSET 0) s`, textArray(names))) if err != nil { return res, fmt.Errorf("hopper: live table statistics: %w", err) @@ -481,11 +482,15 @@ var liveTables = []liveTable{ // analyzeDepsEvery is the least time between the dependency table's // statistics refreshes driven by churn. What the planner reads from it, // how many edges a job has and how many jobs depend on one, changes only -// as the shape of the workflows does, and each analyze has every -// connection replan the finalize statement; but a table first analyzed -// empty, or holding one large fan-out, has the planner scan it on every -// lookup, so its statistics are still refreshed now and then. A table -// never analyzed is analyzed at once. +// as the shape of the workflows does, so a refresh now and then corrects a +// first analyze that sampled the table empty or holding one large fan-out, +// which has the planner scan it on every lookup. The cap bounds the cost +// of sampling the table, not replanning: analyzes of hopper_jobs make +// every connection replan the finalize statement at least as often, and +// autovacuum's own analyzes, at the table's default thresholds, keep the +// statistics fresh in between. A table never analyzed is analyzed without +// the wait, once a thousand of its rows have changed: until then it is +// small enough that a scan costs little. const analyzeDepsEvery = 10 * time.Minute // liveTable is a table JobsMaintain keeps, with its analyze's statistics From e6cd4cbb51758725f089ddef9ecacdcd2d1c74ab Mon Sep 17 00:00:00 2001 From: Michael McQuade Date: Mon, 19 Oct 2026 15:17:32 -0500 Subject: [PATCH 19/19] docs(driver): JobsMaintain also refreshes statistics on churn --- driver/driver.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/driver/driver.go b/driver/driver.go index d1c15ab..c74fe76 100644 --- a/driver/driver.go +++ b/driver/driver.go @@ -323,7 +323,8 @@ type Executor interface { // maintenance passes: it reclaims the space and index entries of jobs // that have run and left once enough have accumulated (a hundred // thousand on Postgres), and refreshes planner statistics when the - // planner's idea of the table's size is off by an order of magnitude. It + // planner's idea of the table's size is off by an order of magnitude or + // enough of the table has changed since the last refresh. It // never waits for a lock, so it costs nothing while the engine's own // pass is running. The live table churns at the job rate, and on // Postgres every claim walks the claim index past the entries of jobs