From a4ab3f5b0def54d1efaa22a410cca483951622be Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Mon, 9 Nov 2026 20:22:08 +0000 Subject: [PATCH] reddit firehose: pace pages, survive 'slow down', never lag past the hour Arctic Shift answers 422 'Timeout. Maybe slow down a bit' after ~5 back- to-back pages (measured), and pages are capped at 100 (~6 s of Reddit). So: - pages 4 s apart, 12 per pass, a pass every 60 s (~11/min, just over Reddit's pace); env REDDIT_FIREHOSE_GAP_MS / _PAGES / _MS - a refusal ends the pass but keeps and stores what it read; the next pass resumes from the cursor - more than 50 min behind, skip to 5 min back and log 'skipped' - reddit per-source floor 60 min -> 7 days: the firehose carries new posts, and per-sub asks spend the same per-IP budget - drop the on-view refresh from #278 (bots read /r/* constantly, and each view would spend that budget too) Co-Authored-By: Claude Opus 5.5 --- apps/poller/src/daemon.js | 2 +- apps/poller/src/index.js | 7 ++-- apps/web/src/app/r/[subreddit]/page.jsx | 8 +---- apps/web/src/app/r/u/[username]/page.jsx | 6 +--- apps/web/src/lib/socialRiver.js | 17 --------- packages/db/src/queries.js | 23 ------------ packages/db/test/due-tiers.test.js | 17 --------- packages/ingest/src/redditFirehose.js | 37 +++++++++++++++++-- packages/ingest/test/reddit-firehose.test.js | 38 +++++++++++++++++++- packages/social/src/collect.js | 8 +++-- packages/social/test/facebook-scrape.test.js | 5 +-- 11 files changed, 88 insertions(+), 80 deletions(-) diff --git a/apps/poller/src/daemon.js b/apps/poller/src/daemon.js index 56931fc..f986404 100644 --- a/apps/poller/src/daemon.js +++ b/apps/poller/src/daemon.js @@ -85,7 +85,7 @@ export function buildPollerDaemon({ createDaemon = defaultCreateDaemon, store, r add('import', c.intervalMs, run.import); add('purge', 3_600_000, run.purge); // Outside catch-up too: it is what keeps all of /r/* current. - if (c.redditFirehose && run.redditFirehose) add('reddit-firehose', c.redditFirehoseMs ?? 120_000, run.redditFirehose); + if (c.redditFirehose && run.redditFirehose) add('reddit-firehose', c.redditFirehoseMs ?? 60_000, run.redditFirehose); // Everything the old tick() skipped below `if (catchupOnly) return`. if (!c.catchupOnly) { diff --git a/apps/poller/src/index.js b/apps/poller/src/index.js index d79411f..f3d0fe8 100644 --- a/apps/poller/src/index.js +++ b/apps/poller/src/index.js @@ -530,8 +530,11 @@ async function topicsTick() { // of /r/* current (see packages/ingest/src/redditFirehose.js). REDDIT_FIREHOSE=0 // switches it off. const redditFirehoseEnabled = process.env.REDDIT_FIREHOSE !== '0'; -const redditFirehoseMs = Number(process.env.REDDIT_FIREHOSE_MS) || 120_000; -const redditFirehose = createRedditFirehose(); +const redditFirehoseMs = Number(process.env.REDDIT_FIREHOSE_MS) || 60_000; +const redditFirehose = createRedditFirehose({ + gapMs: Number(process.env.REDDIT_FIREHOSE_GAP_MS) || undefined, + maxPages: Number(process.env.REDDIT_FIREHOSE_PAGES) || undefined, +}); async function redditFirehoseTick() { if (stopping) return; diff --git a/apps/web/src/app/r/[subreddit]/page.jsx b/apps/web/src/app/r/[subreddit]/page.jsx index 38864ca..443c681 100644 --- a/apps/web/src/app/r/[subreddit]/page.jsx +++ b/apps/web/src/app/r/[subreddit]/page.jsx @@ -2,10 +2,7 @@ import { notFound } from 'next/navigation'; import { redditSource } from '@rssamplifier/social'; import { socialFeed, socialMetadata } from '../../../lib/socialPage.js'; -import { liveSubredditPosts, threadSlug } from '../../../lib/redditThread.js'; -import { refreshIfStale } from '../../../lib/socialRiver.js'; -import { db } from '../../../lib/db.js'; -import AddSocialSource from '../../AddSocialSource.jsx'; +import { liveSubredditPosts, threadSlug } from '../../../lib/redditThread.js';import AddSocialSource from '../../AddSocialSource.jsx'; import FeedPage from '../../[slug]/page.jsx'; import { when } from '../RedditComments.jsx'; @@ -64,9 +61,6 @@ export default async function SubredditPage({ params }) { ); } - // Older than an hour: queue a refresh. This view still shows the stored copy. - refreshIfStale(db(), feed); - // In the directory but never read: 36,679 subreddits were in that state on // 2026-10-06, queued behind one host's crawl budget. Rather than an empty // page that says "shortly", show what the archive has right now. The diff --git a/apps/web/src/app/r/u/[username]/page.jsx b/apps/web/src/app/r/u/[username]/page.jsx index cf85faf..24c7b0c 100644 --- a/apps/web/src/app/r/u/[username]/page.jsx +++ b/apps/web/src/app/r/u/[username]/page.jsx @@ -1,10 +1,7 @@ import { notFound } from 'next/navigation'; import { redditSource } from '@rssamplifier/social'; -import { socialFeed, socialMetadata } from '../../../../lib/socialPage.js'; -import { refreshIfStale } from '../../../../lib/socialRiver.js'; -import { db } from '../../../../lib/db.js'; -import AddSocialSource from '../../../AddSocialSource.jsx'; +import { socialFeed, socialMetadata } from '../../../../lib/socialPage.js';import AddSocialSource from '../../../AddSocialSource.jsx'; import FeedPage from '../../../[slug]/page.jsx'; export const dynamic = 'force-dynamic'; @@ -53,6 +50,5 @@ export default async function RedditUserPage({ params }) { ); } - refreshIfStale(db(), feed); return FeedPage({ params: Promise.resolve({ slug: String(feed.slug) }), base: source.path }); } diff --git a/apps/web/src/lib/socialRiver.js b/apps/web/src/lib/socialRiver.js index 284f6f7..f6f6070 100644 --- a/apps/web/src/lib/socialRiver.js +++ b/apps/web/src/lib/socialRiver.js @@ -89,8 +89,6 @@ export async function socialRiver({ ); } - refreshIfStale(client, feed); - const page = `${siteUrl()}${canonical}`; const rows = await q.itemsForFeed(client, String(feed.id), riverLimit(rawLimit)); @@ -121,21 +119,6 @@ export async function socialRiver({ }); } -/** - * Every /r/ URL is promised hourly. Fifty thousand subreddits cannot all be - * asked hourly through one mirror, but the ones somebody reads can: a view or a - * reader's poll of a copy older than an hour puts it at the front of the queue - * (the poller ticks every few minutes). Fire-and-forget, and never fatal: the - * stored copy is served either way. - * - * @param {object} client - * @param {{ id?: unknown, social_network?: string|null }|null} feed - */ -export function refreshIfStale(client, feed) { - if (!feed?.id || feed.social_network !== 'reddit') return; - q.requestRefreshIfStale(client, String(feed.id), 60).catch(() => {}); -} - /** * The canonical ref and path for a `/r/…` request, or null if it is not one. * diff --git a/packages/db/src/queries.js b/packages/db/src/queries.js index 3e43330..709ab9f 100644 --- a/packages/db/src/queries.js +++ b/packages/db/src/queries.js @@ -3059,29 +3059,6 @@ export async function feedsBySocialRefs(db, refs) { return out; } -/** - * `requestRefresh`, but only when the stored copy is older than `maxAgeMinutes` - * and the feed is not already at the front. For sources whose readers expect a - * fixed freshness (every /r/ URL is promised hourly): reading one is the signal - * that it is wanted, so a page view or an RSS reader's poll pulls a stale - * subreddit forward, instead of fifty thousand of them being asked hourly - * whether anyone is looking or not. One guarded UPDATE, a no-op when fresh. - * - * @param {Client} db - * @param {string} id - * @param {number} [maxAgeMinutes] - */ -export async function requestRefreshIfStale(db, id, maxAgeMinutes = 60) { - const cutoff = new Date(Date.now() - maxAgeMinutes * 60_000).toISOString(); - await db.execute({ - sql: `update feeds set next_fetch_at = '1970-01-01T00:00:00.000Z' - where id = ? and status <> 'dead' - and next_fetch_at > '1970-01-01T00:00:00.000Z' - and (last_fetched_at is null or last_fetched_at < ?)`, - args: [id, cutoff], - }); -} - /** * Due feeds per tier, and how late the oldest of each is: the number that says * whether live feeds are being read on time even while the backlog drains. diff --git a/packages/db/test/due-tiers.test.js b/packages/db/test/due-tiers.test.js index 6351199..4dcc4d2 100644 --- a/packages/db/test/due-tiers.test.js +++ b/packages/db/test/due-tiers.test.js @@ -71,20 +71,3 @@ test('requestRefresh puts a feed at the front of its tier', async () => { const due = await q.dueFeeds(db, 1); assert.equal(due[0].slug, 'dormant-39'); }); - -test('requestRefreshIfStale queues only a copy older than the limit', async () => { - const [row] = (await db.execute({ sql: `select id from feeds where slug = 'dormant-38'`, args: [] })).rows; - const id = String(row.id); - const front = '1970-01-01T00:00:00.000Z'; - const later = new Date(Date.now() + DAY).toISOString(); - const nextOf = async () => - String((await db.execute({ sql: `select next_fetch_at from feeds where id = ?`, args: [id] })).rows[0].next_fetch_at); - - await db.execute({ sql: `update feeds set last_fetched_at = ?, next_fetch_at = ? where id = ?`, args: [new Date().toISOString(), later, id] }); - await q.requestRefreshIfStale(db, id, 60); - assert.equal(await nextOf(), later, 'read minutes ago: left alone'); - - await db.execute({ sql: `update feeds set last_fetched_at = ? where id = ?`, args: [new Date(Date.now() - 2 * 3_600_000).toISOString(), id] }); - await q.requestRefreshIfStale(db, id, 60); - assert.equal(await nextOf(), front, 'two hours old: to the front'); -}); diff --git a/packages/ingest/src/redditFirehose.js b/packages/ingest/src/redditFirehose.js index c155709..3c108ea 100644 --- a/packages/ingest/src/redditFirehose.js +++ b/packages/ingest/src/redditFirehose.js @@ -26,6 +26,10 @@ const PAGE = 100; * @param {{ * lookbackSeconds?: number, * maxPages?: number, + * gapMs?: number, + * sleep?: (ms: number) => Promise, + * maxLagSeconds?: number, + * resumeLagSeconds?: number, * concurrency?: number, * env?: Record, * fetchPage?: typeof fetchRedditFirehosePage, @@ -37,8 +41,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 ?? 80; + const maxPages = opts.maxPages ?? 12; 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 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. + const maxLagSeconds = opts.maxLagSeconds ?? 50 * 60; + const resumeLagSeconds = opts.resumeLagSeconds ?? 5 * 60; /** unix seconds; posts created after this are still to be read */ let cursor = Math.floor(now() / 1000) - (opts.lookbackSeconds ?? 900); @@ -53,8 +66,26 @@ export function createRedditFirehose(opts = {}) { return async function runRedditFirehose(db) { const posts = []; let pages = 0; + let skipped = 0; + let limited = null; + + const nowSeconds = Math.floor(now() / 1000); + if (nowSeconds - cursor > maxLagSeconds) { + skipped = nowSeconds - resumeLagSeconds - cursor; + cursor = nowSeconds - resumeLagSeconds; + } + for (; pages < maxPages; pages += 1) { - const page = await fetchPage(cursor, { env: opts.env }); + if (pages > 0 && gapMs > 0) await sleep(gapMs); + let page; + try { + page = await fetchPage(cursor, { env: opts.env }); + } catch (error) { + // The mirror asking us to slow down, or a bad minute: keep what this + // pass already read and carry on from the cursor next pass. + limited = String(error?.message ?? error); + break; + } for (const post of page) { const id = String(post?.id ?? ''); if (!id || seen.has(id)) continue; @@ -123,6 +154,8 @@ export function createRedditFirehose(opts = {}) { failed, items, lagSeconds: Math.max(0, Math.floor(now() / 1000) - cursor), + skipped, + limited, }; }; } diff --git a/packages/ingest/test/reddit-firehose.test.js b/packages/ingest/test/reddit-firehose.test.js index 88a4f99..2eec637 100644 --- a/packages/ingest/test/reddit-firehose.test.js +++ b/packages/ingest/test/reddit-firehose.test.js @@ -57,7 +57,7 @@ test('every new post reaches the subreddit row it belongs to, and nothing else i return asked.length === 1 ? first : asked.length === 2 ? [first[99], ...second] : []; }; - const run = createRedditFirehose({ fetchPage, now: () => (T0 + 300) * 1000, lookbackSeconds: 300 }); + const run = createRedditFirehose({ fetchPage, now: () => (T0 + 300) * 1000, lookbackSeconds: 300, gapMs: 0 }); const res = await run(db); assert.equal(asked.length, 2, 'stopped once a page came back short'); @@ -77,3 +77,39 @@ test('every new post reaches the subreddit row it belongs to, and nothing else i const [row] = (await db.execute({ sql: 'select last_success_at from feeds where id = ?', args: [rss] })).rows; assert.ok(row.last_success_at, 'the row is marked read'); }); + +test('a "slow down" keeps what the pass read and resumes from the cursor', async () => { + const rss = await seedSub('rss'); + const full = Array.from({ length: 100 }, (_, i) => post(`c${i}`, 'rss', T0 + i)); + const asked = []; + const fetchPage = async (after) => { + asked.push(after); + if (asked.length === 2) throw new Error('upstream-422'); + return asked.length === 1 ? full : []; + }; + const run = createRedditFirehose({ fetchPage, now: () => (T0 + 300) * 1000, lookbackSeconds: 300, gapMs: 0 }); + + const res = await run(db); + assert.equal(res.limited, 'upstream-422'); + assert.equal(res.matched, 1, 'the page read before the refusal is still stored'); + const { rows } = await db.execute({ sql: 'select count(*) as n from feed_items where feed_id = ?', args: [rss] }); + assert.equal(Number(rows[0].n), 50); + + await run(db); + assert.equal(asked[2], T0 + 98, 'the next pass carries on where this one stopped'); +}); + +test('more than the lag limit behind, it skips forward so feeds stay inside the hour', async () => { + let clock = T0 * 1000; + const asked = []; + const run = createRedditFirehose({ + fetchPage: async (after) => (asked.push(after), []), + now: () => clock, + lookbackSeconds: 60, + gapMs: 0, + }); + clock += 2 * 3_600_000; + const res = await run(db); + assert.equal(asked[0], T0 + 7200 - 300, 'resumes five minutes back'); + assert.ok(res.skipped > 3_000); +}); diff --git a/packages/social/src/collect.js b/packages/social/src/collect.js index 769827f..108901c 100644 --- a/packages/social/src/collect.js +++ b/packages/social/src/collect.js @@ -63,9 +63,11 @@ export const FLOOR_MINUTES = { x: 5, instagram: 30, facebook: 60, - // A public archive rather than a personal session, but one shared mirror - // serving fifty thousand subreddits; the hourly floor Reddit's RSS had. - reddit: 60, + // A week. Every new post on Reddit already arrives through the firehose + // (ingest/src/redditFirehose.js), which keeps all of /r/* current; asking a + // subreddit by name is now only the first read and a weekly check, and it + // spends the same per-IP budget at Arctic Shift that the firehose needs. + reddit: 7 * 24 * 60, }; /** The default for anything fetched rather than collected. */ diff --git a/packages/social/test/facebook-scrape.test.js b/packages/social/test/facebook-scrape.test.js index c2ac76c..9e74dd3 100644 --- a/packages/social/test/facebook-scrape.test.js +++ b/packages/social/test/facebook-scrape.test.js @@ -197,8 +197,9 @@ test('each platform is asked no faster than it tolerates', () => { assert.equal(floorMinutesFor({ social_network: 'instagram' }), 30); assert.equal(floorMinutesFor({ social_network: 'facebook' }), 60); - // Reddit, collected through a shared public mirror, keeps the hour its RSS had. - assert.equal(floorMinutesFor({ social_network: 'reddit' }), 60); + // Reddit's new posts arrive through the firehose; asking a subreddit by name + // is a first read and a weekly check. + assert.equal(floorMinutesFor({ social_network: 'reddit' }), 7 * 24 * 60); assert.equal(floorMinutesFor({}), 60); // Nothing collected may be asked faster than X.