reddit firehose: resume from stored posts; measure lag against the mirror

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<string|null>}
+ */
+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

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<number|null>,
  *   sleep?: (ms: number) => Promise<unknown>,
  *   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();

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

packages/social/index.js

@@ -98,6 +98,7 @@ export {
   commentTree,
   fetchRedditComments,
   fetchRedditFirehosePage,
+  fetchRedditNewest,
   fetchRedditListing,
   fetchRedditPost,
   isShowable as isShowableRedditPost,

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<typeof arcticGet>[2]} [opts]
+ * @returns {Promise<object[]>}
+ */
+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.
  *