feat: 통일 대기 시계를 원자적 reconciliation으로 전환

This commit is contained in:
2026-09-03 10:18:44 +00:00
parent a9f23703d9
commit 69ac6028df
22 changed files with 1300 additions and 251 deletions
+8
View File
@@ -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"
@@ -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}
+6
View File
@@ -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 {
@@ -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<number>;
clockOperationAuthority?: {
kind: 'DAEMON';
profileName: string;
ownerId: string;
fencingEpoch: bigint;
};
reconcileUnificationWait?: (input: {
db: GamePrisma.TransactionClient;
suspensionId: string;
profileName: string;
authority: NonNullable<Parameters<typeof reconcileClockSuspensionInTransaction>[0]['authority']>;
}) => Promise<ClockReconciliationResult>;
}): Promise<ActionableMessageResponseResult> => {
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<number>;
clockOperationAuthority?: {
kind: 'DAEMON';
profileName: string;
ownerId: string;
fencingEpoch: bigint;
};
reconcileUnificationWait?: (input: {
db: GamePrisma.TransactionClient;
suspensionId: string;
profileName: string;
authority: NonNullable<Parameters<typeof reconcileClockSuspensionInTransaction>[0]['authority']>;
}) => Promise<ClockReconciliationResult>;
}): Promise<ActionableMessageResponseResult> => {
const acceptedAt = await validateActor(options);
const now = options.world.getGameNow(acceptedAt);
@@ -8,7 +8,7 @@ import {
type GamePrismaClient,
} from '@sammo-ts/infra';
interface ClockProjectionRedis {
export interface ClockProjectionRedis {
get(key: string): Promise<string | null>;
eval(script: string, options: { keys: string[]; arguments: string[] }): Promise<unknown>;
}
@@ -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);
+302 -169
View File
@@ -111,10 +111,9 @@ const readDbWall = async (db: GamePrisma.TransactionClient): Promise<Date> => {
return wallNow;
};
const verifyAuthority = async (
db: GamePrisma.TransactionClient,
authority: ClockOperationAuthority
): Promise<void> => {
export const readClockDatabaseWall = readDbWall;
const verifyAuthority = async (db: GamePrisma.TransactionClient, authority: ClockOperationAuthority): Promise<void> => {
const rows = await db.$queryRaw<LeaseFenceRow[]>(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<void> => {
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<ClockReconciliationResult> => {
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<string, unknown>)
: {};
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<void> => {
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<string, unknown>) }
: 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<string, unknown>)
: {};
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 }
);
+141 -9
View File
@@ -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<void>;
applyClockProjection(redis: ClockProjectionRedis, workerId: string): Promise<boolean>;
}
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(),
};
};
+110
View File
@@ -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
@@ -133,6 +133,7 @@ export const createRaiseInvaderHandler = (options: {
env: TurnCommandEnv;
loadArchivedNationMaxId?: (serverId: string) => Promise<number>;
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
+57 -4
View File
@@ -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<boolean>;
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<void> => {
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) {
@@ -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);
@@ -146,6 +146,10 @@ const refreshActorKillturn = (world: InMemoryTurnWorld, actor: TurnGeneral): voi
interface CommandHandlerContext {
world: InMemoryTurnWorld;
commandDb?: GamePrisma.TransactionClient;
clockOperationAuthority?: Extract<
NonNullable<TurnDaemonCommandExecutionContext['clockOperationAuthority']>,
{ kind: 'DAEMON' }
>;
auctionFinalizer?: AuctionFinalizer;
auctionBidder?: AuctionBidder;
tournamentRewardFinalizer?: TournamentRewardFinalizer;
@@ -153,6 +157,7 @@ interface CommandHandlerContext {
reservedTurns?: InMemoryReservedTurnStore;
generalActionModules?: ReadonlyArray<GeneralActionModule>;
loadArchivedNationMaxId?: (serverId: string) => Promise<number>;
reconcileUnificationWait?: Parameters<typeof respondToActionableMessage>[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<number>;
reconcileUnificationWait?: Parameters<typeof respondToActionableMessage>[0]['reconcileUnificationWait'];
}): TurnDaemonCommandHandler => {
let immediateGeneralActionExecutor: Promise<ImmediateGeneralActionExecutor> | 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;
}
},
};
@@ -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<void>) | 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<void> => {
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<Array<{ now: Date }>>(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<void>((resolve) => {
signalWriterLocked = resolve;
});
const writerRelease = new Promise<void>((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<never>((_, 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();
}
});
});
@@ -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' });
});
});
@@ -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<void>) | undefined;
const cleanup = async (): Promise<void> => {
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<string, MonthlyEventActionHandler>();
@@ -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();
}
@@ -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 },
@@ -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,
@@ -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,
+10 -5
View File
@@ -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",
+21 -6
View File
@@ -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.
@@ -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.
+11 -2
View File
@@ -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.