From 7780d8cd95a4b8887558d73fefe2d7760a8bce8d Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Mon, 9 Nov 2026 20:59:05 +0000 Subject: [PATCH] reddit firehose: resume from stored posts; measure lag against the mirror Arctic Shift's own ingestion fell 37 min behind Reddit (2026-11-09), and a restart started the cursor at now-15 min, ahead of the mirror: every page came back empty and posts between the mirror's head and the cursor would never have been read. - first pass resumes from the newest reddit post stored (feeds last_published_at) minus 2 min, not from the clock - the 50-min skip jumps only to the mirror's newest post minus 5 min (one sort=desc limit=1 ask), so it never passes un-ingested posts Co-Authored-By: Claude Opus 5.5 --- packages/db/src/queries.js | 16 +++++++++ packages/ingest/src/redditFirehose.js | 36 +++++++++++++++++--- packages/ingest/test/reddit-firehose.test.js | 25 +++++++++++++- packages/social/index.js | 1 + packages/social/src/reddit/arctic.js | 12 +++++++ 5 files changed, 84 insertions(+), 6 deletions(-) diff --git a/packages/db/src/queries.js b/packages/db/src/queries.js index 709ab9f..50ee98e 100644 --- a/packages/db/src/queries.js +++ b/packages/db/src/queries.js @@ -3035,6 +3035,22 @@ export async function requestRefresh(db, id) { }); } +/** + * When the newest reddit post we hold was published: where the Reddit firehose + * resumes after a restart. One aggregate over the ~50k reddit rows. + * + * @param {Client} db + * @returns {Promise} + */ +export async function newestRedditPost(db) { + const { rows } = await db.execute({ + sql: `select max(last_published_at) as at from feeds + where social_network = 'reddit' and last_published_at <= ?`, + args: [nowIso()], + }); + return rows[0]?.at ? String(rows[0].at) : null; +} + /** * Live rows for a set of social refs (`r:sub:rss`, `r:user:spez`), with the * columns a crawl needs: the Reddit firehose matches every new post on the site diff --git a/packages/ingest/src/redditFirehose.js b/packages/ingest/src/redditFirehose.js index 292c0d1..360dfac 100644 --- a/packages/ingest/src/redditFirehose.js +++ b/packages/ingest/src/redditFirehose.js @@ -16,7 +16,7 @@ */ import { q } from '@rssamplifier/db'; -import { fetchRedditFirehosePage, redditPostToItem, redditResult } from '@rssamplifier/social'; +import { fetchRedditFirehosePage, fetchRedditNewest, redditPostToItem, redditResult } from '@rssamplifier/social'; import { crawlFeed } from './crawl.js'; @@ -28,6 +28,7 @@ const PAGE = 100; * maxPages?: number, * gapMs?: number, * fetchBudgetMs?: number, + * fetchHead?: () => Promise, * sleep?: (ms: number) => Promise, * maxLagSeconds?: number, * resumeLagSeconds?: number, @@ -59,8 +60,23 @@ export function createRedditFirehose(opts = {}) { 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); + // The newest post the mirror holds. Arctic Shift itself can fall behind + // Reddit (37 minutes on 2026-11-09), so "behind" is measured against this, + // never the wall clock: skipping to now - 5 min while the mirror is 37 + // minutes back would jump over posts it has not ingested yet. + const fetchHead = opts.fetchHead ?? (async () => { + const [newest] = await fetchRedditNewest({ env: opts.env }); + return Number(newest?.created_utc) || null; + }); + + /** + * Unix seconds; posts created after this are still to be read. Null until + * the first pass, which starts where the last process stopped: the newest + * reddit post already stored, so a restart neither skips nor re-reads an + * hour. `lookbackSeconds` (tests) starts from the clock instead. + * @type {number|null} + */ + let cursor = opts.lookbackSeconds != null ? Math.floor(now() / 1000) - opts.lookbackSeconds : null; /** ids already handled near the cursor, since pages overlap by a second */ let seen = new Set(); @@ -76,9 +92,19 @@ export function createRedditFirehose(opts = {}) { let limited = null; const nowSeconds = Math.floor(now() / 1000); + if (cursor == null) { + const stored = await q.newestRedditPost(db).catch(() => null); + const from = stored ? Math.floor(Date.parse(stored) / 1000) - 120 : nowSeconds - 900; + cursor = Math.max(from, nowSeconds - maxLagSeconds); + } + if (nowSeconds - cursor > maxLagSeconds) { - skipped = nowSeconds - resumeLagSeconds - cursor; - cursor = nowSeconds - resumeLagSeconds; + const head = await fetchHead().catch(() => null); + const target = head ? head - resumeLagSeconds : null; + if (target && target > cursor) { + skipped = target - cursor; + cursor = target; + } } const started = now(); diff --git a/packages/ingest/test/reddit-firehose.test.js b/packages/ingest/test/reddit-firehose.test.js index 2eec637..2b3d59a 100644 --- a/packages/ingest/test/reddit-firehose.test.js +++ b/packages/ingest/test/reddit-firehose.test.js @@ -101,15 +101,38 @@ test('a "slow down" keeps what the pass read and resumes from the cursor', async test('more than the lag limit behind, it skips forward so feeds stay inside the hour', async () => { let clock = T0 * 1000; + let head = T0 + 7200; const asked = []; const run = createRedditFirehose({ fetchPage: async (after) => (asked.push(after), []), + fetchHead: async () => head, 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.equal(asked[0], T0 + 7200 - 300, 'resumes five minutes behind the mirror'); assert.ok(res.skipped > 3_000); + + // The mirror itself 2 hours behind: nothing to skip to, so nothing is skipped. + clock += 2 * 3_600_000; + head = T0 + 7200 - 100; + const held = await run(db); + assert.equal(held.skipped, 0); + assert.equal(asked[1], T0 + 7200 - 300, 'the cursor waits for the mirror'); +}); + +test('a restart resumes from the newest reddit post already stored', async () => { + const id = await seedSub('rss'); + const stored = new Date((T0 - 600) * 1000).toISOString(); + await db.execute({ sql: 'update feeds set last_published_at = ? where id = ?', args: [stored, id] }); + const asked = []; + const run = createRedditFirehose({ + fetchPage: async (after) => (asked.push(after), []), + now: () => T0 * 1000, + gapMs: 0, + }); + await run(db); + assert.equal(asked[0], T0 - 600 - 120); }); diff --git a/packages/social/index.js b/packages/social/index.js index a297a7e..d0f12c3 100644 --- a/packages/social/index.js +++ b/packages/social/index.js @@ -98,6 +98,7 @@ export { commentTree, fetchRedditComments, fetchRedditFirehosePage, + fetchRedditNewest, fetchRedditListing, fetchRedditPost, isShowable as isShowableRedditPost, diff --git a/packages/social/src/reddit/arctic.js b/packages/social/src/reddit/arctic.js index a2f1139..111c2ec 100644 --- a/packages/social/src/reddit/arctic.js +++ b/packages/social/src/reddit/arctic.js @@ -113,6 +113,18 @@ export async function fetchRedditFirehosePage(after, opts = {}) { return Array.isArray(data) ? data : []; } +/** + * The newest post the mirror holds, anywhere on Reddit: where its ingestion + * has got to, which can trail Reddit by most of an hour. + * + * @param {Parameters[2]} [opts] + * @returns {Promise} + */ +export async function fetchRedditNewest(opts = {}) { + const data = await arcticGet('/api/posts/search', { sort: 'desc', limit: 1 }, opts); + return Array.isArray(data) ? data : []; +} + /** * One post by id, or null when the archive has never seen it. *