reddit firehose: pace pages, survive 'slow down', never lag past the hour

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) {

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;

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

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 });
 }

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.
  *

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.

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');
-});

packages/ingest/src/redditFirehose.js

@@ -26,6 +26,10 @@ const PAGE = 100;
  * @param {{
  *   lookbackSeconds?: number,
  *   maxPages?: number,
+ *   gapMs?: number,
+ *   sleep?: (ms: number) => Promise<unknown>,
+ *   maxLagSeconds?: number,
+ *   resumeLagSeconds?: number,
  *   concurrency?: number,
  *   env?: Record<string, string|undefined>,
  *   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,
     };
   };
 }

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);
+});

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. */

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.