| author |  Ishan <ishansachu1@gmail.com> 2026-07-17 15:28:09+05:30 |
| committer |  GitHub <noreply@github.com> 2026-07-17 11:58:09+02:00 |
| log | c14a5ca492b4da11da81e482e307222946536300 |
| tree | 627df3e60eab1d8c58a4dcce36ebd4933840803f |
| parent | eb733c10f270bcfc2638b72eb25a8eaa314cb08a |
| signature | Signed by PGP key B5690EEEBB952194 |
Fix dropped events during initial thread snapshot (#4079)
2 files changed, 81 insertions(+), 23 deletions(-)
apps
server
apps/server/src/server.test.ts+58
| ... | ... | @@ -5607,6 +5607,64 @@ it.layer(NodeServices.layer)("server router seam", (it) => { |
| 5607 | 5607 | }).pipe(Effect.provide(NodeHttpServer.layerTest)), |
| 5608 | 5608 | ); |
| 5609 | 5609 | |
| 5610 | it.effect("buffers thread events published while the initial snapshot loads", () => | |
| 5611 | Effect.gen(function* () { | |
| 5612 | const thread = makeDefaultOrchestrationReadModel().threads[0]!; | |
| 5613 | const liveEvents = yield* PubSub.unbounded<OrchestrationEvent>(); | |
| 5614 | const messageEvent = { | |
| 5615 | sequence: 2, | |
| 5616 | eventId: EventId.make("event-message"), | |
| 5617 | aggregateKind: "thread", | |
| 5618 | aggregateId: defaultThreadId, | |
| 5619 | occurredAt: "2026-01-01T00:00:01.000Z", | |
| 5620 | commandId: null, | |
| 5621 | causationEventId: null, | |
| 5622 | correlationId: null, | |
| 5623 | metadata: {}, | |
| 5624 | type: "thread.message-sent", | |
| 5625 | payload: { | |
| 5626 | threadId: defaultThreadId, | |
| 5627 | messageId: MessageId.make("message-1"), | |
| 5628 | role: "user", | |
| 5629 | text: "First message", | |
| 5630 | turnId: null, | |
| 5631 | streaming: false, | |
| 5632 | createdAt: "2026-01-01T00:00:01.000Z", | |
| 5633 | updatedAt: "2026-01-01T00:00:01.000Z", | |
| 5634 | }, | |
| 5635 | } satisfies Extract<OrchestrationEvent, { type: "thread.message-sent" }>; | |
| 5636 | ||
| 5637 | yield* buildAppUnderTest({ | |
| 5638 | layers: { | |
| 5639 | orchestrationEngine: { | |
| 5640 | streamDomainEvents: Stream.fromPubSub(liveEvents), | |
| 5641 | }, | |
| 5642 | projectionSnapshotQuery: { | |
| 5643 | getThreadDetailSnapshot: () => | |
| 5644 | Effect.gen(function* () { | |
| 5645 | yield* Effect.sleep("25 millis"); | |
| 5646 | yield* PubSub.publish(liveEvents, messageEvent); | |
| 5647 | return Option.some({ snapshotSequence: 1, thread }); | |
| 5648 | }), | |
| 5649 | }, | |
| 5650 | }, | |
| 5651 | }); | |
| 5652 | ||
| 5653 | const wsUrl = yield* getWsServerUrl("/ws"); | |
| 5654 | const items = yield* Effect.scoped( | |
| 5655 | withWsRpcClient(wsUrl, (client) => | |
| 5656 | client[ORCHESTRATION_WS_METHODS.subscribeThread]({ | |
| 5657 | threadId: defaultThreadId, | |
| 5658 | }).pipe(Stream.take(2), Stream.runCollect), | |
| 5659 | ), | |
| 5660 | ).pipe(Effect.timeout("2 seconds")); | |
| 5661 | ||
| 5662 | assert.equal(items[0]?.kind, "snapshot"); | |
| 5663 | assert.equal(items[1]?.kind, "event"); | |
| 5664 | assert.equal(items[1]?.kind === "event" ? items[1].event.sequence : null, 2); | |
| 5665 | }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive), | |
| 5666 | ); | |
| 5667 | ||
| 5610 | 5668 | it.effect("enriches replayed project events with repository identity metadata", () => |
| 5611 | 5669 | Effect.gen(function* () { |
| 5612 | 5670 | const repositoryIdentity = { |
apps/server/src/ws.ts+23-23
| ... | ... | @@ -1166,6 +1166,14 @@ const makeWsRpcLayer = ( |
| 1166 | 1166 | })), |
| 1167 | 1167 | ); |
| 1168 | 1168 | |
| 1169 | // Attach live delivery before reading either replay or snapshot state. | |
| 1170 | // Otherwise an event published while the snapshot is loading is lost. | |
| 1171 | const liveBuffer = yield* Queue.unbounded<OrchestrationThreadStreamItem>(); | |
| 1172 | yield* Effect.forkScoped( | |
| 1173 | liveStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))), | |
| 1174 | ); | |
| 1175 | const bufferedLiveStream = Stream.fromQueue(liveBuffer); | |
| 1176 | ||
| 1169 | 1177 | // When the client already loaded the snapshot over HTTP it passes |
| 1170 | 1178 | // that snapshot's sequence, and we resume the live subscription by |
| 1171 | 1179 | // replaying persisted events after it instead of re-sending the |
| ... | ... | @@ -1185,28 +1193,20 @@ const makeWsRpcLayer = ( |
| 1185 | 1193 | // so a global cap could otherwise omit this thread's events. |
| 1186 | 1194 | if (input.afterSequence !== undefined) { |
| 1187 | 1195 | const afterSequence = input.afterSequence; |
| 1188 | return Stream.unwrap( | |
| 1189 | Effect.gen(function* () { | |
| 1190 | const liveBuffer = yield* Queue.unbounded<OrchestrationThreadStreamItem>(); | |
| 1191 | yield* Effect.forkScoped( | |
| 1192 | liveStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))), | |
| 1193 | ); | |
| 1194 | const catchUpStream = orchestrationEngine | |
| 1195 | .readEvents(afterSequence, Number.MAX_SAFE_INTEGER) | |
| 1196 | .pipe( | |
| 1197 | Stream.filter(isThisThreadDetailEvent), | |
| 1198 | Stream.map((event) => ({ kind: "event" as const, event })), | |
| 1199 | Stream.mapError( | |
| 1200 | (cause) => | |
| 1201 | new OrchestrationGetSnapshotError({ | |
| 1202 | message: `Failed to replay thread ${input.threadId} events`, | |
| 1203 | cause, | |
| 1204 | }), | |
| 1205 | ), | |
| 1206 | ); | |
| 1207 | return Stream.concat(catchUpStream, Stream.fromQueue(liveBuffer)); | |
| 1208 | }), | |
| 1209 | ); | |
| 1196 | const catchUpStream = orchestrationEngine | |
| 1197 | .readEvents(afterSequence, Number.MAX_SAFE_INTEGER) | |
| 1198 | .pipe( | |
| 1199 | Stream.filter(isThisThreadDetailEvent), | |
| 1200 | Stream.map((event) => ({ kind: "event" as const, event })), | |
| 1201 | Stream.mapError( | |
| 1202 | (cause) => | |
| 1203 | new OrchestrationGetSnapshotError({ | |
| 1204 | message: `Failed to replay thread ${input.threadId} events`, | |
| 1205 | cause, | |
| 1206 | }), | |
| 1207 | ), | |
| 1208 | ); | |
| 1209 | return Stream.concat(catchUpStream, bufferedLiveStream); | |
| 1210 | 1210 | } |
| 1211 | 1211 | |
| 1212 | 1212 | const snapshot = yield* projectionSnapshotQuery |
| ... | ... | @@ -1233,7 +1233,7 @@ const makeWsRpcLayer = ( |
| 1233 | 1233 | kind: "snapshot" as const, |
| 1234 | 1234 | snapshot: snapshot.value, |
| 1235 | 1235 | }), |
| 1236 | liveStream, | |
| 1236 | bufferedLiveStream, | |
| 1237 | 1237 | ); |
| 1238 | 1238 | }), |
| 1239 | 1239 | { "rpc.aggregate": "orchestration" }, |