569 lines
14 KiB
TypeScript
569 lines
14 KiB
TypeScript
import { describe, expect, it, vi } from "vitest";
|
|
|
|
import {
|
|
SSE_CONTINUE,
|
|
createFetchSseConnection,
|
|
} from "../../../src/adapters/realtime/sse/fetch-sse-connection.ts";
|
|
import {
|
|
realtimeTransportRecoveryCommitted,
|
|
type RealtimeRecoveryCheckpoint,
|
|
} from "../../../src/application/ports/realtime/event-authority.ts";
|
|
import type { RealtimeResult } from "../../../src/application/ports/realtime/shared.ts";
|
|
import type {
|
|
StreamRegistrationId,
|
|
} from "../../../src/contracts/realtime-streams.ts";
|
|
import {
|
|
realtimeFailure,
|
|
realtimeSuccess,
|
|
} from "../../../src/adapters/realtime/result.ts";
|
|
|
|
const encoder = new TextEncoder();
|
|
const continueEvent = () => realtimeSuccess(SSE_CONTINUE);
|
|
const RECOVERY_CHECKPOINT = Object.freeze({
|
|
recoveryMode: "CURSOR" as const,
|
|
streamEpoch: "stream-epoch.0001",
|
|
lastAppliedSequence: "1",
|
|
resumeCursor: "cursor-1",
|
|
}) as RealtimeRecoveryCheckpoint;
|
|
const RECOVERY_OUTCOME = realtimeTransportRecoveryCommitted(
|
|
"REFERENCE_STREAM" as StreamRegistrationId,
|
|
RECOVERY_CHECKPOINT,
|
|
);
|
|
|
|
function eventStream(
|
|
chunks: readonly string[],
|
|
options: Readonly<{
|
|
status?: number;
|
|
contentType?: string;
|
|
}> = {},
|
|
): Response {
|
|
return new Response(
|
|
new ReadableStream<Uint8Array>({
|
|
start(controller) {
|
|
for (const chunk of chunks) {
|
|
controller.enqueue(encoder.encode(chunk));
|
|
}
|
|
controller.close();
|
|
},
|
|
}),
|
|
{
|
|
status: options.status ?? 200,
|
|
headers: {
|
|
"Content-Type":
|
|
options.contentType ?? "text/event-stream; charset=utf-8",
|
|
},
|
|
},
|
|
);
|
|
}
|
|
|
|
function connection(
|
|
fetcher: typeof fetch,
|
|
recoveryMode: "CURSOR" | "SESSION_REBUILD" | "SNAPSHOT_ONLY" =
|
|
"CURSOR",
|
|
) {
|
|
return createFetchSseConnection({
|
|
endpoint: "https://app.example.test/events",
|
|
applicationOrigin: "https://app.example.test",
|
|
recoveryMode,
|
|
fetcher,
|
|
});
|
|
}
|
|
|
|
describe("fetch SSE connection", () => {
|
|
it("uses the fixed request profile and streams events sequentially", async () => {
|
|
const response = eventStream([
|
|
": heartbeat\r\n",
|
|
"retry: 2500\n",
|
|
"id: cursor-1\ndata: first\n",
|
|
"data: second\n\n",
|
|
]);
|
|
const fetcher = vi.fn(
|
|
async (_input: RequestInfo | URL, _init?: RequestInit) =>
|
|
response,
|
|
);
|
|
const received: string[] = [];
|
|
const comments = vi.fn();
|
|
const hints = vi.fn();
|
|
const runtime = connection(fetcher as typeof fetch);
|
|
|
|
await expect(
|
|
runtime.read({
|
|
resumeCursor: "cursor-0",
|
|
onEvent: async (event) => {
|
|
received.push(event.data);
|
|
return realtimeSuccess(SSE_CONTINUE);
|
|
},
|
|
onComment: comments,
|
|
onRetryHint: hints,
|
|
}),
|
|
).resolves.toEqual({
|
|
ok: true,
|
|
value: {
|
|
kind: "EOF",
|
|
incompleteEventDiscarded: false,
|
|
retryHintMs: 2_500,
|
|
},
|
|
});
|
|
|
|
expect(received).toEqual(["first\nsecond"]);
|
|
expect(comments).toHaveBeenCalledOnce();
|
|
expect(hints).toHaveBeenCalledWith(2_500);
|
|
const [target, init] = fetcher.mock.calls[0] ?? [];
|
|
expect(target).toBe("https://app.example.test/events");
|
|
expect(init).toMatchObject({
|
|
method: "GET",
|
|
credentials: "same-origin",
|
|
redirect: "error",
|
|
cache: "no-store",
|
|
referrerPolicy: "no-referrer",
|
|
});
|
|
expect(new Headers(init?.headers)).toEqual(
|
|
new Headers({
|
|
Accept: "text/event-stream",
|
|
"Last-Event-ID": "cursor-0",
|
|
}),
|
|
);
|
|
});
|
|
|
|
it("holds event consumption behind the validated open barrier gate", async () => {
|
|
const runtime = connection(
|
|
(async () =>
|
|
eventStream([
|
|
"id: cursor-1\ndata: after-barrier\n\n",
|
|
])) as typeof fetch,
|
|
);
|
|
let release:
|
|
| ((result: RealtimeResult<void>) => void)
|
|
| undefined;
|
|
const gate = new Promise<RealtimeResult<void>>((resolve) => {
|
|
release = resolve;
|
|
});
|
|
const onOpen = vi.fn(() => gate);
|
|
const onEvent = vi.fn(continueEvent);
|
|
|
|
const reading = runtime.read({
|
|
resumeCursor: null,
|
|
onOpen,
|
|
onEvent,
|
|
});
|
|
await vi.waitFor(() => expect(onOpen).toHaveBeenCalledOnce());
|
|
expect(onEvent).not.toHaveBeenCalled();
|
|
|
|
release?.(realtimeSuccess(undefined));
|
|
await expect(reading).resolves.toMatchObject({
|
|
ok: true,
|
|
value: { kind: "EOF" },
|
|
});
|
|
expect(onEvent).toHaveBeenCalledOnce();
|
|
});
|
|
|
|
it("propagates an exact open-barrier failure before reading events", async () => {
|
|
const runtime = connection(
|
|
(async () =>
|
|
eventStream([
|
|
"id: cursor-1\ndata: unreachable\n\n",
|
|
])) as typeof fetch,
|
|
);
|
|
const barrierFailure = realtimeFailure(
|
|
"SCOPE_FENCED",
|
|
"RECOVER",
|
|
false,
|
|
);
|
|
const onEvent = vi.fn(continueEvent);
|
|
|
|
await expect(
|
|
runtime.read({
|
|
resumeCursor: null,
|
|
onOpen: () => barrierFailure,
|
|
onEvent,
|
|
}),
|
|
).resolves.toBe(barrierFailure);
|
|
expect(onEvent).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it("treats 204 as terminal and rejects incorrect media types", async () => {
|
|
const terminal = connection(
|
|
(async () => new Response(null, { status: 204 })) as typeof fetch,
|
|
);
|
|
await expect(
|
|
terminal.read({
|
|
resumeCursor: null,
|
|
onEvent: continueEvent,
|
|
}),
|
|
).resolves.toEqual({
|
|
ok: true,
|
|
value: { kind: "NO_RECONNECT" },
|
|
});
|
|
|
|
const wrongType = connection(
|
|
(async () =>
|
|
eventStream(["data: value\n\n"], {
|
|
contentType: "application/json",
|
|
})) as typeof fetch,
|
|
);
|
|
await expect(
|
|
wrongType.read({
|
|
resumeCursor: null,
|
|
onEvent: continueEvent,
|
|
}),
|
|
).resolves.toEqual({
|
|
ok: false,
|
|
error: {
|
|
kind: "PROTOCOL_MISMATCH",
|
|
operation: "CONNECT",
|
|
retryable: false,
|
|
},
|
|
});
|
|
});
|
|
|
|
it.each([
|
|
[401, "AUTH_REQUIRED", false],
|
|
[403, "FORBIDDEN", false],
|
|
[409, "CURSOR_EXPIRED", false],
|
|
[410, "CURSOR_EXPIRED", false],
|
|
[429, "RATE_LIMITED", true],
|
|
[502, "PROVIDER_UNAVAILABLE", true],
|
|
[503, "PROVIDER_UNAVAILABLE", true],
|
|
[504, "PROVIDER_UNAVAILABLE", true],
|
|
[500, "PROTOCOL_MISMATCH", false],
|
|
] as const)(
|
|
"maps HTTP %i to %s without exposing a response body",
|
|
async (status, kind, retryable) => {
|
|
const runtime = connection(
|
|
(async () =>
|
|
new Response("private backend text", {
|
|
status,
|
|
headers:
|
|
status === 429 || status === 503
|
|
? { "Retry-After": "10" }
|
|
: undefined,
|
|
})) as typeof fetch,
|
|
);
|
|
const result = await runtime.read({
|
|
resumeCursor: null,
|
|
onEvent: continueEvent,
|
|
});
|
|
expect(result).toEqual({
|
|
ok: false,
|
|
error: {
|
|
kind,
|
|
operation: "CONNECT",
|
|
retryable,
|
|
},
|
|
});
|
|
expect(JSON.stringify(result)).not.toContain("private backend");
|
|
},
|
|
);
|
|
|
|
it.each([
|
|
[429, "RATE_LIMITED"],
|
|
[503, "PROVIDER_UNAVAILABLE"],
|
|
] as const)(
|
|
"does not retry HTTP %i without a bounded server hint",
|
|
async (status, kind) => {
|
|
const runtime = connection(
|
|
(async () => new Response(null, { status })) as typeof fetch,
|
|
);
|
|
await expect(
|
|
runtime.read({
|
|
resumeCursor: null,
|
|
onEvent: continueEvent,
|
|
}),
|
|
).resolves.toEqual({
|
|
ok: false,
|
|
error: {
|
|
kind,
|
|
operation: "CONNECT",
|
|
retryable: false,
|
|
},
|
|
});
|
|
},
|
|
);
|
|
|
|
it("does not clamp an excessive Retry-After into an early retry", async () => {
|
|
const runtime = connection(
|
|
(async () =>
|
|
new Response(null, {
|
|
status: 429,
|
|
headers: { "Retry-After": "120" },
|
|
})) as typeof fetch,
|
|
);
|
|
await expect(
|
|
runtime.read({
|
|
resumeCursor: null,
|
|
onEvent: continueEvent,
|
|
}),
|
|
).resolves.toEqual({
|
|
ok: false,
|
|
error: {
|
|
kind: "RATE_LIMITED",
|
|
operation: "CONNECT",
|
|
retryable: false,
|
|
},
|
|
});
|
|
});
|
|
|
|
it("enforces direct event IDs only for CURSOR recovery", async () => {
|
|
const missingCursorId = connection(
|
|
(async () =>
|
|
eventStream(["data: value\n\n"])) as typeof fetch,
|
|
);
|
|
await expect(
|
|
missingCursorId.read({
|
|
resumeCursor: null,
|
|
onEvent: continueEvent,
|
|
}),
|
|
).resolves.toEqual({
|
|
ok: false,
|
|
error: {
|
|
kind: "PROTOCOL_MISMATCH",
|
|
operation: "DECODE",
|
|
retryable: false,
|
|
},
|
|
});
|
|
|
|
const unexpectedCursorId = connection(
|
|
(async () =>
|
|
eventStream([
|
|
"id: cursor-1\ndata: value\n\n",
|
|
])) as typeof fetch,
|
|
"SNAPSHOT_ONLY",
|
|
);
|
|
await expect(
|
|
unexpectedCursorId.read({
|
|
resumeCursor: null,
|
|
onEvent: continueEvent,
|
|
}),
|
|
).resolves.toEqual({
|
|
ok: false,
|
|
error: {
|
|
kind: "PROTOCOL_MISMATCH",
|
|
operation: "DECODE",
|
|
retryable: false,
|
|
},
|
|
});
|
|
});
|
|
|
|
it("stops the old stream after authority recovery commits", async () => {
|
|
const runtime = connection(
|
|
(async () =>
|
|
eventStream([
|
|
"id: cursor-1\ndata: first\n\n",
|
|
"id: cursor-2\ndata: stale-generation\n\n",
|
|
])) as typeof fetch,
|
|
);
|
|
const onEvent = vi.fn(() =>
|
|
realtimeSuccess(RECOVERY_OUTCOME),
|
|
);
|
|
|
|
await expect(
|
|
runtime.read({
|
|
resumeCursor: null,
|
|
onEvent,
|
|
}),
|
|
).resolves.toEqual({
|
|
ok: true,
|
|
value: RECOVERY_OUTCOME,
|
|
});
|
|
expect(onEvent).toHaveBeenCalledOnce();
|
|
});
|
|
|
|
it("propagates a canonical consumer failure and aborts its event generation", async () => {
|
|
const runtime = connection(
|
|
(async () =>
|
|
eventStream([
|
|
"id: cursor-1\ndata: forbidden\n\n",
|
|
])) as typeof fetch,
|
|
);
|
|
let handlerSignal: AbortSignal | undefined;
|
|
|
|
await expect(
|
|
runtime.read({
|
|
resumeCursor: null,
|
|
onEvent: (_event, signal) => {
|
|
handlerSignal = signal;
|
|
return realtimeFailure("FORBIDDEN", "RECEIVE");
|
|
},
|
|
}),
|
|
).resolves.toEqual(
|
|
realtimeFailure("FORBIDDEN", "RECEIVE"),
|
|
);
|
|
expect(handlerSignal?.aborted).toBe(true);
|
|
});
|
|
|
|
it("fences recovery immediately and waits for bounded reader cancellation", async () => {
|
|
let releaseCancellation: (() => void) | undefined;
|
|
let cancellationStarted = false;
|
|
const stream = new ReadableStream<Uint8Array>({
|
|
start(controller) {
|
|
controller.enqueue(
|
|
encoder.encode("id: cursor-1\ndata: first\n\n"),
|
|
);
|
|
},
|
|
cancel() {
|
|
cancellationStarted = true;
|
|
return new Promise<void>((resolve) => {
|
|
releaseCancellation = resolve;
|
|
});
|
|
},
|
|
});
|
|
const runtime = connection(
|
|
(async () =>
|
|
new Response(stream, {
|
|
headers: { "Content-Type": "text/event-stream" },
|
|
})) as typeof fetch,
|
|
);
|
|
const pending = runtime.read({
|
|
resumeCursor: null,
|
|
onEvent: () =>
|
|
realtimeSuccess(RECOVERY_OUTCOME),
|
|
});
|
|
let settled = false;
|
|
void pending.then(() => {
|
|
settled = true;
|
|
});
|
|
for (let turn = 0; turn < 20; turn += 1) {
|
|
await Promise.resolve();
|
|
}
|
|
|
|
expect(cancellationStarted).toBe(true);
|
|
expect(settled).toBe(false);
|
|
releaseCancellation?.();
|
|
await expect(pending).resolves.toEqual({
|
|
ok: true,
|
|
value: RECOVERY_OUTCOME,
|
|
});
|
|
});
|
|
|
|
it("discards incomplete EOF and closes an active reader idempotently", async () => {
|
|
const incomplete = connection(
|
|
(async () =>
|
|
eventStream([
|
|
"id: cursor-1\ndata: incomplete\n",
|
|
])) as typeof fetch,
|
|
);
|
|
await expect(
|
|
incomplete.read({
|
|
resumeCursor: null,
|
|
onEvent() {
|
|
throw new Error("must not run");
|
|
},
|
|
}),
|
|
).resolves.toEqual({
|
|
ok: true,
|
|
value: {
|
|
kind: "EOF",
|
|
incompleteEventDiscarded: true,
|
|
retryHintMs: null,
|
|
},
|
|
});
|
|
|
|
const pendingStream = new ReadableStream<Uint8Array>({
|
|
pull: () => new Promise<void>(() => {}),
|
|
});
|
|
const active = connection(
|
|
(async () =>
|
|
new Response(pendingStream, {
|
|
headers: { "Content-Type": "text/event-stream" },
|
|
})) as typeof fetch,
|
|
);
|
|
const pending = active.read({
|
|
resumeCursor: null,
|
|
onEvent: continueEvent,
|
|
});
|
|
await Promise.resolve();
|
|
await Promise.resolve();
|
|
active.close();
|
|
active.close();
|
|
await expect(pending).resolves.toMatchObject({
|
|
ok: false,
|
|
error: {
|
|
kind: "ABORTED",
|
|
operation: "CONNECT",
|
|
retryable: false,
|
|
},
|
|
});
|
|
await expect(
|
|
active.read({
|
|
resumeCursor: null,
|
|
onEvent: continueEvent,
|
|
}),
|
|
).resolves.toEqual({
|
|
ok: false,
|
|
error: {
|
|
kind: "CLOSED",
|
|
operation: "CONNECT",
|
|
retryable: false,
|
|
},
|
|
});
|
|
});
|
|
|
|
it("dispatches a CR-terminated blank block completed at EOF", async () => {
|
|
const runtime = connection(
|
|
(async () =>
|
|
eventStream([
|
|
"id: cursor-1\rdata: final\r\r",
|
|
])) as typeof fetch,
|
|
);
|
|
const received = vi.fn(() =>
|
|
realtimeSuccess(SSE_CONTINUE),
|
|
);
|
|
await expect(
|
|
runtime.read({
|
|
resumeCursor: null,
|
|
onEvent: received,
|
|
}),
|
|
).resolves.toMatchObject({
|
|
ok: true,
|
|
value: {
|
|
kind: "EOF",
|
|
incompleteEventDiscarded: false,
|
|
},
|
|
});
|
|
expect(received).toHaveBeenCalledWith(
|
|
expect.objectContaining({ data: "final" }),
|
|
expect.any(AbortSignal),
|
|
);
|
|
});
|
|
|
|
it("rejects arbitrary endpoints and mode-incoherent cursors before fetch", async () => {
|
|
expect(() =>
|
|
createFetchSseConnection({
|
|
endpoint: "https://other.example.test/events",
|
|
applicationOrigin: "https://app.example.test",
|
|
recoveryMode: "CURSOR",
|
|
}),
|
|
).toThrow(TypeError);
|
|
expect(() =>
|
|
createFetchSseConnection({
|
|
endpoint:
|
|
"https://app.example.test/events?token=not-allowed",
|
|
applicationOrigin: "https://app.example.test",
|
|
recoveryMode: "CURSOR",
|
|
}),
|
|
).toThrow(TypeError);
|
|
|
|
const fetcher = vi.fn(
|
|
async () => eventStream(["data: unreachable\n\n"]),
|
|
);
|
|
const snapshotOnly = connection(
|
|
fetcher as unknown as typeof fetch,
|
|
"SNAPSHOT_ONLY",
|
|
);
|
|
await expect(
|
|
snapshotOnly.read({
|
|
resumeCursor: "cursor-not-allowed",
|
|
onEvent: continueEvent,
|
|
}),
|
|
).resolves.toEqual({
|
|
ok: false,
|
|
error: {
|
|
kind: "PROTOCOL_MISMATCH",
|
|
operation: "CONNECT",
|
|
retryable: false,
|
|
},
|
|
});
|
|
expect(fetcher).not.toHaveBeenCalled();
|
|
});
|
|
});
|