From 8ea8e6e26fa47e37cb4ff2862f081c62d6ae2343 Mon Sep 17 00:00:00 2001 From: hided62 Date: Fri, 31 Jul 2026 14:10:11 +0000 Subject: [PATCH] fix(engine): roll back scheduled turns when flush fails --- .../src/lifecycle/turnDaemonLifecycle.ts | 33 ++- .../test/turnDaemonLifecycle.test.ts | 205 ++++++++++++++++++ 2 files changed, 218 insertions(+), 20 deletions(-) diff --git a/app/game-engine/src/lifecycle/turnDaemonLifecycle.ts b/app/game-engine/src/lifecycle/turnDaemonLifecycle.ts index 6698114..d8e2e1f 100644 --- a/app/game-engine/src/lifecycle/turnDaemonLifecycle.ts +++ b/app/game-engine/src/lifecycle/turnDaemonLifecycle.ts @@ -315,9 +315,7 @@ export class TurnDaemonLifecycle { await this.commandResponder.publishCommandError(command.requestId, error); } catch (reportError) { const reportMessage = - reportError instanceof Error - ? reportError.message - : 'Unknown command failure reporting error.'; + reportError instanceof Error ? reportError.message : 'Unknown command failure reporting error.'; this.status.lastError = `${this.status.lastError} (failure report: ${reportMessage})`; } } @@ -347,36 +345,31 @@ export class TurnDaemonLifecycle { const budget = pending.budget ?? this.options.defaultBudget; const checkpoint = this.status.checkpoint; let result: TurnRunResult; + let fallbackError = 'Unknown turn daemon error.'; try { - const runProcessor = () => this.processor.run(targetTime, budget, checkpoint); - result = this.stateManager ? await this.stateManager.transaction(runProcessor) : await runProcessor(); + const runAndFlush = async (): Promise => { + const nextResult = await this.processor.run(targetTime, budget, checkpoint); + fallbackError = 'Unknown turn flush error.'; + this.status.state = 'flushing'; + await this.stateStore.saveLastTurnTime(new Date(nextResult.lastTurnTime)); + await this.stateStore.saveCheckpoint(nextResult.checkpoint); + await this.hooks?.flushChanges?.(nextResult); + return nextResult; + }; + result = this.stateManager ? await this.stateManager.transaction(runAndFlush) : await runAndFlush(); } catch (error) { this.status.running = false; this.status.state = 'paused'; this.status.paused = true; this.errorPaused = true; - this.status.lastError = error instanceof Error ? error.message : 'Unknown turn daemon error.'; + this.status.lastError = error instanceof Error ? error.message : fallbackError; await this.hooks?.onRunError?.(error); return; } finally { this.status.running = false; } - this.status.state = 'flushing'; - try { - await this.stateStore.saveLastTurnTime(new Date(result.lastTurnTime)); - await this.stateStore.saveCheckpoint(result.checkpoint); - await this.hooks?.flushChanges?.(result); - } catch (error) { - this.status.state = 'paused'; - this.status.paused = true; - this.errorPaused = true; - this.status.lastError = error instanceof Error ? error.message : 'Unknown turn flush error.'; - await this.hooks?.onRunError?.(error); - return; - } - await this.applyRunResult(result, startMs); this.status.state = 'idle'; try { diff --git a/app/game-engine/test/turnDaemonLifecycle.test.ts b/app/game-engine/test/turnDaemonLifecycle.test.ts index 1577f87..e464256 100644 --- a/app/game-engine/test/turnDaemonLifecycle.test.ts +++ b/app/game-engine/test/turnDaemonLifecycle.test.ts @@ -75,6 +75,211 @@ describe('TurnDaemonLifecycle', () => { await loop; }); + it('restores processor and state-store mutations when scheduled flush fails', async () => { + const now = new Date('2026-01-01T00:10:00.000Z'); + const previousCheckpoint = { + turnTime: '2026-01-01T00:00:00.000Z', + generalId: 7, + year: 203, + month: 1, + }; + const nextCheckpoint = { + turnTime: now.toISOString(), + generalId: 8, + year: 203, + month: 2, + }; + let engineState: { + value: string; + lastTurnTime: string; + checkpoint: TurnRunResult['checkpoint']; + } = { + value: 'before', + lastTurnTime: previousCheckpoint.turnTime, + checkpoint: previousCheckpoint, + }; + const stateManager = new EngineStateManager(); + stateManager.register('test', { + capture: () => structuredClone(engineState), + restore: (snapshot) => { + engineState = snapshot; + }, + }); + const published = vi.fn(); + let resolveError: (() => void) | undefined; + const errorObserved = new Promise((resolve) => { + resolveError = resolve; + }); + const lifecycle = new TurnDaemonLifecycle( + { + clock: new ManualClock(now.getTime()), + controlQueue: new InMemoryControlQueue(), + getNextTickTime: () => new Date('2026-01-01T00:05:00.000Z'), + stateStore: { + loadLastTurnTime: async () => new Date(engineState.lastTurnTime), + loadNextGeneralTurnTime: async () => null, + saveLastTurnTime: async (turnTime) => { + engineState.lastTurnTime = turnTime.toISOString(); + }, + loadCheckpoint: async () => engineState.checkpoint, + saveCheckpoint: async (checkpoint) => { + engineState.checkpoint = checkpoint; + }, + }, + processor: { + run: async (): Promise => { + engineState.value = 'calculated'; + return { + lastTurnTime: now.toISOString(), + processedGenerals: 1, + processedTurns: 1, + durationMs: 0, + partial: false, + checkpoint: nextCheckpoint, + }; + }, + }, + stateManager, + hooks: { + flushChanges: async () => { + expect(engineState).toMatchObject({ + value: 'calculated', + lastTurnTime: now.toISOString(), + checkpoint: nextCheckpoint, + }); + throw new Error('scheduled flush failed'); + }, + publishEvents: published, + onRunError: async () => { + resolveError?.(); + }, + }, + }, + { + profile: 'test', + defaultBudget: { budgetMs: 100, maxGenerals: 1, catchUpCap: 1 }, + } + ); + + const loop = lifecycle.start(); + await errorObserved; + + expect(engineState).toEqual({ + value: 'before', + lastTurnTime: previousCheckpoint.turnTime, + checkpoint: previousCheckpoint, + }); + expect(stateManager.getRevision()).toBe(0); + expect(published).not.toHaveBeenCalled(); + expect(lifecycle.getStatus()).toMatchObject({ + state: 'paused', + paused: true, + lastError: 'scheduled flush failed', + }); + + await lifecycle.stop('done'); + await loop; + }); + + it('keeps processor and state-store mutations after flush succeeds and publishes afterward', async () => { + const now = new Date('2026-01-01T00:10:00.000Z'); + const previousCheckpoint = { + turnTime: '2026-01-01T00:00:00.000Z', + generalId: 7, + year: 203, + month: 1, + }; + const nextCheckpoint = { + turnTime: now.toISOString(), + generalId: 8, + year: 203, + month: 2, + }; + let engineState: { + value: string; + lastTurnTime: string; + checkpoint: TurnRunResult['checkpoint']; + } = { + value: 'before', + lastTurnTime: previousCheckpoint.turnTime, + checkpoint: previousCheckpoint, + }; + const stateManager = new EngineStateManager(); + stateManager.register('test', { + capture: () => structuredClone(engineState), + restore: (snapshot) => { + engineState = snapshot; + }, + }); + const callOrder: string[] = []; + let resolvePublished: (() => void) | undefined; + const published = new Promise((resolve) => { + resolvePublished = resolve; + }); + const lifecycle = new TurnDaemonLifecycle( + { + clock: new ManualClock(now.getTime()), + controlQueue: new InMemoryControlQueue(), + getNextTickTime: (lastTurnTime) => addMinutes(lastTurnTime, 10), + stateStore: { + loadLastTurnTime: async () => new Date(engineState.lastTurnTime), + loadNextGeneralTurnTime: async () => null, + saveLastTurnTime: async (turnTime) => { + callOrder.push('save-last-turn'); + engineState.lastTurnTime = turnTime.toISOString(); + }, + loadCheckpoint: async () => engineState.checkpoint, + saveCheckpoint: async (checkpoint) => { + callOrder.push('save-checkpoint'); + engineState.checkpoint = checkpoint; + }, + }, + processor: { + run: async (): Promise => { + callOrder.push('processor'); + engineState.value = 'calculated'; + return { + lastTurnTime: now.toISOString(), + processedGenerals: 1, + processedTurns: 1, + durationMs: 0, + partial: false, + checkpoint: nextCheckpoint, + }; + }, + }, + stateManager, + hooks: { + flushChanges: async () => { + callOrder.push('flush'); + }, + publishEvents: async () => { + callOrder.push('publish'); + resolvePublished?.(); + }, + }, + }, + { + profile: 'test', + defaultBudget: { budgetMs: 100, maxGenerals: 1, catchUpCap: 1 }, + } + ); + + const loop = lifecycle.start(); + await published; + + expect(engineState).toEqual({ + value: 'calculated', + lastTurnTime: now.toISOString(), + checkpoint: nextCheckpoint, + }); + expect(stateManager.getRevision()).toBe(1); + expect(callOrder).toEqual(['processor', 'save-last-turn', 'save-checkpoint', 'flush', 'publish']); + + await lifecycle.stop('done'); + await loop; + }); + it('runs scheduled turn based on queue front and checkpoint context', async () => { const turnTermMinutes = 10; const lastTurnTime = new Date(2026, 0, 2, 2, 0, 0, 0);