From cbaca333b384ac954f8200a726a4deab3df39dc9 Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Mon, 9 Nov 2026 19:30:14 +0000 Subject: [PATCH] reddit: /r/ URLs refresh hourly; unblock the reddit crawl r/rss had not been read since 2026-10-23 and only ~7 of 50k subreddits were read per hour. Two bugs plus a freshness rule: - One subreddit returning an empty listing (private/banned) was treated as a host-wide 429 and held back ~200 reddit feeds for 20 min, every few minutes. empty-result is now hostWide:false: that feed is throttled, the host queue keeps walking. - NUL bytes in reddit text failed the Postgres write ("invalid byte sequence for encoding UTF8: 0x00") on every crawl. Stripped at parse. - Any /r/, /r/u/ page or .rss/.json read of a copy older than an hour puts the row at the front of its tier (requestRefreshIfStale), so anything someone reads or polls is at most ~1h + one tick stale. Co-Authored-By: Claude Opus 5.5 --- apps/web/src/app/r/[subreddit]/page.jsx | 5 +++++ apps/web/src/app/r/u/[username]/page.jsx | 3 +++ apps/web/src/lib/socialRiver.js | 17 ++++++++++++++++ packages/db/src/queries.js | 23 ++++++++++++++++++++++ packages/db/test/due-tiers.test.js | 17 ++++++++++++++++ packages/ingest/src/crawl.js | 5 ++++- packages/ingest/test/host-throttle.test.js | 17 ++++++++++++++++ packages/social/src/reddit/arctic.js | 6 +++++- packages/social/src/reddit/fetch.js | 6 +++++- packages/social/test/reddit-arctic.test.js | 9 +++++++++ 10 files changed, 105 insertions(+), 3 deletions(-) diff --git a/apps/web/src/app/r/[subreddit]/page.jsx b/apps/web/src/app/r/[subreddit]/page.jsx index 2fc6dab..38864ca 100644 --- a/apps/web/src/app/r/[subreddit]/page.jsx +++ b/apps/web/src/app/r/[subreddit]/page.jsx @@ -3,6 +3,8 @@ 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 FeedPage from '../../[slug]/page.jsx'; import { when } from '../RedditComments.jsx'; @@ -62,6 +64,9 @@ 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 db04422..cf85faf 100644 --- a/apps/web/src/app/r/u/[username]/page.jsx +++ b/apps/web/src/app/r/u/[username]/page.jsx @@ -2,6 +2,8 @@ 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 FeedPage from '../../../[slug]/page.jsx'; @@ -51,5 +53,6 @@ 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 f6f6070..284f6f7 100644 --- a/apps/web/src/lib/socialRiver.js +++ b/apps/web/src/lib/socialRiver.js @@ -89,6 +89,8 @@ export async function socialRiver({ ); } + refreshIfStale(client, feed); + const page = `${siteUrl()}${canonical}`; const rows = await q.itemsForFeed(client, String(feed.id), riverLimit(rawLimit)); @@ -119,6 +121,21 @@ 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 38ebcb4..506d138 100644 --- a/packages/db/src/queries.js +++ b/packages/db/src/queries.js @@ -3035,6 +3035,29 @@ export async function requestRefresh(db, id) { }); } +/** + * `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 4dcc4d2..6351199 100644 --- a/packages/db/test/due-tiers.test.js +++ b/packages/db/test/due-tiers.test.js @@ -71,3 +71,20 @@ 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/crawl.js b/packages/ingest/src/crawl.js index adfa82d..4def9bc 100644 --- a/packages/ingest/src/crawl.js +++ b/packages/ingest/src/crawl.js @@ -347,6 +347,9 @@ export async function crawlFeed(db, feed, opts = {}) { ok: false, newItems: 0, throttled: true, + // False when the collector says the refusal is about this one source, + // not its host; crawlDue then throttles the feed and keeps walking. + hostWide: resolved.hostWide !== false, retryAfter: resolved.retryAfter ?? null, error: resolved.error, }; @@ -784,7 +787,7 @@ export async function crawlDue(db, batchSize = 25, concurrency = 8, onEvent = nu // the same question another eight hundred times is precisely what it // just told us to stop doing — see `markHostThrottled` for the // measurement that put this here. - if (res.throttled) { + if (res.throttled && res.hostWide !== false) { throttled += 1; const rest = queue.slice(position + 1); if (rest.length > 0) { diff --git a/packages/ingest/test/host-throttle.test.js b/packages/ingest/test/host-throttle.test.js index 90aab8a..79f9167 100644 --- a/packages/ingest/test/host-throttle.test.js +++ b/packages/ingest/test/host-throttle.test.js @@ -181,3 +181,20 @@ test('one throttled host does not stop the others in the same batch', async () = ); assert.equal(Number(fine.rows[0].n), 5, 'every feed on the healthy host was crawled'); }); + +test('a throttle about one source (hostWide: false) does not hold back the host', async () => { + // Reddit's empty-result: one subreddit that had posts and now returns none + // (private, banned). Treating it as the mirror refusing us parked ~200 other + // subreddits every few minutes and capped Reddit at ~7 reads an hour. + await seedHost('crowded.example', 10); + + let attempts = 0; + const resolve = async () => { + attempts += 1; + return { ok: false, error: 'empty-result', throttled: true, hostWide: false, retryAfter: 1200 }; + }; + + const result = await crawlDue(db, 10, 4, null, { crawl: { resolve } }); + assert.equal(attempts, 10, 'every feed on the host is still asked'); + assert.equal(result.throttled, 0, 'no host was held back'); +}); diff --git a/packages/social/src/reddit/arctic.js b/packages/social/src/reddit/arctic.js index b9ced80..fadf658 100644 --- a/packages/social/src/reddit/arctic.js +++ b/packages/social/src/reddit/arctic.js @@ -66,7 +66,11 @@ export async function arcticGet(path, params, opts = {}) { let json; try { - json = JSON.parse(body); + // Strip NUL from every string on the way in. Reddit text occasionally carries + // one (JSON can only spell it `\u0000`), and Postgres refuses it in a text + // column with "invalid byte sequence for encoding UTF8: 0x00" -- which failed + // the whole subreddit's write on every crawl, for ever, the same few each hour. + json = JSON.parse(body, (_key, value) => (typeof value === 'string' ? value.replace(/\u0000/g, '') : value)); } catch { throw new XUnavailable('arctic-shift: unparseable-response'); } diff --git a/packages/social/src/reddit/fetch.js b/packages/social/src/reddit/fetch.js index e116e22..a37d5ed 100644 --- a/packages/social/src/reddit/fetch.js +++ b/packages/social/src/reddit/fetch.js @@ -41,7 +41,11 @@ export async function fetchRedditSource(feed, opts = {}) { // none is the mirror having a bad minute, not a community going silent. if (items.length === 0 && Number(feed?.item_count ?? 0) > 0) { onEvent('reddit.fetch.failed', { ref: feed.social_ref, error: 'empty-result' }); - return { ok: false, throttled: true, retryAfter: 20 * 60, error: 'empty-result' }; + // `hostWide: false`: this is one subreddit's answer (gone private, banned, + // or the archive's bad minute for that one query), not the mirror refusing + // us. Holding back the whole reddit.com queue for it parked ~200 other + // subreddits every few minutes and capped Reddit at ~7 reads an hour. + return { ok: false, throttled: true, hostWide: false, retryAfter: 20 * 60, error: 'empty-result' }; } onEvent('reddit.fetch.success', { ref: feed.social_ref, itemCount: items.length }); diff --git a/packages/social/test/reddit-arctic.test.js b/packages/social/test/reddit-arctic.test.js index 5927e5f..de6b5a8 100644 --- a/packages/social/test/reddit-arctic.test.js +++ b/packages/social/test/reddit-arctic.test.js @@ -150,6 +150,14 @@ test('the collector reads a subreddit from the mirror, by name, with no runtime assert.equal(asked.searchParams.get('sort'), 'desc'); }); +test('NUL characters never reach the database, which refuses them', async () => { + const fetch = fakeFetch({ '/api/posts/search': { data: [{ ...POST, title: 'a\u0000b', selftext: 'x\u0000y' }] } }); + const result = await fetchRedditSource({ social_ref: 'r:sub:asktechnology', feed_url: 'x' }, { runtime: { fetch } }); + assert.equal(result.ok, true); + assert.equal(result.feed.items[0].title, 'ab'); + assert.ok(!JSON.stringify(result.feed).includes('\\u0000')); +}); + test('a user source asks by author', async () => { const fetch = fakeFetch({ '/api/posts/search': { data: [POST] } }); const result = await fetchRedditSource({ social_ref: 'r:user:rosybodynimbus', feed_url: 'x' }, { runtime: { fetch } }); @@ -170,6 +178,7 @@ test('a mirror outage or an empty answer for a source that had posts never count { runtime: { fetch: empty } }, ); assert.equal(quiet.throttled, true); + assert.equal(quiet.hostWide, false, 'one quiet subreddit does not hold back the rest of Reddit'); const errored = fakeFetch({ '/api/posts/search': { error: 'Timeout. Maybe slow down a bit' } }); const slow = await fetchRedditSource({ social_ref: 'r:sub:asktechnology', feed_url: 'x' }, { runtime: { fetch: errored } });