authorgravatar for rusirusadathana@gmail.comRusiru Sadathana <rusirusadathana@gmail.com> 2026-07-20 18:14:58+05:30
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-07-20 14:44:58+02:00
logdb4b2d8a0f23b00f19774dda33e0b02627272c90
tree659194c5fadbef8a97d58e9d51559b7155a134a1
parent2c199aacac3244adc93c5f4a9c838683e93946dd
signature Signed by PGP key B5690EEEBB952194

perf(orchestration): speed up new-chat propagation and offline catch-up (#4177)

Co-authored-by: codex <codex@users.noreply.github.com> Co-authored-by: Julius Marminge <julius0216@outlook.com>

7 files changed, 749 insertions(+), 96 deletions(-)

apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts+2
......@@ -218,6 +218,7 @@ describe("OrchestrationEngine", () => {
218218 const runtime = ManagedRuntime.make(layer);
219219
220220 const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService));
221 expect(await runtime.runPromise(engine.latestSequence)).toBe(7);
221222 const result = await runtime.runPromise(
222223 engine.dispatch({
223224 type: "thread.meta.update",
......@@ -228,6 +229,7 @@ describe("OrchestrationEngine", () => {
228229 );
229230
230231 expect(result.sequence).toBe(8);
232 expect(await runtime.runPromise(engine.latestSequence)).toBe(8);
231233 expect(fullSnapshotReadCount).toBe(0);
232234
233235 await runtime.dispose();
apps/server/src/orchestration/Layers/OrchestrationEngine.ts+5
......@@ -329,6 +329,11 @@ const makeOrchestrationEngine = Effect.gen(function* () {
329329 get streamDomainEvents(): OrchestrationEngineShape["streamDomainEvents"] {
330330 return Stream.fromPubSub(eventPubSub);
331331 },
332 // The command read model's snapshotSequence tracks the latest committed
333 // event sequence (updated on the worker fiber). A plain property read is a
334 // consistent, committed value — reassignment of `commandReadModel` is
335 // atomic on the single-threaded event loop.
336 latestSequence: Effect.sync(() => commandReadModel.snapshotSequence),
332337 } satisfies OrchestrationEngineShape;
333338});
334339
apps/server/src/orchestration/Services/OrchestrationEngine.ts+7
......@@ -56,6 +56,13 @@ export interface OrchestrationEngineShape {
5656 * This is a hot runtime stream (new events only), not a historical replay.
5757 */
5858 readonly streamDomainEvents: Stream.Stream<OrchestrationEvent>;
59
60 /**
61 * The latest sequence reflected in the engine's authoritative command read
62 * model (0 if none). Used to gauge how far behind a resuming client is before
63 * choosing between an incremental replay and a fresh projected snapshot.
64 */
65 readonly latestSequence: Effect.Effect<number, never, never>;
5966}
6067
6168/**
apps/server/src/relay/AgentAwarenessRelay.test.ts+2
......@@ -470,6 +470,7 @@ describe.sequential("signRelayAgentActivityPublishProof", () => {
470470 readEvents: () => Stream.empty,
471471 dispatch: () => Effect.succeed({ sequence: 1 }),
472472 streamDomainEvents: Stream.fromQueue(events),
473 latestSequence: Effect.succeed(0),
473474 } satisfies OrchestrationEngineShape;
474475
475476 const snapshotQuery = {
......@@ -659,6 +660,7 @@ describe.sequential("signRelayAgentActivityPublishProof", () => {
659660 readEvents: () => Stream.empty,
660661 dispatch: () => Effect.succeed({ sequence: 1 }),
661662 streamDomainEvents: Stream.fromQueue(events),
663 latestSequence: Effect.succeed(0),
662664 } satisfies OrchestrationEngineShape),
663665 Layer.succeed(ProjectionSnapshotQuery, {
664666 getShellSnapshot: () =>
apps/server/src/server.test.ts+469-1
......@@ -46,6 +46,7 @@ import * as DateTime from "effect/DateTime";
4646import * as Duration from "effect/Duration";
4747import * as Effect from "effect/Effect";
4848import * as FileSystem from "effect/FileSystem";
49import * as Fiber from "effect/Fiber";
4950import * as Layer from "effect/Layer";
5051import * as ManagedRuntime from "effect/ManagedRuntime";
5152import * as Option from "effect/Option";
......@@ -677,6 +678,7 @@ const buildAppUnderTest = (options?: {
677678 readEvents: () => Stream.empty,
678679 dispatch: () => Effect.succeed({ sequence: 0 }),
679680 streamDomainEvents: Stream.empty,
681 latestSequence: Effect.succeed(0),
680682 ...options?.layers?.orchestrationEngine,
681683 }),
682684 ),
......@@ -5688,7 +5690,6 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
56885690 projectionSnapshotQuery: {
56895691 getShellSnapshot: () =>
56905692 Effect.gen(function* () {
5691 yield* Effect.sleep("25 millis");
56925693 yield* PubSub.publish(liveEvents, deletedEvent);
56935694 return {
56945695 snapshotSequence: 1,
......@@ -5774,6 +5775,473 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
57745775 }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive),
57755776 );
57765777
5778 it.effect("subscribeShell sends a fresh snapshot instead of replaying a large gap", () =>
5779 Effect.gen(function* () {
5780 let readEventsCalls = 0;
5781 const snapshotThreadId = ThreadId.make("thread-from-snapshot");
5782 const now = "2026-01-01T00:00:00.000Z";
5783
5784 yield* buildAppUnderTest({
5785 layers: {
5786 orchestrationEngine: {
5787 // Head is far ahead of the client's afterSequence (gap > 1000).
5788 latestSequence: Effect.succeed(100_000),
5789 readEvents: () =>
5790 Stream.sync(() => {
5791 readEventsCalls += 1;
5792 return {
5793 sequence: 1,
5794 eventId: EventId.make("event-should-not-be-read"),
5795 aggregateKind: "thread",
5796 aggregateId: snapshotThreadId,
5797 occurredAt: now,
5798 commandId: null,
5799 causationEventId: null,
5800 correlationId: null,
5801 metadata: {},
5802 type: "thread.created",
5803 payload: {} as never,
5804 } satisfies OrchestrationEvent;
5805 }),
5806 },
5807 projectionSnapshotQuery: {
5808 getShellSnapshot: () =>
5809 Effect.succeed({
5810 snapshotSequence: 100_000,
5811 projects: [],
5812 threads: [makeDefaultOrchestrationThreadShell({ id: snapshotThreadId })],
5813 updatedAt: now,
5814 }),
5815 },
5816 },
5817 });
5818
5819 const wsUrl = yield* getWsServerUrl("/ws");
5820 const items = yield* Effect.scoped(
5821 withWsRpcClient(wsUrl, (client) =>
5822 client[ORCHESTRATION_WS_METHODS.subscribeShell]({
5823 afterSequence: 5,
5824 requestCompletionMarker: true,
5825 }).pipe(Stream.take(2), Stream.runCollect),
5826 ),
5827 );
5828
5829 const [first, second] = Array.from(items);
5830 // Large gap => fresh snapshot, and the unbounded replay is never started.
5831 assert.equal(first?.kind, "snapshot");
5832 if (first?.kind === "snapshot") {
5833 assert.equal(first.snapshot.threads[0]?.id, snapshotThreadId);
5834 }
5835 assert.equal(second?.kind, "synchronized");
5836 assert.equal(readEventsCalls, 0);
5837 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
5838 );
5839
5840 it.effect("subscribeShell replaces a cursor ahead of the authoritative head", () =>
5841 Effect.gen(function* () {
5842 let readEventsCalls = 0;
5843
5844 yield* buildAppUnderTest({
5845 layers: {
5846 orchestrationEngine: {
5847 latestSequence: Effect.succeed(5),
5848 readEvents: () =>
5849 Stream.sync(() => {
5850 readEventsCalls += 1;
5851 return {} as OrchestrationEvent;
5852 }),
5853 },
5854 projectionSnapshotQuery: {
5855 getShellSnapshot: () =>
5856 Effect.succeed({
5857 snapshotSequence: 5,
5858 projects: [],
5859 threads: [],
5860 updatedAt: "2026-01-01T00:00:00.000Z",
5861 }),
5862 },
5863 },
5864 });
5865
5866 const wsUrl = yield* getWsServerUrl("/ws");
5867 const first = yield* Effect.scoped(
5868 withWsRpcClient(wsUrl, (client) =>
5869 client[ORCHESTRATION_WS_METHODS.subscribeShell]({ afterSequence: 10 }).pipe(
5870 Stream.runHead,
5871 ),
5872 ),
5873 );
5874
5875 assert.equal(Option.getOrThrow(first).kind, "snapshot");
5876 assert.equal(readEventsCalls, 0);
5877 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
5878 );
5879
5880 it.effect("subscribeShell coalesces a per-thread burst without stalling other threads", () =>
5881 Effect.gen(function* () {
5882 const busyThreadId = ThreadId.make("thread-busy");
5883 const newThreadId = ThreadId.make("thread-new");
5884 const now = "2026-01-01T00:00:00.000Z";
5885 const shellFetches: Array<string> = [];
5886 let replayLimit: number | undefined;
5887
5888 const messageEvent = (sequence: number): OrchestrationEvent =>
5889 ({
5890 sequence,
5891 eventId: EventId.make(`event-${sequence}`),
5892 aggregateKind: "thread",
5893 aggregateId: busyThreadId,
5894 occurredAt: now,
5895 commandId: null,
5896 causationEventId: null,
5897 correlationId: null,
5898 metadata: {},
5899 type: "thread.message-sent",
5900 payload: {} as never,
5901 }) satisfies OrchestrationEvent;
5902
5903 const createdEvent: OrchestrationEvent = {
5904 sequence: 50,
5905 eventId: EventId.make("event-created"),
5906 aggregateKind: "thread",
5907 aggregateId: newThreadId,
5908 occurredAt: now,
5909 commandId: null,
5910 causationEventId: null,
5911 correlationId: null,
5912 metadata: {},
5913 type: "thread.created",
5914 payload: {} as never,
5915 };
5916
5917 yield* buildAppUnderTest({
5918 layers: {
5919 orchestrationEngine: {
5920 latestSequence: Effect.succeed(50),
5921 // A burst of message-sent deltas for the busy thread, plus one
5922 // thread.created for a different thread, all within one batch.
5923 readEvents: (_afterSequence, limit) => {
5924 replayLimit = limit;
5925 return Stream.fromIterable([
5926 ...Array.from({ length: 20 }, (_unused, index) => messageEvent(index + 1)),
5927 createdEvent,
5928 ]);
5929 },
5930 },
5931 projectionSnapshotQuery: {
5932 getThreadShellById: (threadId) =>
5933 Effect.sync(() => {
5934 shellFetches.push(threadId);
5935 return Option.some(makeDefaultOrchestrationThreadShell({ id: threadId }));
5936 }),
5937 },
5938 },
5939 });
5940
5941 const wsUrl = yield* getWsServerUrl("/ws");
5942 const items = yield* Effect.scoped(
5943 withWsRpcClient(wsUrl, (client) =>
5944 client[ORCHESTRATION_WS_METHODS.subscribeShell]({
5945 afterSequence: 0,
5946 requestCompletionMarker: true,
5947 }).pipe(Stream.take(3), Stream.runCollect),
5948 ),
5949 );
5950
5951 const collected = Array.from(items);
5952 const upsertedIds = collected.flatMap((item) =>
5953 item.kind === "thread-upserted" ? [item.thread.id] : [],
5954 );
5955 // Both threads surface, and the busy thread's 20-event burst collapses to
5956 // a single shell refetch (not 20). The new thread is not stuck behind it.
5957 assert.include(upsertedIds, busyThreadId);
5958 assert.include(upsertedIds, newThreadId);
5959 assert.equal(collected[2]?.kind, "synchronized");
5960 assert.equal(shellFetches.filter((id) => id === busyThreadId).length, 1);
5961 assert.equal(replayLimit, 50);
5962 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
5963 );
5964
5965 it.effect("subscribeShell coalesces live bursts after the synchronization marker", () =>
5966 Effect.gen(function* () {
5967 const busyThreadId = ThreadId.make("thread-live-busy");
5968 const newThreadId = ThreadId.make("thread-live-new");
5969 const now = "2026-01-01T00:00:00.000Z";
5970 const liveEvents = yield* PubSub.unbounded<OrchestrationEvent>();
5971 const synchronized = yield* Deferred.make<void>();
5972 const shellFetches: Array<string> = [];
5973 const observedLiveThreadIds = new Set<string>();
5974
5975 const messageEvent = (sequence: number): OrchestrationEvent =>
5976 ({
5977 sequence,
5978 eventId: EventId.make(`event-live-${sequence}`),
5979 aggregateKind: "thread",
5980 aggregateId: busyThreadId,
5981 occurredAt: now,
5982 commandId: null,
5983 causationEventId: null,
5984 correlationId: null,
5985 metadata: {},
5986 type: "thread.message-sent",
5987 payload: {} as never,
5988 }) satisfies OrchestrationEvent;
5989
5990 const createdEvent: OrchestrationEvent = {
5991 sequence: 50,
5992 eventId: EventId.make("event-live-created"),
5993 aggregateKind: "thread",
5994 aggregateId: newThreadId,
5995 occurredAt: now,
5996 commandId: null,
5997 causationEventId: null,
5998 correlationId: null,
5999 metadata: {},
6000 type: "thread.created",
6001 payload: {} as never,
6002 };
6003
6004 yield* buildAppUnderTest({
6005 layers: {
6006 orchestrationEngine: {
6007 streamDomainEvents: Stream.fromPubSub(liveEvents),
6008 },
6009 projectionSnapshotQuery: {
6010 getThreadShellById: (threadId) =>
6011 Effect.sync(() => {
6012 shellFetches.push(threadId);
6013 return Option.some(makeDefaultOrchestrationThreadShell({ id: threadId }));
6014 }),
6015 },
6016 },
6017 });
6018
6019 const wsUrl = yield* getWsServerUrl("/ws");
6020 const items = yield* Effect.scoped(
6021 Effect.gen(function* () {
6022 const itemsFiber = yield* withWsRpcClient(wsUrl, (client) =>
6023 client[ORCHESTRATION_WS_METHODS.subscribeShell]({
6024 requestCompletionMarker: true,
6025 }).pipe(
6026 Stream.tap((item) =>
6027 item.kind === "synchronized"
6028 ? Deferred.succeed(synchronized, undefined).pipe(Effect.ignore)
6029 : Effect.void,
6030 ),
6031 Stream.takeUntil((item) => {
6032 if (item.kind === "thread-upserted") {
6033 observedLiveThreadIds.add(item.thread.id);
6034 }
6035 return (
6036 observedLiveThreadIds.has(busyThreadId) && observedLiveThreadIds.has(newThreadId)
6037 );
6038 }),
6039 Stream.runCollect,
6040 ),
6041 ).pipe(Effect.forkScoped);
6042
6043 yield* Deferred.await(synchronized);
6044 for (const event of [
6045 ...Array.from({ length: 20 }, (_unused, index) => messageEvent(index + 1)),
6046 createdEvent,
6047 ]) {
6048 yield* PubSub.publish(liveEvents, event);
6049 }
6050
6051 return yield* Fiber.join(itemsFiber);
6052 }),
6053 ).pipe(Effect.timeout("2 seconds"));
6054
6055 assert.equal(items[0]?.kind, "snapshot");
6056 assert.equal(items[1]?.kind, "synchronized");
6057 const liveUpsertedIds = Array.from(items)
6058 .slice(2)
6059 .flatMap((item) => (item.kind === "thread-upserted" ? [item.thread.id] : []));
6060 assert.include(liveUpsertedIds, busyThreadId);
6061 assert.include(liveUpsertedIds, newThreadId);
6062 assert.isBelow(shellFetches.filter((id) => id === busyThreadId).length, 20);
6063 }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive),
6064 );
6065
6066 it.effect("subscribeShell coalescing still emits a removal for a deleted thread", () =>
6067 Effect.gen(function* () {
6068 const goneThreadId = ThreadId.make("thread-gone");
6069 const now = "2026-01-01T00:00:00.000Z";
6070
6071 const makeThreadEvent = (
6072 sequence: number,
6073 type: "thread.deleted" | "thread.message-sent",
6074 ): OrchestrationEvent =>
6075 ({
6076 sequence,
6077 eventId: EventId.make(`event-${sequence}`),
6078 aggregateKind: "thread",
6079 aggregateId: goneThreadId,
6080 occurredAt: now,
6081 commandId: null,
6082 causationEventId: null,
6083 correlationId: null,
6084 metadata: {},
6085 type,
6086 payload: type === "thread.deleted" ? { threadId: goneThreadId, deletedAt: now } : {},
6087 }) as OrchestrationEvent;
6088
6089 yield* buildAppUnderTest({
6090 layers: {
6091 orchestrationEngine: {
6092 latestSequence: Effect.succeed(2),
6093 // A thread.deleted followed, within the same coalescing window, by a
6094 // later refetchable event for the same thread. The later event wins
6095 // coalescing; its shell refetch returns none (the row is gone), which
6096 // must still surface a removal rather than be swallowed.
6097 readEvents: () =>
6098 Stream.fromIterable([
6099 makeThreadEvent(1, "thread.deleted"),
6100 makeThreadEvent(2, "thread.message-sent"),
6101 ]),
6102 },
6103 projectionSnapshotQuery: {
6104 getThreadShellById: () => Effect.succeed(Option.none()),
6105 },
6106 },
6107 });
6108
6109 const wsUrl = yield* getWsServerUrl("/ws");
6110 const items = yield* Effect.scoped(
6111 withWsRpcClient(wsUrl, (client) =>
6112 client[ORCHESTRATION_WS_METHODS.subscribeShell]({ afterSequence: 0 }).pipe(
6113 Stream.take(1),
6114 Stream.runCollect,
6115 ),
6116 ),
6117 );
6118
6119 const [first] = Array.from(items);
6120 assert.equal(first?.kind, "thread-removed");
6121 assert.equal(first?.kind === "thread-removed" ? first.threadId : null, goneThreadId);
6122 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
6123 );
6124
6125 it.effect("subscribeShell retries a transient shell projection refetch failure", () =>
6126 Effect.gen(function* () {
6127 const threadId = ThreadId.make("thread-transient-refetch");
6128 const now = "2026-01-01T00:00:00.000Z";
6129 let attempts = 0;
6130
6131 const event: OrchestrationEvent = {
6132 sequence: 1,
6133 eventId: EventId.make("event-transient-refetch"),
6134 aggregateKind: "thread",
6135 aggregateId: threadId,
6136 occurredAt: now,
6137 commandId: null,
6138 causationEventId: null,
6139 correlationId: null,
6140 metadata: {},
6141 type: "thread.message-sent",
6142 payload: {} as never,
6143 };
6144
6145 yield* buildAppUnderTest({
6146 layers: {
6147 orchestrationEngine: {
6148 latestSequence: Effect.succeed(1),
6149 readEvents: () => Stream.make(event),
6150 },
6151 projectionSnapshotQuery: {
6152 getThreadShellById: () =>
6153 Effect.suspend(() => {
6154 attempts += 1;
6155 return attempts === 1
6156 ? Effect.fail(
6157 new PersistenceSqlError({
6158 operation: "test.shell-refetch",
6159 detail: "transient failure",
6160 }),
6161 )
6162 : Effect.succeed(
6163 Option.some(makeDefaultOrchestrationThreadShell({ id: threadId })),
6164 );
6165 }),
6166 },
6167 },
6168 });
6169
6170 const wsUrl = yield* getWsServerUrl("/ws");
6171 const items = yield* Effect.scoped(
6172 withWsRpcClient(wsUrl, (client) =>
6173 client[ORCHESTRATION_WS_METHODS.subscribeShell]({ afterSequence: 0 }).pipe(
6174 Stream.take(1),
6175 Stream.runCollect,
6176 ),
6177 ),
6178 );
6179
6180 const [first] = Array.from(items);
6181 assert.equal(first?.kind, "thread-upserted");
6182 assert.equal(first?.kind === "thread-upserted" ? first.thread.id : null, threadId);
6183 assert.equal(attempts, 2);
6184 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
6185 );
6186
6187 it.effect("subscribeShell coalescing still removes a project after a trailing update", () =>
6188 Effect.gen(function* () {
6189 const projectId = ProjectId.make("project-gone");
6190 const now = "2026-01-01T00:00:00.000Z";
6191
6192 const makeProjectEvent = (
6193 sequence: number,
6194 type: "project.deleted" | "project.meta-updated",
6195 ): OrchestrationEvent =>
6196 ({
6197 sequence,
6198 eventId: EventId.make(`event-project-${sequence}`),
6199 aggregateKind: "project",
6200 aggregateId: projectId,
6201 occurredAt: now,
6202 commandId: null,
6203 causationEventId: null,
6204 correlationId: null,
6205 metadata: {},
6206 type,
6207 payload:
6208 type === "project.deleted"
6209 ? { projectId, deletedAt: now }
6210 : { projectId, title: "Still deleted", updatedAt: now },
6211 }) as OrchestrationEvent;
6212
6213 yield* buildAppUnderTest({
6214 layers: {
6215 orchestrationEngine: {
6216 latestSequence: Effect.succeed(2),
6217 readEvents: () =>
6218 Stream.fromIterable([
6219 makeProjectEvent(1, "project.deleted"),
6220 makeProjectEvent(2, "project.meta-updated"),
6221 ]),
6222 },
6223 projectionSnapshotQuery: {
6224 getProjectShellById: () => Effect.succeed(Option.none()),
6225 },
6226 },
6227 });
6228
6229 const wsUrl = yield* getWsServerUrl("/ws");
6230 const items = yield* Effect.scoped(
6231 withWsRpcClient(wsUrl, (client) =>
6232 client[ORCHESTRATION_WS_METHODS.subscribeShell]({ afterSequence: 0 }).pipe(
6233 Stream.take(1),
6234 Stream.runCollect,
6235 ),
6236 ),
6237 );
6238
6239 const [first] = Array.from(items);
6240 assert.equal(first?.kind, "project-removed");
6241 assert.equal(first?.kind === "project-removed" ? first.projectId : null, projectId);
6242 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
6243 );
6244
57776245 it.effect("enriches replayed project events with repository identity metadata", () =>
57786246 Effect.gen(function* () {
57796247 const repositoryIdentity = {
apps/server/src/serverRuntimeStartup.test.ts+3
......@@ -168,6 +168,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets returns existing project and threa
168168 Effect.as({ sequence: 1 }),
169169 ),
170170 streamDomainEvents: Stream.empty,
171 latestSequence: Effect.succeed(0),
171172 } satisfies OrchestrationEngine.OrchestrationEngineService["Service"]),
172173 Effect.provide(NodeServices.layer),
173174 );
......@@ -211,6 +212,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets creates a project and thread when
211212 Effect.as({ sequence: 1 }),
212213 ),
213214 streamDomainEvents: Stream.empty,
215 latestSequence: Effect.succeed(0),
214216 } satisfies OrchestrationEngine.OrchestrationEngineService["Service"]),
215217 Effect.provide(NodeServices.layer),
216218 );
......@@ -260,6 +262,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets preserves typed UUID generation fa
260262 Effect.as({ sequence: 1 }),
261263 ),
262264 streamDomainEvents: Stream.empty,
265 latestSequence: Effect.succeed(0),
263266 } satisfies OrchestrationEngine.OrchestrationEngineService["Service"]),
264267 Effect.provideService(Crypto.Crypto, {
265268 ...crypto,
apps/server/src/ws.ts+261-95
......@@ -36,6 +36,7 @@ import {
3636 OrchestrationGetSnapshotError,
3737 OrchestrationGetTurnDiffError,
3838 ORCHESTRATION_WS_METHODS,
39 type ProjectId,
3940 type ProjectEntriesFailure,
4041 type ProjectFileFailure,
4142 type ProjectFileOperation,
......@@ -276,6 +277,13 @@ function isThreadDetailEvent(event: OrchestrationEvent): event is Extract<
276277
277278const PROVIDER_STATUS_DEBOUNCE_MS = 200;
278279
280// When a resuming client's cursor is more than this many events behind the
281// current head, skip the per-event catch-up replay and send a fresh shell
282// snapshot instead. Replaying each intervening event costs a shell refetch;
283// past this gap a single O(active-threads) snapshot is cheaper and bounded.
284// Matches the event store's default page size (DEFAULT_READ_FROM_SEQUENCE_LIMIT).
285const SHELL_RESUME_MAX_GAP = 1_000;
286
279287const RPC_REQUIRED_SCOPE = new Map<string, AuthEnvironmentScope>([
280288 [ORCHESTRATION_WS_METHODS.dispatchCommand, AuthOrchestrationOperateScope],
281289 [ORCHESTRATION_WS_METHODS.getTurnDiff, AuthOrchestrationReadScope],
......@@ -621,16 +629,7 @@ const makeWsRpcLayer = (
621629 switch (event.type) {
622630 case "project.created":
623631 case "project.meta-updated":
624 return projectionSnapshotQuery.getProjectShellById(event.payload.projectId).pipe(
625 Effect.map((project) =>
626 Option.map(project, (nextProject) => ({
627 kind: "project-upserted" as const,
628 sequence: event.sequence,
629 project: nextProject,
630 })),
631 ),
632 Effect.orElseSucceed(() => Option.none()),
633 );
632 return projectUpsertOrRemove(event.payload.projectId, event.sequence);
634633 case "project.deleted":
635634 return Effect.succeed(
636635 Option.some({
......@@ -649,35 +648,192 @@ const makeWsRpcLayer = (
649648 }),
650649 );
651650 case "thread.unarchived":
652 return projectionSnapshotQuery.getThreadShellById(event.payload.threadId).pipe(
653 Effect.map((thread) =>
654 Option.map(thread, (nextThread) => ({
655 kind: "thread-upserted" as const,
656 sequence: event.sequence,
657 thread: nextThread,
658 })),
659 ),
660 Effect.orElseSucceed(() => Option.none()),
661 );
651 return threadUpsertOrRemove(event.payload.threadId, event.sequence);
662652 default:
663653 if (event.aggregateKind !== "thread") {
664654 return Effect.succeed(Option.none());
665655 }
666 return projectionSnapshotQuery
667 .getThreadShellById(ThreadId.make(event.aggregateId))
668 .pipe(
669 Effect.map((thread) =>
670 Option.map(thread, (nextThread) => ({
671 kind: "thread-upserted" as const,
672 sequence: event.sequence,
673 thread: nextThread,
674 })),
675 ),
676 Effect.orElseSucceed(() => Option.none()),
677 );
656 return threadUpsertOrRemove(ThreadId.make(event.aggregateId), event.sequence);
678657 }
679658 };
680659
660 // Coalescing makes each projection read represent every event for that
661 // aggregate in the current window. Retry a typed persistence failure once
662 // so a brief read failure cannot strand the shell at its previous state.
663 // If both attempts fail, log and drop the stream item; treating an error as
664 // a missing row would incorrectly remove a still-active aggregate.
665 const retryShellProjectionRead = <A, E>(
666 aggregateKind: "project" | "thread",
667 aggregateId: string,
668 read: Effect.Effect<A, E>,
669 ): Effect.Effect<Option.Option<A>, never, never> =>
670 read.pipe(
671 Effect.retry({ times: 1 }),
672 Effect.map(Option.some),
673 Effect.tapError((error) =>
674 Effect.logWarning("orchestration shell projection refetch failed", {
675 aggregateKind,
676 aggregateId,
677 error,
678 }),
679 ),
680 Effect.orElseSucceed(() => Option.none()),
681 );
682
683 const projectUpsertOrRemove = (
684 projectId: ProjectId,
685 sequence: number,
686 ): Effect.Effect<Option.Option<OrchestrationShellStreamEvent>, never, never> =>
687 retryShellProjectionRead(
688 "project",
689 projectId,
690 projectionSnapshotQuery.getProjectShellById(projectId),
691 ).pipe(
692 Effect.map(
693 Option.flatMap((project) =>
694 Option.match(project, {
695 onNone: () =>
696 Option.some<OrchestrationShellStreamEvent>({
697 kind: "project-removed" as const,
698 sequence,
699 projectId,
700 }),
701 onSome: (nextProject) =>
702 Option.some<OrchestrationShellStreamEvent>({
703 kind: "project-upserted" as const,
704 sequence,
705 project: nextProject,
706 }),
707 }),
708 ),
709 ),
710 );
711
712 // Refetch a thread's shell and emit an upsert if it is still active, or a
713 // `thread-removed` if the projection has no active row for it. Emitting a
714 // removal on a `none` (rather than dropping the event) is what keeps
715 // coalescing correct: when a burst collapses a `thread.deleted`/`archived`
716 // into a later refetchable event for the same thread, the refetch returns
717 // `none` for the now-inactive row and this still tells the sidebar to drop
718 // it. A `thread-removed` the client does not have is a harmless no-op. The
719 // projection commits in the same transaction before the event publishes,
720 // so a `none` reliably means the thread is deleted or archived, not
721 // not-yet-persisted.
722 const threadUpsertOrRemove = (
723 threadId: ThreadId,
724 sequence: number,
725 ): Effect.Effect<Option.Option<OrchestrationShellStreamEvent>, never, never> =>
726 retryShellProjectionRead(
727 "thread",
728 threadId,
729 projectionSnapshotQuery.getThreadShellById(threadId),
730 ).pipe(
731 Effect.map(
732 Option.flatMap((thread) =>
733 Option.match(thread, {
734 onNone: () =>
735 Option.some<OrchestrationShellStreamEvent>({
736 kind: "thread-removed" as const,
737 sequence,
738 threadId,
739 }),
740 onSome: (nextThread) =>
741 Option.some<OrchestrationShellStreamEvent>({
742 kind: "thread-upserted" as const,
743 sequence,
744 thread: nextThread,
745 }),
746 }),
747 ),
748 ),
749 );
750
751 // Turn a batch of domain events into shell stream items, coalescing by
752 // aggregate first. `toShellStreamEvent` re-reads the *current* projected
753 // shell for an aggregate, so within a batch only the latest event per
754 // aggregate matters: a burst of streaming `thread.message-sent` deltas for
755 // one thread collapses into a single shell refetch, and an unrelated
756 // `thread.created` in the same batch is never stuck behind those DB reads.
757 //
758 // Input events arrive in ascending sequence; we keep the last (highest
759 // sequence) event per aggregate, then re-sort ascending before emitting so
760 // the client — which applies shell items strictly by increasing sequence
761 // and drops any `sequence <= snapshotSequence` — never skips a coalesced
762 // item. The refetch runs with bounded concurrency (order-preserving).
763 const SHELL_REFETCH_CONCURRENCY = 8;
764 const coalesceShellEvents = (
765 events: ReadonlyArray<OrchestrationEvent>,
766 ): Effect.Effect<ReadonlyArray<OrchestrationShellStreamEvent>, never, never> =>
767 Effect.gen(function* () {
768 if (events.length === 0) {
769 return [];
770 }
771 const latestByAggregate = new Map<string, OrchestrationEvent>();
772 for (const event of events) {
773 latestByAggregate.set(`${event.aggregateKind}:${event.aggregateId}`, event);
774 }
775 const survivors = Array.from(latestByAggregate.values()).sort(
776 (left, right) => left.sequence - right.sequence,
777 );
778 const shellEvents = yield* Effect.forEach(survivors, toShellStreamEvent, {
779 concurrency: SHELL_REFETCH_CONCURRENCY,
780 });
781 return shellEvents.flatMap((option) => (Option.isSome(option) ? [option.value] : []));
782 });
783
784 // Small time/size window over which to coalesce shell events. The window
785 // bounds the worst-case added latency for a brand-new thread to appear in
786 // the sidebar (imperceptible), while collapsing high-frequency streaming
787 // traffic so it can't serialize the shell stream behind per-event DB reads.
788 const SHELL_COALESCE_WINDOW = Duration.millis(50);
789 const SHELL_COALESCE_MAX_CHUNK = 512;
790 const coalesceShellStream = <E, R>(
791 stream: Stream.Stream<OrchestrationEvent, E, R>,
792 ): Stream.Stream<OrchestrationShellStreamEvent, E, R> =>
793 stream.pipe(
794 Stream.groupedWithin(SHELL_COALESCE_MAX_CHUNK, SHELL_COALESCE_WINDOW),
795 Stream.mapEffect(coalesceShellEvents),
796 Stream.flatMap((items) => Stream.fromIterable(items)),
797 );
798
799 type ShellLiveInput =
800 | { readonly kind: "event"; readonly event: OrchestrationEvent }
801 | { readonly kind: "synchronized" };
802
803 // A completion marker is queued alongside raw live events so it cannot
804 // overtake an event still waiting in the coalescing window. Split each
805 // batch at markers and coalesce only the event segments on either side.
806 const coalesceShellLiveInputs = (
807 inputs: ReadonlyArray<ShellLiveInput>,
808 ): Effect.Effect<ReadonlyArray<OrchestrationShellStreamItem>, never, never> =>
809 Effect.gen(function* () {
810 const output: Array<OrchestrationShellStreamItem> = [];
811 let pendingEvents: Array<OrchestrationEvent> = [];
812
813 for (const input of inputs) {
814 if (input.kind === "event") {
815 pendingEvents.push(input.event);
816 continue;
817 }
818
819 output.push(...(yield* coalesceShellEvents(pendingEvents)));
820 pendingEvents = [];
821 output.push({ kind: "synchronized" });
822 }
823
824 output.push(...(yield* coalesceShellEvents(pendingEvents)));
825 return output;
826 });
827
828 const coalesceShellLiveStream = <E, R>(
829 stream: Stream.Stream<ShellLiveInput, E, R>,
830 ): Stream.Stream<OrchestrationShellStreamItem, E, R> =>
831 stream.pipe(
832 Stream.groupedWithin(SHELL_COALESCE_MAX_CHUNK, SHELL_COALESCE_WINDOW),
833 Stream.mapEffect(coalesceShellLiveInputs),
834 Stream.flatMap((items) => Stream.fromIterable(items)),
835 );
836
681837 const dispatchBootstrapTurnStart = (
682838 command: Extract<OrchestrationCommand, { type: "thread.turn.start" }>,
683839 ): Effect.Effect<{ readonly sequence: number }, OrchestrationDispatchCommandError> =>
......@@ -1068,68 +1224,29 @@ const makeWsRpcLayer = (
10681224 observeRpcStreamEffect(
10691225 ORCHESTRATION_WS_METHODS.subscribeShell,
10701226 Effect.gen(function* () {
1071 const liveStream = orchestrationEngine.streamDomainEvents.pipe(
1072 Stream.mapEffect(toShellStreamEvent),
1073 Stream.flatMap((event) =>
1074 Option.isSome(event) ? Stream.succeed(event.value) : Stream.empty,
1227 // Coalesce the live shell stream per aggregate over a small window
1228 // so bursts of high-frequency events (streaming message deltas,
1229 // activity appends) collapse into a single shell refetch and never
1230 // serialize a brand-new thread's `thread.created` behind hundreds
1231 // of per-event DB reads. See coalesceShellStream.
1232 // Attach live delivery into a scope-bound buffer BEFORE loading any
1233 // snapshot or draining catch-up, otherwise an event published while
1234 // the snapshot query is in flight is lost (it is past the snapshot's
1235 // sequence but the live subscription is not attached yet). Every
1236 // path below emits from this same buffered live tail. Overlapping
1237 // events are deduped by sequence on the client.
1238 const liveBuffer = yield* Queue.unbounded<ShellLiveInput>();
1239 yield* Effect.forkScoped(
1240 orchestrationEngine.streamDomainEvents.pipe(
1241 Stream.runForEach((event) =>
1242 Queue.offer(liveBuffer, { kind: "event" as const, event }),
1243 ),
10751244 ),
1245 { startImmediately: true },
10761246 );
1247 const bufferedLiveStream = coalesceShellLiveStream(Stream.fromQueue(liveBuffer));
10771248
1078 // When the client already holds a shell snapshot (cached, or loaded
1079 // over HTTP) it passes that snapshot's sequence, and we resume by
1080 // replaying shell events after it instead of re-sending the whole
1081 // projects/threads list over the socket. As in the thread path, the
1082 // live subscription is attached (into a scope-bound buffer) before
1083 // draining the catch-up replay so no event published during the
1084 // replay window is lost; overlapping events are deduped by sequence
1085 // on the client. The full range is read (not the store's default
1086 // page limit) since the shell filter runs after reading.
1087 if (input.afterSequence !== undefined) {
1088 const afterSequence = input.afterSequence;
1089 return Stream.unwrap(
1090 Effect.gen(function* () {
1091 const liveBuffer = yield* Queue.unbounded<OrchestrationShellStreamItem>();
1092 yield* Effect.forkScoped(
1093 liveStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))),
1094 );
1095 const catchUpStream = orchestrationEngine
1096 .readEvents(afterSequence, Number.MAX_SAFE_INTEGER)
1097 .pipe(
1098 Stream.mapEffect(toShellStreamEvent),
1099 Stream.flatMap((event) =>
1100 Option.isSome(event) ? Stream.succeed(event.value) : Stream.empty,
1101 ),
1102 Stream.mapError(
1103 (cause) =>
1104 new OrchestrationGetSnapshotError({
1105 message: "Failed to replay orchestration shell events",
1106 cause,
1107 }),
1108 ),
1109 );
1110 const afterCatchUp =
1111 input.requestCompletionMarker === true
1112 ? Stream.concat(
1113 Stream.fromEffect(
1114 Queue.offer(liveBuffer, { kind: "synchronized" as const }),
1115 ).pipe(Stream.drain),
1116 Stream.fromQueue(liveBuffer),
1117 )
1118 : Stream.fromQueue(liveBuffer);
1119 return Stream.concat(catchUpStream, afterCatchUp);
1120 }),
1121 );
1122 }
1123
1124 // The full-snapshot fallback needs the same replay-window safety
1125 // as the resume path: subscribe before loading the projection so
1126 // events published while the snapshot is read are buffered.
1127 const liveBuffer = yield* Queue.unbounded<OrchestrationShellStreamItem>();
1128 yield* Effect.forkScoped(
1129 liveStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))),
1130 );
1131 const bufferedLiveStream = Stream.fromQueue(liveBuffer);
1132 const snapshot = yield* projectionSnapshotQuery.getShellSnapshot().pipe(
1249 const loadSnapshot = projectionSnapshotQuery.getShellSnapshot().pipe(
11331250 Effect.tapError((cause) =>
11341251 Effect.logError("orchestration shell snapshot load failed", { cause }),
11351252 ),
......@@ -1142,21 +1259,70 @@ const makeWsRpcLayer = (
11421259 ),
11431260 );
11441261
1145 const afterSnapshot =
1262 // Offer the completion marker into the same queue as live events.
1263 // Anything buffered while snapshot/replay work was in flight is
1264 // therefore delivered before the client is told it is synchronized.
1265 const synchronizedThenLive =
11461266 input.requestCompletionMarker === true
11471267 ? Stream.concat(
11481268 Stream.fromEffect(
1149 Queue.offer(liveBuffer, { kind: "synchronized" as const }),
1150 ).pipe(Stream.drain),
1269 Queue.offer(liveBuffer, { kind: "synchronized" as const }).pipe(
1270 Effect.andThen(Queue.takeAll(liveBuffer)),
1271 Effect.flatMap(coalesceShellLiveInputs),
1272 ),
1273 ).pipe(Stream.flatMap((items) => Stream.fromIterable(items))),
11511274 bufferedLiveStream,
11521275 )
11531276 : bufferedLiveStream;
1277
1278 // When the client already holds a shell snapshot (cached, or loaded
1279 // over HTTP) it passes that snapshot's sequence, and we resume by
1280 // replaying shell events after it instead of re-sending the whole
1281 // projects/threads list over the socket. If the client is too far
1282 // behind, we fall back to a fresh snapshot instead of an unbounded
1283 // replay (see below).
1284 if (input.afterSequence !== undefined) {
1285 const afterSequence = input.afterSequence;
1286 const headSequence = yield* orchestrationEngine.latestSequence;
1287 const replayGap = headSequence - afterSequence;
1288 // Gap too large: replaying every intervening event (each a shell
1289 // refetch) is far more expensive than a single O(active-threads)
1290 // snapshot. A cursor ahead of this engine's authoritative state
1291 // is also invalid, so reset it with a snapshot. Send the snapshot
1292 // followed by the buffered live tail, exactly as the
1293 // no-afterSequence path does.
1294 if (replayGap < 0 || replayGap > SHELL_RESUME_MAX_GAP) {
1295 const snapshot = yield* loadSnapshot;
1296 return Stream.concat(
1297 Stream.make({ kind: "snapshot" as const, snapshot }),
1298 synchronizedThenLive,
1299 );
1300 }
1301 const catchUpStream = coalesceShellStream(
1302 // Replay only through the head captured above. Newer events
1303 // are already covered by the live subscription, so this bound
1304 // cannot chase a moving event-store head or grow the live
1305 // buffer indefinitely while waiting for an empty page.
1306 orchestrationEngine.readEvents(afterSequence, replayGap),
1307 ).pipe(
1308 Stream.mapError(
1309 (cause) =>
1310 new OrchestrationGetSnapshotError({
1311 message: "Failed to replay orchestration shell events",
1312 cause,
1313 }),
1314 ),
1315 );
1316 return Stream.concat(catchUpStream, synchronizedThenLive);
1317 }
1318
1319 const snapshot = yield* loadSnapshot;
11541320 return Stream.concat(
11551321 Stream.make({
11561322 kind: "snapshot" as const,
11571323 snapshot,
11581324 }),
1159 afterSnapshot,
1325 synchronizedThenLive,
11601326 );
11611327 }),
11621328 { "rpc.aggregate": "orchestration" },