fix(game-engine): sync clock authority during pause

This commit is contained in:
2026-09-03 17:50:26 +00:00
parent b250e908f4
commit 1cb66f76fb
10 changed files with 766 additions and 172 deletions
@@ -171,7 +171,8 @@ export class DatabaseTurnDaemonCommandQueue implements TurnDaemonControlQueue, T
'changePermission',
'appoint',
'setNationSetting',
'setNpcPolicy'
'setNpcPolicy',
'shiftSchedule'
)
)
OR (
+196 -152
View File
@@ -113,6 +113,41 @@ const readDbWall = async (db: GamePrisma.TransactionClient): Promise<Date> => {
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 <T>(operation: () => Promise<T>): Promise<T> => {
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<void>((resolve) => setTimeout(resolve, attempt * 10));
}
}
};
const verifyAuthority = async (db: GamePrisma.TransactionClient, authority: ClockOperationAuthority): Promise<void> => {
const rows = await db.$queryRaw<LeaseFenceRow[]>(GamePrisma.sql`
SELECT owner_id AS "ownerId",
@@ -185,64 +220,64 @@ const readParticipantSnapshots = async (
): Promise<ParticipantSnapshot[]> => {
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<ClockReconciliationResult> =>
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 }
)
);
@@ -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<void>;
applyClockProjection(redis: ClockProjectionRedis, workerId: string): Promise<boolean>;
synchronizeClockAuthority(): Promise<boolean>;
}
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(),
};
};
+83 -12
View File
@@ -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<DurableClockReconciliationAlignment, 'sourceRevision'>): 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' };
@@ -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<boolean> => {
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()
);
};
+18 -1
View File
@@ -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
@@ -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);
});
});
@@ -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,
});
@@ -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<string, unknown>, 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/
);
});
});
@@ -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<void> => {
const rotateFirst = <T>(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<void> => {
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<void> => {
const state = await readState();
if (state.users?.length !== 10 || !state.actionFixture) {
@@ -986,8 +1044,20 @@ const exercisePausedActions = async (): Promise<void> => {
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<void> => {
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 <reset|wait-reset|deploy|wait-deploy|status|prepare-users|preopen-messages|verified-preopen-messages|repair-preopen-fixture|database-status|prepare-paused-betting|submit-paused-betting|reserve-user-enlistments|prepare-action-fixture|exercise-paused-actions|verify-paused-wall-expiry|exercise-paused-tournament-bet|reserve-actionable-commands|wait-actionable-messages|respond-actionable-messages|reserve-no-aggression-cancellation|wait-no-aggression-cancellation|respond-no-aggression-cancellation|npc-action-audit|repair-opening-clock-runtime|verify-monitor-message|resume-daemon|fast-forward|place-invader-recipients|respond-invader-browser|action|wait-profile-status|wait-runtime-action|monitor-users>'
'usage: live-ten-user-lifecycle.ts <reset|wait-reset|deploy|wait-deploy|status|prepare-users|preopen-messages|verified-preopen-messages|repair-preopen-fixture|database-status|prepare-paused-betting|submit-paused-betting|reserve-user-enlistments|prepare-action-fixture|repair-action-item-fixture|exercise-paused-actions|verify-paused-wall-expiry|exercise-paused-tournament-bet|reserve-actionable-commands|wait-actionable-messages|respond-actionable-messages|reserve-no-aggression-cancellation|wait-no-aggression-cancellation|respond-no-aggression-cancellation|npc-action-audit|repair-opening-clock-runtime|verify-monitor-message|resume-daemon|fast-forward|place-invader-recipients|respond-invader-browser|action|wait-profile-status|wait-runtime-action|monitor-users>'
);