A quarantined avatar or banner only replaces the old one once the worker has written it out (GRYT-1664)

src/db/sqlite/imageJobs.ts

@@ -40,6 +40,13 @@ export async function getImageJob(jobId: string): Promise<ImageJobRecord | null>
   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<ImageJobStatus | null> {
+  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;

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<string, unknown> };
 }
 
+/** 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<boolean>, 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);

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<string, string>();
+
+async function applyWhenSettled(purpose: "avatar" | "banner", serverUserId: string, fileId: string): Promise<void> {
+  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." });

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<T extends { s3_key: string }>(
   fileId: string,
   first: T,
   read: (id: string) => Promise<T | null> = getFile as unknown as (id: string) => Promise<T | null>,
   waitMs = 15_000,
   stepMs = 250,
-): Promise<T | null> {
+  jobStatus: (id: string) => Promise<string | null> = getImageJobStatusForFile,
+  holdProcess = true,
+): Promise<T | "refused" | null> {
   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<void>((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<string | null> = 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";
+}

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<string, unknown>): void {
   if (!_io || !_clientsInfo) return;
   const io = _io;