From e566e9497c4aa066c08c3d9b4fc69b3363e5918b Mon Sep 17 00:00:00 2001 From: Stephen Zweibel Date: Tue, 22 Sep 2026 14:21:15 -0400 Subject: [PATCH] Verify existing migration files before conditional uploads --- .../app/src/agents/mutation-coordinator.ts | 59 ++++++-- packages/app/src/lib/migration.test.ts | 42 ++++++ packages/app/src/lib/migration.ts | 126 +++++++++++++----- 3 files changed, 184 insertions(+), 43 deletions(-) diff --git a/packages/app/src/agents/mutation-coordinator.ts b/packages/app/src/agents/mutation-coordinator.ts index c1f5a2f..8966e39 100644 --- a/packages/app/src/agents/mutation-coordinator.ts +++ b/packages/app/src/agents/mutation-coordinator.ts @@ -16,6 +16,7 @@ import { import { createAgentHistoryPorter, createProjectHistoryLifecycle } from "../lib/agent-porter"; import { createSiteStudioBoundaryContext, + emitDiagnostic, type SiteStudioLoggingContextData, } from "../lib/logging"; import { @@ -103,13 +104,40 @@ export class MutationCoordinator extends DurableObject { subject: string, ): Promise { return this.mutations.run(async () => { - await new OwnerMutationService( - this.env.SITE_STUDIO_BUCKET, - this.ctx.storage, - undefined, - createProjectHistoryLifecycle(this.env), - ).recover(anonUserId); - const { manifest } = await loadRecoveryManifest(this.env.SITE_STUDIO_BUCKET, recoveryId); + const logging = this.env.CAIL_LOG_ENV + ? createSiteStudioBoundaryContext(this.env) + : undefined; + try { + await new OwnerMutationService( + this.env.SITE_STUDIO_BUCKET, + this.ctx.storage, + undefined, + createProjectHistoryLifecycle(this.env), + ).recover(anonUserId); + } catch (error) { + emitDiagnostic( + "error", + "legacy_recovery_source_journal_failed", + {}, + logging, + ); + throw error; + } + let manifest: Awaited>["manifest"]; + try { + ({ manifest } = await loadRecoveryManifest( + this.env.SITE_STUDIO_BUCKET, + recoveryId, + )); + } catch (error) { + emitDiagnostic( + "error", + "legacy_recovery_manifest_load_failed", + {}, + logging, + ); + throw error; + } if (manifest.source.owner !== anonUserId) throw new LegacyRecoveryError("conflict"); // The first source-queue entry must revalidate immediately before copy. // A later entry with a migration claim is a resume: source deletion may @@ -119,14 +147,25 @@ export class MutationCoordinator extends DurableObject { migrationClaimKey(anonUserId), "json", ); - await validateStagedRecovery(this.env.SITE_STUDIO_BUCKET, manifest, { - allowMissing: Boolean(migrationClaim), - }); + try { + await validateStagedRecovery(this.env.SITE_STUDIO_BUCKET, manifest, { + allowMissing: Boolean(migrationClaim), + }); + } catch (error) { + emitDiagnostic( + "error", + "legacy_recovery_source_validation_failed", + {}, + logging, + ); + throw error; + } return migrateAnonymousData({ bucket: this.env.SITE_STUDIO_BUCKET, kv: this.env.SESSION_KV, anonUserId, subject, + logging, }); }); } diff --git a/packages/app/src/lib/migration.test.ts b/packages/app/src/lib/migration.test.ts index ea87ab1..fcc862d 100644 --- a/packages/app/src/lib/migration.test.ts +++ b/packages/app/src/lib/migration.test.ts @@ -289,6 +289,48 @@ describe("migrateAnonymousData", () => { expect(kv.store.has(migrationPendingKey(SUBJECT))).toBe(false); }); + it("validates an already copied file without attempting another upload", async () => { + const content = "

large preserved source

"; + seedAnonProject(bucket, "portfolio", {}, content); + // SAFETY: seedAnonProject serializes this value from metadataFor's ProjectMetadata fixture. + const sourceMetadata = JSON.parse( + bucket.store.get(`projects/${ANON}/portfolio/.metadata.json`)!.data, + ) as ProjectMetadata; + bucket.store.set(`projects/${SUBJECT}/portfolio/.metadata.json`, { + data: JSON.stringify({ + ...sourceMetadata, + importedFrom: ANON, + importedOriginalId: "portfolio", + }), + }); + const destination = `projects/${SUBJECT}/portfolio/index.html`; + bucket.store.set(destination, { data: content }); + kv.store.set( + migrationClaimKey(ANON), + JSON.stringify({ + subject: SUBJECT, + status: "pending", + startedAt: "2026-01-01T00:00:00.000Z", + } satisfies MigrationClaim), + ); + + const originalPut = bucket.put; + const destinationPuts: string[] = []; + bucket.put = vi.fn( + async (key: string, data: string, options?: R2PutOptions) => { + if (key === destination) destinationPuts.push(key); + return originalPut(key, data, options); + }, + ); + + await expect(run()).resolves.toMatchObject({ + status: "migrated", + projects: { portfolio: "portfolio" }, + }); + expect(destinationPuts).toEqual([]); + expect(textOf(bucket.store.get(destination))).toBe(content); + }); + it("is idempotent: a second run is a no-op that changes nothing", async () => { seedAnonProject(bucket, "portfolio"); await run(); diff --git a/packages/app/src/lib/migration.ts b/packages/app/src/lib/migration.ts index 4699a3d..3a6522e 100644 --- a/packages/app/src/lib/migration.ts +++ b/packages/app/src/lib/migration.ts @@ -166,6 +166,22 @@ async function copyIfAbsent( const object = await bucket.get(fromKey); if (!object) return; // source vanished (concurrent run finished it) — fine const sourceBytes = await object.arrayBuffer(); + + // A retry or the defensive second sweep usually finds the destination + // already present. Compare that object before attempting a conditional PUT: + // the R2 transport sends the entire request body before it handles a failed + // precondition, so needlessly re-uploading a large existing object can fail + // before the binding returns the documented null result. + const existing = await bucket.get(toKey); + if (existing) { + if (!bytesEqual(sourceBytes, await existing.arrayBuffer())) { + throw new Error( + "Anonymous-data migration stopped because the destination changed concurrently.", + ); + } + return; + } + const wrote = await bucket.put(toKey, sourceBytes, { httpMetadata: object.httpMetadata, onlyIf: { etagDoesNotMatch: "*" }, @@ -183,6 +199,30 @@ async function copyIfAbsent( } } +const migrationFailureType = { + inventory: "migration_inventory_failed", + firstCopy: "migration_first_copy_failed", + secondCopy: "migration_second_copy_failed", + handle: "migration_handle_failed", + sourceRetirement: "migration_source_retirement_failed", + completion: "migration_completion_failed", +} as const; + +async function runMigrationStage( + stage: keyof typeof migrationFailureType, + logging: SiteStudioLoggingContext | undefined, + operation: () => Promise, +): Promise { + try { + return await operation(); + } catch (error) { + // Static stage categories only: never project an exception message, key, + // owner, subject, or project identifier into operational logs. + emitDiagnostic("error", migrationFailureType[stage], {}, logging); + throw error; + } +} + async function putJsonIfAbsentOrEqual( bucket: R2Bucket, key: string, @@ -611,64 +651,84 @@ export async function migrateAnonymousData(options: { }; // ---- Inventory ---- - const anonProjectIds = await listProjectIds(bucket, anonUserId); - const uploadKeys = await listKeys(bucket, uploadsPrefix(anonUserId)); + const { anonProjectIds, uploadKeys } = await runMigrationStage( + "inventory", + logging, + async () => ({ + anonProjectIds: await listProjectIds(bucket, anonUserId), + uploadKeys: await listKeys(bucket, uploadsPrefix(anonUserId)), + }), + ); if (anonProjectIds.length === 0 && uploadKeys.length === 0) { // There are no project chats to port, so the planned handle can be // committed before retiring the completed anonymous session as well. - await migrateHandle({ bucket, anonUserId, subject, now }); - return finish("nothing-to-migrate", {}); + await runMigrationStage("handle", logging, () => + migrateHandle({ bucket, anonUserId, subject, now }) + ); + return runMigrationStage("completion", logging, () => + finish("nothing-to-migrate", {}) + ); } // ---- First copy sweep ---- - const firstSweep = await copyAnonymousNamespace({ - bucket, - anonUserId, - subject, - porter, - knownProjects: {}, - knownSlugs: {}, - logging, - }); + const firstSweep = await runMigrationStage("firstCopy", logging, () => + copyAnonymousNamespace({ + bucket, + anonUserId, + subject, + porter, + knownProjects: {}, + knownSlugs: {}, + logging, + }) + ); // ---- Second copy sweep, immediately before delete ---- // The owner-scoped MutationCoordinator prevents adopted anonymous writes // during this run. Re-list anyway so an older deployment or out-of-band // bucket writer cannot be deleted merely because it appeared after the first // inventory. Conditional copy-or-equal keeps this sweep idempotent. - const { projectMap } = await copyAnonymousNamespace({ - bucket, - anonUserId, - subject, - porter, - knownProjects: firstSweep.projectMap, - knownSlugs: firstSweep.slugMap, - logging, - }); + const { projectMap } = await runMigrationStage("secondCopy", logging, () => + copyAnonymousNamespace({ + bucket, + anonUserId, + subject, + porter, + knownProjects: firstSweep.projectMap, + knownSlugs: firstSweep.slugMap, + logging, + }) + ); // All project chat ports have now completed. Re-home the handle only after // that last success and before deleting // the anonymous source. A handle failure therefore also leaves the source // and pending claim available for a later retry. - await migrateHandle({ bucket, anonUserId, subject, now }); + await runMigrationStage("handle", logging, () => + migrateHandle({ bucket, anonUserId, subject, now }) + ); // ---- Delete originals ---- // Retire secondary artifacts first. If a snapshot/upload delete fails, the // project inventory remains discoverable as the retry anchor instead of // leaving an orphaned secondary namespace that a later completed marker can // no longer recover. - for (const key of await listKeys(bucket, snapshotUserPrefix(anonUserId))) { - await bucket.delete(key); - } - for (const key of await listKeys(bucket, uploadsPrefix(anonUserId))) { - await bucket.delete(key); - } - for (const key of await listKeys(bucket, projectPrefix(anonUserId))) { - await bucket.delete(key); - } + await runMigrationStage("sourceRetirement", logging, async () => { + for (const key of await listKeys(bucket, snapshotUserPrefix(anonUserId))) { + await bucket.delete(key); + } + for (const key of await listKeys(bucket, uploadsPrefix(anonUserId))) { + await bucket.delete(key); + } + for (const key of await listKeys(bucket, projectPrefix(anonUserId))) { + await bucket.delete(key); + } + }); - return finish("migrated", projectMap); + return runMigrationStage("completion", logging, () => + finish("migrated", projectMap) + ); } /**