From f8d8003fea7160e0f95abb7bdd6ebba14849a75f Mon Sep 17 00:00:00 2001 From: Juan Castro Date: Wed, 30 Sep 2026 12:25:40 -0400 Subject: [PATCH] test: stop the delivery-backoff test reading holds against a moving clock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `holds one it could not answer, for longer each time` freezes `Date` so a hold written as `now + backoff` reads back as exactly `backoff`. But every wait in the file went through `vi.waitFor`, which advances the faked clock once per poll — measured at 50ms a poll. Each poll between the hold being written and read therefore ate into the value: CI read a 2s hold as 525ms and failed, on main one run and on this branch the next. Polling now goes through `waitUntil`, which retries on `performance.now` and real `setTimeout` — neither faked — so the clock only moves where `jump` moves it. The holds then read exactly 2s/4s/8s/16s, so the assertion is an equality rather than a one-second window that a slow enough runner could always slide out of. --- .../events/workerInvoker.integration.test.ts | 89 ++++++++++++------- 1 file changed, 55 insertions(+), 34 deletions(-) diff --git a/src/backend/services/events/workerInvoker.integration.test.ts b/src/backend/services/events/workerInvoker.integration.test.ts index e82d790ee..e06069f88 100644 --- a/src/backend/services/events/workerInvoker.integration.test.ts +++ b/src/backend/services/events/workerInvoker.integration.test.ts @@ -89,7 +89,10 @@ let calls: StubCall[]; let answer: number | 'hang' = 200; /** Whether the stub's answer carries the handled header, as a real one would. */ let answerHandled = true; -/** Whether the stub reports the script missing until it sees the deployed header. */ +/** + * Whether the stub reports the script missing until it sees the deployed + * header. + */ let dispatchMissing = false; const events = () => env.server.services.events; @@ -106,8 +109,7 @@ const startStub = async (): Promise => { stub = http.createServer((req, res) => { void readBody(req).then((raw) => { const deployed = req.headers['x-puter-events-deployed'] as - | string - | undefined; + string | undefined; calls.push({ method: req.method ?? '', path: req.url ?? '', @@ -124,7 +126,9 @@ const startStub = async (): Promise => { // Stands in for the events dispatcher reporting a namespace miss // until the deploy-on-miss retry carries the deployed header. if (dispatchMissing && deployed !== '1') { - res.writeHead(404, { 'x-puter-events-dispatch': 'missing' }).end(); + res.writeHead(404, { + 'x-puter-events-dispatch': 'missing', + }).end(); return; } // Stands in for the dispatcher forwarding a genuine answer from @@ -240,11 +244,32 @@ const touch = async (name: string): Promise => { await env.server.services.fs.touch(userId, { path: `${anchor}/${name}` }); }; +/** + * Retry `check` until it stops throwing, on real time. + * + * `vi.waitFor` advances the faked `Date` once per poll, which drains the very + * backoff windows these tests measure — a loaded run reads a 2s hold as 0.5s. + * `performance.now` and `setTimeout` are unfaked, so polling through them + * leaves the clock exactly where `jump` put it. + */ +const waitUntil = async ( + check: () => unknown, + { timeout = 5_000, interval = 25 } = {}, +): Promise => { + const started = performance.now(); + for (;;) { + try { + await check(); + return; + } catch (err) { + if (performance.now() - started >= timeout) throw err; + await new Promise((resolve) => setTimeout(resolve, interval)); + } + } +}; + const invoked = (count: number): Promise => - vi.waitFor(() => expect(calls.length).toBeGreaterThanOrEqual(count), { - timeout: 5_000, - interval: 25, - }); + waitUntil(() => expect(calls.length).toBeGreaterThanOrEqual(count)); /** Move time past whatever the failed delivery is being held for. */ const jump = (ms: number): void => { @@ -476,7 +501,7 @@ describe('the call an owed delivery makes', () => { // The revocation settle takes the row out of service on its own, // but it is a best-effort listener — put the row back so what is // under test is the delivery path with the consent already gone. - await vi.waitFor(async () => + await waitUntil(async () => expect( (await durable().getBySubId(subId))?.suspendedAt, ).not.toBeNull(), @@ -491,7 +516,7 @@ describe('the call an owed delivery makes', () => { await durable().warmRegion(userId); await touch('revoked.txt'); - await vi.waitFor(async () => + await waitUntil(async () => expect(await pending().depth(subId)).toBe(1), ); await events().sweepPending(); @@ -515,7 +540,7 @@ describe('the call an owed delivery makes', () => { await touch('settled.txt'); await invoked(1); - await vi.waitFor(async () => + await waitUntil(async () => expect(await pending().depth(subId)).toBe(0), ); }); @@ -531,7 +556,7 @@ describe('what each answer does to the delivery', () => { // The event is gone and a marker stands in its place, so the // subscription learns there was one rather than reading silence. - await vi.waitFor(async () => + await waitUntil(async () => expect(await pending().depth(subId)).toBe(1), ); answer = 200; @@ -556,7 +581,7 @@ describe('what each answer does to the delivery', () => { await touch('unmarked.txt'); await invoked(1); - await vi.waitFor(async () => + await waitUntil(async () => expect(await heldForMs(subId)).toBeGreaterThan(0), ); expect(await pending().depth(subId)).toBe(1); @@ -565,8 +590,8 @@ describe('what each answer does to the delivery', () => { it('holds one it could not answer, for longer each time', async () => { answer = 500; const subId = await subscribe(); - // Freeze the clock so the first hold reads exactly, as later ones do - // after `jump`; on real time a slow run reads 2s as 1s. + // Freeze the clock so every hold is read against the same instant it + // was written from. jump(0); await touch('failing.txt'); @@ -578,7 +603,7 @@ describe('what each answer does to the delivery', () => { attempt < EVENTS_CONSECUTIVE_FAILURES; attempt++ ) { - await vi.waitFor(async () => + await waitUntil(async () => expect(await heldForMs(subId)).toBeGreaterThan(0), ); waits.push(await heldForMs(subId)); @@ -592,16 +617,12 @@ describe('what each answer does to the delivery', () => { await invoked(attempt + 1); } - // Each hold sits within a second below its backoff step: a loaded run - // can spend part of the first one before it is read, and the steps - // are far enough apart that the windows never overlap. - for (const [i, expected] of [2, 4, 8, 16].entries()) { - expect(waits[i]).toBeGreaterThan((expected - 1) * 1000); - expect(waits[i]).toBeLessThanOrEqual(expected * 1000); - } + // Exact, not a window: the clock only moves where `jump` moves it, so + // a loaded run reads the same holds a quiet one does. + expect(waits).toEqual([2_000, 4_000, 8_000, 16_000]); // The fifth failure in a row is the one that stops it. - await vi.waitFor(async () => + await waitUntil(async () => expect((await durable().getBySubId(subId))?.suspendedReason).toBe( 'failures', ), @@ -625,7 +646,7 @@ describe('what each answer does to the delivery', () => { await events().sweepPending(); } - await vi.waitFor(() => + await waitUntil(() => expect(notify).toHaveBeenCalledWith( [userId], expect.objectContaining({ handler: HANDLER }), @@ -644,7 +665,7 @@ describe('what each answer does to the delivery', () => { await invoked(1); // Still owed, and held rather than dropped: nobody said no. - await vi.waitFor(async () => + await waitUntil(async () => expect(await heldForMs(subId)).toBeGreaterThan(0), ); expect(await pending().depth(subId)).toBe(1); @@ -658,7 +679,7 @@ describe('what each answer does to the delivery', () => { await touch('busy.txt'); await invoked(1); - await vi.waitFor(async () => + await waitUntil(async () => expect(await heldForMs(subId)).toBeGreaterThan(0), ); expect(await pending().depth(subId)).toBe(1); @@ -679,14 +700,14 @@ describe('with no events worker to address', () => { attempt < EVENTS_CONSECUTIVE_FAILURES; attempt++ ) { - await vi.waitFor(async () => + await waitUntil(async () => expect(await heldForMs(subId)).toBeGreaterThan(0), ); jump(deliveryBackoffMs(attempt) + 50); await events().sweepPending(); } - await vi.waitFor(async () => + await waitUntil(async () => expect( (await durable().getBySubId(subId))?.suspendedReason, ).toBe('failures'), @@ -712,7 +733,7 @@ describe('with no events worker to address', () => { try { await touch('suspended-owner.txt'); - await vi.waitFor(async () => + await waitUntil(async () => expect(await heldForMs(subId)).toBeGreaterThan(0), ); expect(await pending().depth(subId)).toBe(1); @@ -743,7 +764,7 @@ describe("what withdrawing an app's standing does to its worker session", () => EVENTS_BACKGROUND_PERMISSION, ); - await vi.waitFor(async () => + await waitUntil(async () => expect( (await workerSessionFor(app.appUid))?.revoked_at, ).not.toBeNull(), @@ -768,7 +789,7 @@ describe("what withdrawing an app's standing does to its worker session", () => app.appUid, ); - await vi.waitFor(async () => + await waitUntil(async () => expect( (await workerSessionFor(app.appUid))?.revoked_at, ).not.toBeNull(), @@ -815,7 +836,7 @@ describe("what withdrawing an app's standing does to its worker session", () => app.appUid, EVENTS_BACKGROUND_PERMISSION, ); - await vi.waitFor(async () => + await waitUntil(async () => expect( (await workerSessionFor(app.appUid))?.revoked_at, ).not.toBeNull(), @@ -863,7 +884,7 @@ describe('deploying on a dispatcher miss', () => { expect(createSpy).toHaveBeenCalledTimes(1); expect(calls[0].headers.deployed).toBeUndefined(); expect(calls[1].headers.deployed).toBe('1'); - await vi.waitFor(async () => + await waitUntil(async () => expect(await pending().depth(subId)).toBe(0), ); } finally {