6690e26684
The page asked the relay for a 10 s rendezvous on each world and room track
and waited 10 s for the answer. With no publisher, the relay answers when the
rendezvous ends, so the two timers raced: on a loaded host the page gave up
("SUBSCRIBE: no answer in 10000 ms") as the refusal was sent, and world.js,
which follows a departed actor's track on that refusal, never saw it. The
world scenario's "dave catches up after alice left" failed so on the train's
Studio run.
request() takes how long to wait for its answer; subscribe() waits a step
past the rendezvous it asks for. session.test.mjs holds the three cases with
the clock mocked; without the change its first case fails with the e2e's
error.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
76 lines
3 KiB
JavaScript
76 lines
3 KiB
JavaScript
// How long a request waits for its answer (`node --test web/moq/session.test.mjs`), over a
|
|
// transport and a codec of the test's own, the clock mocked.
|
|
import test from "node:test";
|
|
import assert from "node:assert/strict";
|
|
import { MoqSession, RequestRefused } from "./session.js";
|
|
|
|
const enc = new TextEncoder();
|
|
const hex = (s) => [...enc.encode(s)].map((b) => b.toString(16).padStart(2, "0")).join("");
|
|
|
|
// Control messages as JSON text, one per read.
|
|
const moq = {
|
|
encodeControl: (s) => enc.encode(s),
|
|
MessageStream: class {
|
|
constructor() { this.q = []; }
|
|
push(v) { this.q.push(new TextDecoder().decode(v)); }
|
|
next() { return this.q.shift(); }
|
|
free() {}
|
|
},
|
|
};
|
|
|
|
// Each request stream's reader gives what `answer(i, message)` hands it, and nothing before.
|
|
function transport() {
|
|
const answers = [];
|
|
return {
|
|
kind: "test",
|
|
closed: new Promise(() => {}),
|
|
async openBidi() {
|
|
let give;
|
|
const first = new Promise((r) => { give = r; });
|
|
let read = 0;
|
|
answers.push(give);
|
|
return {
|
|
writer: { async write() {}, reset() {} },
|
|
reader: { read: () => (read++ === 0 ? first : new Promise(() => {})) },
|
|
};
|
|
},
|
|
answer(i, message) { answers[i]({ value: enc.encode(JSON.stringify(message)), done: false }); },
|
|
};
|
|
}
|
|
|
|
const settle = async () => { for (let i = 0; i < 8; i++) await new Promise((r) => setImmediate(r)); };
|
|
const refused = { message: "REQUEST_ERROR", request_id: 0, code: 2, reason: hex("no publisher within RENDEZVOUS_TIMEOUT") };
|
|
|
|
test("a SUBSCRIBE that asks for a rendezvous takes the relay's answer when the rendezvous ends", async (t) => {
|
|
t.mock.timers.enable({ apis: ["setTimeout"] });
|
|
const tr = transport();
|
|
const s = new MoqSession(tr, moq);
|
|
const p = s.subscribe(["room", "a", "world"], "0", { rendezvous_timeout: 10000 }, {});
|
|
p.catch(() => {});
|
|
await settle();
|
|
// The relay holds the request for the whole rendezvous, and its answer arrives after it.
|
|
t.mock.timers.tick(10_500);
|
|
await settle();
|
|
tr.answer(0, refused);
|
|
await assert.rejects(p, (e) => e instanceof RequestRefused && /REQUEST_ERROR 2 no publisher within RENDEZVOUS_TIMEOUT/.test(e.message));
|
|
});
|
|
|
|
test("a SUBSCRIBE that asks for a rendezvous waits no longer than the step past it", async (t) => {
|
|
t.mock.timers.enable({ apis: ["setTimeout"] });
|
|
const s = new MoqSession(transport(), moq);
|
|
const p = s.subscribe(["room", "a", "world"], "0", { rendezvous_timeout: 10000 }, {});
|
|
p.catch(() => {});
|
|
await settle();
|
|
t.mock.timers.tick(20_000);
|
|
await assert.rejects(p, (e) => !(e instanceof RequestRefused) && /SUBSCRIBE: no answer in 20000 ms/.test(e.message));
|
|
});
|
|
|
|
test("a SUBSCRIBE without a rendezvous waits one step for its answer", async (t) => {
|
|
t.mock.timers.enable({ apis: ["setTimeout"] });
|
|
const s = new MoqSession(transport(), moq);
|
|
const p = s.subscribe(["room", "a", "world"], "0", { forward: true }, {});
|
|
p.catch(() => {});
|
|
await settle();
|
|
t.mock.timers.tick(10_000);
|
|
await assert.rejects(p, /SUBSCRIBE: no answer in 10000 ms/);
|
|
});
|