perf(deploy): ticket #430 deploy speed — isolated from the CLI split, with review fixes

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,42 +485,34 @@ 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;
   }
   const uploaded = new Map<string, Promise<UploadUnitResult>>();
   // 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<UploadUnitResult>) => {
     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<void>((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}`,
         });
       }
     }

packages/everything-dev/src/services/workspace/build-train.ts

@@ -294,7 +294,12 @@
       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,
+            }),
+          }
         : {}),
     },
   );
@@ -387,7 +392,7 @@
   fn: (item: T, index: number) => Promise<R>,
   limit: number,
 ): Promise<PromiseSettledResult<R>[]> {
   const results: PromiseSettledResult<R>[] = new Array(items.length);
   let next = 0;
   const workers = Array.from({ length: Math.max(1, Math.min(limit, items.length)) }, async () => {
     while (true) {

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<string, { hash?: string }> => {

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

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<void> => {
+      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);
+  });
+});