Fix scheduled backup recovery

scripts/orbit-production-config-tests.mjs

@@ -124,7 +124,7 @@ const expectedResources = {
     },
   },
   triggers: {
-    crons: ['17 3 * * *'],
+    crons: ['17 3 * * *', '0 4 * * *'],
   },
 };
 

scripts/orbit-slice1-test-worker.ts

@@ -657,6 +657,56 @@ async function testRoute(request: Request, env: TestEnv): Promise<Response | nul
     return Response.json({ ok: true });
   }
 
+  if (url.pathname === '/__test/seed-referenced-idempotency') {
+    try {
+      const id = String(body.id);
+      const accountId = '019f64d2-0109-7644-9a4e-a0d25df888e2';
+      const monthUtc = new Date(now).toISOString().slice(0, 7);
+      const usageDay = new Date(now).toISOString().slice(0, 10);
+      await env.DB.batch([
+      env.DB.prepare(`
+        INSERT INTO idempotency_keys (
+          id, principal_type, principal_id, key_digest, operation,
+          request_digest, response_status, resource_type, resource_id,
+          created_at, expires_at, response_json, state, completed_at
+        ) VALUES (?, 'account', ?, ?, 'POST /v1/account/avatar', ?, 0,
+          'media_upload', NULL, ?, ?, '{}', 'in_progress', NULL)
+      `).bind(id, accountId, id, id, now - 1000, now - 1),
+      env.DB.prepare(`
+        INSERT INTO media_transform_usage_monthly (
+          month_utc, attempted_count, succeeded_count, failed_count, updated_at
+        ) VALUES (?, 0, 0, 0, ?)
+        ON CONFLICT(month_utc) DO NOTHING
+      `).bind(monthUtc, now),
+      env.DB.prepare(`
+        INSERT INTO media_transform_claims (
+          id, month_utc, profile, actor_type, actor_id,
+          source_content_type, source_byte_size, status, created_at,
+          usage_day, target_type, target_id, idempotency_id
+        ) VALUES (?, ?, 'avatar', 'account', ?, 'image/png', 128,
+          'reserved', ?, ?, 'account', ?, NULL)
+      `).bind(`${id}-claim`, monthUtc, accountId, now, usageDay, accountId),
+      ]);
+      await env.DB.prepare(`
+        UPDATE media_transform_claims SET idempotency_id = ? WHERE id = ?
+      `).bind(id, `${id}-claim`).run();
+      await env.DB.prepare(`
+        UPDATE idempotency_keys
+        SET state = 'completed', response_status = 201,
+            response_json = '{"ok":true}', completed_at = ?
+        WHERE id = ?
+      `).bind(now, id).run();
+      const claim = await env.DB.prepare(`
+        SELECT idempotency_id FROM media_transform_claims WHERE id = ?
+      `).bind(`${id}-claim`).first<{ idempotency_id: string | null }>();
+      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 });
 }
 

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

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

src/server/backup/r2-backup.ts

@@ -63,12 +63,10 @@ export async function enforceBackupRetention(bucket: R2BucketLike): Promise<Reco
   return deleted;
 }
 
-export async function runR2Backup(
+export async function reconcileStaleBackupRuns(
   env: OrbitBindings,
-  kind: BackupKind,
   now = Date.now(),
-  actorAccountId: string | null = null,
-): Promise<{ runId: string; objectKey: string; manifestChecksum: string; objectChecksum: string }> {
+): Promise<number> {
   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 });

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 {

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<unknown>): void;
 }
 
+interface ScheduledControllerLike {
+  cron?: string;
+  scheduledTime?: number;
+}
+
+interface ScheduledDependencies {
+  runIdentityCleanup(env: OrbitBindings, now: number): Promise<unknown>;
+  runScheduledBackups(env: OrbitBindings, now: number): Promise<unknown>;
+  cleanupMedia(env: OrbitBindings, now: number): Promise<unknown>;
+  reconcileStaleBackupRuns(env: OrbitBindings, now: number): Promise<unknown>;
+}
+
+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<ApiDependencies, 'requestId'> {
   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<void> {
+  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<Response> {
     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));
   },
 };

wrangler.jsonc

@@ -27,6 +27,6 @@
     "enabled": false
   },
   "triggers": {
-    "crons": ["17 3 * * *"]
+    "crons": ["17 3 * * *", "0 4 * * *"]
   }
 }

wrangler.production.dark-launch.jsonc

@@ -56,6 +56,6 @@
     }
   },
   "triggers": {
-    "crons": ["17 3 * * *"]
+    "crons": ["17 3 * * *", "0 4 * * *"]
   }
 }

wrangler.production.live.jsonc

@@ -56,6 +56,6 @@
     }
   },
   "triggers": {
-    "crons": ["17 3 * * *"]
+    "crons": ["17 3 * * *", "0 4 * * *"]
   }
 }

wrangler.staging.jsonc

@@ -50,6 +50,6 @@
     }
   },
   "triggers": {
-    "crons": ["17 3 * * *"]
+    "crons": ["17 3 * * *", "0 4 * * *"]
   }
 }