authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-10-14 05:38:21-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-14 05:38:21-07:00
log737192594f47d87f28b5f5a3fda3ef792a0d7972
tree3cc11d4ea8f384a2861c7f71216d249316722a7b
parentf37f99ca9e04dd0c4d0ee814402ae224175adc8d
signature Signed by PGP key B5690EEEBB952194

perf(server): unchanged node updates no longer write to SQLite (#19429)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

4 files changed, 88 insertions(+), 0 deletions(-)

apps/server/src/orchestration-v2/ProjectionStore.ts+24
......@@ -73,6 +73,7 @@ import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2Pend
7373import * as Context from "effect/Context";
7474import * as DateTime from "effect/DateTime";
7575import * as Effect from "effect/Effect";
76import * as Equal from "effect/Equal";
7677import * as Layer from "effect/Layer";
7778import * as Ref from "effect/Ref";
7879import * as Schema from "effect/Schema";
......@@ -460,6 +461,10 @@ export interface ProjectionStoreV2Shape {
460461 requestId: TurnItemId,
461462 resultId: TurnItemId,
462463 ) => Effect.Effect<boolean, ProjectionStoreV2Error>;
464 /** True when the projection already stores this exact node, so a write would change nothing. */
465 readonly isNodeStored: (
466 node: OrchestrationV2ExecutionNode,
467 ) => Effect.Effect<boolean, ProjectionStoreV2Error>;
463468 readonly getRuntimeRequest: (
464469 threadId: ThreadId,
465470 requestId: RuntimeRequestId,
......@@ -4404,6 +4409,15 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
44044409 return rows[0]?.pending === 1;
44054410 }).pipe(Effect.mapError(controlReadError(threadId)));
44064411
4412 const isNodeStored: ProjectionStoreV2Shape["isNodeStored"] = (node) =>
4413 Effect.gen(function* () {
4414 const payloadJson = yield* encodeNodePayload(node);
4415 const rows = yield* sql<PayloadRow>`
4416 SELECT payload_json FROM orchestration_v2_projection_nodes WHERE node_id = ${node.id}
4417 `;
4418 return rows[0]?.payload_json === payloadJson;
4419 }).pipe(Effect.mapError(controlReadError(node.threadId)));
4420
44074421 const getRuntimeRequest: ProjectionStoreV2Shape["getRuntimeRequest"] = (threadId, requestId) =>
44084422 sql
44094423 .withTransaction(
......@@ -5992,6 +6006,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
59926006 canStartQueuedRun,
59936007 getPendingNativeUserInputs,
59946008 hasUnpairedRunInterruptRequest,
6009 isNodeStored,
59956010 getMessageCount,
59966011 getNextTurnItemOrdinal,
59976012 getTurnItem,
......@@ -6255,6 +6270,15 @@ export const layerMemory: Layer.Layer<ProjectionStoreV2> = Layer.effect(
62556270 !projection.turnItems.some((item) => item.id === resultId)
62566271 );
62576272 }),
6273 isNodeStored: (node) =>
6274 Ref.get(replayState).pipe(
6275 Effect.map((state) => {
6276 const stored = state.projections
6277 .get(node.threadId)
6278 ?.nodes.find((candidate) => candidate.id === node.id);
6279 return stored !== undefined && Equal.equals(stored, node);
6280 }),
6281 ),
62586282 getMessageCount: (threadId) =>
62596283 Ref.get(replayState).pipe(
62606284 Effect.map((state) => state.projections.get(threadId)?.messages.length ?? 0),
apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts+60
......@@ -325,6 +325,66 @@ layer("ProviderEventIngestorV2", (it) => {
325325 }),
326326 );
327327
328 it.effect("skips a node update only when it matches the stored node", () =>
329 Effect.gen(function* () {
330 const now = yield* DateTime.now;
331 const eventSink = yield* EventSink.EventSinkV2;
332 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
333 const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2;
334 const idAllocator = yield* IdAllocator.IdAllocatorV2;
335 const threadEvent = yield* threadCreatedEvent(now);
336 const threadId = threadEvent.threadId;
337 yield* eventSink.write({ events: [threadEvent] });
338 const providerSessionId = yield* idAllocator.allocate.providerSession({
339 providerInstanceId: modelSelection.instanceId,
340 threadId,
341 });
342 const nodeId = NodeId.make(`${threadId}:approval`);
343 const node: OrchestrationV2ExecutionNode = {
344 id: nodeId,
345 threadId,
346 runId: null,
347 parentNodeId: null,
348 rootNodeId: nodeId,
349 kind: "approval_request",
350 status: "waiting",
351 countsForRun: false,
352 providerThreadId: null,
353 providerTurnId: null,
354 nativeItemRef: null,
355 runtimeRequestId: null,
356 checkpointScopeId: null,
357 startedAt: now,
358 completedAt: null,
359 };
360 const ingestNode = (sent: OrchestrationV2ExecutionNode) =>
361 ingestor.ingestNormalized({
362 providerSessionId,
363 providerInstanceId: modelSelection.instanceId,
364 threadId,
365 event: { type: "node.updated", driver: CODEX_DRIVER, node: sent },
366 });
367
368 assert.lengthOf(yield* ingestNode(node), 1);
369 assert.isEmpty(yield* ingestNode({ ...node }));
370 // The server resolves the request, then the provider asks again with the same node.
371 yield* eventSink.write({
372 events: [
373 {
374 id: yield* idAllocator.allocate.event({ threadId }),
375 type: "node.updated",
376 threadId,
377 occurredAt: now,
378 payload: { ...node, status: "completed", completedAt: now },
379 },
380 ],
381 });
382 assert.lengthOf(yield* ingestNode(node), 1);
383 const projection = yield* projectionStore.getThreadProjection(threadId);
384 assert.equal(projection.nodes.find((entry) => entry.id === nodeId)?.status, "waiting");
385 }),
386 );
387
328388 it.effect("rolls back runs a provider rewound off its active branch", () =>
329389 Effect.gen(function* () {
330390 const now = yield* DateTime.now;
apps/server/src/orchestration-v2/ProviderEventIngestor.ts+3
......@@ -611,6 +611,9 @@ export const layer: Layer.Layer<
611611 }),
612612 ];
613613 case "node.updated":
614 // Adapters send an unchanged node again with each text delta. Compare with the
615 // stored node, not the last one sent, because the server also writes nodes.
616 if (yield* projections.isNodeStored(input.event.node)) return [];
614617 return [
615618 yield* makeDomainEvent(input, {
616619 type: "node.updated",
apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts+1
......@@ -255,6 +255,7 @@ it.effect(
255255 getCheckpointContext: () => Effect.die("not used"),
256256 getCheckpointCaptureContext: () => Effect.die("not used"),
257257 getRunMessage: () => Effect.die("not used"),
258 isNodeStored: () => Effect.die("not used"),
258259 canStartQueuedRun: () => Effect.die("not used"),
259260 getRecoveryThreadIds: () => Effect.die("unused getRecoveryThreadIds"),
260261 getUnreadableThreadIds: () => Effect.die("unused getUnreadableThreadIds"),