diff --git a/app/game-engine/src/lifecycle/databaseTurnDaemonLease.ts b/app/game-engine/src/lifecycle/databaseTurnDaemonLease.ts index a40fe2d..b4eeba3 100644 --- a/app/game-engine/src/lifecycle/databaseTurnDaemonLease.ts +++ b/app/game-engine/src/lifecycle/databaseTurnDaemonLease.ts @@ -1,4 +1,5 @@ import { randomUUID } from 'node:crypto'; +import { performance } from 'node:perf_hooks'; import { createGamePostgresConnector, GamePrisma, type GamePrismaClient } from '@sammo-ts/infra'; @@ -46,6 +47,8 @@ export class DatabaseTurnDaemonLease { private readonly heartbeatEnabled: boolean; private token: TurnDaemonLeaseToken | null = null; private heartbeatTimer: NodeJS.Timeout | null = null; + private expiryTimer: NodeJS.Timeout | null = null; + private renewalInFlight = false; private lost = false; private constructor( @@ -71,6 +74,7 @@ export class DatabaseTurnDaemonLease { } async acquire(): Promise { + const requestStartedAt = performance.now(); const rows = await this.db.$queryRaw(GamePrisma.sql` INSERT INTO "turn_daemon_lease" ( "profile", @@ -111,6 +115,7 @@ export class DatabaseTurnDaemonLease { fencingEpoch: BigInt(row.fencing_epoch), }; this.lost = false; + this.scheduleExpiryWatchdog(requestStartedAt); if (this.heartbeatEnabled) { this.startHeartbeat(); } @@ -127,26 +132,36 @@ export class DatabaseTurnDaemonLease { async renew(): Promise { const token = this.token; - if (!token || this.lost) { + if (!token || this.lost || this.renewalInFlight) { return false; } - const rows = await this.db.$queryRaw(GamePrisma.sql` - UPDATE "turn_daemon_lease" - SET - "lease_until" = CURRENT_TIMESTAMP + (${this.leaseDurationMs} * INTERVAL '1 millisecond'), - "heartbeat_at" = CURRENT_TIMESTAMP - WHERE - "profile" = ${token.profile} - AND "owner_id" = ${token.ownerId} - AND "fencing_epoch" = ${token.fencingEpoch} - AND "lease_until" > CURRENT_TIMESTAMP - RETURNING "profile", "owner_id", "fencing_epoch" - `); - if (rows.length === 0) { - this.markLost(); - return false; + const requestStartedAt = performance.now(); + this.renewalInFlight = true; + try { + const rows = await this.db.$queryRaw(GamePrisma.sql` + UPDATE "turn_daemon_lease" + SET + "lease_until" = CURRENT_TIMESTAMP + (${this.leaseDurationMs} * INTERVAL '1 millisecond'), + "heartbeat_at" = CURRENT_TIMESTAMP + WHERE + "profile" = ${token.profile} + AND "owner_id" = ${token.ownerId} + AND "fencing_epoch" = ${token.fencingEpoch} + AND "lease_until" > CURRENT_TIMESTAMP + RETURNING "profile", "owner_id", "fencing_epoch" + `); + if (rows.length === 0) { + this.markLost(); + return false; + } + if (this.lost) { + return false; + } + this.scheduleExpiryWatchdog(requestStartedAt); + return true; + } finally { + this.renewalInFlight = false; } - return true; } async assertActive(transaction?: GamePrisma.TransactionClient): Promise { @@ -173,6 +188,7 @@ export class DatabaseTurnDaemonLease { async release(): Promise { this.stopHeartbeat(); + this.stopExpiryWatchdog(); const token = this.token; this.token = null; if (!token || this.lost) { @@ -216,8 +232,27 @@ export class DatabaseTurnDaemonLease { } } + private scheduleExpiryWatchdog(requestStartedAt: number): void { + // DB가 lease_until을 정하는 시점보다 앞선 요청 시작 시각을 기준으로 + // 잡아, heartbeat 응답이 멈춰도 DB lease 만료보다 늦게 pause하지 않는다. + this.stopExpiryWatchdog(); + const remainingMs = Math.max(0, this.leaseDurationMs - (performance.now() - requestStartedAt)); + this.expiryTimer = setTimeout(() => { + this.markLost(); + }, remainingMs); + this.expiryTimer.unref(); + } + + private stopExpiryWatchdog(): void { + if (this.expiryTimer) { + clearTimeout(this.expiryTimer); + this.expiryTimer = null; + } + } + private markLost(): void { this.lost = true; this.stopHeartbeat(); + this.stopExpiryWatchdog(); } } diff --git a/app/game-engine/test/turnDaemonLease.integration.test.ts b/app/game-engine/test/turnDaemonLease.integration.test.ts index 4531ad6..540620c 100644 --- a/app/game-engine/test/turnDaemonLease.integration.test.ts +++ b/app/game-engine/test/turnDaemonLease.integration.test.ts @@ -57,31 +57,43 @@ const createPartitionableTcpProxy = async ( restore: () => Promise; close: () => Promise; }> => { - const sockets = new Set(); + type ProxyPair = { + client: Socket; + upstream: Socket; + }; + const pairs = new Set(); let server: Server | null = null; let port = 0; + let partitioned = false; + + const resumePair = (pair: ProxyPair): void => { + if (partitioned || pair.client.destroyed || pair.upstream.destroyed || pair.upstream.connecting) { + return; + } + pair.client.pipe(pair.upstream); + pair.upstream.pipe(pair.client); + }; const listen = async (requestedPort: number): Promise => { const nextServer = createServer((client) => { const upstream = createConnection({ host: upstreamHost, port: upstreamPort }); - sockets.add(client); - sockets.add(upstream); + const pair = { client, upstream }; + pairs.add(pair); client.on('error', () => undefined); upstream.on('error', () => undefined); client.once('close', () => { - sockets.delete(client); + pairs.delete(pair); if (!upstream.destroyed) { upstream.destroy(); } }); upstream.once('close', () => { - sockets.delete(upstream); + pairs.delete(pair); if (!client.destroyed) { client.destroy(); } }); - client.pipe(upstream); - upstream.pipe(client); + upstream.once('connect', () => resumePair(pair)); }); await new Promise((resolve, reject) => { const onError = (error: Error): void => reject(error); @@ -99,7 +111,16 @@ const createPartitionableTcpProxy = async ( port = address.port; }; - const partition = async (): Promise => { + const partition = (): Promise => { + partitioned = true; + for (const pair of pairs) { + pair.client.unpipe(pair.upstream); + pair.upstream.unpipe(pair.client); + } + return Promise.resolve(); + }; + + const close = async (): Promise => { const activeServer = server; server = null; if (!activeServer) { @@ -114,10 +135,11 @@ const createPartitionableTcpProxy = async ( } }); }); - for (const socket of sockets) { - socket.destroy(); + for (const pair of pairs) { + pair.client.destroy(); + pair.upstream.destroy(); } - sockets.clear(); + pairs.clear(); await closed; }; @@ -127,12 +149,14 @@ const createPartitionableTcpProxy = async ( return port; }, partition, - restore: async () => { - if (!server) { - await listen(port); + restore: () => { + partitioned = false; + for (const pair of pairs) { + resumePair(pair); } + return Promise.resolve(); }, - close: partition, + close, }; }; @@ -321,15 +345,15 @@ integration('database turn daemon lease and fencing', () => { expect((await partitionedOwner.acquire())?.fencingEpoch).toBe(1n); await proxy.partition(); + successor = await createLease(profile, 'partition-successor'); + expect(await successor.acquire()).toBeNull(); + const lostDeadline = Date.now() + leaseDurationMs * 2; while (!partitionedOwner.isLost() && Date.now() < lostDeadline) { await delay(50); } expect(partitionedOwner.isLost()).toBe(true); - successor = await createLease(profile, 'partition-successor'); - expect(await successor.acquire()).toBeNull(); - const takeoverDeadline = Date.now() + leaseDurationMs * 4; let successorToken = null; while (Date.now() < takeoverDeadline) {