fix(driver): keep finalize plans stable under stale statistics

.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:
@@ -49,9 +51,10 @@ 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"
+      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

docs/PLAN.md

@@ -967,7 +967,27 @@ 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 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, 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. 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
@@ -980,10 +1000,19 @@ 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, except that
+  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.
 
 - **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
@@ -1053,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;

driver/driver.go

@@ -323,16 +323,18 @@ 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
 	// 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)
 }
 

driver/hopperpgx/postgres_test.go

@@ -1201,3 +1201,169 @@ 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. 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`)
+	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)
+	}
+	// 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 {
+		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()
+	// 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") {
+			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)
+	}
+}