lee/puter
1
0 Fork 0
mirror of https://github.com/HeyPuter/puter.git synced 2026-10-12 23:00:06 +00:00
Code Issues Actions 16 Packages Projects Releases Wiki Activity

test: stop the delivery-backoff test reading holds against a moving clock

Browse Source 
`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.
This commit is contained in:
Juan Castro committed 2026-09-30 12:25:40 -04:00
1 parent 23c8f73bcb
commit f8d8003fea
1 file changed
+55 -34
Download Patch File Download Diff File Expand all files Collapse all files
src/backend/services/events/workerInvoker.integration.test.ts
+55 -34
View File
@@ -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<string> => {
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<string> => {
// 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<void> => {
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<void> => {
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<void> =>
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 {