From 5a452915c79b0e7bb71b52d389e35d93d6d9ff29 Mon Sep 17 00:00:00 2001 From: Elliot Braem Date: Fri, 16 Oct 2026 14:52:22 -0500 Subject: [PATCH] =?UTF-8?q?fix(deploy):=20#430=20review=20fixes=20?= =?UTF-8?q?=E2=80=94=20slot-transfer=20semaphore,=20surfaced=20upload=20re?= =?UTF-8?q?jections,=20build-mode=20oracle=20key,=20probe=20fallback?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/services/deploy/train.ts | 68 +++++++++++++------ .../src/services/workspace/build-train.ts | 7 +- packages/everything-dev/src/workspace.ts | 6 +- .../tests/unit/build-staleness.test.ts | 17 +++++ .../tests/unit/upload-semaphore.test.ts | 61 +++++++++++++++++ 5 files changed, 137 insertions(+), 22 deletions(-) create mode 100644 packages/everything-dev/tests/unit/upload-semaphore.test.ts diff --git a/packages/everything-dev/src/services/deploy/train.ts b/packages/everything-dev/src/services/deploy/train.ts index 79808356..20e0f08d 100644 --- a/packages/everything-dev/src/services/deploy/train.ts +++ b/packages/everything-dev/src/services/deploy/train.ts @@ -123,9 +123,11 @@ export async function waitForPublishedConfig(opts: { continue; } } catch (historyError) { - // A history probe failure must not starve the wait — fall back to + // A history probe failure must not starve the wait — the failed + // probe consumes a skip credit, so repeated failures fall back to // the plain equality poll. lastError = historyError; + txProbeSkips += 1; } } @@ -192,6 +194,36 @@ export interface TrainDeps { }; } +/** + * A counting semaphore with slot transfer: `release` hands its slot to a + * waiter (count conserved) instead of freeing it, so a burst of waiters can + * never push concurrency past the limit — the hand-rolled version leaked one + * count per transfer and over-admitted under late build completions (#430). + */ +export function createSemaphore(limit: number): { + acquire: () => Promise<() => void>; +} { + let inFlight = 0; + const waiters: Array<() => void> = []; + return { + acquire: () => { + if (inFlight < limit) { + inFlight++; + return Promise.resolve(release); + } + return new Promise<() => void>((resolve) => waiters.push(() => resolve(release))); + }, + }; + function release(): void { + const wake = waiters.shift(); + if (wake) { + wake(); + } else { + inFlight--; + } + } +} + export interface DeployTrainInput { bosConfig: BosConfig; runtimeConfig: RuntimeConfig | null; @@ -453,6 +485,7 @@ export async function runPublishTrain( // target order (byte-identical to the serial train); the first failure in // that order is the one thrown. const uploadConcurrency = Math.max(1, Number(process.env.BOS_UPLOAD_CONCURRENCY) || 4); + const uploadRejections: Array<{ key: string; error: unknown }> = []; interface UploadUnitResult { entries: DeployResultEntry[]; detail: UploadUnitDetail; @@ -460,35 +493,26 @@ export async function runPublishTrain( const uploaded = new Map>(); // Rejections stay in the map until the drain's allSettled — deleting would // both swallow the failure (the drain substitutes an empty result) and let - // the fallback loop silently re-attempt the upload. The immediate no-op - // catch keeps the pipelined rejection handled while the build continues. + // the fallback loop silently re-attempt the upload. The early catch records + // the reason so a build failure thrown before the drain can still name the + // parallel upload that died. const trackUpload = (key: string, pending: Promise) => { uploaded.set(key, pending); - pending.catch(() => {}); - }; - let inFlight = 0; - const waiters: Array<() => void> = []; - const acquireUploadSlot = () => { - if (inFlight < uploadConcurrency) { - inFlight++; - return Promise.resolve(); - } - return new Promise((resolve) => waiters.push(resolve)); - }; - const releaseUploadSlot = () => { - inFlight--; - waiters.shift()?.(); + pending.catch((error) => { + uploadRejections.push({ key, error }); + }); }; + const uploadSlots = createSemaphore(uploadConcurrency); const startUpload = (key: string) => { if (uploaded.has(key)) return; trackUpload( key, (async () => { - await acquireUploadSlot(); + const release = await uploadSlots.acquire(); try { return await uploadOneWorkspace(key); } finally { - releaseUploadSlot(); + release(); } })(), ); @@ -540,13 +564,17 @@ export async function runPublishTrain( console.log(colors.dim(" Run with --verbose for full build output.")); console.log(); } + const uploadNote = + uploadRejections.length > 0 + ? `; a parallel upload also failed (${uploadRejections[0].error instanceof Error ? uploadRejections[0].error.message : String(uploadRejections[0].error)})` + : ""; throw new PublishTrainError({ kind: "build", registryUrl, built, skipped, deployResults, - error: `${failures.length} of ${total} workspaces failed to deploy`, + error: `${failures.length} of ${total} workspaces failed to deploy${uploadNote}`, }); } } diff --git a/packages/everything-dev/src/services/workspace/build-train.ts b/packages/everything-dev/src/services/workspace/build-train.ts index f13f979f..7b679f25 100644 --- a/packages/everything-dev/src/services/workspace/build-train.ts +++ b/packages/everything-dev/src/services/workspace/build-train.ts @@ -294,7 +294,12 @@ export async function buildWorkspaceTargets( force, env, ...(input.deploy && root && !force - ? { oracle: contentHashStalenessOracle(root.dir, root.members) } + ? { + oracle: contentHashStalenessOracle(root.dir, root.members, { + nodeEnv: input.deploy ? "production" : "development", + deploy: input.deploy, + }), + } : {}), }, ); diff --git a/packages/everything-dev/src/workspace.ts b/packages/everything-dev/src/workspace.ts index 75898fad..5bf64b6c 100644 --- a/packages/everything-dev/src/workspace.ts +++ b/packages/everything-dev/src/workspace.ts @@ -373,9 +373,13 @@ async function memberInputHash( export function contentHashStalenessOracle( rootDir: string, members: readonly WorkspaceMember[], + buildMode?: { nodeEnv?: string; deploy?: boolean }, ): StalenessOracle { const cachePath = join(rootDir, ".bos", "build-cache.json"); - const envKey = `${process.env.NODE_ENV ?? ""}|${process.env.DEPLOY ?? ""}`; + const envKey = + buildMode === undefined + ? `${process.env.NODE_ENV ?? ""}|${process.env.DEPLOY ?? ""}` + : `${buildMode.nodeEnv ?? ""}|${buildMode.deploy === true ? "deploy" : "build"}`; const byName = new Map(members.map((member) => [member.name, member])); const readCache = (): Record => { diff --git a/packages/everything-dev/tests/unit/build-staleness.test.ts b/packages/everything-dev/tests/unit/build-staleness.test.ts index 7d176f9c..3da863c9 100644 --- a/packages/everything-dev/tests/unit/build-staleness.test.ts +++ b/packages/everything-dev/tests/unit/build-staleness.test.ts @@ -177,4 +177,21 @@ describe("contentHashStalenessOracle", () => { const second = contentHashStalenessOracle(root, [member]); expect(await second.isStale(member)).toBe(false); }); + + it("judges against the build mode it was given, not the shell env", async () => { + const root = mkTmp(join(tmpdir(), "bos-hash-root-")); + const dir = makePackageDir(); + const member = makeMember(dir, "a"); + const deploy = { nodeEnv: "production", deploy: true }; + + await contentHashStalenessOracle(root, [member], deploy).recordBuilt?.(member); + const judge = contentHashStalenessOracle(root, [member], deploy); + expect(await judge.isStale(member)).toBe(false); + + const otherMode = contentHashStalenessOracle(root, [member], { + nodeEnv: "development", + deploy: false, + }); + expect(await otherMode.isStale(member)).toBe(true); + }); }); diff --git a/packages/everything-dev/tests/unit/upload-semaphore.test.ts b/packages/everything-dev/tests/unit/upload-semaphore.test.ts new file mode 100644 index 00000000..4f8f2524 --- /dev/null +++ b/packages/everything-dev/tests/unit/upload-semaphore.test.ts @@ -0,0 +1,61 @@ +import { describe, expect, it } from "vitest"; +import { createSemaphore } from "../../src/services/deploy/train"; + +describe("createSemaphore", () => { + it("hands the released slot to a waiter instead of freeing it", async () => { + const sem = createSemaphore(2); + const releaseOne = await sem.acquire(); + const releaseTwo = await sem.acquire(); + + let thirdAcquired = false; + const third = sem.acquire().then((release) => { + thirdAcquired = true; + return release; + }); + let fourthAcquired = false; + const fourth = sem.acquire().then(() => { + fourthAcquired = true; + }); + await Promise.resolve(); + expect(thirdAcquired).toBe(false); + + releaseOne(); + await third; + expect(thirdAcquired).toBe(true); + releaseTwo(); + await fourth; + expect(fourthAcquired).toBe(true); + + let fifthAcquired = false; + const fifth = sem.acquire().then(() => { + fifthAcquired = true; + }); + await Promise.resolve(); + expect(fifthAcquired).toBe(false); + + (await third)(); + await fifth; + expect(fifthAcquired).toBe(true); + }); + + it("never over-admits across a long acquire/release churn", async () => { + const sem = createSemaphore(3); + let current = 0; + let max = 0; + + const work = async (): Promise => { + const release = await sem.acquire(); + current += 1; + max = Math.max(max, current); + await Promise.resolve(); + current -= 1; + release(); + }; + + for (let round = 0; round < 20; round++) { + await Promise.all(Array.from({ length: 10 }, () => work())); + } + + expect(max).toBe(3); + }); +});