Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions scripts/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@ A Git push does not deploy this pipeline. Changes to `scripts/sync/` or `scripts
| [`scripts/sync/bootstrap-book.js`](sync/bootstrap-book.js) + [`lib/book-bootstrap.js`](sync/lib/book-bootstrap.js) | One resumable pass over every show with episode points: maps each point's payload ID (its Podcast Index ID) to a catalog key through PI's episode list, writes the show's book rows in one transaction (served ID = that PI ID), then stamps `eid` and `catalog_key` on the points (`eid_missing` when unmapped). A point whose Podcast Index ID another key already serves gets no row, so the show's transaction can't fail on it: it is stamped with that key's identity when it is in the same show, or `eid_shared` with the owning show for a duplicate feed. Point IDs and vectors never change. `--dry`, `--limit-shows N`, `--minutes N`; progress in `SYNC_BOOTSTRAP_STATE`. `--catch-up` takes only shows that still have points with none of `eid`, `eid_missing` and `eid_shared`. `--retry-missing` (own-episode-IDs P3; monthly on the VPS) runs the pass again for shows with `eid_missing` points: a point mapped this time gets its stamp and loses the mark |
| [`scripts/sync/lib/charted-at.js`](sync/lib/charted-at.js) + [`stamp-charted-at.js`](sync/stamp-charted-at.js) | Last-charted dates on Qdrant points (own-episode-IDs P3b): every point in both collections carries `charted_at`, the UTC day (epoch seconds) its show was last seen on a chart, so a 90-day vector retention can outlive the show's catalog row and state record. Stage 6 stamps every show it keeps from its chart clock (`shows[id].s`) before deleting; a point is rewritten only when it has no date or one more than `CHARTED_AT_REFRESH_DAYS` (default 7) older than its show's, so dates only move forward and a charting show's points are rewritten about weekly. Stages 4 and 5 give new points the day they are embedded. Failures are logged and never block the cleanup. `stamp-charted-at.js` stamps every point once from the catalog and the chart clock without changing anything else (`--dry` counts only; run from the repository root under the shared job lock). Nothing deletes by this date yet |
| [`scripts/sync/lib/retention.js`](sync/lib/retention.js) | Vector retention after a show leaves the charts (own-episode-IDs P3b, decided 24 Oct 2026): each departed show keeps its newest `ARCHIVE_EPISODES` (5) episode points and its show vector for `RETENTION_DAYS` (90) after it was last on a chart, read from `charted_at`; its catalog row still goes after the 7-day grace. `RETENTION_MODE` (stage 6): `count` by default keeps today's deletes and adds what `on` would keep, delete and expire to the report; `on` keeps the newest episodes instead of deleting the show's vectors, stamps their exact last chart day, and deletes points (both collections) last charted more than `RETENTION_DAYS` ago; `off` reports nothing. Size guard on episode points: over `QDRANT_POINTS_ALERT` (1.8M) the run warns; over `QDRANT_POINTS_CEILING` (2.0M) retention steps down to 60 then 30 days, and an urgent alert goes out through `SYNC_NOTIFY_SCRIPT` while it stays over (resolved once under). Fails safe: expiry is held back (counted only) in a run whose date stamps failed; a departed show whose kept points couldn't be dated keeps its catalog row and chart clock until a later run succeeds; with `on`, a retention failure fails the stage. Numeric settings accept whole numbers only. Attribution logs kept episodes on `show-` (`kept`) and expired points as `exp` (`retention` in the report), written only after Qdrant confirms the delete. Qdrant's memory limit is 6 GB for this |
| [`scripts/sync/lib/run-budget.js`](sync/lib/run-budget.js) | Per-run time budget for the VPS runner's hourly runs (own-episode-IDs P3b). Once a run holds the shared job lock, the runner passes `SYNC_STAGE3_DEADLINE_AT` (half the budget) and `SYNC_DEADLINE_AT` (the whole budget) as epoch seconds; stage 3 stops starting checks and stage 4 stops starting shows when theirs passes, and the rest waits for the next run, so runs stay even. Both say so in the run report. Unset means no limit (manual runs). Stages 3 and 4 append one JSON line of counts and timings to `SYNC_METRICS_FILE` for the runner's usage log |
| [`scripts/sync/lib/episode-recipe-id.js`](sync/lib/episode-recipe-id.js) | boxlore episode IDs from (show ID, catalog key); must match the app's `EpisodeRecipeId` and the shared fixture in `core/rss/src/test/resources/fixtures/` |
| [`scripts/package.json`](package.json) | Sync Node deps + `npm run test:sync` |

Expand Down
2 changes: 1 addition & 1 deletion scripts/package.json
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
"test:sync-scalars": "node --test sync/lib/scalars.test.js",
"test:sync-turso-page": "node --test sync/lib/turso-page.test.js",
"test:sync-turso": "node --test sync/lib/turso.test.js",
"test:sync": "node --test sync/lib/csv.test.js sync/lib/pi-unavailable.test.js sync/lib/scalars.test.js sync/lib/turso-page.test.js sync/lib/episode-caps.test.js sync/lib/staleness.test.js sync/lib/podcast-index.test.js sync/lib/chart-countries.test.js sync/lib/log.test.js sync/lib/text.test.js sync/lib/vectorize-lanes.test.js sync/lib/tip-queue.test.js sync/lib/import-missing.test.js sync/lib/episode-recipe-id.test.js sync/lib/attribution.test.js sync/lib/vector-window.test.js sync/lib/trim-episode-caps.test.js sync/lib/remediate-qdrant.test.js sync/lib/turso.test.js sync/lib/catalog-rows.test.js sync/lib/episode-book.test.js sync/lib/book-bootstrap.test.js sync/lib/episode-identity.test.js sync/lib/stale-tips.test.js sync/lib/show-window.test.js sync/lib/cadence.test.js sync/lib/charted-at.test.js sync/lib/retention.test.js",
"test:sync": "node --test sync/lib/csv.test.js sync/lib/pi-unavailable.test.js sync/lib/scalars.test.js sync/lib/turso-page.test.js sync/lib/episode-caps.test.js sync/lib/staleness.test.js sync/lib/podcast-index.test.js sync/lib/chart-countries.test.js sync/lib/log.test.js sync/lib/text.test.js sync/lib/vectorize-lanes.test.js sync/lib/tip-queue.test.js sync/lib/import-missing.test.js sync/lib/episode-recipe-id.test.js sync/lib/attribution.test.js sync/lib/vector-window.test.js sync/lib/trim-episode-caps.test.js sync/lib/remediate-qdrant.test.js sync/lib/turso.test.js sync/lib/catalog-rows.test.js sync/lib/episode-book.test.js sync/lib/book-bootstrap.test.js sync/lib/episode-identity.test.js sync/lib/stale-tips.test.js sync/lib/show-window.test.js sync/lib/cadence.test.js sync/lib/charted-at.test.js sync/lib/retention.test.js sync/lib/run-budget.test.js",
"test:sync-text": "node --test sync/lib/text.test.js",
"test:sync-recipe": "node --test sync/lib/episode-recipe-id.test.js",
"test:sync-attribution": "node --test sync/lib/attribution.test.js",
Expand Down
32 changes: 29 additions & 3 deletions scripts/sync/03-sync-episodes.js
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@ const staleness = require('./lib/staleness');
const staleTips = require('./lib/stale-tips');
const showWindow = require('./lib/show-window');
const cadence = require('./lib/cadence');
const runBudget = require('./lib/run-budget');

const CONCURRENCY = 5;

Expand Down Expand Up @@ -122,6 +123,9 @@ async function countDueByCountry(ids) {
}

async function main() {
const startedMs = Date.now();
// Hourly runs: stop starting checks at the runner's deadline (lib/run-budget.js).
const deadline = runBudget.stageDeadline(runBudget.STAGE3_DEADLINE);
turso.assertEnv();
const book = bookDb(turso);
pi.assertEnv();
Expand Down Expand Up @@ -251,7 +255,13 @@ async function main() {
}
const prog = log.progress(due.length, 'episode-sync', 5);

let timeLeftOver = 0;
for (let i = 0; i < due.length; i += CONCURRENCY) {
if (runBudget.timeUp(deadline)) {
timeLeftOver = due.length - i;
Comment thread
coderabbitai[bot] marked this conversation as resolved.
log.warn(`[BUDGET] Time budget reached: ${log.fmt(timeLeftOver)} checks wait for the next run`);
break;
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
const batch = due.slice(i, i + CONCURRENCY);
let tipUpdatedThisBatch = false;
await Promise.all(batch.map(async (podId) => {
Expand Down Expand Up @@ -342,9 +352,14 @@ async function main() {
state.save(st);
await handoff.flush();

// Checks that ran this run. When time ran out, R5 reads feeds only for those, until the run's
// deadline (not stage 3's half: a checked stale show skipped here would wait a whole tier). When
// Podcast Index paused, R5 still covers every due show, as the feed is then the only source.
const attempted = piPaused ? checkedBeforePause : due.length - timeLeftOver;
const r5 = await staleTips.runStaleTips({
mode: cfg.R5_MODE,
due,
due: timeLeftOver ? due.slice(0, attempted) : due,
deadlineMs: runBudget.deadlineFrom(runBudget.RUN_DEADLINE),
db: turso,
feedsUrl: cfg.FEEDS_URL,
registryPath: cfg.REGISTRY_DB,
Expand Down Expand Up @@ -428,14 +443,25 @@ async function main() {
`${empty} empty, ${errors} errors · due news=${staleNews} core=${staleCore} relaxed=${staleRelaxed}` +
` · ${pi.summary()}` +
(piPaused ? ` · ${due.length - checkedBeforePause} checks left for the next run` : '') +
(timeLeftOver ? ` · time budget: ${timeLeftOver} checks left for the next run` : '') +
(cfg.R5_MODE !== 'off' ? ` · ${r5Line}` : '') +
(windowLine ? ` · ${windowLine}` : '') +
(cadenceLine ? ` · ${cadenceLine}` : ''),
}]);

runBudget.recordMetrics('03-sync-episodes', {
due: allDue.length,
checked: updated + unchanged + empty + errors,
newTips: updated,
errors,
leftForNextRun: deferred + timeLeftOver + (piPaused ? due.length - checkedBeforePause : 0),
timeUp: timeLeftOver > 0,
ms: Date.now() - startedMs,
});

// A pause is "try later", not a failure; only real errors over half the checks fail the stage.
if (due.length > 20 && errors > due.length / 2) {
log.error(`More than half of episode checks failed (${errors}/${due.length})`);
if (attempted > 20 && errors > attempted / 2) {
log.error(`More than half of episode checks failed (${errors}/${attempted})`);
process.exit(1);
}
}
Expand Down
28 changes: 28 additions & 0 deletions scripts/sync/04-vectorize-episodes.js
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@ const scalars = require('./lib/scalars');
const showWindow = require('./lib/show-window');
const attribution = require('./lib/attribution');
const chartedAt = require('./lib/charted-at');
const runBudget = require('./lib/run-budget');

const UPSERT_BATCH = 100;

Expand All @@ -59,6 +60,9 @@ async function ensureIndexes() {

async function main() {
const startedS = Math.floor(Date.now() / 1000);
// Hourly runs: stop starting shows at the runner's deadline (lib/run-budget.js).
const deadline = runBudget.deadlineFrom(runBudget.RUN_DEADLINE);
let timeUp = false;
turso.assertEnv();
qdrant.assertEnv();
pi.assertEnv();
Expand Down Expand Up @@ -184,6 +188,9 @@ async function main() {
'Collection': cfg.EPISODES_COLLECTION,
});
if (drainOrder.length === 0) {
runBudget.recordMetrics('04-vectorize-episodes', {
shows: 0, embedded: 0, stored: 0, backlog: 0, errors: 0, timeUp: false, ms: Date.now() - startedS * 1000,
});
state.save(st);
log.summaryTable('Stage 4: Vectorize Episodes', [{
stage: 'vectorize-episodes', reads: turso.getStats().reads,
Expand Down Expand Up @@ -275,6 +282,16 @@ async function main() {
}

for (const pod of drainOrder) {
if (runBudget.timeUp(deadline)) {
timeUp = true;
runProg.flush();
reportBacklogIfDue(true);
log.info(
`[BUDGET] Time budget reached (${embedded} embedded). ` +
`Remaining backlog: ${drainOrder.length - processedCount} shows wait for the next run`,
);
break;
}
if (budget <= 0) {
runProg.flush();
reportBacklogIfDue(true);
Expand Down Expand Up @@ -627,11 +644,22 @@ async function main() {
`depth ${depthDue.length}, spooled ${spooledPoints}, replayed ${replayed.points}, ` +
`${showsCompleted} done, partial ${partialShows}, skip ${showsFullySkipped}, ` +
`handoff ${handoffHits}, PI ${piFetches}, backlog ${backlog}, ${errors} errors` +
(timeUp ? ', time budget reached' : '') +
(cfg.WINDOWS_MODE !== 'off' ? `, ${windowSummary()}` : '') +
`, ${pi.summary()}` + (waitingForPi ? `, ${waitingForPi} shows wait for PI` : '') +
(reasons ? `, ${reasons}` : ''),
}]);

runBudget.recordMetrics('04-vectorize-episodes', {
Comment thread
ashwkun marked this conversation as resolved.
shows: processedCount,
embedded,
stored,
backlog,
errors,
timeUp,
ms: Date.now() - startedS * 1000,
});

if (processedCount > 20 && errors > processedCount) {
log.error('Error count exceeds processed shows - failing run for visibility');
process.exit(1);
Expand Down
50 changes: 50 additions & 0 deletions scripts/sync/lib/run-budget.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,50 @@
'use strict';

/**
* Per-run time budget and metrics (own-episode-IDs P3b, hourly runs).
*
* The VPS runner (netcup-panel `boxlore-sync/run-sync.sh`) gives each run a time budget once it
* holds the shared job lock: 30 minutes for hourly runs, 45 for the nightly chart and cleanup
* runs. It passes two deadlines as epoch seconds:
* SYNC_STAGE3_DEADLINE_AT — stage 3 stops checking shows (half the budget);
* SYNC_DEADLINE_AT — stage 4 stops embedding.
* Work left over (shows still due, backlog) waits for the next run, so runs stay even instead of
* one long run after a quiet one. Unset deadlines mean no limit (manual runs, CI).
*
* Stages append one JSON line of counts to SYNC_METRICS_FILE for the runner's usage log.
*/

const fs = require('fs');

const STAGE3_DEADLINE = 'SYNC_STAGE3_DEADLINE_AT';
const RUN_DEADLINE = 'SYNC_DEADLINE_AT';

/** Deadline in epoch ms from an env var holding epoch seconds; Infinity when unset or invalid. */
function deadlineFrom(name, env = process.env) {
const raw = String(env[name] ?? '').trim();
const s = /^[0-9]+$/.test(raw) ? Number(raw) : 0;
return s > 0 ? s * 1000 : Infinity;
}

/** The earlier of the stage's own deadline and the run's. */
function stageDeadline(stageVar, env = process.env) {
return Math.min(deadlineFrom(stageVar, env), deadlineFrom(RUN_DEADLINE, env));
}

function timeUp(deadlineMs, now = Date.now()) {
return now >= deadlineMs;
}

/** Appends one metrics line for the runner's usage log. Never throws. */
function recordMetrics(stage, data, env = process.env) {
const file = env.SYNC_METRICS_FILE;
if (!file) return false;
try {
fs.appendFileSync(file, `${JSON.stringify({ stage, t: Math.floor(Date.now() / 1000), ...data })}\n`);
return true;
} catch {
return false;
}
}

module.exports = { STAGE3_DEADLINE, RUN_DEADLINE, deadlineFrom, stageDeadline, timeUp, recordMetrics };
45 changes: 45 additions & 0 deletions scripts/sync/lib/run-budget.test.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,45 @@
'use strict';

const test = require('node:test');
const assert = require('node:assert/strict');
const fs = require('node:fs');
const os = require('node:os');
const path = require('node:path');
const { STAGE3_DEADLINE, RUN_DEADLINE, deadlineFrom, stageDeadline, timeUp, recordMetrics } = require('./run-budget');

test('deadlines come from epoch seconds; unset or malformed means no limit', () => {
assert.equal(deadlineFrom(RUN_DEADLINE, { [RUN_DEADLINE]: '1792800000' }), 1_792_800_000_000);
assert.equal(deadlineFrom(RUN_DEADLINE, { [RUN_DEADLINE]: ' 1792800000 ' }), 1_792_800_000_000);
for (const bad of [undefined, '', '0', '1.5e9', 'soon', '-5']) {
assert.equal(deadlineFrom(RUN_DEADLINE, { [RUN_DEADLINE]: bad }), Infinity, String(bad));
}
});

test('a stage stops at the earlier of its own deadline and the run\'s', () => {
assert.equal(stageDeadline(STAGE3_DEADLINE, { [STAGE3_DEADLINE]: '100', [RUN_DEADLINE]: '200' }), 100_000);
assert.equal(stageDeadline(STAGE3_DEADLINE, { [STAGE3_DEADLINE]: '300', [RUN_DEADLINE]: '200' }), 200_000);
assert.equal(stageDeadline(STAGE3_DEADLINE, { [RUN_DEADLINE]: '200' }), 200_000);
assert.equal(stageDeadline(STAGE3_DEADLINE, {}), Infinity);
});

test('timeUp at and after the deadline only', () => {
assert.equal(timeUp(1000, 999), false);
assert.equal(timeUp(1000, 1000), true);
assert.equal(timeUp(Infinity), false);
});

test('metrics: one JSON line per call; no file configured means nothing written', () => {
const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'run-metrics-'));
const file = path.join(dir, 'm.jsonl');
try {
assert.equal(recordMetrics('03-sync-episodes', { checked: 5, timeUp: true }, { SYNC_METRICS_FILE: file }), true);
assert.equal(recordMetrics('04-vectorize-episodes', { embedded: 2 }, { SYNC_METRICS_FILE: file }), true);
const lines = fs.readFileSync(file, 'utf8').trim().split('\n').map((l) => JSON.parse(l));
assert.deepEqual(lines.map((l) => [l.stage, l.checked ?? l.embedded]), [['03-sync-episodes', 5], ['04-vectorize-episodes', 2]]);
assert.equal(typeof lines[0].t, 'number');
assert.equal(recordMetrics('x', {}, {}), false);
assert.equal(recordMetrics('x', {}, { SYNC_METRICS_FILE: path.join(dir, 'missing', 'm.jsonl') }), false, 'never throws');
} finally {
fs.rmSync(dir, { recursive: true, force: true });
}
});
14 changes: 11 additions & 3 deletions scripts/sync/lib/stale-tips.js
Original file line number Diff line number Diff line change
Expand Up @@ -98,13 +98,15 @@ function feedIdentity(showId, item) {
* couldn't be read (R5 skipped).
* @param {{ mode: 'off'|'shadow'|'on', due: (string|number)[], db: { execute: Function, rows: Function },
* feedsUrl: string, registryPath: string, mediumOf?: (id: number) => string, now?: () => number,
* fetchImpl?: Function, staleIds?: () => Set<number>, warn?: (msg: string) => void }} opts
* fetchImpl?: Function, staleIds?: () => Set<number>, warn?: (msg: string) => void, deadlineMs?: number }} opts
* deadlineMs: no new chunk starts after it (hourly runs, lib/run-budget.js); the rest waits.
*/
async function runStaleTips(opts) {
const { mode, due, db, feedsUrl, registryPath, fetchImpl = fetch, warn = () => {} } = opts;
const mediumOf = opts.mediumOf || (() => 'podcast');
const now = opts.now || Date.now;
const out = { mode, stale: 0, answered: 0, newer: 0, updated: 0, noItem: 0, failedChunks: 0, error: null, updatedIds: [], tipDates: {} };
const out = { mode, stale: 0, answered: 0, newer: 0, updated: 0, noItem: 0, failedChunks: 0, error: null, updatedIds: [], tipDates: {}, timeLeft: 0 };
const deadlineMs = opts.deadlineMs ?? Infinity;
if (mode !== 'shadow' && mode !== 'on') return out;

let flagged;
Expand All @@ -128,6 +130,11 @@ async function runStaleTips(opts) {
return out;
}
for (let i = 0; i < ids.length; i += CHUNK) {
if (Date.now() >= deadlineMs) {
out.timeLeft = ids.length - i;
warn(`[R5] time budget reached: ${out.timeLeft} stale shows wait for the next run`);
break;
}
const chunk = ids.slice(i, i + CHUNK);
let results;
try {
Expand Down Expand Up @@ -187,7 +194,8 @@ function summary(r) {
if (r.error) return `R5 ${r.mode}: skipped (${r.error})`;
return r.mode === 'shadow'
? `R5 shadow: ${r.stale} stale due, ${r.answered} answered, ${r.newer} with a newer feed episode, ${r.failedChunks} failed chunks`
: `R5 on: ${r.stale} stale due, ${r.answered} answered, ${r.updated} tips from the feed, ${r.noItem} without a served ID yet, ${r.failedChunks} failed chunks`;
: `R5 on: ${r.stale} stale due, ${r.answered} answered, ${r.updated} tips from the feed, ${r.noItem} without a served ID yet, ${r.failedChunks} failed chunks` +
(r.timeLeft ? `, ${r.timeLeft} left for the next run` : '');
}

module.exports = { runStaleTips, staleShowIds, shouldReplace, feedIdentity, feedWindow, summary, TIP_SQL, CHUNK };
Loading
Loading