fix(clock): preserve UTC leases and unification recovery
This commit is contained in:
@@ -86,9 +86,9 @@ export class DatabaseTurnDaemonLease {
|
||||
VALUES (
|
||||
${this.profile},
|
||||
${this.ownerId},
|
||||
CURRENT_TIMESTAMP + (${this.leaseDurationMs} * INTERVAL '1 millisecond'),
|
||||
(CURRENT_TIMESTAMP AT TIME ZONE 'UTC') + (${this.leaseDurationMs} * INTERVAL '1 millisecond'),
|
||||
1,
|
||||
CURRENT_TIMESTAMP
|
||||
CURRENT_TIMESTAMP AT TIME ZONE 'UTC'
|
||||
)
|
||||
ON CONFLICT ("profile") DO UPDATE
|
||||
SET
|
||||
@@ -99,10 +99,10 @@ export class DatabaseTurnDaemonLease {
|
||||
THEN "turn_daemon_lease"."fencing_epoch"
|
||||
ELSE "turn_daemon_lease"."fencing_epoch" + 1
|
||||
END,
|
||||
"heartbeat_at" = CURRENT_TIMESTAMP
|
||||
"heartbeat_at" = CURRENT_TIMESTAMP AT TIME ZONE 'UTC'
|
||||
WHERE
|
||||
"turn_daemon_lease"."owner_id" = EXCLUDED."owner_id"
|
||||
OR "turn_daemon_lease"."lease_until" <= CURRENT_TIMESTAMP
|
||||
OR "turn_daemon_lease"."lease_until" <= CURRENT_TIMESTAMP AT TIME ZONE 'UTC'
|
||||
RETURNING "profile", "owner_id", "fencing_epoch"
|
||||
`);
|
||||
const row = rows[0];
|
||||
@@ -141,13 +141,13 @@ export class DatabaseTurnDaemonLease {
|
||||
const rows = await this.db.$queryRaw<LeaseRow[]>(GamePrisma.sql`
|
||||
UPDATE "turn_daemon_lease"
|
||||
SET
|
||||
"lease_until" = CURRENT_TIMESTAMP + (${this.leaseDurationMs} * INTERVAL '1 millisecond'),
|
||||
"heartbeat_at" = CURRENT_TIMESTAMP
|
||||
"lease_until" = (CURRENT_TIMESTAMP AT TIME ZONE 'UTC') + (${this.leaseDurationMs} * INTERVAL '1 millisecond'),
|
||||
"heartbeat_at" = CURRENT_TIMESTAMP AT TIME ZONE 'UTC'
|
||||
WHERE
|
||||
"profile" = ${token.profile}
|
||||
AND "owner_id" = ${token.ownerId}
|
||||
AND "fencing_epoch" = ${token.fencingEpoch}
|
||||
AND "lease_until" > CURRENT_TIMESTAMP
|
||||
AND "lease_until" > CURRENT_TIMESTAMP AT TIME ZONE 'UTC'
|
||||
RETURNING "profile", "owner_id", "fencing_epoch"
|
||||
`);
|
||||
if (rows.length === 0) {
|
||||
@@ -177,7 +177,7 @@ export class DatabaseTurnDaemonLease {
|
||||
"profile" = ${token.profile}
|
||||
AND "owner_id" = ${token.ownerId}
|
||||
AND "fencing_epoch" = ${token.fencingEpoch}
|
||||
AND "lease_until" > CURRENT_TIMESTAMP
|
||||
AND "lease_until" > CURRENT_TIMESTAMP AT TIME ZONE 'UTC'
|
||||
FOR UPDATE
|
||||
`);
|
||||
if (rows.length === 0) {
|
||||
@@ -196,7 +196,8 @@ export class DatabaseTurnDaemonLease {
|
||||
}
|
||||
await this.db.$executeRaw(GamePrisma.sql`
|
||||
UPDATE "turn_daemon_lease"
|
||||
SET "lease_until" = CURRENT_TIMESTAMP, "heartbeat_at" = CURRENT_TIMESTAMP
|
||||
SET "lease_until" = CURRENT_TIMESTAMP AT TIME ZONE 'UTC',
|
||||
"heartbeat_at" = CURRENT_TIMESTAMP AT TIME ZONE 'UTC'
|
||||
WHERE
|
||||
"profile" = ${token.profile}
|
||||
AND "owner_id" = ${token.ownerId}
|
||||
|
||||
@@ -98,11 +98,12 @@ const invalidateMessageIds = async (
|
||||
db: GamePrisma.TransactionClient,
|
||||
world: InMemoryTurnWorld,
|
||||
ids: number[],
|
||||
now: Date
|
||||
now: Date,
|
||||
authoritativeGameTick?: bigint
|
||||
): Promise<void> => {
|
||||
const uniqueIds = [...new Set(ids.filter((id) => Number.isInteger(id) && id > 0))];
|
||||
if (uniqueIds.length === 0) return;
|
||||
const resolvedGameTick = BigInt(world.dateToGameTick(now));
|
||||
const resolvedGameTick = authoritativeGameTick ?? BigInt(world.dateToGameTick(now));
|
||||
await db.messageAction.updateMany({
|
||||
where: { messageId: { in: uniqueIds }, status: 'PENDING' },
|
||||
data: { status: 'RESOLVED', resolvedGameTick },
|
||||
@@ -348,7 +349,27 @@ const respondToRaiseInvader = async (options: {
|
||||
},
|
||||
event
|
||||
);
|
||||
await invalidateMessageIds(db, world, [row.id], world.gameTickToDate(alignment.alignedTick));
|
||||
const resolvedGameTick = BigInt(alignment.alignedTick);
|
||||
if (resolvedGameTick < row.createdGameTick) {
|
||||
throw new Error(
|
||||
`RaiseInvader resolved tick ${resolvedGameTick} precedes prompt tick ${row.createdGameTick}; clock authority is inconsistent.`
|
||||
);
|
||||
}
|
||||
const promptRows = await db.$queryRaw<Array<{ id: number }>>(GamePrisma.sql`
|
||||
SELECT message_id AS id
|
||||
FROM message_action
|
||||
WHERE action_type = 'raiseInvader'
|
||||
AND status = 'PENDING'
|
||||
AND created_game_tick = ${row.createdGameTick}
|
||||
FOR UPDATE
|
||||
`);
|
||||
await invalidateMessageIds(
|
||||
db,
|
||||
world,
|
||||
promptRows.map(({ id }) => id),
|
||||
world.gameTickToDate(alignment.alignedTick),
|
||||
resolvedGameTick
|
||||
);
|
||||
return { ok: true, action: 'raiseInvader', reason: 'success' };
|
||||
};
|
||||
|
||||
|
||||
@@ -207,6 +207,57 @@ integration('database turn daemon lease and fencing', () => {
|
||||
expect(row.ownerId).toBe(firstToken ? 'owner-a' : 'owner-b');
|
||||
});
|
||||
|
||||
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!);
|
||||
const schema = zonedUrl.searchParams.get('schema') ?? 'public';
|
||||
zonedUrl.searchParams.delete('schema');
|
||||
zonedUrl.searchParams.set('options', `-c search_path=${schema} -c TimeZone=Asia/Seoul`);
|
||||
const zonedConnector = createGamePostgresConnector({ url: zonedUrl.toString() });
|
||||
await zonedConnector.connect();
|
||||
const lease = await DatabaseTurnDaemonLease.connect(zonedUrl.toString(), {
|
||||
profile,
|
||||
ownerId: 'utc-owner',
|
||||
leaseDurationMs: 60_000,
|
||||
heartbeat: false,
|
||||
});
|
||||
leases.push(lease);
|
||||
try {
|
||||
await expect(lease.acquire()).resolves.toMatchObject({ profile, ownerId: 'utc-owner' });
|
||||
const [leaseRows, wallRows] = await Promise.all([
|
||||
zonedConnector.prisma.$queryRaw<Array<{ leaseUntil: Date }>>`
|
||||
SELECT lease_until AS "leaseUntil"
|
||||
FROM turn_daemon_lease
|
||||
WHERE profile = ${profile}
|
||||
`,
|
||||
zonedConnector.prisma.$queryRaw<Array<{ zone: string; wallNow: Date }>>`
|
||||
SELECT current_setting('TimeZone') AS zone,
|
||||
(CURRENT_TIMESTAMP AT TIME ZONE 'UTC')::timestamp(3) AS "wallNow"
|
||||
`,
|
||||
]);
|
||||
expect(wallRows[0]?.zone).toBe('Asia/Seoul');
|
||||
expect(leaseRows[0]!.leaseUntil.getTime() - wallRows[0]!.wallNow.getTime()).toBeGreaterThan(50_000);
|
||||
expect(leaseRows[0]!.leaseUntil.getTime() - wallRows[0]!.wallNow.getTime()).toBeLessThanOrEqual(60_000);
|
||||
|
||||
await lease.release();
|
||||
const [releasedRows, releasedWall] = await Promise.all([
|
||||
zonedConnector.prisma.$queryRaw<Array<{ leaseUntil: Date }>>`
|
||||
SELECT lease_until AS "leaseUntil"
|
||||
FROM turn_daemon_lease
|
||||
WHERE profile = ${profile}
|
||||
`,
|
||||
zonedConnector.prisma.$queryRaw<Array<{ wallNow: Date }>>`
|
||||
SELECT (CURRENT_TIMESTAMP AT TIME ZONE 'UTC')::timestamp(3) AS "wallNow"
|
||||
`,
|
||||
]);
|
||||
expect(Math.abs(releasedRows[0]!.leaseUntil.getTime() - releasedWall[0]!.wallNow.getTime())).toBeLessThan(
|
||||
1_000
|
||||
);
|
||||
} finally {
|
||||
await zonedConnector.disconnect();
|
||||
}
|
||||
});
|
||||
|
||||
it('increments the epoch on expiry takeover and fences the stale owner', async () => {
|
||||
const profile = `${profilePrefix}takeover`;
|
||||
const first = await createLease(profile, 'owner-a');
|
||||
|
||||
@@ -312,7 +312,17 @@ integration('unification finalization transaction', () => {
|
||||
const bidWorld = new InMemoryTurnWorld(beforeBid.state, beforeBid.snapshot, {
|
||||
schedule: { entries: [{ startMinute: 0, tickMinutes: 10 }] },
|
||||
});
|
||||
const bidProcessingTick = bidWorld.getGameClockState().tick;
|
||||
const futureCloseTick = BigInt(bidProcessingTick) + 86_400_000n;
|
||||
await db.auction.updateMany({
|
||||
where: { id: { in: [uniqueAuction.id, resourceAuction.id] } },
|
||||
data: { closeTick: futureCloseTick },
|
||||
});
|
||||
const bidder = await createAuctionBidder({ databaseUrl: databaseUrl!, world: bidWorld });
|
||||
const bidClockContext = {
|
||||
processingGameTick: bidProcessingTick,
|
||||
requestedAtWall: new Date('2026-09-03T00:00:00.000Z'),
|
||||
};
|
||||
try {
|
||||
await expect(
|
||||
bidder.bid({
|
||||
@@ -322,7 +332,8 @@ integration('unification finalization transaction', () => {
|
||||
generalId: fixtureId,
|
||||
amount: 30,
|
||||
tryExtendCloseDate: false,
|
||||
})
|
||||
...bidClockContext,
|
||||
} as Parameters<typeof bidder.bid>[0])
|
||||
).resolves.toMatchObject({ ok: true, auctionId: uniqueAuction.id });
|
||||
await expect(
|
||||
bidder.bid({
|
||||
@@ -332,7 +343,8 @@ integration('unification finalization transaction', () => {
|
||||
generalId: fixtureId,
|
||||
amount: 50,
|
||||
tryExtendCloseDate: false,
|
||||
})
|
||||
...bidClockContext,
|
||||
} as Parameters<typeof bidder.bid>[0])
|
||||
).resolves.toMatchObject({ ok: true, auctionId: uniqueAuction.id });
|
||||
} finally {
|
||||
await bidder.close();
|
||||
@@ -699,6 +711,24 @@ integration('unification finalization transaction', () => {
|
||||
});
|
||||
const commandResult = await stateManager.transaction(executeCommand);
|
||||
expect(commandResult).toMatchObject({ type: 'messageRespond', ok: true, action: 'raiseInvader' });
|
||||
const selectedPromptAction = await db.messageAction.findUniqueOrThrow({
|
||||
where: { messageId: invaderPrompt!.id },
|
||||
});
|
||||
const siblingPromptActions = await db.messageAction.findMany({
|
||||
where: {
|
||||
actionType: 'raiseInvader',
|
||||
createdGameTick: selectedPromptAction.createdGameTick,
|
||||
},
|
||||
});
|
||||
expect(siblingPromptActions.length).toBeGreaterThan(1);
|
||||
expect(
|
||||
siblingPromptActions.every(
|
||||
(action) =>
|
||||
action.status === 'RESOLVED' &&
|
||||
action.resolvedGameTick !== null &&
|
||||
action.resolvedGameTick >= action.createdGameTick
|
||||
)
|
||||
).toBe(true);
|
||||
const reconciledWorld = await db.worldState.findUniqueOrThrow({ where: { id: worldRow.id } });
|
||||
const appliedSuspension = await db.clockSuspension.findUniqueOrThrow({ where: { id: suspension.id } });
|
||||
expect(reconciledWorld).toMatchObject({
|
||||
|
||||
@@ -222,6 +222,14 @@ export const planProfileReconcile = (
|
||||
};
|
||||
};
|
||||
|
||||
/**
|
||||
* A stopped process may be restarted while the world remains in unification
|
||||
* wait. Only the daemon-authorized raise-invader response may reconcile that
|
||||
* suspension, so Gateway RESUME must start the runtime without consuming it.
|
||||
*/
|
||||
export const shouldStartRuntimeInUnificationWait = (clockPhase: string, suspensionSource: string): boolean =>
|
||||
clockPhase === 'SUSPENDED' && suspensionSource === 'UNIFICATION_WAIT';
|
||||
|
||||
export const resolveResetLifecycleStatus = (
|
||||
now: Date,
|
||||
preopenAt: Date | null,
|
||||
@@ -1311,6 +1319,9 @@ export class GatewayOrchestrator implements GatewayOrchestratorHandle {
|
||||
orderBy: { createdAt: 'desc' },
|
||||
});
|
||||
if (!suspension) throw new Error('Profile resume requires a durable suspended clock ledger.');
|
||||
if (shouldStartRuntimeInUnificationWait(world.clockPhase, suspension.source)) {
|
||||
return { phase: 'SUSPENDED', revision: clockRevisionAsNumber(world.clockRevision) };
|
||||
}
|
||||
if (world.clockPhase === 'SUSPENDED') {
|
||||
await reconcileClockSuspension({ db: postgres.prisma, suspensionId: suspension.id, authority });
|
||||
} else if (world.clockPhase !== 'RECONCILING') {
|
||||
|
||||
@@ -13,6 +13,7 @@ import {
|
||||
planProfileReconcile,
|
||||
resolveProfileArchiveServerName,
|
||||
resolveResetLifecycleStatus,
|
||||
shouldStartRuntimeInUnificationWait,
|
||||
} from '../src/orchestrator/gatewayOrchestrator.js';
|
||||
import { GATEWAY_PROFILE_ORDER } from '../src/profileOrder.js';
|
||||
import { sanitizeManagedProcessEnv } from '../src/orchestrator/processManager.js';
|
||||
@@ -114,6 +115,20 @@ describe('planProfileReconcile', () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe('shouldStartRuntimeInUnificationWait', () => {
|
||||
it('starts a stopped runtime without consuming its daemon-authorized unification suspension', () => {
|
||||
expect(shouldStartRuntimeInUnificationWait('SUSPENDED', 'UNIFICATION_WAIT')).toBe(true);
|
||||
});
|
||||
|
||||
it.each([
|
||||
['SUSPENDED', 'MAINTENANCE'],
|
||||
['RECONCILING', 'UNIFICATION_WAIT'],
|
||||
['RUNNING', 'UNIFICATION_WAIT'],
|
||||
])('keeps ordinary resume reconciliation for %s / %s', (phase, source) => {
|
||||
expect(shouldStartRuntimeInUnificationWait(phase, source)).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe('resolveResetLifecycleStatus', () => {
|
||||
const now = new Date('2030-01-01T00:00:00.000Z');
|
||||
|
||||
@@ -471,7 +486,9 @@ describe('buildWorkspaceCommands', () => {
|
||||
);
|
||||
const migrationUrl = new URL(command.env?.DATABASE_URL ?? '');
|
||||
|
||||
expect(migrationUrl.searchParams.getAll('options')).toEqual(['-c statement_timeout=30000 -c TimeZone=Asia/Seoul']);
|
||||
expect(migrationUrl.searchParams.getAll('options')).toEqual([
|
||||
'-c statement_timeout=30000 -c TimeZone=Asia/Seoul',
|
||||
]);
|
||||
});
|
||||
|
||||
it('keeps an already explicit KST migration contract without adding another override', () => {
|
||||
|
||||
Reference in New Issue
Block a user