diff --git a/packages/queue/src/index.js b/packages/queue/src/index.js index 987c182..394b1ac 100644 --- a/packages/queue/src/index.js +++ b/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 }), ]), ); diff --git a/packages/queue/src/workers.js b/packages/queue/src/workers.js index 3a1ed6e..0c09953 100644 --- a/packages/queue/src/workers.js +++ b/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,28 +242,38 @@ 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, }), @@ -271,6 +281,7 @@ export function startWorkers({ concurrency = {} } = {}) { // first lever if reminders start landing late under load. new Worker(QUEUES.batch, runBatch, { connection, + streams, concurrency: concurrency.batch ?? 16, }), ]; diff --git a/packages/sports/src/index.js b/packages/sports/src/index.js index ee7d99f..99c12bd 100644 --- a/packages/sports/src/index.js +++ b/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