diff --git a/src/adapters/realtime/live-poll-handoff-coordinator.ts b/src/adapters/realtime/live-poll-handoff-coordinator.ts index 0946e5e..7ea3606 100644 --- a/src/adapters/realtime/live-poll-handoff-coordinator.ts +++ b/src/adapters/realtime/live-poll-handoff-coordinator.ts @@ -173,6 +173,23 @@ export function createLivePollHandoffCoordinator( * running. Membership keeps `close()` honest about quiescence. */ const retiredWriters = new Set>(); + /** + * RT-RR-04. Checkpoint work is an external authority call like a writer tail, + * so it belongs in a physical-task registry from invocation to settlement. + * Racing it against a timeout bounded the public wait but left `close()` free + * to report success while the checkpoint was still running. + */ + const checkpointTasks = new Set>(); + + /** RT-RR-03. Prunes a settled writer without needing another `close()`. */ + function trackRetiredWriter(lease: InternalWriterLease): void { + retiredWriters.add(lease); + void lease.tail + .catch(() => undefined) + .finally(() => { + retiredWriters.delete(lease); + }); + } active = createWriterLease(dependencies.initial.writer); @@ -573,6 +590,12 @@ export function createLivePollHandoffCoordinator( (value) => ({ kind: "VALUE" as const, value }), () => ({ kind: "REJECTED" as const }), ); + // RT-RR-04. Registered at the moment the authority is called, and removed + // only when it settles, so `close()` cannot report quiescence over it. + checkpointTasks.add(operation); + void operation.finally(() => { + checkpointTasks.delete(operation); + }); const timeout = Promise.resolve() .then(async () => { await clock.sleep( @@ -643,6 +666,33 @@ export function createLivePollHandoffCoordinator( return realtimeSuccess(undefined); } + /** + * RT-RR-04. The writer-tail quiescence bound applied to any retained physical + * task, so checkpoint work is proved settled on the same terms. + */ + async function awaitTaskQuiescence( + task: Promise, + ): Promise { + const timer = new AbortController(); + const settled = task.then( + () => "QUIESCED" as const, + () => "QUIESCED" as const, + ); + const timeout = Promise.resolve() + .then(async () => { + await clock.sleep(limits.quiescenceTimeoutMs, timer.signal); + return "TIMED_OUT" as const; + }) + .catch(() => + timer.signal.aborted + ? ("QUIESCED" as const) + : ("TIMER_FAILED" as const), + ); + const outcome = await Promise.race([settled, timeout]); + timer.abort(); + return outcome; + } + async function awaitQuiescence( lease: InternalWriterLease, ): Promise { @@ -670,7 +720,13 @@ export function createLivePollHandoffCoordinator( } function close(): Promise> { - closePromise ??= performClose(); + // RT-RR-03. Only a close that is still running is shared. Caching the first + // timeout forever meant a writer that later settled could never be proved + // quiescent: every subsequent close replayed the stale failure and the + // retained registry could never be pruned. + closePromise ??= performClose().finally(() => { + closePromise = null; + }); return closePromise; } @@ -693,18 +749,23 @@ export function createLivePollHandoffCoordinator( probe = null; selectedProbe?.buffer.splice(0); if (selectedProbe) selectedProbe.bufferedBytes = 0; - for (const writer of writers) writer.controller.abort(); - const outcomes = await Promise.all( - writers.map(async (writer) => await awaitQuiescence(writer)), - ); + for (const writer of writers) { + // RT-RR-03. Every writer this close fences is retained until its tail + // actually settles, so a later close still sees a writer that has not + // finished — and stops seeing it the moment it does. + trackRetiredWriter(writer); + writer.controller.abort(); + } + const outcomes = await Promise.all([ + ...writers.map(async (writer) => await awaitQuiescence(writer)), + // RT-RR-04. Checkpoint work is drained on the same terms as a writer tail. + ...[...checkpointTasks].map( + async (task) => await awaitTaskQuiescence(task), + ), + ]); quiescing = null; transitionCandidate = null; transitioning = false; - for (const [index, writer] of writers.entries()) { - // Only an actually settled writer leaves the retired set; the rest keep - // the coordinator DRAINING. - if (outcomes[index] === "QUIESCED") retiredWriters.delete(writer); - } if (outcomes.includes("TIMED_OUT")) { return handoffFailure("IDLE_TIMEOUT", "CLOSE"); } @@ -802,7 +863,7 @@ export function createLivePollHandoffCoordinator( transitionCandidate, ])) { writer.controller.abort(); - retiredWriters.add(writer); + trackRetiredWriter(writer); } active = null; if (probe) { diff --git a/src/adapters/realtime/stream-coordinator.ts b/src/adapters/realtime/stream-coordinator.ts index 9249f73..4e159c2 100644 --- a/src/adapters/realtime/stream-coordinator.ts +++ b/src/adapters/realtime/stream-coordinator.ts @@ -129,10 +129,23 @@ type StreamState = { closed: boolean; lifecycle: RealtimeStreamLifecycle; /** - * Tasks whose public wait already ended but whose underlying promise has not - * settled. They keep the stream `DRAINING` and block new admission. + * RT-RR-01. Every physical task this coordinator has handed to an external + * authority, from the moment of the call until it settles. Registering only + * after a timeout meant a `close()` that arrived first saw an empty set and + * reported quiescence while the raw task was still running. */ retainedTasks: Set>; + /** + * The subset of `retainedTasks` whose public wait already ended. These are + * what keep the stream `DRAINING` and block new admission. + */ + timedOutTasks: Set>; + /** + * RT-RR-02. Set when a task was abandoned at its deadline. The resume token + * is discarded with it, so the next admitted event cannot skip authoritative + * recovery on the strength of state a timed-out effect may have invalidated. + */ + recoveryRequired: boolean; }; const SNAPSHOT_CHECKPOINT_KEYS = Object.freeze([ @@ -186,6 +199,19 @@ export function createRealtimeStreamCoordinator( revokeAndAbort: () => void, ): Promise { task.catch(() => {}); + // RT-RR-01. The task is a physical effect the moment it is created, so it + // is registered here rather than when its public wait happens to expire. + state.retainedTasks.add(task); + void task + .catch(() => {}) + .finally(() => { + state.retainedTasks.delete(task); + state.timedOutTasks.delete(task); + if (state.timedOutTasks.size === 0 && state.lifecycle === "DRAINING") { + state.lifecycle = state.closed ? "CLOSED" : "OPEN"; + if (!state.closed) state.freshness = "STALE"; + } + }); let handle: unknown; const timeout = new Promise((resolve) => { handle = scheduleTimeout(() => resolve(TASK_TIMED_OUT), timeoutMs); @@ -197,16 +223,12 @@ export function createRealtimeStreamCoordinator( // The commit capability is revoked immediately; the work itself is not. revokeAndAbort(); state.lifecycle = "DRAINING"; - state.retainedTasks.add(task); - void task - .catch(() => {}) - .finally(() => { - state.retainedTasks.delete(task); - if (state.retainedTasks.size === 0 && state.lifecycle === "DRAINING") { - state.lifecycle = state.closed ? "CLOSED" : "OPEN"; - if (!state.closed) state.freshness = "STALE"; - } - }); + state.timedOutTasks.add(task); + // RT-RR-02. An abandoned task may have applied part of its effect, so the + // resume token it was based on is no longer authoritative evidence. + state.recoveryRequired = true; + state.resumeState = null; + state.freshness = "UNKNOWN"; return TASK_TIMED_OUT; } @@ -232,6 +254,8 @@ export function createRealtimeStreamCoordinator( closed: false, lifecycle: "OPEN", retainedTasks: new Set(), + timedOutTasks: new Set(), + recoveryRequired: false, }); } @@ -352,6 +376,10 @@ export function createRealtimeStreamCoordinator( return realtimeFailure("ABORTED", "RECEIVE"); } if (closed || state.closed) return dropped("CLOSED"); + // RT-RR-02. Admission happened when this event was queued; execution is a + // second decision. A queue entry admitted before the stream entered + // DRAINING must not start running inside it. + if (state.lifecycle === "DRAINING") return dropped("CLOSED"); if (expectedGeneration !== state.processingGeneration) { return dropped("SCOPE_FENCED"); } @@ -367,7 +395,11 @@ export function createRealtimeStreamCoordinator( signal, ); } - if (state.freshness === "UNKNOWN" || !state.resumeState) { + if ( + state.recoveryRequired || + state.freshness === "UNKNOWN" || + !state.resumeState + ) { return recoverForAccept(state, "INITIALIZE", true, signal); } if (event.envelope.streamEpoch !== state.resumeState.streamEpoch) { @@ -782,6 +814,9 @@ export function createRealtimeStreamCoordinator( validatedResumeState as RealtimeRecoveryCheckpoint; state.resumeState = resumeState; + // RT-RR-02. Authoritative recovery is the only thing that clears the + // requirement a timed-out task imposed. + state.recoveryRequired = false; state.awaitingTransportBarrier = recoveryRequiresTransportBarrier(state.registration); state.barrierCheckpoint = state.awaitingTransportBarrier diff --git a/tests/unit/realtime/live-poll-handoff-coordinator.test.ts b/tests/unit/realtime/live-poll-handoff-coordinator.test.ts index 1514fce..c60d77f 100644 --- a/tests/unit/realtime/live-poll-handoff-coordinator.test.ts +++ b/tests/unit/realtime/live-poll-handoff-coordinator.test.ts @@ -527,12 +527,87 @@ describe("live/poll authoritative writer handoff", () => { ok: false, error: { kind: "IDLE_TIMEOUT", operation: "CLOSE" }, }); - await expect(harness.coordinator.close()).resolves.toMatchObject({ + // RT-RR-03. A second close re-runs rather than replaying a cached verdict. + // The writer is still hung, so it still reports a timeout. + const second = harness.coordinator.close(); + await flush(); + harness.clock.advance(100); + await expect(second).resolves.toMatchObject({ ok: false, error: { kind: "IDLE_TIMEOUT", operation: "CLOSE" }, }); }); + /** + * RT-RR-03. Caching the first timeout forever meant a writer that later + * settled could never be proved quiescent: every subsequent close replayed + * the stale failure and the retained registry could never be pruned. + */ + it("converges to success once a retired writer finally settles", async () => { + let release: ((value: RealtimeResult) => void) | undefined; + const harness = createHarness({ + apply: vi.fn( + async () => + await new Promise>((resolve) => { + release = resolve; + }), + ), + }); + const live = harness.coordinator.currentWriter()!; + void live.write("late-settle", 8); + await flush(); + + const first = harness.coordinator.close(); + await flush(); + harness.clock.advance(100); + await expect(first).resolves.toMatchObject({ + ok: false, + error: { kind: "IDLE_TIMEOUT", operation: "CLOSE" }, + }); + + // The writer finishes after the first close gave up. + release?.(realtimeSuccess(undefined)); + await flush(); + await flush(); + + const second = harness.coordinator.close(); + await flush(); + harness.clock.advance(100); + await expect(second).resolves.toMatchObject({ ok: true }); + }); + + /** + * RT-RR-04. Checkpoint work is an external authority call like a writer + * tail. Racing it against a timeout bounded the public wait but left it out + * of the retained registry, so `close()` could report quiescence while the + * checkpoint was still running. + */ + it("does not report quiescence while a checkpoint is still running", async () => { + let checkpointSignal: AbortSignal | undefined; + const harness = createHarness({ + recover: vi.fn(async ({ signal }) => { + checkpointSignal = signal; + return await new Promise>(() => {}); + }), + }); + + const transition = harness.coordinator.switchToPoll(); + await flush(); + expect(checkpointSignal).toBeDefined(); + + const closing = harness.coordinator.close(); + await flush(); + harness.clock.advance(100); + await expect(closing).resolves.toMatchObject({ + ok: false, + error: { kind: "IDLE_TIMEOUT", operation: "CLOSE" }, + }); + + harness.clock.advance(100); + await flush(); + await transition; + }); + it("rejects invalid initial authority and resource ceilings", () => { expect(() => createLivePollHandoffCoordinator({ diff --git a/tests/unit/realtime/stream-coordinator.test.ts b/tests/unit/realtime/stream-coordinator.test.ts index 6b12a3b..27a02a3 100644 --- a/tests/unit/realtime/stream-coordinator.test.ts +++ b/tests/unit/realtime/stream-coordinator.test.ts @@ -943,6 +943,118 @@ describe("transport-independent realtime stream coordinator", () => { }); }); +/** + * RT-RR-01 / RT-RR-02. A physical effect exists from the moment the coordinator + * calls the authority, not from the moment its public wait expires. Registering + * only on timeout let a `close()` that arrived first see an empty retained set + * and report quiescence while the raw task was still running against the + * authority. + */ +describe("realtime physical task ownership", () => { + it("does not report quiescence while an apply is still running", async () => { + const applyGate = deferred>(); + const harness = createHarness({ + apply: async () => await applyGate.promise, + taskLimits: { effectTimeoutMs: 10_000, drainTimeoutMs: 20 }, + }); + await harness.initialize(); + + const applying = harness.coordinator.accept(harness.event()); + await Promise.resolve(); + await Promise.resolve(); + + // close() arrives long before the effect deadline. + const closed = await harness.coordinator.close(); + expect(closed.ok).toBe(false); + expect(closed.ok ? null : closed.error.kind).toBe("IDLE_TIMEOUT"); + expect(harness.coordinator.lifecycle(STREAM_ID)).toBe("DRAINING"); + + applyGate.resolve(realtimeSuccess(undefined)); + await applying; + }); + + it("reports quiescence once the raw task settles", async () => { + const applyGate = deferred>(); + const harness = createHarness({ + apply: async () => await applyGate.promise, + taskLimits: { effectTimeoutMs: 10_000, drainTimeoutMs: 200 }, + }); + await harness.initialize(); + + const applying = harness.coordinator.accept(harness.event()); + await Promise.resolve(); + const closing = harness.coordinator.close(); + applyGate.resolve(realtimeSuccess(undefined)); + await applying; + + expect(await closing).toMatchObject({ ok: true }); + }); + + it("does not start a queued event once the stream is DRAINING", async () => { + const firstApply = deferred>(); + let applyCalls = 0; + const harness = createHarness({ + apply: async () => { + applyCalls += 1; + if (applyCalls === 1) return await firstApply.promise; + return realtimeSuccess(undefined); + }, + taskLimits: { effectTimeoutMs: 15, drainTimeoutMs: 50 }, + }); + await harness.initialize(); + + const first = harness.coordinator.accept( + harness.event({ eventId: "event-0001", sequence: "1" }), + ); + // Queued behind the first while it is still inside its deadline. + const second = harness.coordinator.accept( + harness.event({ eventId: "event-0002", sequence: "2" }), + ); + + await first; + expect(harness.coordinator.lifecycle(STREAM_ID)).toBe("DRAINING"); + + // The queued event is dropped at execution time, not applied and not + // recovered: admission happened before DRAINING, execution happens inside + // it, and the second decision is the one that counts. + const secondResult = await second; + expect(secondResult).toMatchObject({ + ok: true, + value: { outcome: "DROPPED", reason: "CLOSED" }, + }); + expect(applyCalls).toBe(1); + + firstApply.resolve(realtimeSuccess(undefined)); + }); + + it("discards the resume token when an effect is abandoned", async () => { + const firstApply = deferred>(); + let applyCalls = 0; + const harness = createHarness({ + apply: async () => { + applyCalls += 1; + if (applyCalls === 1) return await firstApply.promise; + return realtimeSuccess(undefined); + }, + taskLimits: { effectTimeoutMs: 15, drainTimeoutMs: 50 }, + }); + await harness.initialize(); + expect(harness.coordinator.inspect(STREAM_ID).hasResumeState).toBe(true); + + await harness.coordinator.accept( + harness.event({ eventId: "event-0001", sequence: "1" }), + ); + + // The abandoned effect may have applied part of its change, so the token it + // was based on is no longer authoritative evidence. + const inspection = harness.coordinator.inspect(STREAM_ID); + expect(inspection.hasResumeState).toBe(false); + expect(inspection.freshness).toBe("UNKNOWN"); + + firstApply.resolve(realtimeSuccess(undefined)); + }); +}); + type HarnessOptions = Readonly<{ registry?: RealtimePolicyRegistry; mappers?: typeof TEST_MAPPERS | Readonly>;