glass beadian
5808e89dde
Audio-less inputs no longer emit [0:a] maps that ffmpeg rejects; the worker branches to an anullsrc bed configured at the output preset's rate/layout. Producer tests 42/42.
235 lines
10 KiB
TypeScript
235 lines
10 KiB
TypeScript
/**
|
|
* Postgres produce job store: the poll/claim/complete half of the produce worker.
|
|
*
|
|
* - 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.
|
|
*/
|
|
import type { Pool, PoolClient } from "pg";
|
|
import type { Lane, ProduceJobKind } from "./command-builder";
|
|
import { parseProduceRequest } from "./command-builder";
|
|
import type { ClaimedProduceJob, ProduceJobStore, ProduceOutput, ProduceSourceAsset } from "./store";
|
|
|
|
const KINDS: readonly ProduceJobKind[] = ["compose", "mix"];
|
|
|
|
/**
|
|
* 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 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 {
|
|
const text = typeof value === "string" ? value : undefined;
|
|
if (text === undefined || !/^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/.test(text)) {
|
|
throw new Error(`produce job store: row has an invalid ${field}`);
|
|
}
|
|
return text;
|
|
}
|
|
|
|
/** True once 009's probe job has run: the probe writes duration + codecs together, so duration≠NULL means probed. */
|
|
function probed(row: Record<string, unknown>): boolean {
|
|
return row.duration_ms !== null && row.duration_ms !== undefined;
|
|
}
|
|
|
|
function nullableNumber(value: unknown): number | null {
|
|
if (value === null || value === undefined) return null;
|
|
const n = typeof value === "number" ? value : Number(value);
|
|
return Number.isFinite(n) ? n : null;
|
|
}
|
|
|
|
function sourceFromRow(row: Record<string, unknown>): ProduceSourceAsset {
|
|
return {
|
|
id: uuid(row.id, "video_asset id"),
|
|
storageKey: String(row.storage_key ?? ""),
|
|
mimeType: String(row.mime_type ?? ""),
|
|
sizeBytes: Number(row.size_bytes ?? 0),
|
|
width: nullableNumber(row.width),
|
|
height: nullableNumber(row.height),
|
|
durationMs: row.duration_ms === null || row.duration_ms === undefined ? null : Math.round(Number(row.duration_ms)),
|
|
frameRate: nullableNumber(row.frame_rate),
|
|
// An unprobed asset has NULL codec AND NULL duration (009's probe fills all
|
|
// of them at once); a probed one always has a duration. So the probed
|
|
// audio-less case is duration≠NULL + audio_codec NULL/'' — which must become
|
|
// a silent bed, or the graph references [0:a] on a file with no audio track.
|
|
hasAudio: probed(row) ? String(row.audio_codec ?? "") !== "" : null,
|
|
};
|
|
}
|
|
|
|
/** The store's own reads are RLS-scoped to the job's workspace (009 tenant_isolation). */
|
|
export class PostgresProduceJobStore implements ProduceJobStore {
|
|
constructor(private readonly pool: Pool, private readonly now: () => Date = () => new Date()) {}
|
|
|
|
private async tenant<T>(workspaceId: string, fn: (client: PoolClient) => Promise<T>): Promise<T> {
|
|
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");
|
|
return result;
|
|
} catch (error) {
|
|
await client.query("ROLLBACK").catch(() => undefined);
|
|
throw error;
|
|
} finally {
|
|
client.release();
|
|
}
|
|
}
|
|
|
|
async claim(lane: Lane): Promise<ClaimedProduceJob | null> {
|
|
// 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<ClaimRow>(
|
|
"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<ClaimedProduceJob | null>(workspaceId, async (client) => {
|
|
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`,
|
|
[row.video_asset_id],
|
|
);
|
|
if (assets.rowCount === 0) {
|
|
// 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(row.id, "job id"),
|
|
workspaceId,
|
|
kind,
|
|
request,
|
|
source: sourceFromRow(assets.rows[0] as Record<string, unknown>),
|
|
claimedAt: this.now().toISOString(),
|
|
};
|
|
});
|
|
}
|
|
|
|
async progress(jobId: string, percent: number): Promise<void> {
|
|
await this.tenant(await this.workspaceOf(jobId), async (client) => {
|
|
await client.query(
|
|
`UPDATE video_processing_jobs SET progress_percent = $2, updated_at = now()
|
|
WHERE id = $1 AND state = 'transcoding'`,
|
|
[jobId, percent],
|
|
);
|
|
});
|
|
}
|
|
|
|
async complete(jobId: string, output: ProduceOutput, derivedFrom: string): Promise<string> {
|
|
const workspaceId = await this.workspaceOf(jobId);
|
|
return this.tenant(workspaceId, async (client) => {
|
|
const inserted = await client.query<{ id: string }>(
|
|
`INSERT INTO video_assets
|
|
(workspace_id, storage_key, mime_type, size_bytes, width, height, duration_ms, frame_rate, codec, audio_codec, derived_from)
|
|
VALUES ($1, $2, 'video/mp4', $3, $4, $5, $6, $7, 'h264', 'aac', $8)
|
|
RETURNING id`,
|
|
[workspaceId, output.storageKey, output.sizeBytes, output.width, output.height, output.durationMs, output.frameRate, derivedFrom],
|
|
);
|
|
const producedAssetId = inserted.rows[0]!.id;
|
|
const updated = await client.query(
|
|
`UPDATE video_processing_jobs
|
|
SET state = 'complete', progress_percent = 100, output_key = $2,
|
|
produced_asset_id = $3, error = NULL, updated_at = now()
|
|
WHERE id = $1 AND state = 'transcoding'
|
|
RETURNING id`,
|
|
[jobId, output.storageKey, producedAssetId],
|
|
);
|
|
if (updated.rowCount === 0) throw new Error(`produce job store: job ${jobId} lost its claim`);
|
|
return producedAssetId;
|
|
});
|
|
}
|
|
|
|
async fail(jobId: string, reason: string): Promise<void> {
|
|
await this.tenant(await this.workspaceOf(jobId), async (client) => {
|
|
await client.query(
|
|
`UPDATE video_processing_jobs
|
|
SET state = 'failed', error = $2, updated_at = now()
|
|
WHERE id = $1 AND state IN ('queued', 'transcoding')`,
|
|
[jobId, reason],
|
|
);
|
|
});
|
|
}
|
|
|
|
async asset(workspaceId: string, assetId: string): Promise<{ storageKey: string; mimeType: string } | null> {
|
|
return this.tenant(workspaceId, async (client) => {
|
|
const rows = await client.query<{ storage_key: string; mime_type: string }>(
|
|
`SELECT storage_key, mime_type FROM video_assets WHERE id = $1`,
|
|
[assetId],
|
|
);
|
|
if (rows.rowCount === 0) return null;
|
|
return { storageKey: rows.rows[0]!.storage_key, mimeType: rows.rows[0]!.mime_type };
|
|
});
|
|
}
|
|
|
|
/**
|
|
* 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 produce_job_workspace($1) AS workspace_id",
|
|
[jobId],
|
|
);
|
|
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");
|
|
}
|
|
}
|
|
|
|
export const PRODUCE_JOB_KINDS = KINDS;
|