Merge branch 'main' into feature/legacy-db-migration

2026-07-27 01:12:56 +00:00
parent 89013272e1 bb0abcd3de
commit 017d03eb38
8 changed files with 978 additions and 17 deletions
@@ -0,0 +1,31 @@
+import { DatabaseTurnDaemonLease } from '../../src/lifecycle/databaseTurnDaemonLease.ts';
+
+const databaseUrl = process.env.TURN_DAEMON_LEASE_DATABASE_URL;
+const profile = process.env.TURN_DAEMON_LEASE_PROFILE;
+const ownerId = process.env.TURN_DAEMON_LEASE_OWNER_ID;
+const leaseDurationMs = Number(process.env.TURN_DAEMON_LEASE_DURATION_MS);
+
+if (!databaseUrl || !profile || !ownerId || !Number.isInteger(leaseDurationMs)) {
+    throw new Error('lease holder requires database URL, profile, owner, and duration');
+}
+
+const lease = await DatabaseTurnDaemonLease.connect(databaseUrl, {
+    profile,
+    ownerId,
+    leaseDurationMs,
+    heartbeat: true,
+});
+const token = await lease.acquire();
+if (!token) {
+    throw new Error('lease holder could not acquire the profile lease');
+}
+
+process.stdout.write(
+    `${JSON.stringify({
+        profile: token.profile,
+        ownerId: token.ownerId,
+        fencingEpoch: token.fencingEpoch.toString(),
+    })}\n`
+);
+
+setInterval(() => undefined, 60_000);
@@ -1,3 +1,8 @@
+import { spawn } from 'node:child_process';
+import { once } from 'node:events';
+import { createConnection, createServer, type Server, type Socket } from 'node:net';
+import { fileURLToPath } from 'node:url';
+
 import { afterAll, beforeAll, describe, expect, it } from 'vitest';

 import { createGamePostgresConnector, type GamePrisma, type GamePrismaClient } from '@sammo-ts/infra';
@@ -7,6 +12,153 @@ import { DatabaseTurnDaemonLease, TurnDaemonLeaseLostError } from '../src/lifecy
 const databaseUrl = process.env.TURN_DAEMON_LEASE_DATABASE_URL ?? process.env.INPUT_EVENT_DATABASE_URL;
 const integration = describe.skipIf(!databaseUrl);
 const profilePrefix = 'integration:turn-lease:';
+const holderScript = fileURLToPath(new URL('./helpers/turnDaemonLeaseHolder.mjs', import.meta.url));
+
+type HolderReady = {
+    profile: string;
+    ownerId: string;
+    fencingEpoch: string;
+};
+
+const waitForHolderReady = async (child: ReturnType<typeof spawn>, timeoutMs = 10_000): Promise<HolderReady> => {
+    let output = '';
+    const ready = new Promise<HolderReady>((resolve, reject) => {
+        child.stdout?.setEncoding('utf8');
+        child.stdout?.on('data', (chunk: string) => {
+            output += chunk;
+            const newline = output.indexOf('\n');
+            if (newline < 0) {
+                return;
+            }
+            resolve(JSON.parse(output.slice(0, newline)) as HolderReady);
+        });
+        child.once('error', reject);
+        child.once('exit', (code, signal) => {
+            reject(new Error(`lease holder exited before readiness: code=${String(code)} signal=${String(signal)}`));
+        });
+    });
+    const timeout = new Promise<never>((_, reject) => {
+        setTimeout(() => reject(new Error('lease holder readiness timed out')), timeoutMs).unref();
+    });
+    return Promise.race([ready, timeout]);
+};
+
+const delay = (durationMs: number): Promise<void> =>
+    new Promise((resolve) => {
+        setTimeout(resolve, durationMs);
+    });
+
+const createPartitionableTcpProxy = async (
+    upstreamHost: string,
+    upstreamPort: number
+): Promise<{
+    port: number;
+    partition: () => Promise<void>;
+    restore: () => Promise<void>;
+    close: () => Promise<void>;
+}> => {
+    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 });
+            const pair = { client, upstream };
+            pairs.add(pair);
+            client.on('error', () => undefined);
+            upstream.on('error', () => undefined);
+            client.once('close', () => {
+                pairs.delete(pair);
+                if (!upstream.destroyed) {
+                    upstream.destroy();
+                }
+            });
+            upstream.once('close', () => {
+                pairs.delete(pair);
+                if (!client.destroyed) {
+                    client.destroy();
+                }
+            });
+            upstream.once('connect', () => resumePair(pair));
+        });
+        await new Promise<void>((resolve, reject) => {
+            const onError = (error: Error): void => reject(error);
+            nextServer.once('error', onError);
+            nextServer.listen(requestedPort, '127.0.0.1', () => {
+                nextServer.off('error', onError);
+                resolve();
+            });
+        });
+        server = nextServer;
+        const address = nextServer.address();
+        if (!address || typeof address === 'string') {
+            throw new Error('TCP proxy did not expose a numeric port');
+        }
+        port = address.port;
+    };
+
+    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) {
+            return;
+        }
+        const closed = new Promise<void>((resolve, reject) => {
+            activeServer.close((error) => {
+                if (error) {
+                    reject(error);
+                } else {
+                    resolve();
+                }
+            });
+        });
+        for (const pair of pairs) {
+            pair.client.destroy();
+            pair.upstream.destroy();
+        }
+        pairs.clear();
+        await closed;
+    };
+
+    await listen(0);
+    return {
+        get port() {
+            return port;
+        },
+        partition,
+        restore: () => {
+            partitioned = false;
+            for (const pair of pairs) {
+                resumePair(pair);
+            }
+            return Promise.resolve();
+        },
+        close,
+    };
+};

 integration('database turn daemon lease and fencing', () => {
    let db: GamePrismaClient;
@@ -115,4 +267,115 @@ integration('database turn daemon lease and fencing', () => {
        expect((await second.acquire())?.fencingEpoch).toBe(2n);
        await expect(first.assertActive()).rejects.toBeInstanceOf(TurnDaemonLeaseLostError);
    });
+
+    it('takes over after a heartbeat owner process is killed and its lease expires', async () => {
+        const profile = `${profilePrefix}process-kill`;
+        const leaseDurationMs = 1_200;
+        const holder = spawn(process.execPath, ['--experimental-strip-types', holderScript], {
+            env: {
+                ...process.env,
+                TURN_DAEMON_LEASE_DATABASE_URL: databaseUrl!,
+                TURN_DAEMON_LEASE_PROFILE: profile,
+                TURN_DAEMON_LEASE_OWNER_ID: 'process-owner',
+                TURN_DAEMON_LEASE_DURATION_MS: String(leaseDurationMs),
+            },
+            stdio: ['ignore', 'pipe', 'pipe'],
+        });
+        let successor: DatabaseTurnDaemonLease | null = null;
+        try {
+            await expect(waitForHolderReady(holder)).resolves.toEqual({
+                profile,
+                ownerId: 'process-owner',
+                fencingEpoch: '1',
+            });
+            expect(await db.turnDaemonLease.findUniqueOrThrow({ where: { profile } })).toMatchObject({
+                ownerId: 'process-owner',
+                fencingEpoch: 1n,
+            });
+
+            expect(holder.kill('SIGKILL')).toBe(true);
+            const [exitCode, signal] = (await once(holder, 'exit')) as [number | null, NodeJS.Signals | null];
+            expect(exitCode).toBeNull();
+            expect(signal).toBe('SIGKILL');
+
+            successor = await createLease(profile, 'successor-owner');
+            expect(await successor.acquire()).toBeNull();
+
+            const deadline = Date.now() + leaseDurationMs * 4;
+            let successorToken = null;
+            while (Date.now() < deadline) {
+                successorToken = await successor.acquire();
+                if (successorToken) {
+                    break;
+                }
+                await delay(100);
+            }
+            expect(successorToken).toMatchObject({
+                profile,
+                ownerId: 'successor-owner',
+                fencingEpoch: 2n,
+            });
+            expect(await successor.assertActive()).toBeUndefined();
+        } finally {
+            if (holder.exitCode === null && holder.signalCode === null) {
+                holder.kill('SIGKILL');
+                await once(holder, 'exit');
+            }
+            await successor?.release();
+        }
+    }, 15_000);
+
+    it('stays fenced after a database partition outlasts the heartbeat lease and connectivity recovers', async () => {
+        const profile = `${profilePrefix}database-partition`;
+        const leaseDurationMs = 1_800;
+        const directUrl = new URL(databaseUrl!);
+        const upstreamPort = Number(directUrl.port || '5432');
+        const proxy = await createPartitionableTcpProxy(directUrl.hostname, upstreamPort);
+        directUrl.hostname = '127.0.0.1';
+        directUrl.port = String(proxy.port);
+        const partitionedOwner = await DatabaseTurnDaemonLease.connect(directUrl.toString(), {
+            profile,
+            ownerId: 'partitioned-owner',
+            leaseDurationMs,
+            heartbeat: true,
+        });
+        leases.push(partitionedOwner);
+        let successor: DatabaseTurnDaemonLease | null = null;
+        try {
+            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);
+
+            const takeoverDeadline = Date.now() + leaseDurationMs * 4;
+            let successorToken = null;
+            while (Date.now() < takeoverDeadline) {
+                successorToken = await successor.acquire();
+                if (successorToken) {
+                    break;
+                }
+                await delay(100);
+            }
+            expect(successorToken).toMatchObject({
+                profile,
+                ownerId: 'partition-successor',
+                fencingEpoch: 2n,
+            });
+
+            await proxy.restore();
+            expect(await partitionedOwner.renew()).toBe(false);
+            await expect(partitionedOwner.assertActive()).rejects.toBeInstanceOf(TurnDaemonLeaseLostError);
+            await expect(successor.assertActive()).resolves.toBeUndefined();
+        } finally {
+            await proxy.close();
+            await successor?.release();
+        }
+    }, 20_000);
 });