From 69ac6028df4485f456dfad9b48009ae30acc47e4 Mon Sep 17 00:00:00 2001 From: hided62 Date: Thu, 3 Sep 2026 10:18:44 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20=ED=86=B5=EC=9D=BC=20=EB=8C=80=EA=B8=B0?= =?UTF-8?q?=20=EC=8B=9C=EA=B3=84=EB=A5=BC=20=EC=9B=90=EC=9E=90=EC=A0=81=20?= =?UTF-8?q?reconciliation=EC=9C=BC=EB=A1=9C=20=EC=A0=84=ED=99=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/game-engine/package.json | 8 + .../src/lifecycle/databaseCommandQueue.ts | 26 +- app/game-engine/src/lifecycle/types.ts | 6 + .../src/turn/actionableMessageResponse.ts | 41 ++ .../src/turn/clockProjectionOutbox.ts | 10 +- .../src/turn/clockReconciliation.ts | 471 +++++++++++------- app/game-engine/src/turn/databaseHooks.ts | 150 +++++- app/game-engine/src/turn/inMemoryWorld.ts | 110 ++++ .../src/turn/monthlyInvaderAction.ts | 3 +- app/game-engine/src/turn/turnDaemon.ts | 61 ++- .../src/turn/unificationHandler.ts | 10 + .../src/turn/worldCommandHandler.ts | 19 +- .../clockReconciliation.integration.test.ts | 125 ++++- .../databaseCommandQueue.integration.test.ts | 107 +++- ...nificationFinalization.integration.test.ts | 283 ++++++++++- .../test/unificationHandler.test.ts | 4 + .../test/unificationInvaderResume.test.ts | 34 +- .../src/orchestrator/gatewayOrchestrator.ts | 4 +- .../architecture/game-clock-participants.json | 15 +- .../architecture/game-clock-reconciliation.md | 27 +- .../game-clock-reconciliation-plan.md | 24 +- docs/developer/game-clock-recovery.md | 13 +- 22 files changed, 1300 insertions(+), 251 deletions(-) diff --git a/app/game-engine/package.json b/app/game-engine/package.json index 2201ab88..d2edfd97 100644 --- a/app/game-engine/package.json +++ b/app/game-engine/package.json @@ -42,6 +42,14 @@ "types": "./dist/turn/databaseHooks.d.ts", "default": "./dist/turn/databaseHooks.js" }, + "./turn/clockReconciliation.js": { + "types": "./dist/turn/clockReconciliation.d.ts", + "default": "./dist/turn/clockReconciliation.js" + }, + "./turn/clockProjectionOutbox.js": { + "types": "./dist/turn/clockProjectionOutbox.d.ts", + "default": "./dist/turn/clockProjectionOutbox.js" + }, "./turn/inMemoryWorld.js": { "types": "./dist/turn/inMemoryWorld.d.ts", "default": "./dist/turn/inMemoryWorld.js" diff --git a/app/game-engine/src/lifecycle/databaseCommandQueue.ts b/app/game-engine/src/lifecycle/databaseCommandQueue.ts index a3c60035..98b57656 100644 --- a/app/game-engine/src/lifecycle/databaseCommandQueue.ts +++ b/app/game-engine/src/lifecycle/databaseCommandQueue.ts @@ -112,8 +112,7 @@ export class DatabaseTurnDaemonCommandQueue implements TurnDaemonControlQueue, T orderBy: { id: 'asc' }, select: { clockPhase: true, clockRevision: true, deadlineGeneration: true, clockTick: true }, }); - const gameplayAllowed = - !world || world.clockPhase === 'RUNNING' || world.clockPhase === 'MANUAL'; + const gameplayAllowed = !world || world.clockPhase === 'RUNNING' || world.clockPhase === 'MANUAL'; const currentRevision = world?.clockRevision ?? null; const rows = await transaction.$queryRaw< Array<{ @@ -139,7 +138,28 @@ export class DatabaseTurnDaemonCommandQueue implements TurnDaemonControlQueue, T FROM "input_event" WHERE "target" = 'ENGINE'::"InputEventTarget" AND "status" = 'PENDING'::"InputEventStatus" - AND (${gameplayAllowed} OR "event_type" = 'getStatus') + AND ( + ${gameplayAllowed} + OR "event_type" = 'getStatus' + OR ( + ${world?.clockPhase === 'SUSPENDED'} + AND "event_type" = 'messageRespond' + AND "payload" ->> 'messageId' ~ '^[1-9][0-9]*$' + AND EXISTS ( + SELECT 1 + FROM "message" AS pending_message + WHERE pending_message."id" = ("input_event"."payload" ->> 'messageId')::integer + AND pending_message."message" #>> '{option,action}' = 'raiseInvader' + ) + AND EXISTS ( + SELECT 1 + FROM "clock_suspension" AS active_suspension + WHERE active_suspension."status" = 'SUSPENDED' + AND active_suspension."source" = 'UNIFICATION_WAIT' + AND active_suspension."source_revision" = ${currentRevision} + ) + ) + ) ORDER BY "sequence" ASC FOR UPDATE SKIP LOCKED LIMIT ${limit} diff --git a/app/game-engine/src/lifecycle/types.ts b/app/game-engine/src/lifecycle/types.ts index 284790dc..6e720284 100644 --- a/app/game-engine/src/lifecycle/types.ts +++ b/app/game-engine/src/lifecycle/types.ts @@ -29,6 +29,12 @@ export interface TurnDaemonCommandHandler { export interface TurnDaemonCommandExecutionContext { db?: GamePrisma.TransactionClient; + clockOperationAuthority?: { + kind: 'DAEMON'; + profileName: string; + ownerId: string; + fencingEpoch: bigint; + }; } export interface TurnDaemonCommandResponder { diff --git a/app/game-engine/src/turn/actionableMessageResponse.ts b/app/game-engine/src/turn/actionableMessageResponse.ts index d77ec066..7f0d8567 100644 --- a/app/game-engine/src/turn/actionableMessageResponse.ts +++ b/app/game-engine/src/turn/actionableMessageResponse.ts @@ -8,6 +8,7 @@ import type { ImmediateGeneralActionExecutor } from './reservedTurnHandler.js'; import { buildCommandEnv } from './reservedTurnCommands.js'; import type { InMemoryReservedTurnStore } from './reservedTurnStore.js'; import type { TurnEvent } from './types.js'; +import { reconcileClockSuspensionInTransaction, type ClockReconciliationResult } from './clockReconciliation.js'; type ActionableMessageType = 'scout' | 'raiseInvader'; @@ -251,6 +252,18 @@ const respondToRaiseInvader = async (options: { payload: MessagePayload; now: Date; loadArchivedNationMaxId?: (serverId: string) => Promise; + clockOperationAuthority?: { + kind: 'DAEMON'; + profileName: string; + ownerId: string; + fencingEpoch: bigint; + }; + reconcileUnificationWait?: (input: { + db: GamePrisma.TransactionClient; + suspensionId: string; + profileName: string; + authority: NonNullable[0]['authority']>; + }) => Promise; }): Promise => { const { db, world, reservedTurns, actorId, response, row, payload, now } = options; if (row.type !== 'private' || row.mailbox !== actorId || payload.dest.generalId !== actorId) { @@ -272,6 +285,21 @@ const respondToRaiseInvader = async (options: { if (!reservedTurns) { throw new Error('RaiseInvader message response requires the reserved-turn store.'); } + const suspensionId = + typeof state.meta.unificationClockSuspensionId === 'string' ? state.meta.unificationClockSuspensionId : null; + if (!suspensionId || !options.clockOperationAuthority) { + throw new Error('RaiseInvader requires a daemon-authorized UNIFICATION_WAIT suspension.'); + } + const reconcile = + options.reconcileUnificationWait ?? + ((input) => reconcileClockSuspensionInTransaction({ ...input, allowUnificationWait: true })); + const alignment = await reconcile({ + db, + suspensionId, + profileName: options.clockOperationAuthority.profileName, + authority: options.clockOperationAuthority, + }); + world.applyClockReconciliation(alignment); const args = asRecord(payload.option).args; if (!Array.isArray(args) || args.length !== 4 || args.some((value) => typeof value !== 'number')) { return { ok: false, action: 'raiseInvader', reason: '이민족 소환 인자가 올바르지 않습니다.' }; @@ -281,6 +309,7 @@ const respondToRaiseInvader = async (options: { reservedTurns, env: buildCommandEnv(world.getScenarioConfig(), world.getUnitSet()), loadArchivedNationMaxId: options.loadArchivedNationMaxId, + clockWallNow: alignment.resumeWallAt, }); const event: TurnEvent = { id: 0, targetCode: 'month', priority: 0, condition: true, action: [], meta: {} }; await handler( @@ -311,6 +340,18 @@ export const respondToActionableMessage = async (options: { messageId: number; response: boolean; loadArchivedNationMaxId?: (serverId: string) => Promise; + clockOperationAuthority?: { + kind: 'DAEMON'; + profileName: string; + ownerId: string; + fencingEpoch: bigint; + }; + reconcileUnificationWait?: (input: { + db: GamePrisma.TransactionClient; + suspensionId: string; + profileName: string; + authority: NonNullable[0]['authority']>; + }) => Promise; }): Promise => { const acceptedAt = await validateActor(options); const now = options.world.getGameNow(acceptedAt); diff --git a/app/game-engine/src/turn/clockProjectionOutbox.ts b/app/game-engine/src/turn/clockProjectionOutbox.ts index 1c3ca1f0..705631ed 100644 --- a/app/game-engine/src/turn/clockProjectionOutbox.ts +++ b/app/game-engine/src/turn/clockProjectionOutbox.ts @@ -8,7 +8,7 @@ import { type GamePrismaClient, } from '@sammo-ts/infra'; -interface ClockProjectionRedis { +export interface ClockProjectionRedis { get(key: string): Promise; eval(script: string, options: { keys: string[]; arguments: string[] }): Promise; } @@ -128,10 +128,7 @@ const parsePayload = (value: GamePrisma.JsonValue): ProjectionPayload => { targetRevision: safeInteger(payload.targetRevision, 'targetRevision'), deadlineGeneration: safeInteger(payload.deadlineGeneration, 'deadlineGeneration'), shiftTicks: safeInteger(payload.shiftTicks, 'shiftTicks'), - projectionDeltaMilliseconds: safeInteger( - payload.projectionDeltaMilliseconds, - 'projectionDeltaMilliseconds' - ), + projectionDeltaMilliseconds: safeInteger(payload.projectionDeltaMilliseconds, 'projectionDeltaMilliseconds'), clockBaseTime: payload.clockBaseTime, ticksPerSecond: safeInteger(payload.ticksPerSecond, 'ticksPerSecond'), }; @@ -284,7 +281,8 @@ export const applyNextClockProjection = async (options: { if (applied === -1) throw new Error('Redis active clock revision does not match the outbox source revision.'); if (applied === -2) throw new Error('Redis tournament state changed while rebuilding its projection.'); if (applied === -3) throw new Error('Redis target revision exists without the expected projection checksum.'); - if (applied !== 1 && applied !== 2) throw new Error(`Unexpected Redis clock projection result: ${String(result)}`); + if (applied !== 1 && applied !== 2) + throw new Error(`Unexpected Redis clock projection result: ${String(result)}`); await options.db.$transaction(async (transaction) => { await acquireGameSchemaAdvisoryXactLock(transaction, CLOCK_OPERATION_PERSISTENCE_LOCK); diff --git a/app/game-engine/src/turn/clockReconciliation.ts b/app/game-engine/src/turn/clockReconciliation.ts index ae06c761..fc1dde66 100644 --- a/app/game-engine/src/turn/clockReconciliation.ts +++ b/app/game-engine/src/turn/clockReconciliation.ts @@ -111,10 +111,9 @@ const readDbWall = async (db: GamePrisma.TransactionClient): Promise => { return wallNow; }; -const verifyAuthority = async ( - db: GamePrisma.TransactionClient, - authority: ClockOperationAuthority -): Promise => { +export const readClockDatabaseWall = readDbWall; + +const verifyAuthority = async (db: GamePrisma.TransactionClient, authority: ClockOperationAuthority): Promise => { const rows = await db.$queryRaw(GamePrisma.sql` SELECT owner_id AS "ownerId", fencing_epoch AS "fencingEpoch", @@ -133,11 +132,7 @@ const verifyAuthority = async ( } return; } - if ( - !lease?.valid || - lease.ownerId !== authority.ownerId || - lease.fencingEpoch !== authority.fencingEpoch - ) { + if (!lease?.valid || lease.ownerId !== authority.ownerId || lease.fencingEpoch !== authority.fencingEpoch) { throw new Error(`Stale turn-daemon fencing authority for profile ${authority.profileName}.`); } }; @@ -319,6 +314,89 @@ const persistInitialParticipants = async ( } }; +/** + * Locks every registered participant before an enclosing transaction mutates + * the world into a suspended state. The caller must already hold the daemon, + * clock-operation, general-access, and world-row lock prefix. + */ +export const prepareClockSuspensionUnderHeldLocks = async (options: { + db: GamePrisma.TransactionClient; + cutTick: number; + cutWallAt?: Date; +}): Promise<{ cutWallAt: Date }> => { + if (!Number.isSafeInteger(options.cutTick)) { + throw new Error(`Clock suspension cut tick is outside the safe integer range: ${options.cutTick}.`); + } + const cutWallAt = options.cutWallAt ? new Date(options.cutWallAt.getTime()) : await readDbWall(options.db); + await lockParticipants(options.db, BigInt(options.cutTick)); + return { cutWallAt }; +}; + +/** + * Persists the ledger after all suspension-boundary gameplay writes have been + * staged in the same transaction. Lock acquisition belongs to + * prepareClockSuspensionUnderHeldLocks and must happen first. + */ +export const persistClockSuspensionLedgerUnderHeldLocks = async (options: { + db: GamePrisma.TransactionClient; + suspensionId: string; + worldStateId: number; + profileName: string; + source: ClockSuspensionSource; + cutTick: number; + cutWallAt: Date; + rateTicksPerSecond: number; + sourceRevision: number; + policy?: ClockAlignmentPolicy; + catchUpTicks?: number; +}): Promise => { + if (!options.suspensionId.trim() || options.suspensionId.length > 64) { + throw new Error('Clock suspension ID must contain 1-64 characters.'); + } + const policy = options.policy ?? 'EXACT'; + const catchUpTicks = options.catchUpTicks ?? 0; + if (!Number.isSafeInteger(catchUpTicks) || catchUpTicks < 0) { + throw new Error('Clock suspension catch-up ticks must be a non-negative safe integer.'); + } + const existing = await options.db.clockSuspension.findUnique({ where: { id: options.suspensionId } }); + if (existing) { + if ( + existing.worldStateId !== options.worldStateId || + existing.source !== options.source || + existing.sourceRevision !== BigInt(options.sourceRevision) + ) { + throw new Error(`Clock suspension ID ${options.suspensionId} is already bound to another operation.`); + } + return; + } + const world = await options.db.worldState.findUniqueOrThrow({ where: { id: options.worldStateId } }); + if ( + parseGameClockPhase(world.clockPhase) !== 'SUSPENDED' || + world.clockRevision !== BigInt(options.sourceRevision) + ) { + throw new Error('Clock suspension ledger requires a matching durable SUSPENDED world revision.'); + } + const participants = await readParticipantSnapshots(options.db, options.worldStateId, BigInt(options.cutTick)); + await options.db.clockSuspension.create({ + data: { + id: options.suspensionId, + worldStateId: options.worldStateId, + source: options.source, + policy, + status: 'SUSPENDED', + sourceRevision: BigInt(options.sourceRevision), + targetRevision: BigInt(options.sourceRevision + 1), + cutTick: BigInt(options.cutTick), + cutWallAt: options.cutWallAt, + rateTicksPerSecond: options.rateTicksPerSecond, + catchUpTicks: BigInt(catchUpTicks), + participantChecksumBefore: aggregateChecksum(participants), + detail: asJson({ authority: 'DAEMON', profileName: options.profileName }), + }, + }); + await persistInitialParticipants(options.db, options.suspensionId, participants); +}; + const shiftMetaDate = (value: unknown, deltaMilliseconds: number): unknown => { if (typeof value !== 'string' || !value.trim()) return value; const parsed = new Date(value); @@ -512,8 +590,14 @@ export const startClockSuspension = async (options: { 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.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}.`); @@ -585,6 +669,209 @@ export const startClockSuspension = async (options: { ); }; +/** + * Continues a suspension inside a transaction whose caller already verified + * daemon authority and acquired the clock/general-access lock prefix. + */ +export const reconcileClockSuspensionInTransaction = async (options: { + db: GamePrisma.TransactionClient; + suspensionId: string; + profileName: string; + allowUnificationWait?: boolean; + authority?: ClockOperationAuthority; + /** Deterministic fixture seam; production must always use PostgreSQL CURRENT_TIMESTAMP. */ + testResumeWallAt?: Date; +}): Promise => { + const db = options.db; + const worldStateId = await lockWorld(db); + const suspension = await db.clockSuspension.findUniqueOrThrow({ where: { id: options.suspensionId } }); + if (suspension.worldStateId !== worldStateId) { + throw new Error('Clock suspension belongs to another world state.'); + } + if (suspension.status === 'RECONCILING' || suspension.status === 'APPLIED') { + if ( + suspension.gapTicks === null || + suspension.shiftTicks === null || + suspension.alignedTick === null || + !suspension.resumeWallAt + ) { + throw new Error('Persisted clock reconciliation result is incomplete.'); + } + const world = await db.worldState.findUniqueOrThrow({ where: { id: worldStateId } }); + return { + suspensionId: suspension.id, + phase: 'RECONCILING' as const, + sourceRevision: safeNumber(suspension.sourceRevision, 'source revision'), + targetRevision: safeNumber(suspension.targetRevision, 'target revision'), + deadlineGeneration: safeNumber(world.deadlineGeneration, 'deadline generation'), + gapTicks: safeNumber(suspension.gapTicks, 'gap ticks'), + catchUpTicks: safeNumber(suspension.catchUpTicks, 'catch-up ticks'), + shiftTicks: safeNumber(suspension.shiftTicks, 'shift ticks'), + alignedTick: safeNumber(suspension.alignedTick, 'aligned tick'), + resumeWallAt: suspension.resumeWallAt, + }; + } + if (suspension.status !== 'SUSPENDED') { + throw new Error(`Clock suspension cannot reconcile from status ${suspension.status}.`); + } + const world = await db.worldState.findUniqueOrThrow({ where: { id: worldStateId } }); + const phase = parseGameClockPhase(world.clockPhase); + if (phase !== 'SUSPENDED' || world.clockRevision !== suspension.sourceRevision) { + throw new Error('Clock reconciliation phase or source revision fence failed.'); + } + const worldMeta = + world.meta && typeof world.meta === 'object' && !Array.isArray(world.meta) + ? (world.meta as Record) + : {}; + const united = Number(worldMeta.isunited ?? worldMeta.isUnited ?? 0); + if (suspension.source === 'UNIFICATION_WAIT' || united >= 2) { + if (!options.allowUnificationWait || options.authority?.kind !== 'DAEMON') { + throw new Error('Unification wait requires the daemon-authorized atomic alignment-and-invader workflow.'); + } + await verifyAuthority(db, options.authority); + } + const cutTick = safeNumber(suspension.cutTick, 'cut tick'); + await lockParticipants(db, suspension.cutTick); + if (options.testResumeWallAt && process.env.NODE_ENV !== 'test') { + throw new Error('A clock reconciliation wall override is allowed only in tests.'); + } + const resumeWallAt = options.testResumeWallAt ? new Date(options.testResumeWallAt.getTime()) : await readDbWall(db); + const plan = buildClockAlignmentPlan({ + policy: parseClockAlignmentPolicy(suspension.policy), + sourceRevision: safeNumber(suspension.sourceRevision, 'source revision'), + cutTick, + cutWall: suspension.cutWallAt, + resumeWall: resumeWallAt, + ticksPerSecond: suspension.rateTicksPerSecond, + catchUpTicks: safeNumber(suspension.catchUpTicks, 'catch-up ticks'), + }); + const before = await readParticipantSnapshots(db, worldStateId, suspension.cutTick); + assertShiftFits(before, plan.shiftTicks); + await assertScheduleRanges(db, plan.shiftTicks); + const projectionDeltaMilliseconds = Math.trunc((plan.shiftTicks * 1_000) / suspension.rateTicksPerSecond); + if (!Number.isSafeInteger(projectionDeltaMilliseconds)) { + throw new Error('Clock reconciliation projection delta is outside the safe integer range.'); + } + const targetGeneration = world.deadlineGeneration + 1n; + const affected = await applyParticipantShift( + db, + worldStateId, + suspension.cutTick, + BigInt(plan.alignedTick), + BigInt(plan.targetRevision), + targetGeneration, + BigInt(plan.shiftTicks), + projectionDeltaMilliseconds, + resumeWallAt + ); + const after = await readParticipantSnapshots(db, worldStateId, suspension.cutTick); + const afterByKey = new Map(after.map((participant) => [participant.key, participant])); + for (const participant of before) { + const next = afterByKey.get(participant.key); + if (!next) throw new Error(`Missing post-reconciliation participant: ${participant.key}`); + if (participant.policy === 'KEEP' && participant.checksum !== next.checksum) { + throw new Error(`KEEP participant changed during reconciliation: ${participant.key}`); + } + await db.clockReconciliationParticipant.upsert({ + where: { + suspensionId_participantKey: { + suspensionId: suspension.id, + participantKey: participant.key, + }, + }, + create: { + suspensionId: suspension.id, + participantKey: participant.key, + policy: participant.policy, + beforeChecksum: participant.checksum, + afterChecksum: next.checksum, + affectedCount: affected.get(participant.key) ?? 0, + }, + update: { + policy: participant.policy, + beforeChecksum: participant.checksum, + afterChecksum: next.checksum, + affectedCount: affected.get(participant.key) ?? 0, + }, + }); + } + const outboxPayload = { + version: 1, + profileName: options.profileName, + suspensionId: suspension.id, + sourceRevision: plan.sourceRevision, + targetRevision: plan.targetRevision, + deadlineGeneration: safeNumber(targetGeneration, 'deadline generation'), + shiftTicks: plan.shiftTicks, + projectionDeltaMilliseconds, + clockBaseTime: world.clockBaseTime!.toISOString(), + ticksPerSecond: suspension.rateTicksPerSecond, + }; + await db.clockProjectionOutbox.create({ + data: { + worldStateId, + suspensionId: suspension.id, + targetRevision: BigInt(plan.targetRevision), + status: 'PENDING', + payload: asJson(outboxPayload), + checksum: checksum(outboxPayload), + }, + }); + await db.clockSuspension.update({ + where: { id: suspension.id }, + data: { + status: 'RECONCILING', + resumeWallAt, + gapTicks: BigInt(plan.gapTicks), + shiftTicks: BigInt(plan.shiftTicks), + alignedTick: BigInt(plan.alignedTick), + participantChecksumBefore: aggregateChecksum(before), + participantChecksumAfter: aggregateChecksum(after), + }, + }); + return { + suspensionId: suspension.id, + phase: 'RECONCILING', + sourceRevision: plan.sourceRevision, + targetRevision: plan.targetRevision, + deadlineGeneration: safeNumber(targetGeneration, 'deadline generation'), + gapTicks: plan.gapTicks, + catchUpTicks: plan.catchUpTicks, + shiftTicks: plan.shiftTicks, + alignedTick: plan.alignedTick, + resumeWallAt, + }; +}; + +/** Finalizes an atomic unification workflow after an optional rate change. */ +export const refreshClockProjectionForFinalClockUnderHeldLocks = async (options: { + db: GamePrisma.TransactionClient; + suspensionId: string; + clockBaseTime: Date; + tickSeconds: number; +}): Promise => { + if (GAME_TICKS_PER_TURN % options.tickSeconds !== 0) { + throw new Error(`Final clock rate cannot represent an integer tick: ${options.tickSeconds}.`); + } + const outbox = await options.db.clockProjectionOutbox.findFirstOrThrow({ + where: { suspensionId: options.suspensionId, status: 'PENDING' }, + orderBy: { id: 'asc' }, + }); + const payload = + outbox.payload && typeof outbox.payload === 'object' && !Array.isArray(outbox.payload) + ? { ...(outbox.payload as Record) } + : null; + if (!payload || payload.suspensionId !== options.suspensionId) { + throw new Error('Unification clock projection outbox payload is invalid.'); + } + payload.clockBaseTime = options.clockBaseTime.toISOString(); + payload.ticksPerSecond = GAME_TICKS_PER_TURN / options.tickSeconds; + await options.db.clockProjectionOutbox.update({ + where: { id: outbox.id }, + data: { payload: asJson(payload), checksum: checksum(payload) }, + }); +}; + export const reconcileClockSuspension = async (options: { db: GamePrismaClient; suspensionId: string; @@ -597,167 +884,13 @@ export const reconcileClockSuspension = async (options: { 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 suspension = await db.clockSuspension.findUniqueOrThrow({ where: { id: options.suspensionId } }); - if (suspension.worldStateId !== worldStateId) { - throw new Error('Clock suspension belongs to another world state.'); - } - if (suspension.status === 'RECONCILING' || suspension.status === 'APPLIED') { - if ( - suspension.gapTicks === null || - suspension.shiftTicks === null || - suspension.alignedTick === null || - !suspension.resumeWallAt - ) { - throw new Error('Persisted clock reconciliation result is incomplete.'); - } - const world = await db.worldState.findUniqueOrThrow({ where: { id: worldStateId } }); - return { - suspensionId: suspension.id, - phase: 'RECONCILING' as const, - sourceRevision: safeNumber(suspension.sourceRevision, 'source revision'), - targetRevision: safeNumber(suspension.targetRevision, 'target revision'), - deadlineGeneration: safeNumber(world.deadlineGeneration, 'deadline generation'), - gapTicks: safeNumber(suspension.gapTicks, 'gap ticks'), - catchUpTicks: safeNumber(suspension.catchUpTicks, 'catch-up ticks'), - shiftTicks: safeNumber(suspension.shiftTicks, 'shift ticks'), - alignedTick: safeNumber(suspension.alignedTick, 'aligned tick'), - resumeWallAt: suspension.resumeWallAt, - }; - } - if (suspension.status !== 'SUSPENDED') { - throw new Error(`Clock suspension cannot reconcile from status ${suspension.status}.`); - } - const world = await db.worldState.findUniqueOrThrow({ where: { id: worldStateId } }); - const phase = parseGameClockPhase(world.clockPhase); - if (phase !== 'SUSPENDED' || world.clockRevision !== suspension.sourceRevision) { - throw new Error('Clock reconciliation phase or source revision fence failed.'); - } - const worldMeta = - world.meta && typeof world.meta === 'object' && !Array.isArray(world.meta) - ? (world.meta as Record) - : {}; - const united = Number(worldMeta.isunited ?? worldMeta.isUnited ?? 0); - if (suspension.source === 'UNIFICATION_WAIT' || united >= 2) { - throw new Error( - 'Unification wait requires the atomic alignment-and-invader workflow; generic resume is forbidden.' - ); - } - const cutTick = safeNumber(suspension.cutTick, 'cut tick'); - await lockParticipants(db, suspension.cutTick); - if (options.testResumeWallAt && process.env.NODE_ENV !== 'test') { - throw new Error('A clock reconciliation wall override is allowed only in tests.'); - } - const resumeWallAt = options.testResumeWallAt - ? new Date(options.testResumeWallAt.getTime()) - : await readDbWall(db); - const plan = buildClockAlignmentPlan({ - policy: parseClockAlignmentPolicy(suspension.policy), - sourceRevision: safeNumber(suspension.sourceRevision, 'source revision'), - cutTick, - cutWall: suspension.cutWallAt, - resumeWall: resumeWallAt, - ticksPerSecond: suspension.rateTicksPerSecond, - catchUpTicks: safeNumber(suspension.catchUpTicks, 'catch-up ticks'), - }); - const before = await readParticipantSnapshots(db, worldStateId, suspension.cutTick); - assertShiftFits(before, plan.shiftTicks); - await assertScheduleRanges(db, plan.shiftTicks); - const projectionDeltaMilliseconds = Math.trunc( - (plan.shiftTicks * 1_000) / suspension.rateTicksPerSecond - ); - if (!Number.isSafeInteger(projectionDeltaMilliseconds)) { - throw new Error('Clock reconciliation projection delta is outside the safe integer range.'); - } - const targetGeneration = world.deadlineGeneration + 1n; - const affected = await applyParticipantShift( + return reconcileClockSuspensionInTransaction({ db, - worldStateId, - suspension.cutTick, - BigInt(plan.alignedTick), - BigInt(plan.targetRevision), - targetGeneration, - BigInt(plan.shiftTicks), - projectionDeltaMilliseconds, - resumeWallAt - ); - const after = await readParticipantSnapshots(db, worldStateId, suspension.cutTick); - const afterByKey = new Map(after.map((participant) => [participant.key, participant])); - for (const participant of before) { - const next = afterByKey.get(participant.key); - if (!next) throw new Error(`Missing post-reconciliation participant: ${participant.key}`); - if (participant.policy === 'KEEP' && participant.checksum !== next.checksum) { - throw new Error(`KEEP participant changed during reconciliation: ${participant.key}`); - } - await db.clockReconciliationParticipant.upsert({ - where: { - suspensionId_participantKey: { - suspensionId: suspension.id, - participantKey: participant.key, - }, - }, - create: { - suspensionId: suspension.id, - participantKey: participant.key, - policy: participant.policy, - beforeChecksum: participant.checksum, - afterChecksum: next.checksum, - affectedCount: affected.get(participant.key) ?? 0, - }, - update: { - policy: participant.policy, - beforeChecksum: participant.checksum, - afterChecksum: next.checksum, - affectedCount: affected.get(participant.key) ?? 0, - }, - }); - } - const outboxPayload = { - version: 1, + suspensionId: options.suspensionId, profileName: options.authority.profileName, - suspensionId: suspension.id, - sourceRevision: plan.sourceRevision, - targetRevision: plan.targetRevision, - deadlineGeneration: safeNumber(targetGeneration, 'deadline generation'), - shiftTicks: plan.shiftTicks, - projectionDeltaMilliseconds, - clockBaseTime: world.clockBaseTime!.toISOString(), - ticksPerSecond: suspension.rateTicksPerSecond, - }; - await db.clockProjectionOutbox.create({ - data: { - worldStateId, - suspensionId: suspension.id, - targetRevision: BigInt(plan.targetRevision), - status: 'PENDING', - payload: asJson(outboxPayload), - checksum: checksum(outboxPayload), - }, + authority: options.authority, + ...(options.testResumeWallAt ? { testResumeWallAt: options.testResumeWallAt } : {}), }); - await db.clockSuspension.update({ - where: { id: suspension.id }, - data: { - status: 'RECONCILING', - resumeWallAt, - gapTicks: BigInt(plan.gapTicks), - shiftTicks: BigInt(plan.shiftTicks), - alignedTick: BigInt(plan.alignedTick), - participantChecksumBefore: aggregateChecksum(before), - participantChecksumAfter: aggregateChecksum(after), - }, - }); - return { - suspensionId: suspension.id, - phase: 'RECONCILING', - sourceRevision: plan.sourceRevision, - targetRevision: plan.targetRevision, - deadlineGeneration: safeNumber(targetGeneration, 'deadline generation'), - gapTicks: plan.gapTicks, - catchUpTicks: plan.catchUpTicks, - shiftTicks: plan.shiftTicks, - alignedTick: plan.alignedTick, - resumeWallAt, - }; }, { 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 3103df06..82b96171 100644 --- a/app/game-engine/src/turn/databaseHooks.ts +++ b/app/game-engine/src/turn/databaseHooks.ts @@ -56,12 +56,20 @@ import { persistUnificationFinalization } from './unificationPersistence.js'; import { buildOldNationArchiveData } from './oldNationArchive.js'; import { persistYearbookSnapshot } from './yearbookPersistence.js'; import { buildTurnWebPushEvents, captureWebPushTurnBaseline } from './webPushEvents.js'; +import { + persistClockSuspensionLedgerUnderHeldLocks, + prepareClockSuspensionUnderHeldLocks, + readClockDatabaseWall, + refreshClockProjectionForFinalClockUnderHeldLocks, +} from './clockReconciliation.js'; +import { applyNextClockProjection, type ClockProjectionRedis } from './clockProjectionOutbox.js'; export interface DatabaseTurnHooks { hooks: TurnDaemonHooks; takeCommittedReadModelChanges(): RealtimeReadModelChanges | null; takeCommittedReadModelChangeReceipt(): CommittedReadModelChangeReceipt | null; close(): Promise; + applyClockProjection(redis: ClockProjectionRedis, workerId: string): Promise; } export interface CommittedReadModelChangeReceipt { @@ -1140,8 +1148,7 @@ export const createDatabaseTurnHooks = async ( lifecycleEvents.length > 0 || deletedGenerals.length > 0 || generals.some( - (general) => - typeof general.refreshScoreTotal === 'number' && Number.isFinite(general.refreshScoreTotal) + (general) => typeof general.refreshScoreTotal === 'number' && Number.isFinite(general.refreshScoreTotal) ); const persist = async ( prisma: GamePrisma.TransactionClient @@ -1204,12 +1211,27 @@ export const createDatabaseTurnHooks = async ( const expectedPhase = state.clockPhase ?? (state.clockMode === 'realtime' ? 'RUNNING' : 'MANUAL'); const expectedRevision = BigInt(state.clockRevision ?? 1); const expectedGeneration = BigInt(state.deadlineGeneration ?? 1); + const stateMeta = asRecord(state.meta); + const unificationSuspensionId = + typeof stateMeta.unificationClockSuspensionId === 'string' + ? stateMeta.unificationClockSuspensionId + : null; const openingPhaseTransition = - durableClock.clock_phase === 'PREOPEN' && - expectedPhase === 'RUNNING' && - durableClock.opening_reached; + durableClock.clock_phase === 'PREOPEN' && expectedPhase === 'RUNNING' && durableClock.opening_reached; + const unificationSuspensionTransition = + durableClock.clock_phase === 'RUNNING' && + expectedPhase === 'SUSPENDED' && + Number(stateMeta.isunited ?? stateMeta.isUnited ?? 0) === 2 && + Boolean(unificationSuspensionId); + const completionPhaseTransition = + durableClock.clock_phase === 'RUNNING' && + expectedPhase === 'COMPLETED' && + Number(stateMeta.isunited ?? stateMeta.isUnited ?? 0) >= 2; if ( - (!openingPhaseTransition && durableClock.clock_phase !== expectedPhase) || + (!openingPhaseTransition && + !unificationSuspensionTransition && + !completionPhaseTransition && + durableClock.clock_phase !== expectedPhase) || durableClock.clock_revision !== expectedRevision || durableClock.deadline_generation !== expectedGeneration ) { @@ -1218,6 +1240,18 @@ export const createDatabaseTurnHooks = async ( `found ${durableClock.clock_phase}@${durableClock.clock_revision}/${durableClock.deadline_generation}.` ); } + const unificationCutWallAt = unificationSuspensionTransition ? await readClockDatabaseWall(prisma) : null; + const unificationCutTick = unificationCutWallAt + ? world.dateToGameTick(world.getGameNow(unificationCutWallAt)) + : null; + const suspensionPreparation = + unificationCutTick !== null + ? await prepareClockSuspensionUnderHeldLocks({ + db: prisma, + cutTick: unificationCutTick, + cutWallAt: unificationCutWallAt!, + }) + : null; if (commandCompletion) { const commandFence = await prisma.$queryRaw< Array<{ @@ -1235,11 +1269,21 @@ export const createDatabaseTurnHooks = async ( FOR UPDATE `); const event = commandFence[0]; + const unificationRevisionTransition = + commandCompletion.result.type === 'messageRespond' && + commandCompletion.result.ok && + commandCompletion.result.action === 'raiseInvader' && + expectedPhase === 'RECONCILING' && + event?.processing_clock_revision !== null && + event?.processing_deadline_generation !== null && + event?.processing_clock_revision + 1n === expectedRevision && + event?.processing_deadline_generation + 1n === expectedGeneration; if ( !event || event.status !== 'PROCESSING' || - event.processing_clock_revision !== expectedRevision || - event.processing_deadline_generation !== expectedGeneration + (!unificationRevisionTransition && + (event.processing_clock_revision !== expectedRevision || + event.processing_deadline_generation !== expectedGeneration)) ) { throw new Error( `Input event processing clock fence changed before commit: ${commandCompletion.requestId}.` @@ -1350,6 +1394,35 @@ export const createDatabaseTurnHooks = async ( END WHERE start_tick IS NOT NULL OR end_tick IS NOT NULL `); + await prisma.$executeRaw(GamePrisma.sql` + UPDATE select_pool + SET reserved_until = CASE + WHEN reserved_until_tick IS NULL THEN reserved_until + ELSE CAST(${baseTime} AS timestamp) + + (reserved_until_tick / ${ticksPerSecond}) * INTERVAL '1 second' + + (((reserved_until_tick % ${ticksPerSecond}) * 1000) / ${ticksPerSecond}) + * INTERVAL '1 millisecond' + END + WHERE reserved_until_tick IS NOT NULL + `); + await prisma.$executeRaw(GamePrisma.sql` + UPDATE select_npc_token + SET valid_until = CASE + WHEN valid_until_tick IS NULL THEN valid_until + ELSE CAST(${baseTime} AS timestamp) + + (valid_until_tick / ${ticksPerSecond}) * INTERVAL '1 second' + + (((valid_until_tick % ${ticksPerSecond}) * 1000) / ${ticksPerSecond}) + * INTERVAL '1 millisecond' + END, + pick_more_from = CASE + WHEN pick_more_from_tick IS NULL THEN pick_more_from + ELSE CAST(${baseTime} AS timestamp) + + (pick_more_from_tick / ${ticksPerSecond}) * INTERVAL '1 second' + + (((pick_more_from_tick % ${ticksPerSecond}) * 1000) / ${ticksPerSecond}) + * INTERVAL '1 millisecond' + END + WHERE valid_until_tick IS NOT NULL OR pick_more_from_tick IS NOT NULL + `); } for (const betting of pendingNationBettingOpens) { @@ -1817,6 +1890,41 @@ export const createDatabaseTurnHooks = async ( if (options?.reservedTurns && persistedReservedTurnChanges) { await options.reservedTurns.persistChanges(prisma, persistedReservedTurnChanges); } + if (suspensionPreparation && unificationSuspensionId) { + const cutTick = unificationCutTick!; + await prisma.worldState.update({ + where: { id: state.id }, + data: { + clockTick: BigInt(cutTick), + clockWallAnchor: suspensionPreparation.cutWallAt, + }, + }); + await persistClockSuspensionLedgerUnderHeldLocks({ + db: prisma, + suspensionId: unificationSuspensionId, + worldStateId: state.id, + profileName: options?.profileName ?? 'default', + source: 'UNIFICATION_WAIT', + cutTick, + cutWallAt: suspensionPreparation.cutWallAt, + rateTicksPerSecond: GAME_TICKS_PER_TURN / state.tickSeconds, + sourceRevision: state.clockRevision ?? 1, + }); + } + if ( + commandCompletion?.result.type === 'messageRespond' && + commandCompletion.result.ok && + commandCompletion.result.action === 'raiseInvader' && + unificationSuspensionId && + state.clockPhase === 'RECONCILING' + ) { + await refreshClockProjectionForFinalClockUnderHeldLocks({ + db: prisma, + suspensionId: unificationSuspensionId, + clockBaseTime: state.clockBaseTime ?? state.lastTurnTime, + tickSeconds: state.tickSeconds, + }); + } if (commandCompletion) { await prisma.inputEvent.update({ where: { requestId: commandCompletion.requestId }, @@ -1902,6 +2010,10 @@ export const createDatabaseTurnHooks = async ( }, executeCommand: async (requestId, execute) => { const committed = await prisma.$transaction(async (transaction) => { + await options?.turnDaemonLease?.assertActive(transaction); + await acquireGameSchemaAdvisoryXactLock(transaction, CLOCK_OPERATION_PERSISTENCE_LOCK); + await acquireGameSchemaAdvisoryXactLock(transaction, GENERAL_ACCESS_PERSISTENCE_LOCK); + const leaseToken = options?.turnDaemonLease?.getToken(); const directLogFloor = ( await transaction.logEntry.findFirst({ @@ -1909,7 +2021,19 @@ export const createDatabaseTurnHooks = async ( select: { id: true }, }) )?.id ?? 0; - const result = await execute({ db: transaction }); + const result = await execute({ + db: transaction, + ...(leaseToken + ? { + clockOperationAuthority: { + kind: 'DAEMON' as const, + profileName: leaseToken.profile, + ownerId: leaseToken.ownerId, + fencingEpoch: leaseToken.fencingEpoch, + }, + } + : {}), + }); const persisted = await persistChanges(transaction, { requestId, result }, directLogFloor); return { result, persisted }; }, transactionOptions); @@ -1925,6 +2049,14 @@ export const createDatabaseTurnHooks = async ( return takeCommittedReceipt()?.changes ?? null; }, takeCommittedReadModelChangeReceipt: takeCommittedReceipt, + applyClockProjection: async (redis, workerId) => { + await applyNextClockProjection({ db: prisma, redis, workerId }); + const clock = await prisma.worldState.findFirst({ + orderBy: { id: 'asc' }, + select: { clockPhase: true }, + }); + return clock?.clockPhase === 'RUNNING' || clock?.clockPhase === 'MANUAL'; + }, close: () => connector.disconnect(), }; }; diff --git a/app/game-engine/src/turn/inMemoryWorld.ts b/app/game-engine/src/turn/inMemoryWorld.ts index 508d1132..e99dbe73 100644 --- a/app/game-engine/src/turn/inMemoryWorld.ts +++ b/app/game-engine/src/turn/inMemoryWorld.ts @@ -696,6 +696,116 @@ export class InMemoryTurnWorld { return true; } + beginUnificationWait(suspensionId: string): void { + const phase = this.state.clockPhase ?? inferClockPhase(this.state.clockMode ?? 'manual'); + if (phase === 'SUSPENDED' && this.state.meta.unificationClockSuspensionId === suspensionId) { + return; + } + if (phase !== 'RUNNING') { + throw new Error(`UNIFICATION_WAIT can start only from RUNNING; current phase is ${phase}.`); + } + this.state = { + ...this.state, + clockPhase: 'SUSPENDED', + meta: { + ...this.state.meta, + unificationClockSuspensionId: suspensionId, + }, + }; + } + + completeGameClock(): void { + const phase = this.state.clockPhase ?? inferClockPhase(this.state.clockMode ?? 'manual'); + if (phase === 'COMPLETED') return; + if (phase !== 'RUNNING') { + throw new Error(`Game clock can complete only from RUNNING; current phase is ${phase}.`); + } + 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.'); + } + if ( + !Number.isSafeInteger(input.alignedTick) || + !Number.isSafeInteger(input.shiftTicks) || + input.shiftTicks < 0 || + !Number.isSafeInteger(input.targetRevision) || + !Number.isSafeInteger(input.deadlineGeneration) + ) { + throw new Error('In-memory clock reconciliation received an unsafe coordinate.'); + } + const clock = this.getGameClock(); + const shiftedMilliseconds = Math.trunc((input.shiftTicks * 1_000) / clock.ticksPerSecond); + if (!Number.isSafeInteger(shiftedMilliseconds)) { + throw new Error('In-memory clock reconciliation projection delta is unsafe.'); + } + const lastTurnTick = clock.addTicks(this.state.lastTurnTick ?? 0, input.shiftTicks); + const lastTurnTime = clock.tickToDate(lastTurnTick); + this.state = { + ...this.state, + clockTick: input.alignedTick, + clockWallAnchor: new Date(input.resumeWallAt.getTime()), + lastTurnTick, + lastTurnTime, + clockPhase: 'RECONCILING', + clockRevision: input.targetRevision, + deadlineGeneration: input.deadlineGeneration, + meta: { + ...this.state.meta, + lastTurnTime: lastTurnTime.toISOString(), + turntime: shiftGameClockMetaDate(this.state.meta.turntime, shiftedMilliseconds), + starttime: shiftGameClockMetaDate(this.state.meta.starttime, shiftedMilliseconds), + tnmt_time: shiftGameClockMetaDate(this.state.meta.tnmt_time, shiftedMilliseconds), + }, + }; + for (const [generalId, general] of this.generals) { + const turnTick = clock.addTicks(general.turnTick ?? clock.dateToTick(general.turnTime), input.shiftTicks); + this.generals.set(generalId, { + ...general, + turnTick, + turnTime: clock.tickToDate(turnTick), + }); + } + for (const entry of this.generalPoolEntries ?? []) { + if (entry.reservedUntilTick !== null) { + entry.reservedUntilTick = clock.addTicks(entry.reservedUntilTick, input.shiftTicks); + entry.reservedUntil = clock.tickToDate(entry.reservedUntilTick); + } else if (entry.reservedUntil) { + entry.reservedUntil = new Date(entry.reservedUntil.getTime() + shiftedMilliseconds); + } + } + for (const auction of this.pendingNeutralAuctions) { + auction.closeAt = new Date(auction.closeAt.getTime() + shiftedMilliseconds); + } + if (this.checkpoint) { + const checkpointTick = clock.addTicks( + this.checkpoint.turnTick ?? clock.dateToTick(new Date(this.checkpoint.turnTime)), + input.shiftTicks + ); + this.checkpoint = { + ...this.checkpoint, + turnTick: checkpointTick, + turnTime: clock.tickToDate(checkpointTick).toISOString(), + }; + } + } + + completeClockReconciliation(): void { + if (this.state.clockPhase === 'RECONCILING') { + this.state = { ...this.state, clockPhase: 'RUNNING' }; + } + } + getRunnableGameNow(wallNow: Date): Date { const clock = this.getGameClock(); // PREOPEN still needs negative game ticks for cooldowns, but executable diff --git a/app/game-engine/src/turn/monthlyInvaderAction.ts b/app/game-engine/src/turn/monthlyInvaderAction.ts index 9e97dd88..a4537630 100644 --- a/app/game-engine/src/turn/monthlyInvaderAction.ts +++ b/app/game-engine/src/turn/monthlyInvaderAction.ts @@ -133,6 +133,7 @@ export const createRaiseInvaderHandler = (options: { env: TurnCommandEnv; loadArchivedNationMaxId?: (serverId: string) => Promise; maxGeneralsPerMinute?: number; + clockWallNow?: Date; }): MonthlyEventActionHandler => { return async (args, environment) => { const world = options.getWorld(); @@ -166,7 +167,7 @@ export const createRaiseInvaderHandler = (options: { (candidate) => totalGeneralCount <= maxGeneralsPerMinute * candidate ); if (nextTerm !== undefined) { - world.changeTurnTerm(nextTerm); + world.changeTurnTerm(nextTerm, options.clockWallNow); // Reprojection preserves the frozen monthly boundary by tick but // changes its displayed Date. New generals must join that frozen // boundary, not the realtime game clock that kept advancing while diff --git a/app/game-engine/src/turn/turnDaemon.ts b/app/game-engine/src/turn/turnDaemon.ts index 9a8bd064..665ce898 100644 --- a/app/game-engine/src/turn/turnDaemon.ts +++ b/app/game-engine/src/turn/turnDaemon.ts @@ -1,3 +1,5 @@ +import { randomUUID } from 'node:crypto'; + import { loadActionModuleBundle, type TurnCommandProfile, type TurnSchedule } from '@sammo-ts/logic'; import { buildGameEventChannel, @@ -20,7 +22,11 @@ import type { Clock, TurnDaemonControlQueue, TurnDaemonHooks, TurnRunBudget } fr import { TurnDaemonLifecycle } from '../lifecycle/turnDaemonLifecycle.js'; import { DatabaseTurnDaemonCommandQueue } from '../lifecycle/databaseCommandQueue.js'; import type { MapLoaderOptions } from '../scenario/mapLoader.js'; -import { createDatabaseTurnHooks, type CommittedReadModelChangeReceipt } from './databaseHooks.js'; +import { + createDatabaseTurnHooks, + type CommittedReadModelChangeReceipt, + type DatabaseTurnHooks, +} from './databaseHooks.js'; import type { GeneralTurnHandler, InMemoryTurnWorldOptions, TurnCalendarHandler } from './inMemoryWorld.js'; import { InMemoryTurnWorld } from './inMemoryWorld.js'; import { InMemoryTurnProcessor } from './inMemoryTurnProcessor.js'; @@ -497,15 +503,40 @@ const createRealtimeRuntime = async (options: { profileName: string; hooks?: TurnDaemonHooks; takeCommittedReadModelChangeReceipt: (() => CommittedReadModelChangeReceipt | null) | null; -}): Promise<{ redisConnector: RedisConnector | null; hooks?: TurnDaemonHooks }> => { + applyClockProjection?: (redis: RedisConnector['client'], workerId: string) => Promise; + onClockProjectionApplied?: () => void; +}): Promise<{ + redisConnector: RedisConnector | null; + hooks?: TurnDaemonHooks; + stopClockProjectionWorker: () => void; +}> => { const redisConfig = resolveRedisConfig(options.redisUrl); if (!redisConfig) { - return { redisConnector: null, hooks: options.hooks }; + return { redisConnector: null, hooks: options.hooks, stopClockProjectionWorker: () => {} }; } const redisConnector = createRedisConnector(redisConfig); await redisConnector.connect(); const redisClient = redisConnector.client; + const clockProjectionWorkerId = `turn-daemon:${options.profileName}:${randomUUID()}`; + let clockProjectionInFlight = false; + let clockProjectionStopped = false; + const recoverClockProjection = async (): Promise => { + if (!options.applyClockProjection || clockProjectionInFlight || clockProjectionStopped) return; + clockProjectionInFlight = true; + try { + if (await options.applyClockProjection(redisClient, clockProjectionWorkerId)) { + options.onClockProjectionApplied?.(); + } + } finally { + clockProjectionInFlight = false; + } + }; + await recoverClockProjection().catch(() => undefined); + const clockProjectionTimer = options.applyClockProjection + ? setInterval(() => void recoverClockProjection().catch(() => undefined), 1_000) + : null; + clockProjectionTimer?.unref(); const realtimeChannel = buildGameEventChannel(options.profileName); const revisionKey = buildGameReadModelRevisionKey(options.profileName); const domainRevisionKey = buildGameReadModelDomainRevisionKey(options.profileName); @@ -550,6 +581,9 @@ const createRealtimeRuntime = async (options: { await basePublishEvents?.(result); }, publishCommandEvents: async (result) => { + if (result.type === 'messageRespond' && result.ok && result.action === 'raiseInvader') { + await recoverClockProjection(); + } try { const changes = options.takeCommittedReadModelChangeReceipt?.()?.changes; if (changes && hasRealtimeReadModelChanges(changes)) { @@ -569,7 +603,14 @@ const createRealtimeRuntime = async (options: { await basePublishCommandEvents?.(result); }, }; - return { redisConnector, hooks }; + return { + redisConnector, + hooks, + stopClockProjectionWorker: () => { + clockProjectionStopped = true; + if (clockProjectionTimer) clearInterval(clockProjectionTimer); + }, + }; }; const createStartedAdminActionConsumer = async (options: { @@ -672,6 +713,8 @@ const createTurnDaemonRuntimeWithLease = async ( })); let worldRef: InMemoryTurnWorld | null = null; let redisConnector: RedisConnector | null = null; + let stopClockProjectionWorker = () => {}; + let applyClockProjection: DatabaseTurnHooks['applyClockProjection'] | undefined; const nationTraits = await loadNationTraitModules([...NATION_TRAIT_KEYS], new NationTraitLoader()); const nationTraitMap = new Map(nationTraits.map((module) => [module.key, module])); const monthlyActionModules = await loadActionModuleBundle( @@ -876,6 +919,7 @@ const createTurnDaemonRuntimeWithLease = async ( }, }; takeCommittedReadModelChangeReceipt = dbHooks.takeCommittedReadModelChangeReceipt; + applyClockProjection = dbHooks.applyClockProjection; close = async () => { if (auctionBidder) { await auctionBidder.close(); @@ -926,9 +970,17 @@ const createTurnDaemonRuntimeWithLease = async ( profileName: options.profileName ?? options.profile, hooks, takeCommittedReadModelChangeReceipt, + ...(applyClockProjection + ? { + applyClockProjection: (redis: RedisConnector['client'], workerId: string) => + applyClockProjection!(redis, workerId), + onClockProjectionApplied: () => world.completeClockReconciliation(), + } + : {}), }); redisConnector = realtimeRuntime.redisConnector; hooks = realtimeRuntime.hooks; + stopClockProjectionWorker = realtimeRuntime.stopClockProjectionWorker; const commandConnector = hooks ? createGamePostgresConnector({ url: options.databaseUrl }) : null; const databaseCommandQueue = commandConnector ? new DatabaseTurnDaemonCommandQueue(commandConnector.prisma) : null; @@ -950,6 +1002,7 @@ const createTurnDaemonRuntimeWithLease = async ( const baseClose = close; close = async () => { + stopClockProjectionWorker(); await baseClose(); await neutralAuctionRegistrar.close(); if (redisConnector) { diff --git a/app/game-engine/src/turn/unificationHandler.ts b/app/game-engine/src/turn/unificationHandler.ts index ecdafbf8..5e85643f 100644 --- a/app/game-engine/src/turn/unificationHandler.ts +++ b/app/game-engine/src/turn/unificationHandler.ts @@ -1,3 +1,5 @@ +import { createHash } from 'node:crypto'; + import { asNumber, asRecord, JosaUtil } from '@sammo-ts/common'; import { LogCategory, LogFormat, LogScope, type LogEntryDraft } from '@sammo-ts/logic'; @@ -163,6 +165,14 @@ export const createUnificationHandler = (options: { }); } } + const sourceRevision = world.getGameClockState().revision; + const suspensionId = `unification-wait-${createHash('sha256') + .update(`${serverId}:${sourceRevision}`) + .digest('hex') + .slice(0, 32)}`; + world.beginUnificationWait(suspensionId); + } else { + world.completeGameClock(); } queueYearbookSnapshot(world, options.profileName, context.currentYear, context.currentMonth); diff --git a/app/game-engine/src/turn/worldCommandHandler.ts b/app/game-engine/src/turn/worldCommandHandler.ts index 2a4b29c9..0ccda2cc 100644 --- a/app/game-engine/src/turn/worldCommandHandler.ts +++ b/app/game-engine/src/turn/worldCommandHandler.ts @@ -146,6 +146,10 @@ const refreshActorKillturn = (world: InMemoryTurnWorld, actor: TurnGeneral): voi interface CommandHandlerContext { world: InMemoryTurnWorld; commandDb?: GamePrisma.TransactionClient; + clockOperationAuthority?: Extract< + NonNullable, + { kind: 'DAEMON' } + >; auctionFinalizer?: AuctionFinalizer; auctionBidder?: AuctionBidder; tournamentRewardFinalizer?: TournamentRewardFinalizer; @@ -153,6 +157,7 @@ interface CommandHandlerContext { reservedTurns?: InMemoryReservedTurnStore; generalActionModules?: ReadonlyArray; loadArchivedNationMaxId?: (serverId: string) => Promise; + reconcileUnificationWait?: Parameters[0]['reconcileUnificationWait']; } const requireCommandDatabase = (ctx: CommandHandlerContext): DatabaseClient => { @@ -464,10 +469,10 @@ async function handleSelectPoolCreate( typeof processingGameTick === 'number' && Number.isSafeInteger(processingGameTick) ? ctx.world.gameTickToDate(processingGameTick) : command.acceptedGameTick !== undefined - ? ctx.world.gameTickToDate(command.acceptedGameTick) - : command.acceptedGameAt !== undefined - ? new Date(command.acceptedGameAt) - : ctx.world.getGameNow(operationalAcceptedAt); + ? ctx.world.gameTickToDate(command.acceptedGameTick) + : command.acceptedGameAt !== undefined + ? new Date(command.acceptedGameAt) + : ctx.world.getGameNow(operationalAcceptedAt); const turnScheduleAt = ctx.world.getRunnableGameNow(operationalAcceptedAt); try { return { @@ -1834,6 +1839,8 @@ async function handleMessageRespond( messageId: command.messageId, response: command.response, loadArchivedNationMaxId: ctx.loadArchivedNationMaxId, + clockOperationAuthority: ctx.clockOperationAuthority, + reconcileUnificationWait: ctx.reconcileUnificationWait, }); return { type: 'messageRespond', @@ -3101,6 +3108,7 @@ export const createTurnDaemonCommandHandler = (options: { auctionBidder?: AuctionBidder; tournamentRewardFinalizer?: TournamentRewardFinalizer; loadArchivedNationMaxId?: (serverId: string) => Promise; + reconcileUnificationWait?: Parameters[0]['reconcileUnificationWait']; }): TurnDaemonCommandHandler => { let immediateGeneralActionExecutor: Promise | null = null; const ctx: CommandHandlerContext = { @@ -3111,6 +3119,7 @@ export const createTurnDaemonCommandHandler = (options: { reservedTurns: options.reservedTurns, generalActionModules: options.generalActionModules, loadArchivedNationMaxId: options.loadArchivedNationMaxId, + reconcileUnificationWait: options.reconcileUnificationWait, getImmediateGeneralActionExecutor: () => { immediateGeneralActionExecutor ??= createImmediateGeneralActionExecutor({ world: options.world, @@ -3232,6 +3241,7 @@ export const createTurnDaemonCommandHandler = (options: { return null; } ctx.commandDb = executionContext?.db; + ctx.clockOperationAuthority = executionContext?.clockOperationAuthority; try { if (isActorBoundGeneralCommand(command)) { const rejected = await validateActorBoundGeneralCommand(ctx, command); @@ -3242,6 +3252,7 @@ export const createTurnDaemonCommandHandler = (options: { return await handler(command); } finally { ctx.commandDb = undefined; + ctx.clockOperationAuthority = undefined; } }, }; diff --git a/app/game-engine/test/clockReconciliation.integration.test.ts b/app/game-engine/test/clockReconciliation.integration.test.ts index 3cf349b6..df44f46b 100644 --- a/app/game-engine/test/clockReconciliation.integration.test.ts +++ b/app/game-engine/test/clockReconciliation.integration.test.ts @@ -1,9 +1,12 @@ -import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest'; +import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'; import { GameClock } from '@sammo-ts/common'; import { createGamePostgresConnector, createRedisConnector, + GENERAL_ACCESS_PERSISTENCE_LOCK, + GamePrisma, + acquireGameSchemaAdvisoryXactLock, type GamePrismaClient, type RedisConnector, } from '@sammo-ts/infra'; @@ -22,20 +25,7 @@ describeIntegration('durable clock reconciliation', () => { let disconnect: (() => Promise) | undefined; let redis: RedisConnector; - beforeAll(async () => { - const connector = createGamePostgresConnector({ url: process.env.DATABASE_URL! }); - db = connector.prisma; - disconnect = connector.disconnect; - redis = createRedisConnector({ url: process.env.REDIS_URL! }); - await redis.connect(); - }); - - afterAll(async () => { - await redis.disconnect(); - await disconnect?.(); - }); - - beforeEach(async () => { + const clean = async (): Promise => { await redis.client.flushDb(); await db.$transaction([ db.clockProjectionOutbox.deleteMany(), @@ -54,6 +44,24 @@ describeIntegration('durable clock reconciliation', () => { db.turnDaemonLease.deleteMany(), db.worldState.deleteMany(), ]); + }; + + beforeAll(async () => { + const connector = createGamePostgresConnector({ url: process.env.DATABASE_URL! }); + db = connector.prisma; + disconnect = connector.disconnect; + redis = createRedisConnector({ url: process.env.REDIS_URL! }); + await redis.connect(); + }); + + afterAll(async () => { + await clean(); + await redis.disconnect(); + await disconnect?.(); + }); + + beforeEach(async () => { + await clean(); }); it('preserves every remaining deadline and occurrence across a 65m17.250s exact gap', async () => { @@ -246,9 +254,9 @@ describeIntegration('durable clock reconciliation', () => { ); await redis.client.set('sammo:clock-test:clock:active-revision', '1'); - expect( - await applyNextClockProjection({ db, redis: redis.client, workerId: 'clock-projection-success' }) - ).toBe('APPLIED'); + expect(await applyNextClockProjection({ db, redis: redis.client, workerId: 'clock-projection-success' })).toBe( + 'APPLIED' + ); expect(await redis.client.get('sammo:clock-test:clock:active-revision')).toBe('2'); expect(await redis.client.get('sammo:clock-test:clock:deadline-generation')).toBe('8'); expect(await redis.client.get('sammo:clock-test:clock:phase')).toBe('RUNNING'); @@ -355,10 +363,85 @@ describeIntegration('durable clock reconciliation', () => { expect(await db.clockProjectionOutbox.findFirstOrThrow()).toMatchObject({ status: 'FAILED', attempts: 1 }); await db.clockProjectionOutbox.updateMany({ data: { availableAt: new Date(0) } }); - expect( - await applyNextClockProjection({ db, redis: redis.client, workerId: 'clock-projection-restart' }) - ).toBe('RECOVERED'); + expect(await applyNextClockProjection({ db, redis: redis.client, workerId: 'clock-projection-restart' })).toBe( + 'RECOVERED' + ); expect(await db.worldState.findFirstOrThrow()).toMatchObject({ clockPhase: 'RUNNING', clockRevision: 4n }); expect(await db.clockProjectionOutbox.findFirstOrThrow()).toMatchObject({ status: 'APPLIED', attempts: 2 }); }); + + it('uses DB wall time despite host drift and does not deadlock with a general-access writer', async () => { + const [dbWall] = await db.$queryRaw>(GamePrisma.sql` + SELECT (CURRENT_TIMESTAMP AT TIME ZONE 'UTC')::timestamp(3) AS now + `); + const baseTime = new Date('2026-03-01T00:00:00.000Z'); + const clock = new GameClock({ + baseTime, + tick: 42, + mode: 'realtime', + wallAnchor: dbWall!.now, + turnSeconds: 600, + phase: 'RUNNING', + revision: 1, + }); + const world = await db.worldState.create({ + data: { + scenarioCode: 'clock-drift-deadlock-test', + currentYear: 180, + currentMonth: 1, + tickSeconds: 600, + clockBaseTime: baseTime, + clockTick: 42n, + clockMode: 'realtime', + clockWallAnchor: dbWall!.now, + lastTurnTick: 42n, + clockPhase: 'RUNNING', + clockRevision: 1n, + deadlineGeneration: 1n, + }, + }); + await db.general.create({ + data: { id: 1, name: 'lock-general', turnTick: 100n, turnTime: clock.tickToDate(100) }, + }); + + let releaseWriter!: () => void; + let signalWriterLocked!: () => void; + const writerLocked = new Promise((resolve) => { + signalWriterLocked = resolve; + }); + const writerRelease = new Promise((resolve) => { + releaseWriter = resolve; + }); + const writer = db.$transaction(async (transaction) => { + await acquireGameSchemaAdvisoryXactLock(transaction, GENERAL_ACCESS_PERSISTENCE_LOCK); + signalWriterLocked(); + await writerRelease; + await transaction.$queryRaw(GamePrisma.sql` + SELECT id FROM world_state WHERE id = ${world.id} FOR UPDATE + `); + }); + await writerLocked; + const dateNow = vi.spyOn(Date, 'now').mockReturnValue(dbWall!.now.getTime() + 12 * 60 * 60_000); + try { + const suspensionPromise = startClockSuspension({ + db, + suspensionId: 'clock-host-drift-deadlock', + source: 'MAINTENANCE', + authority: { kind: 'OFFLINE', profileName: 'clock-drift-deadlock-test', reason: 'fixture' }, + }); + releaseWriter(); + const suspension = await Promise.race([ + Promise.all([writer, suspensionPromise]).then(([, result]) => result), + new Promise((_, reject) => + setTimeout(() => reject(new Error('general-access/clock-operation deadlock')), 5_000) + ), + ]); + expect(Math.abs(suspension.cutWallAt.getTime() - dbWall!.now.getTime())).toBeLessThan(5_000); + expect(suspension.cutTick).toBeGreaterThanOrEqual(42); + expect(suspension.cutTick).toBeLessThan(42 + 60 * 60_000); + } finally { + dateNow.mockRestore(); + releaseWriter(); + } + }); }); diff --git a/app/game-engine/test/databaseCommandQueue.integration.test.ts b/app/game-engine/test/databaseCommandQueue.integration.test.ts index d1714621..540e386d 100644 --- a/app/game-engine/test/databaseCommandQueue.integration.test.ts +++ b/app/game-engine/test/databaseCommandQueue.integration.test.ts @@ -27,7 +27,13 @@ integration('database command queue', () => { beforeEach(async () => { await db.inputEvent.deleteMany({ where: { requestId: { startsWith: 'integration:engine:' } } }); - await db.clockSuspension.deleteMany({ where: { id: 'integration-queue-revision-8-9' } }); + await db.clockProjectionOutbox.deleteMany({ + where: { suspensionId: { in: ['integration-queue-revision-8-9', 'integration-unification-wait'] } }, + }); + await db.clockSuspension.deleteMany({ + where: { id: { in: ['integration-queue-revision-8-9', 'integration-unification-wait'] } }, + }); + await db.message.deleteMany({ where: { mailbox: 991_199 } }); await db.worldState.updateMany({ data: { clockPhase: 'RUNNING' } }); }); @@ -350,4 +356,103 @@ integration('database command queue', () => { processingDeadlineGeneration: 4n, }); }); + + it('dequeues only the invader decision while an UNIFICATION_WAIT suspension is active', async () => { + const existingWorld = await db.worldState.findFirst({ orderBy: { id: 'asc' } }); + const world = existingWorld + ? await db.worldState.update({ + where: { id: existingWorld.id }, + data: { clockPhase: 'SUSPENDED', clockRevision: 31n, deadlineGeneration: 7n, clockTick: 900n }, + }) + : await db.worldState.create({ + data: { + scenarioCode: 'queue-unification-clock-test', + currentYear: 200, + currentMonth: 1, + tickSeconds: 600, + clockPhase: 'SUSPENDED', + clockRevision: 31n, + deadlineGeneration: 7n, + clockTick: 900n, + }, + }); + const message = await db.message.create({ + data: { + mailbox: 991_199, + type: 'private', + src: 0, + dest: 991_199, + time: new Date(), + validUntil: new Date('9999-12-31T00:00:00.000Z'), + message: { option: { action: 'raiseInvader', used: false } }, + }, + }); + await db.clockSuspension.create({ + data: { + id: 'integration-unification-wait', + worldStateId: world.id, + source: 'UNIFICATION_WAIT', + policy: 'EXACT', + status: 'SUSPENDED', + sourceRevision: 31n, + targetRevision: 32n, + cutTick: 900n, + cutWallAt: new Date(), + rateTicksPerSecond: 60_000, + }, + }); + const messageRequestId = 'integration:engine:unification-message'; + const gameplayRequestId = 'integration:engine:unification-gameplay'; + await db.inputEvent.createMany({ + data: [ + { + requestId: messageRequestId, + target: 'ENGINE', + eventType: 'messageRespond', + actorUserId: 'user-991199', + acceptedGameTick: 900n, + acceptedClockRevision: 31n, + acceptedDeadlineGeneration: 7n, + payload: { + type: 'messageRespond', + requestId: messageRequestId, + userId: 'user-991199', + generalId: 991_199, + messageId: message.id, + response: true, + }, + }, + { + requestId: gameplayRequestId, + target: 'ENGINE', + eventType: 'vacation', + actorUserId: 'user-991199', + acceptedGameTick: 900n, + acceptedClockRevision: 31n, + acceptedDeadlineGeneration: 7n, + payload: { + type: 'vacation', + requestId: gameplayRequestId, + userId: 'user-991199', + generalId: 991_199, + }, + }, + ], + }); + + const queue = new DatabaseTurnDaemonCommandQueue(db); + await expect(queue.drain()).resolves.toEqual([ + { + type: 'messageRespond', + requestId: messageRequestId, + userId: 'user-991199', + generalId: 991_199, + messageId: message.id, + response: true, + }, + ]); + await expect( + db.inputEvent.findUniqueOrThrow({ where: { requestId: gameplayRequestId } }) + ).resolves.toMatchObject({ status: 'PENDING' }); + }); }); diff --git a/app/game-engine/test/unificationFinalization.integration.test.ts b/app/game-engine/test/unificationFinalization.integration.test.ts index fd18c1e9..0b0761ec 100644 --- a/app/game-engine/test/unificationFinalization.integration.test.ts +++ b/app/game-engine/test/unificationFinalization.integration.test.ts @@ -1,7 +1,7 @@ import { afterAll, beforeAll, describe, expect, it } from 'vitest'; -import { normalizeArchivedGeneral, type ArchivedJsonValue } from '@sammo-ts/common'; -import { createGamePostgresConnector, type GamePrismaClient } from '@sammo-ts/infra'; +import { GAME_TICKS_PER_TURN, GameClock, normalizeArchivedGeneral, type ArchivedJsonValue } from '@sammo-ts/common'; +import { createGamePostgresConnector, createRedisConnector, type GamePrismaClient } from '@sammo-ts/infra'; import { LogCategory, LogScope } from '@sammo-ts/logic'; import { createAuctionBidder } from '../src/auction/bidder.js'; @@ -14,6 +14,10 @@ import { createMergeInheritPointRankHandler } from '../src/turn/monthlyUniqueInh import { loadPendingUnificationAuctionCancellations } from '../src/turn/unificationAuctionCancellation.js'; import { createUnificationHandler } from '../src/turn/unificationHandler.js'; import { loadTurnWorldFromDatabase } from '../src/turn/worldLoader.js'; +import { InMemoryReservedTurnStore } from '../src/turn/reservedTurnStore.js'; +import { createTurnDaemonCommandHandler } from '../src/turn/worldCommandHandler.js'; +import { reconcileClockSuspensionInTransaction } from '../src/turn/clockReconciliation.js'; +import { applyNextClockProjection } from '../src/turn/clockProjectionOutbox.js'; const databaseUrl = process.env.INPUT_EVENT_DATABASE_URL; const integration = describe.skipIf(!databaseUrl); @@ -22,15 +26,26 @@ const serverId = 'che_unification_atomicity_fixture'; const profileName = 'che'; const userId = 'unification-atomicity-user'; const legacyOfficerPicture = 'users/core/a369f064a434262b025bd2ebc70c60d5.jpg?=20260814'; +const invaderCityId = fixtureId + 1; +const invaderNationId = fixtureId + 1; +const invaderGeneralIds = Array.from({ length: 10 }, (_, index) => fixtureId + 1 + index); integration('unification finalization transaction', () => { let db: GamePrismaClient; let closeDb: (() => Promise) | undefined; const cleanup = async (): Promise => { + await db.clockProjectionOutbox.deleteMany({ + where: { suspension: { worldState: { scenarioCode: 'unification-atomicity-fixture' } } }, + }); + await db.clockSuspension.deleteMany({ + where: { worldState: { scenarioCode: 'unification-atomicity-fixture' } }, + }); + await db.inputEvent.deleteMany({ where: { requestId: { startsWith: 'unification-clock:' } } }); + await db.turnDaemonLease.deleteMany({ where: { profile: profileName } }); await db.message.deleteMany({ where: { mailbox: fixtureId } }); await db.auction.deleteMany({ where: { hostGeneralId: fixtureId } }); - await db.event.deleteMany({ where: { id: fixtureId } }); + await db.event.deleteMany({ where: { id: { in: [fixtureId, fixtureId + 1, fixtureId + 2] } } }); await db.unificationFinalization.deleteMany({ where: { serverId } }); await db.yearbookHistory.deleteMany({ where: { profileName: serverId } }); await db.emperor.deleteMany({ where: { serverId } }); @@ -44,10 +59,15 @@ integration('unification finalization transaction', () => { await db.logEntry.deleteMany({ where: { OR: [{ generalId: fixtureId }, { year: 190, month: 7 }] }, }); - await db.rankData.deleteMany({ where: { generalId: fixtureId } }); - await db.general.deleteMany({ where: { id: fixtureId } }); - await db.city.deleteMany({ where: { id: fixtureId } }); - await db.nation.deleteMany({ where: { id: fixtureId } }); + await db.generalTurn.deleteMany({ where: { generalId: { in: invaderGeneralIds } } }); + await db.nationTurn.deleteMany({ where: { nationId: invaderNationId } }); + await db.diplomacy.deleteMany({ + where: { OR: [{ srcNationId: invaderNationId }, { destNationId: invaderNationId }] }, + }); + await db.rankData.deleteMany({ where: { generalId: { in: [fixtureId, ...invaderGeneralIds] } } }); + await db.general.deleteMany({ where: { id: { in: [fixtureId, ...invaderGeneralIds] } } }); + await db.city.deleteMany({ where: { id: { in: [fixtureId, invaderCityId] } } }); + await db.nation.deleteMany({ where: { id: { in: [fixtureId, invaderNationId] } } }); await db.worldState.deleteMany({ where: { scenarioCode: 'unification-atomicity-fixture' } }); }; @@ -90,7 +110,31 @@ integration('unification finalization transaction', () => { id: fixtureId, name: '원자도시', nationId: fixtureId, - level: 1, + level: 3, + population: 1_000, + populationMax: 2_000, + agriculture: 100, + agricultureMax: 200, + commerce: 100, + commerceMax: 200, + security: 100, + securityMax: 200, + defence: 100, + defenceMax: 200, + wall: 100, + wallMax: 200, + supplyState: 1, + frontState: 0, + region: 1, + meta: { state: 0 }, + }, + }); + await db.city.create({ + data: { + id: invaderCityId, + name: '남만', + nationId: fixtureId, + level: 4, population: 1_000, populationMax: 2_000, agriculture: 100, @@ -249,6 +293,9 @@ integration('unification finalization transaction', () => { season: 1, scenarioId: 2, refreshLimit: 2, + maxGeneralsPerMinute: 1, + lastGeneralId: fixtureId, + lastNationId: fixtureId, scenarioMeta: { title: '원자성 시나리오', startYear: 190, @@ -311,6 +358,38 @@ integration('unification finalization transaction', () => { expect.objectContaining({ inheritSpentTrackedAmount: 50 }), ]); + const clockBaseTime = new Date('0190-01-01T00:00:00.000Z'); + const clockWallAnchor = new Date('2030-01-01T00:00:00.000Z'); + const fixtureClock = new GameClock({ + baseTime: clockBaseTime, + tick: 0, + mode: 'realtime', + wallAnchor: clockWallAnchor, + turnSeconds: 600, + phase: 'RUNNING', + revision: 1, + }); + const initialClockTick = fixtureClock.dateToTick(new Date('0190-06-01T00:00:00.000Z')); + await db.worldState.update({ + where: { id: worldRow.id }, + data: { + clockBaseTime, + clockTick: BigInt(initialClockTick), + clockMode: 'realtime', + clockWallAnchor, + lastTurnTick: BigInt(initialClockTick), + clockPhase: 'RUNNING', + clockRevision: 1n, + deadlineGeneration: 1n, + }, + }); + await db.auction.updateMany({ + where: { id: { in: [uniqueAuction.id, resourceAuction.id] } }, + data: { + openTick: BigInt(initialClockTick), + closeTick: BigInt(fixtureClock.dateToTick(futureCloseAt)), + }, + }); const loaded = await loadTurnWorldFromDatabase({ databaseUrl: databaseUrl! }); let world: InMemoryTurnWorld | null = null; const actions = new Map(); @@ -326,7 +405,8 @@ integration('unification finalization transaction', () => { schedule: { entries: [{ startMinute: 0, tickMinutes: 10 }] }, calendarHandler: composeCalendarHandlers(events, unification.handler), }); - const hooks = await createDatabaseTurnHooks(databaseUrl!, world, { profileName }); + const reservedTurns = new InMemoryReservedTurnStore(db, { maxGeneralTurns: 30, maxNationTurns: 12 }); + const hooks = await createDatabaseTurnHooks(databaseUrl!, world, { profileName, reservedTurns }); const stateManager = new EngineStateManager(); stateManager.register('world', { capture: () => world!.captureState(), @@ -343,6 +423,7 @@ integration('unification finalization transaction', () => { const beforeFailedTurn = world.captureState(); await expect( stateManager.transaction(async () => { + world!.advanceGameClockTo(new Date('0190-07-01T00:00:00.000Z'), clockWallAnchor); await world!.advanceMonth(new Date('0190-07-01T00:00:00.000Z')); expect(world!.getState().meta).toMatchObject({ isUnited: 2, isunited: 2, refreshLimit: 200 }); expect(world!.peekDirtyState().pendingUnificationFinalizations).toHaveLength(1); @@ -381,6 +462,7 @@ integration('unification finalization transaction', () => { }, }); await stateManager.transaction(async () => { + world!.advanceGameClockTo(new Date('0190-07-01T00:00:00.000Z'), clockWallAnchor); await world!.advanceMonth(new Date('0190-07-01T00:00:00.000Z')); await hooks.hooks.flushChanges?.(runResult); }); @@ -512,7 +594,190 @@ integration('unification finalization transaction', () => { expect(await db.worldState.findUniqueOrThrow({ where: { id: worldRow.id } })).toMatchObject({ currentYear: 190, currentMonth: 7, + clockPhase: 'SUSPENDED', }); + const suspension = await db.clockSuspension.findFirstOrThrow({ where: { worldStateId: worldRow.id } }); + expect(suspension).toMatchObject({ + source: 'UNIFICATION_WAIT', + policy: 'EXACT', + status: 'SUSPENDED', + sourceRevision: 1n, + targetRevision: 2n, + }); + + const invaderPrompt = (await db.message.findMany({ where: { mailbox: fixtureId } })).find((row) => { + const payload = row.message as { option?: { action?: unknown } }; + return payload.option?.action === 'raiseInvader'; + }); + expect(invaderPrompt).toBeDefined(); + const requestId = 'unification-clock:raise-invader'; + await db.inputEvent.create({ + data: { + requestId, + target: 'ENGINE', + eventType: 'messageRespond', + actorUserId: userId, + payload: { + type: 'messageRespond', + requestId, + userId, + generalId: fixtureId, + messageId: invaderPrompt!.id, + response: true, + }, + status: 'PROCESSING', + acceptedGameTick: suspension.cutTick, + acceptedClockRevision: suspension.sourceRevision, + acceptedDeadlineGeneration: 1n, + processingAt: new Date(), + processingGameTick: suspension.cutTick, + processingClockRevision: suspension.sourceRevision, + processingDeadlineGeneration: 999n, + lockedBy: 'unification-clock-fixture', + leaseUntil: new Date(Date.now() + 60_000), + attempts: 1, + }, + }); + const resumeWallAt = new Date(suspension.cutWallAt.getTime() + 36 * 60 * 60_000); + await db.turnDaemonLease.create({ + data: { + profile: profileName, + ownerId: 'unification-clock-fixture', + fencingEpoch: 1n, + leaseUntil: new Date(Date.now() + 60_000), + }, + }); + const commandHandler = createTurnDaemonCommandHandler({ + world, + reservedTurns, + scenarioMeta: loaded.snapshot.scenarioMeta, + map: loaded.snapshot.map, + loadArchivedNationMaxId: async () => fixtureId, + reconcileUnificationWait: (input) => + reconcileClockSuspensionInTransaction({ + ...input, + allowUnificationWait: true, + testResumeWallAt: resumeWallAt, + }), + }); + const command = { + type: 'messageRespond' as const, + requestId, + userId, + generalId: fixtureId, + messageId: invaderPrompt!.id, + response: true, + }; + const executeCommand = () => + hooks.hooks.executeCommand!(requestId, async (context) => { + const result = await commandHandler.handle(command, { + ...context, + clockOperationAuthority: { + kind: 'DAEMON', + profileName, + ownerId: 'unification-clock-fixture', + fencingEpoch: 1n, + }, + }); + if (!result) throw new Error('Fixture command was not handled.'); + return result; + }); + const beforeFailedInvader = world.captureState(); + await expect(stateManager.transaction(executeCommand)).rejects.toThrow( + 'Input event processing clock fence changed before commit' + ); + expect(world.captureState()).toEqual(beforeFailedInvader); + expect(await db.nation.count({ where: { id: invaderNationId } })).toBe(0); + expect(await db.clockProjectionOutbox.count({ where: { suspensionId: suspension.id } })).toBe(0); + expect(await db.clockSuspension.findUniqueOrThrow({ where: { id: suspension.id } })).toMatchObject({ + status: 'SUSPENDED', + sourceRevision: 1n, + }); + await db.inputEvent.update({ + where: { requestId }, + data: { processingDeadlineGeneration: 1n }, + }); + const commandResult = await stateManager.transaction(executeCommand); + expect(commandResult).toMatchObject({ type: 'messageRespond', ok: true, action: 'raiseInvader' }); + const reconciledWorld = await db.worldState.findUniqueOrThrow({ where: { id: worldRow.id } }); + const appliedSuspension = await db.clockSuspension.findUniqueOrThrow({ where: { id: suspension.id } }); + expect(reconciledWorld).toMatchObject({ + clockPhase: 'RECONCILING', + clockRevision: 2n, + deadlineGeneration: 2n, + tickSeconds: 1_200, + meta: expect.objectContaining({ isUnited: 1, isunited: 1 }), + }); + expect(appliedSuspension).toMatchObject({ + status: 'RECONCILING', + gapTicks: BigInt(36 * 60 * 60 * 60_000), + shiftTicks: BigInt(36 * 60 * 60 * 60_000), + }); + expect(await db.nation.findUniqueOrThrow({ where: { id: invaderNationId } })).toMatchObject({ + name: 'ⓞ남만족', + capitalCityId: invaderCityId, + }); + expect(await db.general.count({ where: { id: { in: invaderGeneralIds } } })).toBe(10); + const invaderTurns = await db.general.findMany({ + where: { id: { in: invaderGeneralIds } }, + select: { turnTick: true }, + orderBy: { id: 'asc' }, + }); + expect( + invaderTurns.every((entry) => entry.turnTick !== null && entry.turnTick > reconciledWorld.clockTick!) + ).toBe(true); + expect( + invaderTurns.every( + (entry) => + entry.turnTick !== null && + entry.turnTick <= reconciledWorld.clockTick! + BigInt(GAME_TICKS_PER_TURN) + ) + ).toBe(true); + expect(await db.inputEvent.findUniqueOrThrow({ where: { requestId } })).toMatchObject({ + status: 'SUCCEEDED', + processingClockRevision: 1n, + processingDeadlineGeneration: 1n, + }); + expect( + await db.clockProjectionOutbox.findFirstOrThrow({ where: { suspensionId: suspension.id } }) + ).toMatchObject({ status: 'PENDING', targetRevision: 2n }); + + if (process.env.REDIS_URL) { + const redis = createRedisConnector({ url: process.env.REDIS_URL }); + await redis.connect(); + const prefix = `sammo:${profileName}`; + try { + await redis.client.del([ + `${prefix}:clock:active-revision`, + `${prefix}:clock:deadline-generation`, + `${prefix}:clock:projection-checksum`, + `${prefix}:clock:phase`, + `${prefix}:auction:timer`, + `${prefix}:tournament:state`, + ]); + await redis.client.set(`${prefix}:clock:active-revision`, '1'); + await expect( + applyNextClockProjection({ db, redis: redis.client, workerId: 'unification-clock-fixture' }) + ).resolves.toBe('APPLIED'); + expect(await db.worldState.findUniqueOrThrow({ where: { id: worldRow.id } })).toMatchObject({ + clockPhase: 'RUNNING', + clockRevision: 2n, + deadlineGeneration: 2n, + }); + world.completeClockReconciliation(); + expect(world.getGameClockState().phase).toBe('RUNNING'); + } finally { + await redis.client.del([ + `${prefix}:clock:active-revision`, + `${prefix}:clock:deadline-generation`, + `${prefix}:clock:projection-checksum`, + `${prefix}:clock:phase`, + `${prefix}:auction:timer`, + `${prefix}:tournament:state`, + ]); + await redis.disconnect(); + } + } } finally { await hooks.close(); } diff --git a/app/game-engine/test/unificationHandler.test.ts b/app/game-engine/test/unificationHandler.test.ts index aabd524f..94c800b3 100644 --- a/app/game-engine/test/unificationHandler.test.ts +++ b/app/game-engine/test/unificationHandler.test.ts @@ -149,6 +149,8 @@ describe('unification handler', () => { currentMonth: 6, tickSeconds: 600, lastTurnTime: new Date('0190-06-01T00:00:00.000Z'), + clockMode: 'realtime', + clockPhase: 'RUNNING', meta: { serverId: 'server-1', refreshLimit: 2 }, }; const snapshot: TurnWorldSnapshot = { @@ -203,6 +205,8 @@ describe('unification handler', () => { currentGeneralCount: 1, }, }); + expect(world.getGameClockState().phase).toBe('SUSPENDED'); + expect(world.getState().meta.unificationClockSuspensionId).toMatch(/^unification-wait-[a-f0-9]{32}$/); expect(world.getGeneralById(1)).toMatchObject({ inheritancePoints: { previous: 150, unifier: 2007, tournament: 11 }, meta: { inherit_earned_dyn: 2162.1, inherit_earned: 2167.1, inherit_spent: 20 }, diff --git a/app/game-engine/test/unificationInvaderResume.test.ts b/app/game-engine/test/unificationInvaderResume.test.ts index e9b04cdc..a174b54b 100644 --- a/app/game-engine/test/unificationInvaderResume.test.ts +++ b/app/game-engine/test/unificationInvaderResume.test.ts @@ -147,6 +147,9 @@ describe('HWE-shaped unification invader resume', () => { clockMode: 'realtime', clockWallAnchor: liveWallAnchor, lastTurnTick: 19_944_000_000, + clockPhase: 'SUSPENDED', + clockRevision: 1, + deadlineGeneration: 1, meta: { hiddenSeed: 'hwe-invader-resume-fixture', serverId: 'hwe:default_snapshot', @@ -154,6 +157,7 @@ describe('HWE-shaped unification invader resume', () => { isunited: 2, refreshLimit: 3_000, maxGeneralsPerMinute: 1_000, + unificationClockSuspensionId: 'unification-wait-fixture', }, }; const snapshot: TurnWorldSnapshot = { @@ -232,6 +236,18 @@ describe('HWE-shaped unification invader resume', () => { }, map, loadArchivedNationMaxId: async () => 57, + reconcileUnificationWait: async () => ({ + suspensionId: 'unification-wait-fixture', + phase: 'RECONCILING', + sourceRevision: 1, + targetRevision: 2, + deadlineGeneration: 2, + gapTicks: 108_000_000, + catchUpTicks: 0, + shiftTicks: 108_000_000, + alignedTick: 20_088_000_000, + resumeWallAt: acceptedAt, + }), }); const processor = new InMemoryTurnProcessor(world); const clock = new ManualClock(acceptedAt.getTime()); @@ -253,7 +269,19 @@ describe('HWE-shaped unification invader resume', () => { processor: { run }, commandHandler, hooks: { - executeCommand: async (_commandRequestId, execute) => execute({ db: commandDb }), + executeCommand: async (_commandRequestId, execute) => { + const result = await execute({ + db: commandDb, + clockOperationAuthority: { + kind: 'DAEMON', + profileName: 'hwe:default-snapshot', + ownerId: 'fixture-daemon', + fencingEpoch: 1n, + }, + }); + world.completeClockReconciliation(); + return result; + }, }, }, { @@ -264,9 +292,7 @@ describe('HWE-shaped unification invader resume', () => { await lifecycle.start(); - expect(commandDb.inputEvent.findUnique).toHaveBeenCalledWith( - expect.objectContaining({ where: { requestId } }) - ); + expect(commandDb.inputEvent.findUnique).toHaveBeenCalledWith(expect.objectContaining({ where: { requestId } })); expect(commandDb.$queryRaw).toHaveBeenCalledOnce(); expect(world.getState()).toMatchObject({ currentYear: 226, diff --git a/app/gateway-api/src/orchestrator/gatewayOrchestrator.ts b/app/gateway-api/src/orchestrator/gatewayOrchestrator.ts index 374ae30e..cce492e4 100644 --- a/app/gateway-api/src/orchestrator/gatewayOrchestrator.ts +++ b/app/gateway-api/src/orchestrator/gatewayOrchestrator.ts @@ -5,12 +5,12 @@ import { createHash, randomBytes, randomUUID } from 'node:crypto'; import { stripVTControlCharacters } from 'node:util'; import type { ScenarioInstallOptions } from '@sammo-ts/game-engine/scenario/scenarioSeeder.js'; +import { applyNextClockProjection } from '@sammo-ts/game-engine/turn/clockProjectionOutbox.js'; import { - applyNextClockProjection, reconcileClockSuspension, startClockSuspension, type ClockOperationAuthority, -} from '@sammo-ts/game-engine'; +} from '@sammo-ts/game-engine/turn/clockReconciliation.js'; import { cancelGame as defaultCancelGame, GAME_CANCELLATION_GENERAL_MODES, diff --git a/docs/architecture/game-clock-participants.json b/docs/architecture/game-clock-participants.json index b85bbbb4..b1b41e5d 100644 --- a/docs/architecture/game-clock-participants.json +++ b/docs/architecture/game-clock-participants.json @@ -163,11 +163,16 @@ }, { "key": "unification-wait", - "policy": "FORBID", - "authorityFields": ["world_state.meta.isunited", "world_state.meta.lastTurnTime"], - "projectionFields": [], + "policy": "REBUILD", + "authorityFields": [ + "world_state.meta.isunited", + "world_state.meta.unificationClockSuspensionId", + "clock_suspension.cut_tick", + "clock_suspension.cut_wall_at" + ], + "projectionFields": ["world_state.meta.lastTurnTime"], "owner": "game-engine/unification", - "migration": "Replace the lastTurnTime workaround with a durable UNIFICATION_WAIT suspension." + "migration": "Implemented as one exact alignment, optional rate change, invader creation, and revisioned outbox transaction." }, { "key": "clock-operation-ledger", @@ -190,7 +195,7 @@ { "keyPattern": "sammo:{profile}:clock:active-revision", "policy": "REBUILD", - "status": "planned" + "status": "implemented-with-db-phase-and-deadline-generation-fence" }, { "keyPattern": "sammo:{profile}:auction:timer", diff --git a/docs/architecture/game-clock-reconciliation.md b/docs/architecture/game-clock-reconciliation.md index bad247e9..54e51878 100644 --- a/docs/architecture/game-clock-reconciliation.md +++ b/docs/architecture/game-clock-reconciliation.md @@ -50,6 +50,23 @@ The authoritative registry is architecture gate rejects a new tick/revision field that is absent from that inventory. +## Unification wait + +A unification month with an invader choice changes `RUNNING -> SUSPENDED` and +persists a deterministic `UNIFICATION_WAIT` suspension in the same transaction +as the archive, prompts, and final unification state. Only a `raiseInvader` +message response tied to that active suspension may pass the suspended command +queue; all other gameplay remains pending. + +The response transaction verifies daemon authority, performs the exact +alignment, applies all participant shifts, optionally changes the turn rate, +then creates the invader nation, deterministic general IDs/RNG results, first +turns, and the final target-revision outbox. The optional rate change refreshes +the outbox with the final base/rate before commit. DB remains `RECONCILING` +until the daemon projection worker applies Redis and verifies the target +revision/generation. Games without an invader choice move directly to +`COMPLETED`. + ## DB to Redis boundary The database transaction leaves the phase `RECONCILING` and creates exactly one @@ -114,9 +131,7 @@ future anchored realtime profiles as `PREOPEN`, and other profiles as `RUNNING`. Existing DateTime columns remain projections while tick columns are authoritative. -Exact reconciliation stays disabled while an active registry participant is -`FORBID`. Tournament writes now carry tick/revision/generation coordinates and -are revision-fenced in Redis. The remaining unification wait participant must -be moved from its `lastTurnTime` workaround to a durable suspension before that -workflow can reach `RUNNING`. Removing this guard to make a partial operation -pass is prohibited. +No active participant remains `FORBID`. Tournament writes carry +tick/revision/generation coordinates and are revision-fenced in Redis. +Unification wait uses the same durable ledger and outbox boundary; the former +temporary `lastTurnTime` save/restore workaround is not part of the workflow. diff --git a/docs/developer/game-clock-reconciliation-plan.md b/docs/developer/game-clock-reconciliation-plan.md index f08e0b11..7ab3ea5e 100644 --- a/docs/developer/game-clock-reconciliation-plan.md +++ b/docs/developer/game-clock-reconciliation-plan.md @@ -46,10 +46,10 @@ mean deployment or production validation. - [x] All durable input events record accepted tick and accepted revision. - [x] Processing converts accepted coordinates across revisions or fails closed. - [x] Gateway pause/resume/open orchestration writes the DB clock phase. -- [ ] Unification wait becomes a durable `UNIFICATION_WAIT` suspension. -- [ ] Alignment, optional rate change, invader IDs/RNG, creation, first schedule, +- [x] Unification wait becomes a durable `UNIFICATION_WAIT` suspension. +- [x] Alignment, optional rate change, invader IDs/RNG, creation, first schedule, outbox, verification, and RUNNING transition form one retry-safe workflow. -- [ ] Multi-host drift and general-access/clock-operation deadlock tests. +- [x] Multi-host drift and general-access/clock-operation deadlock tests. ## Milestone 5 - test-branch release gate @@ -57,11 +57,11 @@ mean deployment or production validation. integration suites. - [ ] Dedicated PostgreSQL/Redis conditional integration suite with skip count recorded. -- [ ] Recovery runbook exercised from each incomplete status. +- [x] Recovery runbook exercised from each incomplete status. - [x] Admin status/readiness exposes revision, phase, participant checksums, and incomplete outbox state. - [ ] User-test deployment evidence is recorded separately from Git push. -- [ ] All `FORBID` inventory entries are removed by typed migrations or proven +- [x] All `FORBID` inventory entries are removed by typed migrations or proven inactive preconditions. ## Evidence log @@ -112,3 +112,17 @@ mean deployment or production validation. to `RUNNING`; the Redis clock phase is revision/generation fenced. - `pnpm --filter @sammo-ts/gateway-api test` passed 313 tests with 35 environment-conditional skips. Gateway typecheck and target lint passed. + +### 2026-09-03 - atomic unification wait + +- A unification flush now commits the finalization, actionable prompts, and one + deterministic `UNIFICATION_WAIT` suspension together. A late archive failure + rolls the whole boundary back; retry creates one ledger. +- The suspended command queue admits only an invader decision tied to the + active ledger. The daemon-authorized command transaction applies a 36-hour + exact gap, preserves participant positions, changes the fixture rate from 10 + to 20 minutes, creates one invader nation and ten deterministic generals with + future first turns, and writes one final-rate projection outbox. +- The dedicated PostgreSQL/Redis fixture reached `RUNNING@2/2` only after the + Redis projection. DB-wall versus a mocked 12-hour host drift and concurrent + general-access lock acquisition completed without drift or deadlock. diff --git a/docs/developer/game-clock-recovery.md b/docs/developer/game-clock-recovery.md index 311d223b..6ad668dd 100644 --- a/docs/developer/game-clock-recovery.md +++ b/docs/developer/game-clock-recovery.md @@ -28,6 +28,13 @@ service. The service must re-read participant checksums and either return the already-applied result or resume the pending outbox. Never create a replacement revision to hide a failed target revision. +For `UNIFICATION_WAIT`, never rerun invader creation as a separate repair. +The input event, aligned schedules, optional rate, deterministic invader IDs, +reserved turns, and outbox committed together. A committed command with a +`RECONCILING` world therefore needs only the same outbox retry. If the command +transaction rolled back, the original prompt and source revision remain and +the same response can be retried without changing IDs or RNG results. + When an outbox row is `FAILED`, the profile must remain `RECONCILING`. A retry is safe in both crash locations: @@ -48,5 +55,7 @@ If participant verification shows an unexpected mutation, stop the profile, retain the ledger/outbox evidence, and restore the whole game schema from that backup. Redis projections are then rebuilt from the restored DB revision. -The implementation-plan release gate remains open until these steps have an -automated fixture and an operator-facing status endpoint. +The conditional clock suite exercises `SUSPENDED`, `RECONCILING/PENDING`, +`RECONCILING/FAILED` before and after Redis commit, recovered `APPLIED`, and +final `RUNNING`. The admin status endpoint exposes the phase, revision, +participant checksums, and outbox error needed to choose the matching step.