From a0ce01924c7c4aa8f5a01e390aafdb6bdb2079a6 Mon Sep 17 00:00:00 2001 From: Harlan GitHub Agent Date: Thu, 12 Nov 2026 12:54:14 +1100 Subject: [PATCH] chore(registry): prune finished durable jobs and embedding attempts nuxt-cf-jobs soft-completes a job, so every hourly sync left up to 250 rows in `jobs` forever, with payloads and retry reasons. D1 is a 10 GB store and reached its limit on 2026-11-04. On 2026-11-12 `jobs` held 146,816 rows, `failed_jobs` 2,672, `job_batches` 3,332, and `embedding_attempts` 25,578, none of them pruned. A new hourly task deletes completed jobs after 7 days, failed jobs and embedding attempts after 30 days, and finished batches once no job names them, in bounded index-seeking batches. Co-Authored-By: Claude Opus 5.5 --- docs/arch/cron.md | 3 +- .../server/tasks/prune-finished-jobs.ts | 55 ++++++ .../utils/finished-job-retention.spec.ts | 127 +++++++++++++ .../server/utils/finished-job-retention.ts | 171 ++++++++++++++++++ shared/schedule-policy.ts | 1 + 5 files changed, 356 insertions(+), 1 deletion(-) create mode 100644 layers/registry/server/tasks/prune-finished-jobs.ts create mode 100644 layers/registry/server/utils/finished-job-retention.spec.ts create mode 100644 layers/registry/server/utils/finished-job-retention.ts diff --git a/docs/arch/cron.md b/docs/arch/cron.md index 9d638159..da41ad4a 100644 --- a/docs/arch/cron.md +++ b/docs/arch/cron.md @@ -2,7 +2,7 @@ Generated from literal `defineScheduledTask` declarations. Run `pnpm cron:docs` after schedule changes. -Tasks: 24. Unique Cloudflare triggers: 15. +Tasks: 25. Unique Cloudflare triggers: 16. | Task | Cron | Source | | --- | --- | --- | @@ -16,6 +16,7 @@ Tasks: 24. Unique Cloudflare triggers: 15. | `detect-star-surges` | `30 4 * * *` | `layers/registry/server/tasks/detect-star-surges.ts` | | `drain-skill-dirty` | `*/5 * * * *` | `layers/registry/server/tasks/drain-skill-dirty.ts` | | `embedding-parity-audit` | `0 21 * * *` | `layers/registry/server/tasks/embedding-parity-audit.ts` | +| `prune-finished-jobs` | `40 * * * *` | `layers/registry/server/tasks/prune-finished-jobs.ts` | | `purge-personal-data` | `30 4 * * *` | `layers/identity/server/tasks/purge-personal-data.ts` | | `recompute-skill-scores` | `0 3 * * *` | `layers/registry/server/tasks/recompute-skill-scores.ts` | | `reconcile-rendered` | `20 */6 * * *` | `layers/registry/server/tasks/reconcile-rendered.ts` | diff --git a/layers/registry/server/tasks/prune-finished-jobs.ts b/layers/registry/server/tasks/prune-finished-jobs.ts new file mode 100644 index 00000000..b73f04f9 --- /dev/null +++ b/layers/registry/server/tasks/prune-finished-jobs.ts @@ -0,0 +1,55 @@ +/// + +import { resolveCloudflareBindings } from '@harlan-zw/nuxt-cloudflare/bindings' +import { runObservedScheduledTask } from '~~/server/utils/scheduled-run' +import { reportJobRun } from '~~/server/utils/sync-job-reporter' +import { observedSchedulePolicy } from '#shared/schedule-policy' +import { pruneFinishedJobs } from '../utils/finished-job-retention' + +const CRON = '40 * * * *' + +/** + * Scheduled task: delete finished durable job rows and old embedding attempts. + * + * nuxt-cf-jobs soft-completes a job, so every hourly sync left up to 250 + * `jobs` rows behind, with their payloads and retry reasons, forever. D1 is a + * 10 GB store and reached that limit on 2026-11-04. Append-only history needs + * a retention window, and this task is the window for the job tables. + * + * The minute avoids the top of the hour, when the sync creates its batch. + */ +export default defineScheduledTask({ + name: 'prune-finished-jobs', + cron: '40 * * * *', + description: 'Delete finished durable job rows and old embedding attempts', + async run({ context }) { + const env = resolveCloudflareBindings(context) + const db = env?.DB as D1Database | undefined + if (!env || !db) { + emitOperationalEvent(createWideEvent({ operation: 'prune-finished-jobs', outcome: 'binding-missing' })) + return { result: { error: 'no-db' } } + } + + return await runObservedScheduledTask({ + db, + env, + context, + policy: observedSchedulePolicy('prune-finished-jobs'), + }, async () => { + const startedAt = Date.now() + const report = await pruneFinishedJobs(db, { now: Math.floor(startedAt / 1000) }) + emitOperationalEvent(createWideEvent({ + 'operation': 'prune-finished-jobs', + 'outcome': 'completed', + 'success.count': report.completedJobs + report.failedJobs + report.finishedBatches + report.embeddingAttempts, + 'truncated': report.truncated, + })) + await reportJobRun(db, 'prune-finished-jobs', { + cron: CRON, + status: 'ok', + durationMs: Date.now() - startedAt, + }) + return { result: report } + }) + }, +}) diff --git a/layers/registry/server/utils/finished-job-retention.spec.ts b/layers/registry/server/utils/finished-job-retention.spec.ts new file mode 100644 index 00000000..0d78708c --- /dev/null +++ b/layers/registry/server/utils/finished-job-retention.spec.ts @@ -0,0 +1,127 @@ +import Database from 'better-sqlite3' +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { + COMPLETED_JOB_RETENTION_SECONDS, + EMBEDDING_ATTEMPT_RETENTION_SECONDS, + FAILED_JOB_RETENTION_SECONDS, + pruneFinishedJobs, +} from './finished-job-retention' + +const DAY = 24 * 60 * 60 +const NOW = 1_800_000_000 + +describe('pruneFinishedJobs', () => { + let sqlite: Database.Database + + beforeEach(() => { + sqlite = new Database(':memory:') + sqlite.exec(` + CREATE TABLE job_batches ( + id TEXT PRIMARY KEY, + parent_batch_id TEXT, + finished_at INTEGER + ); + CREATE TABLE jobs ( + id TEXT PRIMARY KEY, + batch_id TEXT REFERENCES job_batches(id), + payload TEXT NOT NULL DEFAULT '{}', + completed_at INTEGER, + failed_at INTEGER + ); + CREATE TABLE failed_jobs ( + id TEXT PRIMARY KEY, + batch_id TEXT, + failed_at INTEGER NOT NULL + ); + CREATE TABLE embedding_attempts ( + attempt_id TEXT PRIMARY KEY, + started_at INTEGER NOT NULL + ); + CREATE INDEX idx_jobs_completed_at ON jobs (completed_at) WHERE completed_at IS NOT NULL; + CREATE INDEX idx_failed_jobs_failed_at ON failed_jobs (failed_at); + CREATE INDEX idx_job_batches_finished_at ON job_batches (finished_at); + `) + }) + + afterEach(() => sqlite.close()) + + it('deletes old finished rows and keeps active and recent ones', async () => { + const old = NOW - COMPLETED_JOB_RETENTION_SECONDS - DAY + sqlite.exec(` + INSERT INTO job_batches (id, finished_at) VALUES + ('batch-old', ${old}), ('batch-open', NULL), ('batch-recent', ${NOW - DAY}); + INSERT INTO jobs (id, batch_id, completed_at) VALUES + ('done-old', 'batch-old', ${old}), + ('done-open-batch', 'batch-open', ${old}), + ('done-recent', NULL, ${NOW - DAY}), + ('active', NULL, NULL); + INSERT INTO failed_jobs (id, batch_id, failed_at) VALUES + ('failed-old', NULL, ${NOW - FAILED_JOB_RETENTION_SECONDS - DAY}), + ('failed-recent', NULL, ${NOW - DAY}); + INSERT INTO embedding_attempts (attempt_id, started_at) VALUES + ('attempt-old', ${NOW - EMBEDDING_ATTEMPT_RETENTION_SECONDS - DAY}), + ('attempt-recent', ${NOW - DAY}); + `) + + const report = await pruneFinishedJobs(wrapSqlite(sqlite), { now: NOW }) + + expect(report).toEqual({ + completedJobs: 1, + failedJobs: 1, + finishedBatches: 1, + embeddingAttempts: 1, + truncated: false, + }) + expect(ids('jobs')).toEqual(['active', 'done-open-batch', 'done-recent']) + expect(ids('failed_jobs')).toEqual(['failed-recent']) + expect(ids('job_batches')).toEqual(['batch-open', 'batch-recent']) + expect(ids('embedding_attempts', 'attempt_id')).toEqual(['attempt-recent']) + }) + + it('keeps a finished batch while a job still names it', async () => { + const old = NOW - COMPLETED_JOB_RETENTION_SECONDS - DAY + sqlite.exec(` + INSERT INTO job_batches (id, finished_at) VALUES ('batch', ${old}); + INSERT INTO jobs (id, batch_id, completed_at) VALUES ('member', 'batch', ${NOW - DAY}); + `) + + const report = await pruneFinishedJobs(wrapSqlite(sqlite), { now: NOW }) + + expect(report.finishedBatches).toBe(0) + expect(ids('job_batches')).toEqual(['batch']) + }) + + it('stops at the batch cap and reports the backlog', async () => { + const old = NOW - COMPLETED_JOB_RETENTION_SECONDS - DAY + const insert = sqlite.prepare(`INSERT INTO jobs (id, completed_at) VALUES (?, ?)`) + for (let i = 0; i < 7; i += 1) + insert.run(`done-${i}`, old) + + const report = await pruneFinishedJobs(wrapSqlite(sqlite), { now: NOW, limit: 2, maxBatches: 2 }) + + expect(report.completedJobs).toBe(4) + expect(report.truncated).toBe(true) + expect(ids('jobs')).toHaveLength(3) + }) + + function ids(table: string, column = 'id'): string[] { + return sqlite.prepare(`SELECT ${column} AS id FROM ${table} ORDER BY id`).all().map(row => (row as { id: string }).id) + } +}) + +function wrapSqlite(sqlite: Database.Database): D1Database { + return { + prepare(sql: string) { + return { + bind(...params: unknown[]) { + return { + async run() { + const info = sqlite.prepare(sql).run(...params) + return { meta: { changes: info.changes } } + }, + } + }, + } + }, + } as unknown as D1Database +} diff --git a/layers/registry/server/utils/finished-job-retention.ts b/layers/registry/server/utils/finished-job-retention.ts new file mode 100644 index 00000000..df40c45f --- /dev/null +++ b/layers/registry/server/utils/finished-job-retention.ts @@ -0,0 +1,171 @@ +/// + +/** + * How long a finished durable job row is kept. + * + * `completeJob` in nuxt-cf-jobs is a soft complete: the `jobs` row stays with + * its payload, `retry_reasons`, and `last_error`. The daily check-in reads a + * day or a week back, so seven days keeps every row it can ask for. + * + * Measured 2026-11-12: `jobs` held 146,816 rows, `failed_jobs` 2,672, and + * `job_batches` 3,332, from an hourly sync that writes up to 250 job rows + * and one batch row. Nothing deleted any of them. + */ +export const COMPLETED_JOB_RETENTION_SECONDS = 7 * 24 * 60 * 60 + +/** + * Failed jobs are the only evidence of a permanent repository failure once + * `repo_sync_progress` is cleared, so they stay longer than completed ones. + */ +export const FAILED_JOB_RETENTION_SECONDS = 30 * 24 * 60 * 60 + +/** + * Embedding attempts are provider evidence for one Vectorize upsert. The + * parity audit reads only the current `skill_generated` marker, so the + * attempt row matters for triage of a recent failure and nothing else. + */ +export const EMBEDDING_ATTEMPT_RETENTION_SECONDS = 30 * 24 * 60 * 60 + +/** + * Rows one DELETE may remove. The subquery seeks an index, and 500 rows keep + * each statement well under D1's row budget. + */ +export const FINISHED_JOB_PRUNE_BATCH = 500 + +/** + * DELETE statements one run may issue per table. Steady state needs one or + * two; the cap only binds while a backlog drains, and it keeps a run inside + * its scheduled runtime. + */ +export const FINISHED_JOB_PRUNE_MAX_BATCHES = 40 + +export interface FinishedJobPruneReport { + completedJobs: number + failedJobs: number + finishedBatches: number + embeddingAttempts: number + /** True when a table still had a full batch to delete when the cap bound. */ + truncated: boolean +} + +/** + * Finished job rows nothing needs any more. + * + * The predicates match nuxt-cf-jobs's own prune methods: a member of a batch + * stays while its batch is unfinished, and a batch stays while any job still + * names it. Member rows go before their batches, so the batch delete never + * violates the `jobs.batch_id` foreign key. + */ +const DELETE_COMPLETED_JOBS = ` + DELETE FROM jobs WHERE id IN ( + SELECT id FROM jobs INDEXED BY idx_jobs_completed_at + WHERE completed_at IS NOT NULL + AND completed_at <= ? + AND ( + batch_id IS NULL + OR NOT EXISTS ( + SELECT 1 FROM job_batches + WHERE job_batches.id = jobs.batch_id AND job_batches.finished_at IS NULL + ) + ) + LIMIT ? + ) +` + +const DELETE_FAILED_JOBS = ` + DELETE FROM failed_jobs WHERE id IN ( + SELECT id FROM failed_jobs INDEXED BY idx_failed_jobs_failed_at + WHERE failed_at <= ? + AND ( + batch_id IS NULL + OR NOT EXISTS ( + SELECT 1 FROM job_batches + WHERE job_batches.id = failed_jobs.batch_id AND job_batches.finished_at IS NULL + ) + ) + LIMIT ? + ) +` + +const DELETE_FINISHED_BATCHES = ` + DELETE FROM job_batches WHERE id IN ( + SELECT id FROM job_batches INDEXED BY idx_job_batches_finished_at + WHERE finished_at IS NOT NULL + AND finished_at <= ? + AND NOT EXISTS (SELECT 1 FROM jobs WHERE jobs.batch_id = job_batches.id) + AND NOT EXISTS (SELECT 1 FROM failed_jobs WHERE failed_jobs.batch_id = job_batches.id) + AND ( + parent_batch_id IS NULL + OR NOT EXISTS ( + SELECT 1 FROM job_batches parent + WHERE parent.id = job_batches.parent_batch_id AND parent.finished_at IS NULL + ) + ) + LIMIT ? + ) +` + +/** + * A started attempt older than the window is one the Worker never finished, + * so it is as settled as a terminal one. + */ +const DELETE_EMBEDDING_ATTEMPTS = ` + DELETE FROM embedding_attempts WHERE attempt_id IN ( + SELECT attempt_id FROM embedding_attempts + WHERE started_at <= ? + LIMIT ? + ) +` + +async function deleteInBatches( + db: D1Database, + sql: string, + cutoff: number, + limit: number, + maxBatches: number, +): Promise<{ deleted: number, truncated: boolean }> { + let deleted = 0 + for (let batch = 0; batch < maxBatches; batch += 1) { + const result = await db.prepare(sql).bind(cutoff, limit).run() + const changes = result.meta.changes ?? 0 + deleted += changes + if (changes < limit) + return { deleted, truncated: false } + } + return { deleted, truncated: true } +} + +/** + * Delete finished durable job rows and old embedding attempts, in bounded + * batches. Returns how many rows each table lost. + */ +export async function pruneFinishedJobs( + db: D1Database, + input: { + now: number + limit?: number + maxBatches?: number + completedRetentionSeconds?: number + failedRetentionSeconds?: number + embeddingAttemptRetentionSeconds?: number + }, +): Promise { + const limit = input.limit ?? FINISHED_JOB_PRUNE_BATCH + const maxBatches = input.maxBatches ?? FINISHED_JOB_PRUNE_MAX_BATCHES + const completedCutoff = input.now - (input.completedRetentionSeconds ?? COMPLETED_JOB_RETENTION_SECONDS) + const failedCutoff = input.now - (input.failedRetentionSeconds ?? FAILED_JOB_RETENTION_SECONDS) + const attemptCutoff = input.now - (input.embeddingAttemptRetentionSeconds ?? EMBEDDING_ATTEMPT_RETENTION_SECONDS) + + const completed = await deleteInBatches(db, DELETE_COMPLETED_JOBS, completedCutoff, limit, maxBatches) + const failed = await deleteInBatches(db, DELETE_FAILED_JOBS, failedCutoff, limit, maxBatches) + const batches = await deleteInBatches(db, DELETE_FINISHED_BATCHES, completedCutoff, limit, maxBatches) + const attempts = await deleteInBatches(db, DELETE_EMBEDDING_ATTEMPTS, attemptCutoff, limit, maxBatches) + + return { + completedJobs: completed.deleted, + failedJobs: failed.deleted, + finishedBatches: batches.deleted, + embeddingAttempts: attempts.deleted, + truncated: completed.truncated || failed.truncated || batches.truncated || attempts.truncated, + } +} diff --git a/shared/schedule-policy.ts b/shared/schedule-policy.ts index 511a7da9..2e7c3ce6 100644 --- a/shared/schedule-policy.ts +++ b/shared/schedule-policy.ts @@ -67,6 +67,7 @@ export const SCHEDULE_POLICY = [ { _tag: 'observed', taskName: 'embedding-parity-audit', cron: '0 21 * * *', maxSilenceSeconds: 36 * 60 * 60, maxRuntimeSeconds: 30 * 60 }, // First cron fire, not the authoring date: a grace anchored to 2026-09-17 // expired before the deployed trigger ever fired and read as missing_run. + { _tag: 'observed', taskName: 'prune-finished-jobs', cron: '40 * * * *', maxSilenceSeconds: 3 * 60 * 60, maxRuntimeSeconds: 4 * 60 }, { _tag: 'observed', taskName: 'purge-personal-data', cron: '30 4 * * *', maxSilenceSeconds: 36 * 60 * 60, maxRuntimeSeconds: 10 * 60, activeFromSeconds: 1790051400 }, { _tag: 'observed', taskName: 'recompute-skill-scores', cron: '0 3 * * *', maxSilenceSeconds: 36 * 60 * 60, maxRuntimeSeconds: 60 * 60 }, { _tag: 'observed', taskName: 'refresh-x-engagement', cron: '10 * * * *', maxSilenceSeconds: 3 * 60 * 60, maxRuntimeSeconds: 30 * 60 },