|
| 1 | +/** |
| 2 | + * How much history BullMQ keeps in Redis, in one place. |
| 3 | + * |
| 4 | + * Three things accumulate if nothing bounds them: finished job hashes (each one |
| 5 | + * holding its data and return value), the `completed` / `failed` sets indexing |
| 6 | + * them, and each queue's `events` stream. The first two are bounded per job by |
| 7 | + * `removeOnComplete` / `removeOnFail`; the stream is bounded per queue by |
| 8 | + * `streams.events.maxLen`, which BullMQ stores in the queue's meta hash when a |
| 9 | + * Queue is constructed. |
| 10 | + * |
| 11 | + * A count is not a size. Every `completed` event carries the processor's return |
| 12 | + * value, so a stream costs maxLen times whatever the fattest job hands back. |
| 13 | + * PR #78 capped the stream at 1,000 after the live tick's return put 15GB in one |
| 14 | + * key; on 2026-10-01 `bull:sync:events` still held 385MB in only 485 entries, |
| 15 | + * because sync-near and sync-all returned the mirror's whole `result` and |
| 16 | + * `cursor` (~18MB per completed event). Hence the byte ceiling and the return |
| 17 | + * value guard below, not just the count. |
| 18 | + */ |
| 19 | + |
| 20 | +/** |
| 21 | + * BullMQ's own default is 10,000. These are for observability -- nothing here |
| 22 | + * reads them back, there is no QueueEvents -- so a thousand is plenty. |
| 23 | + */ |
| 24 | +export const EVENT_STREAM_MAX_LEN = 1000; |
| 25 | + |
| 26 | +export const streams = { events: { maxLen: EVENT_STREAM_MAX_LEN } }; |
| 27 | + |
| 28 | +/** |
| 29 | + * Completed jobs: an hour, at most 5,000 per queue. |
| 30 | + * |
| 31 | + * The hour is load-bearing, not just debugging: the reminder scan dedupes its |
| 32 | + * fan-outs on a retained `fo-<event>-<offset>` jobId, and an event re-matches |
| 33 | + * for several minutes, so a completed fan-out must outlive that window. The |
| 34 | + * count stays generous for the same reason -- evicting a fan-out early by count |
| 35 | + * would let the next scan enqueue it again -- and finished jobs are small now |
| 36 | + * that every processor returns counts. |
| 37 | + */ |
| 38 | +export const KEEP_COMPLETED = { age: 3600, count: 5000 }; |
| 39 | + |
| 40 | +/** |
| 41 | + * Failed jobs: a week, so a failure seen on Monday is still inspectable on |
| 42 | + * Friday, capped at a thousand so a tick that fails every 30 seconds cannot |
| 43 | + * fill Redis with stack traces. Measured 2026-10-01: at most 6 per queue. It |
| 44 | + * was a day with no count before. |
| 45 | + */ |
| 46 | +export const KEEP_FAILED = { age: 7 * 24 * 3600, count: 1000 }; |
| 47 | + |
| 48 | +export const retention = { removeOnComplete: KEEP_COMPLETED, removeOnFail: KEEP_FAILED }; |
| 49 | + |
| 50 | +/** |
| 51 | + * An events stream larger than this is emptied rather than trimmed by count: |
| 52 | + * a thousand entries is only safe while each entry is small, and a stream |
| 53 | + * this big means some processor returned bulk data, and those entries are |
| 54 | + * spread through it, so no count-based trim reaches all of them. |
| 55 | + */ |
| 56 | +export const EVENT_STREAM_MAX_BYTES = 32 * 1024 * 1024; |
| 57 | + |
| 58 | +/** What a processor may hand back before it is replaced by a summary. */ |
| 59 | +export const RETURN_VALUE_MAX_BYTES = 16 * 1024; |
| 60 | + |
| 61 | +/** |
| 62 | + * Wrap a processor so its return value can never be bulk data. |
| 63 | + * |
| 64 | + * BullMQ writes the return value twice -- the job hash and the `completed` |
| 65 | + * event -- and nothing in this app reads either back. Returning counts is the |
| 66 | + * rule; this is the backstop for the next time a payload slips into a return, |
| 67 | + * which has now happened twice (#78's live tick, then sync's `mirror`). |
| 68 | + */ |
| 69 | +export function boundedReturn( |
| 70 | + name, |
| 71 | + processor, |
| 72 | + { log = console.log, maxBytes = RETURN_VALUE_MAX_BYTES } = {}, |
| 73 | +) { |
| 74 | + return async (job, token) => { |
| 75 | + const value = await processor(job, token); |
| 76 | + if (value === undefined) return value; |
| 77 | + let bytes; |
| 78 | + try { |
| 79 | + bytes = JSON.stringify(value)?.length ?? 0; |
| 80 | + } catch { |
| 81 | + bytes = Infinity; |
| 82 | + } |
| 83 | + if (bytes <= maxBytes) return value; |
| 84 | + log(`[queue] ${name} job ${job?.id} returned ${bytes} bytes; storing a summary instead`); |
| 85 | + return { |
| 86 | + oversized: Number.isFinite(bytes) ? bytes : 'unserialisable', |
| 87 | + keys: value && typeof value === 'object' ? Object.keys(value).slice(0, 20) : typeof value, |
| 88 | + }; |
| 89 | + }; |
| 90 | +} |
| 91 | + |
| 92 | +const sleep = (ms) => new Promise((r) => setTimeout(r, ms)); |
| 93 | + |
| 94 | +/** |
| 95 | + * One pass over every queue bringing what is ALREADY in Redis inside the caps |
| 96 | + * above. New options only govern new jobs, so without this a backlog written |
| 97 | + * under the old defaults stays until the stream turns over. |
| 98 | + * |
| 99 | + * Safe to run on every boot: once everything is within bounds each queue costs |
| 100 | + * an XLEN, a MEMORY USAGE and two ZRANGEBYSCOREs, and removes nothing. It works |
| 101 | + * in small steps with a pause between them so no single command holds Redis. |
| 102 | + * Errors are logged, never thrown: this must not stop a worker from booting. |
| 103 | + */ |
| 104 | +export async function pruneQueues( |
| 105 | + queues, |
| 106 | + { |
| 107 | + log = console.log, |
| 108 | + maxLen = EVENT_STREAM_MAX_LEN, |
| 109 | + maxBytes = EVENT_STREAM_MAX_BYTES, |
| 110 | + step = 500, |
| 111 | + pauseMs = 25, |
| 112 | + } = {}, |
| 113 | +) { |
| 114 | + const report = []; |
| 115 | + for (const queue of queues) { |
| 116 | + try { |
| 117 | + const r = await pruneQueue(queue, { maxLen, maxBytes, step, pauseMs }); |
| 118 | + report.push(r); |
| 119 | + if (r.events.removed || r.completed || r.failed) { |
| 120 | + log( |
| 121 | + `[queue] pruned ${r.name}: events ${r.events.before} -> ${r.events.after} ` + |
| 122 | + `(${mb(r.events.bytes)} before), ${r.completed} completed, ${r.failed} failed`, |
| 123 | + ); |
| 124 | + } |
| 125 | + } catch (err) { |
| 126 | + log(`[queue] prune ${queue.name} failed: ${err?.message ?? err}`); |
| 127 | + } |
| 128 | + } |
| 129 | + const removed = report.reduce((n, r) => n + r.events.removed + r.completed + r.failed, 0); |
| 130 | + log(`[queue] retention pass: ${report.length} queue(s), ${removed} entr(ies) removed`); |
| 131 | + return report; |
| 132 | +} |
| 133 | + |
| 134 | +async function pruneQueue(queue, { maxLen, maxBytes, step, pauseMs }) { |
| 135 | + const client = await queue.client; |
| 136 | + const key = queue.keys.events; |
| 137 | + // Clean first: each clean call itself appends a `cleaned` event to the stream. |
| 138 | + const completed = await cleanInSteps(queue, KEEP_COMPLETED.age * 1000, 'completed', { |
| 139 | + step, |
| 140 | + pauseMs, |
| 141 | + }); |
| 142 | + const failed = await cleanInSteps(queue, KEEP_FAILED.age * 1000, 'failed', { step, pauseMs }); |
| 143 | + |
| 144 | + const before = Number(await client.xlen(key)); |
| 145 | + const bytes = Number(await client.call('MEMORY', 'USAGE', key)) || 0; |
| 146 | + const target = bytes > maxBytes ? 0 : maxLen; |
| 147 | + |
| 148 | + // Exact MAXLEN, not `~`: an approximate trim only drops whole radix nodes, |
| 149 | + // and an oversized entry is a node of its own, so `~` can leave the very |
| 150 | + // entries this exists to remove. |
| 151 | + let len = before; |
| 152 | + while (len > target) { |
| 153 | + await client.xtrim(key, 'MAXLEN', Math.max(target, len - step)); |
| 154 | + len = Number(await client.xlen(key)); |
| 155 | + if (len > target) await sleep(pauseMs); |
| 156 | + } |
| 157 | + |
| 158 | + return { |
| 159 | + name: queue.name, |
| 160 | + events: { before, after: len, removed: before - len, bytes }, |
| 161 | + completed, |
| 162 | + failed, |
| 163 | + }; |
| 164 | +} |
| 165 | + |
| 166 | +async function cleanInSteps(queue, graceMs, type, { step, pauseMs }) { |
| 167 | + // Ask first: a clean call appends a `cleaned` event even when it removes |
| 168 | + // nothing, so calling it unconditionally would grow the stream on every boot. |
| 169 | + // Finished sets are scored by finishedOn, the same clock clean() goes by. |
| 170 | + const client = await queue.client; |
| 171 | + const due = Number(await client.zcount(queue.keys[type], '-inf', Date.now() - graceMs)); |
| 172 | + if (!due) return 0; |
| 173 | + let total = 0; |
| 174 | + for (;;) { |
| 175 | + const ids = await queue.clean(graceMs, step, type); |
| 176 | + total += ids.length; |
| 177 | + if (ids.length < step) return total; |
| 178 | + await sleep(pauseMs); |
| 179 | + } |
| 180 | +} |
| 181 | + |
| 182 | +const mb = (n) => `${(n / 1024 / 1024).toFixed(1)}MB`; |
0 commit comments