From 191a3865e8005d841ecc81daada2d6db55ec868d Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Fri, 31 Jul 2026 14:52:01 -0700 Subject: [PATCH] fix(server): stop unbounded event replay and full-DB snapshot hydration subscribeThread's stale-cursor catch-up read the entire global event log (readEvents with Number.MAX_SAFE_INTEGER) and filtered per-thread in JS. Reconnecting clients with cursors hundreds of thousands of events behind forced the server to page and JSON-decode every intervening event's payload, OOM-killing the backend on large databases. The replay is now bounded to the projection head with the same gap cap subscribeShell already uses; past the cap the client is reset with a fresh thread snapshot, which it already handles. GET /api/orchestration/snapshot and the offline project CLI hydrated every message and activity payload in the database via getSnapshot(). The route's only consumer (the project CLI) reads the project list, so both now use the lightweight getCommandReadModel() with the same wire shape. Co-Authored-By: Claude Fable 5 --- apps/server/src/cli/project.ts | 4 +- apps/server/src/orchestration/http.ts | 7 +- apps/server/src/server.test.ts | 143 ++++++++++++++++++++++++++ apps/server/src/ws.ts | 79 ++++++++------ 4 files changed, 201 insertions(+), 32 deletions(-) diff --git a/apps/server/src/cli/project.ts b/apps/server/src/cli/project.ts index 25733a5e35b..39b3b243112 100644 --- a/apps/server/src/cli/project.ts +++ b/apps/server/src/cli/project.ts @@ -337,7 +337,9 @@ const dispatchLiveOrchestrationCommand = ( const getOfflineSnapshot = Effect.fn("getOfflineSnapshot")(function* () { const projectionSnapshotQuery = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery; - return yield* projectionSnapshotQuery.getSnapshot(); + // Project commands only read the project list, so use the lightweight + // command read model instead of hydrating every thread body in the database. + return yield* projectionSnapshotQuery.getCommandReadModel(); }); const tryResolveLiveProjectExecutionMode = Effect.fn("tryResolveLiveProjectExecutionMode")( diff --git a/apps/server/src/orchestration/http.ts b/apps/server/src/orchestration/http.ts index 659665e47b5..9a5c8c0be39 100644 --- a/apps/server/src/orchestration/http.ts +++ b/apps/server/src/orchestration/http.ts @@ -32,8 +32,13 @@ export const orchestrationHttpApiLayer = HttpApiBuilder.group( Effect.fn("environment.orchestration.snapshot")(function* (args) { yield* annotateEnvironmentRequest(args.endpoint.name); yield* requireEnvironmentScope(AuthOrchestrationReadScope); + // Serve the lightweight command read model (thread bodies empty) + // instead of the fully hydrated snapshot. Hydrating every message + // and activity payload in the database has OOM-killed servers, and + // the route's only consumer (the project CLI) reads projects alone — + // UI clients load the shell and per-thread snapshots instead. return yield* projectionSnapshotQuery - .getSnapshot() + .getCommandReadModel() .pipe( Effect.catch((cause) => failEnvironmentInternal("orchestration_snapshot_failed", cause), diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index fff71dbb4e7..569e8a51c37 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -6005,6 +6005,149 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive), ); + it.effect("subscribeThread sends a fresh snapshot instead of replaying a large gap", () => + Effect.gen(function* () { + let readEventsCalls = 0; + const thread = makeDefaultOrchestrationReadModel().threads[0]!; + + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + // Head is far ahead of the client's afterSequence (gap > 1000). + latestSequence: Effect.succeed(100_000), + readEvents: () => + Stream.sync(() => { + readEventsCalls += 1; + return {} as OrchestrationEvent; + }), + }, + projectionSnapshotQuery: { + getThreadDetailSnapshot: () => + Effect.succeed(Option.some({ snapshotSequence: 100_000, thread })), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeThread]({ + threadId: defaultThreadId, + afterSequence: 5, + requestCompletionMarker: true, + }).pipe(Stream.take(2), Stream.runCollect), + ), + ); + + const [first, second] = Array.from(items); + // Large gap => fresh thread snapshot, and the global replay never starts. + assert.equal(first?.kind, "snapshot"); + if (first?.kind === "snapshot") { + assert.equal(first.snapshot.thread.id, defaultThreadId); + assert.equal(first.snapshot.snapshotSequence, 100_000); + } + assert.equal(second?.kind, "synchronized"); + assert.equal(readEventsCalls, 0); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect("subscribeThread replaces a cursor ahead of the authoritative head", () => + Effect.gen(function* () { + let readEventsCalls = 0; + const thread = makeDefaultOrchestrationReadModel().threads[0]!; + + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + latestSequence: Effect.succeed(5), + readEvents: () => + Stream.sync(() => { + readEventsCalls += 1; + return {} as OrchestrationEvent; + }), + }, + projectionSnapshotQuery: { + getThreadDetailSnapshot: () => + Effect.succeed(Option.some({ snapshotSequence: 5, thread })), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const first = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeThread]({ + threadId: defaultThreadId, + afterSequence: 10, + }).pipe(Stream.runHead), + ), + ); + + assert.equal(Option.getOrThrow(first).kind, "snapshot"); + assert.equal(readEventsCalls, 0); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect("subscribeThread bounds catch-up replay to the captured head", () => + Effect.gen(function* () { + let replayLimit: number | undefined; + const now = "2026-01-01T00:00:00.000Z"; + const messageEvent = { + sequence: 3, + eventId: EventId.make("event-replay-message"), + aggregateKind: "thread", + aggregateId: defaultThreadId, + occurredAt: now, + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.message-sent", + payload: { + threadId: defaultThreadId, + messageId: MessageId.make("message-replay"), + role: "user", + text: "Replayed message", + turnId: null, + streaming: false, + createdAt: now, + updatedAt: now, + }, + } satisfies Extract; + + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + latestSequence: Effect.succeed(50), + readEvents: (_afterSequence, limit) => { + replayLimit = limit; + return Stream.make(messageEvent); + }, + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeThread]({ + threadId: defaultThreadId, + afterSequence: 0, + requestCompletionMarker: true, + }).pipe(Stream.take(2), Stream.runCollect), + ), + ); + + const [first, second] = Array.from(items); + assert.equal(first?.kind, "event"); + assert.equal(first?.kind === "event" ? first.event.sequence : null, 3); + assert.equal(second?.kind, "synchronized"); + // The replay is bounded to the head captured before the read, not + // Number.MAX_SAFE_INTEGER. + assert.equal(replayLimit, 50); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect("subscribeShell sends a fresh snapshot instead of replaying a large gap", () => Effect.gen(function* () { let readEventsCalls = 0; diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 06888ef3f70..909a51a4cf5 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -298,6 +298,13 @@ const PROVIDER_STATUS_DEBOUNCE_MS = 200; // Matches the event store's default page size (DEFAULT_READ_FROM_SEQUENCE_LIMIT). const SHELL_RESUME_MAX_GAP = 1_000; +// Same bound for thread resume. The replay reads the *global* event range and +// filters per-thread afterwards, so a stale cursor far behind the head would +// otherwise decode every intervening event's payload — reconnects with cursors +// hundreds of thousands of events behind have OOM-killed servers on large +// databases. Past this gap the client is reset with a fresh thread snapshot. +const THREAD_RESUME_MAX_GAP = 1_000; + function toAuthAccessStreamEvent( change: PairingGrantStore.BootstrapCredentialChange | SessionStore.SessionCredentialChange, revision: number, @@ -1281,38 +1288,50 @@ const makeWsRpcLayer = ( // catch-up followed by the buffered/ongoing live events. Overlapping // events are deduped by sequence on the client. // - // Read the full range after the cursor (not the store's default - // page-bounded limit): the range is normally tiny (a fresh HTTP - // snapshot sequence) and the per-thread filter runs after reading, - // so a global cap could otherwise omit this thread's events. + // The replay is bounded to the projection head captured below. The + // catch-up range is normally tiny (a fresh HTTP snapshot sequence), + // but a stale cached cursor can sit hundreds of thousands of global + // events behind — replaying that decodes every intervening event + // (including every other thread's tool payloads) only to discard + // almost all of them, which has OOM-killed servers on large + // databases. A truncated replay would silently drop this thread's + // events, so past the gap cap we reset the client with a fresh + // thread snapshot instead, exactly like subscribeShell above. if (input.afterSequence !== undefined) { const afterSequence = input.afterSequence; - const catchUpStream = orchestrationEngine - .readEvents(afterSequence, Number.MAX_SAFE_INTEGER) - .pipe( - Stream.filter(isThisThreadDetailEvent), - Stream.map((event) => ({ - kind: "event" as const, - event: projectActivityEvent(event), - })), - Stream.mapError( - (cause) => - new OrchestrationGetSnapshotError({ - message: `Failed to replay thread ${input.threadId} events`, - cause, - }), - ), - ); - const afterCatchUp = - input.requestCompletionMarker === true - ? Stream.concat( - Stream.fromEffect( - Queue.offer(liveBuffer, { kind: "synchronized" as const }), - ).pipe(Stream.drain), - bufferedLiveStream, - ) - : bufferedLiveStream; - return Stream.concat(catchUpStream, afterCatchUp); + const headSequence = yield* orchestrationEngine.latestSequence; + const replayGap = headSequence - afterSequence; + if (replayGap >= 0 && replayGap <= THREAD_RESUME_MAX_GAP) { + const catchUpStream = orchestrationEngine + .readEvents(afterSequence, replayGap) + .pipe( + Stream.filter(isThisThreadDetailEvent), + Stream.map((event) => ({ + kind: "event" as const, + event: projectActivityEvent(event), + })), + Stream.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: `Failed to replay thread ${input.threadId} events`, + cause, + }), + ), + ); + const afterCatchUp = + input.requestCompletionMarker === true + ? Stream.concat( + Stream.fromEffect( + Queue.offer(liveBuffer, { kind: "synchronized" as const }), + ).pipe(Stream.drain), + bufferedLiveStream, + ) + : bufferedLiveStream; + return Stream.concat(catchUpStream, afterCatchUp); + } + // Gap too large (or cursor ahead of authoritative state): fall + // through to the snapshot path so the client converges from a + // fresh thread detail instead of an unbounded replay. } const snapshot = yield* projectionSnapshotQuery