Skip to content

Commit 29677d1

Browse files
cursoragentcursoragent
committed
Fix thread subscription event loss and snapshot sequence race
- Attach the domain-event PubSub subscription before reading the catch-up replay or snapshot baseline in subscribeThread, so events published while the baseline loads are buffered instead of dropped (new OrchestrationEngine.subscribeDomainEvents, scoped to the RPC stream's lifetime via observeRpcStreamEffect). - Make the afterSequence catch-up replay exhaustive instead of truncating at the event store's default 1,000-event cap, which could silently drop thread events under multi-thread activity. - Read the thread detail and projection snapshot sequence inside one transaction (ProjectionSnapshotQuery.getThreadDetailSnapshot), shared by the HTTP threadSnapshot endpoint and the WS snapshot path, so the reported sequence can never run ahead of the embedded thread.
1 parent fad44e1 commit 29677d1

13 files changed

Lines changed: 117 additions & 47 deletions

‎apps/server/src/checkpointing/CheckpointDiffQuery.test.ts‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -107,6 +107,7 @@ describe("CheckpointDiffQuery.layer", () => {
107107
}),
108108
getThreadShellById: () => Effect.succeed(Option.none()),
109109
getThreadDetailById: () => Effect.succeed(Option.none()),
110+
getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
110111
}),
111112
),
112113
);
@@ -199,6 +200,7 @@ describe("CheckpointDiffQuery.layer", () => {
199200
getFullThreadDiffContext: () => Effect.die("unused"),
200201
getThreadShellById: () => Effect.succeed(Option.none()),
201202
getThreadDetailById: () => Effect.succeed(Option.none()),
203+
getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
202204
}),
203205
),
204206
);
@@ -281,6 +283,7 @@ describe("CheckpointDiffQuery.layer", () => {
281283
getFullThreadDiffContext: () => Effect.die("unused"),
282284
getThreadShellById: () => Effect.succeed(Option.none()),
283285
getThreadDetailById: () => Effect.succeed(Option.none()),
286+
getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
284287
}),
285288
),
286289
);
@@ -348,6 +351,7 @@ describe("CheckpointDiffQuery.layer", () => {
348351
getFullThreadDiffContext: () => Effect.die("unused"),
349352
getThreadShellById: () => Effect.succeed(Option.none()),
350353
getThreadDetailById: () => Effect.succeed(Option.none()),
354+
getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
351355
}),
352356
),
353357
);
@@ -400,6 +404,7 @@ describe("CheckpointDiffQuery.layer", () => {
400404
getFullThreadDiffContext: () => Effect.succeed(Option.none()),
401405
getThreadShellById: () => Effect.succeed(Option.none()),
402406
getThreadDetailById: () => Effect.succeed(Option.none()),
407+
getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
403408
}),
404409
),
405410
);

‎apps/server/src/observability/RpcInstrumentation.ts‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import * as Effect from "effect/Effect";
55
import * as Exit from "effect/Exit";
66
import * as Metric from "effect/Metric";
77
import * as References from "effect/References";
8+
import type * as Scope from "effect/Scope";
89
import * as Stream from "effect/Stream";
910

1011
import { outcomeFromExit } from "./Attributes.ts";
@@ -123,7 +124,14 @@ export const observeRpcStreamEffect = <A, StreamError, StreamContext, EffectErro
123124
method: string,
124125
effect: Effect.Effect<Stream.Stream<A, StreamError, StreamContext>, EffectError, EffectContext>,
125126
traceAttributes?: Readonly<Record<string, unknown>>,
126-
): Stream.Stream<A, StreamError | EffectError, StreamContext | EffectContext> => {
127+
// `Stream.unwrap` scopes the setup effect to the stream's lifetime, so a
128+
// `Scope` requirement (e.g. PubSub subscriptions attached before a snapshot
129+
// is loaded) is satisfied by the stream itself rather than the caller.
130+
): Stream.Stream<
131+
A,
132+
StreamError | EffectError,
133+
StreamContext | Exclude<EffectContext, Scope.Scope>
134+
> => {
127135
const instrumented = Stream.unwrap(
128136
Effect.gen(function* () {
129137
const startedAt = yield* Clock.currentTimeNanos;

‎apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -200,6 +200,7 @@ describe("OrchestrationEngine", () => {
200200
getFullThreadDiffContext: () => Effect.succeed(Option.none()),
201201
getThreadShellById: () => Effect.succeed(Option.none()),
202202
getThreadDetailById: () => Effect.succeed(Option.none()),
203+
getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
203204
}),
204205
),
205206
Layer.provide(

‎apps/server/src/orchestration/Layers/OrchestrationEngine.ts‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -306,8 +306,8 @@ const makeOrchestrationEngine = Effect.gen(function* () {
306306
Effect.annotateLogs({ sequence: commandReadModel.snapshotSequence }),
307307
);
308308

309-
const readEvents: OrchestrationEngineShape["readEvents"] = (fromSequenceExclusive) =>
310-
eventStore.readFromSequence(fromSequenceExclusive);
309+
const readEvents: OrchestrationEngineShape["readEvents"] = (fromSequenceExclusive, limit) =>
310+
eventStore.readFromSequence(fromSequenceExclusive, limit);
311311

312312
const dispatch: OrchestrationEngineShape["dispatch"] = (command) =>
313313
Effect.gen(function* () {
@@ -329,6 +329,7 @@ const makeOrchestrationEngine = Effect.gen(function* () {
329329
get streamDomainEvents(): OrchestrationEngineShape["streamDomainEvents"] {
330330
return Stream.fromPubSub(eventPubSub);
331331
},
332+
subscribeDomainEvents: Effect.map(PubSub.subscribe(eventPubSub), Stream.fromSubscription),
332333
} satisfies OrchestrationEngineShape;
333334
});
334335

‎apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2033,6 +2033,23 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
20332033
);
20342034
});
20352035

2036+
const getThreadDetailSnapshot: ProjectionSnapshotQueryShape["getThreadDetailSnapshot"] = (
2037+
threadId,
2038+
) =>
2039+
sql.withTransaction(Effect.all([getThreadDetailById(threadId), getSnapshotSequence()])).pipe(
2040+
Effect.map(([threadDetail, { snapshotSequence }]) =>
2041+
Option.map(threadDetail, (thread) => ({ snapshotSequence, thread })),
2042+
),
2043+
Effect.mapError((error) => {
2044+
if (isPersistenceError(error)) {
2045+
return error;
2046+
}
2047+
return toPersistenceSqlError("ProjectionSnapshotQuery.getThreadDetailSnapshot:query")(
2048+
error,
2049+
);
2050+
}),
2051+
);
2052+
20362053
return {
20372054
getCommandReadModel,
20382055
getSnapshot,
@@ -2047,6 +2064,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
20472064
getFullThreadDiffContext,
20482065
getThreadShellById,
20492066
getThreadDetailById,
2067+
getThreadDetailSnapshot,
20502068
} satisfies ProjectionSnapshotQueryShape;
20512069
});
20522070

‎apps/server/src/orchestration/Services/OrchestrationEngine.ts‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313
import type { OrchestrationCommand, OrchestrationEvent } from "@t3tools/contracts";
1414
import * as Context from "effect/Context";
1515
import type * as Effect from "effect/Effect";
16+
import type * as Scope from "effect/Scope";
1617
import type * as Stream from "effect/Stream";
1718

1819
import type { OrchestrationDispatchError } from "../Errors.ts";
@@ -26,10 +27,14 @@ export interface OrchestrationEngineShape {
2627
* Replay persisted orchestration events from an exclusive sequence cursor.
2728
*
2829
* @param fromSequenceExclusive - Sequence cursor (exclusive).
30+
* @param limit - Optional maximum number of events to replay. Defaults to
31+
* the event store's replay cap; pass `Number.MAX_SAFE_INTEGER` for an
32+
* exhaustive catch-up replay.
2933
* @returns Stream containing ordered events.
3034
*/
3135
readonly readEvents: (
3236
fromSequenceExclusive: number,
37+
limit?: number,
3338
) => Stream.Stream<OrchestrationEvent, OrchestrationEventStoreError, never>;
3439

3540
/**
@@ -49,8 +54,26 @@ export interface OrchestrationEngineShape {
4954
* Stream persisted domain events in dispatch order.
5055
*
5156
* This is a hot runtime stream (new events only), not a historical replay.
57+
* The underlying PubSub subscription is only attached once the stream is
58+
* pulled; use `subscribeDomainEvents` when the subscription must be
59+
* attached before other work (e.g. loading a snapshot baseline).
5260
*/
5361
readonly streamDomainEvents: Stream.Stream<OrchestrationEvent>;
62+
63+
/**
64+
* Attach a domain event subscription immediately and return the stream of
65+
* events it receives.
66+
*
67+
* Unlike `streamDomainEvents`, events published between running this effect
68+
* and pulling the returned stream are buffered by the subscription instead
69+
* of dropped, which is required for snapshot/catch-up + live combinations.
70+
* The subscription is released when the surrounding scope closes.
71+
*/
72+
readonly subscribeDomainEvents: Effect.Effect<
73+
Stream.Stream<OrchestrationEvent>,
74+
never,
75+
Scope.Scope
76+
>;
5477
}
5578

5679
/**

‎apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ import type {
1414
OrchestrationReadModel,
1515
OrchestrationShellSnapshot,
1616
OrchestrationThread,
17+
OrchestrationThreadDetailSnapshot,
1718
OrchestrationThreadShell,
1819
ProjectId,
1920
ThreadId,
@@ -157,6 +158,15 @@ export interface ProjectionSnapshotQueryShape {
157158
readonly getThreadDetailById: (
158159
threadId: ThreadId,
159160
) => Effect.Effect<Option.Option<OrchestrationThread>, ProjectionRepositoryError>;
161+
162+
/**
163+
* Read a single active thread detail together with the projection snapshot
164+
* sequence, both observed inside one transaction so the sequence never runs
165+
* ahead of the thread rows it is paired with.
166+
*/
167+
readonly getThreadDetailSnapshot: (
168+
threadId: ThreadId,
169+
) => Effect.Effect<Option.Option<OrchestrationThreadDetailSnapshot>, ProjectionRepositoryError>;
160170
}
161171

162172
/**

‎apps/server/src/orchestration/http.ts‎

Lines changed: 9 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -45,18 +45,17 @@ export const orchestrationHttpApiLayer = HttpApiBuilder.group(
4545
Effect.fn("environment.orchestration.threadSnapshot")(function* (args) {
4646
yield* annotateEnvironmentRequest(args.endpoint.name);
4747
yield* requireEnvironmentScope(AuthOrchestrationReadScope);
48-
const [threadDetail, { snapshotSequence }] = yield* Effect.all([
49-
projectionSnapshotQuery.getThreadDetailById(args.params.threadId),
50-
projectionSnapshotQuery.getSnapshotSequence(),
51-
]).pipe(
52-
Effect.catch((cause) =>
53-
failEnvironmentInternal("orchestration_thread_snapshot_failed", cause),
54-
),
55-
);
56-
if (Option.isNone(threadDetail)) {
48+
const snapshot = yield* projectionSnapshotQuery
49+
.getThreadDetailSnapshot(args.params.threadId)
50+
.pipe(
51+
Effect.catch((cause) =>
52+
failEnvironmentInternal("orchestration_thread_snapshot_failed", cause),
53+
),
54+
);
55+
if (Option.isNone(snapshot)) {
5756
return yield* failEnvironmentNotFound("thread_not_found");
5857
}
59-
return { snapshotSequence, thread: threadDetail.value };
58+
return snapshot.value;
6059
}),
6160
)
6261
.handle(

‎apps/server/src/project/ProjectSetupScriptRunner.test.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@ const makeProjectionSnapshotQueryLayer = (project: OrchestrationProject) =>
4343
getFullThreadDiffContext: () => Effect.die("unused"),
4444
getThreadShellById: () => Effect.die("unused"),
4545
getThreadDetailById: () => Effect.die("unused"),
46+
getThreadDetailSnapshot: () => Effect.die("unused"),
4647
});
4748

4849
const makeTerminalManagerLayer = (

‎apps/server/src/provider/Layers/ProviderSessionReaper.test.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -209,6 +209,7 @@ describe("ProviderSessionReaper", () => {
209209
: Option.none(),
210210
),
211211
getThreadDetailById: () => Effect.die("unused"),
212+
getThreadDetailSnapshot: () => Effect.die("unused"),
212213
}),
213214
),
214215
Layer.provideMerge(NodeServices.layer),

0 commit comments

Comments
 (0)