fix lease fencing during silent database partitions

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<TurnDaemonLeaseToken | null> {
+        const requestStartedAt = performance.now();
         const rows = await this.db.$queryRaw<LeaseRow[]>(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<boolean> {
         const token = this.token;
-        if (!token || this.lost) {
+        if (!token || this.lost || this.renewalInFlight) {
             return false;
         }
-        const rows = await this.db.$queryRaw<LeaseRow[]>(GamePrisma.sql`
+        const requestStartedAt = performance.now();
-            UPDATE "turn_daemon_lease"
+        this.renewalInFlight = true;
-            SET
+        try {
-                "lease_until" = CURRENT_TIMESTAMP + (${this.leaseDurationMs} * INTERVAL '1 millisecond'),
+            const rows = await this.db.$queryRaw<LeaseRow[]>(GamePrisma.sql`
-                "heartbeat_at" = CURRENT_TIMESTAMP
+                UPDATE "turn_daemon_lease"
-            WHERE
+                SET
-                "profile" = ${token.profile}
+                    "lease_until" = CURRENT_TIMESTAMP + (${this.leaseDurationMs} * INTERVAL '1 millisecond'),
-                AND "owner_id" = ${token.ownerId}
+                    "heartbeat_at" = CURRENT_TIMESTAMP
-                AND "fencing_epoch" = ${token.fencingEpoch}
+                WHERE
-                AND "lease_until" > CURRENT_TIMESTAMP
+                    "profile" = ${token.profile}
-            RETURNING "profile", "owner_id", "fencing_epoch"
+                    AND "owner_id" = ${token.ownerId}
-        `);
+                    AND "fencing_epoch" = ${token.fencingEpoch}
-        if (rows.length === 0) {
+                    AND "lease_until" > CURRENT_TIMESTAMP
-            this.markLost();
+                RETURNING "profile", "owner_id", "fencing_epoch"
-            return false;
+            `);
+            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<void> {
@@ -173,6 +188,7 @@ export class DatabaseTurnDaemonLease {
 
     async release(): Promise<void> {
         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();
     }
 }

app/game-engine/test/turnDaemonLease.integration.test.ts

@@ -57,31 +57,43 @@ const createPartitionableTcpProxy = async (
     restore: () => Promise<void>;
     close: () => Promise<void>;
 }> => {
-    const sockets = new Set<Socket>();
+    type ProxyPair = {
+        client: Socket;
+        upstream: Socket;
+    };
+    const pairs = new Set<ProxyPair>();
     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<void> => {
         const nextServer = createServer((client) => {
             const upstream = createConnection({ host: upstreamHost, port: upstreamPort });
-            sockets.add(client);
+            const pair = { client, upstream };
-            sockets.add(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.once('connect', () => resumePair(pair));
-            upstream.pipe(client);
         });
         await new Promise<void>((resolve, reject) => {
             const onError = (error: Error): void => reject(error);
@@ -99,7 +111,16 @@ const createPartitionableTcpProxy = async (
         port = address.port;
     };
 
-    const partition = async (): Promise<void> => {
+    const partition = (): Promise<void> => {
+        partitioned = true;
+        for (const pair of pairs) {
+            pair.client.unpipe(pair.upstream);
+            pair.upstream.unpipe(pair.client);
+        }
+        return Promise.resolve();
+    };
+
+    const close = async (): Promise<void> => {
         const activeServer = server;
         server = null;
         if (!activeServer) {
@@ -114,10 +135,11 @@ const createPartitionableTcpProxy = async (
                 }
             });
         });
-        for (const socket of sockets) {
+        for (const pair of pairs) {
-            socket.destroy();
+            pair.client.destroy();
+            pair.upstream.destroy();
         }
-        sockets.clear();
+        pairs.clear();
         await closed;
     };
 
@@ -127,12 +149,14 @@ const createPartitionableTcpProxy = async (
             return port;
         },
         partition,
-        restore: async () => {
+        restore: () => {
-            if (!server) {
+            partitioned = false;
-                await listen(port);
+            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) {