From 6324caf11fab2a8e0cd85e575cbd0a93ee3f20d1 Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Mon, 9 Nov 2026 19:54:03 +0000 Subject: [PATCH] reddit: firehose keeps every /r/* feed current; fix nested NUL placeholders All ~50k subreddits hourly by asking each one is ~14 req/s to a volunteer mirror. Instead a poller job (reddit-firehose, every 2 min) pages every new post on Reddit from Arctic Shift (~25 posts/s = ~15 req/min, ~1 min behind Reddit), matches r:sub: and r:user: rows, and runs each through crawlFeed with the posts in hand (dedupe/FTS/keywords/schedule as a normal crawl). REDDIT_FIREHOSE=0 switches it off; REDDIT_FIREHOSE_MS sets the pace. The NUL write failures were our own: redditMarkdown's \u0000N\u0000 placeholders nest (code span inside a link label) and a single restore pass leaked the inner one. Restore until none remain, then strip. Co-Authored-By: Claude Opus 5.5 --- apps/poller/src/daemon.js | 2 + apps/poller/src/index.js | 18 +++ packages/db/src/queries.js | 24 ++++ packages/ingest/index.js | 1 + packages/ingest/src/redditFirehose.js | 128 +++++++++++++++++++ packages/ingest/test/reddit-firehose.test.js | 79 ++++++++++++ packages/social/index.js | 3 +- packages/social/src/reddit/arctic.js | 27 +++- packages/social/src/reddit/fetch.js | 59 +++++---- packages/social/test/reddit-arctic.test.js | 7 + 10 files changed, 323 insertions(+), 25 deletions(-) create mode 100644 packages/ingest/src/redditFirehose.js create mode 100644 packages/ingest/test/reddit-firehose.test.js diff --git a/apps/poller/src/daemon.js b/apps/poller/src/daemon.js index 958f3a1..56931fc 100644 --- a/apps/poller/src/daemon.js +++ b/apps/poller/src/daemon.js @@ -84,6 +84,8 @@ export function buildPollerDaemon({ createDaemon = defaultCreateDaemon, store, r add('crawl', c.intervalMs, run.crawl, { staleAfter: CRAWL_STALE_AFTER }); 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); // 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 24e9abf..d79411f 100644 --- a/apps/poller/src/index.js +++ b/apps/poller/src/index.js @@ -24,6 +24,7 @@ import { drainImport, seedTopRings, verifyRingMembers, + createRedditFirehose, } from '@rssamplifier/ingest'; import { runDueSources, discoverFromOwnTopics } from '@rssamplifier/discover'; import { findFeedCard } from '@rssamplifier/feed'; @@ -525,6 +526,20 @@ async function topicsTick() { } // Hourly, and a single small read on the 23 hours with no finished day to add. +// Every new post on Reddit, matched to the subreddits we hold: what keeps all +// 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(); + +async function redditFirehoseTick() { + if (stopping) return; + const started = Date.now(); + const res = await redditFirehose(db); + log('reddit-firehose', { ...res, ms: Date.now() - started }); +} + async function trendsTick() { if (stopping) return; const started = Date.now(); @@ -1243,8 +1258,11 @@ const daemon = buildPollerDaemon({ sourceIntervalMs, topicSearchIntervalMs, topicsIntervalMs, + redditFirehose: redditFirehoseEnabled, + redditFirehoseMs, }, run: { + redditFirehose: redditFirehoseTick, crawl: tick, import: importTick, purge: purgeTick, diff --git a/packages/db/src/queries.js b/packages/db/src/queries.js index 506d138..3e43330 100644 --- a/packages/db/src/queries.js +++ b/packages/db/src/queries.js @@ -3035,6 +3035,30 @@ export async function requestRefresh(db, id) { }); } +/** + * 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 + * against these and crawls only the rows it found. + * + * @param {Client} db + * @param {string[]} refs + * @returns {Promise} + */ +export async function feedsBySocialRefs(db, refs) { + const out = []; + for (let i = 0; i < refs.length; i += 500) { + const chunk = refs.slice(i, i + 500); + if (chunk.length === 0) continue; + const { rows } = await db.execute({ + sql: `select ${DUE_COLUMNS} from feeds + where social_ref in (${chunk.map(() => '?').join(',')}) and status <> 'dead'`, + args: chunk, + }); + out.push(...rows); + } + 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 diff --git a/packages/ingest/index.js b/packages/ingest/index.js index 5e4b83c..df727c6 100644 --- a/packages/ingest/index.js +++ b/packages/ingest/index.js @@ -8,6 +8,7 @@ export { refreshFeedKeywords, } from './src/crawl.js'; export { importFeeds, importOpml } from './src/import.js'; +export { createRedditFirehose } from './src/redditFirehose.js'; export { queueFeeds } from './src/queue.js'; export { drainImport } from './src/drain.js'; export { diff --git a/packages/ingest/src/redditFirehose.js b/packages/ingest/src/redditFirehose.js new file mode 100644 index 0000000..c155709 --- /dev/null +++ b/packages/ingest/src/redditFirehose.js @@ -0,0 +1,128 @@ +/** + * Every subreddit in the directory, current within minutes, from one stream. + * + * Asking fifty thousand subreddits one at a time every hour is ~14 requests a + * second, around the clock, to a volunteer-run mirror, and the per-host crawler + * managed ~7 an hour. Reddit as a whole produces ~25 posts a second, and Arctic + * Shift will page through all of them, oldest first, a hundred at a time — so + * reading *every* new post on Reddit is ~15 requests a minute. Each post is + * matched to the rows we hold (`r:sub:` and `r:user:`) and + * every matched row goes through `crawlFeed` with its posts already in hand, + * so dedupe, keywords, FTS, alerts and scheduling are the crawl's own. + * + * The cursor lives in memory. A restart starts `lookbackSeconds` back, which is + * safe because items dedupe by guid; the per-source crawl and the on-view + * refresh cover anything older. + */ + +import { q } from '@rssamplifier/db'; +import { fetchRedditFirehosePage, redditPostToItem, redditResult } from '@rssamplifier/social'; + +import { crawlFeed } from './crawl.js'; + +const PAGE = 100; + +/** + * @param {{ + * lookbackSeconds?: number, + * maxPages?: number, + * concurrency?: number, + * env?: Record, + * fetchPage?: typeof fetchRedditFirehosePage, + * crawl?: typeof crawlFeed, + * now?: () => number, + * }} [opts] + */ +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 concurrency = opts.concurrency ?? 6; + + /** unix seconds; posts created after this are still to be read */ + let cursor = Math.floor(now() / 1000) - (opts.lookbackSeconds ?? 900); + /** ids already handled near the cursor, since pages overlap by a second */ + let seen = new Set(); + + /** + * One pass: read pages until caught up or `maxPages`, then store. + * + * @param {import('@rssamplifier/db/src/pg.js').PgClient} db + */ + return async function runRedditFirehose(db) { + const posts = []; + let pages = 0; + for (; pages < maxPages; pages += 1) { + const page = await fetchPage(cursor, { env: opts.env }); + for (const post of page) { + const id = String(post?.id ?? ''); + if (!id || seen.has(id)) continue; + seen.add(id); + posts.push(post); + } + if (page.length === 0) break; + + const last = Number(page[page.length - 1]?.created_utc) || cursor; + // Step back one second so a second split across two pages is read whole; + // `seen` drops the overlap. A page that is all one second (100+ posts in + // it) must move on, or the loop would ask the same page for ever. + cursor = last - 1 > cursor ? last - 1 : last; + if (page.length < PAGE) break; + } + if (seen.size > 20_000) seen = new Set([...seen].slice(-5_000)); + + // Group by the refs a row can have. + const byRef = new Map(); + const push = (ref, post) => { + const list = byRef.get(ref); + if (list) list.push(post); + else byRef.set(ref, [post]); + }; + for (const post of posts) { + if (post?.subreddit) push(`r:sub:${String(post.subreddit).toLowerCase()}`, post); + const author = String(post?.author ?? ''); + if (author && author !== '[deleted]') push(`r:user:${author.toLowerCase()}`, post); + } + + const rows = byRef.size ? await q.feedsBySocialRefs(db, [...byRef.keys()]) : []; + + let crawled = 0; + let failed = 0; + let items = 0; + let next = 0; + const worker = async () => { + while (next < rows.length) { + const row = rows[next++]; + const ref = String(row.social_ref); + const matched = (byRef.get(ref) ?? []).slice().sort((a, b) => b.created_utc - a.created_utc); + const mapped = matched.map(redditPostToItem).filter(Boolean); + if (mapped.length === 0) continue; + const [, mode, name] = /^r:(sub|user):(.+)$/.exec(ref) ?? []; + if (!mode) continue; + try { + const res = await crawl(db, row, { + x: async () => redditResult(row, { mode: /** @type {'sub'|'user'} */ (mode), name }, matched, mapped), + }); + if (res?.ok) { + crawled += 1; + items += Number(res.newItems ?? 0); + } else failed += 1; + } catch { + failed += 1; + } + } + }; + await Promise.all(Array.from({ length: Math.max(1, Math.min(concurrency, rows.length)) }, worker)); + + return { + pages, + posts: posts.length, + matched: rows.length, + crawled, + failed, + items, + lagSeconds: Math.max(0, Math.floor(now() / 1000) - cursor), + }; + }; +} diff --git a/packages/ingest/test/reddit-firehose.test.js b/packages/ingest/test/reddit-firehose.test.js new file mode 100644 index 0000000..88a4f99 --- /dev/null +++ b/packages/ingest/test/reddit-firehose.test.js @@ -0,0 +1,79 @@ +import assert from 'node:assert/strict'; +import { test, beforeEach, after } from 'node:test'; +import { q, social } from '@rssamplifier/db'; +import { connectTest } from '@rssamplifier/db/src/testdb.js'; +import { redditSource } from '@rssamplifier/social'; + +import { createRedditFirehose } from '../src/redditFirehose.js'; + +let db; + +beforeEach(async () => { + db?.close(); + db = await connectTest(); +}); + +after(async () => { + db?.close(); +}); + +async function seedSub(name) { + const source = redditSource(`r/${name}`); + const stored = await social.upsertSocialSource(db, { + network: 'reddit', + ref: source.ref, + slug: source.slug, + title: source.title, + feedUrl: source.feedUrl, + siteUrl: source.feedUrl, + }); + return String(stored.id); +} + +const T0 = 1_794_000_000; +const post = (id, subreddit, at) => ({ + id, + title: `post ${id}`, + author: 'someone', + subreddit, + permalink: `/r/${subreddit}/comments/${id}/x/`, + created_utc: at, + is_self: true, + selftext: 'hello', + num_comments: 0, + removed_by_category: null, +}); + +test('every new post reaches the subreddit row it belongs to, and nothing else is stored', async () => { + const rss = await seedSub('rss'); + const linux = await seedSub('linux'); + + // Two pages: a full one (forces a second read) then a short one (caught up). + const first = Array.from({ length: 100 }, (_, i) => post(`a${i}`, i % 2 ? 'RSS' : 'notours', T0 + i)); + const second = [post('b1', 'linux', T0 + 200), post('b2', 'rss', T0 + 201)]; + const asked = []; + const fetchPage = async (after) => { + asked.push(after); + return asked.length === 1 ? first : asked.length === 2 ? [first[99], ...second] : []; + }; + + const run = createRedditFirehose({ fetchPage, now: () => (T0 + 300) * 1000, lookbackSeconds: 300 }); + const res = await run(db); + + assert.equal(asked.length, 2, 'stopped once a page came back short'); + assert.equal(asked[1], T0 + 98, 'the next page starts a second back so a split second is read whole'); + assert.equal(res.posts, 102, 'the overlapping post is not counted twice'); + assert.equal(res.matched, 2); + + const { rows: counted } = await db.execute({ sql: 'select guid from feed_items where feed_id = ?', args: [rss] }); + const guids = counted.map((r) => String(r.guid)); + // 51 matched (r/RSS case-insensitively); ITEMS_PER_CRAWL keeps the newest 50, + // so only the oldest is dropped. Two-minute passes never meet that cap. + assert.equal(guids.length, 50); + assert.ok(guids.includes('t3_b2') && !guids.includes('t3_a1')); + const linuxItems = await q.itemsForFeed(db, linux, 100); + assert.deepEqual(linuxItems.map((r) => r.guid), ['t3_b1']); + + 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'); +}); diff --git a/packages/social/index.js b/packages/social/index.js index a24ae7a..a297a7e 100644 --- a/packages/social/index.js +++ b/packages/social/index.js @@ -97,6 +97,7 @@ export { arcticGet, commentTree, fetchRedditComments, + fetchRedditFirehosePage, fetchRedditListing, fetchRedditPost, isShowable as isShowableRedditPost, @@ -105,4 +106,4 @@ export { redditPostId, } from './src/reddit/arctic.js'; -export { fetchRedditSource } from './src/reddit/fetch.js'; +export { fetchRedditSource, redditResult } from './src/reddit/fetch.js'; diff --git a/packages/social/src/reddit/arctic.js b/packages/social/src/reddit/arctic.js index fadf658..a2f1139 100644 --- a/packages/social/src/reddit/arctic.js +++ b/packages/social/src/reddit/arctic.js @@ -97,6 +97,22 @@ export async function fetchRedditListing(spec, opts = {}) { return Array.isArray(data) ? data : []; } +/** + * One page of every new post on Reddit, oldest first, created after `after` + * (unix seconds). The firehose: about 25 posts a second across the whole site, + * so a hundred-post page per ~4 seconds of Reddit keeps every subreddit we hold + * current for ~15 requests a minute, against ~14 a second to ask fifty + * thousand subreddits one by one every hour. + * + * @param {number} after unix seconds + * @param {Parameters[2]} [opts] + * @returns {Promise} + */ +export async function fetchRedditFirehosePage(after, opts = {}) { + const data = await arcticGet('/api/posts/search', { after: Math.floor(after), sort: 'asc', limit: 100 }, opts); + return Array.isArray(data) ? data : []; +} + /** * One post by id, or null when the archive has never seen it. * @@ -360,7 +376,16 @@ function inline(raw) { .replace(/(^|[^*])\*([^*\s][^*]*)\*/g, '$1$2') .replace(/~~([^~]+)~~/g, '$1'); - return s.replace(/\u0000(\d+)\u0000/g, (_, n) => tokens[Number(n)]); + // Tokens nest (a code span inside a link label is kept inside the link's + // token), so restore until none are left. One pass leaked the inner + // placeholder, NULs and all, and Postgres refused the whole subreddit's write + // ("invalid byte sequence for encoding UTF8: 0x00") on every crawl. + const PLACEHOLDER = /\u0000(\d+)\u0000/g; + for (let pass = 0; pass < 8 && PLACEHOLDER.test(s); pass += 1) { + PLACEHOLDER.lastIndex = 0; + s = s.replace(PLACEHOLDER, (_, n) => tokens[Number(n)] ?? ''); + } + return s.replace(/\u0000/g, ''); } function escapeHtml(s) { diff --git a/packages/social/src/reddit/fetch.js b/packages/social/src/reddit/fetch.js index a37d5ed..1af0684 100644 --- a/packages/social/src/reddit/fetch.js +++ b/packages/social/src/reddit/fetch.js @@ -49,31 +49,44 @@ export async function fetchRedditSource(feed, opts = {}) { } onEvent('reddit.fetch.success', { ref: feed.social_ref, itemCount: items.length }); - - // Reddit's casing for the name, when the archive has it, so a row created - // from a lowercased ref still titles itself `r/AskTechnology`. - const named = posts[0]?.[spec.mode === 'user' ? 'author' : 'subreddit']; - const display = { ...spec, name: named && String(named).toLowerCase() === spec.name.toLowerCase() ? String(named) : spec.name }; - - return { - ok: true, - feedUrl: feed.feed_url, - feed: { - title: redditTitle(display), - description: - spec.mode === 'user' - ? `Posts by u/${display.name} on Reddit, mirrored by RSSAmp.` - : `Posts from r/${display.name}, mirrored by RSSAmp.`, - siteUrl: redditSiteUrl(display), - language: null, - imageUrl: null, - categories: [], - kind: 'blog', - items, - }, - }; + return redditResult(feed, spec, posts, items); } catch (error) { onEvent('reddit.fetch.failed', { ref: feed.social_ref, error: String(error?.message ?? error) }); return failureResult(error); } } + +/** + * The `{ok, feedUrl, feed}` a crawl stores, from posts already in hand. Shared + * by the per-source collector and the firehose, which reads every new post on + * Reddit and hands each row its own. + * + * @param {{ feed_url?: string }} feed + * @param {{ mode: 'sub'|'user', name: string }} spec + * @param {object[]} posts raw Arctic Shift posts + * @param {object[]} [items] posts already mapped through postToItem + */ +export function redditResult(feed, spec, posts, items = posts.map(postToItem).filter(Boolean)) { + // Reddit's casing for the name, when the archive has it, so a row created + // from a lowercased ref still titles itself `r/AskTechnology`. + const named = posts[0]?.[spec.mode === 'user' ? 'author' : 'subreddit']; + const display = { ...spec, name: named && String(named).toLowerCase() === spec.name.toLowerCase() ? String(named) : spec.name }; + + return { + ok: true, + feedUrl: feed.feed_url, + feed: { + title: redditTitle(display), + description: + spec.mode === 'user' + ? `Posts by u/${display.name} on Reddit, mirrored by RSSAmp.` + : `Posts from r/${display.name}, mirrored by RSSAmp.`, + siteUrl: redditSiteUrl(display), + language: null, + imageUrl: null, + categories: [], + kind: 'blog', + items, + }, + }; +} diff --git a/packages/social/test/reddit-arctic.test.js b/packages/social/test/reddit-arctic.test.js index de6b5a8..71054be 100644 --- a/packages/social/test/reddit-arctic.test.js +++ b/packages/social/test/reddit-arctic.test.js @@ -156,6 +156,13 @@ test('NUL characters never reach the database, which refuses them', async () => assert.equal(result.ok, true); assert.equal(result.feed.items[0].title, 'ab'); assert.ok(!JSON.stringify(result.feed).includes('\\u0000')); + + // Our own markdown placeholders nest: a code span inside a link label. + const nested = fakeFetch({ '/api/posts/search': { data: [{ ...POST, selftext: 'see [the `rss` tool](https://example.com/x) and r/rss' }] } }); + const out = await fetchRedditSource({ social_ref: 'r:sub:asktechnology', feed_url: 'x' }, { runtime: { fetch: nested } }); + const html = out.feed.items[0].contentHtml; + assert.ok(!html.includes('\u0000'), html); + assert.ok(html.includes('rss'), html); }); test('a user source asks by author', async () => {