fix(sync): refresh stage 3 candidates only when tags are missing

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 |

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

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

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

scripts/sync/lib/state.js

@@ -8,6 +8,9 @@
  *   "version": 2,
  *   "candidatesRefreshedAt": <ms epoch of last DB candidate refresh>,
  *   "chartsRefreshedAt": <ms epoch of last charts refresh>,
+ *   "relaxedTagsCore": "<core check countries the x flags were computed with, e.g. fr,gb,in,us;
+ *                      stage 3 refreshes its candidates when this differs from the config>",
+ *   "cadenceSeededAt": <ms epoch of the last check-cadence pace seed from the episode book>,
  *   "candidateIds": ["<podId>", ...],
  *   "shows": {
  *     "<podId>": {