diff --git a/apps/cli/src/legacy/commands/db/shared/legacy-pgdelta-next-shadow.layer.ts b/apps/cli/src/legacy/commands/db/shared/legacy-pgdelta-next-shadow.layer.ts index 24d6423513..e05d2ddf1a 100644 --- a/apps/cli/src/legacy/commands/db/shared/legacy-pgdelta-next-shadow.layer.ts +++ b/apps/cli/src/legacy/commands/db/shared/legacy-pgdelta-next-shadow.layer.ts @@ -22,9 +22,16 @@ import { import { legacyWaitForShadowReady } from "../../../shared/db-bootstrap/health-check.ts"; import { legacyAcquireShadowDatabase, - type LegacyShadowAcquiredHandle, + legacyPeekShadowBaseline, + type LegacyShadowBaselinePeek, type LegacyShadowCacheOpts, + type LegacyShadowAcquiredHandle, } from "../../../shared/db-bootstrap/shadow-cache.ts"; +import { + legacyBufferedShadowOutput, + legacyResolvePlanShadowStrategy, + legacyRunPlanShadowProvisions, +} from "./legacy-pgdelta-next-shadow.plan.ts"; import { legacyConnectShadowDatabase, legacyMigrateNextShadowDatabase, @@ -82,10 +89,6 @@ interface NativeShadowBase { readonly image: string; } -interface ProvisionedMigrationsShadow extends LegacyPgDeltaNextMigrationsShadow { - readonly snapshotKey: string | undefined; -} - interface ProvisionedDeclarativeShadow { readonly declarativeUrl: string; readonly restoredFromPgDataSnapshot: boolean; @@ -100,11 +103,11 @@ interface ProvisionedDeclarativeShadow { * tar, and the migrations side is that tar's lineage whether it warm-restored FROM the tar or * cold-exported it this very run — the baseline handoff, where requiring the migrations handle * itself to be a warm restore would leave the guard armed against its own clone and fail the - * first cold plan (review: Codex on #6184, P1). A freshly initdb'd declarative shadow always + * first cold plan (review: Codex on #6215, P1). A freshly initdb'd declarative shadow always * carries its own new identity, and different keys mean tars exported from different clusters, * so both of those stay `false` and keep the guard armed. A `true` alongside identities that * happen to differ is harmless by design: pg-delta's bypass only takes effect on an exact - * identity match, never on a same-lineage sibling. + * identity match (`schema-plan.ts`'s `trustedCloneBypass`), never on a same-lineage sibling. */ export function legacyAllowSameDatabaseIdentityForPlanShadows(opts: { readonly declarativeRestoredFromPgDataSnapshot: boolean; @@ -170,19 +173,23 @@ export const legacyPgDeltaNextShadowLayer = Layer.effect( const dbConnection = yield* LegacyDbConnection; const httpClient = yield* HttpClient.HttpClient; - const runtime = Layer.mergeAll( - Layer.succeed(FileSystem.FileSystem, fs), - Layer.succeed(Path.Path, path), - Layer.succeed(LegacyDebugFlag, debugFlag), - Layer.succeed(LegacyExperimentalFlag, experimentalFlag), - Layer.succeed(LegacyNetworkIdFlag, networkIdFlag), - Layer.succeed(CliArgs, cliArgs), - Layer.succeed(Output, output), - Layer.succeed(RuntimeInfo, runtimeInfo), - Layer.succeed(LegacyDockerRun, docker), - Layer.succeed(LegacyDbConnection, dbConnection), - Layer.succeed(HttpClient.HttpClient, httpClient), - ); + // Parameterized on the Output service so `provisionPlan` can hand a concurrently running + // provision a buffering decorator (`legacyBufferedShadowOutput`) instead of the live one. + const runtimeWith = (outputService: typeof Output.Service) => + Layer.mergeAll( + Layer.succeed(FileSystem.FileSystem, fs), + Layer.succeed(Path.Path, path), + Layer.succeed(LegacyDebugFlag, debugFlag), + Layer.succeed(LegacyExperimentalFlag, experimentalFlag), + Layer.succeed(LegacyNetworkIdFlag, networkIdFlag), + Layer.succeed(CliArgs, cliArgs), + Layer.succeed(Output, outputService), + Layer.succeed(RuntimeInfo, runtimeInfo), + Layer.succeed(LegacyDockerRun, docker), + Layer.succeed(LegacyDbConnection, dbConnection), + Layer.succeed(HttpClient.HttpClient, httpClient), + ); + const runtime = runtimeWith(output); const nextPort = (excluded?: number) => Effect.gen(function* () { @@ -263,19 +270,39 @@ export const legacyPgDeltaNextShadowLayer = Layer.effect( }, ); - const provisionMigrations = (input: NativeShadowInput, opts: LegacyShadowCacheOpts) => + const provisionMigrations = ( + input: NativeShadowInput, + opts: LegacyShadowCacheOpts, + onBaselineSeam: Effect.Effect = Effect.void, + ) => Effect.gen(function* () { const handle = yield* acquireShadow(input, opts); - yield* awaitShadowReady(input, handle); - const setup = setupRunInput(input, handle); - yield* legacyMigrateNextShadowDatabase(input.spawner, setup, handle); + // The baseline-handoff strategy (`legacy-pgdelta-next-shadow.plan.ts`) waits on + // `onBaselineSeam` before warm-restoring the declarative shadow. A snapshot-cold handle + // reaches that seam when its export publishes the tar; any other handle (warm because + // another process published between peek and acquire, or uncached) never runs a + // snapshot, so signal immediately — the waiter then just re-peeks current disk state. + const seamWillRun = handle.snapshotRequired && !handle.baselinePresent; + const seamHandle: LegacyShadowAcquiredHandle = seamWillRun + ? { + ...handle, + snapshotBaseline: handle.snapshotBaseline.pipe(Effect.ensuring(onBaselineSeam)), + } + : handle; + if (!seamWillRun) yield* onBaselineSeam; + yield* awaitShadowReady(input, seamHandle); + const setup = setupRunInput(input, seamHandle); + yield* legacyMigrateNextShadowDatabase(input.spawner, setup, seamHandle); return { migrationsUrl: legacyToPostgresURL(setup.connConfig), - snapshotKey: handle.snapshotKey, - } satisfies ProvisionedMigrationsShadow; + } satisfies LegacyPgDeltaNextMigrationsShadow; }).pipe(Effect.provide(runtime), Effect.mapError(nextShadowError)); - const provisionDeclarative = (input: NativeShadowInput, opts: LegacyShadowCacheOpts) => + const provisionDeclarative = ( + input: NativeShadowInput, + opts: LegacyShadowCacheOpts, + outputService: typeof Output.Service = output, + ) => Effect.gen(function* () { const handle = yield* acquireShadow(input, opts); yield* awaitShadowReady(input, handle); @@ -295,7 +322,7 @@ export const legacyPgDeltaNextShadowLayer = Layer.effect( restoredFromPgDataSnapshot: handle.baselinePresent, snapshotKey: handle.snapshotKey, } satisfies ProvisionedDeclarativeShadow; - }).pipe(Effect.provide(runtime), Effect.mapError(nextShadowError)); + }).pipe(Effect.provide(runtimeWith(outputService)), Effect.mapError(nextShadowError)); const cacheOpts = ( opts: LegacyPgDeltaNextShadowInput, @@ -320,22 +347,78 @@ export const legacyPgDeltaNextShadowLayer = Layer.effect( const built = yield* buildNativeBase(opts); const migrationsInput = buildNativeInput(opts, built, migrationsPort); const declarativeInput = buildNativeInput(opts, built, declarativePort); - const migrations = yield* provisionMigrations(migrationsInput, cacheOpts(opts, "config")); - const declarative = yield* provisionDeclarative( - declarativeInput, - cacheOpts(opts, "disabled"), - ); + // The two shadows are independent — anonymous containers on the distinct host ports + // allocated above, per-invocation scoped temp dirs, and a race-tolerant network + // ensure — so warm provisions run fully concurrently. How much can safely overlap + // when a baseline still has to be BUILT is the strategy question: peek both cache + // states up front and dispatch (see `legacy-pgdelta-next-shadow.plan.ts` for the + // three strategies and their transcript guarantees). The peeks also resolve the + // cache-key inputs once; passing them back through `precomputedKeyInputs` keeps the + // acquire from repeating a live JWKS discovery request. + const [migrationsPeek, declarativePeek] = yield* Effect.all([ + legacyPeekShadowBaseline(migrationsInput.base, cacheOpts(opts, "config")), + legacyPeekShadowBaseline(declarativeInput.base, cacheOpts(opts, "disabled")), + ]); + const withPeek = ( + cache: LegacyShadowCacheOpts, + peek: LegacyShadowBaselinePeek, + ): LegacyShadowCacheOpts => + peek.state === "uncachable" + ? cache + : { ...cache, precomputedKeyInputs: peek.keyInputs }; + const strategy = legacyResolvePlanShadowStrategy(migrationsPeek, declarativePeek); + // Peeked inputs are only reused where the acquire follows the peek IMMEDIATELY: the + // migrations acquire always does, the declarative one only under `parallel`. In the + // handoff (waits for the seam) and sequential (waits for the whole migrations + // provision) strategies the declarative acquire is DELAYED, and the key hashes + // `supabase/roles.sql` while the cold setup re-reads that file at its own time — + // reusing a stale peek there could publish a baseline under a key that no longer + // describes it (review: Codex on #6215). Re-resolving at acquire time also + // self-corrects a handoff whose key genuinely changed mid-run: the recomputed key + // misses the just-exported tar and the declarative side correctly cold-provisions + // with the current inputs. The key's OTHER live input, the JWKS resolver, is + // deliberately exempt from this refresh: it is memoized per shadow input + // (`legacyShadowRunInputFromLocalContainerInputs`), so the key and the baked baseline + // always carry the SAME value and cannot diverge; a delayed acquire keeps the + // command-start JWKS, well inside the staleness the snapshot cache accepts by design + // (a warm hit serves a tar up to 14 days old under its matching key). + const migrationsOpts = withPeek(cacheOpts(opts, "config"), migrationsPeek); + const declarativeOpts = + strategy === "parallel" + ? withPeek(cacheOpts(opts, "disabled"), declarativePeek) + : cacheOpts(opts, "disabled"); + + // In the concurrent strategies the declarative fiber's writes are buffered and + // flushed after the join, so nothing can land between two of the migrations fiber's + // live lines; sequential needs no buffer (one fiber at a time). `Effect.ensuring` on + // the JOIN (not the declarative fiber, which can finish first) so anomaly warnings + // survive failures without ever interleaving. + const buffered = + strategy === "sequential" ? undefined : legacyBufferedShadowOutput(output); + const provisions = legacyRunPlanShadowProvisions({ + strategy, + provisionMigrations: (onBaselineSeam) => + provisionMigrations(migrationsInput, migrationsOpts, onBaselineSeam), + provisionDeclarative: provisionDeclarative( + declarativeInput, + declarativeOpts, + buffered === undefined ? output : buffered.output, + ), + }); + const [migrations, declarative] = yield* buffered === undefined + ? provisions + : provisions.pipe(Effect.ensuring(buffered.flush)); return { migrationsUrl: migrations.migrationsUrl, declarativeUrl: declarative.declarativeUrl, - // Key equality is what encodes lineage: the declarative shadow restored the very tar - // the migrations side either restored or exported this run, so the two clusters are - // physical clones. An absent key (uncached/bypassed/uncachable) is never lineage. + // Key equality comes from the peeks (deterministic over inputs, not disk state), + // so a between-fibers eviction or publish cannot make it lie about lineage. allowSameDatabaseIdentity: legacyAllowSameDatabaseIdentityForPlanShadows({ declarativeRestoredFromPgDataSnapshot: declarative.restoredFromPgDataSnapshot, sameSnapshotKey: - migrations.snapshotKey !== undefined && - migrations.snapshotKey === declarative.snapshotKey, + migrationsPeek.state !== "uncachable" && + declarativePeek.state !== "uncachable" && + migrationsPeek.key === declarativePeek.key, }), } satisfies LegacyPgDeltaNextPlanShadows; }).pipe(Effect.mapError(nextShadowError)), diff --git a/apps/cli/src/legacy/commands/db/shared/legacy-pgdelta-next-shadow.plan.ts b/apps/cli/src/legacy/commands/db/shared/legacy-pgdelta-next-shadow.plan.ts new file mode 100644 index 0000000000..10d92a3997 --- /dev/null +++ b/apps/cli/src/legacy/commands/db/shared/legacy-pgdelta-next-shadow.plan.ts @@ -0,0 +1,160 @@ +/** + * Orchestration for pg-delta next's two plan shadows (migrations + declarative) — the strategy + * choice, the concurrency runner, and the output buffering that keeps the user-visible + * transcript free of cross-fiber interleaving. Extracted from + * `legacy-pgdelta-next-shadow.layer.ts` so the branch logic, the baseline-handoff signal, and + * the flush ordering are unit-testable with plain fakes instead of a full Docker/runtime layer + * graph. + * + * The three strategies, chosen from a {@link legacyPeekShadowBaseline} of each shadow: + * + * - `parallel` — both snapshots are published: both provisions warm-restore concurrently. A warm + * provision skips the platform baseline entirely (`legacySetupShadowDatabase`'s + * `baselinePresent` branch), so the declarative fiber prints nothing and the migrations fiber's + * `Applying migration ...` lines stream live and in order. + * - `baseline-handoff` — both are cold with the SAME cache key (webhooks agree): the baseline is + * paid exactly once. The migrations shadow cold-provisions; its snapshot export runs at the + * baseline seam (after platform setup, before migration replay) and signals the declarative + * fiber, which then warm-restores from the just-published tar CONCURRENTLY with the migration + * replay. All normal-mode output still comes from the single migrations fiber. + * - `sequential` — everything else (different keys, mixed warm/cold, `--no-cache`, cache env off, + * PG<=14/OrioleDB): no baseline can be shared, so run migrations then declarative exactly as + * the pre-parallel code did, preserving that transcript byte for byte. + */ + +import { Deferred, Effect } from "effect"; + +import { Output } from "../../../../shared/output/output.service.ts"; +import type { LegacyShadowBaselinePeek } from "../../../shared/db-bootstrap/shadow-cache.ts"; + +export type LegacyPlanShadowStrategy = "parallel" | "baseline-handoff" | "sequential"; + +/** + * Pure strategy choice from the two peeks. Equal-key implies equal warm/cold state (one key = + * one tar), so `cold`+`cold`+equal-keys is the only shareable-baseline shape; a mixed warm/cold + * pair always means different keys, where nothing can be shared and sequential keeps the cold + * side's baseline prints off the migration replay's live stream. + */ +export function legacyResolvePlanShadowStrategy( + migrations: LegacyShadowBaselinePeek, + declarative: LegacyShadowBaselinePeek, +): LegacyPlanShadowStrategy { + if (migrations.state === "warm" && declarative.state === "warm") return "parallel"; + if ( + migrations.state === "cold" && + declarative.state === "cold" && + migrations.key === declarative.key + ) { + return "baseline-handoff"; + } + return "sequential"; +} + +/** + * Runs the two provisions under the chosen strategy. + * + * `provisionMigrations` receives an `onBaselineSeam` effect it MUST arrange to run once its + * baseline seam passes (the snapshot-export point, before migration replay) — the layer wires it + * into the acquired handle's `snapshotBaseline` via `Effect.ensuring`, and fires it immediately + * when the acquired handle will never run a snapshot (a warm or uncached acquire, e.g. when + * another process published the tar between peek and acquire). The runner additionally + * `Effect.ensuring`s the signal onto the WHOLE migrations provision as a liveness backstop, so + * the declarative waiter can never deadlock: seam reached → early signal; provision ends without + * a seam (success, failure, or interruption) → backstop signal, and on failure `Effect.all`'s + * fail-fast interrupts the waiter anyway. + */ +export const legacyRunPlanShadowProvisions = (opts: { + readonly strategy: LegacyPlanShadowStrategy; + readonly provisionMigrations: (onBaselineSeam: Effect.Effect) => Effect.Effect; + readonly provisionDeclarative: Effect.Effect; +}): Effect.Effect => { + switch (opts.strategy) { + case "parallel": + return Effect.all([opts.provisionMigrations(Effect.void), opts.provisionDeclarative], { + concurrency: 2, + }); + case "baseline-handoff": + return Effect.gen(function* () { + const seam = yield* Deferred.make(); + const signal = Deferred.succeed(seam, undefined).pipe(Effect.asVoid); + return yield* Effect.all( + [ + opts.provisionMigrations(signal).pipe(Effect.ensuring(signal)), + Deferred.await(seam).pipe(Effect.andThen(opts.provisionDeclarative)), + ], + { concurrency: 2 }, + ); + }); + case "sequential": + return Effect.gen(function* () { + const migrations = yield* opts.provisionMigrations(Effect.void); + const declarative = yield* opts.provisionDeclarative; + return [migrations, declarative] as const; + }); + } +}; + +export interface LegacyBufferedShadowOutput { + /** The wrapped service to provide to the fiber whose writes must not interleave. */ + readonly output: typeof Output.Service; + /** + * Replays every buffered write to the real output, in order. Run it AFTER the live fiber has + * finished (e.g. `Effect.ensuring` on the join, not on the buffered fiber — the buffered fiber + * can finish first, and flushing then would interleave after all). Idempotent; writes arriving + * after a flush pass straight through live so late teardown warnings are never lost. + */ + readonly flush: Effect.Effect; +} + +/** + * An {@link Output} decorator that buffers `raw`/`rawBytes` (the only channels the shadow + * provisioning paths write to) and delegates everything else live. This is the hard guarantee + * that a concurrently provisioned shadow can never land a line BETWEEN two of the live fiber's + * lines — in normal mode the buffer stays empty (a warm restore prints nothing), so this exists + * for the anomaly paths: cache warnings and cold-fallback baseline prints. + * + * Deliberately NOT covering writes that bypass `Output` entirely (`SUPABASE_SHADOW_DEBUG` timing + * lines and failure-path container-log dumps write straight to `process.stderr`) — those are + * opt-in diagnostics where immediacy beats ordering. + */ +export function legacyBufferedShadowOutput( + real: typeof Output.Service, +): LegacyBufferedShadowOutput { + type BufferedWrite = + | { readonly kind: "raw"; readonly text: string; readonly stream: "stdout" | "stderr" } + | { + readonly kind: "rawBytes"; + readonly bytes: Uint8Array; + readonly stream: "stdout" | "stderr"; + }; + const buffer: Array = []; + let flushed = false; + const output = Output.of({ + ...real, + raw: (text, stream = "stdout") => + Effect.suspend(() => { + if (flushed) return real.raw(text, stream); + buffer.push({ kind: "raw", text, stream }); + return Effect.void; + }), + rawBytes: (bytes, stream = "stdout") => + Effect.suspend(() => { + if (flushed) return real.rawBytes(bytes, stream); + buffer.push({ kind: "rawBytes", bytes, stream }); + return Effect.void; + }), + }); + const flush = Effect.suspend(() => { + flushed = true; + const pending = buffer.splice(0); + return Effect.forEach( + pending, + (write) => + write.kind === "raw" + ? real.raw(write.text, write.stream) + : real.rawBytes(write.bytes, write.stream), + { discard: true }, + ); + }); + return { output, flush }; +} diff --git a/apps/cli/src/legacy/commands/db/shared/legacy-pgdelta-next-shadow.plan.unit.test.ts b/apps/cli/src/legacy/commands/db/shared/legacy-pgdelta-next-shadow.plan.unit.test.ts new file mode 100644 index 0000000000..3bb10e2f12 --- /dev/null +++ b/apps/cli/src/legacy/commands/db/shared/legacy-pgdelta-next-shadow.plan.unit.test.ts @@ -0,0 +1,247 @@ +import { describe, expect, it } from "@effect/vitest"; +import { Deferred, Effect, Exit, Option } from "effect"; + +import { mockOutput } from "../../../../../tests/helpers/mocks.ts"; +import { Output } from "../../../../shared/output/output.service.ts"; +import type { + LegacyShadowBaselinePeek, + LegacyShadowCacheKeyInputs, +} from "../../../shared/db-bootstrap/shadow-cache.ts"; +import { + legacyBufferedShadowOutput, + legacyResolvePlanShadowStrategy, + legacyRunPlanShadowProvisions, +} from "./legacy-pgdelta-next-shadow.plan.ts"; + +const keyInputs = (): LegacyShadowCacheKeyInputs => ({ + postgresImage: "public.ecr.aws/supabase/postgres:17.6.1.158", + majorVersion: 17, + jwtSecret: "super-secret-jwt-token-with-at-least-32-characters-long", + jwtExpiry: 3600, + rootKey: "d4dc5b6d4a1d6a10b2c1e5b6a7c8d9e0", + dbPassword: "postgres", + dbSettings: {}, + autoExposeNewTables: Option.none(), + storageTargetMigration: "", + webhooksEnabled: false, + rolesSql: "", + vault: [], + jwks: "", + services: { + realtime: { enabled: false, image: "" }, + storage: { enabled: false, image: "" }, + auth: { enabled: false, image: "" }, + }, +}); + +const warm = (key: string): LegacyShadowBaselinePeek => ({ + state: "warm", + key, + keyInputs: keyInputs(), +}); +const cold = (key: string): LegacyShadowBaselinePeek => ({ + state: "cold", + key, + keyInputs: keyInputs(), +}); +const uncachable: LegacyShadowBaselinePeek = { state: "uncachable" }; + +describe("legacyResolvePlanShadowStrategy", () => { + it("runs two published snapshots in parallel", () => { + expect(legacyResolvePlanShadowStrategy(warm("k"), warm("k"))).toBe("parallel"); + // Two warm tars under different keys restore independently — still parallel. + expect(legacyResolvePlanShadowStrategy(warm("a"), warm("b"))).toBe("parallel"); + }); + + it("hands the baseline off when both are cold under one key", () => { + expect(legacyResolvePlanShadowStrategy(cold("k"), cold("k"))).toBe("baseline-handoff"); + }); + + it("falls back to sequential when no baseline can be shared", () => { + expect(legacyResolvePlanShadowStrategy(cold("a"), cold("b"))).toBe("sequential"); + expect(legacyResolvePlanShadowStrategy(warm("a"), cold("b"))).toBe("sequential"); + expect(legacyResolvePlanShadowStrategy(cold("a"), warm("b"))).toBe("sequential"); + expect(legacyResolvePlanShadowStrategy(uncachable, uncachable)).toBe("sequential"); + expect(legacyResolvePlanShadowStrategy(uncachable, cold("k"))).toBe("sequential"); + expect(legacyResolvePlanShadowStrategy(warm("k"), uncachable)).toBe("sequential"); + }); +}); + +describe("legacyRunPlanShadowProvisions", () => { + it.effect("parallel: both provisions overlap in flight", () => + Effect.gen(function* () { + const log: string[] = []; + // Each side blocks until the other has started — this completes only under real + // concurrency; a sequential runner would deadlock (and trip the test timeout). + const migrationsStarted = yield* Deferred.make(); + const declarativeStarted = yield* Deferred.make(); + const [migrations, declarative] = yield* legacyRunPlanShadowProvisions({ + strategy: "parallel", + provisionMigrations: () => + Effect.gen(function* () { + log.push("migrations:start"); + yield* Deferred.succeed(migrationsStarted, undefined); + yield* Deferred.await(declarativeStarted); + log.push("migrations:done"); + return "m" as const; + }), + provisionDeclarative: Effect.gen(function* () { + log.push("declarative:start"); + yield* Deferred.succeed(declarativeStarted, undefined); + yield* Deferred.await(migrationsStarted); + log.push("declarative:done"); + return "d" as const; + }), + }); + expect(migrations).toBe("m"); + expect(declarative).toBe("d"); + expect(log.slice(0, 2).sort()).toEqual(["declarative:start", "migrations:start"]); + }), + ); + + it.effect( + "baseline-handoff: declarative starts only after the seam, concurrent with the replay", + () => + Effect.gen(function* () { + const log: string[] = []; + // The migrations side stays in "replay" until the declarative side has finished — + // proving the declarative provision ran BETWEEN the seam and the replay's end (i.e. + // concurrently with the replay), not after the whole migrations provision. + const declarativeDone = yield* Deferred.make(); + yield* legacyRunPlanShadowProvisions({ + strategy: "baseline-handoff", + provisionMigrations: (onBaselineSeam) => + Effect.gen(function* () { + log.push("migrations:baseline"); + yield* onBaselineSeam; + log.push("migrations:replay"); + yield* Deferred.await(declarativeDone); + log.push("migrations:done"); + return "m" as const; + }), + provisionDeclarative: Effect.gen(function* () { + log.push("declarative:start"); + yield* Deferred.succeed(declarativeDone, undefined); + return "d" as const; + }), + }); + expect(log.indexOf("declarative:start")).toBeGreaterThan( + log.indexOf("migrations:baseline"), + ); + expect(log.indexOf("declarative:start")).toBeLessThan(log.indexOf("migrations:done")); + }), + ); + + it.effect( + "baseline-handoff: a provision that never reaches the seam still releases the waiter", + () => + Effect.gen(function* () { + const log: string[] = []; + const [migrations, declarative] = yield* legacyRunPlanShadowProvisions({ + strategy: "baseline-handoff", + // Ignores `onBaselineSeam` entirely — a warm-raced or uncached acquire never runs a + // snapshot. The runner's own `Effect.ensuring` backstop must fire the signal when + // the provision ends, or the declarative waiter deadlocks. + provisionMigrations: () => + Effect.sync(() => { + log.push("migrations"); + return "m" as const; + }), + provisionDeclarative: Effect.sync(() => { + log.push("declarative"); + return "d" as const; + }), + }); + expect(migrations).toBe("m"); + expect(declarative).toBe("d"); + expect(log).toEqual(["migrations", "declarative"]); + }), + ); + + it.effect("baseline-handoff: a pre-seam failure interrupts the waiter instead of hanging", () => + Effect.gen(function* () { + let declarativeRan = false; + const exit = yield* legacyRunPlanShadowProvisions({ + strategy: "baseline-handoff", + provisionMigrations: () => Effect.fail("baseline exploded" as const), + provisionDeclarative: Effect.sync(() => { + declarativeRan = true; + return "d" as const; + }), + }).pipe(Effect.exit); + expect(Exit.isFailure(exit)).toBe(true); + expect(declarativeRan).toBe(false); + }), + ); + + it.effect("sequential: declarative starts only after migrations completes", () => + Effect.gen(function* () { + const log: string[] = []; + yield* legacyRunPlanShadowProvisions({ + strategy: "sequential", + provisionMigrations: () => + Effect.sync(() => { + log.push("migrations"); + return "m" as const; + }), + provisionDeclarative: Effect.sync(() => { + log.push("declarative"); + return "d" as const; + }), + }); + expect(log).toEqual(["migrations", "declarative"]); + }), + ); +}); + +describe("legacyBufferedShadowOutput", () => { + it.effect("holds writes until flush, then replays them after the live lines", () => { + const out = mockOutput(); + return Effect.gen(function* () { + const real = yield* Output; + const buffered = legacyBufferedShadowOutput(real); + // Simulates the parallel window: the buffered fiber's lines arrive in TIME between the + // live fiber's lines, but the flushed transcript keeps each fiber's block contiguous. + yield* real.raw("Applying migration a...\n", "stderr"); + yield* buffered.output.raw("Initialising schema...\n", "stderr"); + yield* real.raw("Applying migration b...\n", "stderr"); + yield* buffered.output.raw("Seeding globals from roles.sql...\n", "stderr"); + yield* buffered.flush; + expect(out.rawChunks.map((chunk) => chunk.text)).toEqual([ + "Applying migration a...\n", + "Applying migration b...\n", + "Initialising schema...\n", + "Seeding globals from roles.sql...\n", + ]); + }).pipe(Effect.provide(out.layer)); + }); + + it.effect("flush is idempotent and later writes pass straight through", () => { + const out = mockOutput(); + return Effect.gen(function* () { + const real = yield* Output; + const buffered = legacyBufferedShadowOutput(real); + yield* buffered.output.raw("buffered\n", "stderr"); + yield* buffered.flush; + yield* buffered.flush; + // A teardown warning arriving after the flush must not be swallowed. + yield* buffered.output.raw("late warning\n", "stderr"); + expect(out.rawChunks.map((chunk) => chunk.text)).toEqual(["buffered\n", "late warning\n"]); + }).pipe(Effect.provide(out.layer)); + }); + + it.effect("buffers rawBytes alongside raw, preserving arrival order and streams", () => { + const out = mockOutput(); + return Effect.gen(function* () { + const real = yield* Output; + const buffered = legacyBufferedShadowOutput(real); + yield* buffered.output.raw("first\n", "stderr"); + yield* buffered.output.rawBytes(new TextEncoder().encode("second\n"), "stderr"); + yield* buffered.flush; + expect(out.rawChunks).toEqual([ + { text: "first\n", stream: "stderr" }, + { text: "second\n", stream: "stderr" }, + ]); + }).pipe(Effect.provide(out.layer)); + }); +}); diff --git a/apps/cli/src/legacy/commands/db/shared/legacy-pgdelta-next-shadow.service.ts b/apps/cli/src/legacy/commands/db/shared/legacy-pgdelta-next-shadow.service.ts index d2e7e4f75e..ef9dcfa729 100644 --- a/apps/cli/src/legacy/commands/db/shared/legacy-pgdelta-next-shadow.service.ts +++ b/apps/cli/src/legacy/commands/db/shared/legacy-pgdelta-next-shadow.service.ts @@ -43,7 +43,12 @@ interface LegacyPgDeltaNextShadowShape { >; /** * Provisions the independent migrated and declarative shadows needed by a - * declarative plan. Both are removed when the current Effect scope closes. + * declarative plan. Concurrency is strategy-driven (see + * `legacy-pgdelta-next-shadow.plan.ts`): warm snapshots restore in parallel, + * a shared cold baseline is built once and handed off, and everything else + * runs sequentially — with the concurrent shapes buffering the declarative + * side's output so progress lines never interleave. Both shadows are removed + * when the current Effect scope closes. */ readonly provisionPlan: ( opts: LegacyPgDeltaNextShadowInput, diff --git a/apps/cli/src/legacy/shared/db-bootstrap/pgdata-snapshot.ts b/apps/cli/src/legacy/shared/db-bootstrap/pgdata-snapshot.ts index 6974cc5996..2ea195830d 100644 --- a/apps/cli/src/legacy/shared/db-bootstrap/pgdata-snapshot.ts +++ b/apps/cli/src/legacy/shared/db-bootstrap/pgdata-snapshot.ts @@ -222,6 +222,12 @@ export const legacyStampPgDataBaselineMarker = ( * stop/start around this call. The `rename` is the LAST step and is what publishes the entry: a * partially written tar must never be observable under the final name. Any failure removes the * temp file; nothing is left behind for a later run to find. + * + * The temp name is scoped by pid alone, so two exports to the same `tarPath` are safe across + * processes but NOT within one: a same-process concurrent writer's pre-clean would unlink this + * writer's live temp file, and the eventual `rename` could publish the other writer's + * half-written bytes under the final name. Callers own that serialization — `shadow-cache.ts` + * holds `legacyShadowExportMutex` around every call. */ export const legacyExportPgDataTar = ( spawner: Spawner, diff --git a/apps/cli/src/legacy/shared/db-bootstrap/shadow-cache.integration.test.ts b/apps/cli/src/legacy/shared/db-bootstrap/shadow-cache.integration.test.ts index a0f8d52e8c..4e90ae87c3 100644 --- a/apps/cli/src/legacy/shared/db-bootstrap/shadow-cache.integration.test.ts +++ b/apps/cli/src/legacy/shared/db-bootstrap/shadow-cache.integration.test.ts @@ -41,6 +41,7 @@ import { LEGACY_SHADOW_BASELINE_KEEP, LEGACY_SHADOW_CACHE_ENV, legacyAcquireShadowDatabase, + legacyPeekShadowBaseline, type LegacyShadowCacheOpts, } from "./shadow-cache.ts"; import { LEGACY_SHADOW_DEBUG_ENV } from "./shadow-debug.ts"; @@ -535,6 +536,57 @@ describe("legacyAcquireShadowDatabase", () => { ).pipe(Effect.provide(Layer.mergeAll(BunServices.layer, out.layer, cluster.layer))); }); + it.live("concurrent same-key cold snapshots publish exactly one intact tar", () => { + const docker = mockLegacyDockerDaemonCliSpawner(); + const cluster = fakeCluster(); + const out = mockOutput(); + return withShadowCacheHome( + "1", + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + // pg-delta next's parallel plan provisioning: both shadows acquire before either has + // published, so both go cold with the SAME key (the host port is not a key input), and + // both snapshot steps then race toward the same tar path. The export mutex must + // serialize them and the loser must skip — without it, both writers share the same + // pid-scoped temp file and the rename can publish half-written bytes. + const [first, second] = yield* Effect.all( + [ + legacyAcquireShadowDatabase(docker.spawner, shadowInput(fs, path)), + legacyAcquireShadowDatabase( + docker.spawner, + shadowInput(fs, path, { shadowPort: 54321 }), + ), + ], + { concurrency: 2 }, + ); + expect(first.baselinePresent).toBe(false); + expect(second.baselinePresent).toBe(false); + + yield* Effect.all([first.snapshotBaseline, second.snapshotBaseline], { concurrency: 2 }); + + // One export, one tar with the exported bytes intact, no leftover partials. + expect(docker.stepCalls("cp-out")).toHaveLength(1); + const tars = yield* soleTarName(fs, path); + expect(tars).toHaveLength(1); + expect(yield* fs.readFileString(path.join(shadowCacheDir(path), tars[0] ?? ""))).toBe( + expectedTarFor(tars[0] ?? ""), + ); + const leftovers = yield* fs.readDirectory(shadowCacheDir(path)); + expect(leftovers.filter((entry) => entry.includes("partial"))).toEqual([]); + + // Both shadows are back up after their own stop/start cycles — the skipped writer's + // container is revived exactly like the exporting one's. + expect(docker.containers.get(first.containerId)?.running).toBe(true); + expect(docker.containers.get(second.containerId)?.running).toBe(true); + + yield* legacyRemoveShadowDatabase(docker.spawner, first.containerId); + yield* legacyRemoveShadowDatabase(docker.spawner, second.containerId); + expect(docker.ids()).toEqual([]); + }), + ).pipe(Effect.provide(Layer.mergeAll(BunServices.layer, out.layer, cluster.layer))); + }); + it.live("publishing distinct keys keeps both tars until LRU/TTL eviction", () => { const docker = mockLegacyDockerDaemonCliSpawner(); const cluster = fakeCluster(); @@ -767,6 +819,9 @@ describe("legacyAcquireShadowDatabase", () => { const cold = yield* coldRun(docker, input); expect(yield* soleTarName(fs, path)).toHaveLength(1); + const [tarName = ""] = yield* soleTarName(fs, path); + const tarPath = path.join(shadowCacheDir(path), tarName); + const fallback = yield* legacyAcquireShadowDatabase(docker.spawner, input); expect(out.stderrText).toContain("cached shadow baseline unusable"); // Falls all the way back to a cold provision — a fresh container with no baseline. @@ -778,10 +833,13 @@ describe("legacyAcquireShadowDatabase", () => { // An extraction failure does NOT implicate the tar's contents (it could just as well be // a daemon hiccup), so the tar survives the fallback decision... expect(yield* soleTarName(fs, path)).toHaveLength(1); - // ...and the cold fallback's own export atomically republishes over it, so a genuinely - // corrupt tar still self-heals within this one run. + // ...and the cold fallback's own export atomically REPLACES it rather than treating the + // retained-but-unusable tar as a sibling's fresh publish and dedupe-skipping — the + // warm-fallback cold path must pass `skipIfPublished: false` (review: Codex on #6215). yield* fallback.snapshotBaseline; + expect(docker.stepCalls("cp-out")).toHaveLength(2); expect(yield* soleTarName(fs, path)).toHaveLength(1); + expect(yield* fs.readFileString(tarPath)).toBe(expectedTarFor(tarName)); }), ).pipe(Effect.provide(Layer.mergeAll(BunServices.layer, out.layer, cluster.layer))); }, @@ -1030,3 +1088,108 @@ describe("SUPABASE_SHADOW_DEBUG phase-timing instrumentation", () => { ).pipe(Effect.provide(Layer.mergeAll(BunServices.layer, out.layer, cluster.layer))); }); }); + +describe("legacyPeekShadowBaseline", () => { + it.live("reports cold before a snapshot exists, warm after, and uncachable on bypass", () => { + const docker = mockLegacyDockerDaemonCliSpawner(); + const cluster = fakeCluster(); + const out = mockOutput(); + return withShadowCacheHome( + "1", + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const input = shadowInput(fs, path); + + const before = yield* legacyPeekShadowBaseline(input); + expect(before.state).toBe("cold"); + + yield* coldRun(docker, input); + const after = yield* legacyPeekShadowBaseline(input); + expect(after.state).toBe("warm"); + // Same input, same key — the tar the cold run published is the one the peek found. + expect(after.state === "uncachable" || before.state === "uncachable").toBe(false); + if (after.state !== "uncachable" && before.state !== "uncachable") { + expect(after.key).toBe(before.key); + } + + // The handoff precondition: with config webhooks OFF, the migrations ("config") and + // declarative ("disabled") opts hash to the SAME key; forcing webhooks on re-keys. + const viaConfig = yield* legacyPeekShadowBaseline(input, { webhooks: "config" }); + const viaDisabled = yield* legacyPeekShadowBaseline(input, { webhooks: "disabled" }); + const viaEnabled = yield* legacyPeekShadowBaseline(input, { webhooks: "enabled" }); + if ( + viaConfig.state !== "uncachable" && + viaDisabled.state !== "uncachable" && + viaEnabled.state !== "uncachable" + ) { + expect(viaConfig.key).toBe(viaDisabled.key); + expect(viaEnabled.key).not.toBe(viaConfig.key); + } else { + expect.unreachable("cache-eligible input peeked as uncachable"); + } + + expect((yield* legacyPeekShadowBaseline(input, { bypassCache: true })).state).toBe( + "uncachable", + ); + }), + ).pipe(Effect.provide(Layer.mergeAll(BunServices.layer, out.layer, cluster.layer))); + }); + + it.live("reports uncachable when the cache env gate is off", () => { + const cluster = fakeCluster(); + const out = mockOutput(); + return withShadowCacheHome( + "0", + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + expect((yield* legacyPeekShadowBaseline(shadowInput(fs, path))).state).toBe("uncachable"); + }), + ).pipe(Effect.provide(Layer.mergeAll(BunServices.layer, out.layer, cluster.layer))); + }); + + it.live("acquire reuses peeked key inputs instead of re-resolving JWKS", () => { + const docker = mockLegacyDockerDaemonCliSpawner(); + const cluster = fakeCluster(); + const out = mockOutput(); + return withShadowCacheHome( + "1", + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + // Realtime enabled on PG17 is the one gate that makes the cache key consume the JWKS + // effect — a live third-party discovery request in production, so resolving it once + // per run (peek) rather than once per peek AND once per acquire is the contract. + let jwksResolutions = 0; + const base = shadowInput(fs, path); + const input: typeof base = { + ...base, + setup: { + ...base.setup, + config: { + ...defaultConfig, + realtime: { ...defaultConfig.realtime, enabled: true }, + }, + jwks: Effect.sync(() => { + jwksResolutions += 1; + return '{"keys":[]}'; + }), + }, + }; + + const peek = yield* legacyPeekShadowBaseline(input); + expect(peek.state).toBe("cold"); + expect(jwksResolutions).toBe(1); + + const handle = yield* legacyAcquireShadowDatabase( + docker.spawner, + input, + peek.state === "uncachable" ? {} : { precomputedKeyInputs: peek.keyInputs }, + ); + expect(jwksResolutions).toBe(1); + yield* legacyRemoveShadowDatabase(docker.spawner, handle.containerId); + }), + ).pipe(Effect.provide(Layer.mergeAll(BunServices.layer, out.layer, cluster.layer))); + }); +}); diff --git a/apps/cli/src/legacy/shared/db-bootstrap/shadow-cache.ts b/apps/cli/src/legacy/shared/db-bootstrap/shadow-cache.ts index c4a852c4c1..4024bafd4a 100644 --- a/apps/cli/src/legacy/shared/db-bootstrap/shadow-cache.ts +++ b/apps/cli/src/legacy/shared/db-bootstrap/shadow-cache.ts @@ -20,7 +20,10 @@ * a warm-path anomaly cold-provisions (deleting the tar only when its contents are implicated — * see `LegacyShadowCacheUnavailable.tarSuspect`), a cold export failure only warns and leaves the * run uncached (with ONE deliberate exception: a shadow that fails to come back up after the - * snapshot fails the run — see `legacyExportShadowBaseline`); tars live under the global + * snapshot fails the run — see `legacyExportShadowBaseline`); same-PROCESS concurrent exports are + * additionally serialized by an in-process mutex, because `legacyExportPgDataTar`'s temp name is + * pid-scoped and two fibers of one process would otherwise share it (see + * {@link legacyWriteShadowBaselineTar}); tars live under the global * `${SUPABASE_HOME}/cache/shadow-baseline/` (shared across worktrees with the same settings), * with LRU (keep 8) + 14-day mtime TTL retention. `SUPABASE_SHADOW_CACHE` is ON by default; * `false`/`0` opts out. @@ -29,7 +32,7 @@ import { createHash } from "node:crypto"; import type { ProjectConfig } from "@supabase/config"; -import { Clock, Effect, Option, Result, type FileSystem } from "effect"; +import { Clock, Effect, Option, Result, Semaphore, type FileSystem } from "effect"; import type { ChildProcessSpawner } from "effect/unstable/process/ChildProcessSpawner"; import { Output } from "../../../shared/output/output.service.ts"; @@ -735,34 +738,65 @@ const legacyAwaitShadowReady = ( // Cold export // --------------------------------------------------------------------------- +/** + * Serializes same-process cold exports. Two shadows provisioned concurrently in one process + * (pg-delta next's plan shadows — `legacy-pgdelta-next-shadow.layer.ts`'s `provisionPlan`) can + * both reach the export step, and with an equal key (the declarative and migrations shadows + * hash identically whenever their effective webhooks booleans agree) they would race on the + * SAME `..partial` temp path — `legacyExportPgDataTar` scopes its temp name by pid + * alone, so the second writer's pre-clean unlinks the first's live temp file, and the first's + * rename would then publish the second's half-written bytes under the final name. One permit + * makes that interleaving impossible; cross-PROCESS writers were never affected (distinct + * pids). Exports run only on the cold path and take seconds, so the serialization is invisible + * outside a double-cold first run. + */ +const legacyShadowExportMutex = Semaphore.makeUnsafe(1); + /** * Ensures the tar's global cache directory exists, delegates the actual export to * {@link legacyExportPgDataTar} (`pgdata-snapshot.ts` — see that function's own doc comment for - * the atomic-publish mechanics), then applies the LRU + TTL retention rule. + * the atomic-publish mechanics), then applies the LRU + TTL retention rule. Runs under + * {@link legacyShadowExportMutex}. + * + * `skipIfPublished` dedupes same-key sibling exports: when the tar was ABSENT at acquire time + * (the `!cached` cold path), one published while this fiber waited on the permit is a sibling's + * snapshot of this same baseline, so re-exporting would only re-move ~90MB to replace equivalent + * bytes. It must be `false` on the warm-fallback cold path, where a tar deliberately RETAINED + * despite an unusable restore (see `LegacyShadowCacheUnavailable.tarSuspect`) is sitting at this + * exact path waiting to be atomically replaced — skipping there would leave a genuinely corrupt + * tar in place forever, failing every later warm restore into another cold provision (review: + * Codex on #6215). */ const legacyWriteShadowBaselineTar = ( spawner: Spawner, input: LegacyShadowSetupInput, tarPath: string, containerId: string, + skipIfPublished: boolean, ): Effect.Effect => - Effect.gen(function* () { - const cacheDir = legacyShadowBaselineCacheDir(input.path); - yield* input.fs - .makeDirectory(cacheDir, { recursive: true, mode: 0o700 }) - .pipe( - Effect.mapError((cause) => - legacyShadowCacheUnavailable(`failed to create ${cacheDir}: ${cause.message}`), + legacyShadowExportMutex.withPermit( + Effect.gen(function* () { + if (skipIfPublished) { + const published = yield* input.fs.exists(tarPath).pipe(Effect.orElseSucceed(() => false)); + if (published) return; + } + const cacheDir = legacyShadowBaselineCacheDir(input.path); + yield* input.fs + .makeDirectory(cacheDir, { recursive: true, mode: 0o700 }) + .pipe( + Effect.mapError((cause) => + legacyShadowCacheUnavailable(`failed to create ${cacheDir}: ${cause.message}`), + ), + ); + yield* legacySweepAbandonedShadowBaselinePartials(input); + yield* legacyExportPgDataTar(spawner, containerId, input.fs, tarPath).pipe( + Effect.mapError((cause: LegacyPgDataSnapshotUnavailable) => + legacyShadowCacheUnavailable(cause.reason), ), ); - yield* legacySweepAbandonedShadowBaselinePartials(input); - yield* legacyExportPgDataTar(spawner, containerId, input.fs, tarPath).pipe( - Effect.mapError((cause: LegacyPgDataSnapshotUnavailable) => - legacyShadowCacheUnavailable(cause.reason), - ), - ); - yield* legacySweepShadowBaselineRetention(input); - }); + yield* legacySweepShadowBaselineRetention(input); + }), + ); /** * The cold path's snapshot step, run at the baseline/migrations seam — after @@ -790,6 +824,7 @@ const legacyExportShadowBaseline = ( key: string, tarPath: string, containerId: string, + skipIfPublished: boolean, ): Effect.Effect => legacyTimeShadowPhase( "baseline-export", @@ -814,7 +849,13 @@ const legacyExportShadowBaseline = ( legacyShadowCacheUnavailable(cause.reason), ), ); - yield* legacyWriteShadowBaselineTar(spawner, input, tarPath, containerId); + yield* legacyWriteShadowBaselineTar( + spawner, + input, + tarPath, + containerId, + skipIfPublished, + ); }), ); // Run-critical phase: the shadow must be back up and answering before this step reports @@ -862,8 +903,67 @@ const legacyExportShadowBaseline = ( export interface LegacyShadowCacheOpts { readonly bypassCache?: boolean; readonly webhooks?: LegacySetupDatabaseOptions["webhooks"]; + /** + * Key inputs a caller already resolved via {@link legacyPeekShadowBaseline}, so + * {@link legacyAcquireShadowDatabase} does not resolve them a second time. Resolution is not + * idempotent-cheap: it can include a live JWKS discovery request (realtime on PG15+), so a + * peek-then-acquire caller passing this through halves that traffic. MUST have been computed + * from the same `input`/`opts` pair, or the acquire keys against the wrong snapshot. + */ + readonly precomputedKeyInputs?: LegacyShadowCacheKeyInputs; } +/** What {@link legacyPeekShadowBaseline} learned about a would-be acquire, without provisioning anything. */ +export type LegacyShadowBaselinePeek = + /** The cache cannot apply: bypassed, env-disabled, or key-ineligible (PG<=14, OrioleDB, unreadable roles.sql). */ + | { readonly state: "uncachable" } + | { + readonly state: "cold" | "warm"; + readonly key: string; + /** Pass back via {@link LegacyShadowCacheOpts.precomputedKeyInputs} to skip re-resolution. */ + readonly keyInputs: LegacyShadowCacheKeyInputs; + }; + +/** + * Answers "what would {@link legacyAcquireShadowDatabase} do for this input right now?" without + * creating a container: `warm` (a snapshot for this key is published), `cold` (cache-enabled but + * no snapshot yet), or `uncachable`. Callers use it to CHOOSE an orchestration (pg-delta next's + * plan provisioning picks parallel / baseline-handoff / sequential — see + * `legacy-pgdelta-next-shadow.plan.ts`), never to skip the acquire's own re-checks: the answer + * can go stale between peek and acquire (another process publishes or evicts the tar), and the + * acquire re-deciding on current state is what keeps that race merely suboptimal rather than + * incorrect. + * + * The error channel is the key resolution's own `E` (a JWKS resolution failure) — same rationale + * as {@link legacyAcquireShadowDatabase}: a real cold provision at this input would have failed + * the same way, so it must not be folded into `uncachable`. + */ +export const legacyPeekShadowBaseline = ( + input: LegacyShadowSetupInput, + opts: LegacyShadowCacheOpts = {}, +): Effect.Effect => + Effect.gen(function* () { + if ( + opts.bypassCache === true || + !legacyShadowCacheEnabled(process.env, input.setup.projectEnvValues) + ) { + return { state: "uncachable" } as const; + } + const keyInputs = yield* legacyResolveShadowCacheKeyInputs(input, opts); + if (Option.isNone(keyInputs)) return { state: "uncachable" } as const; + const key = legacyShadowCacheKey(keyInputs.value); + const tarPath = input.path.join( + legacyShadowBaselineCacheDir(input.path), + legacyShadowBaselineTarFileName(key), + ); + const cached = yield* input.fs.exists(tarPath).pipe(Effect.orElseSucceed(() => false)); + return { + state: cached ? ("warm" as const) : ("cold" as const), + key, + keyInputs: keyInputs.value, + }; + }); + /** * What `Effect.acquireUseRelease`'s `acquire` hands the `use` phase: the container, whether its * cluster already carries the platform baseline, and the snapshot step to run once a fresh @@ -905,12 +1005,17 @@ const legacyUncachedShadow = ( * destroys an `--rm` container the moment it exits. Release still removes it with `docker rm -f * -v`, so the container's lifetime is unchanged — see * {@link LegacyCreateShadowDatabaseInput.autoRemove}. + * + * `skipIfPublished` MUST reflect whether the tar was absent when this cold acquisition began — + * see {@link legacyWriteShadowBaselineTar} for what each value means and why the warm-fallback + * caller must pass `false`. */ const legacyColdCachedShadow = ( spawner: Spawner, input: LegacyShadowSetupInput, key: string, tarPath: string, + skipIfPublished: boolean, ): Effect.Effect => legacyCreateShadowDatabase(spawner, { ...input, autoRemove: false }).pipe( Effect.map(({ containerId }) => ({ @@ -918,7 +1023,14 @@ const legacyColdCachedShadow = ( snapshotKey: key, baselinePresent: false, snapshotRequired: true, - snapshotBaseline: legacyExportShadowBaseline(spawner, input, key, tarPath, containerId), + snapshotBaseline: legacyExportShadowBaseline( + spawner, + input, + key, + tarPath, + containerId, + skipIfPublished, + ), })), ); @@ -1073,8 +1185,13 @@ export const legacyAcquireShadowDatabase = ( // nothing has been acquired yet — and the JWKS effect inside can be a real third-party // discovery request, which must not pin a Ctrl-C for its whole duration (review: Codex on // #6184). Interruption here simply means no container was ever created, so there is nothing - // for a finalizer to release. - const keyInputs = yield* Effect.interruptible(legacyResolveShadowCacheKeyInputs(input, opts)); + // for a finalizer to release. A caller that already peeked passes its resolved inputs + // through ({@link LegacyShadowCacheOpts.precomputedKeyInputs}) so the JWKS request is not + // repeated; the tar-existence check below always re-runs on current state. + const keyInputs = + opts.precomputedKeyInputs !== undefined + ? Option.some(opts.precomputedKeyInputs) + : yield* Effect.interruptible(legacyResolveShadowCacheKeyInputs(input, opts)); if (Option.isNone(keyInputs)) return yield* legacyUncachedShadow(spawner, input); const key = legacyShadowCacheKey(keyInputs.value); const tarPath = input.path.join( @@ -1083,7 +1200,9 @@ export const legacyAcquireShadowDatabase = ( ); const cached = yield* input.fs.exists(tarPath).pipe(Effect.orElseSucceed(() => false)); - if (!cached) return yield* legacyColdCachedShadow(spawner, input, key, tarPath); + // The tar is absent at acquire time, so a tar found at export time can only be a + // same-key sibling's fresh publish — dedupe against it. + if (!cached) return yield* legacyColdCachedShadow(spawner, input, key, tarPath, true); // Warm hits refresh mtime (so frequently used keys survive LRU/TTL) and sweep abandoned // partials — a killed concurrent writer's leftover would otherwise persist indefinitely once @@ -1108,7 +1227,11 @@ export const legacyAcquireShadowDatabase = ( if (cause.tarSuspect === true) { yield* legacyForgetShadowBaselineTar(input.fs, tarPath); } - return yield* legacyColdCachedShadow(spawner, input, key, tarPath); + // No dedupe on this path: unless it was suspect (deleted above), the unusable tar is + // still sitting at this exact path, deliberately retained so this fallback's own + // export atomically REPLACES it — skipping because "a tar exists" would leave a + // genuinely corrupt one in place forever (review: Codex on #6215). + return yield* legacyColdCachedShadow(spawner, input, key, tarPath, false); }), ), );