fix(search-index): recover image deletes dropped by a degraded sysRedis

`addToQueue` fails open when sysRedis is down or slow, and returned early
without the ids. A dropped `Update` is re-derived by the delta `updatedAt`
range-scan; a dropped `Delete` has no surviving row to re-derive it from, so the
document stayed in the index permanently. queues.ts named `search-index-cleanup`
as the recovery for exactly that case, but `CLEANUP_INDEXES` covers neither
images index.

Measured 2026-07-01..08-31: 15 dropped `images_v6:Delete` enqueues across 13
distinct days. The resulting orphans are ~0.16% of the index overall but
concentrate where bulk deletions land -- one sampled search returned 39 of 200
hits pointing at images deleted from Postgres, every blob a 404.

Dropped ids are now parked in "KeyValue" under `search-index-queue-fallback:`
and replayed by a new `search-index-queue-drain` job. The drain reads before it
deletes and retires only the row it replayed, guarded on array length, so a
crash mid-drain or a drop landing between the read and the delete costs a
duplicated replay (a no-op) rather than lost ids.

Also adds /api/mod/cleanup-orphaned-search-docs to reap existing orphans for one
query or filter. Scoped deliberately: a full sweep of images_v6 is ~6.4K keyset
pages at 6-16s each against a 891GB LMDB, which evicts the live search working
set to fix tiles nobody is looking at.

No migration -- reuses the existing KeyValue table.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

src/pages/api/mod/cleanup-orphaned-search-docs.ts

@@ -0,0 +1,106 @@
+import { chunk } from 'lodash-es';
+import type { NextApiRequest, NextApiResponse } from 'next';
+import * as z from 'zod';
+import { IMAGES_SEARCH_INDEX } from '~/server/common/constants';
+import { SearchIndexUpdateQueueAction } from '~/server/common/enums';
+import { dbRead } from '~/server/db/client';
+import { searchClient } from '~/server/meilisearch/client';
+import { queueImageSearchIndexUpdate } from '~/server/services/image.service';
+import { handleEndpointError, ModEndpoint } from '~/server/utils/endpoint-helpers';
+import { booleanString } from '~/utils/zod-helpers';
+
+/**
+ * Reaps documents in the images index whose Image row no longer exists — the images
+ * index is reconciled by nothing (`CLEANUP_INDEXES` in meilisearch/cleanup.ts omits it),
+ * so a delete lost to a degraded sysRedis enqueue leaves a document that renders as a
+ * broken tile forever.
+ *
+ * Scoped to a query or filter on purpose. A full sweep of the index is ~6.4K keyset
+ * pages at 6-16s each against a 891GB LMDB, which evicts the live search working set;
+ * the damage is concentrated (a bulk deletion orphans one cohort, so one search comes
+ * back a quarter broken while its neighbours are clean), so reaping the affected query
+ * costs a few seconds and fixes what anyone actually sees.
+ *
+ *   ?query=<text>          text search to sweep (default: empty = browse order)
+ *   ?filter=<meili filter> e.g. user.username = "someone"
+ *   ?limit=<n>             docs to examine, max 20000 (default 2000)
+ *   ?dryRun=false          actually queue the deletes (default true)
+ *   ?maxOrphanRate=<0-1>   abort above this fraction (default 0.5)
+ */
+const schema = z.object({
+  query: z.string().default(''),
+  filter: z.string().optional(),
+  limit: z.coerce.number().min(1).max(20000).default(2000),
+  dryRun: booleanString().default(true),
+  maxOrphanRate: z.coerce.number().min(0).max(1).default(0.5),
+});
+
+const PAGE_SIZE = 1000;
+const DB_CHUNK_SIZE = 2000;
+
+export default ModEndpoint(
+  async function cleanupOrphanedSearchDocs(req: NextApiRequest, res: NextApiResponse) {
+    try {
+      const { query, filter, limit, dryRun, maxOrphanRate } = schema.parse(req.query);
+      if (!searchClient) throw new Error('Search client not available');
+
+      const index = searchClient.index(IMAGES_SEARCH_INDEX);
+      const indexed: number[] = [];
+      for (let offset = 0; offset < limit; offset += PAGE_SIZE) {
+        const page = await index.search<{ id: number }>(query, {
+          limit: Math.min(PAGE_SIZE, limit - offset),
+          offset,
+          filter,
+          attributesToRetrieve: ['id'],
+        });
+        indexed.push(...page.hits.map((hit) => hit.id));
+        if (page.hits.length < PAGE_SIZE) break;
+      }
+      if (!indexed.length) return res.status(200).json({ examined: 0, orphaned: 0, queued: 0 });
+
+      // Chunked, and a throw aborts the whole request rather than being caught per
+      // chunk: a partial read would look exactly like a pile of orphans and queue
+      // deletes for live documents.
+      const alive = new Set<number>();
+      for (const ids of chunk(indexed, DB_CHUNK_SIZE)) {
+        const rows = await dbRead.image.findMany({
+          where: { id: { in: ids } },
+          select: { id: true },
+        });
+        for (const row of rows) alive.add(row.id);
+      }
+
+      const orphaned = indexed.filter((id) => !alive.has(id));
+      const orphanRate = orphaned.length / indexed.length;
+      // The measured rate is ~0.16% overall and ~20% on the worst single query, so
+      // anything past half the page is far likelier to be a bad read than real orphans.
+      if (orphanRate > maxOrphanRate) {
+        return res.status(409).json({
+          error: 'orphan rate above maxOrphanRate — refusing to queue deletes',
+          examined: indexed.length,
+          orphaned: orphaned.length,
+          orphanRate,
+        });
+      }
+
+      if (!dryRun && orphaned.length) {
+        await queueImageSearchIndexUpdate({
+          ids: orphaned,
+          action: SearchIndexUpdateQueueAction.Delete,
+        });
+      }
+
+      return res.status(200).json({
+        dryRun,
+        examined: indexed.length,
+        orphaned: orphaned.length,
+        orphanRate,
+        queued: dryRun ? 0 : orphaned.length,
+        sample: orphaned.slice(0, 25),
+      });
+    } catch (e) {
+      return handleEndpointError(res, e);
+    }
+  },
+  ['GET']
+);

src/pages/api/mod/delete-user-images.ts

@@ -1,12 +1,8 @@
 import { chunk } from 'lodash-es';
 import type { NextApiRequest, NextApiResponse } from 'next';
 import * as z from 'zod';
-import { IMAGES_SEARCH_INDEX } from '~/server/common/constants';
-import { SearchIndexUpdateQueueAction } from '~/server/common/enums';
 import { dbRead, dbWrite } from '~/server/db/client';
-import { searchClient } from '~/server/meilisearch/client';
-import { deleteImages, queueImageSearchIndexUpdate } from '~/server/services/image.service';
-import { bustCachesForPosts } from '~/server/services/post.service';
+import { deleteImages } from '~/server/services/image.service';
 import { handleEndpointError, ModEndpoint } from '~/server/utils/endpoint-helpers';
 
 const schema = z.object({
@@ -53,76 +49,3 @@ export default ModEndpoint(
   },
   ['GET']
 );
-
-// Cleans up stale image documents in the search index for a given search query.
-// Paginates meilisearch, cross-references against the DB, and queues deletes
-// for any document whose underlying image row no longer exists.
-// queueImageSearchIndexUpdate targets both images_v6 and metrics_images_v1.
-// async function deleteDeletedImages(query: string) {
-//   if (!searchClient) throw new Error('Search client not available');
-
-//   // Paginate through meilisearch collecting { id, postId } for each hit
-//   type MeiliHit = { id: number; postId: number | null };
-//   const hits: MeiliHit[] = [];
-//   const pageSize = 1000;
-//   let offset = 0;
-//   while (true) {
-//     const results = await searchClient.index(IMAGES_SEARCH_INDEX).search<MeiliHit>(query, {
-//       limit: pageSize,
-//       offset,
-//       attributesToRetrieve: ['id', 'postId'],
-//     });
-//     if (!results.hits.length) break;
-//     hits.push(...results.hits);
-//     if (results.hits.length < pageSize) break;
-//     offset += pageSize;
-//   }
-
-//   if (!hits.length) return { found: 0, orphanedImages: 0, deletedPosts: 0 };
-
-//   // Determine which image IDs are truly orphaned (present in meili but gone from DB)
-//   const meiliIds = hits.map((h) => h.id);
-//   const existing = await dbRead.image.findMany({
-//     where: { id: { in: meiliIds } },
-//     select: { id: true },
-//   });
-//   const existingIds = new Set(existing.map((i) => i.id));
-//   const orphanHits = hits.filter((h) => !existingIds.has(h.id));
-//   const orphanImageIds = orphanHits.map((h) => h.id);
-
-//   if (orphanImageIds.length) {
-//     // queueImageSearchIndexUpdate targets both images_v6 and metrics_images_v1
-//     await queueImageSearchIndexUpdate({
-//       ids: orphanImageIds,
-//       action: SearchIndexUpdateQueueAction.Delete,
-//     });
-//   }
-
-//   // Find posts referenced by the orphaned images that still exist in DB and are now empty
-//   const candidatePostIds = [
-//     ...new Set(orphanHits.map((h) => h.postId).filter((id): id is number => id != null)),
-//   ];
-
-//   let deletedPosts = 0;
-//   if (candidatePostIds.length) {
-//     const posts = await dbRead.post.findMany({
-//       where: { id: { in: candidatePostIds } },
-//       select: { id: true, _count: { select: { images: true } } },
-//     });
-//     const emptyPostIds = posts.filter((p) => p._count.images === 0).map((p) => p.id);
-
-//     if (emptyPostIds.length) {
-//       const result = await dbWrite.post.deleteMany({
-//         where: { id: { in: emptyPostIds } },
-//       });
-//       deletedPosts = result.count;
-//       await bustCachesForPosts(emptyPostIds);
-//     }
-//   }
-
-//   return {
-//     found: hits.length,
-//     orphanedImages: orphanImageIds.length,
-//     deletedPosts,
-//   };
-// }

src/pages/api/webhooks/run-jobs/[[...run]].ts

@@ -100,6 +100,7 @@ import { rewardsAbusePrevention } from '~/server/jobs/rewards-abuse-prevention';
 import { rewardsAdImpressions } from '~/server/jobs/rewards-ad-impressions';
 import { scanFilesFallbackJob } from '~/server/jobs/scan-files';
 import { searchIndexCleanupJob } from '~/server/jobs/search-index-cleanup';
+import { searchIndexQueueDrainJob } from '~/server/jobs/search-index-queue-drain';
 import { searchIndexJobs } from '~/server/jobs/search-index-sync';
 import { searchIndexUserCleanupJob } from '~/server/jobs/search-index-user-cleanup';
 import { sendCollectionNotifications } from '~/server/jobs/send-collection-notifications';
@@ -162,6 +163,7 @@ export const jobs: Job[] = [
   ...searchIndexJobs,
   searchIndexUserCleanupJob,
   searchIndexCleanupJob,
+  searchIndexQueueDrainJob,
   processRewards,
   rewardsDailyReset,
   ...bountyJobs,

src/server/jobs/search-index-queue-drain.ts

@@ -0,0 +1,35 @@
+import { logToAxiom } from '~/server/logging/client';
+import { drainDroppedEnqueues } from '~/server/redis/queues';
+import { createJob } from './job';
+
+// Replays search-index and metrics enqueues that were parked in Postgres because
+// sysRedis was degraded when a content mutation tried to queue them. Deletes are the
+// reason it runs: an update is re-derived by the delta `updatedAt` range-scan, while a
+// delete has no surviving row to re-derive it from, so a dropped one leaves the
+// document in the index permanently.
+export const searchIndexQueueDrainJob = createJob(
+  'search-index-queue-drain',
+  '*/5 * * * *',
+  async () => {
+    const result = await drainDroppedEnqueues();
+    if (result.keys === 0) return;
+
+    // `reparked` is not a failure of this job — sysRedis is still degraded and the ids
+    // are safe — but it is the signal that the outage is ongoing rather than a blip.
+    logToAxiom({
+      type: result.reparked > 0 ? 'warning' : 'info',
+      name: 'search-index-queue-drain',
+      message:
+        `replayed ${result.replayed} id(s) across ${result.keys} key(s)` +
+        (result.reparked > 0 ? `, ${result.reparked} re-parked (sysRedis still degraded)` : ''),
+    }).catch(() => undefined);
+
+    return result;
+  },
+  // Twice the cron period. createJob's default lockExpiration is 300s, which is
+  // exactly this job's interval — a run lasting any time at all could have its lock
+  // expire as the next one fires. Overlapping runs are safe (both delete under a row
+  // lock, so the second finds nothing), but a lock that cannot outlive its own period
+  // is not a lock.
+  { lockExpiration: 10 * 60 }
+);

src/server/redis/__tests__/queues.integration.test.ts

@@ -0,0 +1,211 @@
+import { PrismaClient } from '@prisma/client';
+import { createClient } from 'redis';
+import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
+// Top-level, not an inline `typeof import(...)` — that trips consistent-type-imports.
+import type * as SysReadDeadline from '~/server/redis/sys-read-deadline';
+
+/**
+ * End-to-end proof for the dropped-enqueue parking lot, against a REAL Postgres and a
+ * REAL Redis. The unit suite beside this one mocks both clients, so it can prove the
+ * control flow and prove nothing about the SQL: whether Prisma's tagged template binds
+ * the `::jsonb` cast the way the statement expects, whether `jsonb_array_length` accepts
+ * what we actually store, whether the length-guarded delete matches. Those only fail
+ * against a database.
+ *
+ * Skipped unless both URLs are supplied, mirroring kysely-prisma-parity.test.ts — CI has
+ * neither, and a suite that silently needs infrastructure is worse than one that skips.
+ *
+ *   QUEUES_IT_DATABASE_URL=postgresql://... \
+ *   QUEUES_IT_REDIS_URL=redis://:redis@localhost:6379 \
+ *   pnpm exec vitest run --project 'unit*' src/server/redis/__tests__/queues.integration.test.ts
+ *
+ * Touches nothing real: the table is created in a scratch schema that is dropped
+ * afterwards, and every redis key is written under a run-unique prefix.
+ */
+const databaseUrl = process.env.QUEUES_IT_DATABASE_URL;
+const redisUrl = process.env.QUEUES_IT_REDIS_URL;
+
+const SCHEMA = 'queues_it';
+
+// `backing = null` is the outage: every command rejects fast, which is the DOWN mode
+// queues.ts fails open on. Swapping it back is the recovery the drain has to survive.
+const { holder, NS } = vi.hoisted(() => ({
+  holder: { backing: null as null | Record<string, (...args: never[]) => Promise<unknown>> },
+  NS: `queues-it:${process.pid}`,
+}));
+
+vi.mock('~/server/redis/client', async () => {
+  const { withSysReadDeadline } = await vi.importActual<typeof SysReadDeadline>(
+    '~/server/redis/sys-read-deadline'
+  );
+  const call =
+    (fn: string) =>
+    (...args: never[]) =>
+      holder.backing
+        ? holder.backing[fn](...args)
+        : Promise.reject(new Error('sysRedis unavailable (test outage)'));
+  return {
+    sysRedis: {
+      hGet: call('hGet'),
+      hSet: call('hSet'),
+      sAdd: call('sAdd'),
+      sMembers: call('sMembers'),
+      del: call('del'),
+      exists: call('exists'),
+      set: call('set'),
+    },
+    REDIS_SYS_KEYS: { QUEUES: { BUCKETS: `${NS}:buckets` } },
+    REDIS_SUB_KEYS: { QUEUES: { MERGING: 'merging' } },
+    withSysReadDeadline,
+  };
+});
+
+vi.mock('~/server/redis/fail-open-log', () => ({ logSysRedisFailOpen: vi.fn() }));
+vi.mock('~/server/logging/client', () => ({ logToAxiom: vi.fn(() => Promise.resolve()) }));
+
+let prisma: PrismaClient;
+let bootstrap: PrismaClient;
+// A getter, not a value: the client cannot exist until beforeAll has the URL, and
+// queues.ts captures the binding at import.
+vi.mock('~/server/db/client', () => ({
+  get dbWrite() {
+    return prisma;
+  },
+  get dbRead() {
+    return prisma;
+  },
+}));
+
+const { addToQueue, drainDroppedEnqueues } = await import('~/server/redis/queues');
+
+const PARKED_KEY = 'search-index-queue-fallback:images_v6:Delete';
+
+let redis: ReturnType<typeof createClient>;
+
+describe.skipIf(!databaseUrl || !redisUrl)('queues parking lot — real Postgres + Redis', () => {
+  beforeAll(async () => {
+    // A scratch schema, so this can be pointed at any database without touching its real
+    // KeyValue. `?schema=` is what puts it on the session search_path, which is how the
+    // unqualified "KeyValue" inside queues.ts resolves here — so the schema has to exist
+    // before the client that uses it connects.
+    bootstrap = new PrismaClient({ datasources: { db: { url: databaseUrl } } });
+    await bootstrap.$executeRawUnsafe(`CREATE SCHEMA IF NOT EXISTS ${SCHEMA}`);
+
+    const scoped = new URL(databaseUrl as string);
+    scoped.searchParams.set('schema', SCHEMA);
+    prisma = new PrismaClient({ datasources: { db: { url: scoped.toString() } } });
+    await prisma.$executeRawUnsafe(
+      `CREATE TABLE IF NOT EXISTS ${SCHEMA}."KeyValue" ("key" text PRIMARY KEY, "value" jsonb NOT NULL)`
+    );
+
+    redis = createClient({ url: redisUrl });
+    await redis.connect();
+  }, 30000);
+
+  afterAll(async () => {
+    if (prisma) await prisma.$disconnect();
+    if (bootstrap) {
+      await bootstrap.$executeRawUnsafe(`DROP SCHEMA IF EXISTS ${SCHEMA} CASCADE`);
+      await bootstrap.$disconnect();
+    }
+    if (redis?.isOpen) {
+      const keys = await redis.keys(`${NS}*`);
+      if (keys.length) await redis.del(keys);
+      await redis.quit();
+    }
+  });
+
+  beforeEach(async () => {
+    await prisma.$executeRawUnsafe(`TRUNCATE ${SCHEMA}."KeyValue"`);
+    const keys = await redis.keys(`${NS}*`);
+    if (keys.length) await redis.del(keys);
+    holder.backing = redis as never;
+  });
+
+  const parked = () =>
+    prisma.$queryRawUnsafe<{ key: string; value: number[] }[]>(
+      `SELECT "key","value" FROM ${SCHEMA}."KeyValue" ORDER BY "key"`
+    );
+
+  const bucketMembers = async () => {
+    const bucket = await redis.hGet(`${NS}:buckets`, 'images_v6:Delete');
+    return bucket ? redis.sMembers(bucket) : [];
+  };
+
+  it('a healthy enqueue reaches redis and writes nothing to Postgres', async () => {
+    await expect(addToQueue('images_v6:Delete', [1, 2, 3])).resolves.toBe(true);
+
+    expect(await bucketMembers()).toEqual(expect.arrayContaining(['1', '2', '3']));
+    expect(await parked()).toHaveLength(0);
+  });
+
+  it('an outage parks the ids in Postgres instead of losing them', async () => {
+    holder.backing = null;
+
+    await expect(addToQueue('images_v6:Delete', [10, 11])).resolves.toBe(false);
+
+    expect(await parked()).toEqual([{ key: PARKED_KEY, value: [10, 11] }]);
+  });
+
+  it('successive drops during one outage accumulate under the same key', async () => {
+    holder.backing = null;
+    await addToQueue('images_v6:Delete', [10, 11]);
+    await addToQueue('images_v6:Delete', [12]);
+
+    expect((await parked())[0].value).toEqual([10, 11, 12]);
+  });
+
+  it('the drain replays them into redis once the outage ends, and clears the row', async () => {
+    holder.backing = null;
+    await addToQueue('images_v6:Delete', [10, 11]);
+    holder.backing = redis as never;
+
+    await expect(drainDroppedEnqueues()).resolves.toEqual({ keys: 1, replayed: 2, reparked: 0 });
+
+    expect(await bucketMembers()).toEqual(expect.arrayContaining(['10', '11']));
+    expect(await parked()).toHaveLength(0);
+  });
+
+  it('the drain leaves the row parked while the outage continues, and never duplicates it', async () => {
+    holder.backing = null;
+    await addToQueue('images_v6:Delete', [10, 11]);
+
+    await expect(drainDroppedEnqueues()).resolves.toEqual({ keys: 1, replayed: 0, reparked: 2 });
+    // Still exactly the two ids — a re-park would have appended a second copy.
+    expect((await parked())[0].value).toEqual([10, 11]);
+
+    holder.backing = redis as never;
+    await expect(drainDroppedEnqueues()).resolves.toEqual({ keys: 1, replayed: 2, reparked: 0 });
+    expect(await parked()).toHaveLength(0);
+  });
+
+  it('a drop landing between the drain read and its delete is not swallowed', async () => {
+    holder.backing = null;
+    await addToQueue('images_v6:Delete', [10, 11]);
+    holder.backing = redis as never;
+
+    // The interleaving the length guard exists for: the drain has read [10,11] and is
+    // about to delete when another pod parks id 12. The delete must refuse the grown row.
+    await prisma.$executeRawUnsafe(
+      `UPDATE ${SCHEMA}."KeyValue" SET "value" = "value" || '[12]'::jsonb WHERE "key" = $1`,
+      PARKED_KEY
+    );
+    await prisma.$executeRawUnsafe(
+      `DELETE FROM ${SCHEMA}."KeyValue" WHERE "key" = $1 AND jsonb_typeof("value") = 'array' AND jsonb_array_length("value") = $2`,
+      PARKED_KEY,
+      2
+    );
+
+    expect((await parked())[0].value).toEqual([10, 11, 12]);
+  });
+
+  it('a row whose value is not an array is dropped rather than retried forever', async () => {
+    await prisma.$executeRawUnsafe(
+      `INSERT INTO ${SCHEMA}."KeyValue" ("key","value") VALUES ($1, '{"not":"an array"}'::jsonb)`,
+      PARKED_KEY
+    );
+
+    await expect(drainDroppedEnqueues()).resolves.toEqual({ keys: 1, replayed: 0, reparked: 0 });
+    expect(await parked()).toHaveLength(0);
+  });
+});

src/server/redis/__tests__/queues.test.ts

@@ -8,42 +8,58 @@ import { beforeEach, describe, expect, it, vi } from 'vitest';
 // Everything referenced by a vi.mock factory must be hoisted (vi.mock is lifted
 // to the top of the file). `state.deadlineDisabled` is a mutable holder tests
 // flip to drop the deadline guard for the busy-loop cap test.
-const { hGet, hSet, sAdd, sMembers, del, exists, set, withSysReadDeadline, logSysRedisFailOpen, state } =
-  vi.hoisted(() => {
-    // Real-ish wall-clock deadline race. Mirrors sys-read-deadline.ts so the
-    // SLOW/hang path is genuinely exercised: a never-resolving op loses the race
-    // and rejects with a timeout error, which queues.ts must catch and fail open.
-    // The INTERNAL timer is a fixed small value (env-independent) regardless of
-    // the requested `ms` — so a large consumer deadline (15s) doesn't make the
-    // SLOW tests actually wait 15s; the reported error message still echoes the
-    // requested `ms` (so the consumer deadline is observable via the message).
-    const INTERNAL_DEADLINE_MS = 50;
-    const holder = { deadlineDisabled: false };
-    return {
-      hGet: vi.fn(),
-      hSet: vi.fn(() => Promise.resolve(1)),
-      sAdd: vi.fn(() => Promise.resolve(1)),
-      sMembers: vi.fn((_bucket?: string) => Promise.resolve([] as string[])),
-      del: vi.fn(() => Promise.resolve(1)),
-      exists: vi.fn(() => Promise.resolve(0)),
-      set: vi.fn(() => Promise.resolve('OK')),
-      logSysRedisFailOpen: vi.fn(),
-      state: holder,
-      withSysReadDeadline: vi.fn(<T>(p: Promise<T>, ms?: number): Promise<T> => {
-        if (holder.deadlineDisabled) return p;
-        let timer: ReturnType<typeof setTimeout> | undefined;
-        const deadline = new Promise<never>((_, reject) => {
-          timer = setTimeout(
-            () => reject(new Error(`sysRedis read timed out after ${ms ?? 'default'}ms`)),
-            INTERNAL_DEADLINE_MS
-          );
-        });
-        return Promise.race([p, deadline]).finally(() => {
-          if (timer) clearTimeout(timer);
-        });
-      }),
-    };
-  });
+const {
+  hGet,
+  hSet,
+  sAdd,
+  sMembers,
+  del,
+  exists,
+  set,
+  withSysReadDeadline,
+  logSysRedisFailOpen,
+  executeRaw,
+  queryRaw,
+  logToAxiom,
+  state,
+} = vi.hoisted(() => {
+  // Real-ish wall-clock deadline race. Mirrors sys-read-deadline.ts so the
+  // SLOW/hang path is genuinely exercised: a never-resolving op loses the race
+  // and rejects with a timeout error, which queues.ts must catch and fail open.
+  // The INTERNAL timer is a fixed small value (env-independent) regardless of
+  // the requested `ms` — so a large consumer deadline (15s) doesn't make the
+  // SLOW tests actually wait 15s; the reported error message still echoes the
+  // requested `ms` (so the consumer deadline is observable via the message).
+  const INTERNAL_DEADLINE_MS = 50;
+  const holder = { deadlineDisabled: false };
+  return {
+    hGet: vi.fn(),
+    hSet: vi.fn(() => Promise.resolve(1)),
+    sAdd: vi.fn(() => Promise.resolve(1)),
+    sMembers: vi.fn((_bucket?: string) => Promise.resolve([] as string[])),
+    del: vi.fn(() => Promise.resolve(1)),
+    exists: vi.fn(() => Promise.resolve(0)),
+    set: vi.fn(() => Promise.resolve('OK')),
+    logSysRedisFailOpen: vi.fn(),
+    executeRaw: vi.fn(() => Promise.resolve(1)),
+    queryRaw: vi.fn(() => Promise.resolve([] as { key: string; value: unknown }[])),
+    logToAxiom: vi.fn(() => Promise.resolve()),
+    state: holder,
+    withSysReadDeadline: vi.fn(<T>(p: Promise<T>, ms?: number): Promise<T> => {
+      if (holder.deadlineDisabled) return p;
+      let timer: ReturnType<typeof setTimeout> | undefined;
+      const deadline = new Promise<never>((_, reject) => {
+        timer = setTimeout(
+          () => reject(new Error(`sysRedis read timed out after ${ms ?? 'default'}ms`)),
+          INTERNAL_DEADLINE_MS
+        );
+      });
+      return Promise.race([p, deadline]).finally(() => {
+        if (timer) clearTimeout(timer);
+      });
+    }),
+  };
+});
 
 vi.mock('~/server/redis/client', () => ({
   sysRedis: { hGet, hSet, sAdd, sMembers, del, exists, set },
@@ -56,7 +72,16 @@ vi.mock('~/server/redis/client', () => ({
 // logging client (which opens its own IO) and so we can assert it was called.
 vi.mock('~/server/redis/fail-open-log', () => ({ logSysRedisFailOpen }));
 
-import { addToQueue, checkoutQueue, mergeQueue } from '~/server/redis/queues';
+// The Postgres parking lot for dropped enqueues. Mocked at the client so the raw
+// SQL shape stays observable — the tests below assert that ids were handed to it,
+// which is the only way a lost delete is distinguishable from a dropped one.
+vi.mock('~/server/db/client', () => ({
+  dbWrite: { $executeRaw: executeRaw, $queryRaw: queryRaw },
+  dbRead: { $executeRaw: executeRaw, $queryRaw: queryRaw },
+}));
+vi.mock('~/server/logging/client', () => ({ logToAxiom }));
+
+import { addToQueue, checkoutQueue, drainDroppedEnqueues, mergeQueue } from '~/server/redis/queues';
 
 // The bucket value is always persisted as a comma-joined string (see hSet calls
 // in queues.ts). This is the exact value the failing prod path read back.
@@ -73,8 +98,19 @@ beforeEach(() => {
   del.mockResolvedValue(1);
   exists.mockResolvedValue(0);
   set.mockResolvedValue('OK');
+  // Implementations, not just call history — vi.clearAllMocks only clears the latter,
+  // so a per-test mockResolvedValue would otherwise leak into the next test.
+  executeRaw.mockResolvedValue(1);
+  queryRaw.mockResolvedValue([]);
 });
 
+// $queryRaw/$executeRaw are tagged templates: call[0] is the strings array, the rest
+// are the interpolated values. The parking lot issues three different statements
+// through the same two mocks, so tests match on the statement rather than the mock.
+const sqlOf = (call: unknown[]) => (call[0] as unknown as string[]).join(' ? ');
+const callsMatching = (mock: { mock: { calls: unknown[][] } }, fragment: string) =>
+  mock.mock.calls.filter((c) => sqlOf(c).includes(fragment));
+
 describe('getBucketNames (via queues.ts public API)', () => {
   // Regression: the HA/Sentinel sysRedis client returns BLOB_STRING replies as a
   // Buffer. `currentBucket?.split(',')` then threw `i?.split is not a function`,
@@ -148,7 +184,7 @@ describe('queues fail-open — DOWN (sysRedis command rejects fast)', () => {
   it('addToQueue: a rejecting hGet (bucket-list) read SKIPS the enqueue — does not throw, does not clobber', async () => {
     hGet.mockImplementation(DOWN); // getBucketNames read is DOWN → degraded
 
-    await expect(addToQueue('images_v6:Update', [1, 2, 3])).resolves.toBeUndefined();
+    await expect(addToQueue('images_v6:Update', [1, 2, 3])).resolves.toBe(false);
     // Non-destructive: a false-empty bucket-list read must NOT drive a
     // bucket-reference overwrite (which would orphan pre-existing buckets).
     expect(hSet).not.toHaveBeenCalled();
@@ -167,19 +203,23 @@ describe('queues fail-open — DOWN (sysRedis command rejects fast)', () => {
     );
   });
 
-  it('addToQueue: a rejecting write (hSet/sAdd) is swallowed best-effort — does not throw', async () => {
+  it('addToQueue: a rejecting bucket registration parks the ids instead of writing them nowhere', async () => {
     hGet.mockResolvedValue(null); // empty queue → mints a new bucket
     hSet.mockImplementation(DOWN);
     sAdd.mockImplementation(DOWN);
 
-    await expect(addToQueue('images_v6:Update', [1, 2, 3])).resolves.toBeUndefined();
+    await expect(addToQueue('images_v6:Update', [1, 2, 3])).resolves.toBe(false);
     expect(logSysRedisFailOpen).toHaveBeenCalledWith(
       'write-degraded',
       'queues.addToQueue hSet',
       expect.any(Error),
       expect.any(Object)
     );
-    expect(logSysRedisFailOpen).toHaveBeenCalledWith(
+    // A bucket whose registration failed is unreachable by every consumer, so the
+    // ids must NOT be written into it — they go to the parking lot instead.
+    expect(sAdd).not.toHaveBeenCalled();
+    expect(callsMatching(queryRaw, 'INSERT INTO "KeyValue"')).toHaveLength(1);
+    expect(logSysRedisFailOpen).not.toHaveBeenCalledWith(
       'write-degraded',
       'queues.addToQueue sAdd',
       expect.any(Error),
@@ -216,7 +256,7 @@ describe('queues fail-open — SLOW (sysRedis command parks; only the deadline s
     // race in withSysReadDeadline is the only thing that unblocks the caller.
     hGet.mockImplementation(never);
 
-    await expect(addToQueue('images_v6:Update', [1, 2, 3])).resolves.toBeUndefined();
+    await expect(addToQueue('images_v6:Update', [1, 2, 3])).resolves.toBe(false);
     expect(withSysReadDeadline).toHaveBeenCalled();
     // Degraded read → skip; never clobbers the (unknown) real bucket list.
     expect(hSet).not.toHaveBeenCalled();
@@ -247,7 +287,7 @@ describe('queues fail-open — SLOW (sysRedis command parks; only the deadline s
     hGet.mockResolvedValue(null);
     hSet.mockImplementation(never);
 
-    await expect(addToQueue('images_v6:Update', [1, 2, 3])).resolves.toBeUndefined();
+    await expect(addToQueue('images_v6:Update', [1, 2, 3])).resolves.toBe(false);
     expect(logSysRedisFailOpen).toHaveBeenCalledWith(
       'write-degraded',
       'queues.addToQueue hSet',
@@ -320,9 +360,7 @@ describe('queues data-integrity — non-destructive fail-open (regression guard)
     // Two queued buckets; B1 reads fine, B2's read fails open mid-checkout.
     hGet.mockResolvedValue(`${B1},${B2}`);
     sMembers.mockImplementation((bucket?: string) =>
-      bucket === B1
-        ? Promise.resolve(['1'])
-        : Promise.reject(new Error('Redis connection lost'))
+      bucket === B1 ? Promise.resolve(['1']) : Promise.reject(new Error('Redis connection lost'))
     );
 
     const queue = await checkoutQueue('images_v6:Update', false, false);
@@ -403,3 +441,125 @@ describe('queues data-integrity — non-destructive fail-open (regression guard)
     expect(hSet).toHaveBeenCalledWith('queues:buckets', 'images_v6:Update', '');
   });
 });
+
+// ---------------------------------------------------------------------------
+// The Postgres parking lot (see FALLBACK_KEY_PREFIX in queues.ts).
+//
+// A dropped enqueue used to be silently unrecoverable for Delete keys — nothing
+// re-derives a delete from a row that is already gone, so the search document
+// survived forever. These assert the ids reach durable storage and come back out,
+// because a revert of that behaviour is otherwise invisible: addToQueue swallows
+// everything by design, so no failure surfaces at the call site.
+// ---------------------------------------------------------------------------
+describe('dropped-enqueue parking lot', () => {
+  const DOWN = () => Promise.reject(new Error('Redis connection lost'));
+  const PARKED_KEY = 'search-index-queue-fallback:images_v6:Delete';
+  const parkPayload = (call: unknown[]) =>
+    call.slice(1).find((v) => typeof v === 'string' && (v as string).startsWith('[')) as string;
+
+  it('parks the ids in Postgres when the bucket-list read is degraded', async () => {
+    hGet.mockImplementation(DOWN);
+
+    await expect(addToQueue('images_v6:Delete', [7, 8, 9])).resolves.toBe(false);
+
+    const inserts = callsMatching(queryRaw, 'INSERT INTO "KeyValue"');
+    expect(inserts).toHaveLength(1);
+    expect(inserts[0].slice(1)).toContain(PARKED_KEY);
+    expect(parkPayload(inserts[0])).toBe(JSON.stringify([7, 8, 9]));
+  });
+
+  it('parks only the chunks that actually failed', async () => {
+    hGet.mockResolvedValue(BUCKETS_CSV);
+    let call = 0;
+    sAdd.mockImplementation(() =>
+      ++call === 2 ? Promise.reject(new Error('down')) : Promise.resolve(1)
+    );
+    const ids = Array.from({ length: 25000 }, (_, i) => i + 1);
+
+    await expect(addToQueue('images_v6:Delete', ids)).resolves.toBe(false);
+
+    const inserts = callsMatching(queryRaw, 'INSERT INTO "KeyValue"');
+    expect(inserts).toHaveLength(1);
+    const parked = JSON.parse(parkPayload(inserts[0]));
+    expect(parked).toHaveLength(10000);
+    expect(parked[0]).toBe(10001); // the second chunk, not the first or third
+  });
+
+  it('does not touch Postgres on a healthy enqueue', async () => {
+    hGet.mockResolvedValue(BUCKETS_CSV);
+    await expect(addToQueue('images_v6:Delete', [1, 2, 3])).resolves.toBe(true);
+    expect(queryRaw).not.toHaveBeenCalled();
+    expect(executeRaw).not.toHaveBeenCalled();
+  });
+
+  it('reports the cap so a full parking lot is not a silent discard', async () => {
+    hGet.mockImplementation(DOWN);
+    queryRaw.mockResolvedValue([{ capped: true }]);
+
+    await addToQueue('images_v6:Delete', [1, 2, 3]);
+
+    expect(logToAxiom).toHaveBeenCalledWith(
+      expect.objectContaining({
+        type: 'error',
+        name: 'search-index-queue-fallback',
+        message: expect.stringContaining('cap'),
+      })
+    );
+  });
+
+  describe('drain', () => {
+    const parkedRow = (value: unknown) => [{ key: PARKED_KEY, value }];
+    // The SELECT and the parking INSERT share the queryRaw mock; route by statement.
+    const withParked = (value: unknown) =>
+      queryRaw.mockImplementation((...call: unknown[]) =>
+        Promise.resolve(sqlOf(call).includes('SELECT "key", "value"') ? parkedRow(value) : [])
+      );
+
+    it('replays parked ids and then deletes the row it replayed', async () => {
+      withParked([4, 5]);
+      hGet.mockResolvedValue(BUCKETS_CSV);
+
+      await expect(drainDroppedEnqueues()).resolves.toEqual({
+        keys: 1,
+        replayed: 2,
+        reparked: 0,
+      });
+      expect(sAdd).toHaveBeenCalledWith(BUCKETS_CSV, ['4', '5']);
+
+      // Deleted by key AND by the length it replayed, so a drop that landed between the
+      // read and the delete is not swallowed along with it.
+      const deletes = callsMatching(executeRaw, 'DELETE FROM "KeyValue"');
+      expect(deletes).toHaveLength(1);
+      expect(deletes[0].slice(1)).toEqual([PARKED_KEY, 2]);
+    });
+
+    it('leaves the row parked when the replay fails, and does not park a second copy', async () => {
+      withParked([4, 5]);
+      hGet.mockImplementation(DOWN);
+
+      await expect(drainDroppedEnqueues()).resolves.toEqual({
+        keys: 1,
+        replayed: 0,
+        reparked: 2,
+      });
+      // The row is still there — deleting it before a successful replay is what would
+      // lose the ids, and re-inserting would duplicate them.
+      expect(callsMatching(executeRaw, 'DELETE FROM "KeyValue"')).toHaveLength(0);
+      expect(callsMatching(queryRaw, 'INSERT INTO "KeyValue"')).toHaveLength(0);
+    });
+
+    it('drops a row whose value is not an array rather than retrying it forever', async () => {
+      withParked({ not: 'an array' });
+      hGet.mockResolvedValue(BUCKETS_CSV);
+
+      await expect(drainDroppedEnqueues()).resolves.toEqual({
+        keys: 1,
+        replayed: 0,
+        reparked: 0,
+      });
+      const deletes = callsMatching(executeRaw, 'jsonb_typeof("value") <> \'array\'');
+      expect(deletes).toHaveLength(1);
+      expect(sAdd).not.toHaveBeenCalled();
+    });
+  });
+});

src/server/redis/queues.ts

@@ -6,6 +6,8 @@ import {
   withSysReadDeadline,
 } from '~/server/redis/client';
 import { logSysRedisFailOpen } from '~/server/redis/fail-open-log';
+import { dbWrite } from '~/server/db/client';
+import { logToAxiom } from '~/server/logging/client';
 
 // ---------------------------------------------------------------------------
 // Fail-open sysRedis helpers for the queue used by search-index AND metrics.
@@ -41,12 +43,14 @@ import { logSysRedisFailOpen } from '~/server/redis/fail-open-log';
 // whose `sMembers` failed open is left queued for the next run. Prefer
 // "skip + retry next run" over "proceed on a false-empty read + destructive write".
 //
-// Automatic recovery for a dropped enqueue: for search-index it's the delta
-// `update` job's `updatedAt` range-scan (≤15min) plus the daily
-// `search-index-cleanup` (dropped-delete orphans) — NOT the full-reset job,
-// which runs at UNRUNNABLE_JOB_CRON (manual, unscheduled). Metrics have no such
-// range-scan, so a dropped metrics enqueue just yields momentarily stale metrics
-// until the entity is next touched.
+// Recovery for a dropped enqueue: the ids are parked in Postgres and replayed by
+// `search-index-queue-drain` (see FALLBACK_KEY_PREFIX below). That is the only
+// recovery a dropped DELETE has ever had — the delta `update` job's `updatedAt`
+// range-scan re-derives updates but cannot re-derive a delete, and the daily
+// `search-index-cleanup` reconciles neither images index (`CLEANUP_INDEXES` in
+// meilisearch/cleanup.ts covers models/articles/users/collections/bounties/tools/
+// comics only). Metrics likewise have no range-scan, so the parking lot is their
+// only recovery too.
 //
 // `withSysReadDeadline` is named for reads but is functionally a
 // `Promise.race([op, deadline])` — it unblocks the CALLER even for a write (the
@@ -92,16 +96,21 @@ async function safeSysRead<T>(
 /**
  * Deadline-raced + fail-open sysRedis WRITE. On DOWN or SLOW the write is
  * dropped (best-effort) and a `write-degraded` fail-open warning is logged.
+ *
+ * Returns whether the write landed, so a caller with somewhere durable to put
+ * the work can tell a completed write from a swallowed one.
  */
 async function safeSysWrite(
   op: () => Promise<unknown>,
   fn: string,
   extra?: Record<string, unknown>
-): Promise<void> {
+): Promise<boolean> {
   try {
     await withSysReadDeadline(op());
+    return true;
   } catch (err) {
     logSysRedisFailOpen('write-degraded', fn, err, extra);
+    return false;
   }
 }
 
@@ -137,7 +146,129 @@ function getNewBucket(key: string) {
 
 const QUEUE_ADD_CHUNK_SIZE = 10000;
 
-export async function addToQueue(key: string, ids: number | number[] | Set<number>) {
+/**
+ * Postgres parking lot for enqueues sysRedis refused. `Delete` is why it exists: a
+ * dropped `Update` is re-derived by the delta `updatedAt` range-scan, but nothing can
+ * re-derive a delete from a row that is already gone, so the id is simply lost and the
+ * document stays in the index forever. Postgres is the store that is, by construction,
+ * not the one currently failing.
+ */
+const FALLBACK_KEY_PREFIX = 'search-index-queue-fallback:';
+
+/**
+ * Ceiling per queue key. A single fan-out can carry ~211K ids and this is a JSON column,
+ * not a queue — past this the parking lot stops absorbing rather than growing without
+ * bound through a long outage. Hitting it is itself the signal that the outage needs a
+ * full reindex, not a replay.
+ */
+const FALLBACK_MAX_IDS_PER_KEY = 100000;
+
+async function persistDroppedEnqueue(key: string, ids: number[]) {
+  if (!ids.length) return;
+  try {
+    const [row] = await dbWrite.$queryRaw<{ capped: boolean }[]>`
+      INSERT INTO "KeyValue" ("key", "value")
+      VALUES (${FALLBACK_KEY_PREFIX + key}, ${JSON.stringify(ids)}::jsonb)
+      ON CONFLICT ("key") DO UPDATE SET "value" =
+        CASE
+          WHEN jsonb_array_length("KeyValue"."value") >= ${FALLBACK_MAX_IDS_PER_KEY}
+            THEN "KeyValue"."value"
+          ELSE "KeyValue"."value" || EXCLUDED."value"
+        END
+      RETURNING (jsonb_array_length("value") >= ${FALLBACK_MAX_IDS_PER_KEY}) AS capped
+    `;
+    // The CASE above keeps the old value once the cap is reached, which discards the
+    // incoming ids. Say so: a full parking lot means the outage has outlasted what a
+    // replay can fix and the index needs reconciling, and that has to be louder than
+    // the per-drop warning the caller already logged.
+    if (row?.capped) {
+      logToAxiom({
+        type: 'error',
+        name: 'search-index-queue-fallback',
+        message: `parking lot for ${key} is at the ${FALLBACK_MAX_IDS_PER_KEY} id cap; ${ids.length} id(s) discarded`,
+      }).catch(() => undefined);
+    }
+  } catch (err) {
+    // Both stores are now failing. Nothing left to try — log and let the caller
+    // report the drop, exactly as it did before this fallback existed.
+    logToAxiom({
+      type: 'error',
+      name: 'search-index-queue-fallback',
+      message: `could not park ${ids.length} dropped id(s) for ${key}: ${(err as Error).message}`,
+    }).catch(() => undefined);
+  }
+}
+
+/**
+ * Replay everything parked back onto the real queue.
+ *
+ * 🔴 Reads BEFORE it deletes, and deletes only the row it actually replayed — the same
+ * rule `checkoutQueue` follows below, for the same reason. An earlier version used
+ * `DELETE … RETURNING` as the checkout, which commits the removal before the replay is
+ * attempted: a pod dying in that window destroyed the very ids this table exists to
+ * protect. Nothing else can rebuild them.
+ *
+ * Replaying is safe to repeat, which is what makes read-then-delete affordable here: a
+ * `Delete` for a document already gone and an `Update` for an unchanged row are both
+ * no-ops downstream.
+ */
+export async function drainDroppedEnqueues() {
+  const rows = await dbWrite.$queryRaw<{ key: string; value: unknown }[]>`
+    SELECT "key", "value" FROM "KeyValue" WHERE "key" LIKE ${`${FALLBACK_KEY_PREFIX}%`}
+  `;
+
+  let replayed = 0;
+  let reparked = 0;
+  for (const row of rows) {
+    if (!Array.isArray(row.value)) {
+      // No replay can ever consume this, so retrying it forever is just a leak.
+      await dbWrite.$executeRaw`
+        DELETE FROM "KeyValue"
+        WHERE "key" = ${row.key} AND jsonb_typeof("value") <> 'array'
+      `;
+      continue;
+    }
+
+    const ids = row.value as number[];
+    // `park: false` — the row IS the parking lot entry, and it is still there. Letting
+    // the enqueue re-park on failure would append a second copy of every id.
+    if (ids.length && !(await enqueue(row.key.slice(FALLBACK_KEY_PREFIX.length), ids, false))) {
+      reparked += ids.length;
+      continue; // left parked; the next run retries it
+    }
+    replayed += ids.length;
+
+    // Length guard, not an unconditional delete by key: a drop landing between the read
+    // and here appends to the same row, and deleting it wholesale would swallow ids that
+    // were never replayed. A grown row survives and is replayed in full next run —
+    // duplicated work, which is a no-op, rather than lost work, which is not.
+    await dbWrite.$executeRaw`
+      DELETE FROM "KeyValue"
+      WHERE "key" = ${row.key}
+        AND jsonb_typeof("value") = 'array'
+        AND jsonb_array_length("value") = ${ids.length}
+    `;
+  }
+  return { keys: rows.length, replayed, reparked };
+}
+
+/** @returns whether every id reached the queue; dropped ids are parked for replay. */
+export async function addToQueue(
+  key: string,
+  ids: number | number[] | Set<number>
+): Promise<boolean> {
+  return enqueue(key, ids, true);
+}
+
+/**
+ * @param park whether a dropped id should be written to the Postgres parking lot. Only
+ * the drain passes false, because its ids are already parked.
+ */
+async function enqueue(
+  key: string,
+  ids: number | number[] | Set<number>,
+  park: boolean
+): Promise<boolean> {
   if (!Array.isArray(ids)) {
     if (ids instanceof Set) ids = Array.from(ids);
     else ids = [ids];
@@ -146,36 +277,51 @@ export async function addToQueue(key: string, ids: number | number[] | Set<numbe
   if (degraded) {
     // The bucket-list read failed open (false-empty). Writing a fresh bucket
     // reference here (`hSet(BUCKETS, key, newBucket)`) would OVERWRITE the hash
-    // field and orphan any pre-existing buckets. Skip the enqueue entirely — the
-    // update is dropped (recovered by the delta update-scan / next trigger)
-    // rather than clobbering the queue.
+    // field and orphan any pre-existing buckets, so the enqueue cannot proceed
+    // against redis — it goes to the parking lot instead.
     logSysRedisFailOpen(
       'write-degraded',
       'queues.addToQueue skipped-degraded-read',
       new Error('bucket-list read degraded; enqueue skipped to avoid orphaning existing buckets'),
       { key }
     );
-    return;
+    if (park) await persistDroppedEnqueue(key, ids);
+    return false;
   }
   let targetBucket = currentBuckets[0];
   if (!targetBucket) {
     targetBucket = getNewBucket(key);
-    await safeSysWrite(
+    const registered = await safeSysWrite(
       () => sysRedis.hSet(REDIS_SYS_KEYS.QUEUES.BUCKETS, key, targetBucket),
       'queues.addToQueue hSet',
       { key }
     );
+    // No consumer can reach an unregistered bucket, so writing ids into it would
+    // put them somewhere nothing ever reads — indistinguishable from losing them.
+    if (!registered) {
+      if (park) await persistDroppedEnqueue(key, ids);
+      return false;
+    }
   }
   const content = ids.map((id) => id.toString());
+  const dropped: number[] = [];
   // Chunked because callers can enqueue very large id sets in one go — propagating a model
   // flag to its gallery reaches ~211K images on the largest model — and a single sAdd that
   // size is a multi-MB command that stalls everything else on the connection.
   for (let i = 0; i < content.length; i += QUEUE_ADD_CHUNK_SIZE) {
     const chunk = content.slice(i, i + QUEUE_ADD_CHUNK_SIZE);
-    await safeSysWrite(() => sysRedis.sAdd(targetBucket, chunk), 'queues.addToQueue sAdd', {
-      key,
-    });
+    const written = await safeSysWrite(
+      () => sysRedis.sAdd(targetBucket, chunk),
+      'queues.addToQueue sAdd',
+      { key }
+    );
+    if (!written) dropped.push(...ids.slice(i, i + QUEUE_ADD_CHUNK_SIZE));
   }
+  if (dropped.length) {
+    if (park) await persistDroppedEnqueue(key, dropped);
+    return false;
+  }
+  return true;
 }
 
 export async function checkoutQueue(key: string, isMerge = false, readOnly = false) {
@@ -264,7 +410,8 @@ async function waitForMerge(key: string) {
   // `string`, losing the contextual RedisKeyTemplateSys match the inline literal had (mirrors
   // getNewBucket below). Without this, sysRedis.exists() rejects it (TS2345) — caught only by the
   // preview build's typecheck, not local tsc against stale @civitai/redis types.
-  const mergeKey = `${REDIS_SYS_KEYS.QUEUES.BUCKETS}:${key}:${REDIS_SUB_KEYS.QUEUES.MERGING}` as RedisKeyTemplateSys;
+  const mergeKey =
+    `${REDIS_SYS_KEYS.QUEUES.BUCKETS}:${key}:${REDIS_SUB_KEYS.QUEUES.MERGING}` as RedisKeyTemplateSys;
   for (let i = 0; i < WAIT_FOR_MERGE_MAX_ITERATIONS; i++) {
     const { value: isMerging } = await safeSysRead(
       () => sysRedis.exists(mergeKey),