From 64e0c9e32067b27b4316d3090a01092a2a516636 Mon Sep 17 00:00:00 2001 From: Sivert Date: Mon, 5 Oct 2026 14:06:50 +0200 Subject: [PATCH] A quarantined avatar or banner only replaces the old one once the worker has written it out (GRYT-1664) Testing real uploads showed a refused file (a truncated MP4, HTML named .png) became the member's avatar at once, left them with a broken picture, and made every read wait the full 15 seconds. Now the old one stays until the worker's copy is ready, a refused file is deleted, and a read of one answers 404 at once. Co-Authored-By: Claude Opus 5.5 --- src/db/sqlite/imageJobs.ts | 7 ++++ src/routes/uploadQuarantine.test.ts | 54 +++++++++++++++++++++++++++-- src/routes/uploads.ts | 47 +++++++++++++++++++++++-- src/services/quarantineUpload.ts | 35 ++++++++++++++++--- src/socket/utils/server.ts | 6 ++++ 5 files changed, 141 insertions(+), 8 deletions(-) diff --git a/src/db/sqlite/imageJobs.ts b/src/db/sqlite/imageJobs.ts index b70236b8..58359d8a 100644 --- a/src/db/sqlite/imageJobs.ts +++ b/src/db/sqlite/imageJobs.ts @@ -40,6 +40,13 @@ export async function getImageJob(jobId: string): Promise return mapRow(row); } +/** The newest job's status for a file, or null when it never had one. */ +export async function getImageJobStatusForFile(fileId: string): Promise { + const db = getSqliteDb(); + const row = db.prepare("SELECT status FROM image_jobs WHERE file_id = ? ORDER BY created_at DESC LIMIT 1").get(fileId) as { status: ImageJobStatus } | undefined; + return row?.status ?? null; +} + export async function updateImageJobStatus(input: { job_id: string; status: ImageJobStatus; diff --git a/src/routes/uploadQuarantine.test.ts b/src/routes/uploadQuarantine.test.ts index 01235409..741f5e1f 100644 --- a/src/routes/uploadQuarantine.test.ts +++ b/src/routes/uploadQuarantine.test.ts @@ -51,6 +51,21 @@ async function upload(who: Member, path: "banner" | "avatar") { return { status: res.status, body: (await res.json()) as Record }; } +/** What the worker does when it finishes: the row moves out of quarantine, or the job is marked failed. */ +const workerWritesOut = (fileId: string, key: string, mimeType: string) => { + getSqliteDb().prepare("UPDATE files SET s3_key = ?, mime = ? WHERE file_id = ?").run(key, mimeType, fileId); + getSqliteDb().prepare("UPDATE image_jobs SET status = 'done' WHERE file_id = ?").run(fileId); +}; +const workerRefuses = (fileId: string) => + getSqliteDb().prepare("UPDATE image_jobs SET status = 'error', error_message = 'could not be decoded' WHERE file_id = ?").run(fileId); +const until = async (check: () => Promise, ms = 4000) => { + const end = Date.now() + ms; + while (!(await check())) { + if (Date.now() > end) throw new Error("timed out waiting"); + await new Promise((resolve) => setTimeout(resolve, 100)); + } +}; + const read = (who: Member, fileId: string) => fetch(`${base}/api/uploads/files/${fileId}?t=${encodeURIComponent(generateFileToken({ ...who, serverHost: host, tokenVersion: 0 }))}`); @@ -83,11 +98,43 @@ describe("uploads with a worker that clears quarantine", () => { assert.equal(res.status, 201, JSON.stringify(res.body)); const fileId = res.body.bannerFileId as string; assert.equal(res.body.processing, true); - assert.equal((await getUserByServerId(alice.serverUserId))?.banner_file_id, fileId); + assert.equal((await getUserByServerId(alice.serverUserId))?.banner_file_id ?? null, null, "not theirs until written out"); const file = await getFile(fileId); assert.ok(file?.s3_key.startsWith("quarantine/banners/"), file?.s3_key); const job = getSqliteDb().prepare("SELECT raw_s3_key, status FROM image_jobs WHERE file_id = ?").get(fileId) as { raw_s3_key: string; status: string }; assert.deepEqual({ ...job }, { raw_s3_key: file!.s3_key, status: "queued" }); + + workerWritesOut(fileId, `banners/${fileId}.avif`, "image/avif"); + await until(async () => (await getUserByServerId(alice.serverUserId))?.banner_file_id === fileId); + }); + + it("keeps the working avatar when the worker refuses the new one, and drops the refused file", async () => { + await workerSays(["quarantine-v1"]); + const erin = await member("erin"); + const first = (await upload(erin, "avatar")).body.avatarFileId as string; + workerWritesOut(first, `avatars/${first}.avif`, "image/avif"); + await until(async () => (await getUserByServerId(erin.serverUserId))?.avatar_file_id === first); + + const second = (await upload(erin, "avatar")).body.avatarFileId as string; + workerRefuses(second); + const started = Date.now(); + const refused = await read(erin, second); + assert.ok(refused.status === 404 || refused.status === 401, String(refused.status)); + assert.ok(Date.now() - started < 1000, "a refused file is answered at once, not after the wait"); + await until(async () => (await getFile(second)) === null); + assert.equal((await getUserByServerId(erin.serverUserId))?.avatar_file_id, first); + }); + + it("applies only the newest of two uploads, whichever the worker finishes first", async () => { + await workerSays(["quarantine-v1"]); + const finn = await member("finn"); + const older = (await upload(finn, "banner")).body.bannerFileId as string; + const newer = (await upload(finn, "banner")).body.bannerFileId as string; + workerWritesOut(newer, `banners/${newer}.avif`, "image/avif"); + await until(async () => (await getUserByServerId(finn.serverUserId))?.banner_file_id === newer); + workerWritesOut(older, `banners/${older}.avif`, "image/avif"); + await until(async () => (await getFile(older)) === null); + assert.equal((await getUserByServerId(finn.serverUserId))?.banner_file_id, newer); }); it("never serves the original: a read waits for the worker's copy, and says not ready if it never comes", async () => { @@ -136,9 +183,12 @@ describe("uploads with a worker that clears quarantine", () => { await workerSays(["quarantine-v1", "video-v1"]); const taken = await sendVideo(); assert.equal(taken.status, 201, JSON.stringify(taken.body)); - const file = await getFile(taken.body.bannerFileId as string); + const videoId = taken.body.bannerFileId as string; + const file = await getFile(videoId); assert.ok(file?.s3_key.startsWith("quarantine/banners/")); assert.equal(file?.mime, "video/mp4"); + workerWritesOut(videoId, `banners/${videoId}.mp4`, "video/mp4"); + await until(async () => (await getUserByServerId(dave.serverUserId))?.banner_file_id === videoId); const row = (await buildMemberList({})).find((m) => m.serverUserId === dave.serverUserId); assert.equal(row?.bannerVideo, true, "the member list says it plays"); assert.equal(row?.avatarVideo, false); diff --git a/src/routes/uploads.ts b/src/routes/uploads.ts index a683aeb9..86eab7bb 100644 --- a/src/routes/uploads.ts +++ b/src/routes/uploads.ts @@ -19,10 +19,44 @@ import { RangeNotSatisfiableError } from "../utils/byteRange"; import { sendStoredBody } from "../utils/sendStoredBody"; import { ensurePermission } from "../middleware/requirePermission"; import { sanitizeSvg } from "../utils/svgSanitize"; -import { quarantineUpload, waitOutOfQuarantine } from "../services/quarantineUpload"; +import { quarantineUpload, settleQuarantine, waitOutOfQuarantine } from "../services/quarantineUpload"; +import { broadcastMembersUpdate } from "../socket/utils/server"; /** How long a read waits for the worker to write a fresh upload out; a test sets it short. */ const QUARANTINE_WAIT_MS = Number(process.env.GRYT_QUARANTINE_WAIT_MS) || 15_000; + +/* The newest pending avatar and banner per member. An older upload that settles after a + newer one is dropped rather than applied over it. */ +const pendingMedia = new Map(); + +async function applyWhenSettled(purpose: "avatar" | "banner", serverUserId: string, fileId: string): Promise { + const slot = `${purpose}:${serverUserId}`; + pendingMedia.set(slot, fileId); + try { + const verdict = await settleQuarantine(fileId); + if (pendingMedia.get(slot) !== fileId) { + await deleteUnreferencedFiles([fileId]); + return; + } + pendingMedia.delete(slot); + if (verdict !== "ready") { + // A timeout leaves it in quarantine for the worker to finish; a refusal goes now. + if (verdict === "refused") await deleteUnreferencedFiles([fileId]); + consola.info(`[uploads] ${purpose} ${fileId} not applied: ${verdict}`); + return; + } + if (purpose === "avatar") { + await updateUserAvatar(serverUserId, fileId); + } else { + const previous = (await getUserByServerId(serverUserId))?.banner_file_id ?? null; + await setUserBanner(serverUserId, fileId); + if (previous && previous !== fileId) await deleteUnreferencedFiles([previous]); + } + broadcastMembersUpdate(); + } catch (err) { + consola.warn(`[uploads] Could not apply ${purpose} ${fileId}`, err); + } +} import { storeUploadedFile } from "../services/storeUploadedFile"; import { workerClearsQuarantine, workerTranscodesVideo } from "../services/workerCapabilities"; @@ -253,7 +287,11 @@ const storeAvatarImage = (purpose: "avatar" | "group" | "webhook" | "banner") => return; } - if (purpose === "avatar") { + if ((purpose === "avatar" || purpose === "banner") && stored.processing) { + // Not theirs until the worker has written it out, so a refused file never replaces a working one. + void applyWhenSettled(purpose, serverUserId, stored.fileId); + res.status(201).json({ [purpose === "avatar" ? "avatarFileId" : "bannerFileId"]: stored.fileId, processing: true }); + } else if (purpose === "avatar") { await updateUserAvatar(serverUserId, stored.fileId); res.status(201).json({ avatarFileId: stored.fileId, processing: stored.processing }); } else if (purpose === "banner") { @@ -448,6 +486,11 @@ uploadsRouter.get( if (!found) { res.status(404).json({ error: "File not found" }); return; } // Never the quarantined original: wait for the worker's copy, or say it isn't ready. const fileMeta = await waitOutOfQuarantine(fileId, found, undefined, QUARANTINE_WAIT_MS); + if (fileMeta === "refused") { + res.setHeader("Cache-Control", "no-store"); + res.status(404).json({ error: "media_refused", message: "This file could not be processed." }); + return; + } if (!fileMeta) { res.setHeader("Cache-Control", "no-store"); res.status(503).json({ error: "media_not_ready", message: "This file is still being processed." }); diff --git a/src/services/quarantineUpload.ts b/src/services/quarantineUpload.ts index c6d28729..47b774ee 100644 --- a/src/services/quarantineUpload.ts +++ b/src/services/quarantineUpload.ts @@ -1,6 +1,6 @@ import { v4 as uuidv4 } from "uuid"; -import { getFile, insertFile, insertImageJob } from "../db"; +import { getFile, getImageJobStatusForFile, insertFile, insertImageJob } from "../db"; import { putObject } from "../storage"; export const QUARANTINE_PREFIX = "quarantine/"; @@ -39,20 +39,47 @@ export async function quarantineUpload(input: { export const isQuarantined = (key: string | null | undefined): boolean => !!key && key.startsWith(QUARANTINE_PREFIX); /** A file just uploaded is usually written out within a second or two, so a read waits - for it rather than failing. Null when it is still in quarantine after `waitMs`. */ + for it. "refused" as soon as the worker gives up on it, null if still waiting after `waitMs`. */ export async function waitOutOfQuarantine( fileId: string, first: T, read: (id: string) => Promise = getFile as unknown as (id: string) => Promise, waitMs = 15_000, stepMs = 250, -): Promise { + jobStatus: (id: string) => Promise = getImageJobStatusForFile, + holdProcess = true, +): Promise { let file: T | null = first; const until = Date.now() + waitMs; while (file && isQuarantined(file.s3_key)) { + if ((await jobStatus(fileId)) === "error") return "refused"; if (Date.now() >= until) return null; - await new Promise((r) => setTimeout(r, stepMs)); + await new Promise((r) => { + const timer = setTimeout(r, stepMs); + if (!holdProcess) timer.unref(); + }); file = await read(fileId); } return file; } + +/* Long enough for a ten-second video on a slow worker. Past it the upload is left alone, + still in quarantine, and the member keeps what they had. */ +const SETTLE_WAIT_MS = 10 * 60_000; + +/** Waits for the worker's verdict on a quarantined avatar or banner, so it only replaces + the old one once it has been written out (GRYT-1664). */ +export async function settleQuarantine( + fileId: string, + read: (id: string) => Promise<{ s3_key: string } | null> = getFile, + waitMs = SETTLE_WAIT_MS, + stepMs = 1_000, + jobStatus: (id: string) => Promise = getImageJobStatusForFile, +): Promise<"ready" | "refused" | "timeout"> { + const first = await read(fileId); + if (!first) return "refused"; + // Background work, so it never keeps a shutting-down server (or a test run) alive. + const out = await waitOutOfQuarantine(fileId, first, read, waitMs, stepMs, jobStatus, false); + if (out === "refused" || (out === null && (await read(fileId)) === null)) return "refused"; + return out ? "ready" : "timeout"; +} diff --git a/src/socket/utils/server.ts b/src/socket/utils/server.ts index bcbceecb..28b09643 100644 --- a/src/socket/utils/server.ts +++ b/src/socket/utils/server.ts @@ -66,6 +66,12 @@ export function broadcastServerUiUpdate(reason: "settings" | "icon" | "other" = void pushVoiceCapabilities(); } +/** For a REST route that changed a member, such as an avatar the worker has just cleared. */ +export function broadcastMembersUpdate(): void { + if (!_io || !_serverId || !_clientsInfo) return; + broadcastMemberList(_io, _clientsInfo, _serverId); +} + export function broadcastChatNew(message: Record): void { if (!_io || !_clientsInfo) return; const io = _io;