Skip to content

perf(protocol): bound stale thread replay #89

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tarik02 merged 1 commit into from
Jul 29, 2026
Merged

perf(protocol): bound stale thread replay #89

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions FORK.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,19 @@ This repository is a fork of `pingdotgg/t3code`. Keep this file focused on fork
- Exception: upstream actualization PRs may preserve upstream commit structure when that makes future syncs easier to audit.
- Staged formatting tolerates chunks containing only ignored files so large upstream actualization commits can pass the pre-commit hook.

### Protocol Compatibility

- Fork protocol extensions use separate extension RPC methods and additive optional capabilities.
- Existing upstream RPC methods keep their upstream request, response, and behavior contracts so upstream clients remain compatible with the fork backend.
- Fork clients call a fork RPC only when the backend advertises its capability and otherwise retain the upstream RPC path.
- Fork backends keep the upstream RPC handler alongside each fork extension.

### Thread Delta Subscription

- Fork backends advertise `threadDeltaSubscription` and expose `orchestration.subscribeThread.withDelta` alongside the upstream thread subscription.
- Fork clients use the fork subscription only when advertised. Upstream backends therefore continue receiving the upstream subscription.
- The fork subscription replays gaps of at most 1,000 orchestration events and sends a fresh thread snapshot for larger or invalid gaps. If the thread no longer exists, it sends `not-found` so clients clear stale cached state.

### Release And CI

- Fork workflows create/update a daily stable release PR while main-branch pushes produce nightly releases.
Expand Down
1 change: 1 addition & 0 deletions apps/server/src/environment/ServerEnvironment.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -142,6 +142,7 @@ export const make = Effect.gen(function* () {
connectionProbe: true,
threadSettlement: true,
threadSnooze: true,
threadDeltaSubscription: true,
...(serverSelfUpdate === null ? {} : { serverSelfUpdate }),
},
};
Expand Down
239 changes: 135 additions & 104 deletions apps/server/src/ws.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ import {
type OrchestrationEvent,
type OrchestrationShellStreamEvent,
type OrchestrationShellStreamItem,
type OrchestrationSubscribeThreadInput,
type OrchestrationThreadStreamItem,
OrchestrationGetFullThreadDiffError,
OrchestrationGetSnapshotError,
Expand Down Expand Up @@ -294,6 +295,7 @@ const PROVIDER_STATUS_DEBOUNCE_MS = 200;
// past this gap a single O(active-threads) snapshot is cheaper and bounded.
// Matches the event store's default page size (DEFAULT_READ_FROM_SEQUENCE_LIMIT).
const SHELL_RESUME_MAX_GAP = 1_000;
const THREAD_DELTA_SUBSCRIPTION_MAX_GAP = 1_000;

const RPC_REQUIRED_SCOPE = new Map<string, AuthEnvironmentScope>([
[ORCHESTRATION_WS_METHODS.dispatchCommand, AuthOrchestrationOperateScope],
Expand All @@ -302,6 +304,7 @@ const RPC_REQUIRED_SCOPE = new Map<string, AuthEnvironmentScope>([
[ORCHESTRATION_WS_METHODS.subscribeShell, AuthOrchestrationReadScope],
[ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot, AuthOrchestrationReadScope],
[ORCHESTRATION_WS_METHODS.subscribeThread, AuthOrchestrationReadScope],
[ORCHESTRATION_WS_METHODS.subscribeThreadWithDelta, AuthOrchestrationReadScope],
[WS_METHODS.serverProbe, AuthOrchestrationReadScope],
[WS_METHODS.serverGetConfig, AuthOrchestrationReadScope],
[WS_METHODS.serverRefreshProviders, AuthOrchestrationOperateScope],
Expand Down Expand Up @@ -1026,6 +1029,122 @@ const makeWsRpcLayer = (
);
};

const loadThreadSnapshot = Effect.fn("Ws.loadThreadSnapshot")(function* (threadId: ThreadId) {
const snapshot = yield* projectionSnapshotQuery.getThreadDetailSnapshot(threadId).pipe(
Effect.mapError(
(cause) =>
new OrchestrationGetSnapshotError({
message: `Failed to load thread ${threadId}`,
cause,
}),
),
);

return Option.map(snapshot, projectThreadDetailSnapshot);
});

const makeThreadSubscription = Effect.fn("Ws.makeThreadSubscription")(function* (input: {
readonly request: OrchestrationSubscribeThreadInput;
readonly maxReplayGap: number | null;
}) {
const isThisThreadDetailEvent = (event: OrchestrationEvent) =>
event.aggregateKind === "thread" &&
event.aggregateId === input.request.threadId &&
isThreadDetailEvent(event);

const liveStream = orchestrationEngine.streamDomainEvents.pipe(
Stream.filter(isThisThreadDetailEvent),
Stream.map((event) => ({
kind: "event" as const,
event: projectActivityEvent(event),
})),
);

// Attach live delivery before reading either replay or snapshot state.
// Otherwise an event published while the snapshot is loading is lost.
const liveBuffer = yield* Queue.unbounded<OrchestrationThreadStreamItem>();
yield* Effect.forkScoped(
liveStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))),
);
const bufferedLiveStream = Stream.fromQueue(liveBuffer);
const synchronizedThenLive =
input.request.requestCompletionMarker === true
? Stream.concat(
Stream.fromEffect(Queue.offer(liveBuffer, { kind: "synchronized" as const })).pipe(
Stream.drain,
),
bufferedLiveStream,
)
: bufferedLiveStream;

if (input.request.afterSequence !== undefined) {
const afterSequence = input.request.afterSequence;
if (input.maxReplayGap !== null) {
const headSequence = yield* orchestrationEngine.latestSequence;
const replayGap = headSequence - afterSequence;
if (replayGap < 0 || replayGap > input.maxReplayGap) {
const snapshot = yield* loadThreadSnapshot(input.request.threadId);
Comment thread
tarik02 marked this conversation as resolved.
if (Option.isNone(snapshot)) {
return Stream.concat(
Stream.make({ kind: "not-found" as const }),
synchronizedThenLive,
);
}
return Stream.concat(
Stream.make({ kind: "snapshot" as const, snapshot: snapshot.value }),
synchronizedThenLive,
);
}

const catchUpStream = orchestrationEngine.readEvents(afterSequence, replayGap).pipe(
Stream.filter(isThisThreadDetailEvent),
Stream.map((event) => ({
kind: "event" as const,
event: projectActivityEvent(event),
})),
Stream.mapError(
(cause) =>
new OrchestrationGetSnapshotError({
message: `Failed to replay thread ${input.request.threadId} events`,
cause,
}),
),
);
return Stream.concat(catchUpStream, synchronizedThenLive);
}

const catchUpStream = orchestrationEngine
.readEvents(afterSequence, Number.MAX_SAFE_INTEGER)
.pipe(
Stream.filter(isThisThreadDetailEvent),
Stream.map((event) => ({
kind: "event" as const,
event: projectActivityEvent(event),
})),
Stream.mapError(
(cause) =>
new OrchestrationGetSnapshotError({
message: `Failed to replay thread ${input.request.threadId} events`,
cause,
}),
),
);
return Stream.concat(catchUpStream, synchronizedThenLive);
}

const snapshot = yield* loadThreadSnapshot(input.request.threadId);
if (Option.isNone(snapshot)) {
return yield* new OrchestrationGetSnapshotError({
message: `Thread ${input.request.threadId} was not found`,
cause: input.request.threadId,
});
}
return Stream.concat(
Stream.make({ kind: "snapshot" as const, snapshot: snapshot.value }),
synchronizedThenLive,
);
});

const loadServerConfig = Effect.gen(function* () {
const keybindingsConfig = yield* keybindings.loadConfigState;
const providers = yield* providerRegistry.getProviders;
Expand Down Expand Up @@ -1289,110 +1408,22 @@ const makeWsRpcLayer = (
[ORCHESTRATION_WS_METHODS.subscribeThread]: (input) =>
observeRpcStreamEffect(
ORCHESTRATION_WS_METHODS.subscribeThread,
Effect.gen(function* () {
const isThisThreadDetailEvent = (event: OrchestrationEvent) =>
event.aggregateKind === "thread" &&
event.aggregateId === input.threadId &&
isThreadDetailEvent(event);

const liveStream = orchestrationEngine.streamDomainEvents.pipe(
Stream.filter(isThisThreadDetailEvent),
Stream.map((event) => ({
kind: "event" as const,
event: projectActivityEvent(event),
})),
);

// Attach live delivery before reading either replay or snapshot state.
// Otherwise an event published while the snapshot is loading is lost.
const liveBuffer = yield* Queue.unbounded<OrchestrationThreadStreamItem>();
yield* Effect.forkScoped(
liveStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))),
);
const bufferedLiveStream = Stream.fromQueue(liveBuffer);

// When the client already loaded the snapshot over HTTP it passes
// that snapshot's sequence, and we resume the live subscription by
// replaying persisted events after it instead of re-sending the
// (potentially multi-KB) snapshot frame over the socket.
//
// The live PubSub subscription must be attached *before* draining
// the catch-up replay, otherwise events published during the replay
// window are dropped (they are past the persisted tail the replay
// read, but the live stream is not yet subscribed). So fork the
// live stream into a buffer bound to this stream's scope, then emit
// catch-up followed by the buffered/ongoing live events. Overlapping
// events are deduped by sequence on the client.
//
// Read the full range after the cursor (not the store's default
// page-bounded limit): the range is normally tiny (a fresh HTTP
// snapshot sequence) and the per-thread filter runs after reading,
// so a global cap could otherwise omit this thread's events.
if (input.afterSequence !== undefined) {
const afterSequence = input.afterSequence;
const catchUpStream = orchestrationEngine
.readEvents(afterSequence, Number.MAX_SAFE_INTEGER)
.pipe(
Stream.filter(isThisThreadDetailEvent),
Stream.map((event) => ({
kind: "event" as const,
event: projectActivityEvent(event),
})),
Stream.mapError(
(cause) =>
new OrchestrationGetSnapshotError({
message: `Failed to replay thread ${input.threadId} events`,
cause,
}),
),
);
const afterCatchUp =
input.requestCompletionMarker === true
? Stream.concat(
Stream.fromEffect(
Queue.offer(liveBuffer, { kind: "synchronized" as const }),
).pipe(Stream.drain),
bufferedLiveStream,
)
: bufferedLiveStream;
return Stream.concat(catchUpStream, afterCatchUp);
}

const snapshot = yield* projectionSnapshotQuery
.getThreadDetailSnapshot(input.threadId)
.pipe(
Effect.mapError(
(cause) =>
new OrchestrationGetSnapshotError({
message: `Failed to load thread ${input.threadId}`,
cause,
}),
),
);

if (Option.isNone(snapshot)) {
return yield* new OrchestrationGetSnapshotError({
message: `Thread ${input.threadId} was not found`,
cause: input.threadId,
});
}

const afterSnapshot =
input.requestCompletionMarker === true
? Stream.concat(
Stream.fromEffect(
Queue.offer(liveBuffer, { kind: "synchronized" as const }),
).pipe(Stream.drain),
bufferedLiveStream,
)
: bufferedLiveStream;
return Stream.concat(
Stream.make({
kind: "snapshot" as const,
snapshot: projectThreadDetailSnapshot(snapshot.value),
}),
afterSnapshot,
);
makeThreadSubscription({ request: input, maxReplayGap: null }).pipe(
Effect.map((stream) =>
Stream.filter(
stream,
(item): item is OrchestrationThreadStreamItem => item.kind !== "not-found",
),
),
),
{ "rpc.aggregate": "orchestration" },
),
[ORCHESTRATION_WS_METHODS.subscribeThreadWithDelta]: (input) =>
observeRpcStreamEffect(
ORCHESTRATION_WS_METHODS.subscribeThreadWithDelta,
makeThreadSubscription({
request: input,
maxReplayGap: THREAD_DELTA_SUBSCRIPTION_MAX_GAP,
}),
{ "rpc.aggregate": "orchestration" },
),
Expand Down
Loading
Loading