From 9e2324c2e567d75f37b305fbaf53b545178db5b0 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 25 Feb 2026 20:59:21 -0800 Subject: [PATCH] reconcile on failure --- .plans/16c-pr89-remediation-checklist.md | 6 +- .../Layers/OrchestrationEngine.test.ts | 101 ++++++++++++++++++ .../Layers/OrchestrationEngine.ts | 40 ++++++- 3 files changed, 141 insertions(+), 6 deletions(-) diff --git a/.plans/16c-pr89-remediation-checklist.md b/.plans/16c-pr89-remediation-checklist.md index 61b8c2c9c5b99848fb7fff9d563a6b40957d947c..ba7686dfd603dd2fb4a986c67ac7c3cf8c69dd5c 100644 --- a/.plans/16c-pr89-remediation-checklist.md +++ b/.plans/16c-pr89-remediation-checklist.md @@ -11,7 +11,7 @@ Status values: - `DONE`: Implemented and verified - `CLOSED_INVALID`: Stale/invalid review finding -Counts: active `53` (`valid=34`, `partially-valid=19`), closed-invalid `5` +Counts: active `52` (`valid=33`, `partially-valid=19`), closed-invalid `5` ## Active Checklist @@ -163,8 +163,8 @@ Counts: active `53` (`valid=34`, `partially-valid=19`), closed-invalid `5` - Threads: PRRT_kwDORLtfbc5wzp4T - Audit note: Append/project/receipt are non-atomic; retry can duplicate events. -- [ ] `C013` If `projectionPipeline.projectEvent` fails after `eventStore.append` succeeds, the event is persisted but `readModel` isn't updated, causing desync. Consider updating the in-memory `readModel` immediately after append (before the external projection), so local state stays consistent regardless of downstream failures. - - Status: `TODO` +- [x] `C013` If `projectionPipeline.projectEvent` fails after `eventStore.append` succeeds, the event is persisted but `readModel` isn't updated, causing desync. Consider updating the in-memory `readModel` immediately after append (before the external projection), so local state stays consistent regardless of downstream failures. + - Status: `DONE` - Verdict: `valid` - Severity: `Medium` - Area: `Event ordering and state consistency` diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts index e2c8ab35f5789e1a5281cce93367d7a310ada88c..81c46f07d810be9d0ecedc08508aa8ff12cf088e 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts @@ -467,6 +467,107 @@ describe("OrchestrationEngine", () => { await runtime.dispose(); }); + it("reconciles in-memory state when append persists but projection fails", async () => { + type StoredEvent = + ReturnType extends Effect.Effect + ? A + : never; + const events: StoredEvent[] = []; + let nextSequence = 1; + + const nonTransactionalStore: OrchestrationEventStoreShape = { + append(event) { + const savedEvent = { + ...event, + sequence: nextSequence, + } as StoredEvent; + nextSequence += 1; + events.push(savedEvent); + return Effect.succeed(savedEvent); + }, + readFromSequence(sequenceExclusive) { + return Stream.fromIterable(events.filter((event) => event.sequence > sequenceExclusive)); + }, + readAll() { + return Stream.fromIterable(events); + }, + }; + + let shouldFailProjection = true; + const flakyProjectionPipeline: OrchestrationProjectionPipelineShape = { + bootstrap: Effect.void, + projectEvent: (event) => { + if ( + shouldFailProjection && + event.commandId === CommandId.makeUnsafe("cmd-thread-meta-sync-fail") + ) { + shouldFailProjection = false; + return Effect.fail( + new PersistenceSqlError({ + operation: "test.projection", + detail: "projection failed", + }), + ); + } + return Effect.void; + }, + }; + + const runtime = ManagedRuntime.make( + OrchestrationEngineLive.pipe( + Layer.provide(Layer.succeed(OrchestrationProjectionPipeline, flakyProjectionPipeline)), + Layer.provide(Layer.succeed(OrchestrationEventStore, nonTransactionalStore)), + Layer.provide(OrchestrationCommandReceiptRepositoryLive), + Layer.provide(SqlitePersistenceMemory), + ), + ); + const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); + const createdAt = now(); + + await runtime.runPromise( + engine.dispatch({ + type: "project.create", + commandId: CommandId.makeUnsafe("cmd-project-sync-create"), + projectId: asProjectId("project-sync"), + title: "Sync Project", + workspaceRoot: "/tmp/project-sync", + defaultModel: "gpt-5-codex", + createdAt, + }), + ); + await runtime.runPromise( + engine.dispatch({ + type: "thread.create", + commandId: CommandId.makeUnsafe("cmd-thread-sync-create"), + threadId: ThreadId.makeUnsafe("thread-sync"), + projectId: asProjectId("project-sync"), + title: "sync-before", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt, + }), + ); + + await expect( + runtime.runPromise( + engine.dispatch({ + type: "thread.meta.update", + commandId: CommandId.makeUnsafe("cmd-thread-meta-sync-fail"), + threadId: ThreadId.makeUnsafe("thread-sync"), + title: "sync-after-failed-projection", + }), + ), + ).rejects.toThrow("projection failed"); + + const readModelAfterFailure = await runtime.runPromise(engine.getReadModel()); + const updatedThread = readModelAfterFailure.threads.find((thread) => thread.id === "thread-sync"); + expect(readModelAfterFailure.snapshotSequence).toBe(3); + expect(updatedThread?.title).toBe("sync-after-failed-projection"); + + await runtime.dispose(); + }); + it("fails command dispatch when command invariants are violated", async () => { const system = await createOrchestrationSystem(); const { engine } = system; diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts index ebe5547700892ad3deb86eacfaf83af5990f4a5e..ce48f1e1b38a6ba07f03aedfb3d231f56c2872c1 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts @@ -67,8 +67,28 @@ const makeOrchestrationEngine = Effect.gen(function* () { const commandQueue = yield* Queue.unbounded(); const eventPubSub = yield* PubSub.unbounded(); - const processEnvelope = (envelope: CommandEnvelope): Effect.Effect => - Effect.gen(function* () { + const processEnvelope = (envelope: CommandEnvelope): Effect.Effect => { + const dispatchStartSequence = readModel.snapshotSequence; + const reconcileReadModelAfterDispatchFailure = Effect.gen(function* () { + const persistedEvents = yield* Stream.runCollect( + eventStore.readFromSequence(dispatchStartSequence), + ).pipe(Effect.map((chunk): OrchestrationEvent[] => Array.from(chunk))); + if (persistedEvents.length === 0) { + return; + } + + let nextReadModel = readModel; + for (const persistedEvent of persistedEvents) { + nextReadModel = yield* projectEvent(nextReadModel, persistedEvent); + } + readModel = nextReadModel; + + for (const persistedEvent of persistedEvents) { + yield* PubSub.publish(eventPubSub, persistedEvent); + } + }); + + return Effect.gen(function* () { const existingReceipt = yield* commandReceiptRepository.getByCommandId({ commandId: envelope.command.commandId, }); @@ -102,8 +122,8 @@ const makeOrchestrationEngine = Effect.gen(function* () { for (const nextEvent of eventBases) { const savedEvent = yield* eventStore.append(nextEvent); - yield* projectionPipeline.projectEvent(savedEvent); nextReadModel = yield* projectEvent(nextReadModel, savedEvent); + yield* projectionPipeline.projectEvent(savedEvent); committedEvents.push(savedEvent); } @@ -148,6 +168,19 @@ const makeOrchestrationEngine = Effect.gen(function* () { }).pipe( Effect.catch((error) => Effect.gen(function* () { + yield* reconcileReadModelAfterDispatchFailure.pipe( + Effect.catch(() => + Effect.logWarning( + "failed to reconcile orchestration read model after dispatch failure", + ).pipe( + Effect.annotateLogs({ + commandId: envelope.command.commandId, + snapshotSequence: readModel.snapshotSequence, + }), + ), + ), + ); + if (error instanceof OrchestrationCommandInvariantError) { const aggregateRef = commandToAggregateRef(envelope.command); yield* commandReceiptRepository @@ -166,6 +199,7 @@ const makeOrchestrationEngine = Effect.gen(function* () { }), ), ); + }; yield* projectionPipeline.bootstrap; -- 2.54.0