-
Notifications
You must be signed in to change notification settings - Fork 515
perf(cli): strategy-driven parallel provisioning for pg-delta next plan shadows #6215
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Open
Open
Changes from all commits
Commits
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
155 changes: 155 additions & 0 deletions
155
apps/cli/src/legacy/commands/db/shared/legacy-pgdelta-next-shadow.plan.ts
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,155 @@ | ||
| /** | ||
| * 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, 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. | ||
| * - `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. | ||
| */ | ||
|
|
||
| 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"; | ||
|
avallete marked this conversation as resolved.
|
||
| } | ||
| 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). The runner | ||
| * additionally `Effect.ensuring`s the signal onto the whole migrations provision as a liveness | ||
| * backstop, so the declarative waiter can never deadlock. | ||
| */ | ||
| export const legacyRunPlanShadowProvisions = <M, D, EM, ED, RM, RD>(opts: { | ||
| readonly strategy: LegacyPlanShadowStrategy; | ||
| readonly provisionMigrations: (onBaselineSeam: Effect.Effect<void>) => Effect.Effect<M, EM, RM>; | ||
| readonly provisionDeclarative: Effect.Effect<D, ED, RD>; | ||
| }): Effect.Effect<readonly [M, D], EM | ED, RM | RD> => { | ||
| 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<void>(); | ||
| 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 (`Effect.ensuring` on the join, not on the buffered fiber — the buffered fiber can | ||
| * finish first). Idempotent; writes arriving after a flush pass straight through live so late | ||
| * teardown warnings are never lost. | ||
| */ | ||
| readonly flush: Effect.Effect<void>; | ||
| } | ||
|
|
||
| /** | ||
| * 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`). | ||
| */ | ||
| 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<BufferedWrite> = []; | ||
| 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 }; | ||
| } | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.