From bc1bb52cb1b5e66346714843197a1f38bc642a39 Mon Sep 17 00:00:00 2001 From: Hieu Le Date: Thu, 29 Oct 2026 00:11:50 +0100 Subject: [PATCH 1/2] perf(server): replay stored events about twice as fast Replaying stored events validated every event twice (once from JSON, then again as a stored event), parsed row JSON through a schema wrapper, traced one span per replayed row into the server trace file, and waited for each page query before decoding it. Decode each event once and check only the sequence/commandId header, use JSON.parse with the schema decoder as the error fallback, drop the per-row span, and read the next page on the reader while the current one decodes. Co-Authored-By: Claude Opus 5.5 --- .../persistence/OrchestrationEventStore.ts | 54 ++++++++++++------- 1 file changed, 35 insertions(+), 19 deletions(-) diff --git a/apps/server/src/persistence/OrchestrationEventStore.ts b/apps/server/src/persistence/OrchestrationEventStore.ts index 51c3861a9fca0..8c531bdf8b3d2 100644 --- a/apps/server/src/persistence/OrchestrationEventStore.ts +++ b/apps/server/src/persistence/OrchestrationEventStore.ts @@ -19,6 +19,7 @@ import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; import * as SqlClient from "effect/sql/SqlClient"; import * as SqlSchema from "effect/sql/SqlSchema"; import * as Effect from "effect/Effect"; +import { identity } from "effect/Function"; import * as Layer from "effect/Layer"; import * as PubSub from "effect/PubSub"; import * as Option from "effect/Option"; @@ -174,8 +175,21 @@ interface ApplicationEventRow { const decodeV2EventJson = Schema.decodeUnknownEffect(OrchestrationV2DomainEventJson); const encodeV2EventJson = Schema.encodeEffect(OrchestrationV2DomainEventJson); const decodeV2StoredEvent = Schema.decodeUnknownEffect(OrchestrationV2StoredEvent); +// Rows carry an event decodeV2EventJson already validated; revalidating it nearly doubled replay cost. +const decodeV2StoredEventHeader = Schema.decodeUnknownEffect( + OrchestrationV2StoredEvent.mapFields(({ event: _event, ...fields }) => fields), +); const decodeJson = Schema.decodeUnknownEffect(Schema.fromJsonString(Schema.Unknown)); const encodeJson = Schema.encodeEffect(Schema.fromJsonString(Schema.Unknown)); +// Same result as decodeJson; only malformed input pays for the schema's error. +const parseJson = (json: string) => { + if (typeof json !== "string") return decodeJson(json); + try { + return Effect.succeed(JSON.parse(json)); + } catch { + return decodeJson(json); + } +}; function metadataForV2Event(event: OrchestrationV2DomainEvent): Record { return { @@ -189,11 +203,10 @@ function metadataForV2Event(event: OrchestrationV2DomainEvent): Record) : {}; const event = yield* decodeV2EventJson({ @@ -210,16 +223,14 @@ const rowToV2StoredEvent = Effect.fn("OrchestrationEventStore.rowToV2StoredEvent : { providerInstanceId: values.providerInstanceId }), ...(values.rawEventId === undefined ? {} : { rawEventId: values.rawEventId }), }); - return yield* decodeV2StoredEvent({ + const { sequence, commandId } = yield* decodeV2StoredEventHeader({ sequence: row.sequence, commandId: row.command_id, - event, }); + return { sequence, commandId, event }; }); -const rowToProjectEvent = Effect.fn("OrchestrationEventStore.rowToProjectEvent")(function* ( - row: ApplicationEventRow, -) { +const rowToProjectEvent = Effect.fnUntraced(function* (row: ApplicationEventRow) { return yield* decodeProjectEvent({ sequence: row.sequence, eventId: row.event_id, @@ -230,8 +241,8 @@ const rowToProjectEvent = Effect.fn("OrchestrationEventStore.rowToProjectEvent") commandId: row.command_id, causationEventId: row.causation_event_id, correlationId: row.correlation_id, - payload: yield* decodeJson(row.payload_json), - metadata: yield* decodeJson(row.metadata_json), + payload: yield* parseJson(row.payload_json), + metadata: yield* parseJson(row.metadata_json), }); }); @@ -243,11 +254,12 @@ function rowToApplicationStoredEvent( const isKnownProjectEventType = Schema.is(ProjectEventType); -const keepReplayRow = (row: ApplicationEventRow, skipUnknownEventTypes?: boolean) => - skipUnknownEventTypes !== true || - (row.aggregate_kind === "project" - ? isKnownProjectEventType(row.event_type) - : isKnownOrchestrationV2EventType(row.event_type)) +const keepKnownEventTypeRow = (row: ApplicationEventRow) => + ( + row.aggregate_kind === "project" + ? isKnownProjectEventType(row.event_type) + : isKnownOrchestrationV2EventType(row.event_type) + ) ? Effect.succeed(true) : Effect.logWarning("Skipping an application event row with an unknown event type", { aggregateKind: row.aggregate_kind, @@ -574,7 +586,9 @@ const makeEventStore = Effect.gen(function* () { ); }, ).pipe( - Stream.filterEffect((row) => keepReplayRow(row, input?.skipUnknownEventTypes)), + // Query the next page on the reader while this one decodes. + Stream.bufferArray({ capacity: 1 }), + input?.skipUnknownEventTypes === true ? Stream.filterEffect(keepKnownEventTypeRow) : identity, Stream.mapEffect((row) => rowToV2StoredEvent(row).pipe( Effect.mapError( @@ -699,7 +713,9 @@ const makeEventStore = Effect.gen(function* () { }), ), ).pipe( - Stream.filterEffect((row) => keepReplayRow(row, input.skipUnknownEventTypes)), + // Query the next page on the reader while this one decodes. + Stream.bufferArray({ capacity: 1 }), + input.skipUnknownEventTypes === true ? Stream.filterEffect(keepKnownEventTypeRow) : identity, Stream.mapEffect((row) => rowToApplicationStoredEvent(row).pipe( Effect.mapError( From b9291c27fad20330a259eb8670266d6cb9437849 Mon Sep 17 00:00:00 2001 From: Hieu Le Date: Thu, 29 Oct 2026 00:25:53 +0100 Subject: [PATCH 2/2] perf(server): keep one span per event replay Replay no longer traces each row, so wrap each replay stream in a single span instead. Co-Authored-By: Claude Sonnet 5.5 --- apps/server/src/persistence/OrchestrationEventStore.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/apps/server/src/persistence/OrchestrationEventStore.ts b/apps/server/src/persistence/OrchestrationEventStore.ts index 8c531bdf8b3d2..129ce850f08f8 100644 --- a/apps/server/src/persistence/OrchestrationEventStore.ts +++ b/apps/server/src/persistence/OrchestrationEventStore.ts @@ -596,6 +596,7 @@ const makeEventStore = Effect.gen(function* () { ), ), ), + Stream.withSpan("OrchestrationEventStore.readAgentEvents"), ); }; @@ -723,6 +724,7 @@ const makeEventStore = Effect.gen(function* () { ), ), ), + Stream.withSpan("OrchestrationEventStore.readApplicationEvents"), ); const streamProjectedApplicationEvents: OrchestrationEventStore["Service"]["streamProjectedApplicationEvents"] =