fix reserved turn daemon api serialization

2026-07-26 19:23:25 +00:00
parent d1e9d5fc75
commit 8f1fe73bad
12 changed files with 1003 additions and 74 deletions
@@ -1,4 +1,5 @@
 import { describe, expect, it, vi } from 'vitest';
+import { asRecord } from '@sammo-ts/common';

 import {
    InMemoryControlQueue,
@@ -27,7 +28,34 @@ const processor: TurnProcessor = {
 describe('input event atomicity', () => {
    it('keeps reserved-turn dirty state when persistence fails', async () => {
        let failCreate = true;
-        const revisionUpsert = vi.fn(async () => ({ revision: 1 }));
+        let revision: { revision: number; leaseOwner: string | null; leaseExpiresAt: Date | null } | null = null;
+        const revisionUpdate = vi.fn(async (rawArgs: unknown) => {
+            if (!revision) {
+                return { count: 0 };
+            }
+            const data = asRecord(asRecord(rawArgs).data);
+            const revisionChange = asRecord(data.revision);
+            if (typeof revisionChange.increment === 'number') {
+                revision.revision += revisionChange.increment;
+            }
+            if ('leaseOwner' in data) {
+                revision.leaseOwner = typeof data.leaseOwner === 'string' ? data.leaseOwner : null;
+            }
+            if ('leaseExpiresAt' in data) {
+                revision.leaseExpiresAt = data.leaseExpiresAt instanceof Date ? data.leaseExpiresAt : null;
+            }
+            return { count: 1 };
+        });
+        const revisionCreate = vi.fn(async (rawArgs: unknown) => {
+            const rawData = asRecord(rawArgs).data;
+            const row = Array.isArray(rawData) ? asRecord(rawData[0]) : asRecord(rawData);
+            revision = {
+                revision: typeof row.revision === 'number' ? row.revision : 0,
+                leaseOwner: typeof row.leaseOwner === 'string' ? row.leaseOwner : null,
+                leaseExpiresAt: row.leaseExpiresAt instanceof Date ? row.leaseExpiresAt : null,
+            };
+            return { count: 1 };
+        });
        const prisma = {
            generalTurn: {
                findMany: vi.fn(async () => []),
@@ -40,7 +68,9 @@ describe('input event atomicity', () => {
                }),
            },
            generalTurnRevision: {
-                upsert: revisionUpsert,
+                findUnique: vi.fn(async () => null),
+                createMany: revisionCreate,
+                updateMany: revisionUpdate,
            },
            nationTurn: {
                findMany: vi.fn(async () => []),
@@ -51,6 +81,7 @@ describe('input event atomicity', () => {
        const store = new InMemoryReservedTurnStore(prisma, {
            maxGeneralTurns: 1,
            maxNationTurns: 1,
+            leaseOwner: 'test-daemon',
        });
        store.shiftGeneralTurns(7, -1);
        store.ensureGeneralTurns(8);
@@ -59,23 +90,38 @@ describe('input event atomicity', () => {
        expect(store.peekDirtyState()).toEqual({
            generalIds: [7],
            generalInitializationIds: [8],
+            generalLeaseIds: [],
            nationKeys: [],
            nationInitializationKeys: [],
+            nationLeaseKeys: [],
        });

        failCreate = false;
        await store.flushChanges();
-        expect(revisionUpsert).toHaveBeenCalledOnce();
-        expect(revisionUpsert).toHaveBeenCalledWith({
-            where: { generalId: 7 },
-            create: { generalId: 7, revision: 1 },
-            update: { revision: { increment: 1 } },
+        expect(revisionCreate).toHaveBeenCalledOnce();
+        expect(revisionCreate).toHaveBeenCalledWith({
+            data: [
+                {
+                    generalId: 7,
+                    revision: 0,
+                    leaseOwner: 'test-daemon',
+                    leaseExpiresAt: expect.any(Date),
+                },
+            ],
+            skipDuplicates: true,
+        });
+        expect(revision).toMatchObject({
+            revision: 1,
+            leaseOwner: null,
+            leaseExpiresAt: null,
        });
        expect(store.peekDirtyState()).toEqual({
            generalIds: [],
            generalInitializationIds: [],
+            generalLeaseIds: [],
            nationKeys: [],
            nationInitializationKeys: [],
+            nationLeaseKeys: [],
        });
    });

@@ -197,8 +197,10 @@ describe('CreateManyNPC monthly action', () => {
        expect(reservedTurns.peekDirtyState()).toEqual({
            generalIds: [],
            generalInitializationIds: [created.id],
+            generalLeaseIds: [],
            nationKeys: [],
            nationInitializationKeys: [],
+            nationLeaseKeys: [],
        });
        expect(world.peekDirtyState().logs).toMatchInlineSnapshot(`
          [
@@ -0,0 +1,135 @@
+import { afterAll, beforeAll, describe, expect, it } from 'vitest';
+import { createGamePostgresConnector } from '@sammo-ts/infra';
+
+import {
+    InMemoryReservedTurnStore,
+    ReservedTurnLeaseConflictError,
+} from '../src/turn/reservedTurnStore.js';
+
+const databaseUrl = process.env.RESERVED_TURN_DATABASE_URL;
+const describeIntegration = databaseUrl ? describe : describe.skip;
+const GENERAL_ID = 2_147_400_002;
+
+describeIntegration('reserved turn daemon/API lease integration', () => {
+    const connector = databaseUrl ? createGamePostgresConnector({ url: databaseUrl }) : null;
+
+    beforeAll(async () => {
+        if (!connector) {
+            return;
+        }
+        await connector.connect();
+        await connector.prisma.generalTurn.deleteMany({ where: { generalId: GENERAL_ID } });
+        await connector.prisma.generalTurnRevision.deleteMany({ where: { generalId: GENERAL_ID } });
+        await connector.prisma.generalTurn.createMany({
+            data: [
+                { generalId: GENERAL_ID, turnIdx: 0, actionCode: 'che_훈련', arg: {} },
+                { generalId: GENERAL_ID, turnIdx: 1, actionCode: 'che_사기진작', arg: {} },
+            ],
+        });
+    });
+
+    afterAll(async () => {
+        if (!connector) {
+            return;
+        }
+        await connector.prisma.generalTurn.deleteMany({ where: { generalId: GENERAL_ID } });
+        await connector.prisma.generalTurnRevision.deleteMany({ where: { generalId: GENERAL_ID } });
+        await connector.disconnect();
+    });
+
+    it('blocks the API CAS during execution and releases the lease atomically with the shifted queue', async () => {
+        if (!connector) {
+            throw new Error('integration connector is unavailable');
+        }
+        const store = new InMemoryReservedTurnStore(connector.prisma, {
+            maxGeneralTurns: 2,
+            maxNationTurns: 1,
+            leaseOwner: 'integration-daemon',
+            leaseDurationMs: 60_000,
+        });
+
+        await store.prepareTurnsForExecution(GENERAL_ID);
+        const leased = await connector.prisma.generalTurnRevision.findUniqueOrThrow({
+            where: { generalId: GENERAL_ID },
+        });
+        expect(leased).toMatchObject({ revision: 0, leaseOwner: 'integration-daemon' });
+
+        const blockedApiClaim = await connector.prisma.generalTurnRevision.updateMany({
+            where: {
+                generalId: GENERAL_ID,
+                revision: 0,
+                OR: [{ leaseOwner: null }, { leaseExpiresAt: { lte: new Date() } }],
+            },
+            data: {
+                revision: 1,
+                leaseOwner: null,
+                leaseExpiresAt: null,
+            },
+        });
+        expect(blockedApiClaim.count).toBe(0);
+
+        store.shiftGeneralTurns(GENERAL_ID, -1);
+        const changes = store.peekDirtyState();
+        await connector.prisma.$transaction((transaction) => store.persistChanges(transaction, changes));
+        store.acknowledgeDirtyState(changes);
+
+        const committed = await connector.prisma.generalTurnRevision.findUniqueOrThrow({
+            where: { generalId: GENERAL_ID },
+        });
+        expect(committed).toMatchObject({
+            revision: 1,
+            leaseOwner: null,
+            leaseExpiresAt: null,
+        });
+        const turns = await connector.prisma.generalTurn.findMany({
+            where: { generalId: GENERAL_ID },
+            orderBy: { turnIdx: 'asc' },
+        });
+        expect(turns.map((turn) => turn.actionCode)).toEqual(['che_사기진작', '휴식']);
+
+        const nextApiClaim = await connector.prisma.generalTurnRevision.updateMany({
+            where: {
+                generalId: GENERAL_ID,
+                revision: 1,
+                OR: [{ leaseOwner: null }, { leaseExpiresAt: { lte: new Date() } }],
+            },
+            data: { revision: 2 },
+        });
+        expect(nextApiClaim.count).toBe(1);
+    });
+
+    it('blocks a daemon-side replacement before deleting turns while an API lease is active', async () => {
+        if (!connector) {
+            throw new Error('integration connector is unavailable');
+        }
+        await connector.prisma.generalTurnRevision.update({
+            where: { generalId: GENERAL_ID },
+            data: {
+                leaseOwner: 'integration-api',
+                leaseExpiresAt: new Date(Date.now() + 60_000),
+            },
+        });
+        const before = await connector.prisma.generalTurn.findMany({
+            where: { generalId: GENERAL_ID },
+            orderBy: { turnIdx: 'asc' },
+        });
+        const store = new InMemoryReservedTurnStore(connector.prisma, {
+            maxGeneralTurns: 2,
+            maxNationTurns: 1,
+            leaseOwner: 'integration-daemon-replacement',
+            leaseDurationMs: 60_000,
+        });
+        store.replaceGeneralTurns(GENERAL_ID, { action: 'che_훈련', args: {} });
+        const changes = store.peekDirtyState();
+
+        await expect(
+            connector.prisma.$transaction((transaction) => store.persistChanges(transaction, changes))
+        ).rejects.toBeInstanceOf(ReservedTurnLeaseConflictError);
+
+        const after = await connector.prisma.generalTurn.findMany({
+            where: { generalId: GENERAL_ID },
+            orderBy: { turnIdx: 'asc' },
+        });
+        expect(after.map((turn) => turn.actionCode)).toEqual(before.map((turn) => turn.actionCode));
+    });
+});
@@ -0,0 +1,288 @@
+import { describe, expect, it, vi } from 'vitest';
+import { asRecord } from '@sammo-ts/common';
+
+import {
+    InMemoryReservedTurnStore,
+    ReservedTurnLeaseConflictError,
+} from '../src/turn/reservedTurnStore.js';
+
+interface RevisionRow {
+    revision: number;
+    leaseOwner: string | null;
+    leaseExpiresAt: Date | null;
+}
+
+const buildHarness = (initialRevision: RevisionRow | null = null) => {
+    let revision = initialRevision;
+    let nationRevision: RevisionRow | null = null;
+    const writtenTurns: unknown[] = [];
+    const generalFindMany = vi.fn(async () => [
+        {
+            id: 1,
+            generalId: 7,
+            turnIdx: 0,
+            actionCode: 'che_훈련',
+            arg: {},
+            createdAt: new Date(),
+        },
+    ]);
+    const generalDeleteMany = vi.fn(async () => ({ count: 1 }));
+    const generalCreateMany = vi.fn(async (args: unknown) => {
+        writtenTurns.push(args);
+        return { count: 2 };
+    });
+    const updateMany = vi.fn(async (rawArgs: unknown) => {
+        const args = asRecord(rawArgs);
+        const where = asRecord(args.where);
+        const data = asRecord(args.data);
+        if (!revision) {
+            return { count: 0 };
+        }
+        if (typeof where.generalId === 'number' && where.generalId !== 7) {
+            return { count: 0 };
+        }
+        if (typeof where.leaseOwner === 'string' && where.leaseOwner !== revision.leaseOwner) {
+            return { count: 0 };
+        }
+        if (Array.isArray(where.OR)) {
+            const now = new Date();
+            const eligible =
+                revision.leaseOwner === null ||
+                revision.leaseOwner === 'daemon-1' ||
+                (revision.leaseExpiresAt !== null && revision.leaseExpiresAt.getTime() <= now.getTime());
+            if (!eligible) {
+                return { count: 0 };
+            }
+        }
+        const revisionChange = asRecord(data.revision);
+        if (typeof revisionChange.increment === 'number') {
+            revision.revision += revisionChange.increment;
+        }
+        if ('leaseOwner' in data) {
+            revision.leaseOwner = typeof data.leaseOwner === 'string' ? data.leaseOwner : null;
+        }
+        if ('leaseExpiresAt' in data) {
+            revision.leaseExpiresAt = data.leaseExpiresAt instanceof Date ? data.leaseExpiresAt : null;
+        }
+        return { count: 1 };
+    });
+    const createMany = vi.fn(async (rawArgs: unknown) => {
+        if (revision) {
+            return { count: 0 };
+        }
+        const args = asRecord(rawArgs);
+        const rows = Array.isArray(args.data) ? args.data : [];
+        const data = asRecord(rows[0]);
+        revision = {
+            revision: typeof data.revision === 'number' ? data.revision : 0,
+            leaseOwner: typeof data.leaseOwner === 'string' ? data.leaseOwner : null,
+            leaseExpiresAt: data.leaseExpiresAt instanceof Date ? data.leaseExpiresAt : null,
+        };
+        return { count: 1 };
+    });
+    const nationRevisionUpdateMany = vi.fn(async (rawArgs: unknown) => {
+        const args = asRecord(rawArgs);
+        const where = asRecord(args.where);
+        const data = asRecord(args.data);
+        if (!nationRevision) {
+            return { count: 0 };
+        }
+        if (typeof where.leaseOwner === 'string' && where.leaseOwner !== nationRevision.leaseOwner) {
+            return { count: 0 };
+        }
+        const revisionChange = asRecord(data.revision);
+        if (typeof revisionChange.increment === 'number') {
+            nationRevision.revision += revisionChange.increment;
+        }
+        if ('leaseOwner' in data) {
+            nationRevision.leaseOwner = typeof data.leaseOwner === 'string' ? data.leaseOwner : null;
+        }
+        if ('leaseExpiresAt' in data) {
+            nationRevision.leaseExpiresAt = data.leaseExpiresAt instanceof Date ? data.leaseExpiresAt : null;
+        }
+        return { count: 1 };
+    });
+    const nationRevisionCreateMany = vi.fn(async (rawArgs: unknown) => {
+        if (nationRevision) {
+            return { count: 0 };
+        }
+        const args = asRecord(rawArgs);
+        const rows = Array.isArray(args.data) ? args.data : [];
+        const data = asRecord(rows[0]);
+        nationRevision = {
+            revision: typeof data.revision === 'number' ? data.revision : 0,
+            leaseOwner: typeof data.leaseOwner === 'string' ? data.leaseOwner : null,
+            leaseExpiresAt: data.leaseExpiresAt instanceof Date ? data.leaseExpiresAt : null,
+        };
+        return { count: 1 };
+    });
+    const prisma = {
+        generalTurn: {
+            findMany: generalFindMany,
+            deleteMany: generalDeleteMany,
+            createMany: generalCreateMany,
+        },
+        generalTurnRevision: {
+            findUnique: vi.fn(async () => revision),
+            createMany,
+            updateMany,
+        },
+        nationTurn: {
+            findMany: vi.fn(async () => [
+                {
+                    id: 1,
+                    nationId: 3,
+                    officerLevel: 12,
+                    turnIdx: 0,
+                    actionCode: 'che_포상',
+                    arg: {},
+                    createdAt: new Date(),
+                },
+            ]),
+            deleteMany: vi.fn(async () => ({ count: 0 })),
+            createMany: vi.fn(async () => ({ count: 0 })),
+        },
+        nationTurnRevision: {
+            findUnique: vi.fn(async () => nationRevision),
+            createMany: nationRevisionCreateMany,
+            updateMany: nationRevisionUpdateMany,
+        },
+    };
+    const store = new InMemoryReservedTurnStore(prisma, {
+        maxGeneralTurns: 2,
+        maxNationTurns: 1,
+        leaseOwner: 'daemon-1',
+        leaseDurationMs: 60_000,
+    });
+    return {
+        store,
+        prisma,
+        generalFindMany,
+        generalDeleteMany,
+        generalCreateMany,
+        writtenTurns,
+        getRevision: () => revision,
+        getNationRevision: () => nationRevision,
+        stealLease: () => {
+            if (revision) {
+                revision.leaseOwner = 'other-writer';
+            }
+        },
+    };
+};
+
+describe('reserved turn daemon lease', () => {
+    it('holds the queue lease from refresh through shift and releases it with the revision increment', async () => {
+        const harness = buildHarness();
+
+        await harness.store.prepareTurnsForExecution(7);
+        expect(harness.getRevision()).toMatchObject({ revision: 0, leaseOwner: 'daemon-1' });
+        expect(harness.store.getGeneralTurn(7, 0).action).toBe('che_훈련');
+
+        harness.store.shiftGeneralTurns(7, -1);
+        const changes = harness.store.peekDirtyState();
+        expect(changes).toMatchObject({ generalIds: [7], generalLeaseIds: [7] });
+        await harness.store.persistChanges(harness.prisma, changes);
+        harness.store.acknowledgeDirtyState(changes);
+
+        expect(harness.getRevision()).toMatchObject({
+            revision: 1,
+            leaseOwner: null,
+            leaseExpiresAt: null,
+        });
+        expect(harness.generalDeleteMany).toHaveBeenCalledOnce();
+        expect(harness.generalCreateMany).toHaveBeenCalledOnce();
+        expect(harness.store.peekDirtyState()).toEqual({
+            generalIds: [],
+            generalInitializationIds: [],
+            generalLeaseIds: [],
+            nationKeys: [],
+            nationInitializationKeys: [],
+            nationLeaseKeys: [],
+        });
+    });
+
+    it('rejects an active foreign lease before reading the queue', async () => {
+        const harness = buildHarness({
+            revision: 4,
+            leaseOwner: 'api-writer',
+            leaseExpiresAt: new Date(Date.now() + 60_000),
+        });
+
+        await expect(harness.store.prepareTurnsForExecution(7)).rejects.toBeInstanceOf(
+            ReservedTurnLeaseConflictError
+        );
+        expect(harness.generalFindMany).not.toHaveBeenCalled();
+    });
+
+    it('takes over an expired foreign lease before reading the queue', async () => {
+        const harness = buildHarness({
+            revision: 4,
+            leaseOwner: 'stopped-daemon',
+            leaseExpiresAt: new Date(Date.now() - 1_000),
+        });
+
+        await harness.store.prepareTurnsForExecution(7);
+
+        expect(harness.getRevision()).toMatchObject({
+            revision: 4,
+            leaseOwner: 'daemon-1',
+        });
+        expect(harness.generalFindMany).toHaveBeenCalledOnce();
+    });
+
+    it('leases and releases the matching nation officer queue with the general queue', async () => {
+        const harness = buildHarness();
+
+        await harness.store.prepareTurnsForExecution(7, { nationId: 3, officerLevel: 12 });
+        expect(harness.getRevision()).toMatchObject({ revision: 0, leaseOwner: 'daemon-1' });
+        expect(harness.getNationRevision()).toMatchObject({ revision: 0, leaseOwner: 'daemon-1' });
+
+        harness.store.shiftNationTurns(3, 12, -1);
+        const changes = harness.store.peekDirtyState();
+        expect(changes).toMatchObject({
+            generalLeaseIds: [7],
+            nationKeys: ['3:12'],
+            nationLeaseKeys: ['3:12'],
+        });
+        await harness.store.persistChanges(harness.prisma, changes);
+        harness.store.acknowledgeDirtyState(changes);
+
+        expect(harness.getRevision()).toMatchObject({ revision: 0, leaseOwner: null });
+        expect(harness.getNationRevision()).toMatchObject({ revision: 1, leaseOwner: null });
+    });
+
+    it('detects a lost lease before deleting or replacing turns', async () => {
+        const harness = buildHarness();
+        await harness.store.prepareTurnsForExecution(7);
+        harness.store.shiftGeneralTurns(7, -1);
+        harness.stealLease();
+        const changes = harness.store.peekDirtyState();
+
+        await expect(harness.store.persistChanges(harness.prisma, changes)).rejects.toBeInstanceOf(
+            ReservedTurnLeaseConflictError
+        );
+        expect(harness.generalDeleteMany).not.toHaveBeenCalled();
+        expect(harness.generalCreateMany).not.toHaveBeenCalled();
+        expect(harness.store.peekDirtyState()).toMatchObject({
+            generalIds: [7],
+            generalLeaseIds: [7],
+        });
+    });
+
+    it('blocks a daemon-side queue replacement behind an active foreign lease', async () => {
+        const harness = buildHarness({
+            revision: 6,
+            leaseOwner: 'api-writer',
+            leaseExpiresAt: new Date(Date.now() + 60_000),
+        });
+        harness.store.replaceGeneralTurns(7, { action: 'che_훈련', args: {} });
+        const changes = harness.store.peekDirtyState();
+
+        await expect(harness.store.persistChanges(harness.prisma, changes)).rejects.toBeInstanceOf(
+            ReservedTurnLeaseConflictError
+        );
+        expect(harness.generalDeleteMany).not.toHaveBeenCalled();
+        expect(harness.generalCreateMany).not.toHaveBeenCalled();
+    });
+});