feat: 이전 기수 감사 표본을 제한된 batch로 정리

2026-09-16 04:22:09 +00:00
parent 9e14306fe4
commit 10f3f0ebc6
10 changed files with 475 additions and 3 deletions
@@ -0,0 +1,72 @@
+import { asRecord } from '@sammo-ts/common';
+import type { GamePrismaClient } from '@sammo-ts/infra';
+
+export const AUDIT_RETENTION_BATCH_SIZE = 200;
+export type AuditRetentionResult = { status: 'progress' | 'complete' | 'busy' | 'identityChanged'; deleted: number };
+
+/** 새 기수가 활성화된 뒤에만 이전 감사 projection을 작은 transaction으로 정리한다. */
+export const prunePreviousAuditBatch = async (
+    db: GamePrismaClient,
+    expectedServerId: string
+): Promise<AuditRetentionResult> => {
+    if (!expectedServerId.trim()) return { status: 'identityChanged', deleted: 0 };
+    return db.$transaction(
+        async (tx) => {
+            await tx.$executeRaw`SET LOCAL statement_timeout = '2000ms'`;
+            // seeder와 동일한 잠금이다. RESET을 기다리게 하지 않고 다음 batch에서 재시도한다.
+            const [lock] = await tx.$queryRaw<{ locked: boolean }[]>`
+            SELECT pg_try_advisory_xact_lock(hashtextextended(current_schema(), 0)) AS locked
+        `;
+            if (!lock?.locked) return { status: 'busy', deleted: 0 };
+            const world = await tx.worldState.findFirst({ orderBy: { id: 'asc' }, select: { meta: true } });
+            if (asRecord(world?.meta).serverId !== expectedServerId) return { status: 'identityChanged', deleted: 0 };
+            // 부모를 잠가 늦은 child INSERT와 빈 header 삭제의 경쟁도 차단한다.
+            const [sample] = await tx.$queryRaw<{ id: string }[]>`
+            SELECT id FROM play_audit_month WHERE server_id <> ${expectedServerId}
+            ORDER BY id LIMIT 1 FOR UPDATE
+        `;
+            if (!sample) return { status: 'complete', deleted: 0 };
+            const where = { sampleId: sample.id };
+            const generals = await tx.playAuditGeneral.findMany({
+                where,
+                orderBy: { generalId: 'asc' },
+                take: AUDIT_RETENTION_BATCH_SIZE,
+                select: { generalId: true },
+            });
+            if (generals.length) {
+                const deleted = await tx.playAuditGeneral.deleteMany({
+                    where: { ...where, generalId: { in: generals.map((row) => row.generalId) } },
+                });
+                return { status: 'progress', deleted: deleted.count };
+            }
+            const cities = await tx.playAuditCity.findMany({
+                where,
+                orderBy: { cityId: 'asc' },
+                take: AUDIT_RETENTION_BATCH_SIZE,
+                select: { cityId: true },
+            });
+            if (cities.length) {
+                const deleted = await tx.playAuditCity.deleteMany({
+                    where: { ...where, cityId: { in: cities.map((row) => row.cityId) } },
+                });
+                return { status: 'progress', deleted: deleted.count };
+            }
+            const nations = await tx.playAuditNation.findMany({
+                where,
+                orderBy: { nationId: 'asc' },
+                take: AUDIT_RETENTION_BATCH_SIZE,
+                select: { nationId: true },
+            });
+            if (nations.length) {
+                const deleted = await tx.playAuditNation.deleteMany({
+                    where: { ...where, nationId: { in: nations.map((row) => row.nationId) } },
+                });
+                return { status: 'progress', deleted: deleted.count };
+            }
+            // 모든 child가 빈 뒤 부모만 삭제한다. 거대한 FK cascade를 정리 수단으로 쓰지 않는다.
+            await tx.playAuditMonth.delete({ where: { id: sample.id } });
+            return { status: 'progress', deleted: 1 };
+        },
+        { maxWait: 1000, timeout: 5000 }
+    );
+};
@@ -0,0 +1,40 @@
+import type { AuditRetentionResult } from './retention.js';
+
+/** 진행할 자료가 있을 때만 계속 실행한다. 일반 게임 턴/페이지 요청에 정리를 결합하지 않는다. */
+export const startAuditRetentionWorker = (options: {
+    prune: () => Promise<AuditRetentionResult>;
+    onError: (error: unknown) => void;
+}): { stop: () => Promise<void> } => {
+    let stopped = false;
+    let timer: ReturnType<typeof setTimeout> | undefined;
+    let running: Promise<void> | undefined;
+    const schedule = (delay: number) => {
+        if (stopped) return;
+        timer = setTimeout(() => {
+            timer = undefined;
+            running = run();
+        }, delay);
+        timer.unref();
+    };
+    const run = async () => {
+        let delay = 1000;
+        try {
+            const result = await options.prune();
+            if (result.status === 'complete' || result.status === 'identityChanged') stopped = true;
+            if (result.status === 'busy') delay = 30_000;
+        } catch (error) {
+            delay = 30_000;
+            options.onError(error);
+        } finally {
+            schedule(delay);
+        }
+    };
+    schedule(0);
+    return {
+        stop: async () => {
+            stopped = true;
+            if (timer) clearTimeout(timer);
+            await running;
+        },
+    };
+};
@@ -1,3 +1,4 @@
+import { prunePreviousAuditBatch } from '../playAudit/retention.js';
 import { randomBytes } from 'node:crypto';

 import {
@@ -81,7 +82,9 @@ export interface ScenarioSeedOptions {

 export interface ScenarioSeedResult {
    seed: WorldSeedPayload;
-    warnings: ScenarioBootstrapWarning[];
+    warnings: Array<
+        ScenarioBootstrapWarning | { code: 'audit_retention_pending' | 'audit_retention_failed'; message: string }
+    >;
    applied: boolean;
 }

@@ -372,7 +375,7 @@ export const seedScenarioToDatabase = async (options: ScenarioSeedOptions): Prom
    }
    await connector.connect();
    try {
-        const result: ScenarioSeedResult = { seed, warnings, applied: true };
+        const result: ScenarioSeedResult = { seed, warnings: [...warnings], applied: true };
        const applied = await connector.prisma.$transaction(
            async (prisma) => {
                await prisma.$queryRawUnsafe(
@@ -765,6 +768,24 @@ export const seedScenarioToDatabase = async (options: ScenarioSeedOptions): Prom
            { maxWait: 10_000, timeout: 60_000 }
        );
        result.applied = applied;
+        if ((options.resetTables ?? true) && typeof worldMeta.serverId === 'string' && worldMeta.serverId.trim()) {
+            // RESERVED에는 daemon이 없을 수 있으므로 commit 뒤 작은 batch 하나를 시작한다.
+            // 정리 실패는 이미 확정된 seed를 되돌리지 않으며, 나머지는 runtime 시작 시 재시도한다.
+            try {
+                const cleanup = await prunePreviousAuditBatch(connector.prisma, worldMeta.serverId);
+                if (cleanup.status === 'progress' || cleanup.status === 'busy') {
+                    result.warnings.push({
+                        code: 'audit_retention_pending',
+                        message: '이전 플레이 감사 자료의 나머지는 서버 시작 후 정리합니다.',
+                    });
+                }
+            } catch {
+                result.warnings.push({
+                    code: 'audit_retention_failed',
+                    message: '이전 플레이 감사 자료 정리를 완료하지 못했습니다. 서버 시작 후 재시도합니다.',
+                });
+            }
+        }
        return result;
    } finally {
        await connector.disconnect();
@@ -1,3 +1,4 @@
+import { prunePreviousAuditBatch, type AuditRetentionResult } from '../playAudit/retention.js';
 import { persistAuditMonth } from '../playAudit/persistence.js';
 import { persistGeneralAccessScores, persistGeneralUpdates } from './generalBatchPersistence.js';
 import { areSeasonRecordsFinalized } from './seasonRecords.js';
@@ -78,6 +79,7 @@ export interface DatabaseTurnHooks {
    applyClockProjection(redis: ClockProjectionRedis, workerId: string): Promise<boolean>;
    synchronizeClockAuthority(): Promise<boolean>;
    prepareRealtimeRecovery(options?: { paused?: boolean }): Promise<void>;
+    prunePreviousAudit(expectedServerId: string): Promise<AuditRetentionResult>;
 }

 export interface CommittedReadModelChangeReceipt {
@@ -2150,6 +2152,7 @@ export const createDatabaseTurnHooks = async (
                await acquireGameSchemaAdvisoryXactLock(transaction, CLOCK_OPERATION_PERSISTENCE_LOCK);
                return synchronizeRuntimeClockAuthorityUnderHeldLock(transaction, world);
            }, transactionOptions),
+        prunePreviousAudit: (expectedServerId) => prunePreviousAuditBatch(prisma, expectedServerId),
        close: () => connector.disconnect(),
    };
 };
@@ -1,3 +1,4 @@
+import { startAuditRetentionWorker } from '../playAudit/retentionWorker.js';
 import { createPlayAuditHandler } from '../playAudit/collection.js';
 import { randomUUID } from 'node:crypto';
 import { createRuntimePauseGate } from './runtimePauseGate.js';
@@ -720,6 +721,7 @@ const createTurnDaemonRuntimeWithLease = async (
    let applyClockProjection: DatabaseTurnHooks['applyClockProjection'] | undefined;
    let synchronizeClockAuthority: DatabaseTurnHooks['synchronizeClockAuthority'] | undefined;
    let prepareClockRecovery: DatabaseTurnHooks['prepareRealtimeRecovery'] | undefined;
+    let prunePreviousAudit: DatabaseTurnHooks['prunePreviousAudit'] | undefined;
    const nationTraits = await loadNationTraitModules([...NATION_TRAIT_KEYS], new NationTraitLoader());
    const nationTraitMap = new Map(nationTraits.map((module) => [module.key, module]));
    const monthlyActionModules = await loadActionModuleBundle(
@@ -955,6 +957,7 @@ const createTurnDaemonRuntimeWithLease = async (
        applyClockProjection = dbHooks.applyClockProjection;
        synchronizeClockAuthority = dbHooks.synchronizeClockAuthority;
        prepareClockRecovery = dbHooks.prepareRealtimeRecovery;
+        prunePreviousAudit = dbHooks.prunePreviousAudit;
        close = async () => {
            if (auctionBidder) {
                await auctionBidder.close();
@@ -1110,6 +1113,22 @@ const createTurnDaemonRuntimeWithLease = async (
        controlQueue: resolvedControlQueue,
    });

+    const auditServerId = world.getState().meta.serverId;
+    const pruneAudit = prunePreviousAudit;
+    const auditRetention =
+        pruneAudit && typeof auditServerId === 'string' && auditServerId.trim()
+            ? startAuditRetentionWorker({
+                  prune: () =>
+                      turnDaemonLease?.isLost()
+                          ? Promise.resolve({ status: 'identityChanged', deleted: 0 })
+                          : pruneAudit(auditServerId),
+                  onError: () => {
+                      // 원문 DB 오류에는 연결 정보가 포함될 수 있어 고정된 운영 신호만 남긴다.
+                      console.warn('[play-audit] Previous-season cleanup failed; retrying in 30 seconds.');
+                  },
+              })
+            : null;
+
    return {
        lifecycle,
        world,
@@ -1119,7 +1138,10 @@ const createTurnDaemonRuntimeWithLease = async (
        processor,
        reservedTurns: reservedTurnStoreHandle?.store ?? null,
        hooks,
-        close,
+        close: async () => {
+            await auditRetention?.stop();
+            await close();
+        },
    };
 };

@@ -0,0 +1,187 @@
+import { seedScenarioToDatabase } from '../src/scenario/scenarioSeeder.js';
+import { afterAll, beforeAll, describe, expect, it } from 'vitest';
+import { createGamePostgresConnector, type GamePrismaClient } from '@sammo-ts/infra';
+import { AUDIT_RETENTION_BATCH_SIZE, prunePreviousAuditBatch } from '../src/playAudit/retention.js';
+
+const databaseUrl = process.env.PLAY_AUDIT_RETENTION_DATABASE_URL;
+const integration = describe.skipIf(!databaseUrl);
+integration('bounded previous-season audit retention', () => {
+    let db: GamePrismaClient;
+    let close: () => Promise<void>;
+    beforeAll(async () => {
+        if (!new URL(databaseUrl!).searchParams.get('schema')?.endsWith('_retention_fixture')) {
+            throw new Error('Audit retention requires its dedicated fixture schema');
+        }
+        const connector = createGamePostgresConnector({ url: databaseUrl! });
+        await connector.connect();
+        db = connector.prisma;
+        close = () => connector.disconnect();
+    });
+    afterAll(async () => {
+        await close?.();
+    });
+
+    it('keeps the active season, bounds each transaction and retries after rollback', async () => {
+        await db.playAuditMonth.deleteMany();
+        await db.worldState.deleteMany();
+        const world = await db.worldState.create({
+            data: {
+                scenarioCode: 'audit-retention',
+                currentYear: 190,
+                currentMonth: 1,
+                tickSeconds: 600,
+                config: {},
+                meta: { serverId: 'active' },
+            },
+        });
+        const sample = (id: string, serverId: string) => ({
+            id,
+            serverId,
+            year: 190,
+            month: 1,
+            kind: 'MONTH_END',
+            hash: id,
+            settlementsComplete: true,
+        });
+        await db.playAuditMonth.createMany({ data: [sample('old-sample', 'old'), sample('active-sample', 'active')] });
+        await db.playAuditGeneral.createMany({
+            data: Array.from({ length: 401 }, (_, index) => ({
+                sampleId: 'old-sample',
+                generalId: index + 1,
+                nationId: 1,
+                cityId: 1,
+                npcState: 2,
+                data: { marker: 'old' },
+            })),
+        });
+        await db.playAuditCity.createMany({
+            data: Array.from({ length: 201 }, (_, index) => ({
+                sampleId: 'old-sample',
+                cityId: index + 1,
+                nationId: 1,
+                data: {},
+            })),
+        });
+        await db.playAuditNation.create({ data: { sampleId: 'old-sample', nationId: 1, data: {} } });
+        await db.playAuditGeneral.create({
+            data: {
+                sampleId: 'active-sample',
+                generalId: 1,
+                nationId: 1,
+                cityId: 1,
+                npcState: 2,
+                data: { marker: 'preserved' },
+            },
+        });
+        expect(await prunePreviousAuditBatch(db, 'wrong')).toEqual({ status: 'identityChanged', deleted: 0 });
+        expect(await prunePreviousAuditBatch(db, '')).toEqual({ status: 'identityChanged', deleted: 0 });
+        expect(await prunePreviousAuditBatch(db, 'active')).toEqual({
+            status: 'progress',
+            deleted: AUDIT_RETENTION_BATCH_SIZE,
+        });
+        expect(await db.playAuditGeneral.count({ where: { sampleId: 'old-sample' } })).toBe(201);
+        await db.$executeRawUnsafe(
+            `CREATE OR REPLACE FUNCTION audit_retention_test_failure() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN RAISE EXCEPTION 'retention fixture rollback'; END $$`
+        );
+        await db.$executeRawUnsafe(
+            `CREATE TRIGGER audit_retention_test_failure BEFORE DELETE ON play_audit_general FOR EACH ROW EXECUTE FUNCTION audit_retention_test_failure()`
+        );
+        try {
+            await expect(prunePreviousAuditBatch(db, 'active')).rejects.toThrow();
+            expect(await db.playAuditGeneral.count({ where: { sampleId: 'old-sample' } })).toBe(201);
+        } finally {
+            await db.$executeRawUnsafe('DROP TRIGGER audit_retention_test_failure ON play_audit_general');
+            await db.$executeRawUnsafe('DROP FUNCTION audit_retention_test_failure()');
+        }
+        // A RESET holder makes cleanup defer without waiting for the reset transaction.
+        await db.$transaction(async (tx) => {
+            await tx.$queryRaw`SELECT pg_advisory_xact_lock(hashtextextended(current_schema(), 0))::text`;
+            expect(await prunePreviousAuditBatch(db, 'active')).toEqual({ status: 'busy', deleted: 0 });
+        });
+        const deleted = [];
+        for (let attempt = 0; attempt < 10; attempt++) {
+            const result = await prunePreviousAuditBatch(db, 'active');
+            if (result.status === 'complete') break;
+            expect(result.status).toBe('progress');
+            expect(result.deleted).toBeLessThanOrEqual(AUDIT_RETENTION_BATCH_SIZE);
+            deleted.push(result.deleted);
+        }
+        expect(deleted).toEqual([200, 1, 200, 1, 1, 1]);
+        expect(await db.playAuditMonth.findUnique({ where: { id: 'old-sample' } })).toBeNull();
+        expect(
+            await db.playAuditGeneral.findUnique({
+                where: { sampleId_generalId: { sampleId: 'active-sample', generalId: 1 } },
+            })
+        ).toMatchObject({ data: { marker: 'preserved' } });
+        expect(await prunePreviousAuditBatch(db, 'active')).toEqual({ status: 'complete', deleted: 0 });
+        // A worker from the old runtime must stop after the world identity changes.
+        await db.worldState.update({ where: { id: world.id }, data: { meta: { serverId: 'next' } } });
+        expect(await prunePreviousAuditBatch(db, 'active')).toEqual({ status: 'identityChanged', deleted: 0 });
+        expect(await db.playAuditMonth.count()).toBe(1);
+    });
+    it('starts bounded cleanup only after seed commits, including a reserved opening', async () => {
+        await db.playAuditMonth.deleteMany();
+        await db.worldState.deleteMany();
+        await db.worldState.create({
+            data: {
+                scenarioCode: 'before-reset',
+                currentYear: 190,
+                currentMonth: 1,
+                tickSeconds: 600,
+                config: {},
+                meta: { serverId: 'before-reset' },
+            },
+        });
+        await db.playAuditMonth.create({
+            data: {
+                id: 'reset-old-sample',
+                serverId: 'before-reset',
+                year: 190,
+                month: 1,
+                kind: 'MONTH_END',
+                hash: 'reset',
+                settlementsComplete: true,
+            },
+        });
+        await db.playAuditGeneral.createMany({
+            data: Array.from({ length: 201 }, (_, index) => ({
+                sampleId: 'reset-old-sample',
+                generalId: index + 1,
+                nationId: 1,
+                cityId: 1,
+                npcState: 2,
+                data: {},
+            })),
+        });
+        const options = {
+            scenarioId: 1010,
+            databaseUrl: databaseUrl!,
+            resetTables: true,
+            now: new Date('2030-01-01T00:00:00Z'),
+            wallNow: new Date('2030-01-01T00:00:00Z'),
+            installOptions: {
+                serverId: 'after-reset',
+                preopenAt: new Date('2030-01-03T00:00:00Z'),
+                openAt: new Date('2030-01-04T00:00:00Z'),
+            },
+        };
+        await expect(
+            seedScenarioToDatabase({
+                ...options,
+                onBeforeCommit: async () => {
+                    throw new Error('seed fixture rollback');
+                },
+            })
+        ).rejects.toThrow('seed fixture rollback');
+        expect(await db.playAuditGeneral.count({ where: { sampleId: 'reset-old-sample' } })).toBe(201);
+        expect(await db.worldState.findFirst()).toMatchObject({ meta: { serverId: 'before-reset' } });
+        const result = await seedScenarioToDatabase(options);
+        expect(result.applied).toBe(true);
+        expect(await db.worldState.findFirst()).toMatchObject({ meta: { serverId: 'after-reset' } });
+        expect(await db.playAuditGeneral.count({ where: { sampleId: 'reset-old-sample' } })).toBe(1);
+        expect(result.warnings).toContainEqual({
+            code: 'audit_retention_pending',
+            message: '이전 플레이 감사 자료의 나머지는 서버 시작 후 정리합니다.',
+        });
+    });
+});
@@ -0,0 +1,61 @@
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import { startAuditRetentionWorker } from '../src/playAudit/retentionWorker.js';
+import type { AuditRetentionResult } from '../src/playAudit/retention.js';
+
+afterEach(() => vi.useRealTimers());
+describe('audit retention scheduling', () => {
+    it('backs off errors and lock contention, then stops polling when no previous season remains', async () => {
+        vi.useFakeTimers();
+        const prune = vi
+            .fn<() => Promise<AuditRetentionResult>>()
+            .mockRejectedValueOnce(new Error('database failure'))
+            .mockResolvedValueOnce({ status: 'busy', deleted: 0 })
+            .mockResolvedValueOnce({ status: 'progress', deleted: 200 })
+            .mockResolvedValue({ status: 'complete', deleted: 0 });
+        const onError = vi.fn();
+        const worker = startAuditRetentionWorker({ prune, onError });
+        await vi.advanceTimersByTimeAsync(0);
+        expect(onError).toHaveBeenCalledOnce();
+        await vi.advanceTimersByTimeAsync(29_999);
+        expect(prune).toHaveBeenCalledTimes(1);
+        await vi.advanceTimersByTimeAsync(30_001);
+        expect(prune).toHaveBeenCalledTimes(3);
+        await vi.advanceTimersByTimeAsync(1000);
+        expect(prune).toHaveBeenCalledTimes(4);
+        await vi.advanceTimersByTimeAsync(600_000);
+        expect(prune).toHaveBeenCalledTimes(4);
+        await worker.stop();
+    });
+    it('never overlaps batches and waits for the in-flight transaction when closing', async () => {
+        vi.useFakeTimers();
+        let finish!: (result: AuditRetentionResult) => void;
+        const prune = vi.fn(
+            () =>
+                new Promise<AuditRetentionResult>((resolve) => {
+                    finish = resolve;
+                })
+        );
+        const worker = startAuditRetentionWorker({ prune, onError: vi.fn() });
+        await vi.advanceTimersByTimeAsync(0);
+        await vi.advanceTimersByTimeAsync(60_000);
+        expect(prune).toHaveBeenCalledOnce();
+        let closed = false;
+        const stop = worker.stop().then(() => {
+            closed = true;
+        });
+        await Promise.resolve();
+        expect(closed).toBe(false);
+        finish({ status: 'progress', deleted: 200 });
+        await stop;
+        await vi.advanceTimersByTimeAsync(60_000);
+        expect(prune).toHaveBeenCalledOnce();
+    });
+    it('does not use an old runtime identity after reset', async () => {
+        vi.useFakeTimers();
+        const prune = vi.fn(async (): Promise<AuditRetentionResult> => ({ status: 'identityChanged', deleted: 0 }));
+        const worker = startAuditRetentionWorker({ prune, onError: vi.fn() });
+        await vi.advanceTimersByTimeAsync(600_000);
+        expect(prune).toHaveBeenCalledOnce();
+        await worker.stop();
+    });
+});