fix: produce worker claim — SECURITY DEFINER fn fixes 08P01 poll crash + stale-lease reclaim (040)

The store's hand-built claim interpolated the lane predicate into the SQL
text while still passing a bind parameter — node-postgres sent 1 param for
a 0-param statement (08P01) on every poll, so the live produce worker
store-errored every ~2s and never claimed. Migration 040 replaces the
inline claim with produce_job_claim(p_lane, p_stale_after): SECURITY
DEFINER (the 035/oidc_refresh_lookup pattern — the poller is lane-global,
FORCE RLS hides every row without a tenant context), atomic FOR UPDATE
SKIP LOCKED with the queued→transcoding bump as the lease, and stale-
transcoding reclaim so a killed worker's rows return to the queue.
produce_job_workspace(p_job_id) discovers the owning tenant for the
follow-up writes, which stay RLS-bound as sp_app. pg-store rewired; store
errors now surface err.message instead of err.name; pg-store tests added.
38/38 producer tests green.

apps/api/migrations/040_produce_claim.sql

@@ -0,0 +1,88 @@
+-- 040: produce-worker job claim — the queue half of 039's producer feature.
+--
+-- WHY A SECURITY DEFINER CLAIM: video_processing_jobs is FORCE ROW LEVEL SECURITY
+-- with a tenant_isolation policy (009), but the produce worker is a lane-global
+-- poller with NO tenant context: it must see queued rows in EVERY workspace, then
+-- work each claimed row inside that workspace's scope. 006/012/017/035 established
+-- the pattern for exactly this shape — a narrow definer-rights claim that
+-- discovers the tenant and nothing else; the caller enters the tenant context
+-- afterwards and every subsequent statement stays RLS-bound as sp_app.
+--
+-- The same migration also fixes the production bug this claim surfaced
+-- (2026-10 produce worker logging a store error every ~2 s): the store's inline
+-- claim interpolated its lane predicate into the SQL text while still passing a
+-- bind parameter for it, so node-postgres sent `bind message supplies 1
+-- parameters, but prepared statement "" requires 0` (08P01) on EVERY poll. A
+-- stored function with a real $1 takes the driver out of the hand-built-SQL
+-- business entirely; lane becomes a typed parameter that cannot desynchronise.
+--
+-- Leasing: the `queued -> transcoding` bump inside FOR UPDATE SKIP LOCKED is the
+-- lease (no separate lease column). A worker that dies mid-encode leaves its row
+-- in `transcoding` forever, so the claim also requeues rows whose lease has gone
+-- stale (transcoding for longer than p_stale_after with no progress): the row
+-- returns to `queued` and any worker may retake it. Requeue happens BEFORE the
+-- claim sub-select in the same statement snapshot, so a row released by the sweep
+-- is claimable by the very same call.
+
+CREATE FUNCTION produce_job_claim(p_lane text, p_stale_after interval)
+RETURNS TABLE (
+  id uuid,
+  workspace_id uuid,
+  kind text,
+  input_json jsonb,
+  video_asset_id uuid
+)
+LANGUAGE plpgsql SECURITY DEFINER SET search_path = public AS $$
+BEGIN
+  -- Lease reclaim: a crashed/killed worker's rows go back to the queue. The
+  -- progress write (the runner's per-stage checkpoint) is the liveness signal;
+  -- `updated_at` is bumped by every progress/claim write, so `updated_at` older
+  -- than the stale window means the owner is gone, not merely slow.
+  UPDATE video_processing_jobs stale
+     SET state = 'queued', progress_percent = 0, error = NULL, updated_at = now()
+   WHERE stale.state = 'transcoding'
+     AND stale.updated_at < now() - p_stale_after;
+
+  -- The claim: one lane-eligible queued row, locked so a concurrent worker cannot
+  -- take the same one. The kind check + lane check live in the sub-select, so a
+  -- worker never locks a row it would only have to release.
+  RETURN QUERY
+  UPDATE video_processing_jobs j
+     SET state = 'transcoding', progress_percent = 5, error = NULL, updated_at = now()
+   WHERE j.id IN (
+     SELECT c.id
+       FROM video_processing_jobs c
+      WHERE c.kind IN ('compose', 'mix')
+        AND c.state = 'queued'
+        AND (
+              (p_lane = 'remote' AND c.input_json->>'lane' = 'remote')
+           OR (p_lane <> 'remote' AND c.input_json->>'lane' IS DISTINCT FROM 'remote')
+            )
+      ORDER BY c.created_at, c.id
+        FOR UPDATE SKIP LOCKED
+        LIMIT 1
+   )
+  RETURNING j.id, j.workspace_id, j.kind, j.input_json, j.video_asset_id;
+END
+$$;
+
+REVOKE ALL ON FUNCTION produce_job_claim(text, interval) FROM PUBLIC;
+GRANT EXECUTE ON FUNCTION produce_job_claim(text, interval) TO sp_app;
+
+-- The worker's follow-up writes are workspace-scoped already (SET LOCAL
+-- app.workspace_id inside a transaction), so no definer rights are needed there:
+-- FORCE RLS binds those statements as sp_app, which is the point.
+
+-- The claim's follow-up writes (progress / complete / fail / the derived asset
+-- insert) run inside the job's tenant context, so they stay RLS-bound as sp_app —
+-- but the worker only holds a job id at that point, and an ordinary SELECT cannot
+-- discover which workspace owns it (FORCE RLS hides every row without a tenant
+-- context). This is the 012 live_monitor_source_workspace shape verbatim: a narrow
+-- definer lookup that discovers the owning tenant and nothing else.
+CREATE FUNCTION produce_job_workspace(p_job_id uuid) RETURNS uuid
+LANGUAGE sql STABLE SECURITY DEFINER SET search_path = public AS $$
+  SELECT workspace_id FROM video_processing_jobs WHERE id = p_job_id
+$$;
+
+REVOKE ALL ON FUNCTION produce_job_workspace(uuid) FROM PUBLIC;
+GRANT EXECUTE ON FUNCTION produce_job_workspace(uuid) TO sp_app;

apps/media-worker/src/producer/pg-store.test.ts

@@ -0,0 +1,284 @@
+/**
+ * Postgres produce job store, against a fake pool: the claim must go through the
+ * 040 SECURITY DEFINER function (never hand-built lane SQL — that is the bug that
+ * logged `bind message supplies 1 parameters, but prepared statement "" requires 0`
+ * every ~2 s in production), and every write after the claim must run inside the
+ * job's tenant transaction (`SET LOCAL app.workspace_id`) so FORCE RLS binds it.
+ *
+ * The fake pool is deliberately dumb: it records (sql, params) verbatim and
+ * replies from a script, so a claim whose SQL text and parameter list disagree —
+ * the exact production failure — cannot pass by accident.
+ */
+import assert from "node:assert/strict";
+import test from "node:test";
+import type { Pool as PgPool, PoolClient, QueryResult } from "pg";
+import { PostgresProduceJobStore, PRODUCE_LEASE_STALE_AFTER } from "./pg-store";
+import type { ClaimedProduceJob } from "./store";
+
+const JOB = "11111111-1111-4111-8111-111111111111";
+const ASSET = "22222222-2222-4222-8222-222222222222";
+const WS = "8f0c5a8e-3b1d-4c3e-9a5b-0c1d2e3f4a5b";
+const VOICEOVER = "33333333-3333-4333-8333-333333333333";
+const DERIVED = "44444444-4444-4444-8444-444444444444";
+const WORKSPACE_OF = "SELECT produce_job_workspace($1) AS workspace_id";
+
+interface Call { sql: string; params: unknown[] }
+
+function empty(): QueryResult<never> {
+  return { rows: [], rowCount: 0, command: "", oid: 0, fields: [] };
+}
+
+/** One scripted reply for a single statement. */
+type Reply = { rows: Array<Record<string, unknown>> } | { error: Error };
+
+/**
+ * A PoolClient double: the store only drives `query` and `release`, so the rest
+ * of the interface is an inert stub rather than a re-implementation.
+ */
+function fakeClient(): { bound: PoolClient; calls: Call[]; script: Array<Reply>; released: () => number } {
+  const calls: Call[] = [];
+  const script: Array<Reply> = [];
+  let releaseCount = 0;
+  const client = {
+    async query(sql: string, params: unknown[] = []): Promise<QueryResult> {
+      calls.push({ sql, params });
+      // Transaction control (BEGIN/COMMIT/ROLLBACK and the tenant set_config) is
+      // synthesized: a script queues replies for the transaction's REAL statements
+      // only, in order.
+      if (/^(BEGIN|COMMIT|ROLLBACK)\b/.test(sql.trim()) || /set_config\('app\.workspace_id'/.test(sql)) {
+        return empty();
+      }
+      const reply = script.shift();
+      if (reply !== undefined && "error" in reply) throw reply.error;
+      return { ...empty(), rowCount: reply?.rows.length ?? 0, rows: reply?.rows ?? [] };
+    },
+    async release(): Promise<void> { releaseCount += 1; },
+  } as unknown as PoolClient;
+  return { bound: client, calls, script, released: () => releaseCount };
+}
+
+/**
+ * Scripted pool: pool-level statements (`produce_job_claim`, the workspace
+ * lookup) draw from `reply(...)`; each `connect()` hands out the client whose
+ * script was queued by `expectClient(...)`, in order.
+ */
+class FakePool {
+  readonly calls: Call[] = [];
+  readonly clients: Array<{ calls: Call[]; script: Array<Reply>; released: () => number; bound: PoolClient }> = [];
+  private readonly scripted: Array<Reply> = [];
+
+  reply(rows: Array<Record<string, unknown>>): void { this.scripted.push({ rows }) }
+  replyError(error: Error): void { this.scripted.push({ error }) }
+
+  /** Queue the transaction script for the NEXT client the store borrows. */
+  expectClient(...replies: Reply[]): { calls: Call[]; script: Array<Reply>; released: () => number; bound: PoolClient } {
+    const doubled = fakeClient();
+    doubled.script.push(...replies);
+    this.clients.push(doubled);
+    return doubled;
+  }
+
+  async query(sql: string, params: unknown[] = []): Promise<QueryResult> {
+    this.calls.push({ sql, params });
+    const reply = this.scripted.shift();
+    if (reply !== undefined && "error" in reply) throw reply.error;
+    return { ...empty(), rowCount: reply?.rows.length ?? 0, rows: reply?.rows ?? [] };
+  }
+
+  async connect(): Promise<PoolClient> {
+    const doubled = this.clients.shift();
+    if (doubled !== undefined) return doubled.bound;
+    const fresh = fakeClient();
+    this.clients.push(fresh);
+    return fresh.bound;
+  }
+}
+
+function storeOf(pool: FakePool): PostgresProduceJobStore {
+  return new PostgresProduceJobStore(pool as unknown as PgPool);
+}
+
+function claimRow(overrides: Record<string, unknown> = {}): Record<string, unknown> {
+  return {
+    id: JOB, workspace_id: WS, kind: "compose",
+    input_json: { overlays: [{ kind: "lower_third", text: "Ada Lovelace" }], lane: "local" },
+    video_asset_id: ASSET,
+    ...overrides,
+  };
+}
+
+function assetRow(): Record<string, unknown> {
+  return {
+    id: ASSET, storage_key: "sha256/" + "a".repeat(64) + ".mp4", mime_type: "video/mp4",
+    size_bytes: 12, width: 320, height: 180, duration_ms: 2000, frame_rate: 30, audio_codec: "aac",
+  };
+}
+
+/** The rows one healthy claim's tenant transaction sees: the source asset. */
+function healthyClaim(): { pool: FakePool; client: ReturnType<FakePool["expectClient"]> } {
+  const pool = new FakePool();
+  const client = pool.expectClient({ rows: [assetRow()] });
+  pool.reply([claimRow()]);
+  return { pool, client };
+}
+
+test("claim goes through the definer function with typed lane + lease params", async () => {
+  const { pool } = healthyClaim();
+
+  const job = await storeOf(pool).claim("local");
+
+  const claim = pool.calls[0]!;
+  assert.match(claim.sql, /FROM produce_job_claim\(\$1, \$2::interval\)/, "claim must be the 040 definer function");
+  assert.equal(claim.params.length, 2, "claim SQL has exactly $1 and $2 — no stray binds (the 08P01 bug)");
+  assert.equal(claim.params[0], "local");
+  assert.equal(claim.params[1], PRODUCE_LEASE_STALE_AFTER);
+  assert.ok(job !== null);
+  assert.equal(job!.jobId, JOB);
+  assert.equal(job!.workspaceId, WS);
+  assert.equal(job!.source.storageKey.startsWith("sha256/"), true);
+  assert.equal(job!.source.hasAudio, true);
+  assert.equal(job!.request.overlays[0]!.kind, "lower_third");
+});
+
+test("claim passes the lane through: a remote worker binds lane='remote'", async () => {
+  const { pool } = healthyClaim();
+  await storeOf(pool).claim("remote");
+  assert.equal(pool.calls[0]!.params[0], "remote");
+});
+
+test("claim SQL never interpolates the lane predicate into the statement text", async () => {
+  const { pool } = healthyClaim();
+  await storeOf(pool).claim("local");
+  const sql = pool.calls[0]!.sql;
+  assert.equal(/lane IS DISTINCT FROM/.test(sql), false, "lane filtering belongs to the function, not hand-built SQL");
+  assert.equal(sql.includes("$3"), false, "the statement's parameter count matches its placeholders");
+});
+
+test("an empty claim returns null and opens no tenant transaction", async () => {
+  const pool = new FakePool();
+  pool.reply([]);
+  const job = await storeOf(pool).claim("local");
+  assert.equal(job, null);
+  assert.equal(pool.calls.length, 1, "an idle poll is exactly one statement");
+});
+
+test("every post-claim write runs inside SET LOCAL app.workspace_id", async () => {
+  const pool = new FakePool();
+  const store = storeOf(pool);
+
+  // progress: workspace discovery, then the tenant transaction
+  pool.reply([{ workspace_id: WS }]);
+  const progress = pool.expectClient();
+  await store.progress(JOB, 50);
+  assert.equal(pool.calls[0]!.sql, WORKSPACE_OF, "workspace discovery must be the 040 definer lookup");
+  assert.deepEqual(pool.calls[0]!.params, [JOB]);
+  const setLocal = progress.calls.find((c) => /set_config\('app\.workspace_id'/.test(c.sql))!;
+  assert.ok(setLocal !== undefined, "tenant context is set inside the transaction");
+  assert.equal(setLocal.params[0], WS);
+  assert.match(progress.calls[0]!.sql, /BEGIN/);
+  assert.match(progress.calls.at(-1)!.sql, /COMMIT/);
+  assert.ok(progress.calls.some((c) => /UPDATE video_processing_jobs[\s\S]*state = 'transcoding'/.test(c.sql)));
+  assert.equal(progress.released(), 1);
+
+  // fail: same shape
+  pool.reply([{ workspace_id: WS }]);
+  const fail = pool.expectClient();
+  await store.fail(JOB, "produce encode failed");
+  assert.ok(fail.calls.some((c) => c.params.includes("produce encode failed")));
+
+  // complete: derived asset insert + job update, both inside the tenant scope
+  pool.reply([{ workspace_id: WS }]);
+  const complete = pool.expectClient({ rows: [{ id: DERIVED }] }, { rows: [{ id: JOB }] });
+  await store.complete(JOB, {
+    storageKey: "sha256/" + "b".repeat(64) + ".mp4", sizeBytes: 99, durationMs: 2000,
+    width: 1280, height: 720, frameRate: 30,
+  }, ASSET);
+  const insert = complete.calls.find((c) => /INSERT INTO video_assets/.test(c.sql))!;
+  assert.equal(insert.params[0], WS, "the derived row is written under the job's workspace");
+  assert.ok(complete.calls.some((c) => /produced_asset_id = \$3/.test(c.sql)));
+});
+
+test("complete throws when the job has lost its claim", async () => {
+  const pool = new FakePool();
+  pool.reply([{ workspace_id: WS }]);
+  pool.expectClient({ rows: [{ id: DERIVED }] }, { rows: [] }); // the UPDATE matches nothing
+  await assert.rejects(
+    () => storeOf(pool).complete(JOB, {
+      storageKey: "sha256/" + "b".repeat(64) + ".mp4", sizeBytes: 1, durationMs: 1,
+      width: 1280, height: 720, frameRate: 30,
+    }, ASSET),
+    /lost its claim/,
+  );
+});
+
+test("an undecodable input_json fails the row instead of stranding it in transcoding", async () => {
+  const pool = new FakePool();
+  pool.reply([claimRow({ input_json: { overlays: [{ kind: "clock", text: "not allowed here" }] } })]);
+  const client = pool.expectClient();
+
+  const job = await storeOf(pool).claim("local");
+
+  assert.equal(job, null, "an invalid row claims nothing for this tick");
+  const failing = client.calls.find((c) => /state = 'failed'/.test(c.sql))!;
+  assert.ok(failing !== undefined, "the row is failed inside the claim");
+  assert.equal(failing.params[0], JOB);
+  assert.match(failing.params[1] as string, /invalid request/);
+  assert.match(client.calls.at(-1)!.sql, /COMMIT/, "the failure commits (a null return, not a throw)");
+});
+
+test("a claimed row whose source asset is missing fails with asset unavailable", async () => {
+  const pool = new FakePool();
+  pool.reply([claimRow()]);
+  // The asset lookup sees nothing, then the failure UPDATE matches the claimed row.
+  const client = pool.expectClient({ rows: [] }, { rows: [{ id: JOB }] });
+
+  const job = await storeOf(pool).claim("local");
+  assert.equal(job, null);
+  const failing = client.calls.find((c) => /state = 'failed'/.test(c.sql))!;
+  assert.match(failing.sql, /asset unavailable/, "the reason is written on the row");
+  assert.deepEqual(failing.params, [JOB]);
+});
+
+test("a claim that returns an unusable workspace id throws a bounded store error", async () => {
+  const pool = new FakePool();
+  pool.reply([claimRow({ workspace_id: "not-a-uuid" })]);
+  await assert.rejects(() => storeOf(pool).claim("local"), /invalid workspace/);
+});
+
+test("progress for a job whose row no longer resolves throws job not found", async () => {
+  const pool = new FakePool();
+  pool.reply([]); // produce_job_workspace returns NULL
+  await assert.rejects(() => storeOf(pool).progress(JOB, 50), /job .* not found/);
+});
+
+test("a store/transport error inside the tenant transaction rolls back and releases the client", async () => {
+  const pool = new FakePool();
+  pool.reply([{ workspace_id: WS }]);
+  const client = pool.expectClient({ error: new Error("connection terminated") });
+
+  await assert.rejects(() => storeOf(pool).progress(JOB, 50), /connection terminated/);
+  assert.ok(client.calls.some((c) => /ROLLBACK/.test(c.sql)), "the transaction is rolled back");
+  assert.equal(client.released(), 1, "the pooled client is always released");
+});
+
+test("asset lookups are tenant-scoped too", async () => {
+  const pool = new FakePool();
+  const client = pool.expectClient({ rows: [{ storage_key: "sha256/" + "c".repeat(64) + ".m4a", mime_type: "audio/mp4" }] });
+  const found = await storeOf(pool).asset(WS, VOICEOVER);
+  assert.deepEqual(found, { storageKey: "sha256/" + "c".repeat(64) + ".m4a", mimeType: "audio/mp4" });
+  const setLocal = client.calls.find((c) => /app\.workspace_id/.test(c.sql))!;
+  assert.equal(setLocal.params[0], WS);
+});
+
+test("a mix row's voiceover request survives the claim", async () => {
+  const pool = new FakePool();
+  pool.expectClient({ rows: [assetRow()] });
+  pool.reply([claimRow({
+    kind: "mix",
+    input_json: { voiceover: { assetId: VOICEOVER, gainDb: -6, duckOriginal: true } },
+  })]);
+  const job: ClaimedProduceJob | null = await storeOf(pool).claim("local");
+  assert.equal(job!.kind, "mix");
+  assert.equal(job!.request.voiceover!.assetId, VOICEOVER);
+  assert.equal(job!.request.voiceover!.duckOriginal, true);
+});

apps/media-worker/src/producer/pg-store.ts

@@ -1,16 +1,16 @@
 /**
  * Postgres produce job store: the poll/claim/complete half of the produce worker.
  *
- * - Claim: `UPDATE ... SET state='transcoding' WHERE id IN (SELECT id ... FOR UPDATE
+ * - Claim: `SELECT * FROM produce_job_claim(lane, stale_after)` — a SECURITY
- *   SKIP LOCKED) RETURNING` — one row, one statement, lane-filtered, so concurrent
+ *   DEFINER function (040, the 035 submission_box_lookup / 017 invite_link_for_claim
- *   workers never race and a claimed row is invisible to everyone else until its
+ *   pattern) because the worker polls with NO tenant context while
- *   owner finishes it. The `queued → transcoding` bump inside SKIP LOCKED is the
+ *   video_processing_jobs is FORCE ROW LEVEL SECURITY (009 tenant_isolation). The
- *   lease: there is no separate lease column to renew, because a claimed row is
+ *   definer claim is the one statement that may cross tenants: it bumps
- *   owned until its owner writes complete/failed.
+ *   `queued → transcoding` inside FOR UPDATE SKIP LOCKED as the lease (no separate
- * - Every write happens inside the job's tenant transaction (`app.workspace_id`),
+ *   lease column) and requeues rows whose lease has gone stale, then returns the
- *   because both tables are FORCE ROW LEVEL SECURITY with a `tenant_isolation`
+ *   claimed row WITH its workspace_id.
- *   policy (009). The claim reads the workspace off the row first, then re-claims
+ * - Every write after the claim happens inside the job's tenant transaction
- *   inside that tenant's scope.
+ *   (`SET LOCAL app.workspace_id`), so those statements stay RLS-bound as sp_app.
  * - A row whose `input_json` or source asset fails re-validation is FAILED by THE
  *   CLAIM ITSELF (never thrown out of `claim`), so an undecodable row can never sit
  *   in `transcoding` forever holding its lease.
@@ -23,19 +23,20 @@ import type { ClaimedProduceJob, ProduceJobStore, ProduceOutput, ProduceSourceAs
 const KINDS: readonly ProduceJobKind[] = ["compose", "mix"];
 
 /**
- * Lane → which rows this worker may claim (039 lanes: default worker = local or NULL,
+ * A claimed `transcoding` row older than this is assumed ownerless (the worker
- * i.e. anything but an explicit `lane:'remote'`; a remote worker takes only remote).
+ * died mid-encode) and is requeued by the claim itself. Long enough that no real
+ * encode loses its lease to the sweep: the runner reports progress at every stage
+ * (claimed/downloaded/encoded/published) and each write bumps `updated_at`, so a
+ * live lease is touched far more often than this window.
  */
-export function lanePredicate(lane: Lane): { laneSql: string; params: [boolean] } {
+export const PRODUCE_LEASE_STALE_AFTER = "45 minutes";
-  // $1 binds the boolean flag: a local/default worker skips remote rows, a remote
+
-  // worker claims ONLY remote rows. The predicate never matches a local row to a
+interface ClaimRow {
-  // remote worker, so nothing is locked and then released.
+  id: string;
-  return {
+  workspace_id: string;
-    laneSql: lane === "remote"
+  kind: string;
-      ? "(j.input_json->>'lane' = 'remote')"
+  input_json: unknown;
-      : "(j.input_json->>'lane' IS DISTINCT FROM 'remote')",
+  video_asset_id: string;
-    params: [lane === "remote"],
-  };
 }
 
 function uuid(value: unknown, field: string): string {
@@ -76,6 +77,8 @@ export class PostgresProduceJobStore implements ProduceJobStore {
     const client = await this.pool.connect();
     try {
       await client.query("BEGIN");
+      // SET LOCAL: the tenant context dies with this transaction, so a pooled
+      // client can never leak one workspace's scope into the next borrower.
       await client.query("SELECT set_config('app.workspace_id', $1, true)", [workspaceId]);
       const result = await fn(client);
       await client.query("COMMIT");
@@ -89,69 +92,56 @@ export class PostgresProduceJobStore implements ProduceJobStore {
   }
 
   async claim(lane: Lane): Promise<ClaimedProduceJob | null> {
-    // Lane filter in the sub-select, so a worker never locks a row it would have to
+    // Lane and the stale-lease window are typed bind parameters of the definer
-    // release: a remote worker only ever locks remote rows, the default worker only
+    // function (040): never interpolated into the SQL text, so the parameter list
-    // ever locks non-remote rows.
+    // and the statement can no longer disagree (the 08P01 bind-message bug).
-    const { laneSql, params } = lanePredicate(lane);
+    const claimed = await this.pool.query<ClaimRow>(
-    const claimed = await this.pool.query<{ id: string; workspace_id: string }>(
+      "SELECT id, workspace_id, kind, input_json, video_asset_id FROM produce_job_claim($1, $2::interval)",
-      `UPDATE video_processing_jobs j
+      [lane, PRODUCE_LEASE_STALE_AFTER],
-         SET state = 'transcoding', progress_percent = 5, error = NULL, updated_at = now()
-       WHERE j.id IN (
-         SELECT id FROM video_processing_jobs
-          WHERE kind IN ('compose', 'mix')
-            AND state = 'queued'
-            AND ${laneSql}
-          ORDER BY created_at, id
-          FOR UPDATE SKIP LOCKED
-          LIMIT 1
-       )
-       RETURNING j.id, j.workspace_id`,
-      params,
     );
     if (claimed.rowCount === 0) return null;
     const row = claimed.rows[0]!;
     const workspaceId = uuid(row.workspace_id, "workspace id");
+    const kind = row.kind === "mix" ? "mix" : row.kind === "compose" ? "compose" : null;
+    let request = null;
+    if (kind !== null) {
+      try {
+        request = parseProduceRequest(row.input_json, kind);
+      } catch {
+        request = null;
+      }
+    }
+    if (kind === null || request === null) {
+      // A row the worker cannot honour must FAIL here — never strand in `transcoding`.
+      // A null return (not a throw) commits the failure instead of rolling it back.
+      await this.tenant(workspaceId, (client) =>
+        client.query(
+          `UPDATE video_processing_jobs
+              SET state = 'failed', progress_percent = 100, error = $2, updated_at = now()
+            WHERE id = $1 AND state = 'transcoding'`,
+          [row.id, kind === null ? "produce job store: unknown kind" : "produce invalid request"],
+        ));
+      return null;
+    }
 
     return this.tenant<ClaimedProduceJob | null>(workspaceId, async (client) => {
-      // Re-claim inside the tenant's RLS scope (the pool-level row above is outside
-      // it; claiming again is the cheap way to stay correct under FORCE RLS).
-      const mine = await client.query(
-        `UPDATE video_processing_jobs
-            SET state = 'transcoding', progress_percent = 5, error = NULL, updated_at = now()
-          WHERE id = $1 AND state IN ('queued', 'transcoding')
-          RETURNING id, kind, input_json, video_asset_id`,
-        [row.id],
-      );
-      if (mine.rowCount === 0) return null;
-      const job = mine.rows[0]! as Record<string, unknown>;
-      const kind = job.kind === "mix" ? "mix" : job.kind === "compose" ? "compose" : null;
-      let request = null;
-      if (kind !== null) {
-        try {
-          request = parseProduceRequest(job.input_json, kind);
-        } catch {
-          request = null;
-        }
-      }
-      if (kind === null || request === null) {
-        // A row the worker cannot honour must FAIL here — never strand in `transcoding`.
-        // Written inside this transaction and COMMITTED by returning null (not thrown:
-        // a throw would roll the failure back and re-strand the row).
-        await this.failInvalid(client, job, kind === null ? "produce job store: unknown kind" : "produce invalid request");
-        return null;
-      }
-
       const assets = await client.query(
         `SELECT id, storage_key, mime_type, size_bytes, width, height, duration_ms, frame_rate, audio_codec
            FROM video_assets WHERE id = $1`,
-        [job.video_asset_id],
+        [row.video_asset_id],
       );
       if (assets.rowCount === 0) {
-        await this.failInvalid(client, job, "produce asset unavailable");
+        // Same rule as an undecodable payload: fail the row, release the lease.
+        await client.query(
+          `UPDATE video_processing_jobs
+              SET state = 'failed', progress_percent = 100, error = 'produce asset unavailable', updated_at = now()
+            WHERE id = $1 AND state = 'transcoding'`,
+          [row.id],
+        );
         return null;
       }
       return {
-        jobId: uuid(job.id, "job id"),
+        jobId: uuid(row.id, "job id"),
         workspaceId,
         kind,
         request,
@@ -161,20 +151,6 @@ export class PostgresProduceJobStore implements ProduceJobStore {
     });
   }
 
-  /**
-   * Fail a just-claimed row whose payload cannot be honoured, inside the claim's
-   * own transaction so the failure commits with it. Callers then return `null`:
-   * the row is done (failed), there is simply nothing for this tick to encode.
-   */
-  private async failInvalid(client: PoolClient, job: Record<string, unknown>, reason: string): Promise<void> {
-    await client.query(
-      `UPDATE video_processing_jobs
-          SET state = 'failed', progress_percent = 100, error = $2, updated_at = now()
-        WHERE id = $1 AND state = 'transcoding'`,
-      [job.id, reason],
-    );
-  }
-
   async progress(jobId: string, percent: number): Promise<void> {
     await this.tenant(await this.workspaceOf(jobId), async (client) => {
       await client.query(
@@ -231,14 +207,21 @@ export class PostgresProduceJobStore implements ProduceJobStore {
     });
   }
 
-  /** A claimed job's workspace, for the tenant-scoped follow-up writes. */
+  /**
+   * A claimed job's workspace, for the tenant-scoped follow-up writes: the 040
+   * definer lookup (012's live_monitor_source_workspace shape), because an
+   * ordinary SELECT as sp_app with no tenant context sees no rows at all under
+   * FORCE RLS. `job not found` here is exactly what it means: the row is gone
+   * (tenant-deleted) or the lease was reclaimed and the job finished elsewhere.
+   */
   private async workspaceOf(jobId: string): Promise<string> {
     const rows = await this.pool.query<{ workspace_id: string }>(
-      `SELECT workspace_id FROM video_processing_jobs WHERE id = $1`,
+      "SELECT produce_job_workspace($1) AS workspace_id",
       [jobId],
     );
-    if (rows.rowCount === 0) throw new Error(`produce job store: job ${jobId} not found`);
+    const workspaceId = rows.rows[0]?.workspace_id;
-    return uuid(rows.rows[0]!.workspace_id, "workspace id");
+    if (workspaceId === undefined || workspaceId === null) throw new Error(`produce job store: job ${jobId} not found`);
+    return uuid(workspaceId, "workspace id");
   }
 }
 

apps/media-worker/src/producer/runner.test.ts

@@ -193,4 +193,59 @@ test("scratch is removed after a job, success or failure", async () => {
   assert.equal(leftovers.stdout.trim(), "");
 });
 
+test("a store error is reported with the driver's real message, not just its name", async () => {
+  const d = doubles();
+  // node-postgres errors carry name:"error" and the cause in the message — the
+  // production symptom was detail:"error" with the cause swallowed.
+  const boom = new Error("bind message supplies 1 parameters, but prepared statement \"\" requires 0");
+  boom.name = "error";
+  const broken: ProduceJobStore = { ...d.store, claim: async () => { throw boom; } };
+  // pollOnce is the direct path (the loop swallows into the same event), so the
+  // rejection itself proves the error escaped the store call.
+  await assert.rejects(() => runner(d, { jobs: broken }).pollOnce(), /bind message supplies/);
+  const looping = runner(d, { jobs: broken, pollIntervalMs: 5, scheduler: { setTimeout: (_ms, action) => { const t = setTimeout(action, 0); return { cancel: () => clearTimeout(t) }; } } });
+  looping.start();
+  await new Promise((resolve) => setTimeout(resolve, 30));
+  await looping.stop();
+  const event = d.events.find((e) => e.type === "store-error");
+  assert.ok(event !== undefined);
+  assert.match(event.detail ?? "", /bind message supplies 1 parameters/);
+  assert.notEqual(event.detail, "error");
+});
+
+test("a store error while failing a job is also reported with its real message", async () => {
+  const d = doubles();
+  d.setQueue([job("sha256/" + "0".repeat(64) + ".mp4", "compose", {})]); // the input object does not exist
+  const boom = new Error("connection terminated unexpectedly");
+  boom.name = "error";
+  const failingStore: ProduceJobStore = { ...d.store, fail: async () => { throw boom; } };
+  const r = runner(d, { jobs: failingStore });
+  await r.pollOnce(); // run() swallows the fail() error into an event
+  const event = d.events.find((e) => e.type === "store-error");
+  assert.ok(event !== undefined);
+  assert.match(event.detail ?? "", /connection terminated/);
+});
+
+test("the poll loop survives a store error and idles on the next tick", async () => {
+  const d = doubles();
+  let failures = 0;
+  const flaky: ProduceJobStore = {
+    ...d.store,
+    claim: async (lane) => { if (failures++ < 2) throw new Error("connection refused"); return d.store.claim(lane); },
+  };
+  const r = new ProduceRunner({
+    jobs: flaky, storage: d.storage, fontFile: FONT, scratchRoot: SCRATCH, lane: "remote",
+    redactor: { redact: (s) => s }, onStderr: () => undefined, onEvent: (e) => d.events.push(e),
+    pollIntervalMs: 5, scheduler: { setTimeout: (_ms, action) => { const t = setTimeout(action, 0); return { cancel: () => clearTimeout(t) }; } },
+  });
+  r.start();
+  await new Promise((resolve) => setTimeout(resolve, 30));
+  await r.stop();
+  assert.equal(d.laneOf(), "remote");
+  const storeErrors = d.events.filter((e) => e.type === "store-error");
+  assert.ok(storeErrors.length >= 2);
+  for (const event of storeErrors) assert.match(event.detail ?? "", /connection refused/);
+  assert.ok(d.events.some((e) => e.type === "idle"));
+});
+
 test.after(() => { rmSync(SCRATCH, { recursive: true, force: true }); });

apps/media-worker/src/producer/runner.ts

@@ -97,7 +97,10 @@ export class ProduceRunner {
       }
     } catch (error: unknown) {
       // A store/poll error must never kill the loop (the supervisor's DEGRADED path).
-      this.deps.onEvent?.({ type: "store-error", detail: error instanceof Error ? error.name : "unknown error" });
+      // The detail is the real driver message: `error.name` is node-postgres's
+      // generic "error" for everything, which is how this event was being logged
+      // as detail:"error" in production with the actual cause swallowed.
+      this.deps.onEvent?.({ type: "store-error", detail: errorMessage(error) });
     }
     this.scheduleNext();
   }
@@ -134,7 +137,9 @@ export class ProduceRunner {
       const reason = this.reason(error);
       this.deps.onStderr(`produce job ${job.jobId}: ${reason}`);
       await this.deps.jobs.fail(job.jobId, reason).catch((failError: unknown) => {
-        this.deps.onEvent?.({ type: "store-error", jobId: job.jobId, detail: failError instanceof Error ? failError.name : "unknown error" });
+        // A failed `fail()` leaves the row in `transcoding`; the claim's stale-lease
+        // sweep (040) is what requeues it, so say what actually went wrong.
+        this.deps.onEvent?.({ type: "store-error", jobId: job.jobId, detail: errorMessage(failError) });
       });
       this.deps.onEvent?.({ type: "failed", jobId: job.jobId, detail: reason });
     } finally {
@@ -267,6 +272,12 @@ export class ProduceRunner {
   }
 }
 
+/** The real message behind a store failure (err.name alone is "error" for pg). */
+function errorMessage(error: unknown): string {
+  if (error instanceof Error && error.message.trim() !== "") return error.message.slice(0, 300);
+  return error instanceof Error && error.name ? error.name : "unknown error";
+}
+
 function withTimeout<T>(promise: Promise<T>, ms: number, name: ProduceFailure): Promise<T> {
   return new Promise<T>((resolve, reject) => {
     const timer = setTimeout(() => reject(produceError(name)), ms);