diff --git a/app/game-engine/src/lifecycle/databaseCommandQueue.ts b/app/game-engine/src/lifecycle/databaseCommandQueue.ts index 2243d139..c4936c12 100644 --- a/app/game-engine/src/lifecycle/databaseCommandQueue.ts +++ b/app/game-engine/src/lifecycle/databaseCommandQueue.ts @@ -171,7 +171,8 @@ export class DatabaseTurnDaemonCommandQueue implements TurnDaemonControlQueue, T 'changePermission', 'appoint', 'setNationSetting', - 'setNpcPolicy' + 'setNpcPolicy', + 'shiftSchedule' ) ) OR ( diff --git a/app/game-engine/src/turn/clockReconciliation.ts b/app/game-engine/src/turn/clockReconciliation.ts index 607ef1a6..2d3248c2 100644 --- a/app/game-engine/src/turn/clockReconciliation.ts +++ b/app/game-engine/src/turn/clockReconciliation.ts @@ -113,6 +113,41 @@ const readDbWall = async (db: GamePrisma.TransactionClient): Promise => { export const readClockDatabaseWall = readDbWall; +const isRetryableSerializableClockError = (error: unknown): boolean => { + if (!error || typeof error !== 'object') return false; + const value = error as { code?: unknown; message?: unknown; meta?: unknown }; + const meta = + value.meta && typeof value.meta === 'object' + ? (value.meta as { code?: unknown; message?: unknown }) + : undefined; + const code = typeof value.code === 'string' ? value.code : ''; + const databaseCode = typeof meta?.code === 'string' ? meta.code : ''; + const message = [value.message, meta?.message] + .filter((entry): entry is string => typeof entry === 'string') + .join(' '); + return ( + code === 'P2034' || + code === '40001' || + databaseCode === '40001' || + message.includes('40001') || + message.includes('could not serialize access') + ); +}; + +const runSerializableClockOperation = async (operation: () => Promise): Promise => { + const maxAttempts = 3; + for (let attempt = 1; ; attempt += 1) { + try { + return await operation(); + } catch (error) { + if (attempt >= maxAttempts || !isRetryableSerializableClockError(error)) { + throw error; + } + await new Promise((resolve) => setTimeout(resolve, attempt * 10)); + } + } +}; + const verifyAuthority = async (db: GamePrisma.TransactionClient, authority: ClockOperationAuthority): Promise => { const rows = await db.$queryRaw(GamePrisma.sql` SELECT owner_id AS "ownerId", @@ -185,64 +220,64 @@ const readParticipantSnapshots = async ( ): Promise => { const [world, generals, auctions, auctionBids, messages, inheritanceEffects, votes, pool, npcTokens, commands] = await Promise.all([ - db.worldState.findUniqueOrThrow({ - where: { id: worldStateId }, - select: { - clockTick: true, - clockRevision: true, - deadlineGeneration: true, - lastTurnTick: true, - meta: true, - }, - }), - db.general.findMany({ - orderBy: { id: 'asc' }, - select: { id: true, turnTick: true, recentWarTick: true, meta: true }, - }), - db.auction.findMany({ - where: { status: { in: ['OPEN', 'FINALIZING'] } }, - orderBy: { id: 'asc' }, - select: { id: true, status: true, openTick: true, closeTick: true }, - }), - db.auctionBid.findMany({ - where: { auction: { status: { in: ['OPEN', 'FINALIZING'] } } }, - orderBy: { id: 'asc' }, - select: { id: true, occurredGameTick: true }, - }), - db.messageAction.findMany({ - where: { status: 'PENDING' }, - orderBy: { messageId: 'asc' }, - select: { - messageId: true, - createdGameTick: true, - expiresGameTick: true, - clockRevision: true, - deadlineGeneration: true, - }, - }), - db.inheritanceLedger.findMany({ - orderBy: { id: 'asc' }, - select: { id: true, appliedClockRevision: true, appliedDeadlineGeneration: true }, - }), - db.votePoll.findMany({ - where: { closedAt: null }, - orderBy: { id: 'asc' }, - select: { id: true, startTick: true, endTick: true }, - }), - db.selectPoolEntry.findMany({ - where: { generalId: null }, - orderBy: { id: 'asc' }, - select: { id: true, reservedUntilTick: true }, - }), - db.npcSelectionToken.findMany({ - orderBy: { ownerUserId: 'asc' }, - select: { ownerUserId: true, validUntilTick: true, pickMoreFromTick: true }, - }), - db.inputEvent.findMany({ - where: { status: { in: ['PENDING', 'PROCESSING'] } }, - orderBy: { sequence: 'asc' }, - select: { sequence: true, acceptedGameTick: true, acceptedClockRevision: true }, - }), + db.worldState.findUniqueOrThrow({ + where: { id: worldStateId }, + select: { + clockTick: true, + clockRevision: true, + deadlineGeneration: true, + lastTurnTick: true, + meta: true, + }, + }), + db.general.findMany({ + orderBy: { id: 'asc' }, + select: { id: true, turnTick: true, recentWarTick: true, meta: true }, + }), + db.auction.findMany({ + where: { status: { in: ['OPEN', 'FINALIZING'] } }, + orderBy: { id: 'asc' }, + select: { id: true, status: true, openTick: true, closeTick: true }, + }), + db.auctionBid.findMany({ + where: { auction: { status: { in: ['OPEN', 'FINALIZING'] } } }, + orderBy: { id: 'asc' }, + select: { id: true, occurredGameTick: true }, + }), + db.messageAction.findMany({ + where: { status: 'PENDING' }, + orderBy: { messageId: 'asc' }, + select: { + messageId: true, + createdGameTick: true, + expiresGameTick: true, + clockRevision: true, + deadlineGeneration: true, + }, + }), + db.inheritanceLedger.findMany({ + orderBy: { id: 'asc' }, + select: { id: true, appliedClockRevision: true, appliedDeadlineGeneration: true }, + }), + db.votePoll.findMany({ + where: { closedAt: null }, + orderBy: { id: 'asc' }, + select: { id: true, startTick: true, endTick: true }, + }), + db.selectPoolEntry.findMany({ + where: { generalId: null }, + orderBy: { id: 'asc' }, + select: { id: true, reservedUntilTick: true }, + }), + db.npcSelectionToken.findMany({ + orderBy: { ownerUserId: 'asc' }, + select: { ownerUserId: true, validUntilTick: true, pickMoreFromTick: true }, + }), + db.inputEvent.findMany({ + where: { status: { in: ['PENDING', 'PROCESSING'] } }, + orderBy: { sequence: 'asc' }, + select: { sequence: true, acceptedGameTick: true, acceptedClockRevision: true }, + }), ]); const snapshot = (key: string, policy: ParticipantSnapshot['policy'], rows: unknown[]): ParticipantSnapshot => ({ key, @@ -686,90 +721,97 @@ export const startClockSuspension = async (options: { if (!Number.isSafeInteger(catchUpTicks) || catchUpTicks < 0) { throw new Error('Clock suspension catch-up ticks must be a non-negative safe integer.'); } - return options.db.$transaction( - async (db) => { - await verifyAuthority(db, options.authority); - await acquireGameSchemaAdvisoryXactLock(db, CLOCK_OPERATION_PERSISTENCE_LOCK); - await acquireGameSchemaAdvisoryXactLock(db, GENERAL_ACCESS_PERSISTENCE_LOCK); - const worldStateId = await lockWorld(db); - const existing = await db.clockSuspension.findUnique({ where: { id: options.suspensionId } }); - if (existing) { - if ( - existing.worldStateId !== worldStateId || - existing.source !== options.source || - existing.policy !== policy - ) { - throw new Error( - `Clock suspension ID ${options.suspensionId} is already bound to another operation.` - ); + return runSerializableClockOperation(() => + options.db.$transaction( + async (db) => { + await verifyAuthority(db, options.authority); + await acquireGameSchemaAdvisoryXactLock(db, CLOCK_OPERATION_PERSISTENCE_LOCK); + await acquireGameSchemaAdvisoryXactLock(db, GENERAL_ACCESS_PERSISTENCE_LOCK); + const worldStateId = await lockWorld(db); + const existing = await db.clockSuspension.findUnique({ where: { id: options.suspensionId } }); + if (existing) { + if ( + existing.worldStateId !== worldStateId || + existing.source !== options.source || + existing.policy !== policy + ) { + throw new Error( + `Clock suspension ID ${options.suspensionId} is already bound to another operation.` + ); + } + if (existing.status !== 'SUSPENDED') { + throw new Error( + `Clock suspension ${options.suspensionId} already advanced to ${existing.status}.` + ); + } + return { + suspensionId: existing.id, + phase: 'SUSPENDED' as const, + sourceRevision: safeNumber(existing.sourceRevision, 'source revision'), + targetRevision: safeNumber(existing.targetRevision, 'target revision'), + cutTick: safeNumber(existing.cutTick, 'cut tick'), + cutWallAt: existing.cutWallAt, + }; } - if (existing.status !== 'SUSPENDED') { - throw new Error(`Clock suspension ${options.suspensionId} already advanced to ${existing.status}.`); + const world = await db.worldState.findUniqueOrThrow({ where: { id: worldStateId } }); + const phase = parseGameClockPhase(world.clockPhase); + if (phase !== 'RUNNING') { + throw new Error(`Clock suspension can start only from RUNNING; current phase is ${phase}.`); } + if (!world.clockBaseTime || world.clockTick === null || !world.clockWallAnchor) { + throw new Error('Clock suspension requires a fully initialized logical game clock.'); + } + const cutWallAt = await readDbWall(db); + const storedTick = safeNumber(world.clockTick, 'world clock tick'); + const sourceRevision = safeNumber(world.clockRevision, 'world clock revision'); + const clock = new GameClock({ + baseTime: world.clockBaseTime, + tick: storedTick, + mode: world.clockMode === 'manual' ? 'manual' : 'realtime', + wallAnchor: world.clockWallAnchor, + turnSeconds: world.tickSeconds, + phase, + revision: sourceRevision, + }); + const cutTick = clock.nowTick(cutWallAt); + await lockParticipants(db, BigInt(cutTick)); + await db.worldState.update({ + where: { id: worldStateId }, + data: { clockPhase: 'SUSPENDED', clockTick: BigInt(cutTick), clockWallAnchor: cutWallAt }, + }); + const participants = await readParticipantSnapshots(db, worldStateId, BigInt(cutTick)); + await db.clockSuspension.create({ + data: { + id: options.suspensionId, + worldStateId, + source: options.source, + policy, + status: 'SUSPENDED', + sourceRevision: BigInt(sourceRevision), + targetRevision: BigInt(sourceRevision + 1), + cutTick: BigInt(cutTick), + cutWallAt, + rateTicksPerSecond: GAME_TICKS_PER_TURN / world.tickSeconds, + catchUpTicks: BigInt(catchUpTicks), + participantChecksumBefore: aggregateChecksum(participants), + detail: asJson({ + authority: options.authority.kind, + profileName: options.authority.profileName, + }), + }, + }); + await persistInitialParticipants(db, options.suspensionId, participants); return { - suspensionId: existing.id, - phase: 'SUSPENDED' as const, - sourceRevision: safeNumber(existing.sourceRevision, 'source revision'), - targetRevision: safeNumber(existing.targetRevision, 'target revision'), - cutTick: safeNumber(existing.cutTick, 'cut tick'), - cutWallAt: existing.cutWallAt, - }; - } - const world = await db.worldState.findUniqueOrThrow({ where: { id: worldStateId } }); - const phase = parseGameClockPhase(world.clockPhase); - if (phase !== 'RUNNING') { - throw new Error(`Clock suspension can start only from RUNNING; current phase is ${phase}.`); - } - if (!world.clockBaseTime || world.clockTick === null || !world.clockWallAnchor) { - throw new Error('Clock suspension requires a fully initialized logical game clock.'); - } - const cutWallAt = await readDbWall(db); - const storedTick = safeNumber(world.clockTick, 'world clock tick'); - const sourceRevision = safeNumber(world.clockRevision, 'world clock revision'); - const clock = new GameClock({ - baseTime: world.clockBaseTime, - tick: storedTick, - mode: world.clockMode === 'manual' ? 'manual' : 'realtime', - wallAnchor: world.clockWallAnchor, - turnSeconds: world.tickSeconds, - phase, - revision: sourceRevision, - }); - const cutTick = clock.nowTick(cutWallAt); - await lockParticipants(db, BigInt(cutTick)); - await db.worldState.update({ - where: { id: worldStateId }, - data: { clockPhase: 'SUSPENDED', clockTick: BigInt(cutTick), clockWallAnchor: cutWallAt }, - }); - const participants = await readParticipantSnapshots(db, worldStateId, BigInt(cutTick)); - await db.clockSuspension.create({ - data: { - id: options.suspensionId, - worldStateId, - source: options.source, - policy, - status: 'SUSPENDED', - sourceRevision: BigInt(sourceRevision), - targetRevision: BigInt(sourceRevision + 1), - cutTick: BigInt(cutTick), + suspensionId: options.suspensionId, + phase: 'SUSPENDED', + sourceRevision, + targetRevision: sourceRevision + 1, + cutTick, cutWallAt, - rateTicksPerSecond: GAME_TICKS_PER_TURN / world.tickSeconds, - catchUpTicks: BigInt(catchUpTicks), - participantChecksumBefore: aggregateChecksum(participants), - detail: asJson({ authority: options.authority.kind, profileName: options.authority.profileName }), - }, - }); - await persistInitialParticipants(db, options.suspensionId, participants); - return { - suspensionId: options.suspensionId, - phase: 'SUSPENDED', - sourceRevision, - targetRevision: sourceRevision + 1, - cutTick, - cutWallAt, - }; - }, - { isolationLevel: 'Serializable', maxWait: 10_000, timeout: 30_000 } + }; + }, + { isolationLevel: 'Serializable', maxWait: 10_000, timeout: 30_000 } + ) ); }; @@ -983,18 +1025,20 @@ export const reconcileClockSuspension = async (options: { /** Deterministic fixture seam; production must always use PostgreSQL CURRENT_TIMESTAMP. */ testResumeWallAt?: Date; }): Promise => - options.db.$transaction( - async (db) => { - await verifyAuthority(db, options.authority); - await acquireGameSchemaAdvisoryXactLock(db, CLOCK_OPERATION_PERSISTENCE_LOCK); - await acquireGameSchemaAdvisoryXactLock(db, GENERAL_ACCESS_PERSISTENCE_LOCK); - return reconcileClockSuspensionInTransaction({ - db, - suspensionId: options.suspensionId, - profileName: options.authority.profileName, - authority: options.authority, - ...(options.testResumeWallAt ? { testResumeWallAt: options.testResumeWallAt } : {}), - }); - }, - { isolationLevel: 'Serializable', maxWait: 10_000, timeout: 30_000 } + runSerializableClockOperation(() => + options.db.$transaction( + async (db) => { + await verifyAuthority(db, options.authority); + await acquireGameSchemaAdvisoryXactLock(db, CLOCK_OPERATION_PERSISTENCE_LOCK); + await acquireGameSchemaAdvisoryXactLock(db, GENERAL_ACCESS_PERSISTENCE_LOCK); + return reconcileClockSuspensionInTransaction({ + db, + suspensionId: options.suspensionId, + profileName: options.authority.profileName, + authority: options.authority, + ...(options.testResumeWallAt ? { testResumeWallAt: options.testResumeWallAt } : {}), + }); + }, + { isolationLevel: 'Serializable', maxWait: 10_000, timeout: 30_000 } + ) ); diff --git a/app/game-engine/src/turn/databaseHooks.ts b/app/game-engine/src/turn/databaseHooks.ts index 7fa31c39..e6adbdd3 100644 --- a/app/game-engine/src/turn/databaseHooks.ts +++ b/app/game-engine/src/turn/databaseHooks.ts @@ -64,6 +64,7 @@ import { refreshClockProjectionForFinalClockUnderHeldLocks, } from './clockReconciliation.js'; import { applyNextClockProjection, type ClockProjectionRedis } from './clockProjectionOutbox.js'; +import { synchronizeRuntimeClockAuthorityUnderHeldLock } from './runtimeClockAuthoritySync.js'; export interface DatabaseTurnHooks { hooks: TurnDaemonHooks; @@ -71,6 +72,7 @@ export interface DatabaseTurnHooks { takeCommittedReadModelChangeReceipt(): CommittedReadModelChangeReceipt | null; close(): Promise; applyClockProjection(redis: ClockProjectionRedis, workerId: string): Promise; + synchronizeClockAuthority(): Promise; } export interface CommittedReadModelChangeReceipt { @@ -2010,6 +2012,7 @@ export const createDatabaseTurnHooks = async ( await options?.turnDaemonLease?.assertActive(transaction); await acquireGameSchemaAdvisoryXactLock(transaction, CLOCK_OPERATION_PERSISTENCE_LOCK); await acquireGameSchemaAdvisoryXactLock(transaction, GENERAL_ACCESS_PERSISTENCE_LOCK); + await synchronizeRuntimeClockAuthorityUnderHeldLock(transaction, world); const leaseToken = options?.turnDaemonLease?.getToken(); const directLogFloor = ( @@ -2054,6 +2057,12 @@ export const createDatabaseTurnHooks = async ( }); return clock?.clockPhase === 'RUNNING' || clock?.clockPhase === 'MANUAL'; }, + synchronizeClockAuthority: () => + prisma.$transaction(async (transaction) => { + await options?.turnDaemonLease?.assertActive(transaction); + await acquireGameSchemaAdvisoryXactLock(transaction, CLOCK_OPERATION_PERSISTENCE_LOCK); + return synchronizeRuntimeClockAuthorityUnderHeldLock(transaction, world); + }, transactionOptions), close: () => connector.disconnect(), }; }; diff --git a/app/game-engine/src/turn/inMemoryWorld.ts b/app/game-engine/src/turn/inMemoryWorld.ts index e99dbe73..1ac2ea93 100644 --- a/app/game-engine/src/turn/inMemoryWorld.ts +++ b/app/game-engine/src/turn/inMemoryWorld.ts @@ -135,6 +135,20 @@ export interface InMemoryGameClockState { deadlineGeneration: number; } +export interface DurableGameClockSnapshot extends InMemoryGameClockState { + lastTurnTick: number; +} + +export interface DurableClockReconciliationAlignment { + suspensionId: string; + sourceRevision: number; + targetRevision: number; + deadlineGeneration: number; + alignedTick: number; + shiftTicks: number; + resumeWallAt: Date; +} + export type InheritancePersistencePhase = 'before_lifecycle' | 'after_lifecycle'; export interface PendingInheritancePointAdjustment { @@ -723,23 +737,14 @@ export class InMemoryTurnWorld { this.state = { ...this.state, clockPhase: 'COMPLETED' }; } - applyClockReconciliation(input: { - suspensionId: string; - alignedTick: number; - shiftTicks: number; - targetRevision: number; - deadlineGeneration: number; - resumeWallAt: Date; - }): void { - const phase = this.state.clockPhase ?? inferClockPhase(this.state.clockMode ?? 'manual'); - if (phase !== 'SUSPENDED' || this.state.meta.unificationClockSuspensionId !== input.suspensionId) { - throw new Error('In-memory clock reconciliation requires the matching UNIFICATION_WAIT suspension.'); - } + private applyClockReconciliationAlignment(input: DurableClockReconciliationAlignment): void { if ( !Number.isSafeInteger(input.alignedTick) || !Number.isSafeInteger(input.shiftTicks) || input.shiftTicks < 0 || + !Number.isSafeInteger(input.sourceRevision) || !Number.isSafeInteger(input.targetRevision) || + input.targetRevision !== input.sourceRevision + 1 || !Number.isSafeInteger(input.deadlineGeneration) ) { throw new Error('In-memory clock reconciliation received an unsafe coordinate.'); @@ -800,6 +805,72 @@ export class InMemoryTurnWorld { } } + applyClockReconciliation(input: Omit): void { + const phase = this.state.clockPhase ?? inferClockPhase(this.state.clockMode ?? 'manual'); + if (phase !== 'SUSPENDED' || this.state.meta.unificationClockSuspensionId !== input.suspensionId) { + throw new Error('In-memory clock reconciliation requires the matching UNIFICATION_WAIT suspension.'); + } + this.applyClockReconciliationAlignment({ + ...input, + sourceRevision: this.state.clockRevision ?? 1, + }); + } + + applyDurableClockReconciliation(input: DurableClockReconciliationAlignment): void { + const clock = this.getGameClockState(); + if (clock.revision === input.targetRevision && clock.phase === 'RECONCILING') { + return; + } + if (clock.revision !== input.sourceRevision) { + throw new Error( + `Durable clock reconciliation source mismatch: memory ${clock.revision}, ledger ${input.sourceRevision}.` + ); + } + if (clock.phase !== 'RUNNING' && clock.phase !== 'SUSPENDED') { + throw new Error(`Durable clock reconciliation cannot apply from in-memory phase ${clock.phase}.`); + } + this.applyClockReconciliationAlignment(input); + } + + synchronizeDurableClockSnapshot(input: DurableGameClockSnapshot): void { + const current = this.getGameClockState(); + if (current.revision !== input.revision || current.deadlineGeneration !== input.deadlineGeneration) { + throw new Error( + `Durable clock snapshot generation mismatch: memory ${current.revision}/${current.deadlineGeneration}, ` + + `database ${input.revision}/${input.deadlineGeneration}.` + ); + } + if ((this.state.lastTurnTick ?? 0) !== input.lastTurnTick) { + throw new Error( + `Durable clock snapshot turn cursor mismatch: memory ${this.state.lastTurnTick ?? 0}, database ${input.lastTurnTick}.` + ); + } + const currentBaseTime = this.state.clockBaseTime ?? this.state.lastTurnTime; + if (currentBaseTime.getTime() !== input.baseTime.getTime()) { + throw new Error( + `Durable clock snapshot base mismatch: memory ${currentBaseTime.toISOString()}, ` + + `database ${input.baseTime.toISOString()}.` + ); + } + const validTransition = + current.phase === input.phase || + (current.phase === 'RUNNING' && input.phase === 'SUSPENDED') || + (current.phase === 'RECONCILING' && input.phase === 'RUNNING'); + if (!validTransition) { + throw new Error(`Durable clock snapshot phase mismatch: memory ${current.phase}, database ${input.phase}.`); + } + this.state = { + ...this.state, + clockBaseTime: new Date(input.baseTime.getTime()), + clockTick: input.tick, + clockMode: input.mode, + clockWallAnchor: new Date(input.wallAnchor.getTime()), + clockPhase: input.phase, + clockRevision: input.revision, + deadlineGeneration: input.deadlineGeneration, + }; + } + completeClockReconciliation(): void { if (this.state.clockPhase === 'RECONCILING') { this.state = { ...this.state, clockPhase: 'RUNNING' }; diff --git a/app/game-engine/src/turn/runtimeClockAuthoritySync.ts b/app/game-engine/src/turn/runtimeClockAuthoritySync.ts new file mode 100644 index 00000000..d3d96647 --- /dev/null +++ b/app/game-engine/src/turn/runtimeClockAuthoritySync.ts @@ -0,0 +1,125 @@ +import { parseGameClockPhase } from '@sammo-ts/common'; +import type { GamePrisma } from '@sammo-ts/infra'; + +import type { InMemoryTurnWorld } from './inMemoryWorld.js'; + +const safeNumber = (value: bigint, label: string): number => { + const result = Number(value); + if (!Number.isSafeInteger(result)) { + throw new Error(`${label} is outside the JavaScript safe integer range: ${value}.`); + } + return result; +}; + +/** + * Replays durable suspension ledgers into the already-running daemon before it + * handles a command or resumes scheduled turns. The caller must hold the clock + * operation advisory lock so the world row and ledger chain are one snapshot. + */ +export const synchronizeRuntimeClockAuthorityUnderHeldLock = async ( + db: GamePrisma.TransactionClient, + world: InMemoryTurnWorld +): Promise => { + const durable = await db.worldState.findFirst({ + orderBy: { id: 'asc' }, + select: { + id: true, + clockBaseTime: true, + clockTick: true, + clockMode: true, + clockWallAnchor: true, + lastTurnTick: true, + clockPhase: true, + clockRevision: true, + deadlineGeneration: true, + }, + }); + if ( + !durable || + !durable.clockBaseTime || + durable.clockTick === null || + !durable.clockWallAnchor || + durable.lastTurnTick === null + ) { + throw new Error('Runtime clock synchronization requires one fully initialized world clock.'); + } + + const before = world.getGameClockState(); + const durableRevision = safeNumber(durable.clockRevision, 'durable clock revision'); + const durableGeneration = safeNumber(durable.deadlineGeneration, 'durable deadline generation'); + if (before.revision > durableRevision) { + throw new Error(`In-memory clock revision ${before.revision} is ahead of durable revision ${durableRevision}.`); + } + + if (before.revision < durableRevision) { + const ledgers = await db.clockSuspension.findMany({ + where: { + worldStateId: durable.id, + sourceRevision: { gte: BigInt(before.revision) }, + targetRevision: { lte: durable.clockRevision }, + status: { in: ['RECONCILING', 'APPLIED'] }, + }, + orderBy: { sourceRevision: 'asc' }, + select: { + id: true, + sourceRevision: true, + targetRevision: true, + shiftTicks: true, + alignedTick: true, + resumeWallAt: true, + }, + }); + let expectedRevision = before.revision; + let expectedGeneration = before.deadlineGeneration; + for (const ledger of ledgers) { + const sourceRevision = safeNumber(ledger.sourceRevision, `clock suspension ${ledger.id} source revision`); + const targetRevision = safeNumber(ledger.targetRevision, `clock suspension ${ledger.id} target revision`); + if (sourceRevision !== expectedRevision || targetRevision !== sourceRevision + 1) { + throw new Error( + `Clock suspension ledger chain is discontinuous at ${ledger.id}: ` + + `expected ${expectedRevision}->${expectedRevision + 1}, found ${sourceRevision}->${targetRevision}.` + ); + } + if (ledger.shiftTicks === null || ledger.alignedTick === null || !ledger.resumeWallAt) { + throw new Error(`Clock suspension ${ledger.id} has no completed reconciliation coordinate.`); + } + expectedGeneration += 1; + world.applyDurableClockReconciliation({ + suspensionId: ledger.id, + sourceRevision, + targetRevision, + deadlineGeneration: expectedGeneration, + alignedTick: safeNumber(ledger.alignedTick, `clock suspension ${ledger.id} aligned tick`), + shiftTicks: safeNumber(ledger.shiftTicks, `clock suspension ${ledger.id} shift ticks`), + resumeWallAt: ledger.resumeWallAt, + }); + expectedRevision = targetRevision; + } + if (expectedRevision !== durableRevision || expectedGeneration !== durableGeneration) { + throw new Error( + `Clock suspension ledger chain ended at ${expectedRevision}/${expectedGeneration}, ` + + `but durable clock is ${durableRevision}/${durableGeneration}.` + ); + } + } + + world.synchronizeDurableClockSnapshot({ + baseTime: durable.clockBaseTime, + tick: safeNumber(durable.clockTick, 'durable clock tick'), + mode: durable.clockMode === 'manual' ? 'manual' : 'realtime', + wallAnchor: durable.clockWallAnchor, + lastTurnTick: safeNumber(durable.lastTurnTick, 'durable last turn tick'), + phase: parseGameClockPhase(durable.clockPhase), + revision: durableRevision, + deadlineGeneration: durableGeneration, + }); + + const after = world.getGameClockState(); + return ( + before.phase !== after.phase || + before.revision !== after.revision || + before.deadlineGeneration !== after.deadlineGeneration || + before.tick !== after.tick || + before.wallAnchor.getTime() !== after.wallAnchor.getTime() + ); +}; diff --git a/app/game-engine/src/turn/turnDaemon.ts b/app/game-engine/src/turn/turnDaemon.ts index 665ce898..d4b44cb2 100644 --- a/app/game-engine/src/turn/turnDaemon.ts +++ b/app/game-engine/src/turn/turnDaemon.ts @@ -715,6 +715,7 @@ const createTurnDaemonRuntimeWithLease = async ( let redisConnector: RedisConnector | null = null; let stopClockProjectionWorker = () => {}; let applyClockProjection: DatabaseTurnHooks['applyClockProjection'] | undefined; + let synchronizeClockAuthority: DatabaseTurnHooks['synchronizeClockAuthority'] | undefined; const nationTraits = await loadNationTraitModules([...NATION_TRAIT_KEYS], new NationTraitLoader()); const nationTraitMap = new Map(nationTraits.map((module) => [module.key, module])); const monthlyActionModules = await loadActionModuleBundle( @@ -920,6 +921,7 @@ const createTurnDaemonRuntimeWithLease = async ( }; takeCommittedReadModelChangeReceipt = dbHooks.takeCommittedReadModelChangeReceipt; applyClockProjection = dbHooks.applyClockProjection; + synchronizeClockAuthority = dbHooks.synchronizeClockAuthority; close = async () => { if (auctionBidder) { await auctionBidder.close(); @@ -1031,6 +1033,7 @@ const createTurnDaemonRuntimeWithLease = async ( maxGenerals: 200, catchUpCap: 1, }; + let lastObservedGatewayPause: boolean | null = null; const lifecycle = new TurnDaemonLifecycle( { @@ -1041,7 +1044,21 @@ const createTurnDaemonRuntimeWithLease = async ( stateStore, processor, hooks, - pauseGate: async () => turnDaemonLease?.isLost() || ((await pauseGate?.()) ?? false), + pauseGate: async () => { + if (turnDaemonLease?.isLost()) { + return true; + } + const gatewayPaused = (await pauseGate?.()) ?? false; + const phase = world.getGameClockState().phase; + const phaseNeedsSync = gatewayPaused + ? phase !== 'SUSPENDED' + : phase === 'SUSPENDED' || phase === 'RECONCILING'; + if (synchronizeClockAuthority && (lastObservedGatewayPause !== gatewayPaused || phaseNeedsSync)) { + await synchronizeClockAuthority(); + } + lastObservedGatewayPause = gatewayPaused; + return gatewayPaused; + }, commandHandler, commandResponder: options.controlQueue ? undefined : (databaseCommandQueue ?? undefined), // The exclusive fixture runner aborts the entire in-memory runtime diff --git a/app/game-engine/test/clockReconciliationRetry.test.ts b/app/game-engine/test/clockReconciliationRetry.test.ts new file mode 100644 index 00000000..971b6f25 --- /dev/null +++ b/app/game-engine/test/clockReconciliationRetry.test.ts @@ -0,0 +1,71 @@ +import { describe, expect, it, vi } from 'vitest'; + +import type { GamePrismaClient } from '@sammo-ts/infra'; +import { + reconcileClockSuspension, + startClockSuspension, + type ClockReconciliationResult, + type ClockSuspensionResult, +} from '../src/turn/clockReconciliation.js'; + +const authority = { kind: 'OFFLINE' as const, profileName: 'retry-test', reason: 'isolated unit test' }; + +describe('serializable clock operation retries', () => { + it('retries a PostgreSQL serialization conflict while starting suspension', async () => { + const result: ClockSuspensionResult = { + suspensionId: 'retry-start', + phase: 'SUSPENDED', + sourceRevision: 7, + targetRevision: 8, + cutTick: 123, + cutWallAt: new Date('2026-09-03T10:00:00.000Z'), + }; + const transaction = vi + .fn() + .mockRejectedValueOnce(Object.assign(new Error('could not serialize access'), { code: 'P2034' })) + .mockResolvedValueOnce(result); + const db = { $transaction: transaction } as unknown as GamePrismaClient; + + await expect( + startClockSuspension({ db, suspensionId: result.suspensionId, source: 'MAINTENANCE', authority }) + ).resolves.toEqual(result); + expect(transaction).toHaveBeenCalledTimes(2); + }); + + it('retries a raw 40001 conflict while reconciling suspension', async () => { + const result: ClockReconciliationResult = { + suspensionId: 'retry-resume', + phase: 'RECONCILING', + sourceRevision: 7, + targetRevision: 8, + deadlineGeneration: 8, + gapTicks: 100, + catchUpTicks: 0, + shiftTicks: 100, + alignedTick: 223, + resumeWallAt: new Date('2026-09-03T10:10:00.000Z'), + }; + const transaction = vi + .fn() + .mockRejectedValueOnce( + Object.assign(new Error('SQLSTATE 40001'), { code: 'P2010', meta: { code: '40001' } }) + ) + .mockResolvedValueOnce(result); + const db = { $transaction: transaction } as unknown as GamePrismaClient; + + await expect(reconcileClockSuspension({ db, suspensionId: result.suspensionId, authority })).resolves.toEqual( + result + ); + expect(transaction).toHaveBeenCalledTimes(2); + }); + + it('does not retry a non-serialization failure', async () => { + const transaction = vi.fn().mockRejectedValue(new Error('authority denied')); + const db = { $transaction: transaction } as unknown as GamePrismaClient; + + await expect( + startClockSuspension({ db, suspensionId: 'no-retry', source: 'MAINTENANCE', authority }) + ).rejects.toThrow('authority denied'); + expect(transaction).toHaveBeenCalledTimes(1); + }); +}); diff --git a/app/game-engine/test/databaseCommandQueue.integration.test.ts b/app/game-engine/test/databaseCommandQueue.integration.test.ts index 058c2924..a5bf9f0c 100644 --- a/app/game-engine/test/databaseCommandQueue.integration.test.ts +++ b/app/game-engine/test/databaseCommandQueue.integration.test.ts @@ -608,6 +608,12 @@ integration('database command queue', () => { expectedUpdatedAt: null, mutation: { kind: 'nationPriority', priority: ['develop'] }, }, + { + type: 'shiftSchedule', + requestId: 'integration:engine:suspended-shift-schedule', + actionId: '00000000-0000-4000-8000-000000000023', + deltaMinutes: -15, + }, ]; await db.inputEvent.createMany({ data: commands.map((command) => ({ @@ -637,14 +643,18 @@ integration('database command queue', () => { const claimed = await new DatabaseTurnDaemonCommandQueue(db).drain(); expect(claimed.map(({ type }) => type)).toEqual(commands.map(({ type }) => type)); for (const command of commands) { - await expect(db.inputEvent.findUniqueOrThrow({ where: { requestId: command.requestId! } })).resolves.toMatchObject({ + await expect( + db.inputEvent.findUniqueOrThrow({ where: { requestId: command.requestId! } }) + ).resolves.toMatchObject({ status: 'PROCESSING', processingGameTick: 777n, processingClockRevision: 23n, processingDeadlineGeneration: 9n, }); } - await expect(db.inputEvent.findUniqueOrThrow({ where: { requestId: blockedRequestId } })).resolves.toMatchObject({ + await expect( + db.inputEvent.findUniqueOrThrow({ where: { requestId: blockedRequestId } }) + ).resolves.toMatchObject({ status: 'PENDING', processingClockRevision: null, }); diff --git a/app/game-engine/test/runtimeClockAuthoritySync.test.ts b/app/game-engine/test/runtimeClockAuthoritySync.test.ts new file mode 100644 index 00000000..b7eb9122 --- /dev/null +++ b/app/game-engine/test/runtimeClockAuthoritySync.test.ts @@ -0,0 +1,174 @@ +import { describe, expect, it, vi } from 'vitest'; + +import type { GamePrisma } from '@sammo-ts/infra'; +import { InMemoryTurnWorld } from '../src/turn/inMemoryWorld.js'; +import { synchronizeRuntimeClockAuthorityUnderHeldLock } from '../src/turn/runtimeClockAuthoritySync.js'; +import type { TurnGeneral, TurnWorldSnapshot, TurnWorldState } from '../src/turn/types.js'; + +const baseTime = new Date('2026-09-03T10:00:00.000Z'); + +const buildWorld = (phase: 'RUNNING' | 'SUSPENDED' = 'RUNNING'): InMemoryTurnWorld => { + const general = { + id: 1, + name: 'clock-sync-general', + nationId: 1, + cityId: 1, + troopId: 0, + stats: { leadership: 50, strength: 50, intelligence: 50 }, + turnTime: new Date('2026-09-03T10:10:00.000Z'), + turnTick: 36_000_000, + role: { items: { horse: null, weapon: null, book: null, item: null } }, + triggerState: { flags: {}, counters: {}, modifiers: {}, meta: {} }, + meta: { killturn: 24 }, + officerLevel: 5, + experience: 0, + dedication: 0, + injury: 0, + gold: 1000, + rice: 1000, + crew: 0, + crewTypeId: 0, + train: 0, + atmos: 0, + age: 30, + npcState: 0, + } as TurnGeneral; + const state: TurnWorldState = { + id: 1, + currentYear: 190, + currentMonth: 1, + tickSeconds: 600, + lastTurnTime: baseTime, + clockBaseTime: baseTime, + clockTick: 0, + clockMode: 'realtime', + clockWallAnchor: baseTime, + lastTurnTick: 0, + clockPhase: phase, + clockRevision: 3, + deadlineGeneration: 5, + meta: { lastTurnTime: baseTime.toISOString() }, + }; + const snapshot: TurnWorldSnapshot = { + generals: [general], + cities: [], + nations: [], + troops: [], + diplomacy: [], + events: [], + initialEvents: [], + map: { + id: 'clock-sync', + name: 'clock-sync', + cities: [], + defaults: { trust: 50, trade: 100, supplyState: 1, frontState: 0 }, + }, + scenarioConfig: { + stat: { total: 300, min: 10, max: 100, npcTotal: 150, npcMax: 50, npcMin: 10, chiefMin: 70 }, + iconPath: '', + map: {}, + const: {}, + environment: { mapName: 'test', unitSet: 'default' }, + }, + }; + return new InMemoryTurnWorld(state, snapshot, { + schedule: { entries: [{ startMinute: 0, tickMinutes: 10 }] }, + }); +}; + +const buildDb = (worldState: Record, ledgers: unknown[] = []): GamePrisma.TransactionClient => + ({ + worldState: { findFirst: vi.fn().mockResolvedValue(worldState) }, + clockSuspension: { findMany: vi.fn().mockResolvedValue(ledgers) }, + }) as unknown as GamePrisma.TransactionClient; + +describe('runtime clock authority synchronization', () => { + it('adopts a maintenance suspension cut without advancing the game schedule', async () => { + const world = buildWorld('RUNNING'); + const cutWallAt = new Date('2026-09-03T10:02:00.000Z'); + const beforeTurnTick = world.getGeneralById(1)!.turnTick; + const db = buildDb({ + id: 1, + clockBaseTime: baseTime, + clockTick: 7_200_000n, + clockMode: 'realtime', + clockWallAnchor: cutWallAt, + lastTurnTick: 0n, + clockPhase: 'SUSPENDED', + clockRevision: 3n, + deadlineGeneration: 5n, + }); + + await expect(synchronizeRuntimeClockAuthorityUnderHeldLock(db, world)).resolves.toBe(true); + + expect(world.getGameClockState()).toMatchObject({ + phase: 'SUSPENDED', + tick: 7_200_000, + revision: 3, + deadlineGeneration: 5, + wallAnchor: cutWallAt, + }); + expect(world.getGeneralById(1)!.turnTick).toBe(beforeTurnTick); + }); + + it('replays the durable reconciliation shift before returning to RUNNING', async () => { + const world = buildWorld('SUSPENDED'); + const resumeWallAt = new Date('2026-09-03T11:00:00.000Z'); + const shiftTicks = 5_000; + const beforeTurnTick = world.getGeneralById(1)!.turnTick!; + const db = buildDb( + { + id: 1, + clockBaseTime: baseTime, + clockTick: 5_000n, + clockMode: 'realtime', + clockWallAnchor: resumeWallAt, + lastTurnTick: 5_000n, + clockPhase: 'RUNNING', + clockRevision: 4n, + deadlineGeneration: 6n, + }, + [ + { + id: 'maintenance-revision-3', + sourceRevision: 3n, + targetRevision: 4n, + shiftTicks: BigInt(shiftTicks), + alignedTick: 5_000n, + resumeWallAt, + }, + ] + ); + + await expect(synchronizeRuntimeClockAuthorityUnderHeldLock(db, world)).resolves.toBe(true); + + expect(world.getGameClockState()).toMatchObject({ + phase: 'RUNNING', + tick: 5_000, + lastTurnTick: 5_000, + revision: 4, + deadlineGeneration: 6, + wallAnchor: resumeWallAt, + }); + expect(world.getGeneralById(1)!.turnTick).toBe(beforeTurnTick + shiftTicks); + }); + + it('rejects a revision jump when the durable ledger chain is incomplete', async () => { + const world = buildWorld('SUSPENDED'); + const db = buildDb({ + id: 1, + clockBaseTime: baseTime, + clockTick: 1n, + clockMode: 'realtime', + clockWallAnchor: baseTime, + lastTurnTick: 1n, + clockPhase: 'RUNNING', + clockRevision: 4n, + deadlineGeneration: 6n, + }); + + await expect(synchronizeRuntimeClockAuthorityUnderHeldLock(db, world)).rejects.toThrow( + /ledger chain ended at 3\/5/ + ); + }); +}); diff --git a/tools/integration-tests/scripts/live-ten-user-lifecycle.ts b/tools/integration-tests/scripts/live-ten-user-lifecycle.ts index d89737ef..96be7a0f 100644 --- a/tools/integration-tests/scripts/live-ten-user-lifecycle.ts +++ b/tools/integration-tests/scripts/live-ten-user-lifecycle.ts @@ -8,7 +8,7 @@ import { chromium, type BrowserContext, type Page } from '@playwright/test'; import { createTRPCProxyClient, httpBatchLink } from '@trpc/client'; import type { AppRouter as GatewayAppRouter } from '@sammo-ts/gateway-api'; import type { AppRouter as GameAppRouter } from '@sammo-ts/game-api'; -import { createGamePostgresConnector, createRedisConnector } from '@sammo-ts/infra'; +import { createGamePostgresConnector, createRedisConnector, type GamePrisma } from '@sammo-ts/infra'; import { createTurnDaemonRuntime } from '../../../app/game-engine/src/turn/turnDaemon.js'; const gatewayUrl = process.env.SAMMO_LIVE_GATEWAY_URL ?? 'http://caddy/gateway/api/trpc'; @@ -923,6 +923,64 @@ const prepareActionFixture = async (): Promise => { const rotateFirst = (values: readonly T[]): T[] => (values.length < 2 ? [...values] : [...values.slice(1), values[0]!]); +const readHiddenBuffLevel = (rawMeta: unknown, key: string): number => { + if (!rawMeta || typeof rawMeta !== 'object' || Array.isArray(rawMeta)) return 0; + const rawBuff = Reflect.get(rawMeta, 'inheritBuff'); + let buff: unknown = rawBuff; + if (typeof rawBuff === 'string') { + try { + buff = JSON.parse(rawBuff) as unknown; + } catch { + return 0; + } + } + if (!buff || typeof buff !== 'object' || Array.isArray(buff)) return 0; + const value = Reflect.get(buff, key); + return typeof value === 'number' && Number.isFinite(value) ? value : 0; +}; + +const repairActionItemFixture = async (): Promise => { + const state = await readState(); + if (!state.actionFixture) throw new Error('The action fixture is required.'); + const gateway = await loginAdmin(); + const profiles = (await gateway.admin.profiles.list.query()) as unknown as Array<{ + profileName: string; + status: string; + }>; + const profile = profiles.find((entry) => entry.profileName === profileName); + if (profile?.status !== 'STOPPED') { + throw new Error(`Item fixture repair requires STOPPED, found ${profile?.status ?? 'missing'}.`); + } + const db = createGamePostgresConnector({ url: gameDatabaseUrl() }); + await db.connect(); + try { + const targetGeneralId = state.actionFixture.generals[2]!.id; + const repaired = await db.prisma.$transaction(async (tx) => { + const world = await tx.worldState.findFirstOrThrow({ orderBy: { id: 'asc' } }); + if (world.clockPhase !== 'SUSPENDED') { + throw new Error(`Item fixture repair requires SUSPENDED, found ${world.clockPhase}.`); + } + const general = await tx.general.findUniqueOrThrow({ where: { id: targetGeneralId } }); + const meta = + general.meta && typeof general.meta === 'object' && !Array.isArray(general.meta) + ? { ...general.meta } + : {}; + delete meta.itemInventory; + return tx.general.update({ + where: { id: targetGeneralId }, + data: { + weaponCode: 'che_무기_01_단도', + meta: meta as GamePrisma.InputJsonValue, + }, + select: { id: true, weaponCode: true }, + }); + }); + log('action-item-fixture-repaired', repaired); + } finally { + await db.disconnect(); + } +}; + const exercisePausedActions = async (): Promise => { const state = await readState(); if (state.users?.length !== 10 || !state.actionFixture) { @@ -986,8 +1044,20 @@ const exercisePausedActions = async (): Promise => { expectedRevision: turnSnapshot.revision, }); - const inheritance = await clients[3]!.inherit.buyHiddenBuff.mutate({ type: 'warAvoidRatio', level: 1 }); - const dropped = await clients[2]!.general.dropItem.mutate({ itemType: 'weapon' }); + const inheritanceRows = await db.prisma.general.findMany({ + where: { id: { in: generals.map(({ id }) => id) } }, + select: { id: true, meta: true }, + }); + const inheritanceById = new Map(inheritanceRows.map((general) => [general.id, general.meta])); + const inheritanceGeneralIndex = generals.findIndex( + (general, index) => + index >= 3 && readHiddenBuffLevel(inheritanceById.get(general.id), 'warAvoidRatio') < 1 + ); + if (inheritanceGeneralIndex < 0) throw new Error('No lifecycle user remains for the inheritance purchase.'); + const inheritance = await clients[inheritanceGeneralIndex]!.inherit.buyHiddenBuff.mutate({ + type: 'warAvoidRatio', + level: 1, + }); const permission = await clients[0]!.nation.changePermission.mutate({ isAmbassador: true, targetGeneralIds: [generals[2]!.id], @@ -1018,6 +1088,7 @@ const exercisePausedActions = async (): Promise => { detail: `SUSPENDED 상태의 WALL_TIME 외교문서 ${runSlug}`, }); const letterResponse = await clients[5]!.diplomacy.respondLetter.mutate({ letterId: letter.id, agree: true }); + const dropped = await clients[2]!.general.dropItem.mutate({ itemType: 'weapon' }); const [worldAfter, persistedGenerals, persistedMessages, persistedLetter, inheritanceLogs, pendingEvents] = await Promise.all([ @@ -2186,6 +2257,7 @@ else if (command === 'prepare-paused-betting') await preparePausedBetting(); else if (command === 'submit-paused-betting') await submitPausedBetting(); else if (command === 'reserve-user-enlistments') await reserveUserEnlistments(); else if (command === 'prepare-action-fixture') await prepareActionFixture(); +else if (command === 'repair-action-item-fixture') await repairActionItemFixture(); else if (command === 'exercise-paused-actions') await exercisePausedActions(); else if (command === 'verify-paused-wall-expiry') await verifyPausedWallExpiry(); else if (command === 'exercise-paused-tournament-bet') await exercisePausedTournamentBet(); @@ -2208,5 +2280,5 @@ else if (command === 'wait-runtime-action') await waitRuntimeAction(); else if (command === 'monitor-users') await monitorUsers(); else throw new Error( - 'usage: live-ten-user-lifecycle.ts ' + 'usage: live-ten-user-lifecycle.ts ' );