From e73698f2b5741ad49ed4bc24faf7f955a375e95b Mon Sep 17 00:00:00 2001 From: David Torcivia Date: Mon, 3 Aug 2026 22:46:34 -0400 Subject: [PATCH] Write exports through storage instead of a disk Exports were the last part of the job pipeline that opened a filesystem: the finished document went to blobRoot/exports/... with writeFile, and a PDF read its stills back off disk. There is no `node:fs` in that file at all now. Every path involved was already a key. The still a report waits on is requested at a key the export chooses, and the result is stored at a key the job row records, so nothing here needed a filesystem in the first place -- it just happened to have one. runDueExport takes one queued export and runs it, claiming by conditional update so two drivers racing cannot both take the same row. The node pump's export slot calls it rather than keeping its own copy of the query. Not yet wired into the Workers cron, and the reason is recorded there: exports still import the PDF report builder and the SVG compositor from @onelight/worker, whose index reaches media.ts and therefore node:child_process. Mounting them on that target means lifting those two out of the media graph first, exactly as the rendition planner was lifted. A comment report requested there stays queued until then. --- apps/cf/src/index.ts | 7 ++ apps/server/src/comment-exports.ts | 150 ++++++++++++++++++++++------ apps/server/src/index.ts | 2 +- apps/server/src/pump-loop.ts | 48 ++------- apps/server/src/worker-pump.test.ts | 2 +- 5 files changed, 134 insertions(+), 75 deletions(-) diff --git a/apps/cf/src/index.ts b/apps/cf/src/index.ts index 34ac966..5a5b35e 100644 --- a/apps/cf/src/index.ts +++ b/apps/cf/src/index.ts @@ -137,5 +137,12 @@ export default { await sweepUnclaimedWork(db, now, { mediaEnabled: Boolean(env.WORKER_SECRET), }); + /* Exports are NOT run here yet. They no longer need a filesystem -- the + result goes to storage by key -- but they still import the PDF report + builder and the SVG compositor from @onelight/worker, whose index + reaches media.ts and therefore node:child_process. Mounting them here + means lifting those two out of that graph first, exactly as the + rendition planner was lifted. Until then a comment report requested on + this target stays queued. */ }, }; diff --git a/apps/server/src/comment-exports.ts b/apps/server/src/comment-exports.ts index b7ca779..a44d5a0 100644 --- a/apps/server/src/comment-exports.ts +++ b/apps/server/src/comment-exports.ts @@ -1,5 +1,3 @@ -import { mkdir, readFile, rm, writeFile } from "node:fs/promises"; -import path from "node:path"; import { and, asc, eq, inArray, isNull, lt } from "drizzle-orm"; import { exportAvidText, @@ -31,19 +29,60 @@ import { shareAssets, } from "@onelight/db/schema"; import type { AppDb } from "@onelight/db"; + +/** + * Storage, as an export needs it. + * + * Exports used to open the filesystem directly: the finished document went to + * `blobRoot/exports/...` with writeFile, and a PDF read its stills back off + * disk. That is the last thing in the job pipeline that assumed a mounted + * volume, and it is why exports were the one kind of work the Workers target + * could not do at all -- there is no disk there to write to. + * + * Every path involved was already a key. The still a report waits on is + * requested at a key the export chooses, and the result is stored at a key the + * job row records, so nothing here needed a filesystem in the first place. + */ +export interface ExportBlobStore { + putStream( + key: string, + stream: ReadableStream, + meta: { contentType?: string; size?: number }, + ): Promise; + getStream(key: string): Promise; + delete(key: string): Promise; +} + +const bytesOf = async (stream: ReadableStream): Promise => + new Uint8Array(await new Response(stream).arrayBuffer()); + +const streamOf = (value: Uint8Array | string): ReadableStream => { + /* Through the buffer rather than the view: a Uint8Array is not BodyInit, + and handing Response the underlying ArrayBuffer is what every other + stream in this codebase does. */ + const body = new Response( + typeof value === "string" ? value : (value.buffer as ArrayBuffer), + ).body; + if (!body) throw new Error("Could not stream the export."); + return body; +}; import { parseObject } from "@onelight/job-protocol"; import type { JobPayload } from "@onelight/job-protocol"; /* A comment report, and the marker files an edit suite reads. * - * Split out of the pump because this is the only part of it that still needs a - * filesystem: a PDF is assembled from stills the workers wrote to the blob - * root, and the finished document is written back there as a file. The job - * protocol beside it names storage and nothing else, which is what lets it be - * mounted on a deployment that has no disk to speak of. + * This was split out of the pump as the one part that still needed a + * filesystem. It no longer does: the finished document goes to storage by key + * and a PDF reads its stills back the same way, so there is no `node:fs` here + * at all. Every path involved was already a key -- the still a report waits on + * is requested at a key the export chooses, and the result is stored at a key + * the job row records. * - * The pump still drives all of this: it owns the single export slot and the - * reclaim pass, and calls in here for the work itself. */ + * That is what lets the Workers target run exports, which it previously could + * not do at any speed: there is no disk there to write to. + * + * Whoever drives it owns the single export slot and the reclaim pass, and + * calls in here for the work itself. */ interface ExportFilter { version_id?: string; @@ -208,7 +247,7 @@ const summarizeFilter = (filter: ExportFilter): string => { const buildPdfExport = async ( db: AppDb, job: typeof exportJobs.$inferSelect, - blobRoot: string, + store: ExportBlobStore, mediaEnabled: boolean, allRows: ExportRow[], selected: ExportRow[], @@ -251,7 +290,10 @@ const buildPdfExport = async ( console.warn( `[onelight] pdf export ${job.id}: media processing is not configured; the report falls back to text-only blocks.`, ); - const stillsDir = path.join(blobRoot, "exports", `.stills-${job.id}`); + /* The namespace the report's stills are written into, as a key prefix. The + still jobs are handed exactly these keys, so reading them back is asking + storage for what it was told to store. */ + const stillsPrefix = `exports/.stills-${job.id}/`; const proxyByVersion = new Map(); const proxyFor = async (versionId: string): Promise => { if (proxyByVersion.has(versionId)) return proxyByVersion.get(versionId); @@ -292,11 +334,7 @@ const buildPdfExport = async ( await enqueueExportStill(db, key, { version_id: version.id, blob_key: proxyKey, - output_key: path.posix.join( - "exports", - `.stills-${job.id}`, - `${comment.id}.png`, - ), + output_key: `${stillsPrefix}${comment.id}.png`, frame: comment.frameIn, rate: version.frameRateNum && version.frameRateDen @@ -325,8 +363,8 @@ const buildPdfExport = async ( let stillPng: Uint8Array | undefined; if (renderedStills.has(stillKeyFor(comment.id))) { try { - stillPng = new Uint8Array( - await readFile(path.join(stillsDir, `${comment.id}.png`)), + stillPng = await bytesOf( + await store.getStream(`${stillsPrefix}${comment.id}.png`), ); } catch (error) { stillPng = undefined; @@ -393,7 +431,12 @@ const buildPdfExport = async ( comments: reportComments, }); } finally { - await rm(stillsDir, { recursive: true, force: true }); + /* Scratch the report read and no longer needs. Deleted by key, one per + comment that had one, because storage has no directories to remove. */ + for (const key of stillKeys.map( + (key) => `${stillsPrefix}${key.split(":").pop() ?? ""}.png`, + )) + await store.delete(key).catch(() => undefined); await discardExportStillJobs(db, stillKeys); } }; @@ -474,10 +517,59 @@ const zipTextFiles = async ( return new Uint8Array(await new Response(zipStream(entries)).arrayBuffer()); }; +/** + * Take one queued export and run it, or do nothing. + * + * One per call, deliberately: a long PDF report must not hold the caller open + * behind it, and both drivers -- the node pump's own slot and the Workers cron + * -- want the same "make some progress and return" shape rather than a loop. + * + * Claims by conditional update, so two drivers racing cannot both run the same + * export. A failure is written onto the row rather than thrown, because the + * caller is a timer with nobody watching it. + */ +export const runDueExport = async ( + db: AppDb, + store: ExportBlobStore, + options: { mediaEnabled: boolean; now: number }, +): Promise => { + const pending = ( + await db + .select() + .from(exportJobs) + .where(eq(exportJobs.status, "queued")) + .orderBy(asc(exportJobs.createdAt)) + .limit(1) + .all() + )[0]; + if (!pending) return false; + const claimed = await db + .update(exportJobs) + .set({ status: "processing" }) + .where(and(eq(exportJobs.id, pending.id), eq(exportJobs.status, "queued"))) + .run(); + /* Somebody else took it between the read and the write. */ + if ((claimed as { changes?: number }).changes === 0) return false; + try { + await processExportJob(db, pending, store, options.mediaEnabled); + } catch (error) { + await db + .update(exportJobs) + .set({ + status: "failed", + error: error instanceof Error ? error.message : "Export failed.", + finishedAt: options.now, + }) + .where(eq(exportJobs.id, pending.id)) + .run(); + } + return true; +}; + export const processExportJob = async ( db: AppDb, job: typeof exportJobs.$inferSelect, - blobRoot: string, + store: ExportBlobStore, mediaEnabled: boolean, ): Promise => { const filter = parseObject(job.filtersJson) as ExportFilter; @@ -633,14 +725,7 @@ export const processExportJob = async ( let output: string | Uint8Array; let outputName: string; if (job.format === "pdf") { - output = await buildPdfExport( - db, - job, - blobRoot, - mediaEnabled, - rows, - selected, - ); + output = await buildPdfExport(db, job, store, mediaEnabled, rows, selected); outputName = "onelight-comment-report.pdf"; } else { const extension = exportExtension(job.format); @@ -661,9 +746,12 @@ export const processExportJob = async ( } } const key = `exports/${job.id}/${outputName}`; - const directory = path.dirname(path.join(blobRoot, key)); - await mkdir(directory, { recursive: true }); - await writeFile(path.join(blobRoot, key), output); + await store.putStream(key, streamOf(output), { + size: + typeof output === "string" + ? new TextEncoder().encode(output).byteLength + : output.byteLength, + }); await db .update(exportJobs) .set({ diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts index b052c94..712fcdf 100644 --- a/apps/server/src/index.ts +++ b/apps/server/src/index.ts @@ -456,7 +456,7 @@ const start = async (): Promise => { }); const stopWorkerPump = startWorkerPump(db, { ...(config.WORKER_SECRET ? { workerSecret: config.WORKER_SECRET } : {}), - blobRoot, + store: blobStore, }); const stopBackups = backupConfig ? startBackups(sqlite, db, backupConfig) diff --git a/apps/server/src/pump-loop.ts b/apps/server/src/pump-loop.ts index e8b223f..5be2db5 100644 --- a/apps/server/src/pump-loop.ts +++ b/apps/server/src/pump-loop.ts @@ -1,12 +1,11 @@ -import { and, asc, eq } from "drizzle-orm"; -import { exportJobs } from "@onelight/db/schema"; import { buryAbandonedJob, findAbandonedJobs } from "@onelight/db"; import type { AppDb } from "@onelight/db"; import { EXPORT_RECLAIM_STALE_MS, - processExportJob, reclaimStuckExports, + runDueExport, } from "./comment-exports.js"; +import type { ExportBlobStore } from "./comment-exports.js"; import { FINGERPRINT_SWEEP_INTERVAL_MS, markAbandonedVersionFailed, @@ -37,7 +36,9 @@ import { export const startWorkerPump = ( db: AppDb, - options: { workerSecret?: string; blobRoot: string }, + /* The store rather than a path: exports write their result through storage + now, so this loop no longer needs to know where storage happens to be. */ + options: { workerSecret?: string; store: ExportBlobStore }, ): (() => void) => { /* Exports are pure DB-to-file work, so the pump runs them whether or not any worker exists; media jobs are no longer run here at all. The server @@ -67,44 +68,7 @@ export const startWorkerPump = ( if (exporting || stopped) return; exporting = true; try { - const pendingExport = ( - await db - .select() - .from(exportJobs) - .where(eq(exportJobs.status, "queued")) - .orderBy(asc(exportJobs.createdAt)) - .limit(1) - .all() - )[0]; - if (!pendingExport) return; - try { - await db - .update(exportJobs) - .set({ status: "processing" }) - .where( - and( - eq(exportJobs.id, pendingExport.id), - eq(exportJobs.status, "queued"), - ), - ) - .run(); - await processExportJob( - db, - pendingExport, - options.blobRoot, - mediaEnabled, - ); - } catch (error) { - await db - .update(exportJobs) - .set({ - status: "failed", - error: error instanceof Error ? error.message : "Export failed.", - finishedAt: Date.now(), - }) - .where(eq(exportJobs.id, pendingExport.id)) - .run(); - } + await runDueExport(db, options.store, { mediaEnabled, now: Date.now() }); } catch (error) { console.warn( `[onelight] export pump failed: ${ diff --git a/apps/server/src/worker-pump.test.ts b/apps/server/src/worker-pump.test.ts index de969a1..0b8ccfc 100644 --- a/apps/server/src/worker-pump.test.ts +++ b/apps/server/src/worker-pump.test.ts @@ -1404,7 +1404,7 @@ describe("a comment report's frames", () => { stop = startWorkerPump(db, { workerSecret: WORKER_SECRET, - blobRoot, + store: new LocalBlobStore(blobRoot), }); const deadline = Date.now() + 20_000;