authorgravatar for rcawston@users.noreply.github.comRoss Cawston <rcawston@users.noreply.github.com> 2026-09-01 18:07:56-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-01 18:07:56-07:00
log7e460f429b740180cd72730418262a2df971ba54
tree1a36f8e34c227eeafd33d2f7671212bb0da66e8a
parentc2283ce14628127ef1758974ddab4f5b1f89d47a
signature Signed by PGP key B5690EEEBB952194

fix(server): bound orchestration replay payloads (#8992)

WebSocket reconnect catch-up only checked the number of missing events, then loaded and decoded the whole replay range before sending it. A handful of events with large tool payloads could allocate gigabytes and OOM the server. Before replaying, run a SQL preflight that counts the events in the range and sums their serialized payload bytes with octet_length(). If the range is over the existing row limit or an 8 MiB byte budget, fall back to the fresh snapshot path. Applies to both thread and shell subscriptions. Authored by @rcawston. Review, octet_length fix, and merge by Claude Fable 5.1 in Claude Code on behalf of @t3dotgg.

10 files changed, 230 insertions(+), 2 deletions(-)

apps/server/src/checkpointing/CheckpointDiffQuery.test.ts+5
......@@ -85,6 +85,7 @@ describe("CheckpointDiffQuery.layer", () => {
8585 Effect.die("CheckpointDiffQuery should not request archived shell snapshots"),
8686 getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }),
8787 getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
88 getEventReplayStats: () => Effect.die("unused"),
8889 getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
8990 getProjectShellById: () => Effect.succeed(Option.none()),
9091 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
......@@ -194,6 +195,7 @@ describe("CheckpointDiffQuery.layer", () => {
194195 Effect.die("CheckpointDiffQuery should not request archived shell snapshots"),
195196 getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }),
196197 getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
198 getEventReplayStats: () => Effect.die("unused"),
197199 getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
198200 getProjectShellById: () => Effect.succeed(Option.none()),
199201 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
......@@ -278,6 +280,7 @@ describe("CheckpointDiffQuery.layer", () => {
278280 Effect.die("CheckpointDiffQuery should not request archived shell snapshots"),
279281 getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }),
280282 getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
283 getEventReplayStats: () => Effect.die("unused"),
281284 getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
282285 getProjectShellById: () => Effect.succeed(Option.none()),
283286 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
......@@ -347,6 +350,7 @@ describe("CheckpointDiffQuery.layer", () => {
347350 Effect.die("CheckpointDiffQuery should not request archived shell snapshots"),
348351 getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }),
349352 getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
353 getEventReplayStats: () => Effect.die("unused"),
350354 getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
351355 getProjectShellById: () => Effect.succeed(Option.none()),
352356 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
......@@ -401,6 +405,7 @@ describe("CheckpointDiffQuery.layer", () => {
401405 Effect.die("CheckpointDiffQuery should not request archived shell snapshots"),
402406 getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }),
403407 getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
408 getEventReplayStats: () => Effect.die("unused"),
404409 getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
405410 getProjectShellById: () => Effect.succeed(Option.none()),
406411 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts+1
......@@ -206,6 +206,7 @@ describe("OrchestrationEngine", () => {
206206 getSnapshotSequence: () =>
207207 Effect.succeed({ snapshotSequence: projectionSnapshot.snapshotSequence }),
208208 getCounts: () => Effect.succeed({ projectCount: 1, threadCount: 1 }),
209 getEventReplayStats: () => Effect.die("unused"),
209210 getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
210211 getProjectShellById: () => Effect.succeed(Option.none()),
211212 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts+46
......@@ -922,6 +922,52 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
922922 }),
923923 );
924924
925 it.effect("measures replay payload bytes without decoding event bodies", () =>
926 Effect.gen(function* () {
927 const snapshotQuery = yield* ProjectionSnapshotQuery;
928 const sql = yield* SqlClient.SqlClient;
929
930 yield* sql`DELETE FROM orchestration_events`;
931 yield* sql`
932 INSERT INTO orchestration_events (
933 event_id, aggregate_kind, stream_id, stream_version, event_type, occurred_at,
934 command_id, causation_event_id, correlation_id, actor_kind, payload_json, metadata_json
935 )
936 VALUES
937 (
938 'replay-event-1', 'thread', 'thread-replay', 1, 'thread.activity-appended',
939 '2026-03-01T00:00:00.000Z', NULL, NULL, NULL, 'provider',
940 json_object('output', printf('%.*c', 1000, 'x')), '{}'
941 ),
942 (
943 'replay-event-2', 'thread', 'thread-replay', 2, 'thread.activity-appended',
944 '2026-03-01T00:00:01.000Z', NULL, NULL, NULL, 'provider',
945 json_object('output', printf('%.*c', 2000, 'x')), '{}'
946 ),
947 (
948 'replay-event-3', 'thread', 'thread-replay', 3, 'thread.activity-appended',
949 '2026-03-01T00:00:02.000Z', NULL, NULL, NULL, 'provider',
950 json_object('output', printf('%.*c', 3000, 'x')), '{}'
951 ),
952 (
953 'replay-event-4', 'thread', 'thread-replay', 4, 'thread.activity-appended',
954 '2026-03-01T00:00:03.000Z', NULL, NULL, NULL, 'provider',
955 json_object('output', '😀'), '{}'
956 )
957 `;
958
959 // Bytes, not code points: the 4-byte emoji row is {"output":"😀"}, 17 bytes.
960 const stats = yield* snapshotQuery.getEventReplayStats({
961 fromSequenceExclusive: 1,
962 toSequenceInclusive: 4,
963 });
964 assert.deepStrictEqual(stats, {
965 eventCount: 3,
966 payloadBytes: 5043,
967 });
968 }),
969 );
970
925971 it.effect("reads single-thread checkpoint context without hydrating unrelated threads", () =>
926972 Effect.gen(function* () {
927973 const snapshotQuery = yield* ProjectionSnapshotQuery;
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts+40
......@@ -62,6 +62,7 @@ import * as RepositoryIdentityResolver from "../../project/RepositoryIdentityRes
6262import { ORCHESTRATION_PROJECTOR_NAMES } from "./ProjectionPipeline.ts";
6363import {
6464 ProjectionSnapshotQuery,
65 type ProjectionEventReplayStats,
6566 type ProjectionFullThreadDiffContext,
6667 type ProjectionSnapshotCounts,
6768 type ProjectionThreadCheckpointContext,
......@@ -129,6 +130,14 @@ const ProjectionCountsRowSchema = Schema.Struct({
129130 projectCount: Schema.Number,
130131 threadCount: Schema.Number,
131132});
133const EventReplayStatsInput = Schema.Struct({
134 fromSequenceExclusive: NonNegativeInt,
135 toSequenceInclusive: NonNegativeInt,
136});
137const EventReplayStatsRowSchema = Schema.Struct({
138 eventCount: Schema.Number,
139 payloadBytes: Schema.Number,
140});
132141const ProjectionThreadSearchRequest = Schema.Struct({
133142 pattern: Schema.String,
134143 limit: Schema.Int,
......@@ -812,6 +821,20 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
812821 `,
813822 });
814823
824 const readEventReplayStats = SqlSchema.findOne({
825 Request: EventReplayStatsInput,
826 Result: EventReplayStatsRowSchema,
827 execute: ({ fromSequenceExclusive, toSequenceInclusive }) =>
828 sql`
829 SELECT
830 COUNT(*) AS "eventCount",
831 COALESCE(SUM(octet_length(payload_json)), 0) AS "payloadBytes"
832 FROM orchestration_events
833 WHERE sequence > ${fromSequenceExclusive}
834 AND sequence <= ${toSequenceInclusive}
835 `,
836 });
837
815838 const searchActiveThreadRows = SqlSchema.findAll({
816839 Request: ProjectionThreadSearchRequest,
817840 Result: ProjectionThreadSearchRow,
......@@ -2448,6 +2471,22 @@ pending_approval_requests AS (
24482471 ),
24492472 );
24502473
2474 const getEventReplayStats: ProjectionSnapshotQueryShape["getEventReplayStats"] = (input) =>
2475 readEventReplayStats(input).pipe(
2476 Effect.mapError(
2477 toPersistenceSqlOrDecodeError(
2478 "ProjectionSnapshotQuery.getEventReplayStats:query",
2479 "ProjectionSnapshotQuery.getEventReplayStats:decodeRow",
2480 ),
2481 ),
2482 Effect.map(
2483 (row): ProjectionEventReplayStats => ({
2484 eventCount: row.eventCount,
2485 payloadBytes: row.payloadBytes,
2486 }),
2487 ),
2488 );
2489
24512490 const searchThreads: ProjectionSnapshotQueryShape["searchThreads"] = Effect.fn(
24522491 "ProjectionSnapshotQuery.searchThreads",
24532492 )(function* (input) {
......@@ -3109,6 +3148,7 @@ pending_approval_requests AS (
31093148 searchThreads,
31103149 getSnapshotSequence,
31113150 getCounts,
3151 getEventReplayStats,
31123152 getActiveProjectByWorkspaceRoot,
31133153 getProjectShellById,
31143154 getFirstActiveThreadIdByProjectId,
apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts+13
......@@ -37,6 +37,11 @@ export interface ProjectionSnapshotSequence {
3737 readonly snapshotSequence: number;
3838}
3939
40export interface ProjectionEventReplayStats {
41 readonly eventCount: number;
42 readonly payloadBytes: number;
43}
44
4045export interface ProjectionThreadCheckpointContext {
4146 readonly threadId: ThreadId;
4247 readonly projectId: ProjectId;
......@@ -128,6 +133,14 @@ export interface ProjectionSnapshotQueryShape {
128133 */
129134 readonly getCounts: () => Effect.Effect<ProjectionSnapshotCounts, ProjectionRepositoryError>;
130135
136 /**
137 * Measure a persisted event range without decoding its payload bodies.
138 */
139 readonly getEventReplayStats: (input: {
140 readonly fromSequenceExclusive: number;
141 readonly toSequenceInclusive: number;
142 }) => Effect.Effect<ProjectionEventReplayStats, ProjectionRepositoryError>;
143
131144 /**
132145 * Read the active project for an exact workspace root match.
133146 */
apps/server/src/project/ProjectSetupScriptRunner.test.ts+1
......@@ -32,6 +32,7 @@ const makeProjectionSnapshotQueryLayer = (project: OrchestrationProject) =>
3232 getArchivedShellSnapshot: () => Effect.die("unused"),
3333 getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 1 }),
3434 getCounts: () => Effect.die("unused"),
35 getEventReplayStats: () => Effect.die("unused"),
3536 getActiveProjectByWorkspaceRoot: (workspaceRoot) =>
3637 Effect.succeed(
3738 workspaceRoot === project.workspaceRoot ? Option.some(project) : Option.none(),
apps/server/src/provider/Layers/ProviderSessionReaper.test.ts+1
......@@ -210,6 +210,7 @@ describe("ProviderSessionReaper", () => {
210210 getSnapshotSequence: () =>
211211 Effect.succeed({ snapshotSequence: input.readModel.snapshotSequence }),
212212 getCounts: () => Effect.die("unused"),
213 getEventReplayStats: () => Effect.die("unused"),
213214 getActiveProjectByWorkspaceRoot: () => Effect.die("unused"),
214215 getProjectShellById: () => Effect.die("unused"),
215216 getFirstActiveThreadIdByProjectId: () => Effect.die("unused"),
apps/server/src/server.test.ts+69
......@@ -888,6 +888,11 @@ const buildAppUnderTest = (options?: {
888888 getThreadDetailById: () => Effect.succeed(Option.none()),
889889 getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
890890 getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
891 getEventReplayStats: ({ fromSequenceExclusive, toSequenceInclusive }) =>
892 Effect.succeed({
893 eventCount: Math.max(0, toSequenceInclusive - fromSequenceExclusive),
894 payloadBytes: 0,
895 }),
891896 getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
892897 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
893898 getThreadCheckpointContext: () => Effect.succeed(Option.none()),
......@@ -7366,6 +7371,70 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
73667371 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
73677372 );
73687373
7374 it.effect("subscriptions snapshot instead of decoding an oversized replay range", () =>
7375 Effect.gen(function* () {
7376 let readEventsCalls = 0;
7377 let replayStatsCalls = 0;
7378 const thread = makeDefaultOrchestrationReadModel().threads[0]!;
7379 const shell = makeDefaultOrchestrationThreadShell({ id: thread.id });
7380
7381 yield* buildAppUnderTest({
7382 layers: {
7383 orchestrationEngine: {
7384 latestSequence: Effect.succeed(5),
7385 readEvents: () =>
7386 Stream.sync(() => {
7387 readEventsCalls += 1;
7388 return {} as OrchestrationEvent;
7389 }),
7390 },
7391 projectionSnapshotQuery: {
7392 getEventReplayStats: () =>
7393 Effect.sync(() => {
7394 replayStatsCalls += 1;
7395 return { eventCount: 5, payloadBytes: 8 * 1024 * 1024 + 1 };
7396 }),
7397 getThreadDetailSnapshot: () =>
7398 Effect.succeed(Option.some({ snapshotSequence: 5, thread })),
7399 getShellSnapshot: () =>
7400 Effect.succeed({
7401 snapshotSequence: 5,
7402 projects: [],
7403 threads: [shell],
7404 updatedAt: "2026-01-01T00:00:00.000Z",
7405 }),
7406 },
7407 },
7408 });
7409
7410 const wsUrl = yield* getWsServerUrl("/ws");
7411 const threadItems = yield* Effect.scoped(
7412 withWsRpcClient(wsUrl, (client) =>
7413 client[ORCHESTRATION_WS_METHODS.subscribeThread]({
7414 threadId: thread.id,
7415 afterSequence: 0,
7416 requestCompletionMarker: true,
7417 }).pipe(Stream.take(2), Stream.runCollect),
7418 ),
7419 );
7420 const shellItems = yield* Effect.scoped(
7421 withWsRpcClient(wsUrl, (client) =>
7422 client[ORCHESTRATION_WS_METHODS.subscribeShell]({
7423 afterSequence: 0,
7424 requestCompletionMarker: true,
7425 }).pipe(Stream.take(2), Stream.runCollect),
7426 ),
7427 );
7428
7429 assert.equal(threadItems[0]?.kind, "snapshot");
7430 assert.equal(threadItems[1]?.kind, "synchronized");
7431 assert.equal(shellItems[0]?.kind, "snapshot");
7432 assert.equal(shellItems[1]?.kind, "synchronized");
7433 assert.equal(replayStatsCalls, 2);
7434 assert.equal(readEventsCalls, 0);
7435 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
7436 );
7437
73697438 it.effect("subscribeShell replaces a cursor ahead of the authoritative head", () =>
73707439 Effect.gen(function* () {
73717440 let readEventsCalls = 0;
apps/server/src/serverRuntimeStartup.test.ts+4
......@@ -83,6 +83,7 @@ it.effect("launchStartupHeartbeat does not block the caller while counts are loa
8383 getShellSnapshot: () => Effect.die("unused"),
8484 getArchivedShellSnapshot: () => Effect.die("unused"),
8585 getSnapshotSequence: () => Effect.die("unused"),
86 getEventReplayStats: () => Effect.die("unused"),
8687 getCounts: () =>
8788 Deferred.succeed(countsStarted, undefined).pipe(
8889 Effect.andThen(Deferred.await(releaseCounts)),
......@@ -150,6 +151,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets returns existing project and threa
150151 getArchivedShellSnapshot: () => Effect.die("unused"),
151152 getSnapshotSequence: () => Effect.die("unused"),
152153 getCounts: () => Effect.die("unused"),
154 getEventReplayStats: () => Effect.die("unused"),
153155 getActiveProjectByWorkspaceRoot: () =>
154156 Effect.succeed(
155157 Option.some({
......@@ -207,6 +209,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets creates a project and thread when
207209 getArchivedShellSnapshot: () => Effect.die("unused"),
208210 getSnapshotSequence: () => Effect.die("unused"),
209211 getCounts: () => Effect.die("unused"),
212 getEventReplayStats: () => Effect.die("unused"),
210213 getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
211214 getProjectShellById: () => Effect.die("unused"),
212215 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
......@@ -258,6 +261,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets preserves typed UUID generation fa
258261 getArchivedShellSnapshot: () => Effect.die("unused"),
259262 getSnapshotSequence: () => Effect.die("unused"),
260263 getCounts: () => Effect.die("unused"),
264 getEventReplayStats: () => Effect.die("unused"),
261265 getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
262266 getProjectShellById: () => Effect.die("unused"),
263267 getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
apps/server/src/ws.ts+50-2
......@@ -335,6 +335,10 @@ const SHELL_RESUME_MAX_GAP = 1_000;
335335// hundreds of thousands of events behind have OOM-killed servers on large
336336// databases. Past this gap the client is reset with a fresh thread snapshot.
337337const THREAD_RESUME_MAX_GAP = 1_000;
338// Row count alone does not bound replay memory: a few events with large tool
339// payloads can decode to gigabytes. Before replaying, sum the serialized
340// payload bytes of the range in SQL and reset with a snapshot past this budget.
341const ORCHESTRATION_REPLAY_PAYLOAD_BUDGET_BYTES = 8 * 1024 * 1024;
338342
339343function toAuthAccessStreamEvent(
340344 change: PairingGrantStore.BootstrapCredentialChange | SessionStore.SessionCredentialChange,
......@@ -514,6 +518,42 @@ const makeWsRpcLayer = (
514518 const startup = yield* ServerRuntimeStartup.ServerRuntimeStartup;
515519 const workspaceEntries = yield* WorkspaceEntries.WorkspaceEntries;
516520 const workspaceFileSystem = yield* WorkspaceFileSystem.WorkspaceFileSystem;
521 const canReplayPersistedRange = Effect.fnUntraced(function* (
522 afterSequence: number,
523 headSequence: number,
524 maxGap: number,
525 ) {
526 const replayGap = headSequence - afterSequence;
527 if (replayGap < 0 || replayGap > maxGap) {
528 return false;
529 }
530 const stats = yield* projectionSnapshotQuery
531 .getEventReplayStats({
532 fromSequenceExclusive: afterSequence,
533 toSequenceInclusive: headSequence,
534 })
535 .pipe(
536 Effect.mapError(
537 (cause) =>
538 new OrchestrationGetSnapshotError({
539 message: "Failed to measure orchestration replay range",
540 cause,
541 }),
542 ),
543 );
544 if (stats.payloadBytes > ORCHESTRATION_REPLAY_PAYLOAD_BUDGET_BYTES) {
545 yield* Effect.logDebug("orchestration replay replaced by snapshot", {
546 afterSequence,
547 headSequence,
548 replayGap,
549 eventCount: stats.eventCount,
550 payloadBytes: stats.payloadBytes,
551 payloadBudgetBytes: ORCHESTRATION_REPLAY_PAYLOAD_BUDGET_BYTES,
552 });
553 return false;
554 }
555 return true;
556 });
517557 const projectSetupScriptRunner = yield* ProjectSetupScriptRunner.ProjectSetupScriptRunner;
518558 const serverEnvironment = yield* ServerEnvironment.ServerEnvironment;
519559 const backgroundPolicy = yield* BackgroundPolicy.BackgroundPolicy;
......@@ -1419,7 +1459,13 @@ const makeWsRpcLayer = (
14191459 // is also invalid, so reset it with a snapshot. Send the snapshot
14201460 // followed by the buffered live tail, exactly as the
14211461 // no-afterSequence path does.
1422 if (replayGap < 0 || replayGap > SHELL_RESUME_MAX_GAP) {
1462 if (
1463 !(yield* canReplayPersistedRange(
1464 afterSequence,
1465 headSequence,
1466 SHELL_RESUME_MAX_GAP,
1467 ))
1468 ) {
14231469 const snapshot = yield* loadSnapshot;
14241470 return Stream.concat(
14251471 Stream.make({ kind: "snapshot" as const, snapshot }),
......@@ -1521,7 +1567,9 @@ const makeWsRpcLayer = (
15211567 const afterSequence = input.afterSequence;
15221568 const headSequence = yield* orchestrationEngine.latestSequence;
15231569 const replayGap = headSequence - afterSequence;
1524 if (replayGap >= 0 && replayGap <= THREAD_RESUME_MAX_GAP) {
1570 if (
1571 yield* canReplayPersistedRange(afterSequence, headSequence, THREAD_RESUME_MAX_GAP)
1572 ) {
15251573 const catchUpStream = orchestrationEngine
15261574 .readEvents(afterSequence, replayGap)
15271575 .pipe(