VM 복귀 후 예약턴 잠금 복구와 Windows 부팅 식별자 확인을 추가한다
This commit is contained in:
Hide_D committed
1 parent
fd8145d6eb
commit
f36bc2c8d7
24 files changed
+1048
-28
No files matched your search
@@ -2,6 +2,7 @@ import { randomUUID } from 'node:crypto';
|
||||
import { performance } from 'node:perf_hooks';
|
||||
|
||||
import { createGamePostgresConnector, GamePrisma, type GamePrismaClient } from '@sammo-ts/infra';
|
||||
import type { HostBootMonitor } from '../turn/hostBootMonitor.js';
|
||||
|
||||
export interface TurnDaemonLeaseToken {
|
||||
profile: string;
|
||||
@@ -14,6 +15,7 @@ export interface DatabaseTurnDaemonLeaseOptions {
|
||||
ownerId?: string;
|
||||
leaseDurationMs?: number;
|
||||
heartbeat?: boolean;
|
||||
hostBootMonitor?: HostBootMonitor;
|
||||
}
|
||||
|
||||
export class TurnDaemonLeaseUnavailableError extends Error {
|
||||
@@ -45,6 +47,7 @@ export class DatabaseTurnDaemonLease {
|
||||
private readonly ownerId: string;
|
||||
private readonly leaseDurationMs: number;
|
||||
private readonly heartbeatEnabled: boolean;
|
||||
private readonly hostBootMonitor?: HostBootMonitor;
|
||||
private token: TurnDaemonLeaseToken | null = null;
|
||||
private heartbeatTimer: NodeJS.Timeout | null = null;
|
||||
private expiryTimer: NodeJS.Timeout | null = null;
|
||||
@@ -63,6 +66,7 @@ export class DatabaseTurnDaemonLease {
|
||||
this.ownerId = options.ownerId ?? randomUUID();
|
||||
this.leaseDurationMs = normalizeLeaseDuration(options.leaseDurationMs);
|
||||
this.heartbeatEnabled = options.heartbeat ?? true;
|
||||
this.hostBootMonitor = options.hostBootMonitor;
|
||||
}
|
||||
|
||||
static async connect(
|
||||
@@ -75,6 +79,8 @@ export class DatabaseTurnDaemonLease {
|
||||
}
|
||||
|
||||
async acquire(): Promise<TurnDaemonLeaseToken | null> {
|
||||
await this.checkHostBoot();
|
||||
if (this.lost) throw this.getLossError();
|
||||
const requestStartedAt = performance.now();
|
||||
const rows = await this.db.$queryRaw<LeaseRow[]>(GamePrisma.sql`
|
||||
INSERT INTO "turn_daemon_lease" (
|
||||
@@ -145,6 +151,12 @@ export class DatabaseTurnDaemonLease {
|
||||
return new TurnDaemonLeaseLostError(this.profile, this.lossReason);
|
||||
}
|
||||
|
||||
async checkHostBoot(): Promise<void> {
|
||||
if (!this.lost && (await this.hostBootMonitor?.hasChanged())) {
|
||||
this.markLost('Host boot identity changed; restart with a fresh database snapshot.');
|
||||
}
|
||||
}
|
||||
|
||||
async renew(): Promise<boolean> {
|
||||
const token = this.token;
|
||||
if (!token || this.lost || this.renewalInFlight) {
|
||||
@@ -153,6 +165,8 @@ export class DatabaseTurnDaemonLease {
|
||||
const requestStartedAt = performance.now();
|
||||
this.renewalInFlight = true;
|
||||
try {
|
||||
await this.checkHostBoot();
|
||||
if (this.lost) return false;
|
||||
const rows = await this.db.$queryRaw<LeaseRow[]>(GamePrisma.sql`
|
||||
UPDATE "turn_daemon_lease"
|
||||
SET
|
||||
@@ -180,6 +194,7 @@ export class DatabaseTurnDaemonLease {
|
||||
}
|
||||
|
||||
async assertActive(transaction?: GamePrisma.TransactionClient): Promise<void> {
|
||||
await this.checkHostBoot();
|
||||
const token = this.token;
|
||||
if (!token || this.lost) {
|
||||
throw this.getLossError();
|
||||
@@ -207,12 +222,13 @@ export class DatabaseTurnDaemonLease {
|
||||
this.stopExpiryWatchdog();
|
||||
const token = this.token;
|
||||
this.token = null;
|
||||
if (!token || this.lost) {
|
||||
if (!token) {
|
||||
return;
|
||||
}
|
||||
await this.db.$executeRaw(GamePrisma.sql`
|
||||
UPDATE "turn_daemon_lease"
|
||||
SET "lease_until" = CURRENT_TIMESTAMP AT TIME ZONE 'UTC',
|
||||
SET "clock_ready" = FALSE,
|
||||
"lease_until" = CURRENT_TIMESTAMP AT TIME ZONE 'UTC',
|
||||
"heartbeat_at" = CURRENT_TIMESTAMP AT TIME ZONE 'UTC'
|
||||
WHERE
|
||||
"profile" = ${token.profile}
|
||||
|
||||
@@ -45,6 +45,7 @@ export interface TurnDaemonLifecycleDeps {
|
||||
commandResponder?: TurnDaemonCommandResponder;
|
||||
pauseGate?: () => Promise<boolean>;
|
||||
stateManager?: Pick<EngineStateManager, 'transaction'>;
|
||||
runErrorRetry?: { wait(error: unknown): Promise<boolean>; reset(): void };
|
||||
}
|
||||
|
||||
export class TurnDaemonLifecycle {
|
||||
@@ -59,6 +60,7 @@ export class TurnDaemonLifecycle {
|
||||
private readonly commandResponder?: TurnDaemonCommandResponder;
|
||||
private readonly pauseGate?: () => Promise<boolean>;
|
||||
private readonly stateManager?: Pick<EngineStateManager, 'transaction'>;
|
||||
private readonly runErrorRetry?: TurnDaemonLifecycleDeps['runErrorRetry'];
|
||||
private readonly options: TurnDaemonLifecycleOptions;
|
||||
|
||||
private status: TurnDaemonStatus;
|
||||
@@ -79,6 +81,7 @@ export class TurnDaemonLifecycle {
|
||||
this.commandResponder = deps.commandResponder;
|
||||
this.pauseGate = deps.pauseGate;
|
||||
this.stateManager = deps.stateManager;
|
||||
this.runErrorRetry = deps.runErrorRetry;
|
||||
this.options = options;
|
||||
this.status = {
|
||||
state: 'idle',
|
||||
@@ -192,8 +195,9 @@ export class TurnDaemonLifecycle {
|
||||
// 수동 실행 요청도 가오픈·정지·재조정의 턴 실행 gate를 통과해야 한다.
|
||||
// 사용자 명령 처리는 루프 시작에서 계속하되 시간 진행은 여기서 분리한다.
|
||||
if (this.pendingRun) {
|
||||
await this.runOnce(this.pendingRun);
|
||||
const pending = this.pendingRun;
|
||||
this.pendingRun = null;
|
||||
await this.runOnce(pending);
|
||||
continue;
|
||||
}
|
||||
|
||||
@@ -447,6 +451,15 @@ export class TurnDaemonLifecycle {
|
||||
};
|
||||
result = this.stateManager ? await this.stateManager.transaction(runAndFlush) : await runAndFlush();
|
||||
} catch (error) {
|
||||
// savepoint 복원이 성공한 획득 충돌만 대기한다. 복원 실패/AggregateError나
|
||||
// 저장 중 소유권 상실을 자동 재실행하면 중복 효과를 숨길 수 있다.
|
||||
if (this.stateManager && (await this.runErrorRetry?.wait(error))) {
|
||||
this.status.running = false;
|
||||
this.status.state = 'idle';
|
||||
this.pendingRun = pending;
|
||||
return;
|
||||
}
|
||||
this.runErrorRetry?.reset();
|
||||
this.status.running = false;
|
||||
this.status.state = 'paused';
|
||||
this.status.paused = true;
|
||||
@@ -459,6 +472,7 @@ export class TurnDaemonLifecycle {
|
||||
}
|
||||
|
||||
await this.applyRunResult(result, startMs);
|
||||
this.runErrorRetry?.reset();
|
||||
this.status.state = 'idle';
|
||||
try {
|
||||
await this.hooks?.publishEvents?.(result);
|
||||
|
||||
@@ -8,6 +8,7 @@ import { createTurnDaemonMemoryReporter } from './turnDaemonMemoryReporter.js';
|
||||
import { createGatewayProfileGate } from './gatewayProfileGate.js';
|
||||
import { retryTurnDaemonLeaseStartup } from './leaseStartupRetry.js';
|
||||
import { TurnDaemonLeaseUnavailableError } from '../lifecycle/databaseTurnDaemonLease.js';
|
||||
import { loadHostBootMonitor } from './hostBootMonitor.js';
|
||||
|
||||
export interface TurnDaemonCliOptions {
|
||||
profile?: string;
|
||||
@@ -80,6 +81,7 @@ export const runTurnDaemonCli = async (options: TurnDaemonCliOptions = {}): Prom
|
||||
throw new Error(`GAME_CLOCK_MODE must be realtime or manual: ${rawGameClockMode}`);
|
||||
}
|
||||
const gameClockMode = rawGameClockMode as GameClockMode | undefined;
|
||||
const hostBootMonitor = await loadHostBootMonitor(env);
|
||||
|
||||
const runtime = await retryTurnDaemonLeaseStartup(() =>
|
||||
createTurnDaemonRuntime({
|
||||
@@ -95,6 +97,7 @@ export const runTurnDaemonCli = async (options: TurnDaemonCliOptions = {}): Prom
|
||||
pauseGateIntervalMs,
|
||||
adminActionIntervalMs,
|
||||
gameClockMode,
|
||||
hostBootMonitor,
|
||||
})
|
||||
).catch(async (error: unknown) => {
|
||||
// 중복 starter가 정상 owner를 멈추면 안 된다. 그 밖의 초기화 실패는
|
||||
|
||||
@@ -0,0 +1,142 @@
|
||||
import { createHmac, randomBytes, timingSafeEqual } from 'node:crypto';
|
||||
import { readFile } from 'node:fs/promises';
|
||||
import { performance } from 'node:perf_hooks';
|
||||
import { isRecord } from '@sammo-ts/common';
|
||||
|
||||
export interface HostBootMonitor {
|
||||
hasChanged(): Promise<boolean>;
|
||||
}
|
||||
|
||||
const mac = (key: Buffer, value: string): string => createHmac('sha256', key).update(value).digest('hex');
|
||||
|
||||
/** 부팅 세대는 첫 인증 응답에 고정한다. 통신 실패는 기존 lease 복구를 막지 않는다. */
|
||||
export const createHostBootMonitor = (options: {
|
||||
url: string;
|
||||
key: Buffer;
|
||||
pollMs?: number;
|
||||
timeoutMs?: number;
|
||||
now?: () => number;
|
||||
wallNow?: () => number;
|
||||
onUnavailable?: () => void;
|
||||
onVerified?: () => void;
|
||||
}): HostBootMonitor => {
|
||||
const url = new URL(options.url);
|
||||
if (
|
||||
!['http:', 'https:'].includes(url.protocol) ||
|
||||
url.username ||
|
||||
url.password ||
|
||||
url.search ||
|
||||
url.hash ||
|
||||
url.pathname !== '/v1/boot-id'
|
||||
) {
|
||||
throw new Error('Host boot URL must use HTTP(S) and /v1/boot-id without credentials or query.');
|
||||
}
|
||||
if (options.key.length !== 32) throw new Error('Host boot key must contain exactly 32 bytes.');
|
||||
const now = options.now ?? (() => performance.now());
|
||||
const wallNow = options.wallNow ?? (() => Date.now());
|
||||
let baseline: string | null = null;
|
||||
let changed = false;
|
||||
let lastAttempt = -Infinity;
|
||||
let lastWall = 0;
|
||||
let lastWarning = -Infinity;
|
||||
let inFlight: Promise<boolean> | null = null;
|
||||
const sample = async (): Promise<boolean> => {
|
||||
const nonce = randomBytes(32).toString('hex');
|
||||
const response = await fetch(url, {
|
||||
method: 'GET',
|
||||
redirect: 'error',
|
||||
signal: AbortSignal.timeout(options.timeoutMs ?? 1_000),
|
||||
headers: {
|
||||
'x-sammo-challenge': nonce,
|
||||
'x-sammo-signature': mac(options.key, `sammo-host-boot-request-v1\n${nonce}`),
|
||||
},
|
||||
});
|
||||
if (!response.ok || !response.body) {
|
||||
await response.body?.cancel();
|
||||
throw new Error('Host boot response unavailable.');
|
||||
}
|
||||
const reader = response.body.getReader();
|
||||
const chunks: Uint8Array[] = [];
|
||||
let size = 0;
|
||||
try {
|
||||
for (;;) {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) break;
|
||||
size += value.length;
|
||||
if (size > 2_048) throw new Error('Host boot response exceeds its limit.');
|
||||
chunks.push(value);
|
||||
}
|
||||
} finally {
|
||||
await reader.cancel();
|
||||
}
|
||||
const body: unknown = JSON.parse(Buffer.concat(chunks).toString('utf8'));
|
||||
if (
|
||||
!isRecord(body) ||
|
||||
body.version !== 1 ||
|
||||
body.challenge !== nonce ||
|
||||
typeof body.bootId !== 'string' ||
|
||||
!/^[a-f0-9]{64}$/.test(body.bootId) ||
|
||||
typeof body.signature !== 'string' ||
|
||||
!/^[a-f0-9]{64}$/.test(body.signature)
|
||||
) {
|
||||
throw new Error('Host boot response is invalid.');
|
||||
}
|
||||
const expected = mac(options.key, `sammo-host-boot-response-v1\n${nonce}\n${body.bootId}`);
|
||||
if (!timingSafeEqual(Buffer.from(expected, 'hex'), Buffer.from(body.signature, 'hex'))) {
|
||||
throw new Error('Host boot response signature is invalid.');
|
||||
}
|
||||
if (baseline === null) {
|
||||
baseline = body.bootId;
|
||||
options.onVerified?.();
|
||||
}
|
||||
changed ||= body.bootId !== baseline;
|
||||
return changed;
|
||||
};
|
||||
return {
|
||||
hasChanged(): Promise<boolean> {
|
||||
if (changed) return Promise.resolve(true);
|
||||
if (inFlight) return inFlight;
|
||||
const current = now();
|
||||
const wall = wallNow();
|
||||
const discontinuity = Math.abs(wall - lastWall - (current - lastAttempt)) > 1_000;
|
||||
if (current - lastAttempt < (options.pollMs ?? 2_000) && !discontinuity) return Promise.resolve(false);
|
||||
lastAttempt = current;
|
||||
lastWall = wall;
|
||||
inFlight = sample()
|
||||
.catch(() => {
|
||||
if (current - lastWarning >= 60_000) {
|
||||
options.onUnavailable?.();
|
||||
lastWarning = current;
|
||||
}
|
||||
return false;
|
||||
})
|
||||
.finally(() => {
|
||||
inFlight = null;
|
||||
});
|
||||
return inFlight;
|
||||
},
|
||||
};
|
||||
};
|
||||
|
||||
export const loadHostBootMonitor = async (env: NodeJS.ProcessEnv): Promise<HostBootMonitor | undefined> => {
|
||||
const url = env.TURN_HOST_BOOT_URL?.trim();
|
||||
const keyFile = env.TURN_HOST_BOOT_KEY_FILE?.trim();
|
||||
if (!url && !keyFile) return undefined;
|
||||
try {
|
||||
if (!url || !keyFile) throw new Error('Incomplete optional configuration.');
|
||||
const encoded = (await readFile(keyFile, 'utf8')).trim();
|
||||
if (!/^[a-f0-9]{64}$/.test(encoded)) throw new Error('Invalid optional key file.');
|
||||
return createHostBootMonitor({
|
||||
url,
|
||||
key: Buffer.from(encoded, 'hex'),
|
||||
onVerified: () => console.info('[turn-daemon] host boot identity verified.'),
|
||||
onUnavailable: () => {
|
||||
console.warn('[turn-daemon] host boot identity unavailable; ordinary lease recovery remains active.');
|
||||
},
|
||||
});
|
||||
} catch {
|
||||
// 보조 관측기의 설정/파일 부재도 게임 기동의 필수 의존성이 되지 않는다.
|
||||
console.warn('[turn-daemon] host boot observation disabled; check optional configuration and key file.');
|
||||
return undefined;
|
||||
}
|
||||
};
|
||||
@@ -0,0 +1,38 @@
|
||||
import { performance } from 'node:perf_hooks';
|
||||
import { setTimeout } from 'node:timers/promises';
|
||||
import { ReservedTurnLeaseBusyError } from './reservedTurnStore.js';
|
||||
|
||||
/** rollback 완료 뒤에만 대기한다. 다음 loop에서 pause/lease gate와 최신 큐를 다시 읽는다. */
|
||||
export const createReservedTurnLeaseRetry = (
|
||||
options: {
|
||||
now?: () => number;
|
||||
sleep?: (ms: number) => Promise<void>;
|
||||
maxWaitMs?: number;
|
||||
onWait?: (queueKey: string) => void;
|
||||
} = {}
|
||||
) => {
|
||||
const now = options.now ?? (() => performance.now());
|
||||
const sleep = options.sleep ?? ((ms: number) => setTimeout(ms));
|
||||
const maxWaitMs = options.maxWaitMs ?? 5 * 60_000 + 1_000;
|
||||
let startedAt: number | null = null;
|
||||
let lastReportedAt = -Infinity;
|
||||
return {
|
||||
reset(): void {
|
||||
startedAt = null;
|
||||
lastReportedAt = -Infinity;
|
||||
},
|
||||
async wait(error: unknown): Promise<boolean> {
|
||||
if (!(error instanceof ReservedTurnLeaseBusyError)) return false;
|
||||
const current = now();
|
||||
startedAt ??= current;
|
||||
const remaining = maxWaitMs - (current - startedAt);
|
||||
if (remaining <= 0) return false;
|
||||
if (current - lastReportedAt >= 30_000) {
|
||||
options.onWait?.(error.queueKey);
|
||||
lastReportedAt = current;
|
||||
}
|
||||
await sleep(Math.min(1_000, remaining));
|
||||
return true;
|
||||
},
|
||||
};
|
||||
};
|
||||
@@ -129,6 +129,14 @@ export class ReservedTurnLeaseConflictError extends Error {
|
||||
}
|
||||
}
|
||||
|
||||
/** 실행 전 획득 충돌만 재시도한다. flush/finalize의 소유권 상실과 구분한다. */
|
||||
export class ReservedTurnLeaseBusyError extends ReservedTurnLeaseConflictError {
|
||||
constructor(queueKey: string) {
|
||||
super(queueKey);
|
||||
this.name = 'ReservedTurnLeaseBusyError';
|
||||
}
|
||||
}
|
||||
|
||||
export class InMemoryReservedTurnStore {
|
||||
private readonly generalTurns = new Map<number, ReservedTurnEntry[]>();
|
||||
private readonly nationTurns = new Map<string, ReservedTurnEntry[]>();
|
||||
@@ -249,11 +257,7 @@ export class InMemoryReservedTurnStore {
|
||||
}
|
||||
};
|
||||
pruneNationKeys(Array.from(this.nationTurns.keys()), (key) => this.nationTurns.delete(key));
|
||||
for (const keys of [
|
||||
this.dirtyNationKeys,
|
||||
this.pendingNationInitializationKeys,
|
||||
this.leasedNationKeys,
|
||||
]) {
|
||||
for (const keys of [this.dirtyNationKeys, this.pendingNationInitializationKeys, this.leasedNationKeys]) {
|
||||
for (const key of Array.from(keys)) {
|
||||
const nationId = Number(key.split(':', 1)[0]);
|
||||
if (deletedNations.has(nationId)) {
|
||||
@@ -376,7 +380,7 @@ export class InMemoryReservedTurnStore {
|
||||
}
|
||||
}
|
||||
if (claimed.count !== 1) {
|
||||
throw new ReservedTurnLeaseConflictError(`general:${generalId}`);
|
||||
throw new ReservedTurnLeaseBusyError(`general:${generalId}`);
|
||||
}
|
||||
this.leasedGeneralIds.add(generalId);
|
||||
return !retainedExistingLease;
|
||||
@@ -427,7 +431,7 @@ export class InMemoryReservedTurnStore {
|
||||
}
|
||||
}
|
||||
if (claimed.count !== 1) {
|
||||
throw new ReservedTurnLeaseConflictError(`nation:${nationId}:${officerLevel}`);
|
||||
throw new ReservedTurnLeaseBusyError(`nation:${nationId}:${officerLevel}`);
|
||||
}
|
||||
this.leasedNationKeys.add(buildNationKey(nationId, officerLevel));
|
||||
return !retainedExistingLease;
|
||||
@@ -904,6 +908,24 @@ export class InMemoryReservedTurnStore {
|
||||
await this.persistChanges(this.prisma, changes);
|
||||
this.acknowledgeDirtyState(changes);
|
||||
}
|
||||
|
||||
/** 실행 루프 종료 뒤 호출한다. rollback으로 journal에서 빠진 DB 임대도 해제한다. */
|
||||
async releaseOwnedLeases(): Promise<void> {
|
||||
const results = await Promise.allSettled([
|
||||
this.prisma.generalTurnRevision?.updateMany({
|
||||
where: { leaseOwner: this.leaseOwner },
|
||||
data: { leaseOwner: null, leaseExpiresAt: null },
|
||||
}),
|
||||
this.prisma.nationTurnRevision?.updateMany({
|
||||
where: { leaseOwner: this.leaseOwner },
|
||||
data: { leaseOwner: null, leaseExpiresAt: null },
|
||||
}),
|
||||
]);
|
||||
const errors = results.flatMap((result) => (result.status === 'rejected' ? [result.reason] : []));
|
||||
if (errors.length) throw new AggregateError(errors, 'Reserved turn lease cleanup failed.');
|
||||
this.leasedGeneralIds.clear();
|
||||
this.leasedNationKeys.clear();
|
||||
}
|
||||
}
|
||||
|
||||
export const createReservedTurnStore = async (options: ReservedTurnStoreOptions): Promise<ReservedTurnStoreHandle> => {
|
||||
@@ -918,6 +940,12 @@ export const createReservedTurnStore = async (options: ReservedTurnStoreOptions)
|
||||
await store.loadAll();
|
||||
return {
|
||||
store,
|
||||
close: () => connector.disconnect(),
|
||||
close: async () => {
|
||||
try {
|
||||
await store.releaseOwnedLeases();
|
||||
} finally {
|
||||
await connector.disconnect();
|
||||
}
|
||||
},
|
||||
};
|
||||
};
|
||||
@@ -2,7 +2,7 @@ import type { GameClockPhase } from '@sammo-ts/common';
|
||||
|
||||
/** Gateway의 실행 gate와 durable 시계를 명령 claim 전에 맞춘다. */
|
||||
export const createRuntimePauseGate = (options: {
|
||||
assertLease(): void;
|
||||
assertLease(): void | Promise<void>;
|
||||
shouldPause(): Promise<boolean>;
|
||||
getPhase(): GameClockPhase;
|
||||
isExplicitlyPaused(): boolean;
|
||||
@@ -11,7 +11,7 @@ export const createRuntimePauseGate = (options: {
|
||||
}): (() => Promise<boolean>) => {
|
||||
let lastPaused: boolean | null = null;
|
||||
return async () => {
|
||||
options.assertLease();
|
||||
await options.assertLease();
|
||||
const paused = await options.shouldPause();
|
||||
const phase = options.getPhase();
|
||||
// 오류 정지는 Gateway 상태만 PAUSED로 바꿀 수 있다. 그대로 두면 가입은
|
||||
|
||||
@@ -108,6 +108,8 @@ import { DatabaseTurnDaemonLease, TurnDaemonLeaseUnavailableError } from '../lif
|
||||
import { EngineStateManager } from './engineStateManager.js';
|
||||
import { applyRuntimeClockShift } from './runtimeClockShift.js';
|
||||
import { applyRuntimeGameSettings } from './runtimeGameSettings.js';
|
||||
import { createReservedTurnLeaseRetry } from './reservedTurnLeaseRetry.js';
|
||||
import type { HostBootMonitor } from './hostBootMonitor.js';
|
||||
|
||||
export interface TurnDaemonRuntimeOptions {
|
||||
profile: string;
|
||||
@@ -134,6 +136,7 @@ export interface TurnDaemonRuntimeOptions {
|
||||
leaseDurationMs?: number;
|
||||
leaseOwnerId?: string;
|
||||
enableLeaseHeartbeat?: boolean;
|
||||
hostBootMonitor?: HostBootMonitor;
|
||||
/**
|
||||
* Isolated, single-process fixture acceleration only. Reserved turns are
|
||||
* loaded once and no concurrent API writer may touch this database.
|
||||
@@ -985,6 +988,7 @@ const createTurnDaemonRuntimeWithLease = async (
|
||||
prepareClockRecovery = dbHooks.prepareRealtimeRecovery;
|
||||
prunePreviousAudit = dbHooks.prunePreviousAudit;
|
||||
close = async () => {
|
||||
await adminActionConsumer?.stop();
|
||||
if (auctionBidder) {
|
||||
await auctionBidder.close();
|
||||
}
|
||||
@@ -995,12 +999,15 @@ const createTurnDaemonRuntimeWithLease = async (
|
||||
await tournamentRewardFinalizer.close();
|
||||
}
|
||||
await dbHooks.close();
|
||||
if (reservedTurnStoreHandle) {
|
||||
await reservedTurnStoreHandle.close();
|
||||
try {
|
||||
await reservedTurnStoreHandle?.close();
|
||||
} finally {
|
||||
try {
|
||||
await gatewayGate?.close();
|
||||
} finally {
|
||||
await turnDaemonLease?.close();
|
||||
}
|
||||
}
|
||||
await gatewayGate?.close();
|
||||
await adminActionConsumer?.stop();
|
||||
await turnDaemonLease?.close();
|
||||
};
|
||||
} else if (reservedTurnStoreHandle) {
|
||||
hooks = {
|
||||
@@ -1009,9 +1016,12 @@ const createTurnDaemonRuntimeWithLease = async (
|
||||
},
|
||||
};
|
||||
close = async () => {
|
||||
await reservedTurnStoreHandle.close();
|
||||
await gatewayGate?.close();
|
||||
await adminActionConsumer?.stop();
|
||||
try {
|
||||
await reservedTurnStoreHandle.close();
|
||||
} finally {
|
||||
await gatewayGate?.close();
|
||||
}
|
||||
};
|
||||
} else if (gatewayGate) {
|
||||
hooks = {
|
||||
@@ -1070,12 +1080,17 @@ const createTurnDaemonRuntimeWithLease = async (
|
||||
const baseClose = close;
|
||||
close = async () => {
|
||||
stopClockProjectionWorker();
|
||||
await baseClose();
|
||||
await neutralAuctionRegistrar.close();
|
||||
if (redisConnector) {
|
||||
await redisConnector.disconnect();
|
||||
}
|
||||
await commandConnector?.disconnect();
|
||||
const baseResults = await Promise.allSettled([baseClose()]);
|
||||
// queue 임대 정리 실패도 다른 연결/worker의 종료를 막지 않게 한다.
|
||||
const results = await Promise.allSettled([
|
||||
neutralAuctionRegistrar.close(),
|
||||
redisConnector?.disconnect(),
|
||||
commandConnector?.disconnect(),
|
||||
]);
|
||||
const failures = [...baseResults, ...results].flatMap((result) =>
|
||||
result.status === 'rejected' ? [result.reason] : []
|
||||
);
|
||||
if (failures.length) throw new AggregateError(failures, 'Turn daemon connection cleanup failed.');
|
||||
};
|
||||
|
||||
const resolvedControlQueue = options.controlQueue ?? databaseCommandQueue ?? controlQueue;
|
||||
@@ -1109,7 +1124,8 @@ const createTurnDaemonRuntimeWithLease = async (
|
||||
processor,
|
||||
hooks,
|
||||
pauseGate: createRuntimePauseGate({
|
||||
assertLease: () => {
|
||||
assertLease: async () => {
|
||||
await turnDaemonLease?.checkHostBoot();
|
||||
if (turnDaemonLease?.isLost()) throw turnDaemonLease.getLossError();
|
||||
},
|
||||
shouldPause: async () => (await pauseGate?.()) ?? false,
|
||||
@@ -1127,6 +1143,9 @@ const createTurnDaemonRuntimeWithLease = async (
|
||||
// accumulated world before every due tick in that isolated mode;
|
||||
// production and gateway-managed runtimes keep rollback savepoints.
|
||||
stateManager: options.exclusiveFastForward ? undefined : stateManager,
|
||||
runErrorRetry: createReservedTurnLeaseRetry({
|
||||
onWait: (queueKey) => console.info(`[turn-daemon] waiting for reserved turn lease: ${queueKey}`),
|
||||
}),
|
||||
},
|
||||
{ profile: options.profile, defaultBudget }
|
||||
);
|
||||
@@ -1179,6 +1198,7 @@ export const createTurnDaemonRuntime = async (options: TurnDaemonRuntimeOptions)
|
||||
ownerId: options.leaseOwnerId,
|
||||
leaseDurationMs: options.leaseDurationMs,
|
||||
heartbeat: options.enableLeaseHeartbeat,
|
||||
hostBootMonitor: options.hostBootMonitor,
|
||||
})
|
||||
: null;
|
||||
if (turnDaemonLease && !(await turnDaemonLease.acquire())) {
|
||||
|
||||
@@ -0,0 +1,134 @@
|
||||
import { createHmac, randomBytes } from 'node:crypto';
|
||||
import { once } from 'node:events';
|
||||
import { createServer } from 'node:http';
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest';
|
||||
import { createHostBootMonitor, loadHostBootMonitor } from '../src/turn/hostBootMonitor.js';
|
||||
|
||||
const cleanups: Array<() => Promise<void>> = [];
|
||||
afterEach(async () => {
|
||||
await Promise.all(cleanups.splice(0).map((cleanup) => cleanup()));
|
||||
});
|
||||
|
||||
const fixture = async () => {
|
||||
const key = randomBytes(32);
|
||||
let bootId = 'a'.repeat(64);
|
||||
let replay: string | null = null;
|
||||
let mode: 'valid' | 'replay' | 'forged' | 'oversized' | 'offline' | 'hang' = 'valid';
|
||||
const requests: string[] = [];
|
||||
const server = createServer((request, response) => {
|
||||
const nonce = String(request.headers['x-sammo-challenge']);
|
||||
requests.push(nonce);
|
||||
expect(request.headers['x-sammo-signature']).toBe(
|
||||
createHmac('sha256', key).update(`sammo-host-boot-request-v1\n${nonce}`).digest('hex')
|
||||
);
|
||||
if (mode === 'hang') return;
|
||||
if (mode === 'offline') {
|
||||
response.writeHead(503).end();
|
||||
return;
|
||||
}
|
||||
if (mode === 'oversized') {
|
||||
response.end('x'.repeat(4096));
|
||||
return;
|
||||
}
|
||||
if (mode === 'replay') {
|
||||
response.end(replay);
|
||||
return;
|
||||
}
|
||||
const body = JSON.stringify({
|
||||
version: 1,
|
||||
challenge: nonce,
|
||||
bootId,
|
||||
signature:
|
||||
mode === 'forged'
|
||||
? '0'.repeat(64)
|
||||
: createHmac('sha256', key)
|
||||
.update(`sammo-host-boot-response-v1\n${nonce}\n${bootId}`)
|
||||
.digest('hex'),
|
||||
});
|
||||
replay ??= body;
|
||||
response.end(body);
|
||||
});
|
||||
server.listen(0, '127.0.0.1');
|
||||
await once(server, 'listening');
|
||||
cleanups.push(
|
||||
() =>
|
||||
new Promise<void>((resolve) => {
|
||||
server.close(() => resolve());
|
||||
server.closeAllConnections();
|
||||
})
|
||||
);
|
||||
const address = server.address();
|
||||
if (!address || typeof address === 'string') throw new Error('Missing fixture listener');
|
||||
return {
|
||||
key,
|
||||
requests,
|
||||
url: `http://127.0.0.1:${address.port}/v1/boot-id`,
|
||||
setBoot: (value: string) => {
|
||||
bootId = value;
|
||||
},
|
||||
setMode: (value: typeof mode) => {
|
||||
mode = value;
|
||||
},
|
||||
};
|
||||
};
|
||||
|
||||
describe('Windows boot observation', () => {
|
||||
it('uses a stable baseline, survives outages, and latches a signed boot change', async () => {
|
||||
const f = await fixture();
|
||||
const monitor = createHostBootMonitor({ ...f, pollMs: 0 });
|
||||
expect(await monitor.hasChanged()).toBe(false);
|
||||
f.setMode('offline');
|
||||
expect(await monitor.hasChanged()).toBe(false);
|
||||
f.setBoot('b'.repeat(64));
|
||||
f.setMode('valid');
|
||||
expect(await monitor.hasChanged()).toBe(true);
|
||||
f.setMode('offline');
|
||||
expect(await monitor.hasChanged()).toBe(true);
|
||||
});
|
||||
|
||||
it.each(['replay', 'forged', 'oversized', 'hang'] as const)(
|
||||
'rejects %s without changing the baseline',
|
||||
async (mode) => {
|
||||
const f = await fixture();
|
||||
const onUnavailable = vi.fn();
|
||||
const monitor = createHostBootMonitor({ ...f, pollMs: 0, timeoutMs: 100, onUnavailable });
|
||||
expect(await monitor.hasChanged()).toBe(false);
|
||||
f.setBoot('b'.repeat(64));
|
||||
f.setMode(mode);
|
||||
expect(await monitor.hasChanged()).toBe(false);
|
||||
expect(onUnavailable).toHaveBeenCalledOnce();
|
||||
f.setMode('valid');
|
||||
expect(await monitor.hasChanged()).toBe(true);
|
||||
}
|
||||
);
|
||||
|
||||
it('samples immediately after a wall-clock jump despite a frozen monotonic cache and coalesces requests', async () => {
|
||||
const f = await fixture();
|
||||
let wall = 10_000;
|
||||
const monitor = createHostBootMonitor({ ...f, now: () => 0, wallNow: () => wall });
|
||||
expect(await monitor.hasChanged()).toBe(false);
|
||||
expect(await monitor.hasChanged()).toBe(false);
|
||||
expect(f.requests).toHaveLength(1);
|
||||
wall += 300_000;
|
||||
f.setBoot('b'.repeat(64));
|
||||
expect(await Promise.all([monitor.hasChanged(), monitor.hasChanged()])).toEqual([true, true]);
|
||||
expect(f.requests).toHaveLength(2);
|
||||
expect(new Set(f.requests).size).toBe(2);
|
||||
});
|
||||
|
||||
it('does not block game startup when optional configuration or its key file is absent', async () => {
|
||||
const warning = vi.spyOn(console, 'warn').mockImplementation(() => {});
|
||||
expect(await loadHostBootMonitor({})).toBeUndefined();
|
||||
expect(await loadHostBootMonitor({ TURN_HOST_BOOT_URL: 'http://host/v1/boot-id' })).toBeUndefined();
|
||||
expect(
|
||||
await loadHostBootMonitor({
|
||||
TURN_HOST_BOOT_URL: 'http://host/v1/boot-id',
|
||||
TURN_HOST_BOOT_KEY_FILE: '/nonexistent-host-boot-key',
|
||||
})
|
||||
).toBeUndefined();
|
||||
expect(() =>
|
||||
createHostBootMonitor({ url: 'http://user:password@host/v1/boot-id', key: randomBytes(32) })
|
||||
).toThrow('without credentials');
|
||||
warning.mockRestore();
|
||||
});
|
||||
});
|
||||
@@ -4,7 +4,9 @@ import { createGamePostgresConnector } from '@sammo-ts/infra';
|
||||
import {
|
||||
InMemoryReservedTurnStore,
|
||||
ReservedTurnLeaseConflictError,
|
||||
createReservedTurnStore,
|
||||
} from '../src/turn/reservedTurnStore.js';
|
||||
import { createReservedTurnLeaseRetry } from '../src/turn/reservedTurnLeaseRetry.js';
|
||||
|
||||
const databaseUrl = process.env.RESERVED_TURN_DATABASE_URL;
|
||||
const describeIntegration = databaseUrl ? describe : describe.skip;
|
||||
@@ -132,4 +134,98 @@ describeIntegration('reserved turn daemon/API lease integration', () => {
|
||||
});
|
||||
expect(after.map((turn) => turn.actionCode)).toEqual(before.map((turn) => turn.actionCode));
|
||||
});
|
||||
|
||||
it('cleans persisted ownership after rollback loses the journal, preserves other writers, and reloads once', async () => {
|
||||
if (!connector || !databaseUrl) throw new Error('integration connector is unavailable');
|
||||
await connector.prisma.generalTurnRevision.update({
|
||||
where: { generalId: GENERAL_ID },
|
||||
data: { leaseOwner: null, leaseExpiresAt: null },
|
||||
});
|
||||
const before = await connector.prisma.generalTurn.findMany({
|
||||
where: { generalId: GENERAL_ID },
|
||||
orderBy: { turnIdx: 'asc' },
|
||||
});
|
||||
const revision = await connector.prisma.generalTurnRevision.findUniqueOrThrow({
|
||||
where: { generalId: GENERAL_ID },
|
||||
});
|
||||
const old = await createReservedTurnStore({ databaseUrl, leaseOwner: 'interrupted-daemon' });
|
||||
const savepoint = old.store.captureTransactionState();
|
||||
await old.store.prepareTurnsForExecution(GENERAL_ID);
|
||||
old.store.shiftGeneralTurns(GENERAL_ID, -1);
|
||||
old.store.restoreState(savepoint);
|
||||
expect(old.store.getQueueCounts().leasedGeneralQueues).toBe(0);
|
||||
const next = await createReservedTurnStore({ databaseUrl, leaseOwner: 'new-daemon' });
|
||||
try {
|
||||
await expect(next.store.prepareTurnsForExecution(GENERAL_ID)).rejects.toBeInstanceOf(
|
||||
ReservedTurnLeaseConflictError
|
||||
);
|
||||
await old.close();
|
||||
const released = await connector.prisma.generalTurnRevision.findUniqueOrThrow({
|
||||
where: { generalId: GENERAL_ID },
|
||||
});
|
||||
expect(released).toMatchObject({ leaseOwner: null, leaseExpiresAt: null, revision: revision.revision });
|
||||
await next.store.prepareTurnsForExecution(GENERAL_ID);
|
||||
// A delayed close of an old owner cannot clear the new owner's lease.
|
||||
const delayedCleaner = new InMemoryReservedTurnStore(connector.prisma, {
|
||||
maxGeneralTurns: 2,
|
||||
maxNationTurns: 1,
|
||||
leaseOwner: 'interrupted-daemon',
|
||||
});
|
||||
await delayedCleaner.releaseOwnedLeases();
|
||||
expect(
|
||||
await connector.prisma.generalTurnRevision.findUniqueOrThrow({ where: { generalId: GENERAL_ID } })
|
||||
).toMatchObject({ leaseOwner: 'new-daemon', revision: revision.revision });
|
||||
await next.store.releaseOwnedLeases();
|
||||
const actual = await connector.prisma.generalTurn.findMany({
|
||||
where: { generalId: GENERAL_ID },
|
||||
orderBy: { turnIdx: 'asc' },
|
||||
});
|
||||
expect(actual.map((row) => row.actionCode)).toEqual(before.map((row) => row.actionCode));
|
||||
} finally {
|
||||
await next.close();
|
||||
}
|
||||
});
|
||||
|
||||
it('waits out an orphaned lease without deleting it and reloads an intervening API edit', async () => {
|
||||
if (!connector) throw new Error('integration connector is unavailable');
|
||||
await connector.prisma.generalTurnRevision.update({
|
||||
where: { generalId: GENERAL_ID },
|
||||
data: { leaseOwner: 'killed-daemon', leaseExpiresAt: new Date(Date.now() + 300_000) },
|
||||
});
|
||||
const store = new InMemoryReservedTurnStore(connector.prisma, {
|
||||
maxGeneralTurns: 2,
|
||||
maxNationTurns: 1,
|
||||
leaseOwner: 'recovery-daemon',
|
||||
});
|
||||
await store.loadAll();
|
||||
const retry = createReservedTurnLeaseRetry({
|
||||
sleep: async () => {
|
||||
// Simulate the recorded expiry, then an authorized API edit before the next claim.
|
||||
expect(
|
||||
await connector.prisma.generalTurnRevision.findUniqueOrThrow({ where: { generalId: GENERAL_ID } })
|
||||
).toMatchObject({ leaseOwner: 'killed-daemon' });
|
||||
await connector.prisma.generalTurnRevision.update({
|
||||
where: { generalId: GENERAL_ID },
|
||||
data: { leaseExpiresAt: new Date(0) },
|
||||
});
|
||||
await connector.prisma.generalTurn.updateMany({
|
||||
where: { generalId: GENERAL_ID, turnIdx: 0 },
|
||||
data: { actionCode: 'che_수련' },
|
||||
});
|
||||
},
|
||||
});
|
||||
try {
|
||||
let failure: unknown;
|
||||
try {
|
||||
await store.prepareTurnsForExecution(GENERAL_ID);
|
||||
} catch (error) {
|
||||
failure = error;
|
||||
}
|
||||
expect(await retry.wait(failure)).toBe(true);
|
||||
await store.prepareTurnsForExecution(GENERAL_ID);
|
||||
expect(store.getGeneralTurns(GENERAL_ID)[0]?.action).toBe('che_수련');
|
||||
} finally {
|
||||
await store.releaseOwnedLeases();
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -169,6 +169,18 @@ const buildHarness = (initialRevision: RevisionRow | null = null) => {
|
||||
};
|
||||
|
||||
describe('reserved turn daemon lease', () => {
|
||||
it('cleans both persisted lease kinds even after their journal is rolled back', async () => {
|
||||
const harness = buildHarness();
|
||||
const snapshot = harness.store.captureTransactionState();
|
||||
await harness.store.prepareTurnsForExecution(7, { nationId: 3, officerLevel: 12 });
|
||||
harness.store.restoreState(snapshot);
|
||||
expect(harness.store.getQueueCounts()).toMatchObject({ leasedGeneralQueues: 0, leasedNationQueues: 0 });
|
||||
await harness.store.releaseOwnedLeases();
|
||||
expect(harness.getRevision()).toMatchObject({ leaseOwner: null, leaseExpiresAt: null, revision: 0 });
|
||||
expect(harness.getNationRevision()).toMatchObject({ leaseOwner: null, leaseExpiresAt: null, revision: 0 });
|
||||
expect(harness.generalDeleteMany).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('prunes deleted general and nation queues together with their journal state', () => {
|
||||
const harness = buildHarness();
|
||||
harness.store.ensureGeneralTurns(7);
|
||||
|
||||
@@ -0,0 +1,38 @@
|
||||
import { describe, expect, it, vi } from 'vitest';
|
||||
import { createReservedTurnLeaseRetry } from '../src/turn/reservedTurnLeaseRetry.js';
|
||||
import { ReservedTurnLeaseBusyError, ReservedTurnLeaseConflictError } from '../src/turn/reservedTurnStore.js';
|
||||
|
||||
describe('reserved turn acquisition retry', () => {
|
||||
it('bounds the entire contention episode across different queues and resets after progress', async () => {
|
||||
let now = 0;
|
||||
const onWait = vi.fn();
|
||||
const retry = createReservedTurnLeaseRetry({
|
||||
now: () => now,
|
||||
sleep: async (ms) => {
|
||||
now += ms;
|
||||
},
|
||||
maxWaitMs: 2_500,
|
||||
onWait,
|
||||
});
|
||||
expect(await retry.wait(new ReservedTurnLeaseBusyError('general:1'))).toBe(true);
|
||||
expect(await retry.wait(new ReservedTurnLeaseBusyError('nation:2:12'))).toBe(true);
|
||||
expect(await retry.wait(new ReservedTurnLeaseBusyError('general:3'))).toBe(true);
|
||||
expect(now).toBe(2_500);
|
||||
expect(await retry.wait(new ReservedTurnLeaseBusyError('general:1'))).toBe(false);
|
||||
expect(onWait).toHaveBeenCalledTimes(1);
|
||||
retry.reset();
|
||||
expect(await retry.wait(new ReservedTurnLeaseBusyError('general:1'))).toBe(true);
|
||||
});
|
||||
|
||||
it('never retries flush ownership loss, calculation errors, or failed savepoint restoration', async () => {
|
||||
const sleep = vi.fn();
|
||||
const retry = createReservedTurnLeaseRetry({ sleep });
|
||||
for (const error of [
|
||||
new ReservedTurnLeaseConflictError('general:1'),
|
||||
new Error('calculation failed'),
|
||||
new AggregateError([new ReservedTurnLeaseBusyError('general:1')], 'restore failed'),
|
||||
])
|
||||
expect(await retry.wait(error)).toBe(false);
|
||||
expect(sleep).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@@ -3,12 +3,36 @@ import type { GameClockPhase } from '@sammo-ts/common';
|
||||
import { createRuntimePauseGate } from '../src/turn/runtimePauseGate.js';
|
||||
|
||||
describe('runtime pause clock boundary', () => {
|
||||
it('awaits a host authority check before reading profile state or changing the clock', async () => {
|
||||
const calls: string[] = [];
|
||||
const gate = createRuntimePauseGate({
|
||||
assertLease: async () => {
|
||||
await Promise.resolve();
|
||||
calls.push('host');
|
||||
throw new Error('host changed');
|
||||
},
|
||||
shouldPause: async () => {
|
||||
calls.push('profile');
|
||||
return false;
|
||||
},
|
||||
isExplicitlyPaused: () => false,
|
||||
getPhase: () => 'RUNNING',
|
||||
prepareRecovery: async () => {
|
||||
calls.push('recovery');
|
||||
},
|
||||
synchronize: async () => {},
|
||||
});
|
||||
await expect(gate()).rejects.toThrow('host changed');
|
||||
expect(calls).toEqual(['host']);
|
||||
});
|
||||
it('freezes a live error pause before commands, resumes without restart, and does not freeze twice', async () => {
|
||||
let phase: GameClockPhase = 'RUNNING';
|
||||
let paused = true;
|
||||
const calls: string[] = [];
|
||||
const gate = createRuntimePauseGate({
|
||||
assertLease: () => calls.push('lease'),
|
||||
assertLease: () => {
|
||||
calls.push('lease');
|
||||
},
|
||||
shouldPause: async () => paused,
|
||||
isExplicitlyPaused: () => paused,
|
||||
getPhase: () => phase,
|
||||
|
||||
@@ -207,6 +207,39 @@ integration('database turn daemon lease and fencing', () => {
|
||||
expect(row.ownerId).toBe(firstToken ? 'owner-a' : 'owner-b');
|
||||
});
|
||||
|
||||
it('fences a host boot change inside a transaction before mutation, without expiring the DB lease', async () => {
|
||||
const profile = `${profilePrefix}host-boot`;
|
||||
let changed = false;
|
||||
const lease = await DatabaseTurnDaemonLease.connect(databaseUrl!, {
|
||||
profile,
|
||||
ownerId: 'saved-vm-owner',
|
||||
heartbeat: false,
|
||||
leaseDurationMs: 60_000,
|
||||
hostBootMonitor: { hasChanged: async () => changed },
|
||||
});
|
||||
leases.push(lease);
|
||||
await lease.acquire();
|
||||
await lease.markClockReady();
|
||||
changed = true;
|
||||
let committed = false;
|
||||
await expect(
|
||||
db.$transaction(async (transaction) => {
|
||||
await lease.assertActive(transaction);
|
||||
committed = true;
|
||||
})
|
||||
).rejects.toBeInstanceOf(TurnDaemonLeaseLostError);
|
||||
expect(committed).toBe(false);
|
||||
expect(lease.isLost()).toBe(true);
|
||||
expect(await lease.renew()).toBe(false);
|
||||
// The monitor does not force-clear ownership or bypass the successor's fencing.
|
||||
expect((await db.turnDaemonLease.findUniqueOrThrow({ where: { profile } })).ownerId).toBe('saved-vm-owner');
|
||||
const next = await createLease(profile, 'fresh-vm-owner');
|
||||
expect(await next.acquire()).toBeNull();
|
||||
await lease.release();
|
||||
expect((await db.turnDaemonLease.findUniqueOrThrow({ where: { profile } })).clockReady).toBe(false);
|
||||
expect((await next.acquire())?.fencingEpoch).toBe(2n);
|
||||
});
|
||||
|
||||
it('persists and evaluates lease wall time as UTC under a non-UTC database session', async () => {
|
||||
const profile = `${profilePrefix}utc-wall`;
|
||||
const zonedUrl = new URL(databaseUrl!);
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
import { describe, expect, it, vi } from 'vitest';
|
||||
import { ManualClock } from '@sammo-ts/common';
|
||||
import { TurnDaemonLeaseLostError } from '../src/lifecycle/databaseTurnDaemonLease.js';
|
||||
import { ReservedTurnLeaseBusyError } from '../src/turn/reservedTurnStore.js';
|
||||
import { createReservedTurnLeaseRetry } from '../src/turn/reservedTurnLeaseRetry.js';
|
||||
|
||||
import {
|
||||
InMemoryControlQueue,
|
||||
@@ -15,6 +17,68 @@ import {
|
||||
const addMinutes = (time: Date, minutes: number): Date => new Date(time.getTime() + minutes * 60_000);
|
||||
|
||||
describe('TurnDaemonLifecycle', () => {
|
||||
it('rolls back a busy queue, checks the operator gate, and retries an explicit run exactly once', async () => {
|
||||
const now = new Date('2026-11-12T00:00:00Z');
|
||||
const queue = new InMemoryControlQueue();
|
||||
queue.enqueue({ type: 'run', reason: 'manual', targetTime: now.toISOString() });
|
||||
let resource = 10;
|
||||
let runs = 0;
|
||||
const stateManager = new EngineStateManager();
|
||||
stateManager.register('resource', {
|
||||
capture: () => resource,
|
||||
restore: (value) => {
|
||||
resource = value;
|
||||
},
|
||||
});
|
||||
const onRunError = vi.fn();
|
||||
const flushChanges = vi.fn();
|
||||
const gate = vi.fn(async () => false);
|
||||
const retry = createReservedTurnLeaseRetry({
|
||||
sleep: async () => {
|
||||
expect(resource).toBe(10);
|
||||
},
|
||||
});
|
||||
const lifecycle = new TurnDaemonLifecycle(
|
||||
{
|
||||
clock: new ManualClock(now.getTime()),
|
||||
controlQueue: queue,
|
||||
stateManager,
|
||||
runErrorRetry: retry,
|
||||
getNextTickTime: (value) => addMinutes(value, 5),
|
||||
pauseGate: gate,
|
||||
stateStore: {
|
||||
loadLastTurnTime: async () => now,
|
||||
loadNextGeneralTurnTime: async () => null,
|
||||
saveLastTurnTime: async () => {},
|
||||
loadCheckpoint: async () => undefined,
|
||||
saveCheckpoint: async () => {},
|
||||
},
|
||||
processor: {
|
||||
run: async () => {
|
||||
resource -= 2;
|
||||
if (++runs === 1) throw new ReservedTurnLeaseBusyError('general:7');
|
||||
queue.enqueue({ type: 'shutdown' });
|
||||
return {
|
||||
lastTurnTime: now.toISOString(),
|
||||
processedGenerals: 1,
|
||||
processedTurns: 0,
|
||||
durationMs: 0,
|
||||
partial: false,
|
||||
};
|
||||
},
|
||||
},
|
||||
hooks: { onRunError, flushChanges },
|
||||
},
|
||||
{ profile: 'che', defaultBudget: { budgetMs: 100, maxGenerals: 1, catchUpCap: 1 } }
|
||||
);
|
||||
await lifecycle.start();
|
||||
expect(runs).toBe(2);
|
||||
expect(resource).toBe(8);
|
||||
expect(gate.mock.calls.length).toBeGreaterThanOrEqual(2);
|
||||
expect(flushChanges).toHaveBeenCalledOnce();
|
||||
expect(onRunError).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it.each([false, true])('reports a fatal lease gate and exits even if reporting fails (%s)', async (reportFails) => {
|
||||
const now = new Date('2026-09-09T17:30:00Z');
|
||||
const error = new TurnDaemonLeaseLostError('che:default');
|
||||
|
||||
Reference in new issue
Block a user