diff --git a/apps/api/migrations/040_produce_claim.sql b/apps/api/migrations/040_produce_claim.sql new file mode 100644 index 0000000..31b20b5 --- /dev/null +++ b/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; diff --git a/apps/media-worker/src/producer/pg-store.test.ts b/apps/media-worker/src/producer/pg-store.test.ts new file mode 100644 index 0000000..c598c21 --- /dev/null +++ b/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 { + return { rows: [], rowCount: 0, command: "", oid: 0, fields: [] }; +} + +/** One scripted reply for a single statement. */ +type Reply = { rows: Array> } | { 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; released: () => number } { + const calls: Call[] = []; + const script: Array = []; + let releaseCount = 0; + const client = { + async query(sql: string, params: unknown[] = []): Promise { + 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 { 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; released: () => number; bound: PoolClient }> = []; + private readonly scripted: Array = []; + + reply(rows: Array>): 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; released: () => number; bound: PoolClient } { + const doubled = fakeClient(); + doubled.script.push(...replies); + this.clients.push(doubled); + return doubled; + } + + async query(sql: string, params: unknown[] = []): Promise { + 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 { + 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 = {}): Record { + 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 { + 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 } { + 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); +}); diff --git a/apps/media-worker/src/producer/pg-store.ts b/apps/media-worker/src/producer/pg-store.ts index 117a585..7358ab3 100644 --- a/apps/media-worker/src/producer/pg-store.ts +++ b/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 - * SKIP LOCKED) RETURNING` — one row, one statement, lane-filtered, so concurrent - * workers never race and a claimed row is invisible to everyone else until its - * owner finishes it. The `queued → transcoding` bump inside SKIP LOCKED is the - * lease: there is no separate lease column to renew, because a claimed row is - * owned until its owner writes complete/failed. - * - Every write happens inside the job's tenant transaction (`app.workspace_id`), - * because both tables are FORCE ROW LEVEL SECURITY with a `tenant_isolation` - * policy (009). The claim reads the workspace off the row first, then re-claims - * inside that tenant's scope. + * - Claim: `SELECT * FROM produce_job_claim(lane, stale_after)` — a SECURITY + * DEFINER function (040, the 035 submission_box_lookup / 017 invite_link_for_claim + * pattern) because the worker polls with NO tenant context while + * video_processing_jobs is FORCE ROW LEVEL SECURITY (009 tenant_isolation). The + * definer claim is the one statement that may cross tenants: it bumps + * `queued → transcoding` inside FOR UPDATE SKIP LOCKED as the lease (no separate + * lease column) and requeues rows whose lease has gone stale, then returns the + * claimed row WITH its workspace_id. + * - Every write after the claim happens inside the job's tenant transaction + * (`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, - * i.e. anything but an explicit `lane:'remote'`; a remote worker takes only remote). + * A claimed `transcoding` row older than this is assumed ownerless (the worker + * 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] } { - // $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 - // remote worker, so nothing is locked and then released. - return { - laneSql: lane === "remote" - ? "(j.input_json->>'lane' = 'remote')" - : "(j.input_json->>'lane' IS DISTINCT FROM 'remote')", - params: [lane === "remote"], - }; +export const PRODUCE_LEASE_STALE_AFTER = "45 minutes"; + +interface ClaimRow { + id: string; + workspace_id: string; + kind: string; + input_json: unknown; + video_asset_id: string; } 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 { - // Lane filter in the sub-select, so a worker never locks a row it would have to - // release: a remote worker only ever locks remote rows, the default worker only - // ever locks non-remote rows. - const { laneSql, params } = lanePredicate(lane); - const claimed = await this.pool.query<{ id: string; workspace_id: string }>( - `UPDATE video_processing_jobs j - 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, + // Lane and the stale-lease window are typed bind parameters of the definer + // function (040): never interpolated into the SQL text, so the parameter list + // and the statement can no longer disagree (the 08P01 bind-message bug). + const claimed = await this.pool.query( + "SELECT id, workspace_id, kind, input_json, video_asset_id FROM produce_job_claim($1, $2::interval)", + [lane, PRODUCE_LEASE_STALE_AFTER], ); 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(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; - 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, reason: string): Promise { - 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 { 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 { 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`); - return uuid(rows.rows[0]!.workspace_id, "workspace id"); + const workspaceId = rows.rows[0]?.workspace_id; + if (workspaceId === undefined || workspaceId === null) throw new Error(`produce job store: job ${jobId} not found`); + return uuid(workspaceId, "workspace id"); } } diff --git a/apps/media-worker/src/producer/runner.test.ts b/apps/media-worker/src/producer/runner.test.ts index eb38332..112f585 100644 --- a/apps/media-worker/src/producer/runner.test.ts +++ b/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 }); }); diff --git a/apps/media-worker/src/producer/runner.ts b/apps/media-worker/src/producer/runner.ts index 7b748f7..bb0bef2 100644 --- a/apps/media-worker/src/producer/runner.ts +++ b/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(promise: Promise, ms: number, name: ProduceFailure): Promise { return new Promise((resolve, reject) => { const timer = setTimeout(() => reject(produceError(name)), ms);