authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-07-31 16:14:15-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-07-31 16:14:15-07:00
logca72e381c64f25d771236eecf70219f68e5f365b
treef290b483b804b277e3a4ad263e79ab8c557fb8d9
parent491219bf1f4db7144b747fe02af0cbc075668b12
signature Signed by PGP key B5690EEEBB952194

fix(server): bound thread catch-up replay and stop full-DB snapshot hydration (#5147)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

4 files changed, 201 insertions(+), 32 deletions(-)

apps/server/src/cli/project.ts+3-1
......@@ -337,7 +337,9 @@ const dispatchLiveOrchestrationCommand = (
337337
338338const getOfflineSnapshot = Effect.fn("getOfflineSnapshot")(function* () {
339339 const projectionSnapshotQuery = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery;
340 return yield* projectionSnapshotQuery.getSnapshot();
340 // Project commands only read the project list, so use the lightweight
341 // command read model instead of hydrating every thread body in the database.
342 return yield* projectionSnapshotQuery.getCommandReadModel();
341343});
342344
343345const tryResolveLiveProjectExecutionMode = Effect.fn("tryResolveLiveProjectExecutionMode")(
apps/server/src/orchestration/http.ts+6-1
......@@ -32,8 +32,13 @@ export const orchestrationHttpApiLayer = HttpApiBuilder.group(
3232 Effect.fn("environment.orchestration.snapshot")(function* (args) {
3333 yield* annotateEnvironmentRequest(args.endpoint.name);
3434 yield* requireEnvironmentScope(AuthOrchestrationReadScope);
35 // Serve the lightweight command read model (thread bodies empty)
36 // instead of the fully hydrated snapshot. Hydrating every message
37 // and activity payload in the database has OOM-killed servers, and
38 // the route's only consumer (the project CLI) reads projects alone —
39 // UI clients load the shell and per-thread snapshots instead.
3540 return yield* projectionSnapshotQuery
36 .getSnapshot()
41 .getCommandReadModel()
3742 .pipe(
3843 Effect.catch((cause) =>
3944 failEnvironmentInternal("orchestration_snapshot_failed", cause),
apps/server/src/server.test.ts+143
......@@ -6005,6 +6005,149 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
60056005 }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive),
60066006 );
60076007
6008 it.effect("subscribeThread sends a fresh snapshot instead of replaying a large gap", () =>
6009 Effect.gen(function* () {
6010 let readEventsCalls = 0;
6011 const thread = makeDefaultOrchestrationReadModel().threads[0]!;
6012
6013 yield* buildAppUnderTest({
6014 layers: {
6015 orchestrationEngine: {
6016 // Head is far ahead of the client's afterSequence (gap > 1000).
6017 latestSequence: Effect.succeed(100_000),
6018 readEvents: () =>
6019 Stream.sync(() => {
6020 readEventsCalls += 1;
6021 return {} as OrchestrationEvent;
6022 }),
6023 },
6024 projectionSnapshotQuery: {
6025 getThreadDetailSnapshot: () =>
6026 Effect.succeed(Option.some({ snapshotSequence: 100_000, thread })),
6027 },
6028 },
6029 });
6030
6031 const wsUrl = yield* getWsServerUrl("/ws");
6032 const items = yield* Effect.scoped(
6033 withWsRpcClient(wsUrl, (client) =>
6034 client[ORCHESTRATION_WS_METHODS.subscribeThread]({
6035 threadId: defaultThreadId,
6036 afterSequence: 5,
6037 requestCompletionMarker: true,
6038 }).pipe(Stream.take(2), Stream.runCollect),
6039 ),
6040 );
6041
6042 const [first, second] = Array.from(items);
6043 // Large gap => fresh thread snapshot, and the global replay never starts.
6044 assert.equal(first?.kind, "snapshot");
6045 if (first?.kind === "snapshot") {
6046 assert.equal(first.snapshot.thread.id, defaultThreadId);
6047 assert.equal(first.snapshot.snapshotSequence, 100_000);
6048 }
6049 assert.equal(second?.kind, "synchronized");
6050 assert.equal(readEventsCalls, 0);
6051 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
6052 );
6053
6054 it.effect("subscribeThread replaces a cursor ahead of the authoritative head", () =>
6055 Effect.gen(function* () {
6056 let readEventsCalls = 0;
6057 const thread = makeDefaultOrchestrationReadModel().threads[0]!;
6058
6059 yield* buildAppUnderTest({
6060 layers: {
6061 orchestrationEngine: {
6062 latestSequence: Effect.succeed(5),
6063 readEvents: () =>
6064 Stream.sync(() => {
6065 readEventsCalls += 1;
6066 return {} as OrchestrationEvent;
6067 }),
6068 },
6069 projectionSnapshotQuery: {
6070 getThreadDetailSnapshot: () =>
6071 Effect.succeed(Option.some({ snapshotSequence: 5, thread })),
6072 },
6073 },
6074 });
6075
6076 const wsUrl = yield* getWsServerUrl("/ws");
6077 const first = yield* Effect.scoped(
6078 withWsRpcClient(wsUrl, (client) =>
6079 client[ORCHESTRATION_WS_METHODS.subscribeThread]({
6080 threadId: defaultThreadId,
6081 afterSequence: 10,
6082 }).pipe(Stream.runHead),
6083 ),
6084 );
6085
6086 assert.equal(Option.getOrThrow(first).kind, "snapshot");
6087 assert.equal(readEventsCalls, 0);
6088 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
6089 );
6090
6091 it.effect("subscribeThread bounds catch-up replay to the captured head", () =>
6092 Effect.gen(function* () {
6093 let replayLimit: number | undefined;
6094 const now = "2026-01-01T00:00:00.000Z";
6095 const messageEvent = {
6096 sequence: 3,
6097 eventId: EventId.make("event-replay-message"),
6098 aggregateKind: "thread",
6099 aggregateId: defaultThreadId,
6100 occurredAt: now,
6101 commandId: null,
6102 causationEventId: null,
6103 correlationId: null,
6104 metadata: {},
6105 type: "thread.message-sent",
6106 payload: {
6107 threadId: defaultThreadId,
6108 messageId: MessageId.make("message-replay"),
6109 role: "user",
6110 text: "Replayed message",
6111 turnId: null,
6112 streaming: false,
6113 createdAt: now,
6114 updatedAt: now,
6115 },
6116 } satisfies Extract<OrchestrationEvent, { type: "thread.message-sent" }>;
6117
6118 yield* buildAppUnderTest({
6119 layers: {
6120 orchestrationEngine: {
6121 latestSequence: Effect.succeed(50),
6122 readEvents: (_afterSequence, limit) => {
6123 replayLimit = limit;
6124 return Stream.make(messageEvent);
6125 },
6126 },
6127 },
6128 });
6129
6130 const wsUrl = yield* getWsServerUrl("/ws");
6131 const items = yield* Effect.scoped(
6132 withWsRpcClient(wsUrl, (client) =>
6133 client[ORCHESTRATION_WS_METHODS.subscribeThread]({
6134 threadId: defaultThreadId,
6135 afterSequence: 0,
6136 requestCompletionMarker: true,
6137 }).pipe(Stream.take(2), Stream.runCollect),
6138 ),
6139 );
6140
6141 const [first, second] = Array.from(items);
6142 assert.equal(first?.kind, "event");
6143 assert.equal(first?.kind === "event" ? first.event.sequence : null, 3);
6144 assert.equal(second?.kind, "synchronized");
6145 // The replay is bounded to the head captured before the read, not
6146 // Number.MAX_SAFE_INTEGER.
6147 assert.equal(replayLimit, 50);
6148 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
6149 );
6150
60086151 it.effect("subscribeShell sends a fresh snapshot instead of replaying a large gap", () =>
60096152 Effect.gen(function* () {
60106153 let readEventsCalls = 0;
apps/server/src/ws.ts+49-30
......@@ -298,6 +298,13 @@ const PROVIDER_STATUS_DEBOUNCE_MS = 200;
298298// Matches the event store's default page size (DEFAULT_READ_FROM_SEQUENCE_LIMIT).
299299const SHELL_RESUME_MAX_GAP = 1_000;
300300
301// Same bound for thread resume. The replay reads the *global* event range and
302// filters per-thread afterwards, so a stale cursor far behind the head would
303// otherwise decode every intervening event's payload — reconnects with cursors
304// hundreds of thousands of events behind have OOM-killed servers on large
305// databases. Past this gap the client is reset with a fresh thread snapshot.
306const THREAD_RESUME_MAX_GAP = 1_000;
307
301308function toAuthAccessStreamEvent(
302309 change: PairingGrantStore.BootstrapCredentialChange | SessionStore.SessionCredentialChange,
303310 revision: number,
......@@ -1281,38 +1288,50 @@ const makeWsRpcLayer = (
12811288 // catch-up followed by the buffered/ongoing live events. Overlapping
12821289 // events are deduped by sequence on the client.
12831290 //
1284 // Read the full range after the cursor (not the store's default
1285 // page-bounded limit): the range is normally tiny (a fresh HTTP
1286 // snapshot sequence) and the per-thread filter runs after reading,
1287 // so a global cap could otherwise omit this thread's events.
1291 // The replay is bounded to the projection head captured below. The
1292 // catch-up range is normally tiny (a fresh HTTP snapshot sequence),
1293 // but a stale cached cursor can sit hundreds of thousands of global
1294 // events behind — replaying that decodes every intervening event
1295 // (including every other thread's tool payloads) only to discard
1296 // almost all of them, which has OOM-killed servers on large
1297 // databases. A truncated replay would silently drop this thread's
1298 // events, so past the gap cap we reset the client with a fresh
1299 // thread snapshot instead, exactly like subscribeShell above.
12881300 if (input.afterSequence !== undefined) {
12891301 const afterSequence = input.afterSequence;
1290 const catchUpStream = orchestrationEngine
1291 .readEvents(afterSequence, Number.MAX_SAFE_INTEGER)
1292 .pipe(
1293 Stream.filter(isThisThreadDetailEvent),
1294 Stream.map((event) => ({
1295 kind: "event" as const,
1296 event: projectActivityEvent(event),
1297 })),
1298 Stream.mapError(
1299 (cause) =>
1300 new OrchestrationGetSnapshotError({
1301 message: `Failed to replay thread ${input.threadId} events`,
1302 cause,
1303 }),
1304 ),
1305 );
1306 const afterCatchUp =
1307 input.requestCompletionMarker === true
1308 ? Stream.concat(
1309 Stream.fromEffect(
1310 Queue.offer(liveBuffer, { kind: "synchronized" as const }),
1311 ).pipe(Stream.drain),
1312 bufferedLiveStream,
1313 )
1314 : bufferedLiveStream;
1315 return Stream.concat(catchUpStream, afterCatchUp);
1302 const headSequence = yield* orchestrationEngine.latestSequence;
1303 const replayGap = headSequence - afterSequence;
1304 if (replayGap >= 0 && replayGap <= THREAD_RESUME_MAX_GAP) {
1305 const catchUpStream = orchestrationEngine
1306 .readEvents(afterSequence, replayGap)
1307 .pipe(
1308 Stream.filter(isThisThreadDetailEvent),
1309 Stream.map((event) => ({
1310 kind: "event" as const,
1311 event: projectActivityEvent(event),
1312 })),
1313 Stream.mapError(
1314 (cause) =>
1315 new OrchestrationGetSnapshotError({
1316 message: `Failed to replay thread ${input.threadId} events`,
1317 cause,
1318 }),
1319 ),
1320 );
1321 const afterCatchUp =
1322 input.requestCompletionMarker === true
1323 ? Stream.concat(
1324 Stream.fromEffect(
1325 Queue.offer(liveBuffer, { kind: "synchronized" as const }),
1326 ).pipe(Stream.drain),
1327 bufferedLiveStream,
1328 )
1329 : bufferedLiveStream;
1330 return Stream.concat(catchUpStream, afterCatchUp);
1331 }
1332 // Gap too large (or cursor ahead of authoritative state): fall
1333 // through to the snapshot path so the client converges from a
1334 // fresh thread detail instead of an unbounded replay.
13161335 }
13171336
13181337 const snapshot = yield* projectionSnapshotQuery