diff --git a/scripts/orbit-production-config-tests.mjs b/scripts/orbit-production-config-tests.mjs index ec61ac5..7a3a9ef 100644 --- a/scripts/orbit-production-config-tests.mjs +++ b/scripts/orbit-production-config-tests.mjs @@ -124,7 +124,7 @@ const expectedResources = { }, }, triggers: { - crons: ['17 3 * * *'], + crons: ['17 3 * * *', '0 4 * * *'], }, }; diff --git a/scripts/orbit-slice1-test-worker.ts b/scripts/orbit-slice1-test-worker.ts index b76f265..5631542 100644 --- a/scripts/orbit-slice1-test-worker.ts +++ b/scripts/orbit-slice1-test-worker.ts @@ -657,6 +657,56 @@ async function testRoute(request: Request, env: TestEnv): Promise(); + return Response.json({ ok: true, idempotencyId: claim?.idempotency_id ?? null }); + } catch (error) { + return Response.json({ + error: error instanceof Error ? error.message : 'seed_referenced_idempotency_failed', + }, { status: 500 }); + } + } + return Response.json({ error: 'not_found' }, { status: 404 }); } diff --git a/scripts/orbit-slice1-tests.ts b/scripts/orbit-slice1-tests.ts index 9739d836..1851498 100644 --- a/scripts/orbit-slice1-tests.ts +++ b/scripts/orbit-slice1-tests.ts @@ -1556,9 +1556,15 @@ let firstCredentialToken = ''; assert.equal(expired.status, 401); }); - test('daily cleanup removes retained OAuth/session rows but leaves audit evidence', async () => { + test('daily cleanup retains referenced media idempotency while removing other expired rows', async () => { const cleanupAt = NOW + 62 * 24 * 60 * 60 * 1000; await postJson('/__test/seed-idempotency', { id: 'cleanup-key' }, {}, cleanupAt); + const referencedSeed = await postJson('/__test/seed-referenced-idempotency', { + id: 'cleanup-referenced-key', + }, {}, cleanupAt); + assert.equal(referencedSeed.status, 200, await referencedSeed.clone().text()); + assert.equal((await referencedSeed.json() as { idempotencyId: string }).idempotencyId, + 'cleanup-referenced-key'); const beforeResponse = await postJson('/__test/state', {}, {}, cleanupAt); const before = await beforeResponse.json() as { counts: { audit_events: number; idempotency_keys: number }; @@ -1573,12 +1579,12 @@ let firstCredentialToken = ''; assert.ok(body.oauthFlows > 0); assert.ok(body.sessions > 0); assert.ok(before.counts.idempotency_keys > 0); - assert.equal(body.idempotencyKeys, before.counts.idempotency_keys); + assert.equal(body.idempotencyKeys, before.counts.idempotency_keys - 1); const afterResponse = await postJson('/__test/state', {}, {}, cleanupAt); const after = await afterResponse.json() as { counts: { audit_events: number; idempotency_keys: number }; }; - assert.equal(after.counts.idempotency_keys, 0); + assert.equal(after.counts.idempotency_keys, 1); assert.equal(after.counts.audit_events, before.counts.audit_events); }); }); diff --git a/scripts/orbit-slice1-worker-tests.ts b/scripts/orbit-slice1-worker-tests.ts index 3b225cf..d8c841c 100644 --- a/scripts/orbit-slice1-worker-tests.ts +++ b/scripts/orbit-slice1-worker-tests.ts @@ -1,6 +1,10 @@ import assert from 'node:assert/strict'; import { describe, test } from 'node:test'; -import worker from '../src/worker'; +import worker, { + BACKUP_CRON, + BACKUP_RECONCILIATION_CRON, + runScheduledMaintenance, +} from '../src/worker'; import { assertDeploymentBindings, assertIdentityBindings, @@ -331,4 +335,54 @@ describe('Orbit V6 deployment-mode contract', () => { assert.equal(response.headers.get('x-robots-tag'), null); } }); + + test('waits for independent scheduled tasks before reporting one task failure', async () => { + const calls: string[] = []; + const scheduledTime = 1_785_727_021_000; + await assert.rejects( + runScheduledMaintenance( + { cron: BACKUP_CRON, scheduledTime }, + productionBindings('live'), + { + async runIdentityCleanup(_env, now) { + assert.equal(now, scheduledTime); + calls.push('identity_cleanup'); + throw new Error('simulated_cleanup_failure'); + }, + async runScheduledBackups(_env, now) { + assert.equal(now, scheduledTime); + calls.push('backup'); + }, + async cleanupMedia(_env, now) { + assert.equal(now, scheduledTime); + calls.push('media_cleanup'); + }, + async reconcileStaleBackupRuns() { + calls.push('reconciliation'); + }, + }, + ), + /scheduled_maintenance_failed:identity_cleanup/u, + ); + assert.deepEqual(calls.sort(), ['backup', 'identity_cleanup', 'media_cleanup']); + }); + + test('routes the follow-up cron only to stale backup reconciliation', async () => { + const calls: string[] = []; + const scheduledTime = 1_785_729_600_000; + await runScheduledMaintenance( + { cron: BACKUP_RECONCILIATION_CRON, scheduledTime }, + productionBindings('live'), + { + async runIdentityCleanup() { calls.push('identity_cleanup'); }, + async runScheduledBackups() { calls.push('backup'); }, + async cleanupMedia() { calls.push('media_cleanup'); }, + async reconcileStaleBackupRuns(_env, now) { + assert.equal(now, scheduledTime); + calls.push('reconciliation'); + }, + }, + ); + assert.deepEqual(calls, ['reconciliation']); + }); }); diff --git a/src/server/backup/r2-backup.ts b/src/server/backup/r2-backup.ts index ba9d8b2..fc2b151 100644 --- a/src/server/backup/r2-backup.ts +++ b/src/server/backup/r2-backup.ts @@ -63,12 +63,10 @@ export async function enforceBackupRetention(bucket: R2BucketLike): Promise { +): Promise { const repository = new D1PlatformRepository(env.DB); const reconciledRuns = await repository.failStaleBackupRuns({ before: now - STALE_BACKUP_RUN_MS, @@ -82,6 +80,17 @@ export async function runR2Backup( count: reconciledRuns, })); } + return reconciledRuns; +} + +export async function runR2Backup( + env: OrbitBindings, + kind: BackupKind, + now = Date.now(), + actorAccountId: string | null = null, +): Promise<{ runId: string; objectKey: string; manifestChecksum: string; objectChecksum: string }> { + const repository = new D1PlatformRepository(env.DB); + await reconcileStaleBackupRuns(env, now); const runId = createEntityId(); const startedAt = Date.now(); await repository.startBackupRun({ id: runId, kind, actorAccountId, now }); diff --git a/src/server/repositories/d1/d1-identity-repository.ts b/src/server/repositories/d1/d1-identity-repository.ts index 2b60f4c..83319d6 100644 --- a/src/server/repositories/d1/d1-identity-repository.ts +++ b/src/server/repositories/d1/d1-identity-repository.ts @@ -528,7 +528,12 @@ export class D1IdentityRepository implements IdentityRepository { OR (MIN(idle_expires_at, absolute_expires_at) <= ?) `).bind(sessionCutoff, sessionCutoff), this.#db.prepare(` - DELETE FROM idempotency_keys WHERE expires_at <= ? + DELETE FROM idempotency_keys + WHERE expires_at <= ? + AND NOT EXISTS ( + SELECT 1 FROM media_transform_claims claim + WHERE claim.idempotency_id = idempotency_keys.id + ) `).bind(now), ]); return { diff --git a/src/worker.ts b/src/worker.ts index 8fa6fe8..9226741 100644 --- a/src/worker.ts +++ b/src/worker.ts @@ -5,7 +5,10 @@ import { } from './server/identity/bindings'; import { handleApiRequest, runIdentityCleanup, type ApiDependencies } from './server/http/api'; import { dashboardAssetResponse } from './server/dashboard/response'; -import { runScheduledBackups } from './server/backup/r2-backup'; +import { + reconcileStaleBackupRuns, + runScheduledBackups, +} from './server/backup/r2-backup'; import { bumpPublicCacheEpoch, mutationInvalidatesPublicCache, @@ -25,6 +28,32 @@ interface ExecutionContextLike { waitUntil(promise: Promise): void; } +interface ScheduledControllerLike { + cron?: string; + scheduledTime?: number; +} + +interface ScheduledDependencies { + runIdentityCleanup(env: OrbitBindings, now: number): Promise; + runScheduledBackups(env: OrbitBindings, now: number): Promise; + cleanupMedia(env: OrbitBindings, now: number): Promise; + reconcileStaleBackupRuns(env: OrbitBindings, now: number): Promise; +} + +export const BACKUP_CRON = '17 3 * * *'; +export const BACKUP_RECONCILIATION_CRON = '0 4 * * *'; + +const scheduledDependencies: ScheduledDependencies = { + runIdentityCleanup, + runScheduledBackups, + cleanupMedia: async (env, now) => await cleanupMedia( + env, + new D1MediaRepository(env.DB), + now, + ), + reconcileStaleBackupRuns, +}; + interface WorkerDependencies extends Omit { publicRepository?: PublicRepository; agentRepository?: AgentRepository; @@ -140,17 +169,46 @@ export async function handleWorkerRequest( return protectFromIndexing(response, env); } +export async function runScheduledMaintenance( + controller: ScheduledControllerLike, + env: OrbitBindings, + dependencies: ScheduledDependencies = scheduledDependencies, +): Promise { + const now = controller.scheduledTime ?? Date.now(); + if (controller.cron === BACKUP_RECONCILIATION_CRON) { + await dependencies.reconcileStaleBackupRuns(env, now); + return; + } + + const tasks = [ + ['identity_cleanup', () => dependencies.runIdentityCleanup(env, now)], + ['backup', () => dependencies.runScheduledBackups(env, now)], + ['media_cleanup', () => dependencies.cleanupMedia(env, now)], + ] as const; + const results = await Promise.allSettled(tasks.map(([, run]) => Promise.resolve().then(run))); + const failedTasks: string[] = []; + results.forEach((result, index) => { + if (result.status === 'fulfilled') return; + const task = tasks[index]?.[0] ?? 'unknown'; + failedTasks.push(task); + console.error(JSON.stringify({ + event: 'scheduled.task', + task, + status: 'failed', + })); + }); + if (failedTasks.length > 0) { + throw new Error(`scheduled_maintenance_failed:${failedTasks.join(',')}`); + } +} + export default { async fetch(request: Request, env: OrbitBindings): Promise { return await handleWorkerRequest(request, env); }, - scheduled(_controller: unknown, env: OrbitBindings, ctx: ExecutionContextLike): void { + scheduled(controller: ScheduledControllerLike, env: OrbitBindings, ctx: ExecutionContextLike): void { assertDeploymentBindings(env); - ctx.waitUntil(Promise.all([ - runIdentityCleanup(env), - runScheduledBackups(env), - cleanupMedia(env, new D1MediaRepository(env.DB)), - ])); + ctx.waitUntil(runScheduledMaintenance(controller, env)); }, }; diff --git a/wrangler.jsonc b/wrangler.jsonc index 77b35c6..fc249c3 100644 --- a/wrangler.jsonc +++ b/wrangler.jsonc @@ -27,6 +27,6 @@ "enabled": false }, "triggers": { - "crons": ["17 3 * * *"] + "crons": ["17 3 * * *", "0 4 * * *"] } } diff --git a/wrangler.production.dark-launch.jsonc b/wrangler.production.dark-launch.jsonc index ca963f5..ef651fd 100644 --- a/wrangler.production.dark-launch.jsonc +++ b/wrangler.production.dark-launch.jsonc @@ -56,6 +56,6 @@ } }, "triggers": { - "crons": ["17 3 * * *"] + "crons": ["17 3 * * *", "0 4 * * *"] } } diff --git a/wrangler.production.live.jsonc b/wrangler.production.live.jsonc index d63de32..9651baa 100644 --- a/wrangler.production.live.jsonc +++ b/wrangler.production.live.jsonc @@ -56,6 +56,6 @@ } }, "triggers": { - "crons": ["17 3 * * *"] + "crons": ["17 3 * * *", "0 4 * * *"] } } diff --git a/wrangler.staging.jsonc b/wrangler.staging.jsonc index 98973b2..8645722 100644 --- a/wrangler.staging.jsonc +++ b/wrangler.staging.jsonc @@ -50,6 +50,6 @@ } }, "triggers": { - "crons": ["17 3 * * *"] + "crons": ["17 3 * * *", "0 4 * * *"] } }