감사 상세를 무손실 압축하고 이전 기수 정리 처리량을 개선한다
This commit is contained in:
Hide_D committed
1 parent
da79013226
commit
078b9aa205
23 files changed
+504
-95
No files matched your search
@@ -76,7 +76,7 @@ export class ReadModelOutboxWorker implements ReadModelOutboxWakeup {
|
||||
this.leaseMs = normalizePositiveInteger(options.leaseMs, 30_000);
|
||||
this.retentionMs = normalizePositiveInteger(options.retentionMs, 24 * 60 * 60 * 1_000);
|
||||
this.pruneIntervalMs = normalizePositiveInteger(options.pruneIntervalMs, 60_000);
|
||||
this.pruneLimit = normalizePositiveInteger(options.pruneLimit, 100);
|
||||
this.pruneLimit = Math.min(500, normalizePositiveInteger(options.pruneLimit, 500));
|
||||
this.owner = options.owner ?? `game-api:${profileName}:${process.pid}:${randomUUID()}`;
|
||||
this.now = options.now ?? (() => new Date());
|
||||
this.onError = options.onError ?? (() => undefined);
|
||||
@@ -109,11 +109,17 @@ export class ReadModelOutboxWorker implements ReadModelOutboxWakeup {
|
||||
|
||||
const now = this.now();
|
||||
if (now.getTime() >= this.nextPruneAt) {
|
||||
this.nextPruneAt = now.getTime() + this.pruneIntervalMs;
|
||||
await pruneDeliveredReadModelOutbox(this.db, {
|
||||
deliveredBefore: new Date(now.getTime() - this.retentionMs),
|
||||
limit: this.pruneLimit,
|
||||
});
|
||||
// backlog가 있으면 다음 dispatch에서도 이어서 정리한다. 평상시에는 낮은 빈도로 확인한다.
|
||||
// 한 호출은 최대 10 batch/50ms이며 각 SQL도 최대 500행이다.
|
||||
const started = performance.now();
|
||||
for (let batch = 0; batch < 10; batch += 1) {
|
||||
const deleted = await pruneDeliveredReadModelOutbox(this.db, {
|
||||
deliveredBefore: new Date(now.getTime() - this.retentionMs),
|
||||
limit: this.pruneLimit,
|
||||
});
|
||||
this.nextPruneAt = deleted < this.pruneLimit ? now.getTime() + this.pruneIntervalMs : 0;
|
||||
if (deleted < this.pruneLimit || performance.now() - started >= 50) break;
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { TRPCError } from '@trpc/server';
|
||||
import { z } from 'zod';
|
||||
import type { GamePrisma } from '@sammo-ts/infra';
|
||||
import { decodeAuditDecisionSteps, type GamePrisma } from '@sammo-ts/infra';
|
||||
import { auditProcedure, monthOrdinal, readAudit, readAuditWorld, zAuditMonth } from './shared.js';
|
||||
|
||||
const zEffectivePolicy = z.object({
|
||||
@@ -275,7 +275,7 @@ export const decisionDetail = auditProcedure
|
||||
decision: project(row),
|
||||
chunks: chunks.slice(0, input.limit).map((chunk) => ({
|
||||
ordinal: chunk.ordinal,
|
||||
steps: z.array(zStep).max(128).parse(chunk.steps),
|
||||
steps: z.array(zStep).max(128).parse(decodeAuditDecisionSteps(chunk.steps)),
|
||||
})),
|
||||
nextCursor:
|
||||
chunks.length && chunks.at(-1)!.ordinal + 1 < Math.ceil(row.stepCount / 128)
|
||||
|
||||
@@ -2,7 +2,12 @@ import { afterAll, beforeAll, expect, it } from 'vitest';
|
||||
import fastify from 'fastify';
|
||||
import { fastifyTRPCPlugin } from '@trpc/server/adapters/fastify';
|
||||
import { asRecord } from '@sammo-ts/common';
|
||||
import { createGamePostgresConnector, createRedisConnector, GamePrisma } from '@sammo-ts/infra';
|
||||
import {
|
||||
decodeAuditDecisionSteps,
|
||||
createGamePostgresConnector,
|
||||
createRedisConnector,
|
||||
GamePrisma,
|
||||
} from '@sammo-ts/infra';
|
||||
import { createTurnDaemonRuntime, seedScenarioToDatabase, type TurnDaemonRuntime } from '@sammo-ts/game-engine';
|
||||
import { appRouter } from '../src/router.js';
|
||||
import { DatabaseTurnDaemonTransport } from '../src/daemon/databaseTransport.js';
|
||||
@@ -255,12 +260,10 @@ integration(
|
||||
include: { chunks: { orderBy: { ordinal: 'asc' } } },
|
||||
});
|
||||
const procedures = nationDecision.chunks.flatMap((chunk) =>
|
||||
Array.isArray(chunk.steps)
|
||||
? chunk.steps
|
||||
.map(asRecord)
|
||||
.filter((step) => step.kind === 'PROCEDURE_START')
|
||||
.map((step) => step.procedure)
|
||||
: []
|
||||
decodeAuditDecisionSteps(chunk.steps)
|
||||
.map(asRecord)
|
||||
.filter((step) => step.kind === 'PROCEDURE_START')
|
||||
.map((step) => step.procedure)
|
||||
);
|
||||
expect(procedures).toEqual(['NPC포상']);
|
||||
expect(configuration(await call('getPolicy', {}))).toEqual(configuration(withoutRecruitment));
|
||||
|
||||
@@ -45,6 +45,28 @@ const createFixture = (rows: readonly object[]) => {
|
||||
};
|
||||
|
||||
describe('ReadModelOutboxWorker', () => {
|
||||
it('drains more than one full retention batch rather than leaving a permanent 100-per-minute backlog', async () => {
|
||||
let now = new Date('2026-11-09T00:00:00Z');
|
||||
const fixture = createFixture([]);
|
||||
fixture.queryRaw
|
||||
.mockReset()
|
||||
.mockResolvedValueOnce([])
|
||||
.mockResolvedValueOnce([{ id: 1n }, { id: 2n }])
|
||||
.mockResolvedValueOnce([{ id: 3n }, { id: 4n }])
|
||||
.mockResolvedValueOnce([]);
|
||||
const worker = new ReadModelOutboxWorker(fixture.db, fixture.redis, 'che:default', {
|
||||
intervalMs: 60000,
|
||||
pruneLimit: 2,
|
||||
now: () => now,
|
||||
});
|
||||
now = new Date('2026-11-09T00:01:00Z');
|
||||
worker.start();
|
||||
await vi.waitFor(() => expect(fixture.queryRaw).toHaveBeenCalledTimes(4));
|
||||
await worker.stop();
|
||||
const query = fixture.queryRaw.mock.calls[1]?.[0] as { sql: string };
|
||||
expect(query.sql).toContain('FOR UPDATE SKIP LOCKED');
|
||||
expect(query.sql).toContain('ORDER BY "delivered_at", "available_at", "id"');
|
||||
});
|
||||
it('publishes a survey-style global front-status invalidation without a turn daemon', async () => {
|
||||
const fixture = createFixture([{ id: 10n, payload: payload('front.global'), attempts: 1 }]);
|
||||
const worker = new ReadModelOutboxWorker(fixture.db, fixture.redis, 'che:default', {
|
||||
|
||||
@@ -13,6 +13,7 @@ import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest';
|
||||
import { encryptGameSessionToken, type GameSessionTokenPayload } from '@sammo-ts/common/auth/gameToken';
|
||||
import {
|
||||
createGamePostgresConnector,
|
||||
encodeAuditDecisionSteps,
|
||||
GamePrisma,
|
||||
hashAuditDiplomacyDocument,
|
||||
createRedisConnector,
|
||||
@@ -2345,50 +2346,52 @@ integration('game API security over HTTP transport', () => {
|
||||
{
|
||||
decisionId: decisionIds[0]!,
|
||||
ordinal: 0,
|
||||
steps: Array.from({ length: 128 }, (_, sequence) =>
|
||||
sequence === 127
|
||||
? {
|
||||
...step,
|
||||
sequence,
|
||||
kind: 'DECISION_START',
|
||||
reservedAction: '휴식',
|
||||
effectivePolicy: {
|
||||
schemaVersion: 1,
|
||||
general: { priority: ['징병'], flags: { 징병: true, 출병: false } },
|
||||
nation: {
|
||||
priority: ['천도'],
|
||||
flags: { 천도: true },
|
||||
values: {
|
||||
reqNationGold: 4321,
|
||||
reqNationRice: 100,
|
||||
reqHumanWarUrgentGold: 100,
|
||||
reqHumanWarUrgentRice: 100,
|
||||
reqHumanWarRecommandGold: 100,
|
||||
reqHumanWarRecommandRice: 100,
|
||||
reqHumanDevelGold: 100,
|
||||
reqHumanDevelRice: 100,
|
||||
reqNpcWarGold: 100,
|
||||
reqNpcWarRice: 100,
|
||||
reqNpcDevelGold: 100,
|
||||
reqNpcDevelRice: 100,
|
||||
minimumResourceActionAmount: 100,
|
||||
maximumResourceActionAmount: 100,
|
||||
minNpcWarLeadership: 100,
|
||||
minWarCrew: 100,
|
||||
minNpcRecruitCityPopulation: 100,
|
||||
safeRecruitCityPopulationRatio: 100,
|
||||
properWarTrainAtmos: 100,
|
||||
cureThreshold: 100,
|
||||
steps: encodeAuditDecisionSteps(
|
||||
Array.from({ length: 128 }, (_, sequence) =>
|
||||
sequence === 127
|
||||
? {
|
||||
...step,
|
||||
sequence,
|
||||
kind: 'DECISION_START',
|
||||
reservedAction: '휴식',
|
||||
effectivePolicy: {
|
||||
schemaVersion: 1,
|
||||
general: { priority: ['징병'], flags: { 징병: true, 출병: false } },
|
||||
nation: {
|
||||
priority: ['천도'],
|
||||
flags: { 천도: true },
|
||||
values: {
|
||||
reqNationGold: 4321,
|
||||
reqNationRice: 100,
|
||||
reqHumanWarUrgentGold: 100,
|
||||
reqHumanWarUrgentRice: 100,
|
||||
reqHumanWarRecommandGold: 100,
|
||||
reqHumanWarRecommandRice: 100,
|
||||
reqHumanDevelGold: 100,
|
||||
reqHumanDevelRice: 100,
|
||||
reqNpcWarGold: 100,
|
||||
reqNpcWarRice: 100,
|
||||
reqNpcDevelGold: 100,
|
||||
reqNpcDevelRice: 100,
|
||||
minimumResourceActionAmount: 100,
|
||||
maximumResourceActionAmount: 100,
|
||||
minNpcWarLeadership: 100,
|
||||
minWarCrew: 100,
|
||||
minNpcRecruitCityPopulation: 100,
|
||||
safeRecruitCityPopulationRatio: 100,
|
||||
properWarTrainAtmos: 100,
|
||||
cureThreshold: 100,
|
||||
},
|
||||
combatForce: { '1': [2, 3] },
|
||||
supportForce: [4],
|
||||
developForce: [5],
|
||||
secret: 'decision-secret',
|
||||
},
|
||||
combatForce: { '1': [2, 3] },
|
||||
supportForce: [4],
|
||||
developForce: [5],
|
||||
secret: 'decision-secret',
|
||||
},
|
||||
secret: 'decision-secret',
|
||||
},
|
||||
}
|
||||
: { ...step, sequence }
|
||||
}
|
||||
: { ...step, sequence }
|
||||
)
|
||||
),
|
||||
},
|
||||
{
|
||||
|
||||
Reference in new issue
Block a user