From 2f29ccbf1a43200a4e8b92840585dd591a3337fa Mon Sep 17 00:00:00 2001 From: DongHyeonka Date: Thu, 13 Aug 2026 23:52:46 +0900 Subject: [PATCH] fix: retain realtime work through draining R-02: add an OPEN/DRAINING/CLOSED lifecycle orthogonal to freshness. Effect and recovery authorities are now awaited under a deadline: on expiry the commit capability is revoked and the work aborted, the caller gets a bounded non-retryable IDLE_TIMEOUT, and the underlying task is retained rather than dropped. A draining stream refuses new events and recovery, and close() returns a Promise that succeeds only once every retained task actually settled, reporting IDLE_TIMEOUT otherwise. R-03: a handoff fail-close moves active, probe, quiescing and transition leases into a retired-writer set before clearing their references, and close() waits on current and retired writers together, so an abandoned non-cooperative writer can no longer make teardown report a false success. Co-Authored-By: Claude Opus 5 (1M context) --- ...ime-events-web-push-and-bounded-polling.md | 20 ++ ...ime-events-web-push-and-bounded-polling.md | 10 + docs/operations/adapter-remediation-ledger.md | 4 +- .../realtime/live-poll-handoff-coordinator.ts | 35 ++- src/adapters/realtime/stream-coordinator.ts | 236 ++++++++++++++++-- .../live-poll-handoff-coordinator.test.ts | 41 +++ .../unit/realtime/stream-coordinator.test.ts | 91 +++++++ 7 files changed, 409 insertions(+), 28 deletions(-) diff --git a/docs/architecture/decisions/VD-28-realtime-events-web-push-and-bounded-polling.md b/docs/architecture/decisions/VD-28-realtime-events-web-push-and-bounded-polling.md index d009671..7c6d32d 100644 --- a/docs/architecture/decisions/VD-28-realtime-events-web-push-and-bounded-polling.md +++ b/docs/architecture/decisions/VD-28-realtime-events-web-push-and-bounded-polling.md @@ -11,6 +11,26 @@ 첫 제품 stream/Web Push를 선택할 때, 또는 backend replay/hosting/provider protocol이 바뀔 때 + +## 스트림 lifecycle은 freshness와 직교한다 (R-02, R-03) + +`RealtimeStreamLifecycle = OPEN | DRAINING | CLOSED`는 freshness +(`UNKNOWN/CURRENT/STALE/RESYNCING`)와 별개다. + +- effect/recovery deadline에 도달하면 commit capability를 즉시 영구 무효화하고 + abort한다. caller에는 bounded `IDLE_TIMEOUT`(non-retryable, operation + `APPLY`/`RECOVER`)을 반환하되 **실제 task는 버리지 않고 retain**한다. +- retain된 task가 하나라도 있으면 stream은 `DRAINING`이고 새 event/recovery + admission을 거절한다. 실제 settlement가 일어나야 `STALE`로 돌아가 + authoritative recovery를 요구하거나, close 요청이면 `CLOSED`가 된다. +- `close()`는 `Promise>`다. 모든 retain task가 실제로 + settle해야 success이고, drain bound를 넘기면 `IDLE_TIMEOUT/CLOSE`를 반환하며 + stream은 계속 `DRAINING`이다. teardown success가 곧 quiescence다. +- LIVE↔POLL overflow fail-close는 active/probe/quiescing/transition lease를 + 모두 abort한 뒤 **retired writer set**으로 옮기고 나서 reference를 지운다. + `close()`는 current와 retired를 dedupe해 함께 기다리므로, 버려진 + non-cooperative writer가 아직 실행 중인데 close가 성공을 보고할 수 없다. + ## 배경 현재 optional recipe catalog는 realtime capability에 diff --git a/docs/architecture/realtime-events-web-push-and-bounded-polling.md b/docs/architecture/realtime-events-web-push-and-bounded-polling.md index a1fa0b8..27c735f 100644 --- a/docs/architecture/realtime-events-web-push-and-bounded-polling.md +++ b/docs/architecture/realtime-events-web-push-and-bounded-polling.md @@ -34,6 +34,16 @@ capability가 설치됐거나 production-ready라는 뜻이 아니다. [Frontend ports, adapters, and boundaries](./frontend-ports-adapters-and-boundaries.md) 를 따른다. + +## Bounded task lease와 DRAINING (R-02, R-03) + +non-cooperative effect/recovery authority 하나가 stream tail 전체를 영구 +wedge하지 못하도록, common coordinator는 각 task를 deadline으로 감싼다. deadline +초과 시 commit capability는 즉시 취소되지만 task 자체는 `retainedTasks`에 남아 +stream을 `DRAINING`으로 유지한다. `close()`는 이 retain 집합이 실제로 settle해야 +성공을 반환한다. handoff coordinator도 같은 원칙으로 fail-close된 writer를 +`retiredWriters`에 보존한다. + ## 0. 현재 상태와 목표 delta 이 문서에서 설계 승인, reference source 존재, production 조합과 target browser의 diff --git a/docs/operations/adapter-remediation-ledger.md b/docs/operations/adapter-remediation-ledger.md index 5814003..9fdc46e 100644 --- a/docs/operations/adapter-remediation-ledger.md +++ b/docs/operations/adapter-remediation-ledger.md @@ -105,8 +105,8 @@ Rollout state starts at `NOT_STARTED`; documented-unimplemented items start at | ID | Activation | Red test command | Fix commit/PR | Rollout state | Rollback trigger | Evidence | | --- | --- | --- | --- | --- | --- | --- | | R-01 | Browser RPC `AVAILABLE_NOT_COMPOSED` | `corepack pnpm exec vitest run tests/unit/browser-rpc/browser-rpc-runtime.test.ts` | — | `NOT_STARTED` | stream lease deadlock | — | -| R-02 | Realtime `NOT_SELECTED` | `corepack pnpm exec vitest run tests/unit/realtime/stream-coordinator.test.ts` | — | `NOT_STARTED` | stream stuck in `DRAINING` | — | -| R-03 | Realtime `NOT_SELECTED` | `corepack pnpm exec vitest run tests/unit/realtime/live-poll-handoff-coordinator.test.ts` | — | `NOT_STARTED` | retired-writer set growth | — | +| R-02 | Realtime `NOT_SELECTED` | `corepack pnpm exec vitest run tests/unit/realtime/stream-coordinator.test.ts` | `fix: retain realtime work through draining` | `FIXED_NOT_RELEASED` | stream stuck in `DRAINING` | Red never-settling effect and recovery → green 27/27; `close()` returns `IDLE_TIMEOUT` while a task is retained and success only after actual settlement | +| R-03 | Realtime `NOT_SELECTED` | `corepack pnpm exec vitest run tests/unit/realtime/live-poll-handoff-coordinator.test.ts` | `fix: retain realtime work through draining` | `FIXED_NOT_RELEASED` | retired-writer set growth | Red overflow fail-close then `close()` → green 11/11; retired writers are waited on and only removed once actually quiesced | | R-04 | Browser RPC `AVAILABLE_NOT_COMPOSED` | `corepack pnpm exec vitest run tests/unit/browser-rpc/browser-rpc-contract.test.ts` | — | `NOT_STARTED` | binding install rejection | — | | R-05 | WebSocket protocol codec | `corepack pnpm exec vitest run tests/unit/realtime/websocket-protocol.test.ts` | — | `NOT_STARTED` | frame rejection regression | — | | R-06 | Browser RPC `AVAILABLE_NOT_COMPOSED` | `corepack pnpm exec vitest run tests/unit/browser-rpc/browser-rpc-runtime.test.ts` | — | `NOT_STARTED` | closed-failure taxonomy drift | — | diff --git a/src/adapters/realtime/live-poll-handoff-coordinator.ts b/src/adapters/realtime/live-poll-handoff-coordinator.ts index 83dd54e..0946e5e 100644 --- a/src/adapters/realtime/live-poll-handoff-coordinator.ts +++ b/src/adapters/realtime/live-poll-handoff-coordinator.ts @@ -168,6 +168,11 @@ export function createLivePollHandoffCoordinator( let quiescing: InternalWriterLease | null = null; let transitionCandidate: InternalWriterLease | null = null; let closePromise: Promise> | null = null; + /** + * R-03. Writers whose lease was fail-closed but whose tail may still be + * running. Membership keeps `close()` honest about quiescence. + */ + const retiredWriters = new Set>(); active = createWriterLease(dependencies.initial.writer); @@ -673,11 +678,15 @@ export function createLivePollHandoffCoordinator( lifecycleGeneration += 1; state = "CLOSED"; transitioning = true; + // R-03. Current and previously retired writers are waited on together and + // deduplicated, so a writer dropped by an overflow fail-close is still + // proved quiescent before close reports success. const writers = uniqueLeases([ active, probe?.lease ?? null, quiescing, transitionCandidate, + ...retiredWriters, ]); active = null; const selectedProbe = probe; @@ -691,6 +700,11 @@ export function createLivePollHandoffCoordinator( 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"); } @@ -771,20 +785,33 @@ export function createLivePollHandoffCoordinator( ); } + /** + * R-03. A fail-close aborts every lease, but the underlying writers may be + * non-cooperative and still running. They move into the retired set before + * their references are cleared, so a later `close()` cannot report success + * while an abandoned writer is still executing. + */ function failClosed(): void { lifecycleGeneration += 1; state = "CLOSED"; transitioning = false; - active?.controller.abort(); - probe?.lease.controller.abort(); - quiescing?.controller.abort(); - transitionCandidate?.controller.abort(); + for (const writer of uniqueLeases([ + active, + probe?.lease ?? null, + quiescing, + transitionCandidate, + ])) { + writer.controller.abort(); + retiredWriters.add(writer); + } active = null; if (probe) { probe.buffer.length = 0; probe.bufferedBytes = 0; } probe = null; + quiescing = null; + transitionCandidate = null; } return Object.freeze({ diff --git a/src/adapters/realtime/stream-coordinator.ts b/src/adapters/realtime/stream-coordinator.ts index d890c77..9249f73 100644 --- a/src/adapters/realtime/stream-coordinator.ts +++ b/src/adapters/realtime/stream-coordinator.ts @@ -41,6 +41,27 @@ import { type RealtimeDataSnapshot, } from "./result.ts"; +/** + * R-02. Lifecycle is orthogonal to freshness. `DRAINING` means the coordinator + * has revoked commit capability and stopped admitting work, but a + * non-cooperative task it started is still running and is deliberately retained + * until it actually settles. + */ +export type RealtimeStreamLifecycle = "OPEN" | "DRAINING" | "CLOSED"; + +export type RealtimeStreamTaskLimits = Readonly<{ + effectTimeoutMs: number; + recoveryTimeoutMs: number; + drainTimeoutMs: number; +}>; + +export const DEFAULT_REALTIME_STREAM_TASK_LIMITS: RealtimeStreamTaskLimits = + Object.freeze({ + effectTimeoutMs: 5_000, + recoveryTimeoutMs: 10_000, + drainTimeoutMs: 5_000, + }); + export type RealtimeStreamCoordinatorDependencies = Readonly<{ registry: RealtimePolicyRegistry; mappers: Readonly>; @@ -48,6 +69,10 @@ export type RealtimeStreamCoordinatorDependencies = Readonly<{ scope: RealtimeScopeSnapshot; now?: () => number; observe?: RealtimeEventObservationSink; + taskLimits?: Partial; + /** Test seam for the bounded task deadline. */ + scheduleTimeout?: (callback: () => void, delayMs: number) => unknown; + clearScheduledTimeout?: (handle: unknown) => void; }>; export type RealtimeStreamCoordinator = Readonly<{ @@ -66,7 +91,13 @@ export type RealtimeStreamCoordinator = Readonly<{ ): RealtimeResult; getResumeState(streamId: StreamRegistrationId): RealtimeResumeState | null; inspect(streamId: StreamRegistrationId): RealtimeStreamInspection; - close(): void; + lifecycle(streamId: StreamRegistrationId): RealtimeStreamLifecycle; + /** + * R-02. Bounded quiescence. Success means every tracked task actually + * settled; `IDLE_TIMEOUT` means the coordinator is still `DRAINING` and the + * caller must not assume a clean teardown. + */ + close(): Promise>; }>; type DedupeEntry = Readonly<{ @@ -96,6 +127,12 @@ type StreamState = { awaitingTransportBarrier: boolean; barrierCheckpoint: RealtimeRecoveryCheckpoint | null; 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. + */ + retainedTasks: Set>; }; const SNAPSHOT_CHECKPOINT_KEYS = Object.freeze([ @@ -120,9 +157,59 @@ export function createRealtimeStreamCoordinator( ): RealtimeStreamCoordinator { assertDependencies(dependencies); const now = dependencies.now ?? Date.now; + const limits: RealtimeStreamTaskLimits = Object.freeze({ + ...DEFAULT_REALTIME_STREAM_TASK_LIMITS, + ...dependencies.taskLimits, + }); + const scheduleTimeout = + dependencies.scheduleTimeout ?? + ((callback: () => void, delayMs: number) => setTimeout(callback, delayMs)); + const clearScheduledTimeout = + dependencies.clearScheduledTimeout ?? + ((handle: unknown) => { + clearTimeout(handle as ReturnType); + }); const states = new Map(); let closed = false; + const TASK_TIMED_OUT = Symbol("REALTIME_TASK_TIMED_OUT"); + + /** + * R-02. Bounds the public wait without discarding the task. A task that + * outlives its deadline is retained so `close()` can report honestly whether + * the stream is actually quiescent. + */ + async function awaitTaskWithinDeadline( + state: StreamState, + task: Promise, + timeoutMs: number, + revokeAndAbort: () => void, + ): Promise { + task.catch(() => {}); + let handle: unknown; + const timeout = new Promise((resolve) => { + handle = scheduleTimeout(() => resolve(TASK_TIMED_OUT), timeoutMs); + }); + const outcome = await Promise.race([task, timeout]); + clearScheduledTimeout(handle); + if (outcome !== TASK_TIMED_OUT) return outcome; + + // 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"; + } + }); + return TASK_TIMED_OUT; + } + for (const registration of dependencies.registry.listStreams()) { states.set(registration.id, { registration, @@ -143,6 +230,8 @@ export function createRealtimeStreamCoordinator( awaitingTransportBarrier: false, barrierCheckpoint: null, closed: false, + lifecycle: "OPEN", + retainedTasks: new Set(), }); } @@ -158,6 +247,11 @@ export function createRealtimeStreamCoordinator( realtimeFailure("MALFORMED_EVENT", "RECEIVE"), ); } + const draining = states.get(event.envelope.streamId); + if (draining && draining.lifecycle === "DRAINING") { + // R-02. New work is refused while a retained task is still running. + return Promise.resolve(realtimeFailure("CLOSED", "RECEIVE")); + } const state = states.get(event.envelope.streamId); if (!state) { return Promise.resolve( @@ -384,19 +478,37 @@ export function createRealtimeStreamCoordinator( !effectAbort.signal.aborted && scopeIsCurrent(dependencies.scope); let effect: unknown; + let effectTimedOut = false; try { - effect = await dependencies.authority.effects.apply( - eventType.effectProfileId, - mapped.value, - Object.freeze({ - streamId: state.registration.id, - eventType: eventType.id, - occurredAt: event.envelope.occurredAt, - scopeGeneration: dependencies.scope.generation, - isCurrent: effectIsCurrent, - }), - effectAbort.signal, + const applied = await awaitTaskWithinDeadline( + state, + Promise.resolve( + dependencies.authority.effects.apply( + eventType.effectProfileId, + mapped.value, + Object.freeze({ + streamId: state.registration.id, + eventType: eventType.id, + occurredAt: event.envelope.occurredAt, + scopeGeneration: dependencies.scope.generation, + isCurrent: effectIsCurrent, + }), + effectAbort.signal, + ), + ), + limits.effectTimeoutMs, + () => { + // Commit capability is revoked permanently for this attempt. + effectLeaseActive = false; + effectAbort.abort(); + }, ); + if (applied === TASK_TIMED_OUT) { + effectTimedOut = true; + effect = null; + } else { + effect = applied; + } } catch { effect = null; } finally { @@ -406,6 +518,17 @@ export function createRealtimeStreamCoordinator( state.activeEffectAbort = null; } } + if (effectTimedOut) { + // R-02. Bounded for the caller; the underlying task stays tracked. + observe({ + operation: "APPLY", + outcome: "FAILED", + streamId: state.registration.id, + eventType: event.envelope.eventType, + reason: "IDLE_TIMEOUT", + }); + return realtimeFailure("IDLE_TIMEOUT", "APPLY", false); + } if (closed || state.closed) { return dropped("CLOSED"); @@ -483,7 +606,7 @@ export function createRealtimeStreamCoordinator( calledFromCurrentJob: boolean, signal?: AbortSignal, ): Promise> { - if (closed || state.closed) { + if (closed || state.closed || state.lifecycle === "DRAINING") { return Promise.resolve(realtimeFailure("CLOSED", "RECOVER")); } if (!scopeIsCurrent(dependencies.scope)) { @@ -542,22 +665,50 @@ export function createRealtimeStreamCoordinator( let recovered: unknown; let recoveryThrew = false; + let recoveryTimedOut = false; try { - recovered = await dependencies.authority.recovery.recover( - Object.freeze({ - streamId: state.registration.id, - reason, - scopeGeneration: dependencies.scope.generation, - signal: recoveryAbort.signal, - isCurrent: recoveryIsCurrent, - }), + const outcome = await awaitTaskWithinDeadline( + state, + Promise.resolve( + dependencies.authority.recovery.recover( + Object.freeze({ + streamId: state.registration.id, + reason, + scopeGeneration: dependencies.scope.generation, + signal: recoveryAbort.signal, + isCurrent: recoveryIsCurrent, + }), + ), + ), + limits.recoveryTimeoutMs, + () => { + recoveryLeaseActive = false; + recoveryAbort.abort(); + }, ); + if (outcome === TASK_TIMED_OUT) { + recoveryTimedOut = true; + recovered = null; + } else { + recovered = outcome; + } } catch { recovered = null; recoveryThrew = true; } finally { recoveryLeaseActive = false; } + if (recoveryTimedOut) { + // R-02. A late checkpoint from this attempt can never commit. + state.freshness = "UNKNOWN"; + observe({ + operation: "RECOVER", + outcome: "FAILED", + streamId: state.registration.id, + reason: "IDLE_TIMEOUT", + }); + return realtimeFailure("IDLE_TIMEOUT", "RECOVER", false); + } if (closed || state.closed) { state.freshness = "UNKNOWN"; return realtimeFailure("CLOSED", "RECOVER"); @@ -806,7 +957,45 @@ export function createRealtimeStreamCoordinator( }); } - function close(): void { + function lifecycleOf(streamId: StreamRegistrationId): RealtimeStreamLifecycle { + const state = states.get(streamId); + if (!state) return "CLOSED"; + return state.lifecycle; + } + + /** + * R-02. `close()` fences immediately but reports honestly: success only when + * every retained task actually settled within the drain bound. + */ + async function close(): Promise> { + fenceAllStates(); + const retained = [...states.values()].flatMap((state) => [ + ...state.retainedTasks, + ]); + if (retained.length === 0) { + for (const state of states.values()) state.lifecycle = "CLOSED"; + return realtimeSuccess(undefined); + } + let handle: unknown; + const drained = await Promise.race([ + Promise.allSettled(retained).then(() => true), + new Promise((resolve) => { + handle = scheduleTimeout( + () => resolve(false), + limits.drainTimeoutMs, + ); + }), + ]); + clearScheduledTimeout(handle); + if (!drained) { + // Still DRAINING: the caller must not treat this as quiescence. + return realtimeFailure("IDLE_TIMEOUT", "CLOSE", false); + } + for (const state of states.values()) state.lifecycle = "CLOSED"; + return realtimeSuccess(undefined); + } + + function fenceAllStates(): void { if (closed) return; closed = true; for (const state of states.values()) { @@ -826,6 +1015,8 @@ export function createRealtimeStreamCoordinator( state.eventIds.clear(); state.sequences.clear(); state.dedupeBytes = 0; + state.lifecycle = + state.retainedTasks.size > 0 ? "DRAINING" : "CLOSED"; } observe({ operation: "CLOSE", @@ -911,6 +1102,7 @@ export function createRealtimeStreamCoordinator( confirmTransportBarrier, getResumeState, inspect, + lifecycle: lifecycleOf, close, }); } diff --git a/tests/unit/realtime/live-poll-handoff-coordinator.test.ts b/tests/unit/realtime/live-poll-handoff-coordinator.test.ts index 9e0a319..1514fce 100644 --- a/tests/unit/realtime/live-poll-handoff-coordinator.test.ts +++ b/tests/unit/realtime/live-poll-handoff-coordinator.test.ts @@ -214,6 +214,47 @@ describe("live/poll authoritative writer handoff", () => { void second; }); + it("tracks a retired active writer after handoff queue overflow", async () => { + let release: ((result: RealtimeResult) => void) | undefined; + const harness = createHarness({ + limits: { + ...limits, + maxActiveQueueCount: 2, + maxActiveQueueBytes: 10, + }, + apply: vi.fn(async ({ value }) => { + if (value === "first") { + return await new Promise>((resolve) => { + release = resolve; + }); + } + return realtimeSuccess(undefined); + }), + }); + const writer = harness.coordinator.currentWriter()!; + const first = writer.write("first", 5); + const second = writer.write("second", 5); + await flush(); + await expect(writer.write("overflow", 1)).resolves.toMatchObject({ + ok: false, + error: { kind: "QUEUE_OVERFLOW" }, + }); + + // R-03. The fail-close dropped the active reference, but the writer is + // still running, so close() must not claim quiescence. + const closing = harness.coordinator.close(); + await flush(); + harness.clock.advance(100); + await expect(closing).resolves.toMatchObject({ + ok: false, + error: { kind: "IDLE_TIMEOUT", operation: "CLOSE" }, + }); + + release?.(realtimeSuccess(undefined)); + void first; + void second; + }); + it("fences and aborts live, waits for quiescence, then recovers before activating poll", async () => { const liveEffect = deferred>(); const checkpoint = deferred>(); diff --git a/tests/unit/realtime/stream-coordinator.test.ts b/tests/unit/realtime/stream-coordinator.test.ts index ec7347c..6b12a3b 100644 --- a/tests/unit/realtime/stream-coordinator.test.ts +++ b/tests/unit/realtime/stream-coordinator.test.ts @@ -29,6 +29,91 @@ import { } from "./fixture.ts"; describe("transport-independent realtime stream coordinator", () => { + it("keeps the stream DRAINING until a non-cooperative effect settles", async () => { + const wedged = deferred>(); + const harness = createHarness({ + apply: async () => wedged.promise, + taskLimits: { effectTimeoutMs: 5, drainTimeoutMs: 5 }, + }); + await harness.initialize(); + + const applied = await harness.coordinator.accept( + harness.event({ + eventId: "event-00000001", + sequence: "1", + resumeCursor: "cursor-00000001", + }), + ); + // Bounded for the caller, and explicitly non-retryable. + expect(applied).toMatchObject({ + ok: false, + error: { kind: "IDLE_TIMEOUT", operation: "APPLY", retryable: false }, + }); + expect(harness.coordinator.lifecycle(STREAM_ID)).toBe("DRAINING"); + + // DRAINING refuses new admission rather than queueing behind the wedge. + await expect( + harness.coordinator.accept( + harness.event({ + eventId: "event-00000002", + sequence: "2", + resumeCursor: "cursor-00000002", + }), + ), + ).resolves.toMatchObject({ ok: false, error: { kind: "CLOSED" } }); + + // close() cannot claim quiescence while the task is still running. + await expect(harness.coordinator.close()).resolves.toMatchObject({ + ok: false, + error: { kind: "IDLE_TIMEOUT", operation: "CLOSE" }, + }); + + // Only actual settlement ends DRAINING. + wedged.resolve(realtimeSuccess(undefined)); + await expect(harness.coordinator.close()).resolves.toMatchObject({ + ok: true, + }); + expect(harness.coordinator.lifecycle(STREAM_ID)).toBe("CLOSED"); + }); + + it("bounds non-cooperative recovery and rejects its late checkpoint", async () => { + const wedged = deferred>(); + let recoveries = 0; + const harness = createHarness({ + recover: async () => { + recoveries += 1; + return recoveries === 1 + ? realtimeSuccess(snapshotCommit("0")) + : wedged.promise; + }, + taskLimits: { recoveryTimeoutMs: 5, drainTimeoutMs: 5 }, + }); + await harness.initialize(); + + const recovered = await harness.coordinator.recover( + STREAM_ID, + "SEQUENCE_GAP", + ); + expect(recovered).toMatchObject({ + ok: false, + error: { kind: "IDLE_TIMEOUT", operation: "RECOVER", retryable: false }, + }); + expect(harness.coordinator.lifecycle(STREAM_ID)).toBe("DRAINING"); + + const beforeLateCommit = harness.coordinator.getResumeState(STREAM_ID); + + // A late checkpoint from the abandoned attempt cannot commit. + wedged.resolve(realtimeSuccess(snapshotCommit("9"))); + for (let flush = 0; flush < 10; flush += 1) await Promise.resolve(); + expect(harness.coordinator.getResumeState(STREAM_ID)).toEqual( + beforeLateCommit, + ); + // Settlement returns the stream to OPEN, marked STALE for an authoritative + // recovery rather than silently trusting the abandoned attempt. + expect(harness.coordinator.lifecycle(STREAM_ID)).toBe("OPEN"); + expect(harness.coordinator.inspect(STREAM_ID).freshness).toBe("STALE"); + }); + it("applies one stream sequentially and commits each cursor after its effect", async () => { const first = deferred>(); const applied: string[] = []; @@ -866,6 +951,11 @@ type HarnessOptions = Readonly<{ request: RealtimeRecoveryRequest, ) => Promise>; observe?: (observation: RealtimeObservation) => void; + taskLimits?: Readonly<{ + effectTimeoutMs?: number; + recoveryTimeoutMs?: number; + drainTimeoutMs?: number; + }>; }>; function createHarness(options: HarnessOptions = {}) { @@ -911,6 +1001,7 @@ function createHarness(options: HarnessOptions = {}) { }, now: () => 10_000, observe: options.observe, + ...(options.taskLimits ? { taskLimits: options.taskLimits } : {}), }); return {