The live tick was storing every synced row in Redis, sixty times an hour

packages/queue/src/index.js

@@ -40,10 +40,26 @@ const defaults = {
   backoff: { type: 'exponential', delay: 2000 },
 };
 
+/**
+ * How many entries a queue's event stream keeps.
+ *
+ * BullMQ's own default is 10,000, which is a sensible count and a dangerous size:
+ * a `completed` event carries the processor's return value, so the stream costs
+ * 10,000 times whatever the biggest job hands back. When the live tick briefly
+ * returned the mirror's whole payload that came to 15GB in one key, which is what
+ * filled the volume and stopped Redis saving at all. These are for observability
+ * -- nothing here reads them back -- so a thousand is plenty, and the smaller
+ * ceiling means a future fat return value shows up as a slow query rather than an
+ * outage.
+ */
+export const EVENT_STREAM_MAX_LEN = 1000;
+
+export const streams = { events: { maxLen: EVENT_STREAM_MAX_LEN } };
+
 export const queues = Object.fromEntries(
   Object.entries(QUEUES).map(([k, name]) => [
     k,
-    new Queue(name, { connection, defaultJobOptions: defaults }),
+    new Queue(name, { connection, defaultJobOptions: defaults, streams }),
   ]),
 );
 

packages/queue/src/workers.js

@@ -12,7 +12,7 @@ import {
   syncPlays,
 } from '@tipoff/sports';
 import { Worker } from 'bullmq';
-import { connection, QUEUES, queues } from './index.js';
+import { connection, QUEUES, queues, streams } from './index.js';
 
 const log = (...a) => console.log('[worker]', ...a);
 
@@ -214,7 +214,7 @@ async function runBatch(job) {
 
 export function startWorkers({ concurrency = {} } = {}) {
   const workers = [
-    new Worker(QUEUES.scan, runScan, { connection, concurrency: 1 }),
+    new Worker(QUEUES.scan, runScan, { connection, streams, concurrency: 1 }),
 
     new Worker(
       QUEUES.sync,
@@ -242,35 +242,46 @@ export function startWorkers({ concurrency = {} } = {}) {
       },
       {
         connection,
+        streams,
         concurrency: 1,
       },
     ),
 
     // Scores only; concurrency 1 because it already fans out internally and a
     // second overlapping tick would just refetch the same leagues.
-    new Worker(QUEUES.live, () => syncLiveScores(), { connection, concurrency: 1 }),
+    new Worker(QUEUES.live, () => syncLiveScores(), { connection, streams, concurrency: 1 }),
 
     // Concurrency 1: these responses are large and the point is to stagger them.
-    new Worker(QUEUES.plays, () => syncPlays(), { connection, concurrency: 1 }),
+    new Worker(QUEUES.plays, () => syncPlays(), { connection, streams, concurrency: 1 }),
 
     // Concurrency 1, and the poller itself is sequential inside. These are other
     // people's subscriptions: several ~800KB pulls at once from one datacenter IP
     // is the traffic pattern that gets a line cut off.
-    new Worker(QUEUES.playlists, () => refreshDuePlaylists(), { connection, concurrency: 1 }),
+    new Worker(QUEUES.playlists, () => refreshDuePlaylists(), {
+      connection,
+      streams,
+      concurrency: 1,
+    }),
 
     // Down only: lapsed managed lists are removed or handed back. Nothing here
     // talks to the line provider.
-    new Worker(QUEUES.livePasses, () => reconcileLapsed({ log }), { connection, concurrency: 1 }),
+    new Worker(QUEUES.livePasses, () => reconcileLapsed({ log }), {
+      connection,
+      streams,
+      concurrency: 1,
+    }),
 
     new Worker(QUEUES.fanout, runFanout, {
       connection,
+      streams,
       concurrency: concurrency.fanout ?? 4,
     }),
 
     // The delivery tier is the one that scales horizontally. Raising this is the
     // first lever if reminders start landing late under load.
     new Worker(QUEUES.batch, runBatch, {
       connection,
+      streams,
       concurrency: concurrency.batch ?? 16,
     }),
   ];

packages/sports/src/index.js

@@ -512,7 +512,15 @@ export async function syncLiveScores({ log = console.log } = {}) {
   // and nichedb's own live pass is what makes the answer minute-fresh.
   if (mirrorEnabled()) {
     const out = await nichedbsports.syncSince({ log });
-    return { leagues: 0, events: out.fixtures ?? 0, failed: 0, mirror: out };
+    /*
+     * Counts only. What a processor returns is written twice -- into the job hash
+     * and into the queue's `completed` event -- so handing back the mirror's whole
+     * payload put a copy of every synced row in Redis sixty times an hour. Nothing
+     * reads it: `bull:live-scores:events` reached 15GB of ~1.5MB entries, the RDB
+     * grew past what a background save could write to the volume, and each failed
+     * save left a multi-GB temp file behind until the disk filled.
+     */
+    return { leagues: 0, events: out.fixtures ?? 0, failed: 0 };
   }
   const leagues = await q.leaguesWithLiveGames();
   if (leagues.length === 0) return { leagues: 0, events: 0 };
@@ -568,7 +576,6 @@ export async function syncPlays({ log = console.log, limit = 8 } = {}) {
       plays: out.plays ?? 0,
       recaps: out.recaps ?? 0,
       failed: 0,
-      mirror: out,
     };
   }
   // A reserved share for the catch-up reads, but only while there is live work to