976 lines
29 KiB
TypeScript
976 lines
29 KiB
TypeScript
import { describe, expect, it, vi } from "vitest";
|
|
|
|
import type {
|
|
ExternalRealtimeEventContext,
|
|
RealtimeEventAuthority,
|
|
RealtimeObservation,
|
|
RealtimeRecoveryCommit,
|
|
RealtimeRecoveryRequest,
|
|
} from "../../../src/application/ports/realtime/event-authority.ts";
|
|
import type { RealtimeResult } from "../../../src/application/ports/realtime/shared.ts";
|
|
import {
|
|
createRealtimeStreamCoordinator,
|
|
} from "../../../src/adapters/realtime/stream-coordinator.ts";
|
|
import {
|
|
realtimeFailure,
|
|
realtimeSuccess,
|
|
} from "../../../src/adapters/realtime/result.ts";
|
|
import type { ValidatedRealtimeEventDto } from "../../../src/adapters/realtime/event-codec.ts";
|
|
import type { RealtimePolicyRegistry } from "../../../src/contracts/realtime-streams.ts";
|
|
import {
|
|
STREAM_ID,
|
|
TEST_LIMITS,
|
|
TEST_MAPPER,
|
|
TEST_MAPPERS,
|
|
createTestRealtimeCodec,
|
|
createTestRealtimeRegistry,
|
|
realtimeEventJson,
|
|
type EventOverrides,
|
|
} from "./fixture.ts";
|
|
|
|
describe("transport-independent realtime stream coordinator", () => {
|
|
it("applies one stream sequentially and commits each cursor after its effect", async () => {
|
|
const first = deferred<RealtimeResult<void>>();
|
|
const applied: string[] = [];
|
|
const harness = createHarness({
|
|
async apply(_profile, value) {
|
|
const selected = (value as Readonly<{ value: string }>).value;
|
|
applied.push(selected);
|
|
return selected === "first"
|
|
? first.promise
|
|
: realtimeSuccess(undefined);
|
|
},
|
|
});
|
|
await harness.initialize();
|
|
|
|
const firstResult = harness.coordinator.accept(
|
|
harness.event({
|
|
eventId: "event-00000001",
|
|
sequence: "1",
|
|
resumeCursor: "cursor-00000001",
|
|
payload: { value: "first" },
|
|
}),
|
|
);
|
|
const secondResult = harness.coordinator.accept(
|
|
harness.event({
|
|
eventId: "event-00000002",
|
|
sequence: "2",
|
|
resumeCursor: "cursor-00000002",
|
|
payload: { value: "second" },
|
|
}),
|
|
);
|
|
|
|
await vi.waitFor(() => expect(applied).toEqual(["first"]));
|
|
expect(harness.coordinator.getResumeState(STREAM_ID)).toMatchObject({
|
|
lastAppliedSequence: "0",
|
|
resumeCursor: "cursor-snapshot-0",
|
|
});
|
|
|
|
first.resolve(realtimeSuccess(undefined));
|
|
await expect(firstResult).resolves.toMatchObject({
|
|
ok: true,
|
|
value: { outcome: "APPLIED" },
|
|
});
|
|
await expect(secondResult).resolves.toMatchObject({
|
|
ok: true,
|
|
value: { outcome: "APPLIED" },
|
|
});
|
|
expect(applied).toEqual(["first", "second"]);
|
|
expect(harness.coordinator.getResumeState(STREAM_ID)).toEqual({
|
|
recoveryMode: "CURSOR",
|
|
streamEpoch: "stream-epoch-0001",
|
|
lastAppliedSequence: "2",
|
|
resumeCursor: "cursor-00000002",
|
|
});
|
|
});
|
|
|
|
it("drops exact duplicates and stale events without replaying effects", async () => {
|
|
const harness = createHarness();
|
|
await harness.initialize();
|
|
const event = harness.event();
|
|
|
|
await harness.coordinator.accept(event);
|
|
await expect(harness.coordinator.accept(event)).resolves.toEqual({
|
|
ok: true,
|
|
value: {
|
|
outcome: "DROPPED",
|
|
reason: "DUPLICATE_EVENT",
|
|
},
|
|
});
|
|
await expect(
|
|
harness.coordinator.accept(
|
|
harness.event({
|
|
eventId: "event-stale-0001",
|
|
sequence: "0",
|
|
resumeCursor: "cursor-stale-0001",
|
|
}),
|
|
),
|
|
).resolves.toEqual({
|
|
ok: true,
|
|
value: {
|
|
outcome: "DROPPED",
|
|
reason: "STALE_EVENT",
|
|
},
|
|
});
|
|
expect(harness.effects).toHaveBeenCalledTimes(1);
|
|
expect(harness.coordinator.inspect(STREAM_ID).dedupeEntries).toBe(1);
|
|
});
|
|
|
|
it("recovers without applying a conflicting ID, sequence gap or epoch", async () => {
|
|
for (const [overrides, expectedReason] of [
|
|
[
|
|
{
|
|
eventId: "event-00000001",
|
|
sequence: "1",
|
|
payload: { value: "conflict" },
|
|
},
|
|
"EVENT_CONFLICT",
|
|
],
|
|
[
|
|
{
|
|
eventId: "event-gap-000001",
|
|
sequence: "3",
|
|
resumeCursor: "cursor-gap-000001",
|
|
},
|
|
"SEQUENCE_GAP",
|
|
],
|
|
[
|
|
{
|
|
eventId: "event-epoch-0001",
|
|
sequence: "2",
|
|
streamEpoch: "stream-epoch-0002",
|
|
resumeCursor: "cursor-epoch-0001",
|
|
},
|
|
"STREAM_EPOCH_CHANGED",
|
|
],
|
|
] as const) {
|
|
const harness = createHarness();
|
|
await harness.initialize();
|
|
await harness.coordinator.accept(harness.event());
|
|
|
|
await expect(
|
|
harness.coordinator.accept(harness.event(overrides)),
|
|
).resolves.toMatchObject({
|
|
ok: true,
|
|
value: {
|
|
outcome: "RECOVERED",
|
|
reason: expectedReason,
|
|
},
|
|
});
|
|
expect(harness.recoveryReasons).toEqual([
|
|
"INITIALIZE",
|
|
expectedReason,
|
|
]);
|
|
expect(harness.effects).toHaveBeenCalledTimes(1);
|
|
}
|
|
});
|
|
|
|
it("does not apply or advance when mapping fails", async () => {
|
|
const harness = createHarness({
|
|
mappers: {
|
|
ReferenceRealtimeMapper: {
|
|
...TEST_MAPPER,
|
|
map: () => ({
|
|
ok: false as const,
|
|
code: "MAPPING_INVARIANT_REJECTED" as const,
|
|
}),
|
|
},
|
|
},
|
|
});
|
|
await harness.initialize();
|
|
|
|
await expect(
|
|
harness.coordinator.accept(harness.event()),
|
|
).resolves.toMatchObject({
|
|
ok: true,
|
|
value: {
|
|
outcome: "RECOVERED",
|
|
reason: "MAPPING_CONTRACT_VIOLATION",
|
|
},
|
|
});
|
|
expect(harness.effects).not.toHaveBeenCalled();
|
|
expect(harness.recoveryReasons).toEqual([
|
|
"INITIALIZE",
|
|
"MAPPING_CONTRACT_VIOLATION",
|
|
]);
|
|
});
|
|
|
|
it("leaves the prior checkpoint when an effect and its recovery fail", async () => {
|
|
let recoveryCount = 0;
|
|
const harness = createHarness({
|
|
apply: async () => realtimeFailure("APPLY_FAILED", "APPLY"),
|
|
recover: async () => {
|
|
recoveryCount += 1;
|
|
return recoveryCount === 1
|
|
? realtimeSuccess(snapshotCommit("0"))
|
|
: realtimeFailure("PROVIDER_UNAVAILABLE", "RECOVER");
|
|
},
|
|
});
|
|
await harness.initialize();
|
|
|
|
await expect(
|
|
harness.coordinator.accept(harness.event()),
|
|
).resolves.toMatchObject({
|
|
ok: false,
|
|
error: {
|
|
kind: "PROVIDER_UNAVAILABLE",
|
|
operation: "RECOVER",
|
|
},
|
|
});
|
|
expect(harness.coordinator.getResumeState(STREAM_ID)).toMatchObject({
|
|
lastAppliedSequence: "0",
|
|
resumeCursor: "cursor-snapshot-0",
|
|
});
|
|
expect(harness.coordinator.inspect(STREAM_ID).freshness).toBe(
|
|
"UNKNOWN",
|
|
);
|
|
await harness.coordinator.accept(
|
|
harness.event({
|
|
eventId: "event-00000002",
|
|
sequence: "2",
|
|
resumeCursor: "cursor-00000002",
|
|
}),
|
|
);
|
|
expect(harness.effects).toHaveBeenCalledTimes(1);
|
|
});
|
|
|
|
it("rejects malformed authority results without advancing a checkpoint", async () => {
|
|
let recoveryCount = 0;
|
|
const harness = createHarness({
|
|
apply: async () =>
|
|
({
|
|
ok: true,
|
|
value: "not-void",
|
|
}) as unknown as RealtimeResult<void>,
|
|
recover: async () => {
|
|
recoveryCount += 1;
|
|
return recoveryCount === 1
|
|
? realtimeSuccess(snapshotCommit("0"))
|
|
: ({
|
|
ok: false,
|
|
error: {
|
|
kind: "FORBIDDEN",
|
|
operation: "APPLY",
|
|
retryable: false,
|
|
},
|
|
} as RealtimeResult<RealtimeRecoveryCommit>);
|
|
},
|
|
});
|
|
await harness.initialize();
|
|
|
|
await expect(
|
|
harness.coordinator.accept(harness.event()),
|
|
).resolves.toEqual(
|
|
realtimeFailure("PROTOCOL_MISMATCH", "RECOVER"),
|
|
);
|
|
expect(harness.coordinator.getResumeState(STREAM_ID)).toMatchObject({
|
|
lastAppliedSequence: "0",
|
|
});
|
|
expect(harness.coordinator.inspect(STREAM_ID).freshness).toBe(
|
|
"UNKNOWN",
|
|
);
|
|
});
|
|
|
|
it("rejects recovery accessors without invoking them", async () => {
|
|
let checkpointReads = 0;
|
|
const accessorCommit = Object.defineProperties(
|
|
{},
|
|
{
|
|
checkpoint: {
|
|
enumerable: true,
|
|
get() {
|
|
checkpointReads += 1;
|
|
return snapshotCommit("0");
|
|
},
|
|
},
|
|
kind: {
|
|
enumerable: true,
|
|
value: "SNAPSHOT_RESET",
|
|
},
|
|
},
|
|
) as RealtimeRecoveryCommit;
|
|
const harness = createHarness({
|
|
recover: async () => realtimeSuccess(accessorCommit),
|
|
});
|
|
|
|
await expect(
|
|
harness.coordinator.recover(STREAM_ID, "INITIALIZE"),
|
|
).resolves.toEqual(
|
|
realtimeFailure("PROTOCOL_MISMATCH", "RECOVER"),
|
|
);
|
|
expect(checkpointReads).toBe(0);
|
|
expect(harness.coordinator.getResumeState(STREAM_ID)).toBeNull();
|
|
});
|
|
|
|
it("commits only the one-shot descriptor snapshot of a recovery proxy", async () => {
|
|
const descriptorCommit = snapshotCommit("0");
|
|
const propertyReads: PropertyKey[] = [];
|
|
const proxyCommit = new Proxy(descriptorCommit, {
|
|
get(_target, key) {
|
|
propertyReads.push(key);
|
|
if (key === "kind") return "SESSION_REBUILD";
|
|
if (key === "checkpoint") return snapshotCommit("99");
|
|
return undefined;
|
|
},
|
|
});
|
|
const harness = createHarness({
|
|
recover: async () => realtimeSuccess(proxyCommit),
|
|
});
|
|
|
|
const recovered = await harness.coordinator.recover(
|
|
STREAM_ID,
|
|
"INITIALIZE",
|
|
);
|
|
|
|
expect(recovered).toMatchObject({
|
|
ok: true,
|
|
value: {
|
|
lastAppliedSequence: "0",
|
|
resumeCursor: "cursor-snapshot-0",
|
|
},
|
|
});
|
|
expect(propertyReads).toEqual([]);
|
|
});
|
|
|
|
it.each([
|
|
Object.freeze({
|
|
...snapshotCommit("0"),
|
|
unexpected: true,
|
|
}),
|
|
Object.assign(
|
|
Object.create({ inherited: true }) as Record<string, unknown>,
|
|
snapshotCommit("0"),
|
|
),
|
|
])(
|
|
"rejects extra and inherited recovery commit shapes",
|
|
async (commit) => {
|
|
const harness = createHarness({
|
|
recover: async () =>
|
|
realtimeSuccess(commit as RealtimeRecoveryCommit),
|
|
});
|
|
|
|
await expect(
|
|
harness.coordinator.recover(STREAM_ID, "INITIALIZE"),
|
|
).resolves.toEqual(
|
|
realtimeFailure("PROTOCOL_MISMATCH", "RECOVER"),
|
|
);
|
|
expect(harness.coordinator.getResumeState(STREAM_ID)).toBeNull();
|
|
},
|
|
);
|
|
|
|
it("expires effect and recovery commit leases when callbacks settle", async () => {
|
|
let effectContext: ExternalRealtimeEventContext | undefined;
|
|
let recoveryRequest: RealtimeRecoveryRequest | undefined;
|
|
const harness = createHarness({
|
|
async apply(_profile, _value, context) {
|
|
effectContext = context;
|
|
expect(context.isCurrent()).toBe(true);
|
|
return realtimeSuccess(undefined);
|
|
},
|
|
async recover(request) {
|
|
recoveryRequest = request;
|
|
expect(request.isCurrent()).toBe(true);
|
|
return realtimeSuccess(snapshotCommit("0"));
|
|
},
|
|
});
|
|
|
|
await harness.initialize();
|
|
expect(recoveryRequest?.isCurrent()).toBe(false);
|
|
await harness.coordinator.accept(harness.event());
|
|
expect(effectContext?.isCurrent()).toBe(false);
|
|
});
|
|
|
|
it("fences an in-flight successful effect before cursor commit", async () => {
|
|
const pending = deferred<RealtimeResult<void>>();
|
|
let effectContext: ExternalRealtimeEventContext | undefined;
|
|
const harness = createHarness({
|
|
apply: async (_profile, _value, context) => {
|
|
effectContext = context;
|
|
return pending.promise;
|
|
},
|
|
});
|
|
await harness.initialize();
|
|
|
|
const accepted = harness.coordinator.accept(harness.event());
|
|
await vi.waitFor(() =>
|
|
expect(harness.effects).toHaveBeenCalledOnce(),
|
|
);
|
|
harness.setCurrent(false);
|
|
expect(effectContext?.isCurrent()).toBe(false);
|
|
pending.resolve(realtimeSuccess(undefined));
|
|
|
|
await expect(accepted).resolves.toEqual({
|
|
ok: true,
|
|
value: {
|
|
outcome: "DROPPED",
|
|
reason: "SCOPE_FENCED",
|
|
},
|
|
});
|
|
expect(harness.coordinator.getResumeState(STREAM_ID)).toBeNull();
|
|
expect(harness.coordinator.inspect(STREAM_ID)).toMatchObject({
|
|
freshness: "UNKNOWN",
|
|
dedupeEntries: 0,
|
|
queuedEvents: 0,
|
|
awaitingTransportBarrier: false,
|
|
});
|
|
});
|
|
|
|
it("aborts and quiesces the active effect before queue-overflow recovery", async () => {
|
|
const pending = deferred<RealtimeResult<void>>();
|
|
let capturedSignal: AbortSignal | undefined;
|
|
const registry = createTestRealtimeRegistry({
|
|
limits: {
|
|
...TEST_LIMITS,
|
|
maxQueueEvents: 1,
|
|
},
|
|
});
|
|
const harness = createHarness({
|
|
registry,
|
|
async apply(_profile, _value, _context, signal) {
|
|
capturedSignal = signal;
|
|
return pending.promise;
|
|
},
|
|
});
|
|
await harness.initialize();
|
|
|
|
const first = harness.coordinator.accept(harness.event());
|
|
await vi.waitFor(() =>
|
|
expect(harness.effects).toHaveBeenCalledOnce(),
|
|
);
|
|
const overflow = harness.coordinator.accept(
|
|
harness.event({
|
|
eventId: "event-00000002",
|
|
sequence: "2",
|
|
resumeCursor: "cursor-00000002",
|
|
}),
|
|
);
|
|
|
|
expect(capturedSignal?.aborted).toBe(true);
|
|
expect(harness.recoveryReasons).toEqual(["INITIALIZE"]);
|
|
pending.resolve(realtimeSuccess(undefined));
|
|
await expect(first).resolves.toMatchObject({
|
|
ok: true,
|
|
value: { outcome: "DROPPED" },
|
|
});
|
|
await expect(overflow).resolves.toMatchObject({
|
|
ok: true,
|
|
value: {
|
|
outcome: "RECOVERED",
|
|
reason: "QUEUE_OVERFLOW",
|
|
},
|
|
});
|
|
expect(harness.recoveryReasons).toEqual([
|
|
"INITIALIZE",
|
|
"QUEUE_OVERFLOW",
|
|
]);
|
|
});
|
|
|
|
it("treats a current scope mismatch as a protocol violation, not a stale callback", async () => {
|
|
const harness = createHarness();
|
|
await harness.initialize();
|
|
|
|
await expect(
|
|
harness.coordinator.accept(
|
|
harness.event({ scopeBinding: "other-scope-binding" }),
|
|
),
|
|
).resolves.toMatchObject({
|
|
ok: true,
|
|
value: {
|
|
outcome: "RECOVERED",
|
|
reason: "SCOPE_PROTOCOL_VIOLATION",
|
|
},
|
|
});
|
|
expect(harness.effects).not.toHaveBeenCalled();
|
|
expect(harness.recoveryReasons).toEqual([
|
|
"INITIALIZE",
|
|
"SCOPE_PROTOCOL_VIOLATION",
|
|
]);
|
|
});
|
|
|
|
it("never claims CURRENT for a SNAPSHOT_ONLY stream without a barrier", async () => {
|
|
const registry = createTestRealtimeRegistry({
|
|
recovery: {
|
|
mode: "SNAPSHOT_ONLY",
|
|
snapshotOperationId: "GET_REFERENCE_REALTIME_SNAPSHOT",
|
|
checkpointCodecId: "ReferenceRealtimeCheckpoint",
|
|
barrier: "NONE",
|
|
},
|
|
delivery: "INVALIDATION_HINT",
|
|
stateBearing: false,
|
|
});
|
|
const harness = createHarness({
|
|
registry,
|
|
recover: async () =>
|
|
realtimeSuccess({
|
|
kind: "SNAPSHOT_RESET",
|
|
checkpoint: {
|
|
recoveryMode: "SNAPSHOT_ONLY",
|
|
streamEpoch: "stream-epoch-0001",
|
|
lastAppliedSequence: "0",
|
|
resumeCursor: null,
|
|
snapshotRevision: "snapshot-revision-0",
|
|
},
|
|
}),
|
|
});
|
|
await harness.initialize();
|
|
expect(harness.coordinator.inspect(STREAM_ID).freshness).toBe("STALE");
|
|
|
|
await expect(
|
|
harness.coordinator.accept(
|
|
harness.event({
|
|
recoveryMode: "SNAPSHOT_ONLY",
|
|
resumeCursor: null,
|
|
}),
|
|
),
|
|
).resolves.toMatchObject({
|
|
ok: true,
|
|
value: { outcome: "APPLIED" },
|
|
});
|
|
expect(harness.coordinator.inspect(STREAM_ID).freshness).toBe("STALE");
|
|
});
|
|
|
|
it("requires an exact transport barrier before promoting a recovered stream to CURRENT", async () => {
|
|
const harness = createHarness();
|
|
const recovered = await harness.coordinator.recover(
|
|
STREAM_ID,
|
|
"INITIALIZE",
|
|
);
|
|
expect(recovered.ok).toBe(true);
|
|
if (!recovered.ok) return;
|
|
await Promise.resolve();
|
|
|
|
expect(harness.coordinator.inspect(STREAM_ID)).toMatchObject({
|
|
freshness: "STALE",
|
|
awaitingTransportBarrier: true,
|
|
});
|
|
expect(
|
|
harness.coordinator.confirmTransportBarrier(
|
|
STREAM_ID,
|
|
Object.freeze({
|
|
...recovered.value,
|
|
}) as typeof recovered.value,
|
|
),
|
|
).toMatchObject({
|
|
ok: false,
|
|
error: { kind: "PROTOCOL_MISMATCH" },
|
|
});
|
|
const publicResumeState =
|
|
harness.coordinator.getResumeState(STREAM_ID);
|
|
expect(publicResumeState).not.toBe(recovered.value);
|
|
expect(
|
|
harness.coordinator.confirmTransportBarrier(
|
|
STREAM_ID,
|
|
publicResumeState as typeof recovered.value,
|
|
),
|
|
).toMatchObject({
|
|
ok: false,
|
|
error: { kind: "PROTOCOL_MISMATCH" },
|
|
});
|
|
expect(
|
|
harness.coordinator.confirmTransportBarrier(
|
|
STREAM_ID,
|
|
recovered.value,
|
|
),
|
|
).toEqual(realtimeSuccess(undefined));
|
|
expect(harness.coordinator.inspect(STREAM_ID)).toMatchObject({
|
|
freshness: "CURRENT",
|
|
awaitingTransportBarrier: false,
|
|
});
|
|
});
|
|
|
|
it("marks a successful invalidation hint stale until authoritative refresh", async () => {
|
|
const registry = createTestRealtimeRegistry({
|
|
recovery: {
|
|
mode: "SNAPSHOT_ONLY",
|
|
snapshotOperationId: "GET_REFERENCE_REALTIME_SNAPSHOT",
|
|
checkpointCodecId: "ReferenceRealtimeCheckpoint",
|
|
barrier: "CONNECT_BUFFER",
|
|
},
|
|
delivery: "INVALIDATION_HINT",
|
|
stateBearing: false,
|
|
});
|
|
const harness = createHarness({
|
|
registry,
|
|
recover: async () =>
|
|
realtimeSuccess({
|
|
kind: "SNAPSHOT_RESET",
|
|
checkpoint: {
|
|
recoveryMode: "SNAPSHOT_ONLY",
|
|
streamEpoch: "stream-epoch-0001",
|
|
lastAppliedSequence: "0",
|
|
resumeCursor: null,
|
|
snapshotRevision: "snapshot-revision-0",
|
|
},
|
|
}),
|
|
});
|
|
await harness.initialize();
|
|
expect(harness.coordinator.inspect(STREAM_ID).freshness).toBe(
|
|
"CURRENT",
|
|
);
|
|
|
|
await expect(
|
|
harness.coordinator.accept(
|
|
harness.event({
|
|
recoveryMode: "SNAPSHOT_ONLY",
|
|
resumeCursor: null,
|
|
}),
|
|
),
|
|
).resolves.toMatchObject({
|
|
ok: true,
|
|
value: { outcome: "APPLIED" },
|
|
});
|
|
expect(harness.coordinator.inspect(STREAM_ID).freshness).toBe("STALE");
|
|
});
|
|
|
|
it("recovers instead of evicting and continuing after dedupe capacity", async () => {
|
|
const harness = createHarness({
|
|
registry: createTestRealtimeRegistry({
|
|
limits: {
|
|
...TEST_LIMITS,
|
|
maxDedupeEntries: 1,
|
|
},
|
|
}),
|
|
});
|
|
await harness.initialize();
|
|
await harness.coordinator.accept(harness.event());
|
|
|
|
await expect(
|
|
harness.coordinator.accept(
|
|
harness.event({
|
|
eventId: "event-00000002",
|
|
sequence: "2",
|
|
resumeCursor: "cursor-00000002",
|
|
}),
|
|
),
|
|
).resolves.toMatchObject({
|
|
ok: true,
|
|
value: {
|
|
outcome: "RECOVERED",
|
|
reason: "DEDUPE_OVERFLOW",
|
|
},
|
|
});
|
|
expect(harness.effects).toHaveBeenCalledTimes(1);
|
|
expect(harness.coordinator.inspect(STREAM_ID).dedupeEntries).toBe(0);
|
|
});
|
|
|
|
it("rejects a regressing same-epoch recovery checkpoint atomically", async () => {
|
|
let checkpoint = "5";
|
|
const harness = createHarness({
|
|
recover: async () => realtimeSuccess(snapshotCommit(checkpoint)),
|
|
});
|
|
await harness.initialize();
|
|
checkpoint = "4";
|
|
|
|
await expect(
|
|
harness.coordinator.recover(STREAM_ID, "CURSOR_EXPIRED"),
|
|
).resolves.toMatchObject({
|
|
ok: false,
|
|
error: { kind: "PROTOCOL_MISMATCH" },
|
|
});
|
|
expect(harness.coordinator.getResumeState(STREAM_ID)).toMatchObject({
|
|
lastAppliedSequence: "5",
|
|
});
|
|
});
|
|
|
|
it("coalesces recovery, closes idempotently and rejects forged DTOs", async () => {
|
|
const pending = deferred<RealtimeResult<RealtimeRecoveryCommit>>();
|
|
const recover = vi.fn(async () => pending.promise);
|
|
const harness = createHarness({ recover });
|
|
|
|
const first = harness.coordinator.recover(STREAM_ID, "INITIALIZE");
|
|
const second = harness.coordinator.recover(STREAM_ID, "SEQUENCE_GAP");
|
|
await vi.waitFor(() => expect(recover).toHaveBeenCalledOnce());
|
|
pending.resolve(realtimeSuccess(snapshotCommit("0")));
|
|
await expect(first).resolves.toMatchObject({ ok: true });
|
|
await expect(second).resolves.toMatchObject({ ok: true });
|
|
|
|
harness.coordinator.close();
|
|
harness.coordinator.close();
|
|
expect(harness.coordinator.getResumeState(STREAM_ID)).toBeNull();
|
|
await expect(
|
|
harness.coordinator.accept(
|
|
Object.freeze({}) as ValidatedRealtimeEventDto,
|
|
),
|
|
).resolves.toMatchObject({
|
|
ok: false,
|
|
error: { kind: "MALFORMED_EVENT" },
|
|
});
|
|
await expect(
|
|
harness.coordinator.accept(harness.event()),
|
|
).resolves.toEqual({
|
|
ok: true,
|
|
value: { outcome: "DROPPED", reason: "CLOSED" },
|
|
});
|
|
});
|
|
|
|
it("returns the exact shared checkpoint to an event waiting on active recovery", async () => {
|
|
const pending = deferred<RealtimeResult<RealtimeRecoveryCommit>>();
|
|
const recover = vi.fn(async () => pending.promise);
|
|
const harness = createHarness({ recover });
|
|
|
|
const recovering = harness.coordinator.recover(
|
|
STREAM_ID,
|
|
"INITIALIZE",
|
|
);
|
|
await vi.waitFor(() => expect(recover).toHaveBeenCalledOnce());
|
|
const accepting = harness.coordinator.accept(harness.event());
|
|
pending.resolve(realtimeSuccess(snapshotCommit("0")));
|
|
|
|
const recovered = await recovering;
|
|
const accepted = await accepting;
|
|
expect(recovered.ok).toBe(true);
|
|
expect(accepted.ok).toBe(true);
|
|
if (!recovered.ok || !accepted.ok) return;
|
|
expect(accepted.value).toMatchObject({
|
|
outcome: "RECOVERED",
|
|
reason: "INITIALIZE",
|
|
});
|
|
if (accepted.value.outcome !== "RECOVERED") return;
|
|
expect(accepted.value.resumeState).toBe(recovered.value);
|
|
});
|
|
|
|
it("aborts only an event waiter without cancelling shared recovery", async () => {
|
|
const pending = deferred<RealtimeResult<RealtimeRecoveryCommit>>();
|
|
let recoverySignal: AbortSignal | undefined;
|
|
const harness = createHarness({
|
|
recover: async (request) => {
|
|
recoverySignal = request.signal;
|
|
return pending.promise;
|
|
},
|
|
});
|
|
const recovering = harness.coordinator.recover(
|
|
STREAM_ID,
|
|
"INITIALIZE",
|
|
);
|
|
await vi.waitFor(() => expect(recoverySignal).toBeDefined());
|
|
const controller = new AbortController();
|
|
const accepting = harness.coordinator.accept(
|
|
harness.event(),
|
|
controller.signal,
|
|
);
|
|
|
|
controller.abort();
|
|
await expect(accepting).resolves.toEqual(
|
|
realtimeFailure("ABORTED", "RECEIVE"),
|
|
);
|
|
expect(recoverySignal?.aborted).toBe(false);
|
|
|
|
pending.resolve(realtimeSuccess(snapshotCommit("0")));
|
|
await expect(recovering).resolves.toMatchObject({ ok: true });
|
|
});
|
|
|
|
it("aborts the active recovery authority when closed", async () => {
|
|
let recoverySignal: AbortSignal | undefined;
|
|
const harness = createHarness({
|
|
recover: async (request) => {
|
|
recoverySignal = request.signal;
|
|
return new Promise((resolve) => {
|
|
request.signal.addEventListener(
|
|
"abort",
|
|
() => resolve(realtimeFailure("ABORTED", "RECOVER")),
|
|
{ once: true },
|
|
);
|
|
});
|
|
},
|
|
});
|
|
|
|
const recovering = harness.coordinator.recover(
|
|
STREAM_ID,
|
|
"INITIALIZE",
|
|
);
|
|
await vi.waitFor(() => expect(recoverySignal).toBeDefined());
|
|
harness.coordinator.close();
|
|
|
|
expect(recoverySignal?.aborted).toBe(true);
|
|
await expect(recovering).resolves.toMatchObject({
|
|
ok: false,
|
|
error: { kind: "CLOSED", operation: "RECOVER" },
|
|
});
|
|
});
|
|
|
|
it("does not commit a non-cooperative recovery after caller cancellation", async () => {
|
|
const pending = deferred<RealtimeResult<RealtimeRecoveryCommit>>();
|
|
let recoveryRequest: RealtimeRecoveryRequest | undefined;
|
|
const harness = createHarness({
|
|
recover: async (request) => {
|
|
recoveryRequest = request;
|
|
return pending.promise;
|
|
},
|
|
});
|
|
const controller = new AbortController();
|
|
const recovering = harness.coordinator.recover(
|
|
STREAM_ID,
|
|
"INITIALIZE",
|
|
controller.signal,
|
|
);
|
|
await vi.waitFor(() => expect(recoveryRequest).toBeDefined());
|
|
|
|
controller.abort();
|
|
expect(recoveryRequest?.isCurrent()).toBe(false);
|
|
pending.resolve(realtimeSuccess(snapshotCommit("0")));
|
|
|
|
await expect(recovering).resolves.toEqual(
|
|
realtimeFailure("ABORTED", "RECOVER"),
|
|
);
|
|
expect(recoveryRequest?.signal.aborted).toBe(true);
|
|
expect(recoveryRequest?.isCurrent()).toBe(false);
|
|
expect(harness.coordinator.getResumeState(STREAM_ID)).toBeNull();
|
|
expect(harness.coordinator.inspect(STREAM_ID).freshness).toBe(
|
|
"UNKNOWN",
|
|
);
|
|
});
|
|
|
|
it("emits only closed redacted observations and ignores sink failure", async () => {
|
|
const observations: RealtimeObservation[] = [];
|
|
const harness = createHarness({
|
|
observe(observation) {
|
|
observations.push(observation);
|
|
throw new Error("diagnostic sink unavailable");
|
|
},
|
|
});
|
|
await harness.initialize();
|
|
await harness.coordinator.accept(
|
|
harness.event({
|
|
eventId: "sensitive-event-id",
|
|
resumeCursor: "sensitive-cursor",
|
|
payload: { value: "sensitive-payload" },
|
|
}),
|
|
);
|
|
|
|
const serialized = JSON.stringify(observations);
|
|
expect(serialized).not.toContain("sensitive-event-id");
|
|
expect(serialized).not.toContain("sensitive-cursor");
|
|
expect(serialized).not.toContain("sensitive-payload");
|
|
expect(serialized).not.toContain("scope-binding-0001");
|
|
expect(observations).toEqual(
|
|
expect.arrayContaining([
|
|
expect.objectContaining({
|
|
operation: "RECOVER",
|
|
outcome: "RECOVERED",
|
|
reason: "INITIALIZE",
|
|
}),
|
|
expect.objectContaining({
|
|
operation: "APPLY",
|
|
outcome: "APPLIED",
|
|
}),
|
|
]),
|
|
);
|
|
});
|
|
});
|
|
|
|
type HarnessOptions = Readonly<{
|
|
registry?: RealtimePolicyRegistry;
|
|
mappers?: typeof TEST_MAPPERS | Readonly<Record<string, typeof TEST_MAPPER>>;
|
|
apply?: RealtimeEventAuthority["effects"]["apply"];
|
|
recover?: (
|
|
request: RealtimeRecoveryRequest,
|
|
) => Promise<RealtimeResult<RealtimeRecoveryCommit>>;
|
|
observe?: (observation: RealtimeObservation) => void;
|
|
}>;
|
|
|
|
function createHarness(options: HarnessOptions = {}) {
|
|
const registry = options.registry ?? createTestRealtimeRegistry();
|
|
const codec = createTestRealtimeCodec(registry);
|
|
let current = true;
|
|
const recoveryReasons: string[] = [];
|
|
let defaultRecoveryCount = 0;
|
|
const effects = vi.fn(
|
|
options.apply ??
|
|
(async () => realtimeSuccess(undefined)),
|
|
);
|
|
const recover = vi.fn(
|
|
options.recover ??
|
|
(async (request: RealtimeRecoveryRequest) => {
|
|
recoveryReasons.push(request.reason);
|
|
defaultRecoveryCount += 1;
|
|
return realtimeSuccess(
|
|
snapshotCommit(
|
|
"0",
|
|
`stream-epoch-${String(defaultRecoveryCount).padStart(4, "0")}`,
|
|
),
|
|
);
|
|
}),
|
|
);
|
|
const authority: RealtimeEventAuthority = {
|
|
effects: { apply: effects },
|
|
recovery: {
|
|
async recover(request) {
|
|
if (options.recover) recoveryReasons.push(request.reason);
|
|
return recover(request);
|
|
},
|
|
},
|
|
};
|
|
const coordinator = createRealtimeStreamCoordinator({
|
|
registry,
|
|
mappers: options.mappers ?? TEST_MAPPERS,
|
|
authority,
|
|
scope: {
|
|
generation: 7,
|
|
scopeBinding: "scope-binding-0001",
|
|
isCurrent: () => current,
|
|
},
|
|
now: () => 10_000,
|
|
observe: options.observe,
|
|
});
|
|
|
|
return {
|
|
coordinator,
|
|
effects,
|
|
recoveryReasons,
|
|
event(overrides: EventOverrides = {}) {
|
|
const decoded = codec.decode(realtimeEventJson(overrides));
|
|
if (!decoded.ok) {
|
|
throw new Error(`test event failed: ${decoded.error.kind}`);
|
|
}
|
|
return decoded.value;
|
|
},
|
|
async initialize() {
|
|
const result = await coordinator.recover(
|
|
STREAM_ID,
|
|
"INITIALIZE",
|
|
);
|
|
await Promise.resolve();
|
|
if (
|
|
result.ok &&
|
|
coordinator.inspect(STREAM_ID).awaitingTransportBarrier
|
|
) {
|
|
const confirmed = coordinator.confirmTransportBarrier(
|
|
STREAM_ID,
|
|
result.value,
|
|
);
|
|
if (!confirmed.ok) {
|
|
throw new Error("test transport barrier failed");
|
|
}
|
|
}
|
|
return result;
|
|
},
|
|
setCurrent(value: boolean) {
|
|
current = value;
|
|
},
|
|
};
|
|
}
|
|
|
|
function snapshotCommit(
|
|
sequence: string,
|
|
streamEpoch = "stream-epoch-0001",
|
|
): RealtimeRecoveryCommit {
|
|
return Object.freeze({
|
|
kind: "SNAPSHOT_RESET",
|
|
checkpoint: Object.freeze({
|
|
recoveryMode: "CURSOR",
|
|
streamEpoch,
|
|
lastAppliedSequence: sequence,
|
|
resumeCursor: `cursor-snapshot-${sequence}`,
|
|
snapshotRevision: `snapshot-revision-${sequence}`,
|
|
}),
|
|
});
|
|
}
|
|
|
|
function deferred<Value>() {
|
|
let resolve!: (value: Value) => void;
|
|
const promise = new Promise<Value>((next) => {
|
|
resolve = next;
|
|
});
|
|
return { promise, resolve };
|
|
}
|