From 02b978adf90fb85fa2f6d9f34e47084a8c40f638 Mon Sep 17 00:00:00 2001 From: Aswin C Date: Fri, 23 Oct 2026 01:54:13 +0530 Subject: [PATCH] fix(sync): refresh stage 3 candidates only when tags are missing --- scripts/README.md | 2 +- scripts/sync/03-sync-episodes.js | 40 ++++----------- scripts/sync/lib/staleness.js | 40 +++++++++++++++ scripts/sync/lib/staleness.test.js | 78 ++++++++++++++++++++++++++++++ scripts/sync/lib/state.js | 3 ++ 5 files changed, 131 insertions(+), 32 deletions(-) diff --git a/scripts/README.md b/scripts/README.md index af3c2e7e4..0d1e68684 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -20,7 +20,7 @@ A Git push does not deploy this pipeline. Changes to `scripts/sync/` or `scripts | [`scripts/sync/lib/vectorize-lanes.js`](sync/lib/vectorize-lanes.js) | Tip-first drain order + partial-complete flag rules for stage 4 | | [`scripts/sync/lib/pi-handoff.js`](sync/lib/pi-handoff.js) | Same-run episode payloads stage 3→4 (not cross-run durable) | | [`scripts/sync/lib/turso.js`](sync/lib/turso.js) + [`turso-page.js`](sync/lib/turso-page.js) | HTTP Turso client; **always page** large SELECTs (`RESPONSE_TOO_LARGE`). Digit-only strings bind as integers by default; wrap catalog keys and other opaque IDs in `turso.text()` to keep them text. Objects and arrays are refused instead of being stored as `[object Object]`; `JSON.stringify` structured values on purpose. `atomic()` runs statements all-or-nothing | -| [`scripts/sync/lib/staleness.js`](sync/lib/staleness.js) | Core vs relaxed episode-check windows | +| [`scripts/sync/lib/staleness.js`](sync/lib/staleness.js) | Core vs relaxed episode-check windows, and when stage 3 re-reads its cached candidate list from Turso (`candidateRefreshReason`): no list, charts refreshed since, the list is over 20 hours old, the core countries differ from the state's `relaxedTagsCore` (the list the relaxed `x` tags were computed with, recorded by each refresh), or, with the check cadence on, more than 5% of candidates (at least 50) have no iTunes ID. Core shows carry no `x`, so a missing tag alone never triggers a refresh. Otherwise the run reads no candidates from Turso | | [`scripts/sync/lib/episode-caps.js`](sync/lib/episode-caps.js) | Per-storefront episode vector caps (us/gb 50, else 20), plus earned caps: a show keeps the highest cap it had while its vectors are kept (`shows[id].k` in the sync state), so leaving the US/GB charts doesn't delete and later re-embed 30 vectors | | [`scripts/sync/lib/trim-episode-caps.js`](sync/lib/trim-episode-caps.js) | Nightly per-show trim (stage 6 and `remediate-qdrant.js`). Uses the larger of the chart cap and the earned cap, and **fails closed**: if the cap lookup fails, nothing is trimmed. Returns the trimmed show IDs, which `remediate-qdrant.js` re-flags for healing | | [`scripts/sync/lib/vector-window.js`](sync/lib/vector-window.js) | Stage 4's window: shows with no vectors get their newest `NEW_ENTRANT_EPISODES` (5) first and the rest after `NEW_ENTRANT_DEPTH_DAYS` (7; `shows[id].d`, cleared only after the show's re-flag is stored, so a failed write retries next run); pruning only against a full window; stage 3 never hands stage 4 a one-item fallback | diff --git a/scripts/sync/03-sync-episodes.js b/scripts/sync/03-sync-episodes.js index 97ba5f3b3..861a303d3 100644 --- a/scripts/sync/03-sync-episodes.js +++ b/scripts/sync/03-sync-episodes.js @@ -4,7 +4,8 @@ /** * Stage 3: Sync latest-episode metadata for chart shows from the PI API. * - * - Candidate list cached in sync state; re-queried after charts refresh or 20h. + * - Candidate list cached in sync state; re-queried after charts refresh, after 20h, or when the + * core countries differ from those its relaxed tags were computed with (lib/staleness.js). * Countries come from a one-shot charts pre-agg (see lib/chart-countries.js), * not a per-podcast correlated subquery. * - Staleness: News 8h · core countries 24h · relaxed (secondary) countries 48h @@ -39,11 +40,10 @@ const staleTips = require('./lib/stale-tips'); const showWindow = require('./lib/show-window'); const cadence = require('./lib/cadence'); -const CANDIDATE_CACHE_MAX_AGE_MS = 20 * 60 * 60 * 1000; const CONCURRENCY = 5; -async function refreshCandidates(st) { - log.info('[CANDIDATES] Refreshing candidate list from Turso'); +async function refreshCandidates(st, reason) { + log.info(`[CANDIDATES] Refreshing candidate list from Turso (${reason})`); // Pre-aggregate charts once (~one linear scan). Never use a correlated // CAST(c.itunes_id AS INTEGER) subquery — that was multi-billion rows_read. const countriesByItunes = await loadCountriesByItunesId(turso); @@ -84,6 +84,7 @@ async function refreshCandidates(st) { } st.candidateIds = ids; st.candidatesRefreshedAt = Date.now(); + st.relaxedTagsCore = staleness.relaxedTagsKey(); const coreRegular = ids.length - newsCount - relaxedCount; log.info( `[CANDIDATES] ${log.fmt(ids.length)} chart shows cached · ` + @@ -151,33 +152,10 @@ async function main() { }); const cacheAge = Date.now() - (st.candidatesRefreshedAt || 0); - const chartsNewerThanCache = (st.chartsRefreshedAt || 0) > (st.candidatesRefreshedAt || 0); - let needRefresh = !st.candidateIds || chartsNewerThanCache || cacheAge > CANDIDATE_CACHE_MAX_AGE_MS; - if (!needRefresh && st.candidateIds && cfg.RELAXED_CHECK_COUNTRIES.length > 0) { - let missingX = 0; - for (const id of st.candidateIds) { - const rec = st.shows[id]; - if (rec && rec.x === undefined && rec.n !== 1) missingX++; - } - // One-shot: apply relaxed(48h) tags without waiting for charts / 20h cache expiry - if (missingX > Math.max(50, st.candidateIds.length * 0.05)) { - log.info( - `[CANDIDATES] Refreshing to apply relaxed(48h) tags ` + - `(${log.fmt(missingX)} non-news shows untagged)`, - ); - needRefresh = true; - } - } - if (!needRefresh && st.candidateIds && cfg.CADENCE_MODE !== 'off') { - // One-shot: the cadence reads Apple's dates by iTunes ID, which only a refresh records. - const missingI = st.candidateIds.filter((id) => !st.shows[id]?.i).length; - if (missingI > Math.max(50, st.candidateIds.length * 0.05)) { - log.info(`[CANDIDATES] Refreshing to record iTunes IDs (${log.fmt(missingI)} shows without one)`); - needRefresh = true; - } - } + const refreshReason = staleness.candidateRefreshReason(st, { cadence: cfg.CADENCE_MODE !== 'off' }); + const needRefresh = refreshReason !== null; if (needRefresh) { - await refreshCandidates(st); + await refreshCandidates(st, refreshReason); } else { log.info(`[CANDIDATES] Using cached list (${log.fmt(st.candidateIds.length)} shows, age ${Math.round(cacheAge / 3600000)}h) - 0 Turso reads`); } @@ -187,7 +165,7 @@ async function main() { let shadow = null; let appleError = null; if (cfg.CADENCE_MODE === 'shadow') { - if (needRefresh || Date.now() - (st.cadenceSeededAt || 0) > CANDIDATE_CACHE_MAX_AGE_MS) { + if (needRefresh || Date.now() - (st.cadenceSeededAt || 0) > staleness.CANDIDATE_CACHE_MAX_AGE_MS) { try { const seeded = await cadence.seedFromBook(book, st.shows, st.candidateIds); st.cadenceSeededAt = Date.now(); diff --git a/scripts/sync/lib/staleness.js b/scripts/sync/lib/staleness.js index c34d9776c..ce84572c0 100644 --- a/scripts/sync/lib/staleness.js +++ b/scripts/sync/lib/staleness.js @@ -9,11 +9,14 @@ * relaxed — only on relaxed check countries, non-News → RELAXED_STALE_MS (48h) * * State flags on show rec: n=1 (news), x=1 (relaxed-only). Core wins over relaxed. + * + * Also decides when stage 3 re-reads its cached candidate list (candidateRefreshReason). */ const cfg = require('./config'); const CORE_SET = new Set(cfg.CORE_CHECK_COUNTRIES.map((c) => String(c).toLowerCase())); +const CANDIDATE_CACHE_MAX_AGE_MS = 20 * 60 * 60 * 1000; function hoursLabel(ms) { const h = ms / 3_600_000; @@ -61,6 +64,40 @@ function applyCountryCheckFlag(rec, countryCsvOrList) { return rec; } +/** + * The core-country list the x flags were computed with, stored as the state's `relaxedTagsCore` + * by each candidate refresh. + * @param {string[]} [coreCountries] + */ +function relaxedTagsKey(coreCountries = cfg.CORE_CHECK_COUNTRIES) { + return [...new Set(coreCountries.map((c) => String(c).toLowerCase()))].sort().join(','); +} + +/** + * Why stage 3 must re-read its candidates from Turso (a charts pre-aggregate plus a paged podcasts + * read), or null when the cached list will do. + * @param {object} st sync state + * @param {{ now?: number, cadence?: boolean, tagsKey?: string }} [opts] cadence: the check cadence + * is on, so candidates need iTunes IDs + * @returns {string|null} + */ +function candidateRefreshReason(st, { now = Date.now(), cadence = false, tagsKey = relaxedTagsKey() } = {}) { + if (!st.candidateIds) return 'no cached list'; + const refreshedAt = st.candidatesRefreshedAt || 0; + if ((st.chartsRefreshedAt || 0) > refreshedAt) return 'charts refreshed since'; + if (now - refreshedAt > CANDIDATE_CACHE_MAX_AGE_MS) return 'cache expired'; + // Core shows carry no x, so a missing x can't tell "core" from "never tagged"; the key can. + if (st.relaxedTagsCore !== tagsKey) return 'apply relaxed(48h) tags'; + if (cadence) { + // The cadence reads Apple's dates by iTunes ID, which only a refresh records. + const missing = st.candidateIds.filter((id) => !st.shows?.[id]?.i).length; + if (missing > Math.max(50, st.candidateIds.length * 0.05)) { + return `record iTunes IDs (${missing.toLocaleString()} shows without one)`; + } + } + return null; +} + /** @returns {'news'|'core'|'relaxed'} */ function checkTier(rec) { if (rec?.n === 1) return 'news'; @@ -137,9 +174,12 @@ module.exports = { policySummary, isRelaxedOnly, applyCountryCheckFlag, + relaxedTagsKey, + candidateRefreshReason, checkTier, jitterFactor, thresholdMs, planDue, CORE_SET, + CANDIDATE_CACHE_MAX_AGE_MS, }; diff --git a/scripts/sync/lib/staleness.test.js b/scripts/sync/lib/staleness.test.js index 72982dea6..adb166fbf 100644 --- a/scripts/sync/lib/staleness.test.js +++ b/scripts/sync/lib/staleness.test.js @@ -6,12 +6,39 @@ const cfg = require('./config'); const { isRelaxedOnly, applyCountryCheckFlag, + relaxedTagsKey, + candidateRefreshReason, + CANDIDATE_CACHE_MAX_AGE_MS, checkTier, thresholdMs, planDue, policySummary, } = require('./staleness'); +const NOW = 1_792_700_000_000; + +/** A freshly refreshed cache: `core` core shows, `news` news shows, `relaxed` relaxed-only shows. */ +function cachedState({ core = 0, news = 0, relaxed = 0, tagsKey = relaxedTagsKey() } = {}) { + const shows = {}; + const ids = []; + const add = (rec) => { + const id = String(ids.length + 1); + ids.push(id); + shows[id] = { c: NOW - 1000, i: 900000 + ids.length, ...rec }; + }; + for (let k = 0; k < core; k++) add({}); + for (let k = 0; k < news; k++) add({ n: 1 }); + for (let k = 0; k < relaxed; k++) add({ x: 1 }); + return { + version: 2, + candidatesRefreshedAt: NOW - 60 * 60 * 1000, + chartsRefreshedAt: NOW - 2 * 60 * 60 * 1000, + relaxedTagsCore: tagsKey, + candidateIds: ids, + shows, + }; +} + test('core vs relaxed country sets are disjoint and cover ALL', () => { const core = new Set(cfg.CORE_CHECK_COUNTRIES); const relaxed = new Set(cfg.RELAXED_CHECK_COUNTRIES); @@ -92,3 +119,54 @@ test('policySummary mentions all three windows', () => { assert.ok(p.coreCountries.includes('us')); assert.ok(p.relaxedCountries.includes('de')); }); + +test('relaxedTagsKey: the sorted, lowercased core list', () => { + assert.equal(relaxedTagsKey(['US', 'gb', 'us']), 'gb,us'); + assert.equal(relaxedTagsKey(), relaxedTagsKey([...cfg.CORE_CHECK_COUNTRIES].reverse())); +}); + +test('candidate refresh: core shows without x keep the cached list (regression: refreshed every run)', () => { + // Shaped like the runner's cache: about 10,400 core shows, and core shows carry no x. + const st = cachedState({ core: 10_400, news: 235, relaxed: 15_400 }); + const untagged = st.candidateIds.filter((id) => st.shows[id].x === undefined && st.shows[id].n !== 1).length; + assert.ok(untagged > st.candidateIds.length * 0.05, 'the old count called these untagged'); + assert.equal(candidateRefreshReason(st, { now: NOW }), null); + assert.equal(candidateRefreshReason(st, { now: NOW, cadence: true }), null); +}); + +test('candidate refresh: a cache never tagged, or tagged for other core countries, refreshes once', () => { + const st = cachedState({ core: 300, relaxed: 40 }); + delete st.relaxedTagsCore; + assert.equal(candidateRefreshReason(st, { now: NOW }), 'apply relaxed(48h) tags'); + assert.equal( + candidateRefreshReason(cachedState({ core: 10, tagsKey: 'gb,us' }), { now: NOW, tagsKey: 'fr,gb,us' }), + 'apply relaxed(48h) tags', + ); + + // What the refresh does: tag every candidate and record the key. The next run uses the cache. + for (const id of st.candidateIds) applyCountryCheckFlag(st.shows[id], st.shows[id].x ? 'de' : 'us,de'); + st.candidatesRefreshedAt = NOW; + st.relaxedTagsCore = relaxedTagsKey(); + assert.equal(candidateRefreshReason(st, { now: NOW + 1000 }), null); +}); + +test('candidate refresh: the cadence still refreshes when most candidates lack an iTunes ID', () => { + const st = cachedState({ core: 1000 }); + for (const id of st.candidateIds.slice(0, 60)) delete st.shows[id].i; + assert.match(candidateRefreshReason(st, { now: NOW, cadence: true }), /^record iTunes IDs \(60 shows/); + assert.equal(candidateRefreshReason(st, { now: NOW }), null, 'cadence off: iTunes IDs not needed'); + for (const id of st.candidateIds.slice(0, 10)) st.shows[id].i = 1; + assert.equal(candidateRefreshReason(st, { now: NOW, cadence: true }), null, '50 or fewer missing: cached'); +}); + +test('candidate refresh: no list, newer charts and an expired cache still refresh', () => { + const st = cachedState({ core: 5 }); + assert.equal(candidateRefreshReason({ version: 2, shows: {} }, { now: NOW }), 'no cached list'); + assert.equal(candidateRefreshReason({ ...st, chartsRefreshedAt: NOW - 1000 }, { now: NOW }), 'charts refreshed since'); + assert.equal( + candidateRefreshReason(st, { now: st.candidatesRefreshedAt + CANDIDATE_CACHE_MAX_AGE_MS + 1 }), + 'cache expired', + ); + // Stage 2 forces a refresh by zeroing the refresh time. + assert.notEqual(candidateRefreshReason({ ...st, candidatesRefreshedAt: 0 }, { now: NOW }), null); +}); diff --git a/scripts/sync/lib/state.js b/scripts/sync/lib/state.js index a47167566..3d7b1428c 100644 --- a/scripts/sync/lib/state.js +++ b/scripts/sync/lib/state.js @@ -8,6 +8,9 @@ * "version": 2, * "candidatesRefreshedAt": , * "chartsRefreshedAt": , + * "relaxedTagsCore": "", + * "cadenceSeededAt": , * "candidateIds": ["", ...], * "shows": { * "": {