443 lines
14 KiB
TypeScript
443 lines
14 KiB
TypeScript
/**
|
|
* `HttpsTransport` long-poll receive loop — covers happy dispatch,
|
|
* cursor advance, malformed-body retry, transient-failure backoff,
|
|
* auth-failure termination, batch-yield, and abort.
|
|
*/
|
|
|
|
import { describe, expect, it } from "@suaveplan/testing/runner";
|
|
import type { MessageEnvelope } from "@suaveplan/types";
|
|
import {
|
|
HTTPS_TRANSPORT_CODES,
|
|
HttpsTransport,
|
|
RECV_BATCH_CAP,
|
|
} from "../src/index.js";
|
|
import {
|
|
type FakeFetchCall,
|
|
jsonResponse,
|
|
makeEnvelope,
|
|
makeFakeFetch,
|
|
waitUntil,
|
|
} from "./_helpers.js";
|
|
|
|
const RELAY = "https://relay.test";
|
|
|
|
describe("HttpsTransport.recv — long-poll loop", () => {
|
|
it("dispatches envelopes returned by /recv to handlers and advances cursor", {
|
|
timeout: 5000,
|
|
}, async () => {
|
|
let recvCalls = 0;
|
|
const envelopes = [
|
|
makeEnvelope({ id: "a" }),
|
|
makeEnvelope({ id: "b" }),
|
|
];
|
|
const fetchFn = makeFakeFetch({
|
|
responder: (call) => {
|
|
if (call.url.includes("/recv")) {
|
|
recvCalls++;
|
|
if (recvCalls === 1) {
|
|
return jsonResponse({
|
|
envelopes,
|
|
nextCursor: "cursor-1",
|
|
});
|
|
}
|
|
return jsonResponse({
|
|
envelopes: [],
|
|
nextCursor: "cursor-1",
|
|
});
|
|
}
|
|
return jsonResponse({ ok: true });
|
|
},
|
|
});
|
|
const t = new HttpsTransport({
|
|
relayUrl: RELAY,
|
|
ourPeerId: "peer-a",
|
|
fetchFn,
|
|
longPollTimeoutMs: 50,
|
|
pollIdleDelayMs: 10,
|
|
});
|
|
const seen: MessageEnvelope[] = [];
|
|
t.onMessage((env) => seen.push(env));
|
|
t.start();
|
|
await waitUntil(() => seen.length === 2);
|
|
expect(seen[0]?.id).toBe("a");
|
|
expect(seen[1]?.id).toBe("b");
|
|
expect(t.cursor).toBe("cursor-1");
|
|
await t.stop();
|
|
});
|
|
|
|
it("includes since=cursor on subsequent /recv calls", {
|
|
timeout: 5000,
|
|
}, async () => {
|
|
const history: FakeFetchCall[] = [];
|
|
let recvCalls = 0;
|
|
const fetchFn = makeFakeFetch({
|
|
responder: (call) => {
|
|
if (call.url.includes("/recv")) {
|
|
recvCalls++;
|
|
if (recvCalls === 1) {
|
|
return jsonResponse({
|
|
envelopes: [makeEnvelope()],
|
|
nextCursor: "cursor-1",
|
|
});
|
|
}
|
|
return jsonResponse({
|
|
envelopes: [],
|
|
nextCursor: "cursor-1",
|
|
});
|
|
}
|
|
return jsonResponse({ ok: true });
|
|
},
|
|
history,
|
|
});
|
|
const t = new HttpsTransport({
|
|
relayUrl: RELAY,
|
|
ourPeerId: "peer-a",
|
|
fetchFn,
|
|
longPollTimeoutMs: 25,
|
|
pollIdleDelayMs: 5,
|
|
});
|
|
t.onMessage(() => undefined);
|
|
t.start();
|
|
await waitUntil(() => recvCalls >= 2);
|
|
const secondRecv = history.filter((c) => c.url.includes("/recv"))[1];
|
|
expect(secondRecv?.url).toContain("since=cursor-1");
|
|
await t.stop();
|
|
});
|
|
|
|
it("AUTH_FAILED on /recv stops the loop", { timeout: 5000 }, async () => {
|
|
let recvCalls = 0;
|
|
const fetchFn = makeFakeFetch({
|
|
responder: (call) => {
|
|
if (call.url.includes("/recv")) {
|
|
recvCalls++;
|
|
return new Response("nope", { status: 403 });
|
|
}
|
|
return jsonResponse({ ok: true });
|
|
},
|
|
});
|
|
const t = new HttpsTransport({
|
|
relayUrl: RELAY,
|
|
ourPeerId: "peer-a",
|
|
fetchFn,
|
|
longPollTimeoutMs: 25,
|
|
});
|
|
t.start();
|
|
await waitUntil(() => t.state === "stopped");
|
|
expect(recvCalls).toBe(1);
|
|
await t.stop();
|
|
});
|
|
|
|
it("backs off on 503 and retries", { timeout: 5000 }, async () => {
|
|
let recvCalls = 0;
|
|
const fetchFn = makeFakeFetch({
|
|
responder: (call) => {
|
|
if (call.url.includes("/recv")) {
|
|
recvCalls++;
|
|
if (recvCalls < 3) {
|
|
return new Response("down", { status: 503 });
|
|
}
|
|
return jsonResponse({ envelopes: [], nextCursor: "ok" });
|
|
}
|
|
return jsonResponse({ ok: true });
|
|
},
|
|
});
|
|
const t = new HttpsTransport({
|
|
relayUrl: RELAY,
|
|
ourPeerId: "peer-a",
|
|
fetchFn,
|
|
longPollTimeoutMs: 25,
|
|
backoffBaseMs: 1,
|
|
backoffMaxMs: 5,
|
|
pollIdleDelayMs: 5,
|
|
});
|
|
t.start();
|
|
await waitUntil(() => t.cursor === "ok");
|
|
expect(recvCalls).toBeGreaterThanOrEqual(3);
|
|
await t.stop();
|
|
});
|
|
|
|
it("backs off on 4xx other than auth (RECV_FAILED) and retries", {
|
|
timeout: 5000,
|
|
}, async () => {
|
|
let recvCalls = 0;
|
|
const fetchFn = makeFakeFetch({
|
|
responder: (call) => {
|
|
if (call.url.includes("/recv")) {
|
|
recvCalls++;
|
|
if (recvCalls < 2) {
|
|
return new Response("bad", { status: 418 });
|
|
}
|
|
return jsonResponse({ envelopes: [], nextCursor: "ok" });
|
|
}
|
|
return jsonResponse({ ok: true });
|
|
},
|
|
});
|
|
const t = new HttpsTransport({
|
|
relayUrl: RELAY,
|
|
ourPeerId: "peer-a",
|
|
fetchFn,
|
|
longPollTimeoutMs: 25,
|
|
backoffBaseMs: 1,
|
|
backoffMaxMs: 5,
|
|
pollIdleDelayMs: 5,
|
|
});
|
|
t.start();
|
|
await waitUntil(() => t.cursor === "ok");
|
|
expect(recvCalls).toBeGreaterThanOrEqual(2);
|
|
await t.stop();
|
|
});
|
|
|
|
it("backs off on malformed JSON body (MALFORMED_RESPONSE) and retries", {
|
|
timeout: 5000,
|
|
}, async () => {
|
|
let recvCalls = 0;
|
|
const fetchFn = makeFakeFetch({
|
|
responder: (call) => {
|
|
if (call.url.includes("/recv")) {
|
|
recvCalls++;
|
|
if (recvCalls < 2) {
|
|
return new Response("not-json{{", {
|
|
status: 200,
|
|
headers: { "content-type": "application/json" },
|
|
});
|
|
}
|
|
return jsonResponse({ envelopes: [], nextCursor: "ok" });
|
|
}
|
|
return jsonResponse({ ok: true });
|
|
},
|
|
});
|
|
const t = new HttpsTransport({
|
|
relayUrl: RELAY,
|
|
ourPeerId: "peer-a",
|
|
fetchFn,
|
|
longPollTimeoutMs: 25,
|
|
backoffBaseMs: 1,
|
|
backoffMaxMs: 5,
|
|
pollIdleDelayMs: 5,
|
|
});
|
|
t.start();
|
|
await waitUntil(() => t.cursor === "ok");
|
|
expect(recvCalls).toBeGreaterThanOrEqual(2);
|
|
await t.stop();
|
|
});
|
|
|
|
it("backs off on unexpected JSON shape (MALFORMED_RESPONSE) and retries", {
|
|
timeout: 5000,
|
|
}, async () => {
|
|
let recvCalls = 0;
|
|
const fetchFn = makeFakeFetch({
|
|
responder: (call) => {
|
|
if (call.url.includes("/recv")) {
|
|
recvCalls++;
|
|
if (recvCalls < 2) {
|
|
return jsonResponse({ wrong: "shape" });
|
|
}
|
|
return jsonResponse({ envelopes: [], nextCursor: "ok" });
|
|
}
|
|
return jsonResponse({ ok: true });
|
|
},
|
|
});
|
|
const t = new HttpsTransport({
|
|
relayUrl: RELAY,
|
|
ourPeerId: "peer-a",
|
|
fetchFn,
|
|
longPollTimeoutMs: 25,
|
|
backoffBaseMs: 1,
|
|
backoffMaxMs: 5,
|
|
pollIdleDelayMs: 5,
|
|
});
|
|
t.start();
|
|
await waitUntil(() => t.cursor === "ok");
|
|
expect(recvCalls).toBeGreaterThanOrEqual(2);
|
|
await t.stop();
|
|
});
|
|
|
|
it("yields after RECV_BATCH_CAP envelopes per long-poll batch", {
|
|
timeout: 10000,
|
|
}, async () => {
|
|
let dispatched = 0;
|
|
const big = Array.from({ length: RECV_BATCH_CAP + 5 }, (_, i) =>
|
|
makeEnvelope({ id: `e-${i}` })
|
|
);
|
|
let recvCalls = 0;
|
|
const fetchFn = makeFakeFetch({
|
|
responder: (call) => {
|
|
if (call.url.includes("/recv")) {
|
|
recvCalls++;
|
|
if (recvCalls === 1) {
|
|
return jsonResponse({
|
|
envelopes: big,
|
|
nextCursor: "post-batch",
|
|
});
|
|
}
|
|
return jsonResponse({
|
|
envelopes: [],
|
|
nextCursor: "post-batch",
|
|
});
|
|
}
|
|
return jsonResponse({ ok: true });
|
|
},
|
|
});
|
|
const t = new HttpsTransport({
|
|
relayUrl: RELAY,
|
|
ourPeerId: "peer-a",
|
|
fetchFn,
|
|
longPollTimeoutMs: 50,
|
|
pollIdleDelayMs: 5,
|
|
});
|
|
t.onMessage(() => {
|
|
dispatched++;
|
|
});
|
|
t.start();
|
|
await waitUntil(() => dispatched === big.length);
|
|
await t.stop();
|
|
});
|
|
|
|
it("a handler that throws does not poison the loop", {
|
|
timeout: 5000,
|
|
}, async () => {
|
|
let recvCalls = 0;
|
|
const fetchFn = makeFakeFetch({
|
|
responder: (call) => {
|
|
if (call.url.includes("/recv")) {
|
|
recvCalls++;
|
|
if (recvCalls <= 2) {
|
|
return jsonResponse({
|
|
envelopes: [makeEnvelope({ id: `e-${recvCalls}` })],
|
|
nextCursor: `c-${recvCalls}`,
|
|
});
|
|
}
|
|
return jsonResponse({
|
|
envelopes: [],
|
|
nextCursor: `c-${recvCalls}`,
|
|
});
|
|
}
|
|
return jsonResponse({ ok: true });
|
|
},
|
|
});
|
|
const t = new HttpsTransport({
|
|
relayUrl: RELAY,
|
|
ourPeerId: "peer-a",
|
|
fetchFn,
|
|
longPollTimeoutMs: 25,
|
|
pollIdleDelayMs: 5,
|
|
});
|
|
const seen: string[] = [];
|
|
t.onMessage(() => {
|
|
throw new Error("subscriber-A");
|
|
});
|
|
t.onMessage((e) => {
|
|
seen.push(e.id);
|
|
});
|
|
t.start();
|
|
await waitUntil(() => seen.length >= 2);
|
|
await t.stop();
|
|
});
|
|
|
|
it("unsubscribed handlers stop receiving envelopes", {
|
|
timeout: 5000,
|
|
}, async () => {
|
|
// Single-shot fetch: return one envelope then empty forever so
|
|
// the loop quiesces rather than spinning.
|
|
let recvCalls = 0;
|
|
const fetchFn = makeFakeFetch({
|
|
responder: (call) => {
|
|
if (call.url.includes("/recv")) {
|
|
recvCalls++;
|
|
if (recvCalls === 1) {
|
|
return jsonResponse({
|
|
envelopes: [makeEnvelope({ id: "first" })],
|
|
nextCursor: "c-1",
|
|
});
|
|
}
|
|
return jsonResponse({ envelopes: [], nextCursor: "c-1" });
|
|
}
|
|
return jsonResponse({ ok: true });
|
|
},
|
|
});
|
|
const t = new HttpsTransport({
|
|
relayUrl: RELAY,
|
|
ourPeerId: "peer-a",
|
|
fetchFn,
|
|
longPollTimeoutMs: 25,
|
|
pollIdleDelayMs: 5,
|
|
});
|
|
const seen: string[] = [];
|
|
const off = t.onMessage((e) => seen.push(e.id));
|
|
t.start();
|
|
await waitUntil(() => seen.length >= 1);
|
|
expect(seen).toEqual(["first"]);
|
|
// After unsubscribing, the same handler should never be called
|
|
// again — but we still need to verify the loop kept running.
|
|
off();
|
|
const recvBeforeStop = recvCalls;
|
|
await waitUntil(() => recvCalls > recvBeforeStop);
|
|
await t.stop();
|
|
expect(seen).toEqual(["first"]);
|
|
});
|
|
|
|
it("AUTH_FAILED on /recv propagates via lastAuthError", {
|
|
timeout: 5000,
|
|
}, async () => {
|
|
const fetchFn = makeFakeFetch({
|
|
responder: (call) => {
|
|
if (call.url.includes("/recv")) {
|
|
return new Response("nope", { status: 401 });
|
|
}
|
|
return jsonResponse({ ok: true });
|
|
},
|
|
});
|
|
const t = new HttpsTransport({
|
|
relayUrl: RELAY,
|
|
ourPeerId: "peer-a",
|
|
fetchFn,
|
|
longPollTimeoutMs: 25,
|
|
});
|
|
expect(t.lastAuthError).toBeUndefined();
|
|
t.start();
|
|
await waitUntil(() => t.lastAuthError !== undefined);
|
|
expect(t.state).toBe("stopped");
|
|
const last = t.lastAuthError;
|
|
expect(last).toBeDefined();
|
|
expect(last?.code).toBe(HTTPS_TRANSPORT_CODES.AUTH_FAILED);
|
|
await t.stop();
|
|
});
|
|
|
|
it("handles zero-ms backoff sleep deterministically", {
|
|
timeout: 5000,
|
|
}, async () => {
|
|
let recvCalls = 0;
|
|
const fetchFn = makeFakeFetch({
|
|
responder: (call) => {
|
|
if (call.url.includes("/recv")) {
|
|
recvCalls++;
|
|
if (recvCalls < 2) {
|
|
return new Response("bad", { status: 500 });
|
|
}
|
|
return jsonResponse({
|
|
envelopes: [],
|
|
nextCursor: "ok",
|
|
});
|
|
}
|
|
return jsonResponse({ ok: true });
|
|
},
|
|
});
|
|
const t = new HttpsTransport({
|
|
relayUrl: RELAY,
|
|
ourPeerId: "peer-zero",
|
|
fetchFn,
|
|
longPollTimeoutMs: 25,
|
|
backoffBaseMs: 1,
|
|
backoffMaxMs: 5,
|
|
pollIdleDelayMs: 5,
|
|
// Zero jitter — `computeBackoff(1)` = floor(1/2 + 0) = 0,
|
|
// forcing the `sleep(0, signal)` early-return path.
|
|
random: () => 0,
|
|
});
|
|
t.start();
|
|
await waitUntil(() => t.cursor === "ok");
|
|
expect(recvCalls).toBeGreaterThanOrEqual(2);
|
|
await t.stop();
|
|
});
|
|
});
|