고빈도 저장 ID를 64비트로 확장하고 랭킹 갱신의 시퀀스 낭비를 막는다

This commit is contained in:
Hide_D committed 2026-11-14 15:34:21 +09:00
1 parent 087517821e
commit bc54b157c1
25 files changed
+518 -86

No files matched your search

+18 -7
View File
@@ -119,7 +119,7 @@ export interface RealtimeReadModelBaseline {
}
export type PersistedVisibleLogRow = {
id: number;
id: number | bigint;
scope: LogScope;
category: LogCategory;
generalId: number | null;
@@ -831,17 +831,28 @@ const buildPersistedGeneralMeta = (
const RANK_DATA_UPSERT_BATCH_SIZE = 1_000;
const upsertRankRows = async (
export const upsertRankRows = async (
prisma: GamePrisma.TransactionClient,
rows: Array<{ generalId: number; nationId: number; type: string; value: number }>
): Promise<void> => {
for (let offset = 0; offset < rows.length; offset += RANK_DATA_UPSERT_BATCH_SIZE) {
const batch = rows.slice(offset, offset + RANK_DATA_UPSERT_BATCH_SIZE);
// Ref와 같은 (장수, 항목) 값을 저장한다. 이미 있는 행의 ID를 재사용해
// 매 턴 44종 갱신마다 nextval을 소비하지 않고, 신규 행만 시퀀스를 사용한다.
// 동시 신규 insert 경합은 기존 ON CONFLICT 계약으로 처리한다.
await prisma.$executeRaw(GamePrisma.sql`
INSERT INTO "rank_data" ("general_id", "nation_id", "type", "value")
VALUES ${GamePrisma.join(
batch.map((row) => GamePrisma.sql`(${row.generalId}, ${row.nationId}, ${row.type}, ${row.value})`)
)}
INSERT INTO "rank_data" ("id", "general_id", "nation_id", "type", "value")
SELECT
COALESCE(existing."id", nextval(pg_get_serial_sequence('rank_data', 'id'))),
incoming."general_id", incoming."nation_id", incoming."type", incoming."value"
FROM (VALUES ${GamePrisma.join(
batch.map(
(row) =>
GamePrisma.sql`(${row.generalId}::integer, ${row.nationId}::integer, ${row.type}::text, ${row.value}::integer)`
)
)}) AS incoming("general_id", "nation_id", "type", "value")
LEFT JOIN "rank_data" AS existing
ON existing."general_id" = incoming."general_id" AND existing."type" = incoming."type"
ON CONFLICT ("general_id", "type") DO UPDATE
SET
"nation_id" = EXCLUDED."nation_id",
@@ -1129,7 +1140,7 @@ export const createDatabaseTurnHooks = async (
const persistChanges = async (
transaction?: GamePrisma.TransactionClient,
commandCompletion?: { requestId: string; result: TurnDaemonCommandResult },
directLogFloor?: number
directLogFloor?: number | bigint
): Promise<{
acknowledge: () => void;
readModelChanges: RealtimeReadModelChanges;
@@ -0,0 +1,145 @@
import { afterAll, beforeAll, describe, expect, it } from 'vitest';
import { createGamePostgresConnector, type GamePrismaClient } from '@sammo-ts/infra';
import { upsertRankRows } from '../src/turn/databaseHooks.js';
const databaseUrl = process.env.IMMEDIATE_ACTION_DATABASE_URL;
// The conditional runner owns this schema and runs integration files serially.
describe.skipIf(!databaseUrl)('rank data bigint PostgreSQL persistence', () => {
let db: GamePrismaClient;
let disconnect: () => Promise<void>;
const generalId = 7_319;
beforeAll(async () => {
const schema = new URL(databaseUrl!).searchParams.get('schema');
if (!schema?.endsWith('immediate_action_integration')) {
throw new Error('Refusing to mutate a non-dedicated rank persistence schema');
}
const connector = createGamePostgresConnector({ url: databaseUrl! });
await connector.connect();
db = connector.prisma;
disconnect = () => connector.disconnect();
await db.rankData.deleteMany({ where: { generalId } });
// Even conflict-update on an existing row used to exhaust int4 IDs.
await db.$executeRaw`SELECT setval(pg_get_serial_sequence('rank_data', 'id'), 2147483647)`;
});
afterAll(async () => {
if (db) await db.rankData.deleteMany({ where: { generalId } });
await disconnect?.();
});
it('reuses IDs across batches, updates ownership and values, and allocates bigint IDs only for new rows', async () => {
const rows = Array.from({ length: 1_001 }, (_, index) => ({
generalId,
nationId: 10,
type: `rank_fixture_${index}`,
value: index,
}));
await db.$transaction((tx) => upsertRankRows(tx, rows));
const original = await db.rankData.findMany({ where: { generalId }, orderBy: { type: 'asc' } });
expect(original).toHaveLength(rows.length);
expect(original.every((row) => row.id > 2_147_483_647n)).toBe(true);
const sequenceBefore = await db.$queryRaw<
Array<{ last_value: bigint }>
>`SELECT last_value FROM rank_data_id_seq`;
for (let count = 0; count < 3; count++) {
await db.$transaction((tx) =>
upsertRankRows(
tx,
rows.map((row) => ({ ...row, nationId: 20, value: row.value + 3 }))
)
);
}
const updated = await db.rankData.findMany({ where: { generalId }, orderBy: { type: 'asc' } });
expect(updated.map((row) => row.id)).toEqual(original.map((row) => row.id));
expect(updated.map((row) => [row.nationId, row.value])).toEqual(original.map((row) => [20, row.value + 3]));
expect(await db.$queryRaw`SELECT last_value FROM rank_data_id_seq`).toEqual(sequenceBefore);
await db.$transaction((tx) =>
upsertRankRows(tx, [rows[0]!, { generalId, nationId: 30, type: 'new_rank', value: 17 }])
);
expect(await db.rankData.count({ where: { generalId } })).toBe(1_002);
expect(await db.$queryRaw`SELECT last_value FROM rank_data_id_seq`).toEqual([
{ last_value: sequenceBefore[0]!.last_value + 1n },
]);
});
it('reloads bigint IDs for high churn queues, logs, upserts and the traffic foreign key', async () => {
const tables = [
'general_turn',
'nation_turn',
'general_access_log',
'diplomacy',
'inheritance_point',
'log_entry',
'error_log',
'inheritance_log',
'traffic_period',
];
for (const table of tables) {
await db.$executeRaw`SELECT setval(pg_get_serial_sequence(${table}, 'id'), 2147483647)`;
}
await expect(
db.$transaction(async (tx) => {
const now = new Date();
const world = await tx.worldState.create({
data: {
scenarioCode: 'bigint-fixture',
currentYear: 200,
currentMonth: 1,
tickSeconds: 600,
},
});
const rows = await Promise.all([
tx.generalTurn.create({ data: { generalId, turnIdx: 0, actionCode: 'fixture' } }),
tx.nationTurn.create({
data: { nationId: 3, officerLevel: 12, turnIdx: 0, actionCode: 'fixture' },
}),
tx.generalAccessLog.create({ data: { generalId } }),
tx.diplomacy.create({ data: { srcNationId: 3, destNationId: 4, stateCode: 0 } }),
tx.inheritancePoint.create({ data: { userId: 'bigint-fixture', key: 'previous', value: 123 } }),
tx.logEntry.create({
data: { scope: 'SYSTEM', category: 'HISTORY', year: 200, month: 1, text: 'fixture' },
}),
tx.errorLog.create({ data: { category: 'fixture', message: 'fixture' } }),
tx.inheritanceLog.create({
data: { userId: 'bigint-fixture', year: 200, month: 1, text: 'fixture' },
}),
]);
expect(rows.every((row) => row.id > 2_147_483_647n)).toBe(true);
const period = await tx.trafficPeriod.create({
data: {
worldStateId: world.id,
year: 200,
month: 1,
startedAt: now,
lastRefresh: now,
generals: { create: { generalId, lastRefresh: now } },
},
include: { generals: true },
});
expect(period.id).toBeGreaterThan(2_147_483_647n);
expect(period.generals[0]?.periodId).toBe(period.id);
expect((await tx.logEntry.findUniqueOrThrow({ where: { id: rows[5]!.id } })).text).toBe('fixture');
throw new Error('high churn fixture rollback');
})
).rejects.toThrow('high churn fixture rollback');
});
it('rolls value updates back atomically and retains the logical unique constraint', async () => {
const where = { generalId_type: { generalId, type: 'new_rank' } };
const before = await db.rankData.findUniqueOrThrow({ where });
await expect(
db.$transaction(async (tx) => {
await upsertRankRows(tx, [{ generalId, nationId: 99, type: 'new_rank', value: 999 }]);
throw new Error('fixture rollback');
})
).rejects.toThrow('fixture rollback');
expect(await db.rankData.findUniqueOrThrow({ where })).toEqual(before);
await expect(
db.rankData.create({ data: { generalId, nationId: 99, type: 'new_rank', value: 999 } })
).rejects.toThrow();
});
});