diff --git a/apps/poller/src/index.js b/apps/poller/src/index.js index 9d51d3b..988b60f 100644 --- a/apps/poller/src/index.js +++ b/apps/poller/src/index.js @@ -1,4 +1,13 @@ -import { connect, createWriteWorker, migrate, q, accounts, alerts, takeWriteTally } from '@rssamplifier/db'; +import { + connect, + createWriteWorker, + migrate, + q, + accounts, + alerts, + takeWriteTally, + warmStatsCache, +} from '@rssamplifier/db'; import { crawlDue, enrichDue, @@ -135,6 +144,16 @@ const authorIntervalMs = (Number(env['AUTHOR_INTERVAL_SECONDS']) || 60) * 1000; // minutes stale on a page that refreshes every fifteen seconds. const queueSampleMs = (Number(env['QUEUE_SAMPLE_SECONDS']) || 600) * 1000; +// How often the category breakdown is recomputed into Redis for /crawlstats. +// +// It is a ~59-second read of every feed, which is why it cannot live on a +// request: with a 30-second deadline in front of it, it could only ever fail, +// so the cache it was supposed to fill stayed empty and every visitor paid the +// full timeout. Five minutes is far more often than the number moves -- it is +// the shape of a directory of half a million feeds -- and the point is only +// that Redis is never empty, not that it is current to the second. +const statsWarmMs = (Number(env['STATS_WARM_SECONDS']) || 300) * 1000; + // Where the write queue lives. Absent, `connect()` falls back to the // in-process serialiser and this daemon starts no worker — the system as it // shipped before the queue existed, which is the right thing to degrade to. @@ -831,6 +850,34 @@ async function queueTick() { } } +/** Guards against a warm that outruns its interval starting a second one. */ +let warming = false; + +/** + * Recompute the category breakdown into Redis, so no page has to. + * + * This read takes about a minute against half a million feeds and cannot be + * made cheap: the columns it groups by are rewritten on every crawl, so an + * index covering them would be paid for on the write path, which is the one + * thing this system has none of to spare. Doing it here instead costs the + * crawler nothing -- it is a read, and reads do not queue behind the single + * writer everything else contends for. + * + * `warmStatsCache` never throws; the result is logged rather than acted on, + * because there is nothing to do about a warm that did not happen except serve + * the previous answer, which is what the cache already does. + */ +async function statsTick() { + if (warming) return; + warming = true; + try { + const result = await warmStatsCache({ log }); + if (!result.ok) return; + } finally { + warming = false; + } +} + /** * The one process that actually writes to Turso. * @@ -881,6 +928,7 @@ const cardTimer = setInterval(cardTick, cardIntervalMs); const alertTimer = setInterval(alertTick, alertIntervalMs); const enrichTimer = setInterval(enrichTick, authorIntervalMs); const queueTimer = setInterval(queueTick, queueSampleMs); +const statsTimer = setInterval(statsTick, statsWarmMs); const searchTimer = setInterval(searchTick, searchIntervalMs); // Same cadence as the queue sample: long enough that the line is a summary // rather than a stream, short enough to bracket an experiment against. @@ -891,6 +939,9 @@ void cardTick(); void alertTick(); void enrichTick(); void queueTick(); +// Run once at boot, so a deploy does not leave the page slow until the first +// interval comes round. +void statsTick(); log('started', { intervalSeconds: intervalMs / 1000, @@ -922,6 +973,7 @@ function shutdown(signal) { clearInterval(enrichTimer); clearInterval(alertTimer); clearInterval(queueTimer); + clearInterval(statsTimer); clearInterval(searchTimer); clearInterval(tallyTimer); diff --git a/apps/web/src/app/api/crawlstats/route.js b/apps/web/src/app/api/crawlstats/route.js index bc52d8f..a17f384 100644 --- a/apps/web/src/app/api/crawlstats/route.js +++ b/apps/web/src/app/api/crawlstats/route.js @@ -1,7 +1,14 @@ -import { q, discovery, alerts } from '@rssamplifier/db'; +import { q, discovery } from '@rssamplifier/db'; import { db } from '../../../lib/db.js'; -import { categoryStats, indexingHistory, jobBacklogs } from '../../../lib/crawlstats.js'; +import { + categoryStats, + indexingHistory, + jobBacklogs, + liveStats, + failingFeeds, + alertingAccounts, +} from '../../../lib/crawlstats.js'; import { toLine } from '../../../lib/crawlLog.js'; import { jobRows } from '../../../lib/jobs.js'; @@ -15,10 +22,15 @@ export const dynamic = 'force-dynamic'; * backlog that should drain within a tick or two, `stale` is active feeds the * crawler has not successfully read in a day. * - * Deliberately uncached — a status endpoint that answers from a cache reports - * that everything was fine a minute ago, which is the one thing it must not do. - * The two additions that are cached, briefly, are the ones nothing would alert - * on: the hourly history and the category breakdown. See lib/crawlstats.js. + * Every read here is cached, and the distinction that keeps that honest is + * between a fact and a derivation. A count is a fact and may be ten seconds + * old; `idleMinutes` is `now - lastSuccessAt`, so caching it would freeze the + * one number a monitor alerts on. `liveStats` caches the timestamp and redoes + * the subtraction — see lib/crawlstats.js. + * + * Before this, the endpoint answered in 118 seconds: `categoryStats` no longer + * completes inside the client's 30s deadline, and since a cache that only + * stores successes never stored it, every request paid the full timeout. */ export async function GET() { const client = db(); @@ -36,8 +48,8 @@ export async function GET() { alertAccounts, operationalErrors, ] = await Promise.all([ - q.crawlStats(client), - q.failingFeeds(client, 20), + liveStats(), + failingFeeds(), q.recentlyCrawled(client, 20), discovery.countQueuedCandidates(client), discovery.countQueuedKeywords(client), @@ -57,7 +69,7 @@ export async function GET() { q.logActivity(client, 1), // See the page: this only tells a sender with nobody to serve from one that // has stopped, which the log alone cannot say. - alerts.alertingAccountCount(client), + alertingAccounts(), q.crawlOperationalErrors(client, { limit: 20, hours: 24 }), ]); diff --git a/apps/web/src/app/crawlstats/page.jsx b/apps/web/src/app/crawlstats/page.jsx index 47e30ef..14aaa7f 100644 --- a/apps/web/src/app/crawlstats/page.jsx +++ b/apps/web/src/app/crawlstats/page.jsx @@ -1,4 +1,4 @@ -import { q, discovery, alerts } from '@rssamplifier/db'; +import { q, discovery } from '@rssamplifier/db'; import { db } from '../../lib/db.js'; import { @@ -6,6 +6,9 @@ import { indexingHistory, jobBacklogs, queueHistory, + liveStats, + failingFeeds, + alertingAccounts, GROWTH_DAYS, } from '../../lib/crawlstats.js'; import { describe, toLine } from '../../lib/crawlLog.js'; @@ -62,8 +65,8 @@ export default async function CrawlStatsPage() { operationalErrors, queues, ] = await Promise.all([ - q.crawlStats(client), - q.failingFeeds(client, 50), + liveStats(), + failingFeeds(50), q.recentlyCrawled(client, 15), discovery.countQueuedCandidates(client), discovery.countQueuedKeywords(client), @@ -90,7 +93,7 @@ export default async function CrawlStatsPage() { // Only to tell a sender with nothing to do from one that has stopped: the // alert pass writes no log line at all when nobody is subscribed, and a // silent job is otherwise indistinguishable from a dead one. - alerts.alertingAccountCount(client), + alertingAccounts(), // Kept separately from the rolling live-log window. At crawler throughput, // 400 successful feed lines can evict an operational failure in minutes. q.crawlOperationalErrors(client, { limit: 20, hours: 24 }), diff --git a/apps/web/src/lib/crawlstats.js b/apps/web/src/lib/crawlstats.js index 6b74c77..4789a2d 100644 --- a/apps/web/src/lib/crawlstats.js +++ b/apps/web/src/lib/crawlstats.js @@ -1,26 +1,42 @@ -import { q } from '@rssamplifier/db'; +import { q, alerts, remember } from '@rssamplifier/db'; import { db } from './db.js'; /** - * The slow half of /crawlstats, cached in the process. + * The slow half of /crawlstats, cached in Redis. * * The status numbers on that page have to be current to the second — a stalled * crawler that reads "healthy" because the answer came from a cache is the one * failure the page exists to catch. The breakdowns below are the opposite kind - * of number: a directory of 48k feeds does not change shape between two - * fifteen-second refreshes, and recomputing a group-by over every feed on each - * one bills a full table scan for an answer that is the same all morning. + * of number: a directory of half a million feeds does not change shape between + * two fifteen-second refreshes, and recomputing a group-by over every feed on + * each one bills a full table scan for an answer that is the same all morning. * - * So they are cached, separately, for as long as each is actually still true: + * ## Why this moved out of the process * - * - the hourly rollup is a handful of rows and moves within the hour, so a - * minute is plenty of staleness to accept for it; - * - the category breakdown is two scans of `feeds` and moves at the speed of - * the crawler finding new blogs, which is nothing like every fifteen seconds. + * These caches used to live in module scope, which was right as far as it went + * and had two holes. It died with the process, so every deploy re-paid the full + * cost on the next request; and it was per instance, so nothing was shared. * - * Per process and dying with it, matching `popularLanguages` in ./languages.js: - * a deploy or a restart is exactly when it is worth looking again. + * The second hole is the one that bit. `categoryStats` stopped merely being + * slow and started *failing* — measured 2026-08-21, it does not finish inside + * the client's 30s deadline against 476,715 feeds — and a cache that only + * stores successes never stores anything. So every request paid 30 seconds, for + * ever, and `/api/crawlstats` answered in 118. Redis plus serve-stale-on-failure + * means one success, any time, is enough for every later reader. + * + * Redis also adds no writes to Turso, which matters more than usual here: the + * write path is this database's binding constraint, so a rollup table would put + * the fix on the hot side of it. + * + * ## What is cached, and what is derived + * + * Facts are cached; anything computed against "now" is derived at serve time. + * An hour label or an idle-minutes count baked into a cached blob is frozen, + * and a frozen liveness number is exactly the lie this page must not tell. So + * `queueHistory` caches the sparse rows and fills the window on the way out, + * and `crawlStats` (in the route) re-derives `idleMinutes` from the cached + * `lastSuccessAt` timestamp. */ /** How long an hourly rollup read is trusted. */ @@ -29,6 +45,21 @@ const HISTORY_TTL_MS = 60 * 1000; /** How long a category breakdown is trusted. */ const CATEGORY_TTL_MS = 5 * 60 * 1000; +/** + * How stale a breakdown may get before a reader waits for a fresh one. + * + * Generous on purpose. These are shape-of-the-directory numbers, and the whole + * reason the window is wide is that the alternative — when the underlying read + * is failing — is no chart at all rather than a slightly old one. + */ +const CHART_MAX_STALE_MS = 24 * 60 * 60 * 1000; + +/** + * Shorter than the client's own 30s deadline, so a read that is going to hang + * gives the page back before the browser gives up on it. + */ +const CHART_TIMEOUT_MS = 20 * 1000; + /** How much history the charts draw. */ export const HISTORY_HOURS = 24; export const GROWTH_DAYS = 30; @@ -43,15 +74,6 @@ export const GROWTH_DAYS = 30; */ export const QUEUE_HOURS = 48; -/** @type {{ at: number, value: Awaited> }|null} */ -let historyCache = null; - -/** @type {{ at: number, value: Awaited> }|null} */ -let categoryCache = null; - -/** @type {{ at: number, value: { hours: string[], series: Record> } }|null} */ -let queueCache = null; - /** * Crawler throughput, hour by hour. * @@ -64,15 +86,12 @@ let queueCache = null; * @returns {Promise>>} */ export async function indexingHistory() { - if (historyCache && Date.now() - historyCache.at < HISTORY_TTL_MS) return historyCache.value; - - try { - const value = await q.indexingHistory(db(), HISTORY_HOURS); - historyCache = { at: Date.now(), value }; - return value; - } catch { - return []; - } + const value = await remember( + 'indexingHistory', + { ttlMs: HISTORY_TTL_MS, maxStaleMs: CHART_MAX_STALE_MS, timeoutMs: CHART_TIMEOUT_MS, fallback: [] }, + () => q.indexingHistory(db(), HISTORY_HOURS), + ); + return value ?? []; } /** @@ -85,6 +104,10 @@ export async function indexingHistory() { * has would show. * * So this fills the window and leaves `null` where nothing was written down. + * The fill happens *after* the cache rather than before it: the hour labels are + * built from the current time, so a cached dense array would still be carrying + * yesterday's axis tomorrow. The sparse rows are the fact worth keeping. + * * Empty rather than throwing, on the same reasoning as `indexingHistory`: the * poller owns migration, so there is a deploy window where `queue_hourly` does * not exist yet and losing a chart must not take the page with it. @@ -92,155 +115,193 @@ export async function indexingHistory() { * @returns {Promise<{ hours: string[], series: Record> }>} */ export async function queueHistory() { - if (queueCache && Date.now() - queueCache.at < HISTORY_TTL_MS) return queueCache.value; - - try { - const rows = await q.queueHistory(db(), QUEUE_HOURS); - const byHour = new Map(rows.map((r) => [r.hour, r])); - - const hours = []; - const now = Date.now(); - for (let i = QUEUE_HOURS - 1; i >= 0; i--) { - hours.push(new Date(now - i * 3_600_000).toISOString().slice(0, 13)); - } - - const pick = (key) => hours.map((h) => (byHour.has(h) ? Number(byHour.get(h)[key]) : null)); - - const value = { - hours, - series: { - due: pick('due'), - firstCrawl: pick('firstCrawl'), - cards: pick('cards'), - authors: pick('authors'), - }, - }; - - queueCache = { at: Date.now(), value }; - return value; - } catch { - return { hours: [], series: {} }; + const rows = await remember( + 'queueHistory', + { ttlMs: HISTORY_TTL_MS, maxStaleMs: CHART_MAX_STALE_MS, timeoutMs: CHART_TIMEOUT_MS, fallback: null }, + () => q.queueHistory(db(), QUEUE_HOURS), + ); + + if (!rows) return { hours: [], series: {} }; + + const byHour = new Map(rows.map((r) => [r.hour, r])); + + const hours = []; + const now = Date.now(); + for (let i = QUEUE_HOURS - 1; i >= 0; i--) { + hours.push(new Date(now - i * 3_600_000).toISOString().slice(0, 13)); } + + const pick = (key) => hours.map((h) => (byHour.has(h) ? Number(byHour.get(h)[key]) : null)); + + return { + hours, + series: { + due: pick('due'), + firstCrawl: pick('firstCrawl'), + cards: pick('cards'), + authors: pick('authors'), + }, + }; } /** * The directory by category, with each category's growth curve. * - * Same bargain as above: a status page missing its breakdown is worth more than - * a status page that 500s. + * The slowest read on the page and the reason this module exists. It wants five + * columns per feed (category, status, created_at, last_success_at, item_count) + * and three of them are rewritten on every crawl, so an index wide enough to + * cover it would cost more on the write path than the read is worth. Measured + * against production it no longer completes at all: a bare `count(*)` of + * `feeds` is 6.9s and `select category, count(*) … group by category` exceeds + * the 30s client deadline. Removing its conditional aggregates — the fix that + * worked for `crawlStats` in PR #96 — does not help, because the cost is + * visiting every row for a column no index covers. + * + * Which is why it is served stale for up to a day rather than recomputed: + * five minutes and five minutes plus a failed thirty-second scan are the same + * answer, and a day-old breakdown beats the empty one this returned before. * * @returns {Promise>>} */ export async function categoryStats() { - const fresh = categoryCache && Date.now() - categoryCache.at < CATEGORY_TTL_MS; - if (fresh) return categoryCache.value; - - // Expired but present: hand back the old answer and refresh behind it. - // - // This read is the slowest thing left on the page — 6.1 seconds against - // production, because it wants five columns per feed (category, status, - // created_at, last_success_at, item_count) and three of them are rewritten on - // every crawl, so an index wide enough to cover it would cost more on the - // write path than the read is worth. A plain TTL therefore does not make the - // page fast, it makes one reader in every five minutes wait six seconds for a - // breakdown that was already almost right. - // - // Serving stale while revalidating is the honest trade for this particular - // number: it is a shape-of-the-directory figure that moves at the speed of - // the crawler finding new blogs, so five minutes and five minutes plus six - // seconds are the same answer. - if (categoryCache) { - if (!categoryRefreshing) { - categoryRefreshing = true; - refreshCategories().finally(() => { - categoryRefreshing = false; - }); - } - return categoryCache.value; - } + const value = await remember( + 'categoryStats', + { + ttlMs: CATEGORY_TTL_MS, + maxStaleMs: CHART_MAX_STALE_MS, + timeoutMs: CHART_TIMEOUT_MS, + fallback: null, + }, + () => q.categoryStats(db(), GROWTH_DAYS), + ); - // Nothing cached at all — the first request after a deploy has to wait. - return (await refreshCategories()) ?? { total: 0, days: [], categories: [] }; + return value ?? { total: 0, days: [], categories: [] }; } -/** Guards against a slow refresh being started once per concurrent request. */ -let categoryRefreshing = false; +/** + * How long the liveness numbers are trusted. + * + * Ten seconds against a page that refreshes every fifteen, so a reader is never + * looking at anything meaningfully older than the last refresh, and a burst of + * concurrent readers costs one read rather than one each. + */ +const STATS_TTL_MS = 10 * 1000; /** - * Re-read the breakdown and store it, returning null rather than throwing. + * The status numbers, cached briefly — and the derivation that makes that safe. + * + * `crawlStats` was deliberately never cached, because a stalled crawler reading + * "healthy" is the one failure /crawlstats exists to catch. That reasoning is + * right about `idleMinutes` and wrong about everything else on the object: the + * counts are counts, but `idleMinutes` is computed as `now - lastSuccessAt` at + * the moment the query runs, so caching the object *freezes it*. A crawler that + * died would go on reporting the same cheerful number until the entry expired. + * + * So the fact is cached and the derivation is redone here. `lastSuccessAt` is a + * timestamp — it does not go stale, it just gets further away — and recomputing + * the gap against the current clock gives a number that keeps climbing while + * the crawler is down, which is exactly the alarm that must not be cacheable. + * `generatedAt` is left as the moment the read actually happened, so the page + * can be honest about how old the counts beside it are. + * + * The read itself measured 4,975ms against production, which is why it is worth + * doing at all. * - * @returns {Promise>|null>} + * @returns {Promise>>} */ -async function refreshCategories() { - try { - const value = await q.categoryStats(db(), GROWTH_DAYS); - categoryCache = { at: Date.now(), value }; - return value; - } catch { - // A failed refresh leaves the previous answer in place and is retried on - // the next request, which is the whole point of keeping the old value. - return null; - } +export async function liveStats() { + const stats = await remember( + 'crawlStats', + { + ttlMs: STATS_TTL_MS, + // Minutes, not hours: these are the numbers that must not drift far, and + // if they cannot be read at all the page should say so by other means. + maxStaleMs: 2 * 60 * 1000, + timeoutMs: CHART_TIMEOUT_MS, + fallback: null, + }, + () => q.crawlStats(db()), + ); + + if (!stats) return await q.crawlStats(db()); + + const lastSuccessAt = stats.lastSuccessAt ? String(stats.lastSuccessAt) : null; + return { + ...stats, + idleMinutes: lastSuccessAt + ? Math.max(0, Math.round((Date.now() - Date.parse(lastSuccessAt)) / 60_000)) + : null, + }; } -/** How long a job-board read is trusted. */ -const JOBS_TTL_MS = 60 * 1000; +/** + * The feeds failing hardest, for the table at the bottom of the page. + * + * 5,172ms against production: it sorts the whole error population. A minute old + * is fine — a feed that has failed eleven times has not stopped failing since + * the last refresh, and this list is read to find a pattern, not to catch an + * event. + * + * Keyed by limit, because the page asks for 50 and the JSON endpoint for 20 -- + * one key would hand whichever asked second the wrong-length list. + * + * @param {number} [limit] + * @returns {Promise>>} + */ +export async function failingFeeds(limit = 20) { + const value = await remember( + `failingFeeds:${limit}`, + { ttlMs: 60 * 1000, maxStaleMs: 6 * 60 * 60 * 1000, timeoutMs: CHART_TIMEOUT_MS, fallback: [] }, + () => q.failingFeeds(db(), limit), + ); + return value ?? []; +} -/** @type {{ at: number, value: Awaited> }|null} */ -let jobsCache = null; +/** + * How many accounts have alerts configured. + * + * Only tells a sender with nobody to serve from one that has stopped, so it + * moves at the speed of people signing up and can be an hour old without + * anybody being misled. + * + * @returns {Promise} + */ +export async function alertingAccounts() { + const value = await remember( + 'alertingAccounts', + { ttlMs: 5 * 60 * 1000, maxStaleMs: CHART_MAX_STALE_MS, timeoutMs: CHART_TIMEOUT_MS, fallback: 0 }, + () => alerts.alertingAccountCount(db()), + ); + return Number(value ?? 0); +} -/** Guards against a slow refresh being started once per concurrent request. */ -let jobsRefreshing = false; +/** How long a job-board read is trusted. */ +const JOBS_TTL_MS = 60 * 1000; /** * The job board's backlogs, cached and served stale while it refreshes. * - * This was the last uncached scan of `feeds` on the page, and it is a scan - * however well it is written: counting the directory by status, by card state - * and by never-crawled means visiting every row, and there are 369,030 of them. - * On an idle database that is 398ms, which is why it shipped uncached. Under - * the crawler's write load the same statement measured **16.9 seconds** -- and - * the page pays it on every request, because /crawlstats is deliberately - * `force-dynamic`. + * A scan of `feeds` however well it is written: counting the directory by + * status, by card state and by never-crawled means visiting every row, and + * there are 476,715 of them. On an idle database that is 398ms, which is why it + * shipped uncached; under the crawler's write load the same statement measured + * 16.9 seconds, and 11.3 in the timing that prompted this change. * * A minute of staleness costs nothing here. These are backlogs of hundreds of * thousands of feeds draining at a few hundred an hour; they do not * meaningfully move between two views of a page that refreshes itself every - * fifteen seconds. The health badge and the live figures beside it are read - * separately and stay current to the second -- see `crawlStats`, which is the - * one thing on this page that must never be served from a cache, because a - * stalled crawler reading "healthy" is the failure the page exists to catch. + * fifteen seconds. + * + * Null rather than zeroes when there is nothing to serve: a job board showing + * "0 waiting" because the read failed reads as "all caught up", where a missing + * board is merely missing. * * @returns {Promise>|null>} */ export async function jobBacklogs() { - if (jobsCache && Date.now() - jobsCache.at < JOBS_TTL_MS) return jobsCache.value; - - if (jobsCache) { - if (!jobsRefreshing) { - jobsRefreshing = true; - refreshJobs().finally(() => { - jobsRefreshing = false; - }); - } - return jobsCache.value; - } - - return refreshJobs(); -} - -/** - * @returns {Promise>|null>} - */ -async function refreshJobs() { - try { - const value = await q.jobBacklogs(db()); - jobsCache = { at: Date.now(), value }; - return value; - } catch { - // The previous answer if there is one, null if there is not. Zeroes would - // be a lie: a job board showing "0 waiting" because the read failed reads - // as "all caught up", where a missing board is merely missing. - return jobsCache?.value ?? null; - } + return remember( + 'jobBacklogs', + { ttlMs: JOBS_TTL_MS, maxStaleMs: 6 * 60 * 60 * 1000, timeoutMs: CHART_TIMEOUT_MS, fallback: null }, + () => q.jobBacklogs(db()), + ); } diff --git a/packages/db/index.js b/packages/db/index.js index 907c9bb..fc62a95 100644 --- a/packages/db/index.js +++ b/packages/db/index.js @@ -1,5 +1,7 @@ export { connect, newId, nowIso } from './src/client.js'; export { createWriteWorker, WRITE_QUEUE, takeWriteTally } from './src/writeQueue.js'; +export { remember, redisClient, primeCache, resetCacheState } from './src/cache.js'; +export { warmStatsCache } from './src/statsWarmer.js'; export { migrate } from './src/migrate.js'; export * as q from './src/queries.js'; export * as accounts from './src/accounts.js'; diff --git a/packages/db/src/cache.js b/packages/db/src/cache.js new file mode 100644 index 0000000..97df9a9 --- /dev/null +++ b/packages/db/src/cache.js @@ -0,0 +1,298 @@ +/** + * A read-through cache for the expensive halves of /crawlstats. + * + * ## Why this exists + * + * `/api/crawlstats` fans out eleven reads and returns when the slowest one + * does. Measured against production on 2026-08-21: + * + * categoryStats 30,005ms (timed out) + * jobBacklogs 11,255ms + * failingFeeds(20) 5,172ms + * crawlStats 4,975ms + * ...the other seven under 600ms each + * Promise.all of all 11 30,008ms + * + * The endpoint answered in 118 seconds. `categoryStats` alone sets the floor, + * and **no rewrite fixes it**: it is a `group by category` over 476,715 rows, + * and on the same connection a bare `count(*)` of that table is 6.9s while + * `select category, count(*) … group by category` does not finish inside the + * client's 30s deadline. A conditional aggregate was the bug in `crawlStats` + * (PR #96) and removing the CASEs here changes nothing, because the cost is + * visiting every row for a column that no index covers. + * + * So the fix is to stop doing it on the request path. + * + * ## Why Redis rather than the per-process cache already here + * + * `apps/web/src/lib/crawlstats.js` caches these in module scope, which is right + * as far as it goes and has two holes this closes. It dies with the process, so + * every deploy re-pays the full cost on the next request; and it is per + * instance, so it cannot be shared. Redis also adds **no writes to Turso**, + * which matters more than usual here: the write path is the binding constraint + * on this database, and a rollup table would put the fix on the hot side of it. + * + * ## Why stale-while-revalidate, and not a plain TTL + * + * A plain cache never fills for the read that needs it most. `categoryStats` + * does not merely run slowly, it *fails* — so a "compute on miss, store on + * success" cache stores nothing, and every single request pays 30 seconds + * forever. That is the state production is in. + * + * Serving stale while refreshing in the background inverts it: one success, any + * time, is enough for every later reader to be served instantly, and the + * refresh that keeps failing costs nobody a wait. A value that cannot be + * recomputed is returned however old it is, because a month-old category + * breakdown is a better answer than a thirty-second hang. + * + * ## What must not be cached this way + * + * The liveness numbers. This page exists to catch a stalled crawler, and one + * that reads "healthy" from a cache is the single failure that would make it + * worthless. Those keys take a short `ttlMs` and a small `maxStaleMs`, so + * staleness is bounded by seconds rather than by whether a refresh succeeds -- + * see the callers for which is which. + */ + +/** @type {Map>} in-flight refreshes, per process */ +const inFlight = new Map(); + +/** @type {{ client: unknown, url: string }|null} */ +let shared = null; + +/** + * The process's Redis client, or null when no `REDIS_URL` is configured. + * + * Lazily imported rather than pulled in at module load: this module is reached + * from Next server code, and `ioredis` at the top level drags a node-only + * dependency into any bundle that so much as touches `@rssamplifier/db`. + * + * @param {string} [url] + * @returns {Promise} + */ +export async function redisClient(url = process.env['REDIS_URL'] ?? '') { + if (!url) return null; + if (shared && shared.url === url) return shared.client; + + try { + const { default: Redis } = await import('ioredis'); + // A cache must never be the reason a page is slow. `commandTimeout` is the + // load-bearing option: without it a Redis that accepts connections but + // stops answering leaves every `get` hanging, and the cache becomes the + // stall it was added to remove. + // + // The offline queue is deliberately left on. Railway's Redis is reached + // over internal DNS, so the first request after a cold start can arrive + // before the socket is ready; queueing briefly is better than failing those + // outright, and `commandTimeout` still bounds the wait either way. No + // `family` option for the same reason the write queue needs none -- the + // same URL already works there. + const client = new Redis(url, { + maxRetriesPerRequest: 1, + commandTimeout: 1_000, + connectTimeout: 3_000, + }); + // Without a listener an ioredis connection error is an unhandled 'error' + // event, which takes the process down -- a cache outage becoming a site + // outage is the opposite of the point. + client.on('error', () => {}); + shared = { client, url }; + return client; + } catch { + return null; + } +} + +/** + * Run `fn`, giving up after `ms`. + * + * The underlying read keeps running -- there is no cancellation in the libSQL + * client -- but nobody is waiting on it any more, which is the part that + * matters for a response deadline. + * + * @template T + * @param {() => Promise} fn + * @param {number} ms + * @returns {Promise} + */ +function withTimeout(fn, ms) { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error(`timed out after ${ms}ms`)), ms); + fn().then( + (v) => { clearTimeout(timer); resolve(v); }, + (e) => { clearTimeout(timer); reject(e); }, + ); + }); +} + +/** + * Read `key` from the cache, recomputing it when it is old enough to matter. + * + * Returns `fallback` (default null) only when there is nothing cached *and* + * the computation failed -- callers already treat that as "chart unavailable" + * rather than as an error, which is why a failed read here must not throw. + * + * @template T + * @param {string} key + * @param {{ + * ttlMs: number, + * maxStaleMs?: number, + * timeoutMs?: number, + * fallback?: T|null, + * client?: any, + * }} opts + * @param {() => Promise} compute + * @returns {Promise} + */ +export async function remember(key, opts, compute) { + const { + ttlMs, + maxStaleMs = 24 * 60 * 60 * 1000, + timeoutMs = 20_000, + fallback = null, + } = opts; + + const client = opts.client !== undefined ? opts.client : await redisClient(); + + // No Redis configured is not an error, it is the local and the test case: + // behave exactly as the uncached code did. + if (!client) { + try { + return await withTimeout(compute, timeoutMs); + } catch { + return fallback; + } + } + + const entry = await readEntry(client, key); + const age = entry ? Date.now() - entry.at : Infinity; + + if (entry && age < ttlMs) return entry.value; + + // Stale but usable: answer now, and refresh behind the reader. This is the + // path that makes a 30-second read invisible. + if (entry && age < maxStaleMs) { + refresh(client, key, compute, timeoutMs); + return entry.value; + } + + // Nothing usable cached, so this reader has to wait for it. + try { + const value = await withTimeout(compute, timeoutMs); + await writeEntry(client, key, value, maxStaleMs); + return value; + } catch { + // It failed, and an expired value is still a better answer than none -- + // this is what keeps `categoryStats` serving after it stops completing. + return entry ? entry.value : fallback; + } +} + +/** + * Recompute in the background, at most once at a time per key. + * + * Deduped because a page that refreshes every fifteen seconds would otherwise + * start a new thirty-second scan on every request and pile them up against the + * database this is trying to spare. + * + * @param {any} client + * @param {string} key + * @param {() => Promise} compute + * @param {number} timeoutMs + */ +function refresh(client, key, compute, timeoutMs) { + if (inFlight.has(key)) return; + + const task = withTimeout(compute, timeoutMs) + .then((value) => writeEntry(client, key, value, 24 * 60 * 60 * 1000)) + // A background refresh that fails is not an event: the reader already has + // an answer, and the next one will try again. + .catch(() => {}) + .finally(() => inFlight.delete(key)); + + inFlight.set(key, task); +} + +/** + * @param {any} client + * @param {string} key + * @returns {Promise<{ at: number, value: any }|null>} + */ +async function readEntry(client, key) { + try { + // Bounded independently of `commandTimeout`, because the client here may be + // a stand-in rather than ioredis, and a cache lookup that can hang is not a + // cache. Well under any of the read timeouts it is protecting. + const raw = await withTimeout(() => client.get(cacheKey(key)), 1_500); + if (!raw) return null; + const parsed = JSON.parse(String(raw)); + if (!parsed || typeof parsed.at !== 'number') return null; + return parsed; + } catch { + // Unreadable or unparseable is the same as absent. A cache cannot be + // allowed to fail a request. + return null; + } +} + +/** + * Stored with an expiry well past `maxStaleMs` so that the *policy* about how + * stale is too stale lives in one place -- `remember` -- rather than being + * split between here and Redis's own eviction. + * + * @param {any} client + * @param {string} key + * @param {unknown} value + * @param {number} maxStaleMs + */ +async function writeEntry(client, key, value, maxStaleMs) { + try { + const body = JSON.stringify({ at: Date.now(), value }, (_, v) => + typeof v === 'bigint' ? Number(v) : v, + ); + await withTimeout( + () => client.set(cacheKey(key), body, 'PX', Math.max(maxStaleMs * 2, 60_000)), + 1_500, + ); + } catch { + // Failing to store is survivable: the value was still computed and is + // being returned. The next reader simply pays for it again. + } +} + +/** + * @param {string} key + * @returns {string} + */ +function cacheKey(key) { + return `rsa:stats:${key}`; +} + +/** + * Store a value under `key` as though a reader had just computed it. + * + * For work that cannot be done on a request: the category breakdown takes ~59 + * seconds, so no page can wait for it and a cache that only fills from readers + * never fills at all. A background job computes it on a patient connection and + * primes the same key, and every reader is then served from Redis. + * + * Goes through `writeEntry` rather than reimplementing the envelope, so the + * warmer and the reader cannot drift apart on the format. + * + * @param {string} key + * @param {unknown} value + * @param {{ client?: any, maxStaleMs?: number }} [opts] + * @returns {Promise} whether it was stored + */ +export async function primeCache(key, value, opts = {}) { + const client = opts.client !== undefined ? opts.client : await redisClient(); + if (!client) return false; + await writeEntry(client, key, value, opts.maxStaleMs ?? 24 * 60 * 60 * 1000); + return true; +} + +/** Test seam: forget the in-flight refreshes between cases. */ +export function resetCacheState() { + inFlight.clear(); + shared = null; +} diff --git a/packages/db/src/client.js b/packages/db/src/client.js index 584ec00..01fb46a 100644 --- a/packages/db/src/client.js +++ b/packages/db/src/client.js @@ -15,7 +15,13 @@ import { queueWrites } from './writeQueue.js'; * A `file:` URL needs no auth token, which is what makes local development and * the test suite work without a Turso account. * - * @param {{ url?: string, authToken?: string, redisUrl?: string, queue?: boolean }} [opts] + * `timeoutMs` overrides the per-request deadline for this connection alone. + * The default is right for anything serving a page, and wrong for the one + * background job that recomputes the category breakdown: that read takes ~59 + * seconds against half a million feeds, so on the default it can only ever + * fail. See `warmStatsCache` in ./statsWarmer.js. + * + * @param {{ url?: string, authToken?: string, redisUrl?: string, queue?: boolean, timeoutMs?: number }} [opts] * @returns {import('@libsql/client').Client} */ export function connect(opts = {}) { @@ -26,7 +32,9 @@ export function connect(opts = {}) { if (!url) throw new Error('TURSO_DATABASE_URL must be set'); const client = createClient( - url.startsWith('file:') ? { url } : { url, authToken, fetch: withTimeout(requestTimeoutMs()) }, + url.startsWith('file:') + ? { url } + : { url, authToken, fetch: withTimeout(opts.timeoutMs ?? requestTimeoutMs()) }, ); // Redis moves the write queue out of the process, which is the only way to diff --git a/packages/db/src/statsWarmer.js b/packages/db/src/statsWarmer.js new file mode 100644 index 0000000..597bc12 --- /dev/null +++ b/packages/db/src/statsWarmer.js @@ -0,0 +1,90 @@ +import { connect } from './client.js'; +import { primeCache } from './cache.js'; +import * as q from './queries.js'; + +/** + * Recompute the numbers no page can afford to wait for, and put them in Redis. + * + * ## Why a background job and not a faster query + * + * `categoryStats` groups half a million feeds by category while reading their + * status, item count and last-success time. Measured against production on + * 2026-08-21, decomposed piece by piece: + * + * feeds per category 6,340ms (index-only, uses the partial index) + * growth by day+category 1,046ms (index-only) + * status x category 23,942ms + * sum(item_count) per category 35,440ms + * crawled-in-last-day 40,020ms + * + * The three slow ones read columns no index covers, so they fetch every row. + * An index that covered them would have to carry `status` and `last_success_at`, + * both rewritten on every crawl — and writes, not reads, are this database's + * binding constraint. Buying a fast breakdown with a slower crawler is the + * wrong trade. + * + * So the query stays as it is and stops being on the request path. The whole + * statement completes in **58.9 seconds** given a deadline long enough to allow + * it, which is the other half of this: the 30s ceiling it kept hitting is this + * codebase's own `TURSO_REQUEST_TIMEOUT_MS`, not a limit of the database. A + * connection with a patient deadline can finish what a page never could. + * + * ## Why this is safe to run beside the crawler + * + * It is a read, and reads bypass the write serialization entirely — they do not + * take the single writer that everything else queues behind. One long read every + * few minutes costs the crawler nothing. + * + * ## Why it primes rather than returns + * + * Nothing here consumes the value. The point is that `remember('categoryStats')` + * on the web side finds it already there, so the reader that would have waited + * 20 seconds and then given up is served from Redis instead. + */ + +/** Long enough for the ~59s read, with room for a bad day. */ +const WARM_TIMEOUT_MS = 150_000; + +/** How much growth history the breakdown carries; matches the web's GROWTH_DAYS. */ +const GROWTH_DAYS = 30; + +/** + * Compute the category breakdown on a patient connection and cache it. + * + * Never throws. A warmer that takes the poller down with it would trade a slow + * chart for a stopped crawler, and the cache simply keeps serving whatever it + * last had. + * + * @param {{ log?: (event: string, fields?: object) => void, client?: any }} [opts] + * @returns {Promise<{ ok: boolean, ms: number, cached: boolean, error?: string }>} + */ +export async function warmStatsCache(opts = {}) { + const started = Date.now(); + const log = opts.log ?? (() => {}); + + /** @type {import('@libsql/client').Client|null} */ + let db = null; + try { + // Its own connection, deliberately. Sharing the poller's would either impose + // this job's 150-second deadline on every crawl write or leave this job with + // the 30 seconds it cannot finish in. + db = connect({ timeoutMs: WARM_TIMEOUT_MS, queue: false }); + + const value = await q.categoryStats(db, GROWTH_DAYS); + const cached = await primeCache('categoryStats', value, { client: opts.client }); + + const ms = Date.now() - started; + log('stats-warmed', { ms, categories: value?.categories?.length ?? 0, cached }); + return { ok: true, ms, cached }; + } catch (error) { + const ms = Date.now() - started; + // Logged as `stats-warm-skipped`, not `…-error`: the page has a cached + // answer and is fine, and the operational-error panel treats any event + // whose name ends in "error" as an alarm. A missed warm is not one. + const reason = error instanceof Error ? error.message : String(error); + log('stats-warm-skipped', { ms, reason }); + return { ok: false, ms, cached: false, error: reason }; + } finally { + try { db?.close(); } catch { /* closing a broken client is not a failure */ } + } +} diff --git a/packages/db/test/cache.test.js b/packages/db/test/cache.test.js new file mode 100644 index 0000000..1dec893 --- /dev/null +++ b/packages/db/test/cache.test.js @@ -0,0 +1,230 @@ +import assert from 'node:assert/strict'; +import { test, beforeEach } from 'node:test'; + +import { remember, primeCache, resetCacheState } from '../src/cache.js'; + +/** + * A Redis stand-in. + * + * Only `get` and `set` are used, and only the `PX` form of `set` -- keeping it + * this small is deliberate, because a fake that drifts from the real client is + * worse than none. Expiry is modelled because "the key vanished" is a state the + * cache has to survive, and it is the one ioredis behaviour these tests depend + * on beyond plain storage. + */ +function fakeRedis() { + const store = new Map(); + return { + store, + calls: { get: 0, set: 0 }, + async get(k) { + this.calls.get += 1; + const hit = store.get(k); + if (!hit) return null; + if (hit.until <= Date.now()) { store.delete(k); return null; } + return hit.body; + }, + async set(k, body, _px, ms) { + this.calls.set += 1; + store.set(k, { body, until: Date.now() + Number(ms) }); + return 'OK'; + }, + }; +} + +/** Rewrite a stored entry's timestamp, to age it without waiting. */ +function age(client, key, ms) { + const full = `rsa:stats:${key}`; + const hit = client.store.get(full); + const parsed = JSON.parse(hit.body); + parsed.at -= ms; + client.store.set(full, { ...hit, body: JSON.stringify(parsed) }); +} + +const settle = () => new Promise((r) => setTimeout(r, 20)); + +beforeEach(() => resetCacheState()); + +test('a miss computes, stores, and returns', async () => { + const client = fakeRedis(); + let ran = 0; + + const got = await remember('k', { ttlMs: 1000, client }, async () => { ran += 1; return { n: 1 }; }); + + assert.deepEqual(got, { n: 1 }); + assert.equal(ran, 1); + assert.equal(client.calls.set, 1, 'the value was stored'); +}); + +test('a fresh hit does not run the computation at all', async () => { + const client = fakeRedis(); + let ran = 0; + const compute = async () => { ran += 1; return { n: ran }; }; + + await remember('k', { ttlMs: 60_000, client }, compute); + const second = await remember('k', { ttlMs: 60_000, client }, compute); + + assert.equal(ran, 1, 'the second read was served from the cache'); + assert.deepEqual(second, { n: 1 }); +}); + +test('a stale hit answers immediately and refreshes behind the reader', async () => { + const client = fakeRedis(); + let ran = 0; + + await remember('k', { ttlMs: 100, client }, async () => { ran += 1; return { n: 1 }; }); + age(client, 'k', 5_000); // now well past ttl, well inside maxStale + + // The point of the whole module: the caller gets the old value now, not in + // however long the recomputation takes. + const got = await remember('k', { ttlMs: 100, maxStaleMs: 60_000, client }, async () => { + ran += 1; + await new Promise((r) => setTimeout(r, 50)); + return { n: 2 }; + }); + + assert.deepEqual(got, { n: 1 }, 'served stale rather than waiting'); + // Longer than the 50ms the refresh itself takes: the point being asserted is + // that it lands eventually, not that it lands within one tick. + await new Promise((r) => setTimeout(r, 150)); + assert.equal(ran, 2, 'and the refresh did run'); + + const after = await remember('k', { ttlMs: 100, maxStaleMs: 60_000, client }, async () => ({ n: 99 })); + assert.deepEqual(after, { n: 2 }, 'the refreshed value replaced it'); +}); + +test('a computation that never succeeds still serves the last good answer', async () => { + // This is production's `categoryStats`: it does not run slowly, it fails. A + // plain cache stores nothing and every request pays the full timeout for + // ever. One success has to be enough. + const client = fakeRedis(); + + await remember('cat', { ttlMs: 10, client }, async () => ({ blogs: 7 })); + age(client, 'cat', 10 * 60 * 1000); + + for (let i = 0; i < 3; i++) { + const got = await remember('cat', { ttlMs: 10, maxStaleMs: 60 * 60 * 1000, client }, async () => { + throw new Error('SQLITE timeout'); + }); + assert.deepEqual(got, { blogs: 7 }, 'the stale answer survives a failing refresh'); + await settle(); + } +}); + +test('a value past maxStale is still returned when it cannot be recomputed', async () => { + const client = fakeRedis(); + await remember('cat', { ttlMs: 10, client }, async () => ({ blogs: 7 })); + age(client, 'cat', 48 * 60 * 60 * 1000); // older than maxStale + + const got = await remember('cat', { ttlMs: 10, maxStaleMs: 60_000, client }, async () => { + throw new Error('still down'); + }); + + assert.deepEqual(got, { blogs: 7 }, 'a very old answer beats no answer'); +}); + +test('a slow computation is abandoned at the timeout rather than held', async () => { + const client = fakeRedis(); + const started = Date.now(); + + const got = await remember('slow', { ttlMs: 1000, timeoutMs: 60, fallback: null, client }, async () => { + await new Promise((r) => setTimeout(r, 5_000)); + return 'too late'; + }); + + assert.equal(got, null, 'gave up and returned the fallback'); + assert.ok(Date.now() - started < 2_000, `returned promptly, took ${Date.now() - started}ms`); +}); + +test('concurrent stale reads start only one refresh', async () => { + const client = fakeRedis(); + let ran = 0; + + await remember('k', { ttlMs: 10, client }, async () => { ran += 1; return 1; }); + age(client, 'k', 5_000); + + const compute = async () => { ran += 1; await new Promise((r) => setTimeout(r, 40)); return 2; }; + await Promise.all( + Array.from({ length: 5 }, () => remember('k', { ttlMs: 10, maxStaleMs: 60_000, client }, compute)), + ); + await settle(); + + assert.equal(ran, 2, 'one initial fill plus exactly one refresh, not five'); +}); + +test('with no client configured it behaves exactly like the uncached read', async () => { + let ran = 0; + const compute = async () => { ran += 1; return { n: ran }; }; + + assert.deepEqual(await remember('k', { ttlMs: 60_000, client: null }, compute), { n: 1 }); + assert.deepEqual(await remember('k', { ttlMs: 60_000, client: null }, compute), { n: 2 }); + assert.equal(ran, 2, 'nothing is cached, every call computes'); +}); + +test('a Redis that throws on every command does not fail the read', async () => { + const broken = { + async get() { throw new Error('ECONNREFUSED'); }, + async set() { throw new Error('ECONNREFUSED'); }, + }; + + const got = await remember('k', { ttlMs: 1000, client: broken }, async () => ({ ok: true })); + assert.deepEqual(got, { ok: true }, 'the value still came back'); +}); + +test('a Redis that accepts commands and never answers does not stall the read', async () => { + // The failure mode that would make this module worse than no cache: a socket + // that is up, so nothing errors, but a GET that never settles. Bounding the + // lookup is what keeps a Redis outage from becoming the stall it was added + // to remove. + const hung = { + async get() { await new Promise(() => {}); }, + async set() { await new Promise(() => {}); }, + }; + + const started = Date.now(); + const got = await remember('k', { ttlMs: 1000, client: hung }, async () => ({ ok: true })); + + assert.deepEqual(got, { ok: true }, 'it fell through to the real read'); + assert.ok(Date.now() - started < 8_000, `did not hang, took ${Date.now() - started}ms`); +}); + +test('a failed computation with nothing cached returns the fallback, not a throw', async () => { + const client = fakeRedis(); + const got = await remember('k', { ttlMs: 1000, fallback: {}, client }, async () => { + throw new Error('nope'); + }); + assert.deepEqual(got, {}, 'callers treat this as "unavailable", so it must not throw'); +}); + +test('a primed value is served to readers without them computing anything', async () => { + // The warmer's whole contract. `categoryStats` takes ~59s, so no reader can + // fill this key; a background job fills it and readers must find it there. + const client = fakeRedis(); + + const stored = await primeCache('categoryStats', { total: 476_715 }, { client }); + assert.equal(stored, true); + + let ran = 0; + const got = await remember('categoryStats', { ttlMs: 5 * 60_000, client }, async () => { + ran += 1; + throw new Error('a reader must never have to run this'); + }); + + assert.deepEqual(got, { total: 476_715 }); + assert.equal(ran, 0, 'the reader did not touch the database'); +}); + +test('priming without a client is a no-op rather than a crash', async () => { + // The poller runs with no REDIS_URL locally and in the test suite. + assert.equal(await primeCache('k', { a: 1 }, { client: null }), false); +}); + +test('bigint counts survive the round trip', async () => { + // libSQL hands back BigInt for some aggregates and JSON.stringify throws on + // it, which would silently disable the cache for exactly the count-heavy + // reads it exists to serve. + const client = fakeRedis(); + await remember('k', { ttlMs: 60_000, client }, async () => ({ feeds: 10n })); + const again = await remember('k', { ttlMs: 60_000, client }, async () => ({ feeds: 0 })); + assert.deepEqual(again, { feeds: 10 }); +});