diff --git a/packages/gatekeeper-scheduler/__tests__/schedule-driver.test.ts b/packages/gatekeeper-scheduler/__tests__/schedule-driver.test.ts index 3c99a8f46..eb2ffacea 100644 --- a/packages/gatekeeper-scheduler/__tests__/schedule-driver.test.ts +++ b/packages/gatekeeper-scheduler/__tests__/schedule-driver.test.ts @@ -30,6 +30,7 @@ type TestHooks = HookInitiator & { }>; release(): Promise; reset(): Promise; + setCallbackBarrier(size: number, limit: number): Promise; waitUntilBlocked(): Promise; }; @@ -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( diff --git a/packages/gatekeeper-scheduler/__tests__/worker.ts b/packages/gatekeeper-scheduler/__tests__/worker.ts index 718b61669..f0d0d4446 100644 --- a/packages/gatekeeper-scheduler/__tests__/worker.ts +++ b/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 { + 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 { 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 { // 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; } }