| author |  Julius Marminge <julius0216@outlook.com> 2026-02-25 20:59:21-08:00 |
| committer |  Julius Marminge <jmarminge@gmail.com> 2026-02-26 13:20:36-08:00 |
| log | 9e2324c2e567d75f37b305fbaf53b545178db5b0 |
| tree | 57ece138ec7edadddfddb106263e86b7a4e5f7e6 |
| parent | 5a08b9dd64e5f13a9ad88c910e9f34cb7598199a |
reconcile on failure
3 files changed, 141 insertions(+), 6 deletions(-)
apps
server
src
orchestration
.plans/16c-pr89-remediation-checklist.md+3-3
| ... | @@ -11,7 +11,7 @@ Status values: | ... | @@ -11,7 +11,7 @@ Status values: |
| 11 | - `DONE`: Implemented and verified | 11 | - `DONE`: Implemented and verified |
| 12 | - `CLOSED_INVALID`: Stale/invalid review finding | 12 | - `CLOSED_INVALID`: Stale/invalid review finding |
| 13 | 13 | ||
| 14 | Counts: active `53` (`valid=34`, `partially-valid=19`), closed-invalid `5` | 14 | Counts: active `52` (`valid=33`, `partially-valid=19`), closed-invalid `5` |
| 15 | 15 | ||
| 16 | ## Active Checklist | 16 | ## Active Checklist |
| 17 | 17 | ||
| ... | @@ -163,8 +163,8 @@ Counts: active `53` (`valid=34`, `partially-valid=19`), closed-invalid `5` | ... | @@ -163,8 +163,8 @@ Counts: active `53` (`valid=34`, `partially-valid=19`), closed-invalid `5` |
| 163 | - Threads: PRRT_kwDORLtfbc5wzp4T | 163 | - Threads: PRRT_kwDORLtfbc5wzp4T |
| 164 | - Audit note: Append/project/receipt are non-atomic; retry can duplicate events. | 164 | - Audit note: Append/project/receipt are non-atomic; retry can duplicate events. |
| 165 | 165 | ||
| 166 | - [ ] `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. | 166 | - [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. |
| 167 | - Status: `TODO` | 167 | - Status: `DONE` |
| 168 | - Verdict: `valid` | 168 | - Verdict: `valid` |
| 169 | - Severity: `Medium` | 169 | - Severity: `Medium` |
| 170 | - Area: `Event ordering and state consistency` | 170 | - Area: `Event ordering and state consistency` |
apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts+101
| ... | @@ -467,6 +467,107 @@ describe("OrchestrationEngine", () => { | ... | @@ -467,6 +467,107 @@ describe("OrchestrationEngine", () => { |
| 467 | await runtime.dispose(); | 467 | await runtime.dispose(); |
| 468 | }); | 468 | }); |
| 469 | 469 | ||
| 470 | it("reconciles in-memory state when append persists but projection fails", async () => { | ||
| 471 | type StoredEvent = | ||
| 472 | ReturnType<OrchestrationEventStoreShape["append"]> extends Effect.Effect<infer A, any, any> | ||
| 473 | ? A | ||
| 474 | : never; | ||
| 475 | const events: StoredEvent[] = []; | ||
| 476 | let nextSequence = 1; | ||
| 477 | |||
| 478 | const nonTransactionalStore: OrchestrationEventStoreShape = { | ||
| 479 | append(event) { | ||
| 480 | const savedEvent = { | ||
| 481 | ...event, | ||
| 482 | sequence: nextSequence, | ||
| 483 | } as StoredEvent; | ||
| 484 | nextSequence += 1; | ||
| 485 | events.push(savedEvent); | ||
| 486 | return Effect.succeed(savedEvent); | ||
| 487 | }, | ||
| 488 | readFromSequence(sequenceExclusive) { | ||
| 489 | return Stream.fromIterable(events.filter((event) => event.sequence > sequenceExclusive)); | ||
| 490 | }, | ||
| 491 | readAll() { | ||
| 492 | return Stream.fromIterable(events); | ||
| 493 | }, | ||
| 494 | }; | ||
| 495 | |||
| 496 | let shouldFailProjection = true; | ||
| 497 | const flakyProjectionPipeline: OrchestrationProjectionPipelineShape = { | ||
| 498 | bootstrap: Effect.void, | ||
| 499 | projectEvent: (event) => { | ||
| 500 | if ( | ||
| 501 | shouldFailProjection && | ||
| 502 | event.commandId === CommandId.makeUnsafe("cmd-thread-meta-sync-fail") | ||
| 503 | ) { | ||
| 504 | shouldFailProjection = false; | ||
| 505 | return Effect.fail( | ||
| 506 | new PersistenceSqlError({ | ||
| 507 | operation: "test.projection", | ||
| 508 | detail: "projection failed", | ||
| 509 | }), | ||
| 510 | ); | ||
| 511 | } | ||
| 512 | return Effect.void; | ||
| 513 | }, | ||
| 514 | }; | ||
| 515 | |||
| 516 | const runtime = ManagedRuntime.make( | ||
| 517 | OrchestrationEngineLive.pipe( | ||
| 518 | Layer.provide(Layer.succeed(OrchestrationProjectionPipeline, flakyProjectionPipeline)), | ||
| 519 | Layer.provide(Layer.succeed(OrchestrationEventStore, nonTransactionalStore)), | ||
| 520 | Layer.provide(OrchestrationCommandReceiptRepositoryLive), | ||
| 521 | Layer.provide(SqlitePersistenceMemory), | ||
| 522 | ), | ||
| 523 | ); | ||
| 524 | const engine = await runtime.runPromise(Effect.service(OrchestrationEngineService)); | ||
| 525 | const createdAt = now(); | ||
| 526 | |||
| 527 | await runtime.runPromise( | ||
| 528 | engine.dispatch({ | ||
| 529 | type: "project.create", | ||
| 530 | commandId: CommandId.makeUnsafe("cmd-project-sync-create"), | ||
| 531 | projectId: asProjectId("project-sync"), | ||
| 532 | title: "Sync Project", | ||
| 533 | workspaceRoot: "/tmp/project-sync", | ||
| 534 | defaultModel: "gpt-5-codex", | ||
| 535 | createdAt, | ||
| 536 | }), | ||
| 537 | ); | ||
| 538 | await runtime.runPromise( | ||
| 539 | engine.dispatch({ | ||
| 540 | type: "thread.create", | ||
| 541 | commandId: CommandId.makeUnsafe("cmd-thread-sync-create"), | ||
| 542 | threadId: ThreadId.makeUnsafe("thread-sync"), | ||
| 543 | projectId: asProjectId("project-sync"), | ||
| 544 | title: "sync-before", | ||
| 545 | model: "gpt-5-codex", | ||
| 546 | branch: null, | ||
| 547 | worktreePath: null, | ||
| 548 | createdAt, | ||
| 549 | }), | ||
| 550 | ); | ||
| 551 | |||
| 552 | await expect( | ||
| 553 | runtime.runPromise( | ||
| 554 | engine.dispatch({ | ||
| 555 | type: "thread.meta.update", | ||
| 556 | commandId: CommandId.makeUnsafe("cmd-thread-meta-sync-fail"), | ||
| 557 | threadId: ThreadId.makeUnsafe("thread-sync"), | ||
| 558 | title: "sync-after-failed-projection", | ||
| 559 | }), | ||
| 560 | ), | ||
| 561 | ).rejects.toThrow("projection failed"); | ||
| 562 | |||
| 563 | const readModelAfterFailure = await runtime.runPromise(engine.getReadModel()); | ||
| 564 | const updatedThread = readModelAfterFailure.threads.find((thread) => thread.id === "thread-sync"); | ||
| 565 | expect(readModelAfterFailure.snapshotSequence).toBe(3); | ||
| 566 | expect(updatedThread?.title).toBe("sync-after-failed-projection"); | ||
| 567 | |||
| 568 | await runtime.dispose(); | ||
| 569 | }); | ||
| 570 | |||
| 470 | it("fails command dispatch when command invariants are violated", async () => { | 571 | it("fails command dispatch when command invariants are violated", async () => { |
| 471 | const system = await createOrchestrationSystem(); | 572 | const system = await createOrchestrationSystem(); |
| 472 | const { engine } = system; | 573 | const { engine } = system; |
apps/server/src/orchestration/Layers/OrchestrationEngine.ts+37-3
| ... | @@ -67,8 +67,28 @@ const makeOrchestrationEngine = Effect.gen(function* () { | ... | @@ -67,8 +67,28 @@ const makeOrchestrationEngine = Effect.gen(function* () { |
| 67 | const commandQueue = yield* Queue.unbounded<CommandEnvelope>(); | 67 | const commandQueue = yield* Queue.unbounded<CommandEnvelope>(); |
| 68 | const eventPubSub = yield* PubSub.unbounded<OrchestrationEvent>(); | 68 | const eventPubSub = yield* PubSub.unbounded<OrchestrationEvent>(); |
| 69 | 69 | ||
| 70 | const processEnvelope = (envelope: CommandEnvelope): Effect.Effect<void> => | 70 | const processEnvelope = (envelope: CommandEnvelope): Effect.Effect<void> => { |
| 71 | Effect.gen(function* () { | 71 | const dispatchStartSequence = readModel.snapshotSequence; |
| 72 | const reconcileReadModelAfterDispatchFailure = Effect.gen(function* () { | ||
| 73 | const persistedEvents = yield* Stream.runCollect( | ||
| 74 | eventStore.readFromSequence(dispatchStartSequence), | ||
| 75 | ).pipe(Effect.map((chunk): OrchestrationEvent[] => Array.from(chunk))); | ||
| 76 | if (persistedEvents.length === 0) { | ||
| 77 | return; | ||
| 78 | } | ||
| 79 | |||
| 80 | let nextReadModel = readModel; | ||
| 81 | for (const persistedEvent of persistedEvents) { | ||
| 82 | nextReadModel = yield* projectEvent(nextReadModel, persistedEvent); | ||
| 83 | } | ||
| 84 | readModel = nextReadModel; | ||
| 85 | |||
| 86 | for (const persistedEvent of persistedEvents) { | ||
| 87 | yield* PubSub.publish(eventPubSub, persistedEvent); | ||
| 88 | } | ||
| 89 | }); | ||
| 90 | |||
| 91 | return Effect.gen(function* () { | ||
| 72 | const existingReceipt = yield* commandReceiptRepository.getByCommandId({ | 92 | const existingReceipt = yield* commandReceiptRepository.getByCommandId({ |
| 73 | commandId: envelope.command.commandId, | 93 | commandId: envelope.command.commandId, |
| 74 | }); | 94 | }); |
| ... | @@ -102,8 +122,8 @@ const makeOrchestrationEngine = Effect.gen(function* () { | ... | @@ -102,8 +122,8 @@ const makeOrchestrationEngine = Effect.gen(function* () { |
| 102 | 122 | ||
| 103 | for (const nextEvent of eventBases) { | 123 | for (const nextEvent of eventBases) { |
| 104 | const savedEvent = yield* eventStore.append(nextEvent); | 124 | const savedEvent = yield* eventStore.append(nextEvent); |
| 105 | yield* projectionPipeline.projectEvent(savedEvent); | ||
| 106 | nextReadModel = yield* projectEvent(nextReadModel, savedEvent); | 125 | nextReadModel = yield* projectEvent(nextReadModel, savedEvent); |
| 126 | yield* projectionPipeline.projectEvent(savedEvent); | ||
| 107 | committedEvents.push(savedEvent); | 127 | committedEvents.push(savedEvent); |
| 108 | } | 128 | } |
| 109 | 129 | ||
| ... | @@ -148,6 +168,19 @@ const makeOrchestrationEngine = Effect.gen(function* () { | ... | @@ -148,6 +168,19 @@ const makeOrchestrationEngine = Effect.gen(function* () { |
| 148 | }).pipe( | 168 | }).pipe( |
| 149 | Effect.catch((error) => | 169 | Effect.catch((error) => |
| 150 | Effect.gen(function* () { | 170 | Effect.gen(function* () { |
| 171 | yield* reconcileReadModelAfterDispatchFailure.pipe( | ||
| 172 | Effect.catch(() => | ||
| 173 | Effect.logWarning( | ||
| 174 | "failed to reconcile orchestration read model after dispatch failure", | ||
| 175 | ).pipe( | ||
| 176 | Effect.annotateLogs({ | ||
| 177 | commandId: envelope.command.commandId, | ||
| 178 | snapshotSequence: readModel.snapshotSequence, | ||
| 179 | }), | ||
| 180 | ), | ||
| 181 | ), | ||
| 182 | ); | ||
| 183 | |||
| 151 | if (error instanceof OrchestrationCommandInvariantError) { | 184 | if (error instanceof OrchestrationCommandInvariantError) { |
| 152 | const aggregateRef = commandToAggregateRef(envelope.command); | 185 | const aggregateRef = commandToAggregateRef(envelope.command); |
| 153 | yield* commandReceiptRepository | 186 | yield* commandReceiptRepository |
| ... | @@ -166,6 +199,7 @@ const makeOrchestrationEngine = Effect.gen(function* () { | ... | @@ -166,6 +199,7 @@ const makeOrchestrationEngine = Effect.gen(function* () { |
| 166 | }), | 199 | }), |
| 167 | ), | 200 | ), |
| 168 | ); | 201 | ); |
| 202 | }; | ||
| 169 | 203 | ||
| 170 | yield* projectionPipeline.bootstrap; | 204 | yield* projectionPipeline.bootstrap; |
| 171 | 205 |