Make the scheduler delivery-bound assertion deterministic

packages/gatekeeper-scheduler/__tests__/schedule-driver.test.ts

@@ -30,6 +30,7 @@ type TestHooks = HookInitiator<ScheduleHookTarget> & {
   }>;
   release(): Promise<void>;
   reset(): Promise<void>;
+  setCallbackBarrier(size: number, limit: number): Promise<void>;
   waitUntilBlocked(): Promise<void>;
 };
 
@@ -696,6 +697,9 @@ describe("ScheduleDriver", () => {
   it("bounds callback concurrency and immediately continues a due backlog", async () => {
     const debug = vi.spyOn(console, "debug").mockImplementation(() => {});
     const driver = testEnv.SCHEDULE_DRIVER.getByName("batching");
+    // The batch of 20 delivers as five full sets of lanes; hold each set so the observed peak is
+    // the runner's bound and not RPC timing. The 21st, a round of one, is left unbarriered.
+    await testEnv.TEST_HOOKS.setCallbackBarrier(4, 20);
     const activationTime = Date.now();
     for (let index = 0; index < 21; index++) {
       await enableSchedule(

packages/gatekeeper-scheduler/__tests__/worker.ts

@@ -24,6 +24,19 @@ let disposedApprovalQueues = 0;
 let disposedCallbacks = 0;
 let blockPoint: BlockPoint | null = null;
 let blockedPoint: BlockPoint | null = null;
+let callbackBarrier = 1;
+let callbackBarrierLimit = 0;
+
+// Holds a callback until `callbackBarrier` of them are in flight, so a test observing peak
+// concurrency measures the delivery bound rather than how RPC round-trips happened to interleave.
+// It applies only to the first `callbackBarrierLimit` callbacks, which the caller sizes to a whole
+// number of full rounds -- a partial round could never reach the barrier, and waiting on a count
+// that will arrive keeps this free of any wall-clock assumption.
+async function waitForCallbackBarrier(): Promise<void> {
+  if (callbackScheduleIds.length > callbackBarrierLimit) return;
+  // eslint-disable-next-line no-unmodified-loop-condition -- sibling callbacks mutate this.
+  while (activeCallbacks < callbackBarrier) await new Promise((resolve) => setTimeout(resolve, 1));
+}
 
 async function pauseIfBlocked(point: BlockPoint): Promise<void> {
   if (blockPoint !== point) return;
@@ -54,6 +67,7 @@ class TestCallback extends RpcTarget {
     try {
       await pauseIfBlocked("callback");
       if (mode === "callback-reject") throw new Error("callback rejected");
+      await waitForCallbackBarrier();
       await new Promise((resolve) => setTimeout(resolve, 10));
     } finally {
       activeCallbacks--;
@@ -83,6 +97,11 @@ export class TestHooks extends WorkerEntrypoint {
     blockedPoint = null;
   }
 
+  setCallbackBarrier(size: number, limit: number): void {
+    callbackBarrier = size;
+    callbackBarrierLimit = limit;
+  }
+
   async waitUntilBlocked(): Promise<void> {
     // eslint-disable-next-line no-unmodified-loop-condition -- pauseIfBlocked() mutates this via RPC.
     while (blockedPoint === null) await new Promise((resolve) => setTimeout(resolve, 1));
@@ -112,6 +131,8 @@ export class TestHooks extends WorkerEntrypoint {
     disposedApprovalQueues = 0;
     disposedCallbacks = 0;
     blockedPoint = null;
+    callbackBarrier = 1;
+    callbackBarrierLimit = 0;
   }
 }