From dfdc9c0bc69cb921d8cda21b6af78c000f5af181 Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Mon, 9 Nov 2026 20:42:32 +0000 Subject: [PATCH] reddit firehose: time-box the read so each pass fits its 60 s slot A pass that ran 67-87 s missed its slot on the daemon's fixed grid and waited for the next one: ~6 pages a minute against Reddit's ~9, so lag grew ~40 s a minute. Reading now stops at 40 s (gap 3.5 s, up to 30 pages), leaving the slot for the writes. Co-Authored-By: Claude Opus 5.5 --- packages/ingest/src/redditFirehose.js | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/packages/ingest/src/redditFirehose.js b/packages/ingest/src/redditFirehose.js index 3c108ea..292c0d1 100644 --- a/packages/ingest/src/redditFirehose.js +++ b/packages/ingest/src/redditFirehose.js @@ -27,6 +27,7 @@ const PAGE = 100; * lookbackSeconds?: number, * maxPages?: number, * gapMs?: number, + * fetchBudgetMs?: number, * sleep?: (ms: number) => Promise, * maxLagSeconds?: number, * resumeLagSeconds?: number, @@ -41,12 +42,17 @@ export function createRedditFirehose(opts = {}) { const now = opts.now ?? (() => Date.now()); const fetchPage = opts.fetchPage ?? fetchRedditFirehosePage; const crawl = opts.crawl ?? crawlFeed; - const maxPages = opts.maxPages ?? 12; + const maxPages = opts.maxPages ?? 30; + // Reading stops at this deadline so the pass (reads plus writes) fits its + // 60 s slot. A pass that overran the slot waited for the *next* one (the + // daemon runs on a fixed grid), which halved throughput to ~6 pages a minute + // against Reddit's ~9 (2026-11-09). + const fetchBudgetMs = opts.fetchBudgetMs ?? 40_000; const concurrency = opts.concurrency ?? 6; // Arctic Shift answers 422 "Timeout. Maybe slow down a bit" after ~5 pages // back to back (measured 2026-11-09), and a page is ~6 s of Reddit, so pages // are spaced rather than fired: ~11 a minute is just over Reddit's own pace. - const gapMs = opts.gapMs ?? 4_000; + const gapMs = opts.gapMs ?? 3_500; const sleep = opts.sleep ?? ((ms) => new Promise((resolve) => setTimeout(resolve, ms))); // Never more than this far behind: past it, skip forward so every feed stays // inside the hour, and say how much was skipped. @@ -75,7 +81,9 @@ export function createRedditFirehose(opts = {}) { cursor = nowSeconds - resumeLagSeconds; } + const started = now(); for (; pages < maxPages; pages += 1) { + if (pages > 0 && now() - started + gapMs >= fetchBudgetMs) break; if (pages > 0 && gapMs > 0) await sleep(gapMs); let page; try {