From 5040691d7ca62d8e3c7bd97c509a6c51cf95298b Mon Sep 17 00:00:00 2001 From: hided62 Date: Sat, 25 Jul 2026 05:36:34 +0000 Subject: [PATCH] feat: implement input event system with durable command handling - Introduced InputEvent model with status tracking (PENDING, PROCESSING, SUCCEEDED, FAILED) and unique request IDs. - Added DatabaseTurnDaemonTransport for sending commands and handling idempotency. - Implemented executeInputEvent function to manage input event lifecycle and error handling. - Created DatabaseTurnDaemonCommandQueue for managing command processing and lease recovery. - Enhanced turn daemon lifecycle to support atomic command execution and error recovery. - Added tests for input event atomicity, command queuing, and error handling scenarios. --- README.md | 8 + app/game-api/src/context.ts | 3 + app/game-api/src/daemon/databaseTransport.ts | 95 ++++++ .../src/daemon/idempotentTransport.ts | 28 ++ app/game-api/src/daemon/inMemoryTransport.ts | 2 +- app/game-api/src/daemon/redisTransport.ts | 2 +- app/game-api/src/index.ts | 11 +- app/game-api/src/inputEventBoundary.ts | 86 ++++++ app/game-api/src/server.ts | 17 +- app/game-api/src/trpc.ts | 44 ++- app/game-api/test/idempotentTransport.test.ts | 22 ++ .../inputEventBoundary.integration.test.ts | 147 +++++++++ app/game-engine/src/auction/bidder.ts | 79 +++-- app/game-engine/src/auction/finalizer.ts | 69 ++--- app/game-engine/src/index.ts | 1 + .../src/lifecycle/databaseCommandQueue.ts | 164 ++++++++++ .../src/lifecycle/turnDaemonLifecycle.ts | 148 ++++----- app/game-engine/src/lifecycle/types.ts | 15 +- app/game-engine/src/tournament/finalizer.ts | 44 +-- app/game-engine/src/turn/commandRegistry.ts | 57 ++-- app/game-engine/src/turn/databaseHooks.ts | 110 +++++-- app/game-engine/src/turn/inMemoryWorld.ts | 78 +++-- app/game-engine/src/turn/reservedTurnStore.ts | 44 ++- app/game-engine/src/turn/turnDaemon.ts | 25 +- .../src/turn/worldCommandHandler.ts | 165 +++++----- .../databaseCommandQueue.integration.test.ts | 95 ++++++ .../test/inputEventAtomicity.test.ts | 285 ++++++++++++++++++ docs/architecture/runtime.md | 64 ++-- docs/architecture/todo.md | 16 +- docs/architecture/turn-daemon-lifecycle.md | 39 ++- packages/common/src/turnDaemon/types.ts | 13 +- packages/infra/prisma/game.prisma | 33 ++ .../migration.sql | 25 ++ packages/infra/src/db.ts | 1 + 34 files changed, 1587 insertions(+), 448 deletions(-) create mode 100644 app/game-api/src/daemon/databaseTransport.ts create mode 100644 app/game-api/src/daemon/idempotentTransport.ts create mode 100644 app/game-api/src/inputEventBoundary.ts create mode 100644 app/game-api/test/idempotentTransport.test.ts create mode 100644 app/game-api/test/inputEventBoundary.integration.test.ts create mode 100644 app/game-engine/src/lifecycle/databaseCommandQueue.ts create mode 100644 app/game-engine/test/databaseCommandQueue.integration.test.ts create mode 100644 app/game-engine/test/inputEventAtomicity.test.ts create mode 100644 packages/infra/prisma/migrations/20260725000000_add_input_event/migration.sql diff --git a/README.md b/README.md index 700a532..082095d 100644 --- a/README.md +++ b/README.md @@ -25,6 +25,14 @@ ## 개발 도구 및 스크립트 - 패키지 매니저: pnpm 워크스페이스 +- 현재 개발 호스트는 `fnm`으로 Node.js와 pnpm을 관리한다. 대화형 zsh가 + 아닌 환경에서 `pnpm`을 찾지 못하면 아래 형태로 실행한다. + + ```sh + /home/letrhee/.local/share/fnm/fnm exec --using=default -- pnpm + ``` + + 2026-07-25 확인 기준 default는 Node.js `v24.18.0`, pnpm은 `11.17.0`이다. - 초기 개발기간 동안 npm 패키지는 가능한 최신버전을 유지 - 공통 스크립트 - `pnpm install` diff --git a/app/game-api/src/context.ts b/app/game-api/src/context.ts index 208f88c..cab4ad9 100644 --- a/app/game-api/src/context.ts +++ b/app/game-api/src/context.ts @@ -61,6 +61,7 @@ export type InputJsonValue = GamePrisma.InputJsonValue; export type DatabaseClient = InfraDatabaseClient; export interface GameApiContext { + requestId?: string; db: DatabaseClient; redis: RedisConnector['client']; turnDaemon: TurnDaemonTransport; @@ -76,6 +77,7 @@ export interface GameApiContext { } export const createGameApiContext = (options: { + requestId?: string; db: DatabaseClient; redis: RedisConnector['client']; turnDaemon: TurnDaemonTransport; @@ -90,6 +92,7 @@ export const createGameApiContext = (options: { gameTokenSecret: string; }): GameApiContext => { return { + requestId: options.requestId, db: options.db, redis: options.redis, turnDaemon: options.turnDaemon, diff --git a/app/game-api/src/daemon/databaseTransport.ts b/app/game-api/src/daemon/databaseTransport.ts new file mode 100644 index 0000000..8aeef74 --- /dev/null +++ b/app/game-api/src/daemon/databaseTransport.ts @@ -0,0 +1,95 @@ +import { randomUUID } from 'node:crypto'; + +import type { GamePrisma } from '@sammo-ts/infra'; + +import type { DatabaseClient } from '../context.js'; +import type { TurnDaemonTransport } from './transport.js'; +import type { TurnDaemonCommand, TurnDaemonCommandResult, TurnDaemonStatus } from './types.js'; + +const asJson = (value: unknown): GamePrisma.InputJsonValue => value as GamePrisma.InputJsonValue; + +const delay = (ms: number): Promise => new Promise((resolve) => setTimeout(resolve, ms)); +const stableJson = (value: unknown): string => { + if (Array.isArray(value)) { + return `[${value.map(stableJson).join(',')}]`; + } + if (value && typeof value === 'object') { + return `{${Object.entries(value) + .sort(([left], [right]) => left.localeCompare(right)) + .map(([key, entry]) => `${JSON.stringify(key)}:${stableJson(entry)}`) + .join(',')}}`; + } + return JSON.stringify(value) ?? 'null'; +}; + +export class ConflictingTurnDaemonCommandError extends Error { + constructor(readonly requestId: string) { + super(`Engine input event ${requestId} already exists with a different payload.`); + this.name = 'ConflictingTurnDaemonCommandError'; + } +} + +export class DatabaseTurnDaemonTransport implements TurnDaemonTransport { + constructor( + private readonly db: DatabaseClient, + private readonly requestTimeoutMs: number + ) {} + + async sendCommand(command: TurnDaemonCommand): Promise { + const requestId = ('requestId' in command ? command.requestId : undefined) ?? randomUUID(); + const durableCommand = JSON.parse(JSON.stringify({ ...command, requestId })) as TurnDaemonCommand; + try { + await this.db.inputEvent.create({ + data: { + requestId, + target: 'ENGINE', + eventType: command.type, + payload: asJson(durableCommand), + }, + }); + } catch (error) { + const isUniqueConflict = + typeof error === 'object' && error !== null && 'code' in error && error.code === 'P2002'; + if (!isUniqueConflict) { + throw error; + } + const existing = await this.db.inputEvent.findUniqueOrThrow({ + where: { requestId }, + select: { eventType: true, payload: true }, + }); + if (existing.eventType !== command.type || stableJson(existing.payload) !== stableJson(durableCommand)) { + throw new ConflictingTurnDaemonCommandError(requestId); + } + } + return requestId; + } + + async requestCommand(command: TurnDaemonCommand, timeoutMs?: number): Promise { + const requestId = await this.sendCommand(command); + return this.waitForResult(requestId, timeoutMs); + } + + async requestStatus(timeoutMs?: number): Promise { + const requestId = await this.sendCommand({ type: 'getStatus', requestId: randomUUID() }); + const payload = await this.waitForResult<{ status: TurnDaemonStatus }>(requestId, timeoutMs); + return payload?.status ?? null; + } + + private async waitForResult(requestId: string, timeoutMs?: number): Promise { + const deadline = Date.now() + (timeoutMs ?? this.requestTimeoutMs); + while (Date.now() < deadline) { + const event = await this.db.inputEvent.findUnique({ + where: { requestId }, + select: { status: true, result: true }, + }); + if (event?.status === 'SUCCEEDED') { + return event.result as T; + } + if (event?.status === 'FAILED') { + return null; + } + await delay(Math.min(50, Math.max(1, deadline - Date.now()))); + } + return null; + } +} diff --git a/app/game-api/src/daemon/idempotentTransport.ts b/app/game-api/src/daemon/idempotentTransport.ts new file mode 100644 index 0000000..d721a5e --- /dev/null +++ b/app/game-api/src/daemon/idempotentTransport.ts @@ -0,0 +1,28 @@ +import type { TurnDaemonTransport } from './transport.js'; +import type { TurnDaemonCommand, TurnDaemonCommandResult, TurnDaemonStatus } from './types.js'; + +export class IdempotentTurnDaemonTransport implements TurnDaemonTransport { + private sequence = 0; + + constructor( + private readonly transport: TurnDaemonTransport, + private readonly parentRequestId: string + ) {} + + async sendCommand(command: TurnDaemonCommand): Promise { + return this.transport.sendCommand(this.scope(command)); + } + + async requestCommand(command: TurnDaemonCommand, timeoutMs?: number): Promise { + return this.transport.requestCommand(this.scope(command), timeoutMs); + } + + async requestStatus(timeoutMs?: number): Promise { + return this.transport.requestStatus(timeoutMs); + } + + private scope(command: TurnDaemonCommand): TurnDaemonCommand { + const requestId = `${this.parentRequestId}:engine:${this.sequence++}:${command.type}`; + return { ...command, requestId } as TurnDaemonCommand; + } +} diff --git a/app/game-api/src/daemon/inMemoryTransport.ts b/app/game-api/src/daemon/inMemoryTransport.ts index 9ce3bc8..3a14d55 100644 --- a/app/game-api/src/daemon/inMemoryTransport.ts +++ b/app/game-api/src/daemon/inMemoryTransport.ts @@ -27,7 +27,7 @@ export class InMemoryTurnDaemonTransport implements TurnDaemonTransport { // 테스트용: 메모리 큐에 명령을 저장하고 requestId를 반환한다. async sendCommand(command: TurnDaemonCommand): Promise { - const requestId = command.type === 'getStatus' && command.requestId ? command.requestId : randomUUID(); + const requestId = command.requestId ?? randomUUID(); this.commands.push({ requestId, sentAt: new Date().toISOString(), diff --git a/app/game-api/src/daemon/redisTransport.ts b/app/game-api/src/daemon/redisTransport.ts index 69d1f3d..65443c8 100644 --- a/app/game-api/src/daemon/redisTransport.ts +++ b/app/game-api/src/daemon/redisTransport.ts @@ -26,7 +26,7 @@ type RedisStreamReadResponse = Array<{ }>; const buildCommandEnvelope = (command: TurnDaemonCommand): TurnDaemonCommandEnvelope => { - const requestId = command.type === 'getStatus' && command.requestId ? command.requestId : randomUUID(); + const requestId = command.requestId ?? randomUUID(); return { requestId, sentAt: new Date().toISOString(), diff --git a/app/game-api/src/index.ts b/app/game-api/src/index.ts index 6aa5373..c69c4e6 100644 --- a/app/game-api/src/index.ts +++ b/app/game-api/src/index.ts @@ -8,11 +8,14 @@ import { runTournamentWorker } from './tournament/worker.js'; export * from './config.js'; export * from './context.js'; +export * from './inputEventBoundary.js'; export * from './router.js'; export * from './server.js'; export * from './daemon/types.js'; export * from './daemon/streamKeys.js'; export * from './daemon/transport.js'; +export * from './daemon/databaseTransport.js'; +export * from './daemon/idempotentTransport.js'; export * from './daemon/inMemoryTransport.js'; export * from './daemon/redisTransport.js'; export * from './auth/flushStore.js'; @@ -51,10 +54,10 @@ if (isMain()) { role === 'battle-sim-worker' ? runBattleSimWorker : role === 'auction-worker' - ? runAuctionWorker - : role === 'tournament-worker' - ? runTournamentWorker - : runGameApiServer; + ? runAuctionWorker + : role === 'tournament-worker' + ? runTournamentWorker + : runGameApiServer; run().catch((error) => { console.error('[game-api] failed to start', error); process.exitCode = 1; diff --git a/app/game-api/src/inputEventBoundary.ts b/app/game-api/src/inputEventBoundary.ts new file mode 100644 index 0000000..4c3c8ad --- /dev/null +++ b/app/game-api/src/inputEventBoundary.ts @@ -0,0 +1,86 @@ +import type { GamePrisma } from '@sammo-ts/infra'; + +import type { DatabaseClient } from './context.js'; + +const asJson = (value: unknown): GamePrisma.InputJsonValue => value as GamePrisma.InputJsonValue; + +export class DuplicateInputEventError extends Error { + constructor(readonly requestId: string) { + super(`Input event ${requestId} was already accepted.`); + this.name = 'DuplicateInputEventError'; + } +} + +export const executeInputEvent = async (options: { + db: DatabaseClient; + requestId: string; + eventType: string; + actorUserId?: string | null; + execute(db: DatabaseClient): Promise; +}): Promise => { + const { db, requestId, eventType, actorUserId, execute } = options; + if (!db.$transaction) { + return execute(db); + } + + const processingAt = new Date(); + try { + await db.inputEvent.create({ + data: { + requestId, + target: 'API', + eventType, + payload: asJson({}), + actorUserId: actorUserId ?? null, + status: 'PROCESSING', + processingAt, + attempts: 1, + }, + }); + } catch (error) { + const isUniqueConflict = + typeof error === 'object' && error !== null && 'code' in error && error.code === 'P2002'; + if (!isUniqueConflict) { + throw error; + } + const claimedRetry = await db.inputEvent.updateMany({ + where: { requestId, status: 'FAILED' }, + data: { + status: 'PROCESSING', + error: null, + processingAt, + completedAt: null, + attempts: { increment: 1 }, + }, + }); + if (claimedRetry.count === 0) { + throw new DuplicateInputEventError(requestId); + } + } + + try { + return await db.$transaction(async (transaction) => { + const result = await execute(transaction); + await transaction.inputEvent.update({ + where: { requestId }, + data: { + status: 'SUCCEEDED', + result: asJson({ ok: true }), + completedAt: new Date(), + }, + }); + return result; + }); + } catch (error) { + const message = error instanceof Error ? error.message : 'Unknown API input event error.'; + await db.inputEvent.update({ + where: { requestId }, + data: { + status: 'FAILED', + error: message, + completedAt: new Date(), + }, + }); + throw error; + } +}; diff --git a/app/game-api/src/server.ts b/app/game-api/src/server.ts index 38e112e..99aaa5b 100644 --- a/app/game-api/src/server.ts +++ b/app/game-api/src/server.ts @@ -14,8 +14,7 @@ import { import { resolveGameApiConfigFromEnv } from './config.js'; import { createGameApiContext, type DatabaseClient as _DatabaseClient } from './context.js'; -import { buildTurnDaemonStreamKeys } from './daemon/streamKeys.js'; -import { RedisTurnDaemonTransport } from './daemon/redisTransport.js'; +import { DatabaseTurnDaemonTransport } from './daemon/databaseTransport.js'; import { InMemoryFlushStore, RedisGatewayFlushSubscriber, type FlushStore } from './auth/flushStore.js'; import { RedisAccessTokenStore } from './auth/accessTokenStore.js'; import { appRouter } from './router.js'; @@ -66,10 +65,7 @@ export const createGameApiServer = async () => { await postgres.connect(); await redis.connect(); - const turnDaemon = new RedisTurnDaemonTransport(redis.client, { - keys: buildTurnDaemonStreamKeys(config.profileName), - requestTimeoutMs: config.daemonRequestTimeoutMs, - }); + const turnDaemon = new DatabaseTurnDaemonTransport(postgres.prisma, config.daemonRequestTimeoutMs); const battleSim = new RedisBattleSimTransport(redis.client, { keys: buildBattleSimQueueKeys(config.profileName), requestTimeoutMs: config.battleSimRequestTimeoutMs, @@ -83,10 +79,7 @@ export const createGameApiServer = async () => { const accessTokenStore = new RedisAccessTokenStore(redis.client, config.profileName); const realtimeSubscriberClient = redis.client.duplicate(); await realtimeSubscriberClient.connect(); - const realtimeHub = new RedisRealtimeEventHub( - realtimeSubscriberClient, - buildGameEventChannel(config.profileName) - ); + const realtimeHub = new RedisRealtimeEventHub(realtimeSubscriberClient, buildGameEventChannel(config.profileName)); await realtimeHub.start(); const app = fastify({ @@ -111,6 +104,10 @@ export const createGameApiServer = async () => { const token = extractBearerToken(req.headers.authorization); const auth = await resolveAuthFromToken(token, accessTokenStore, flushStore); return createGameApiContext({ + requestId: + (Array.isArray(req.headers['idempotency-key']) + ? req.headers['idempotency-key'][0] + : req.headers['idempotency-key']) || undefined, db: postgres.prisma, redis: redis.client, turnDaemon, diff --git a/app/game-api/src/trpc.ts b/app/game-api/src/trpc.ts index 51de120..a8ce3e9 100644 --- a/app/game-api/src/trpc.ts +++ b/app/game-api/src/trpc.ts @@ -1,12 +1,52 @@ +import { randomUUID } from 'node:crypto'; import { initTRPC, TRPCError } from '@trpc/server'; import type { GameApiContext } from './context.js'; +import { IdempotentTurnDaemonTransport } from './daemon/idempotentTransport.js'; +import { DuplicateInputEventError, executeInputEvent } from './inputEventBoundary.js'; const t = initTRPC.context().create(); +const inputEventMiddleware = t.middleware(async ({ ctx, type, path, next }) => { + if (type !== 'mutation' || !ctx.db.$transaction) { + return next(); + } + + const requestId = `${ctx.requestId ?? randomUUID()}:${path}`; + try { + return await executeInputEvent({ + db: ctx.db, + requestId, + eventType: path, + actorUserId: ctx.auth?.user.id, + execute: async (transaction) => { + const result = await next({ + ctx: { + ...ctx, + db: transaction, + turnDaemon: new IdempotentTurnDaemonTransport(ctx.turnDaemon, requestId), + }, + }); + if (!result.ok) { + throw result.error; + } + return result; + }, + }); + } catch (error) { + if (error instanceof DuplicateInputEventError) { + throw new TRPCError({ + code: 'CONFLICT', + message: error.message, + }); + } + throw error; + } +}); + export const router = t.router; -export const procedure = t.procedure; -export const authedProcedure: typeof t.procedure = t.procedure.use(({ ctx, next }) => { +export const procedure = t.procedure.use(inputEventMiddleware); +export const authedProcedure: typeof procedure = procedure.use(({ ctx, next }) => { if (!ctx.auth) { throw new TRPCError({ code: 'UNAUTHORIZED', diff --git a/app/game-api/test/idempotentTransport.test.ts b/app/game-api/test/idempotentTransport.test.ts new file mode 100644 index 0000000..34dce12 --- /dev/null +++ b/app/game-api/test/idempotentTransport.test.ts @@ -0,0 +1,22 @@ +import { describe, expect, it } from 'vitest'; + +import { IdempotentTurnDaemonTransport } from '../src/daemon/idempotentTransport.js'; +import { InMemoryTurnDaemonTransport } from '../src/daemon/inMemoryTransport.js'; + +describe('IdempotentTurnDaemonTransport', () => { + it('derives stable ordered engine request IDs from the API input event', async () => { + const inner = new InMemoryTurnDaemonTransport(); + const firstAttempt = new IdempotentTurnDaemonTransport(inner, 'api-event'); + const retry = new IdempotentTurnDaemonTransport(inner, 'api-event'); + + await firstAttempt.sendCommand({ type: 'vacation', generalId: 7 }); + await firstAttempt.sendCommand({ type: 'dropItem', generalId: 7, itemType: 'weapon' }); + await retry.sendCommand({ type: 'vacation', generalId: 7 }); + + expect(inner.commands.map((entry) => entry.requestId)).toEqual([ + 'api-event:engine:0:vacation', + 'api-event:engine:1:dropItem', + 'api-event:engine:0:vacation', + ]); + }); +}); diff --git a/app/game-api/test/inputEventBoundary.integration.test.ts b/app/game-api/test/inputEventBoundary.integration.test.ts new file mode 100644 index 0000000..f77728d --- /dev/null +++ b/app/game-api/test/inputEventBoundary.integration.test.ts @@ -0,0 +1,147 @@ +import { afterAll, beforeAll, describe, expect, it } from 'vitest'; +import { createGamePostgresConnector, type GamePrismaClient } from '@sammo-ts/infra'; + +import { DuplicateInputEventError, executeInputEvent } from '../src/inputEventBoundary.js'; +import { ConflictingTurnDaemonCommandError, DatabaseTurnDaemonTransport } from '../src/daemon/databaseTransport.js'; + +const databaseUrl = process.env.INPUT_EVENT_DATABASE_URL; +const integration = describe.skipIf(!databaseUrl); + +integration('API input event boundary', () => { + let close: (() => Promise) | undefined; + let db: GamePrismaClient; + + beforeAll(async () => { + const connector = createGamePostgresConnector({ url: databaseUrl! }); + await connector.connect(); + db = connector.prisma; + close = () => connector.disconnect(); + await db.inputEvent.deleteMany({ + where: { requestId: { startsWith: 'integration:api:' } }, + }); + }); + + afterAll(async () => { + await db.inputEvent.deleteMany({ + where: { requestId: { startsWith: 'integration:api:' } }, + }); + await close?.(); + }); + + it('commits a direct DB mutation and its event marker together', async () => { + const requestId = 'integration:api:success'; + const markerId = 'integration:api:success:marker'; + await executeInputEvent({ + db, + requestId, + eventType: 'test.success', + actorUserId: 'user-7', + execute: async (transaction) => { + await transaction.inputEvent.create({ + data: { + requestId: markerId, + target: 'API', + eventType: 'test.marker', + }, + }); + return { ok: true }; + }, + }); + + const [event, marker] = await Promise.all([ + db.inputEvent.findUniqueOrThrow({ where: { requestId } }), + db.inputEvent.findUniqueOrThrow({ where: { requestId: markerId } }), + ]); + expect(event).toMatchObject({ + status: 'SUCCEEDED', + actorUserId: 'user-7', + attempts: 1, + }); + expect(marker.status).toBe('PENDING'); + }); + + it('rolls back business writes, records failure, and permits one explicit retry', async () => { + const requestId = 'integration:api:retry'; + const markerId = 'integration:api:retry:marker'; + await expect( + executeInputEvent({ + db, + requestId, + eventType: 'test.failure', + execute: async (transaction) => { + await transaction.inputEvent.create({ + data: { + requestId: markerId, + target: 'API', + eventType: 'test.marker', + }, + }); + throw new Error('injected transaction failure'); + }, + }) + ).rejects.toThrow('injected transaction failure'); + + expect(await db.inputEvent.findUnique({ where: { requestId: markerId } })).toBeNull(); + expect(await db.inputEvent.findUniqueOrThrow({ where: { requestId } })).toMatchObject({ + status: 'FAILED', + attempts: 1, + error: 'injected transaction failure', + }); + + await executeInputEvent({ + db, + requestId, + eventType: 'test.failure', + execute: async () => ({ ok: true }), + }); + expect(await db.inputEvent.findUniqueOrThrow({ where: { requestId } })).toMatchObject({ + status: 'SUCCEEDED', + attempts: 2, + }); + }); + + it('rejects a concurrent duplicate idempotency key', async () => { + const requestId = 'integration:api:duplicate'; + let releaseFirst: (() => void) | undefined; + let signalStarted: (() => void) | undefined; + const started = new Promise((resolve) => { + signalStarted = resolve; + }); + const release = new Promise((resolve) => { + releaseFirst = resolve; + }); + const first = executeInputEvent({ + db, + requestId, + eventType: 'test.duplicate', + execute: async () => { + signalStarted?.(); + await release; + return { ok: true }; + }, + }); + await started; + + await expect( + executeInputEvent({ + db, + requestId, + eventType: 'test.duplicate', + execute: async () => ({ ok: true }), + }) + ).rejects.toBeInstanceOf(DuplicateInputEventError); + + releaseFirst?.(); + await first; + }); + + it('reuses the same engine child event but rejects a changed retry payload', async () => { + const transport = new DatabaseTurnDaemonTransport(db, 100); + const requestId = 'integration:api:engine-child'; + await transport.sendCommand({ type: 'vacation', requestId, generalId: 7 }); + await expect(transport.sendCommand({ type: 'vacation', requestId, generalId: 7 })).resolves.toBe(requestId); + await expect(transport.sendCommand({ type: 'vacation', requestId, generalId: 8 })).rejects.toBeInstanceOf( + ConflictingTurnDaemonCommandError + ); + }); +}); diff --git a/app/game-engine/src/auction/bidder.ts b/app/game-engine/src/auction/bidder.ts index 1804c54..a876f75 100644 --- a/app/game-engine/src/auction/bidder.ts +++ b/app/game-engine/src/auction/bidder.ts @@ -2,11 +2,14 @@ import { randomUUID } from 'node:crypto'; import { createGamePostgresConnector, GamePrisma, type GamePrismaClient } from '@sammo-ts/infra'; -import type { TurnDaemonCommand, TurnDaemonCommandResult, TurnDaemonHooks } from '../lifecycle/types.js'; +import type { TurnDaemonCommand, TurnDaemonCommandResult } from '../lifecycle/types.js'; import type { InMemoryTurnWorld } from '../turn/inMemoryWorld.js'; export interface AuctionBidder { - bid(command: Extract): Promise; + bid( + command: Extract, + db?: GamePrisma.TransactionClient + ): Promise; close(): Promise; } @@ -37,25 +40,6 @@ interface AuctionDetail { availableLatestBidCloseDate?: string | null; } -const buildFlushResult = (world: InMemoryTurnWorld) => { - const state = world.getState(); - return { - lastTurnTime: state.lastTurnTime.toISOString(), - processedGenerals: 0, - processedTurns: 0, - durationMs: 0, - partial: false, - checkpoint: world.getCheckpoint(), - }; -}; - -const flushWorld = async (world: InMemoryTurnWorld, hooks?: TurnDaemonHooks): Promise => { - if (!hooks?.flushChanges) { - return; - } - await hooks.flushChanges(buildFlushResult(world)); -}; - const parseDetail = (detail: unknown): AuctionDetail => { if (!detail || typeof detail !== 'object') { return {}; @@ -94,7 +78,9 @@ const shouldUsePrevBid = (highestBid: AuctionBidRow | null, myPrevBid: AuctionBi return myPrevBid; }; -const loadAuction = async (prisma: GamePrismaClient, auctionId: number): Promise => { +type QueryClient = Pick; + +const loadAuction = async (prisma: QueryClient, auctionId: number): Promise => { const rows = await prisma.$queryRaw( GamePrisma.sql` SELECT id, @@ -110,7 +96,7 @@ const loadAuction = async (prisma: GamePrismaClient, auctionId: number): Promise }; const loadHighestBid = async ( - prisma: GamePrismaClient, + prisma: QueryClient, auctionId: number, isReverse: boolean ): Promise => { @@ -135,7 +121,7 @@ const loadHighestBid = async ( }; const loadMyPrevBid = async ( - prisma: GamePrismaClient, + prisma: QueryClient, auctionId: number, generalId: number, isReverse: boolean @@ -160,8 +146,6 @@ const loadMyPrevBid = async ( return rows[0] ?? null; }; -type QueryClient = Pick; - const resolveUserId = async (prisma: QueryClient, generalId: number): Promise => { const rows = await prisma.$queryRaw<{ userId: string | null }[]>( GamePrisma.sql`SELECT user_id as "userId" FROM general WHERE id = ${generalId}` @@ -172,17 +156,16 @@ const resolveUserId = async (prisma: QueryClient, generalId: number): Promise => { const connector = createGamePostgresConnector({ url: options.databaseUrl }); await connector.connect(); const prisma = connector.prisma; const world = options.world; - const hooks = options.hooks; return { - bid: async (command): Promise => { - const auction = await loadAuction(prisma, command.auctionId); + bid: async (command, commandDb): Promise => { + const db = commandDb ?? prisma; + const auction = await loadAuction(db, command.auctionId); if (!auction) { return { type: 'auctionBid', ok: false, auctionId: command.auctionId, reason: '경매가 없습니다.' }; } @@ -206,8 +189,8 @@ export const createAuctionBidder = async (options: { const detail = parseDetail(auction.detail); const isReverse = detail.isReverse === true; - const highestBid = await loadHighestBid(prisma, command.auctionId, isReverse); - const myPrevBidRaw = await loadMyPrevBid(prisma, command.auctionId, command.generalId, isReverse); + const highestBid = await loadHighestBid(db, command.auctionId, isReverse); + const myPrevBidRaw = await loadMyPrevBid(db, command.auctionId, command.generalId, isReverse); const myPrevBid = shouldUsePrevBid(highestBid, myPrevBidRaw); if (highestBid) { @@ -292,7 +275,12 @@ export const createAuctionBidder = async (options: { }; } - if (auction.type !== 'UNIQUE_ITEM' && highestBid && highestBid.generalId !== command.generalId && !myPrevBid) { + if ( + auction.type !== 'UNIQUE_ITEM' && + highestBid && + highestBid.generalId !== command.generalId && + !myPrevBid + ) { const prev = world.getGeneralById(highestBid.generalId); if (!prev) { return { @@ -319,7 +307,7 @@ export const createAuctionBidder = async (options: { const eventAt = now; try { - await prisma.$transaction(async (tx) => { + const persistBid = async (tx: GamePrisma.TransactionClient): Promise => { await tx.$executeRaw( GamePrisma.sql` INSERT INTO auction_bid (auction_id, general_id, amount, event_id, event_at, meta) @@ -407,7 +395,20 @@ export const createAuctionBidder = async (options: { } } } - }); + }; + if (commandDb) { + await commandDb.$executeRawUnsafe('SAVEPOINT auction_bid_attempt'); + try { + await persistBid(commandDb); + await commandDb.$executeRawUnsafe('RELEASE SAVEPOINT auction_bid_attempt'); + } catch (error) { + await commandDb.$executeRawUnsafe('ROLLBACK TO SAVEPOINT auction_bid_attempt'); + await commandDb.$executeRawUnsafe('RELEASE SAVEPOINT auction_bid_attempt'); + throw error; + } + } else { + await prisma.$transaction(persistBid); + } } catch (error) { const reason = error instanceof Error ? error.message : 'CONFLICT'; if (reason === 'INSUFFICIENT_POINT') { @@ -445,15 +446,11 @@ export const createAuctionBidder = async (options: { const prev = world.getGeneralById(highestBid.generalId); if (prev) { world.updateGeneral(highestBid.generalId, { - gold: - resourceType === 'gold' ? prev.gold + highestBid.amount : prev.gold, - rice: - resourceType === 'rice' ? prev.rice + highestBid.amount : prev.rice, + gold: resourceType === 'gold' ? prev.gold + highestBid.amount : prev.gold, + rice: resourceType === 'rice' ? prev.rice + highestBid.amount : prev.rice, }); } } - - await flushWorld(world, hooks); } return { diff --git a/app/game-engine/src/auction/finalizer.ts b/app/game-engine/src/auction/finalizer.ts index abe79a7..c115070 100644 --- a/app/game-engine/src/auction/finalizer.ts +++ b/app/game-engine/src/auction/finalizer.ts @@ -1,13 +1,13 @@ -import { createGamePostgresConnector, GamePrisma, type GamePrismaClient } from '@sammo-ts/infra'; +import { createGamePostgresConnector, GamePrisma } from '@sammo-ts/infra'; import { ActionLogger, ItemLoader, LogFormat, UserLogger, isItemKey } from '@sammo-ts/logic'; import { JosaUtil } from '@sammo-ts/common'; -import type { TurnDaemonCommandResult, TurnDaemonHooks } from '../lifecycle/types.js'; +import type { TurnDaemonCommandResult } from '../lifecycle/types.js'; import type { InMemoryTurnWorld } from '../turn/inMemoryWorld.js'; import type { LogEntryDraft } from '@sammo-ts/logic'; export interface AuctionFinalizer { - finalize(auctionId: number): Promise; + finalize(auctionId: number, db?: GamePrisma.TransactionClient): Promise; close(): Promise; } @@ -44,25 +44,6 @@ interface AuctionDetailResource extends AuctionDetailBase { amount?: number; } -const buildFlushResult = (world: InMemoryTurnWorld) => { - const state = world.getState(); - return { - lastTurnTime: state.lastTurnTime.toISOString(), - processedGenerals: 0, - processedTurns: 0, - durationMs: 0, - partial: false, - checkpoint: world.getCheckpoint(), - }; -}; - -const flushWorld = async (world: InMemoryTurnWorld, hooks?: TurnDaemonHooks): Promise => { - if (!hooks?.flushChanges) { - return; - } - await hooks.flushChanges(buildFlushResult(world)); -}; - const parseDetail = (detail: unknown): AuctionDetailResource => { if (!detail || typeof detail !== 'object') { return {}; @@ -72,7 +53,9 @@ const parseDetail = (detail: unknown): AuctionDetailResource => { const toTurnMinutes = (tickSeconds: number): number => Math.max(1, Math.round(tickSeconds / 60)); -const resolveTurnMinutes = async (prisma: GamePrismaClient): Promise => { +type AuctionDb = GamePrisma.TransactionClient; + +const resolveTurnMinutes = async (prisma: AuctionDb): Promise => { const rows = (await prisma.$queryRaw( GamePrisma.sql`SELECT tick_seconds as "tickSeconds" FROM world_state ORDER BY id LIMIT 1` )) as Array<{ tickSeconds: number }>; @@ -104,7 +87,7 @@ const pushLogs = (world: InMemoryTurnWorld, logs: LogEntryDraft[]): void => { }; const refundInheritancePoint = async (options: { - prisma: GamePrismaClient; + prisma: AuctionDb; userId: string; amount: number; }): Promise => { @@ -142,25 +125,24 @@ const refundInheritancePoint = async (options: { export const createAuctionFinalizer = async (options: { databaseUrl: string; world: InMemoryTurnWorld; - hooks?: TurnDaemonHooks; }): Promise => { const connector = createGamePostgresConnector({ url: options.databaseUrl }); await connector.connect(); const prisma = connector.prisma; const world = options.world; - const hooks = options.hooks; const itemLoader = new ItemLoader(); - const getGeneralUserId = async (generalId: number): Promise => { - const rows = await prisma.$queryRaw<{ userId: string | null }[]>( + const getGeneralUserId = async (db: AuctionDb, generalId: number): Promise => { + const rows = await db.$queryRaw<{ userId: string | null }[]>( GamePrisma.sql`SELECT user_id as "userId" FROM general WHERE id = ${generalId}` ); return rows[0]?.userId ?? null; }; return { - finalize: async (auctionId: number): Promise => { - const rows = await prisma.$queryRaw( + finalize: async (auctionId: number, commandDb): Promise => { + const db = commandDb ?? prisma; + const rows = await db.$queryRaw( GamePrisma.sql` SELECT id, type, @@ -200,7 +182,7 @@ export const createAuctionFinalizer = async (options: { const detail = parseDetail(auction.detail); const isReverse = detail.isReverse === true; - const bidRows = await prisma.$queryRaw( + const bidRows = await db.$queryRaw( isReverse ? GamePrisma.sql` SELECT id, general_id as "generalId", amount @@ -224,7 +206,7 @@ export const createAuctionFinalizer = async (options: { const globalLogger = new ActionLogger(); const finalizeStatus = async (status: AuctionStatus) => { - await prisma.$executeRaw( + await db.$executeRaw( GamePrisma.sql` UPDATE auction SET status = ${status}, @@ -263,7 +245,10 @@ export const createAuctionFinalizer = async (options: { [resourceKey]: host[resourceKey] + amount, }); const hostLogger = new ActionLogger({ generalId: host.id, nationId: host.nationId }); - hostLogger.pushGeneralActionLog(`경매가 유찰되어 ${resourceKey === 'rice' ? '쌀' : '금'} ${amount}을 회수했습니다.`, LogFormat.PLAIN); + hostLogger.pushGeneralActionLog( + `경매가 유찰되어 ${resourceKey === 'rice' ? '쌀' : '금'} ${amount}을 회수했습니다.`, + LogFormat.PLAIN + ); logs.push(...hostLogger.flush()); globalLogger.pushGlobalActionLog(`경매 ${auctionId}번이 유찰되었습니다.`, LogFormat.PLAIN); } @@ -271,7 +256,6 @@ export const createAuctionFinalizer = async (options: { logs.push(...globalLogger.flush()); pushLogs(world, logs); - await flushWorld(world, hooks); await finalizeStatus('FINISHED'); return { type: 'auctionFinalize', ok: true, auctionId }; } @@ -358,8 +342,8 @@ export const createAuctionFinalizer = async (options: { if (!itemModule) { await finalizeStatus('CANCELED'); await refundInheritancePoint({ - prisma, - userId: (await getGeneralUserId(bidder.id)) ?? '', + prisma: db, + userId: (await getGeneralUserId(db, bidder.id)) ?? '', amount: highestBid.amount, }); return { @@ -375,7 +359,7 @@ export const createAuctionFinalizer = async (options: { if (currentItem && currentItem !== 'None' && isItemKey(currentItem)) { const currentModule = await itemLoader.load(currentItem).catch(() => null); if (currentModule && !currentModule.buyable) { - const turnMinutes = await resolveTurnMinutes(prisma); + const turnMinutes = await resolveTurnMinutes(db); const availableLatestBidCloseDate = detail.availableLatestBidCloseDate ? new Date(detail.availableLatestBidCloseDate) : null; @@ -385,7 +369,7 @@ export const createAuctionFinalizer = async (options: { turnMinutes, availableLatestBidCloseDate, }); - await prisma.$executeRaw( + await db.$executeRaw( GamePrisma.sql` UPDATE auction SET status = 'OPEN', @@ -400,7 +384,6 @@ export const createAuctionFinalizer = async (options: { ); logs.push(...globalLogger.flush()); pushLogs(world, logs); - await flushWorld(world, hooks); return { type: 'auctionFinalize', ok: false, @@ -427,12 +410,15 @@ export const createAuctionFinalizer = async (options: { ); logs.push(...bidderLogger.flush()); - const bidderUserId = await getGeneralUserId(bidder.id); + const bidderUserId = await getGeneralUserId(db, bidder.id); if (bidderUserId) { const userIdNum = Number(bidderUserId); if (Number.isFinite(userIdNum)) { const userLogger = new UserLogger(userIdNum); - userLogger.push(`유니크 ${itemModule.name} 경매로 ${highestBid.amount} 포인트 사용`, 'inheritPoint'); + userLogger.push( + `유니크 ${itemModule.name} 경매로 ${highestBid.amount} 포인트 사용`, + 'inheritPoint' + ); logs.push(...userLogger.flush()); } } @@ -442,7 +428,6 @@ export const createAuctionFinalizer = async (options: { logs.push(...globalLogger.flush()); pushLogs(world, logs); - await flushWorld(world, hooks); await finalizeStatus('FINISHED'); return { type: 'auctionFinalize', ok: true, auctionId }; diff --git a/app/game-engine/src/index.ts b/app/game-engine/src/index.ts index 189e29b..01096d0 100644 --- a/app/game-engine/src/index.ts +++ b/app/game-engine/src/index.ts @@ -5,6 +5,7 @@ import { runTurnDaemonCli } from './turn/cli.js'; export * from './lifecycle/types.js'; export * from './lifecycle/clock.js'; +export * from './lifecycle/databaseCommandQueue.js'; export * from './lifecycle/inMemoryControlQueue.js'; export * from './lifecycle/turnDaemonLifecycle.js'; export * from './lifecycle/getNextTickTime.js'; diff --git a/app/game-engine/src/lifecycle/databaseCommandQueue.ts b/app/game-engine/src/lifecycle/databaseCommandQueue.ts new file mode 100644 index 0000000..49e70e4 --- /dev/null +++ b/app/game-engine/src/lifecycle/databaseCommandQueue.ts @@ -0,0 +1,164 @@ +import { randomUUID } from 'node:crypto'; +import { GamePrisma, type GamePrismaClient } from '@sammo-ts/infra'; + +import { normalizeTurnDaemonCommand } from '../turn/commandRegistry.js'; +import type { + TurnDaemonCommand, + TurnDaemonCommandResponder, + TurnDaemonCommandResult, + TurnDaemonControlQueue, + TurnDaemonStatus, +} from './types.js'; + +const asJson = (value: unknown): GamePrisma.InputJsonValue => value as GamePrisma.InputJsonValue; +const delay = (ms: number): Promise => new Promise((resolve) => setTimeout(resolve, ms)); + +export class DatabaseTurnDaemonCommandQueue implements TurnDaemonControlQueue, TurnDaemonCommandResponder { + private readonly localQueue: TurnDaemonCommand[] = []; + private readonly workerId = randomUUID(); + private readonly leaseDurationMs = 60_000; + + constructor(private readonly db: GamePrismaClient) {} + + async initialize(): Promise { + await this.recoverExpiredLeases(); + } + + enqueue(command: TurnDaemonCommand): void { + this.localQueue.push(command); + } + + async drain(): Promise { + const local = this.localQueue.splice(0, this.localQueue.length); + const remote = await this.claimPending(); + return local.concat(remote); + } + + async waitUntil(deadlineMs: number | null): Promise { + while (deadlineMs === null || Date.now() < deadlineMs) { + const local = this.localQueue.shift(); + if (local) { + return local; + } + const remote = await this.claimPending(1); + if (remote[0]) { + return remote[0]; + } + const remaining = deadlineMs === null ? 100 : Math.max(1, Math.min(100, deadlineMs - Date.now())); + await delay(remaining); + } + return null; + } + + getDepth(): number { + return this.localQueue.length; + } + + async publishStatus(requestId: string, status: TurnDaemonStatus): Promise { + await this.complete(requestId, { status }); + } + + async publishCommandResult(requestId: string, result: TurnDaemonCommandResult): Promise { + await this.complete(requestId, result); + } + + private async claimPending(limit = 100): Promise { + await this.recoverExpiredLeases(); + return this.db.$transaction(async (transaction) => { + const rows = await transaction.$queryRaw< + Array<{ + sequence: bigint; + requestId: string; + eventType: string; + payload: unknown; + createdAt: Date; + }> + >(GamePrisma.sql` + SELECT + "sequence", + "request_id" AS "requestId", + "event_type" AS "eventType", + "payload", + "created_at" AS "createdAt" + FROM "input_event" + WHERE "target" = 'ENGINE'::"InputEventTarget" + AND "status" = 'PENDING'::"InputEventStatus" + ORDER BY "sequence" ASC + FOR UPDATE SKIP LOCKED + LIMIT ${limit} + `); + if (rows.length === 0) { + return []; + } + await transaction.inputEvent.updateMany({ + where: { + sequence: { in: rows.map((row) => row.sequence) }, + target: 'ENGINE', + status: 'PENDING', + }, + data: { + status: 'PROCESSING', + processingAt: new Date(), + lockedBy: this.workerId, + leaseUntil: new Date(Date.now() + this.leaseDurationMs), + attempts: { increment: 1 }, + }, + }); + + const commands: TurnDaemonCommand[] = []; + for (const row of rows) { + const command = normalizeTurnDaemonCommand({ + requestId: row.requestId, + sentAt: row.createdAt.toISOString(), + command: row.payload as TurnDaemonCommand, + }); + if (!command) { + await transaction.inputEvent.update({ + where: { sequence: row.sequence }, + data: { + status: 'FAILED', + error: `Invalid command payload for ${row.eventType}`, + completedAt: new Date(), + lockedBy: null, + leaseUntil: null, + }, + }); + continue; + } + commands.push(command); + } + return commands; + }); + } + + private async complete(requestId: string, result: unknown): Promise { + await this.db.inputEvent.update({ + where: { requestId }, + data: { + status: 'SUCCEEDED', + result: asJson(result), + completedAt: new Date(), + error: null, + lockedBy: null, + leaseUntil: null, + }, + }); + } + + private async recoverExpiredLeases(): Promise { + const now = new Date(); + await this.db.inputEvent.updateMany({ + where: { + target: 'ENGINE', + status: 'PROCESSING', + leaseUntil: { lt: now }, + }, + data: { + status: 'PENDING', + processingAt: null, + lockedBy: null, + leaseUntil: null, + }, + }); + } +} diff --git a/app/game-engine/src/lifecycle/turnDaemonLifecycle.ts b/app/game-engine/src/lifecycle/turnDaemonLifecycle.ts index d7d2536..11b3e9e 100644 --- a/app/game-engine/src/lifecycle/turnDaemonLifecycle.ts +++ b/app/game-engine/src/lifecycle/turnDaemonLifecycle.ts @@ -13,6 +13,7 @@ import type { TurnDaemonCommandHandler, TurnDaemonCommandResponder, TurnDaemonCommandResult, + TurnDaemonCommandExecutionContext, } from './types.js'; type PendingRun = { @@ -21,6 +22,12 @@ type PendingRun = { budget?: TurnRunBudget; }; +type TurnDaemonControlCommand = Extract< + TurnDaemonCommand, + { type: 'pause' | 'resume' | 'shutdown' | 'getStatus' | 'run' } +>; +type TurnDaemonMutationCommand = Exclude; + export interface TurnDaemonLifecycleOptions { profile: string; defaultBudget: TurnRunBudget; @@ -217,15 +224,24 @@ export class TurnDaemonLifecycle { this.manualPaused = true; this.status.paused = true; this.status.state = 'paused'; + if (command.requestId) { + await this.commandResponder?.publishStatus(command.requestId, this.getStatus()); + } return; case 'resume': this.manualPaused = false; this.status.paused = this.errorPaused; this.status.state = 'idle'; + if (command.requestId) { + await this.commandResponder?.publishStatus(command.requestId, this.getStatus()); + } return; case 'shutdown': this.status.state = 'stopping'; this.stopping = true; + if (command.requestId) { + await this.commandResponder?.publishStatus(command.requestId, this.getStatus()); + } return; case 'getStatus': { if (command.requestId) { @@ -240,79 +256,61 @@ export class TurnDaemonLifecycle { budget: command.budget, }; this.status.pendingReason = command.reason; + if (command.requestId) { + await this.commandResponder?.publishStatus(command.requestId, this.getStatus()); + } return; - case 'troopJoin': - case 'troopExit': - case 'dieOnPrestart': - case 'buildNationCandidate': - case 'instantRetreat': - case 'vacation': - case 'setMySetting': - case 'dropItem': - case 'auctionFinalize': - case 'changePermission': - case 'kick': - case 'appoint': + default: await this.handleMutationCommand(command); return; } } - private async handleMutationCommand( - command: Extract< - TurnDaemonCommand, - | { type: 'troopJoin' } - | { type: 'troopExit' } - | { type: 'dieOnPrestart' } - | { type: 'buildNationCandidate' } - | { type: 'instantRetreat' } - | { type: 'vacation' } - | { type: 'setMySetting' } - | { type: 'dropItem' } - | { type: 'auctionFinalize' } - | { type: 'changePermission' } - | { type: 'kick' } - | { type: 'appoint' } - > - ): Promise { - let result: TurnDaemonCommandResult | null = null; - try { - result = this.commandHandler ? await this.commandHandler.handle(command) : null; - if (!result) { - if (command.type === 'auctionFinalize') { - result = { - type: 'auctionFinalize', - ok: false, - auctionId: command.auctionId, - reason: '턴 데몬이 경매 확정을 처리할 수 없습니다.', - }; - } else { - result = { - type: command.type, - ok: false, - generalId: command.generalId, - reason: '턴 데몬이 명령을 처리할 수 없습니다.', - ...(command.type === 'troopJoin' ? { troopId: command.troopId } : {}), - } as TurnDaemonCommandResult; + private async handleMutationCommand(command: TurnDaemonMutationCommand): Promise { + let result: TurnDaemonCommandResult; + let committedByExecutionBoundary = false; + const executeHandler = async ( + context?: TurnDaemonCommandExecutionContext + ): Promise => { + const handled = this.commandHandler ? await this.commandHandler.handle(command, context) : null; + return ( + handled ?? { + type: 'commandRejected', + ok: false, + commandType: command.type, + reason: '턴 데몬이 명령을 처리할 수 없습니다.', } + ); + }; + try { + if (command.requestId && this.hooks?.executeCommand) { + result = await this.hooks.executeCommand(command.requestId, executeHandler); + committedByExecutionBoundary = true; + } else { + result = await executeHandler(); } } catch (error) { - const reason = error instanceof Error ? error.message : 'Unknown command error.'; - if (command.type === 'auctionFinalize') { - result = { - type: 'auctionFinalize', - ok: false, - auctionId: command.auctionId, - reason, - }; - } else { - result = { - type: command.type, - ok: false, - generalId: command.generalId, - reason, - ...(command.type === 'troopJoin' ? { troopId: command.troopId } : {}), - } as TurnDaemonCommandResult; + // A handler may already have changed the in-memory world. Do not commit + // either those changes or the inbox completion marker after an exception. + // Pausing forces a reload/retry instead of acknowledging a partial event. + this.status.state = 'paused'; + this.status.paused = true; + this.errorPaused = true; + this.status.lastError = error instanceof Error ? error.message : 'Unknown command error.'; + await this.hooks?.onRunError?.(error); + return; + } + + if (!committedByExecutionBoundary && command.requestId && this.hooks?.commitCommand) { + try { + await this.hooks.commitCommand(command.requestId, result); + } catch (error) { + this.status.state = 'paused'; + this.status.paused = true; + this.errorPaused = true; + this.status.lastError = error instanceof Error ? error.message : 'Unknown input event commit error.'; + await this.hooks.onRunError?.(error); + return; } } @@ -347,12 +345,26 @@ export class TurnDaemonLifecycle { } this.status.state = 'flushing'; - await this.stateStore.saveLastTurnTime(new Date(result.lastTurnTime)); - await this.stateStore.saveCheckpoint(result.checkpoint); - await this.hooks?.flushChanges?.(result); - await this.hooks?.publishEvents?.(result); + try { + await this.stateStore.saveLastTurnTime(new Date(result.lastTurnTime)); + await this.stateStore.saveCheckpoint(result.checkpoint); + await this.hooks?.flushChanges?.(result); + } catch (error) { + this.status.state = 'paused'; + this.status.paused = true; + this.errorPaused = true; + this.status.lastError = error instanceof Error ? error.message : 'Unknown turn flush error.'; + await this.hooks?.onRunError?.(error); + return; + } + await this.applyRunResult(result, startMs); this.status.state = 'idle'; + try { + await this.hooks?.publishEvents?.(result); + } catch (error) { + this.status.lastError = error instanceof Error ? error.message : 'Unknown event publication error.'; + } } private async applyRunResult(result: TurnRunResult, startMs: number): Promise { diff --git a/app/game-engine/src/lifecycle/types.ts b/app/game-engine/src/lifecycle/types.ts index 1ffcd21..7307db7 100644 --- a/app/game-engine/src/lifecycle/types.ts +++ b/app/game-engine/src/lifecycle/types.ts @@ -6,6 +6,7 @@ import type { TurnRunBudget, TurnRunResult, } from '@sammo-ts/common'; +import type { GamePrisma } from '@sammo-ts/infra'; export type { RunReason, @@ -19,7 +20,14 @@ export type { } from '@sammo-ts/common'; export interface TurnDaemonCommandHandler { - handle(command: TurnDaemonCommand): Promise; + handle( + command: TurnDaemonCommand, + context?: TurnDaemonCommandExecutionContext + ): Promise; +} + +export interface TurnDaemonCommandExecutionContext { + db?: GamePrisma.TransactionClient; } export interface TurnDaemonCommandResponder { @@ -53,6 +61,11 @@ export interface TurnDaemonControlQueue { export interface TurnDaemonHooks { flushChanges?(result: TurnRunResult): Promise; + commitCommand?(requestId: string, result: TurnDaemonCommandResult): Promise; + executeCommand?( + requestId: string, + execute: (context: TurnDaemonCommandExecutionContext) => Promise + ): Promise; publishEvents?(result: TurnRunResult): Promise; onRunError?(error: unknown): Promise; } diff --git a/app/game-engine/src/tournament/finalizer.ts b/app/game-engine/src/tournament/finalizer.ts index e4b0060..1d80828 100644 --- a/app/game-engine/src/tournament/finalizer.ts +++ b/app/game-engine/src/tournament/finalizer.ts @@ -1,12 +1,15 @@ import { JosaUtil, asRecord } from '@sammo-ts/common'; -import { createGamePostgresConnector } from '@sammo-ts/infra'; +import { createGamePostgresConnector, type GamePrisma } from '@sammo-ts/infra'; import { ActionLogger, LogFormat, type TournamentType, type TriggerValue } from '@sammo-ts/logic'; -import type { TurnDaemonCommand, TurnDaemonCommandResult, TurnDaemonHooks } from '../lifecycle/types.js'; +import type { TurnDaemonCommand, TurnDaemonCommandResult } from '../lifecycle/types.js'; import type { InMemoryTurnWorld } from '../turn/inMemoryWorld.js'; export interface TournamentRewardFinalizer { - finalize(command: Extract): Promise; + finalize( + command: Extract, + db?: GamePrisma.TransactionClient + ): Promise; close(): Promise; } @@ -57,39 +60,22 @@ const pushLogs = (world: InMemoryTurnWorld, logs: ReturnType => { - if (!hooks?.flushChanges) { - return; - } - const state = world.getState(); - await hooks.flushChanges({ - lastTurnTime: state.lastTurnTime.toISOString(), - processedGenerals: 0, - processedTurns: 0, - durationMs: 0, - partial: false, - checkpoint: world.getCheckpoint(), - }); -}; - export const createTournamentRewardFinalizer = async (options: { databaseUrl: string; world: InMemoryTurnWorld; - hooks?: TurnDaemonHooks; }): Promise => { const connector = createGamePostgresConnector({ url: options.databaseUrl }); await connector.connect(); const prisma = connector.prisma; const finalize = async ( - command: Extract + command: Extract, + commandDb?: GamePrisma.TransactionClient ): Promise => { - const { world, hooks } = options; + const { world } = options; + const db = commandDb ?? prisma; const { winnerId, runnerUpId } = command; - const rewardMap = new Map< - number, - { gold: number; exp: number; label: string; inheritPoint: number } - >(); + const rewardMap = new Map(); const applyTier = ( ids: number[], @@ -126,7 +112,7 @@ export const createTournamentRewardFinalizer = async (options: { } const nameMap = new Map(); - const generals = await prisma.general.findMany({ + const generals = await db.general.findMany({ where: { id: { in: Array.from(rewardMap.keys()) } }, select: { id: true, userId: true, name: true }, }); @@ -240,7 +226,7 @@ export const createTournamentRewardFinalizer = async (options: { .filter((entry) => !!entry.userId); for (const entry of pointUpdates) { - await prisma.inheritancePoint.upsert({ + await db.inheritancePoint.upsert({ where: { userId_key: { userId: entry.userId!, key: 'tournament' }, }, @@ -249,8 +235,6 @@ export const createTournamentRewardFinalizer = async (options: { }); } - await flushWorld(world, hooks); - return { type: 'tournamentReward', ok: true, @@ -269,4 +253,4 @@ export const createTournamentRewardFinalizer = async (options: { await connector.disconnect(); }, }; -}; \ No newline at end of file +}; diff --git a/app/game-engine/src/turn/commandRegistry.ts b/app/game-engine/src/turn/commandRegistry.ts index 1f535ef..3f80ae8 100644 --- a/app/game-engine/src/turn/commandRegistry.ts +++ b/app/game-engine/src/turn/commandRegistry.ts @@ -1,10 +1,6 @@ import { z } from 'zod'; -import type { - TurnDaemonCommand, - TurnDaemonCommandType, - TurnDaemonCommandByType, -} from '@sammo-ts/common'; +import type { TurnDaemonCommand, TurnDaemonCommandType, TurnDaemonCommandByType } from '@sammo-ts/common'; export type TurnDaemonCommandEnvelope = { requestId: string; @@ -161,22 +157,21 @@ const zSetNationMeta = z.object({ expectedUpdatedAt: z.string().optional(), }); -const zAdjustGeneralResources = z - .object({ - type: z.literal('adjustGeneralResources'), - reason: z.string().optional(), - adjustments: z - .array( - z - .object({ - generalId: zFiniteNumber, - goldDelta: zFiniteNumber.optional(), - riceDelta: zFiniteNumber.optional(), - }) - .refine((value) => value.goldDelta !== undefined || value.riceDelta !== undefined) - ) - .min(1), - }); +const zAdjustGeneralResources = z.object({ + type: z.literal('adjustGeneralResources'), + reason: z.string().optional(), + adjustments: z + .array( + z + .object({ + generalId: zFiniteNumber, + goldDelta: zFiniteNumber.optional(), + riceDelta: zFiniteNumber.optional(), + }) + .refine((value) => value.goldDelta !== undefined || value.riceDelta !== undefined) + ) + .min(1), +}); const zAdjustGeneralMeta = z.object({ type: z.literal('adjustGeneralMeta'), @@ -430,10 +425,22 @@ const normalizeGetStatus: CommandNormalizer<'getStatus'> = (envelope) => { }; }; -const normalizeRun: CommandNormalizer<'run'> = (envelope) => parseWith(zRun, envelope.command); -const normalizePause: CommandNormalizer<'pause'> = (envelope) => parseWith(zPause, envelope.command); -const normalizeResume: CommandNormalizer<'resume'> = (envelope) => parseWith(zResume, envelope.command); -const normalizeShutdown: CommandNormalizer<'shutdown'> = (envelope) => parseWith(zShutdown, envelope.command); +const normalizeRun: CommandNormalizer<'run'> = (envelope) => { + const command = parseWith(zRun, envelope.command); + return command ? { ...command, requestId: envelope.requestId } : null; +}; +const normalizePause: CommandNormalizer<'pause'> = (envelope) => { + const command = parseWith(zPause, envelope.command); + return command ? { ...command, requestId: envelope.requestId } : null; +}; +const normalizeResume: CommandNormalizer<'resume'> = (envelope) => { + const command = parseWith(zResume, envelope.command); + return command ? { ...command, requestId: envelope.requestId } : null; +}; +const normalizeShutdown: CommandNormalizer<'shutdown'> = (envelope) => { + const command = parseWith(zShutdown, envelope.command); + return command ? { ...command, requestId: envelope.requestId } : null; +}; const normalizers: CommandNormalizerMap = { auctionFinalize: normalizeAuctionFinalize, diff --git a/app/game-engine/src/turn/databaseHooks.ts b/app/game-engine/src/turn/databaseHooks.ts index 4d3efd9..9245fdc 100644 --- a/app/game-engine/src/turn/databaseHooks.ts +++ b/app/game-engine/src/turn/databaseHooks.ts @@ -1,5 +1,6 @@ import { createGamePostgresConnector, + type GamePrisma, type InputJsonValue, type TurnEngineCityUpdateInput, type TurnEngineDiplomacyCreateManyInput, @@ -15,7 +16,7 @@ import { import { finalizeLogEntry, LogCategory, LogScope, type LogEntryDraft } from '@sammo-ts/logic'; import { asRecord, type RankDataType } from '@sammo-ts/common'; -import type { TurnDaemonHooks } from '../lifecycle/types.js'; +import type { TurnDaemonCommandResult, TurnDaemonHooks } from '../lifecycle/types.js'; import type { InMemoryTurnWorld } from './inMemoryWorld.js'; import type { InMemoryReservedTurnStore } from './reservedTurnStore.js'; import { buildDiplomacyMeta } from '@sammo-ts/logic'; @@ -305,32 +306,37 @@ export const createDatabaseTurnHooks = async ( await connector.connect(); const prisma = connector.prisma; - const hooks: TurnDaemonHooks = { - flushChanges: async () => { - const state = world.getState(); - const { - generals, - cities, - nations, - troops, - deletedTroops, - deletedGenerals, - deletedNations, - deletedNationSnapshots, - diplomacy, - logs, - createdGenerals, - createdNations, - createdTroops, - createdDiplomacy, - } = world.consumeDirtyState(); + const persistChanges = async ( + transaction?: GamePrisma.TransactionClient, + commandCompletion?: { requestId: string; result: TurnDaemonCommandResult } + ): Promise<() => void> => { + const state = world.getState(); + const changes = world.peekDirtyState(); + const { + generals, + cities, + nations, + troops, + deletedTroops, + deletedGenerals, + deletedNations, + deletedNationSnapshots, + diplomacy, + logs, + createdGenerals, + createdNations, + createdTroops, + createdDiplomacy, + } = changes; + const reservedTurnChanges = options?.reservedTurns?.peekDirtyState(); - const worldStateUpdate: TurnEngineWorldStateUpdateInput = { - currentYear: state.currentYear, - currentMonth: state.currentMonth, - tickSeconds: state.tickSeconds, - meta: asJson(state.meta), - }; + const worldStateUpdate: TurnEngineWorldStateUpdateInput = { + currentYear: state.currentYear, + currentMonth: state.currentMonth, + tickSeconds: state.tickSeconds, + meta: asJson(state.meta), + }; + const persist = async (prisma: GamePrisma.TransactionClient): Promise => { await prisma.worldState.update({ where: { id: state.id }, data: worldStateUpdate, @@ -462,10 +468,7 @@ export const createDatabaseTurnHooks = async ( if (deletedNations.length > 0) { await prisma.diplomacy.deleteMany({ where: { - OR: [ - { srcNationId: { in: deletedNations } }, - { destNationId: { in: deletedNations } }, - ], + OR: [{ srcNationId: { in: deletedNations } }, { destNationId: { in: deletedNations } }], }, }); await prisma.nationTurn.deleteMany({ @@ -563,9 +566,52 @@ export const createDatabaseTurnHooks = async ( }); } } - if (options?.reservedTurns) { - await options.reservedTurns.flushChanges(); + if (options?.reservedTurns && reservedTurnChanges) { + await options.reservedTurns.persistChanges(prisma, reservedTurnChanges); } + if (commandCompletion) { + await prisma.inputEvent.update({ + where: { requestId: commandCompletion.requestId }, + data: { + status: 'SUCCEEDED', + result: asJson(commandCompletion.result), + completedAt: new Date(), + error: null, + }, + }); + } + }; + if (transaction) { + await persist(transaction); + } else { + await prisma.$transaction(persist); + } + + return () => { + world.acknowledgeDirtyState(changes); + if (options?.reservedTurns && reservedTurnChanges) { + options.reservedTurns.acknowledgeDirtyState(reservedTurnChanges); + } + }; + }; + + const hooks: TurnDaemonHooks = { + flushChanges: async () => { + const acknowledge = await persistChanges(); + acknowledge(); + }, + commitCommand: async (requestId, result) => { + const acknowledge = await persistChanges(undefined, { requestId, result }); + acknowledge(); + }, + executeCommand: async (requestId, execute) => { + const committed = await prisma.$transaction(async (transaction) => { + const result = await execute({ db: transaction }); + const acknowledge = await persistChanges(transaction, { requestId, result }); + return { result, acknowledge }; + }); + committed.acknowledge(); + return committed.result; }, }; diff --git a/app/game-engine/src/turn/inMemoryWorld.ts b/app/game-engine/src/turn/inMemoryWorld.ts index 2f1d31b..531cef3 100644 --- a/app/game-engine/src/turn/inMemoryWorld.ts +++ b/app/game-engine/src/turn/inMemoryWorld.ts @@ -68,6 +68,23 @@ export interface InMemoryTurnWorldOptions { calendarHandler?: TurnCalendarHandler; } +export interface TurnWorldChanges { + generals: TurnGeneral[]; + cities: City[]; + nations: Nation[]; + troops: Troop[]; + deletedTroops: number[]; + deletedGenerals: number[]; + deletedNations: number[]; + deletedNationSnapshots: Array<{ nation: Nation; generalIds: number[]; removedAt: Date }>; + diplomacy: TurnDiplomacy[]; + logs: LogEntryDraft[]; + createdGenerals: TurnGeneral[]; + createdNations: Nation[]; + createdTroops: Troop[]; + createdDiplomacy: TurnDiplomacy[]; +} + const compareTurnOrder = (left: TurnGeneral, right: TurnGeneral): number => { const timeDiff = left.turnTime.getTime() - right.turnTime.getTime(); if (timeDiff !== 0) { @@ -694,22 +711,7 @@ export class InMemoryTurnWorld { } } - consumeDirtyState(): { - generals: TurnGeneral[]; - cities: City[]; - nations: Nation[]; - troops: Troop[]; - deletedTroops: number[]; - deletedGenerals: number[]; - deletedNations: number[]; - deletedNationSnapshots: Array<{ nation: Nation; generalIds: number[]; removedAt: Date }>; - diplomacy: TurnDiplomacy[]; - logs: LogEntryDraft[]; - createdGenerals: TurnGeneral[]; - createdNations: Nation[]; - createdTroops: Troop[]; - createdDiplomacy: TurnDiplomacy[]; - } { + peekDirtyState(): TurnWorldChanges { const generals = Array.from(this.dirtyGeneralIds) .map((id) => this.generals.get(id)) .filter((general): general is TurnGeneral => Boolean(general)); @@ -740,21 +742,8 @@ export class InMemoryTurnWorld { const deletedTroops = Array.from(this.deletedTroopIds); const deletedGenerals = Array.from(this.deletedGeneralIds); const deletedNations = Array.from(this.deletedNationIds); - const deletedNationSnapshots = this.deletedNationSnapshots.splice(0, this.deletedNationSnapshots.length); - const logs = this.logs.splice(0, this.logs.length); - - this.dirtyGeneralIds.clear(); - this.dirtyCityIds.clear(); - this.dirtyNationIds.clear(); - this.dirtyTroopIds.clear(); - this.dirtyDiplomacyKeys.clear(); - this.createdGeneralIds.clear(); - this.createdNationIds.clear(); - this.createdTroopIds.clear(); - this.createdDiplomacyKeys.clear(); - this.deletedTroopIds.clear(); - this.deletedGeneralIds.clear(); - this.deletedNationIds.clear(); + const deletedNationSnapshots = this.deletedNationSnapshots.slice(); + const logs = this.logs.slice(); return { generals, @@ -774,6 +763,33 @@ export class InMemoryTurnWorld { }; } + acknowledgeDirtyState(changes: TurnWorldChanges): void { + for (const general of changes.generals) this.dirtyGeneralIds.delete(general.id); + for (const city of changes.cities) this.dirtyCityIds.delete(city.id); + for (const nation of changes.nations) this.dirtyNationIds.delete(nation.id); + for (const troop of changes.troops) this.dirtyTroopIds.delete(troop.id); + for (const entry of changes.diplomacy) { + this.dirtyDiplomacyKeys.delete(buildDiplomacyKey(entry.fromNationId, entry.toNationId)); + } + for (const general of changes.createdGenerals) this.createdGeneralIds.delete(general.id); + for (const nation of changes.createdNations) this.createdNationIds.delete(nation.id); + for (const troop of changes.createdTroops) this.createdTroopIds.delete(troop.id); + for (const entry of changes.createdDiplomacy) { + this.createdDiplomacyKeys.delete(buildDiplomacyKey(entry.fromNationId, entry.toNationId)); + } + for (const id of changes.deletedTroops) this.deletedTroopIds.delete(id); + for (const id of changes.deletedGenerals) this.deletedGeneralIds.delete(id); + for (const id of changes.deletedNations) this.deletedNationIds.delete(id); + this.deletedNationSnapshots.splice(0, changes.deletedNationSnapshots.length); + this.logs.splice(0, changes.logs.length); + } + + consumeDirtyState(): TurnWorldChanges { + const changes = this.peekDirtyState(); + this.acknowledgeDirtyState(changes); + return changes; + } + private removeCollapsedNations(): void { const collapsedNationIds: number[] = []; for (const nation of this.nations.values()) { diff --git a/app/game-engine/src/turn/reservedTurnStore.ts b/app/game-engine/src/turn/reservedTurnStore.ts index 1d14a28..062b08b 100644 --- a/app/game-engine/src/turn/reservedTurnStore.ts +++ b/app/game-engine/src/turn/reservedTurnStore.ts @@ -72,6 +72,11 @@ const buildNationKey = (nationId: number, officerLevel: number): string => `${na type ReservedTurnDatabaseClient = Pick; +export interface ReservedTurnChanges { + generalIds: number[]; + nationKeys: string[]; +} + export class InMemoryReservedTurnStore { private readonly generalTurns = new Map(); private readonly nationTurns = new Map(); @@ -213,12 +218,27 @@ export class InMemoryReservedTurnStore { this.dirtyNationKeys.add(key); } - async flushChanges(): Promise { - const generalIds = Array.from(this.dirtyGeneralIds); - for (const generalId of generalIds) { + peekDirtyState(): ReservedTurnChanges { + return { + generalIds: Array.from(this.dirtyGeneralIds), + nationKeys: Array.from(this.dirtyNationKeys), + }; + } + + acknowledgeDirtyState(changes: ReservedTurnChanges): void { + for (const generalId of changes.generalIds) { + this.dirtyGeneralIds.delete(generalId); + } + for (const key of changes.nationKeys) { + this.dirtyNationKeys.delete(key); + } + } + + async persistChanges(prisma: ReservedTurnDatabaseClient, changes: ReservedTurnChanges): Promise { + for (const generalId of changes.generalIds) { const turns = this.getGeneralTurns(generalId); - await this.prisma.generalTurn.deleteMany({ where: { generalId } }); - await this.prisma.generalTurn.createMany({ + await prisma.generalTurn.deleteMany({ where: { generalId } }); + await prisma.generalTurn.createMany({ data: turns.map((entry, turnIdx) => ({ generalId, turnIdx, @@ -228,16 +248,15 @@ export class InMemoryReservedTurnStore { }); } - const nationKeys = Array.from(this.dirtyNationKeys); - for (const key of nationKeys) { + for (const key of changes.nationKeys) { const [nationIdRaw, officerLevelRaw] = key.split(':'); const nationId = Number(nationIdRaw); const officerLevel = Number(officerLevelRaw); const turns = this.getNationTurns(nationId, officerLevel); - await this.prisma.nationTurn.deleteMany({ + await prisma.nationTurn.deleteMany({ where: { nationId, officerLevel }, }); - await this.prisma.nationTurn.createMany({ + await prisma.nationTurn.createMany({ data: turns.map((entry, turnIdx) => ({ nationId, officerLevel, @@ -247,9 +266,12 @@ export class InMemoryReservedTurnStore { })), }); } + } - this.dirtyGeneralIds.clear(); - this.dirtyNationKeys.clear(); + async flushChanges(): Promise { + const changes = this.peekDirtyState(); + await this.persistChanges(this.prisma, changes); + this.acknowledgeDirtyState(changes); } } diff --git a/app/game-engine/src/turn/turnDaemon.ts b/app/game-engine/src/turn/turnDaemon.ts index 6e2bf59..62d4fbb 100644 --- a/app/game-engine/src/turn/turnDaemon.ts +++ b/app/game-engine/src/turn/turnDaemon.ts @@ -1,6 +1,6 @@ import type { TurnCommandProfile, TurnSchedule } from '@sammo-ts/logic'; import { buildGameEventChannel, type RealtimeEvent } from '@sammo-ts/common'; -import { createRedisConnector, resolveRedisConfigFromEnv } from '@sammo-ts/infra'; +import { createGamePostgresConnector, createRedisConnector, resolveRedisConfigFromEnv } from '@sammo-ts/infra'; import { NATION_TRAIT_KEYS, NationTraitLoader, loadNationTraitModules } from '@sammo-ts/logic'; import { SystemClock } from '../lifecycle/clock.js'; @@ -8,7 +8,7 @@ import { getNextTickTime } from '../lifecycle/getNextTickTime.js'; import { InMemoryControlQueue } from '../lifecycle/inMemoryControlQueue.js'; import type { Clock, TurnDaemonControlQueue, TurnDaemonHooks, TurnRunBudget } from '../lifecycle/types.js'; import { TurnDaemonLifecycle } from '../lifecycle/turnDaemonLifecycle.js'; -import { buildTurnDaemonStreamKeys, RedisTurnDaemonCommandStream } from '../lifecycle/redisCommandStream.js'; +import { DatabaseTurnDaemonCommandQueue } from '../lifecycle/databaseCommandQueue.js'; import type { MapLoaderOptions } from '../scenario/mapLoader.js'; import { createDatabaseTurnHooks } from './databaseHooks.js'; import type { GeneralTurnHandler, InMemoryTurnWorldOptions, TurnCalendarHandler } from './inMemoryWorld.js'; @@ -201,7 +201,6 @@ export const createTurnDaemonRuntime = async (options: TurnDaemonRuntimeOptions) let auctionFinalizer: Awaited> | null = null; let auctionBidder: Awaited> | null = null; let tournamentRewardFinalizer: Awaited> | null = null; - let redisCommandStream: RedisTurnDaemonCommandStream | null = null; let pauseGate: (() => Promise) | undefined; let adminActionConsumer: Awaited> | null = null; const gatewayGate = options.profileName @@ -222,17 +221,14 @@ export const createTurnDaemonRuntime = async (options: TurnDaemonRuntimeOptions) auctionBidder = await createAuctionBidder({ databaseUrl: options.databaseUrl, world, - hooks: dbHooks.hooks, }); auctionFinalizer = await createAuctionFinalizer({ databaseUrl: options.databaseUrl, world, - hooks: dbHooks.hooks, }); tournamentRewardFinalizer = await createTournamentRewardFinalizer({ databaseUrl: options.databaseUrl, world, - hooks: dbHooks.hooks, }); hooks = { ...dbHooks.hooks, @@ -290,10 +286,6 @@ export const createTurnDaemonRuntime = async (options: TurnDaemonRuntimeOptions) redisConnector = createRedisConnector(redisConfig); await redisConnector.connect(); const redisClient = redisConnector.client; - redisCommandStream = new RedisTurnDaemonCommandStream(redisClient, { - keys: buildTurnDaemonStreamKeys(options.profileName ?? options.profile), - startId: options.commandStreamStartId, - }); const realtimeChannel = buildGameEventChannel(options.profileName ?? options.profile); publishRealtimeEvent = async (event: RealtimeEvent) => { await redisClient.publish(realtimeChannel, JSON.stringify(event)); @@ -320,6 +312,13 @@ export const createTurnDaemonRuntime = async (options: TurnDaemonRuntimeOptions) }; } + const commandConnector = hooks ? createGamePostgresConnector({ url: options.databaseUrl }) : null; + const databaseCommandQueue = commandConnector ? new DatabaseTurnDaemonCommandQueue(commandConnector.prisma) : null; + if (commandConnector && databaseCommandQueue) { + await commandConnector.connect(); + await databaseCommandQueue.initialize(); + } + const baseClose = close; close = async () => { await baseClose(); @@ -330,12 +329,12 @@ export const createTurnDaemonRuntime = async (options: TurnDaemonRuntimeOptions) if (redisConnector) { await redisConnector.disconnect(); } + await commandConnector?.disconnect(); }; - const resolvedControlQueue = options.controlQueue ?? redisCommandStream ?? controlQueue; + const resolvedControlQueue = options.controlQueue ?? databaseCommandQueue ?? controlQueue; const commandHandler = createTurnDaemonCommandHandler({ world, - hooks, auctionFinalizer: auctionFinalizer ?? undefined, auctionBidder: auctionBidder ?? undefined, tournamentRewardFinalizer: tournamentRewardFinalizer ?? undefined, @@ -357,7 +356,7 @@ export const createTurnDaemonRuntime = async (options: TurnDaemonRuntimeOptions) hooks, pauseGate, commandHandler, - commandResponder: redisCommandStream ?? undefined, + commandResponder: options.controlQueue ? undefined : (databaseCommandQueue ?? undefined), }, { profile: options.profile, defaultBudget } ); diff --git a/app/game-engine/src/turn/worldCommandHandler.ts b/app/game-engine/src/turn/worldCommandHandler.ts index 900fb4b..cd0fa29 100644 --- a/app/game-engine/src/turn/worldCommandHandler.ts +++ b/app/game-engine/src/turn/worldCommandHandler.ts @@ -1,10 +1,10 @@ import type { - TurnDaemonHooks, TurnDaemonCommandHandler, TurnDaemonCommand, + TurnDaemonCommandExecutionContext, TurnDaemonCommandResult, - TurnRunResult, } from '../lifecycle/types.js'; +import type { GamePrisma } from '@sammo-ts/infra'; import { asRecord, JosaUtil, LiteHashDRBG, RandUtil } from '@sammo-ts/common'; import { LogCategory, @@ -23,25 +23,6 @@ import { import type { InMemoryTurnWorld } from './inMemoryWorld.js'; import type { TurnGeneral } from './types.js'; -const buildFlushResult = (world: InMemoryTurnWorld): TurnRunResult => { - const state = world.getState(); - return { - lastTurnTime: state.lastTurnTime.toISOString(), - processedGenerals: 0, - processedTurns: 0, - durationMs: 0, - partial: false, - checkpoint: world.getCheckpoint(), - }; -}; - -const flushWorld = async (world: InMemoryTurnWorld, hooks?: TurnDaemonHooks): Promise => { - if (!hooks?.flushChanges) { - return; - } - await hooks.flushChanges(buildFlushResult(world)); -}; - let itemRegistryPromise: Promise> | null = null; const getItemRegistry = async (): Promise> => { @@ -67,29 +48,35 @@ const readMetaNumber = (meta: Record, key: string, fallback: nu interface CommandHandlerContext { world: InMemoryTurnWorld; - hooks?: TurnDaemonHooks; + commandDb?: GamePrisma.TransactionClient; auctionFinalizer?: AuctionFinalizer; auctionBidder?: AuctionBidder; tournamentRewardFinalizer?: TournamentRewardFinalizer; } interface AuctionFinalizer { - finalize(auctionId: number): Promise; + finalize(auctionId: number, db?: GamePrisma.TransactionClient): Promise; } interface AuctionBidder { - bid(command: Extract): Promise; + bid( + command: Extract, + db?: GamePrisma.TransactionClient + ): Promise; } interface TournamentRewardFinalizer { - finalize(command: Extract): Promise; + finalize( + command: Extract, + db?: GamePrisma.TransactionClient + ): Promise; } async function handleSetNationMeta( ctx: CommandHandlerContext, command: Extract ): Promise { - const { world, hooks } = ctx; + const { world } = ctx; const nation = world.getNationById(command.nationId); if (!nation) { return { @@ -122,7 +109,6 @@ async function handleSetNationMeta( world.updateNation(command.nationId, { meta: nextMeta, }); - await flushWorld(world, hooks); return { type: 'setNationMeta', ok: true, @@ -135,7 +121,7 @@ async function handleAdjustGeneralResources( ctx: CommandHandlerContext, command: Extract ): Promise { - const { world, hooks } = ctx; + const { world } = ctx; if (!command.adjustments || command.adjustments.length === 0) { return { type: 'adjustGeneralResources', ok: false, reason: '조정 대상이 없습니다.' }; } @@ -176,8 +162,6 @@ async function handleAdjustGeneralResources( totalGoldDelta += goldDelta; totalRiceDelta += riceDelta; } - - await flushWorld(world, hooks); return { type: 'adjustGeneralResources', ok: true, @@ -192,7 +176,7 @@ async function handleAdjustGeneralMeta( ctx: CommandHandlerContext, command: Extract ): Promise { - const { world, hooks } = ctx; + const { world } = ctx; if (!command.adjustments || command.adjustments.length === 0) { return { type: 'adjustGeneralMeta', @@ -224,8 +208,6 @@ async function handleAdjustGeneralMeta( world.updateGeneral(adjustment.generalId, { meta: nextMeta }); processed += 1; } - - await flushWorld(world, hooks); return { type: 'adjustGeneralMeta', ok: true, @@ -238,7 +220,7 @@ async function handleTournamentMatchResult( ctx: CommandHandlerContext, command: Extract ): Promise { - const { world, hooks } = ctx; + const { world } = ctx; const resolvePrefix = (type: number): string => { switch (type) { case 1: @@ -342,8 +324,6 @@ async function handleTournamentMatchResult( nextMeta[rankKey('g')] = defenderG + defenderGDelta; world.updateGeneral(defender.id, { meta: nextMeta }); } - - await flushWorld(world, hooks); return { type: 'tournamentMatchResult', ok: true, @@ -358,7 +338,7 @@ async function handlePatchGeneral( ctx: CommandHandlerContext, command: Extract ): Promise { - const { world, hooks } = ctx; + const { world } = ctx; const general = world.getGeneralById(command.generalId); if (!general) { return { @@ -396,7 +376,6 @@ async function handlePatchGeneral( } world.updateGeneral(command.generalId, patch); - await flushWorld(world, hooks); return { type: 'patchGeneral', ok: true, generalId: command.generalId }; } @@ -404,7 +383,7 @@ async function handleTroopJoin( ctx: CommandHandlerContext, command: Extract ): Promise { - const { world, hooks } = ctx; + const { world } = ctx; const general = world.getGeneralById(command.generalId); if (!general) { return { @@ -448,7 +427,6 @@ async function handleTroopJoin( world.updateGeneral(command.generalId, { troopId: command.troopId, }); - await flushWorld(world, hooks); return { type: 'troopJoin', ok: true, @@ -461,7 +439,7 @@ async function handleTroopExit( ctx: CommandHandlerContext, command: Extract ): Promise { - const { world, hooks } = ctx; + const { world } = ctx; const general = world.getGeneralById(command.generalId); if (!general) { return { @@ -484,7 +462,6 @@ async function handleTroopExit( world.updateGeneral(command.generalId, { troopId: 0, }); - await flushWorld(world, hooks); return { type: 'troopExit', ok: true, @@ -499,7 +476,6 @@ async function handleTroopExit( world.updateGeneral(member.id, { troopId: 0 }); } world.removeTroop(troopId); - await flushWorld(world, hooks); return { type: 'troopExit', ok: true, @@ -512,7 +488,7 @@ async function handleDieOnPrestart( ctx: CommandHandlerContext, command: Extract ): Promise { - const { world, hooks } = ctx; + const { world } = ctx; const general = world.getGeneralById(command.generalId); if (!general) { return { @@ -533,7 +509,6 @@ async function handleDieOnPrestart( } world.removeGeneral(command.generalId); - await flushWorld(world, hooks); return { type: 'dieOnPrestart', ok: true, generalId: command.generalId }; } @@ -619,7 +594,7 @@ async function handleSetMySetting( ctx: CommandHandlerContext, command: Extract ): Promise { - const { world, hooks } = ctx; + const { world } = ctx; const general = world.getGeneralById(command.generalId); if (!general) { return { @@ -635,7 +610,6 @@ async function handleSetMySetting( ...command.settings, }, }); - await flushWorld(world, hooks); return { type: 'setMySetting', ok: true, generalId: command.generalId }; } @@ -643,7 +617,7 @@ async function handleDropItem( ctx: CommandHandlerContext, command: Extract ): Promise { - const { world, hooks } = ctx; + const { world } = ctx; const general = world.getGeneralById(command.generalId); if (!general) { return { type: 'dropItem', ok: false, generalId: command.generalId, reason: '장수 정보를 찾을 수 없습니다.' }; @@ -664,7 +638,6 @@ async function handleDropItem( items, }, }); - await flushWorld(world, hooks); return { type: 'dropItem', ok: true, generalId: command.generalId }; } @@ -680,7 +653,7 @@ async function handleAuctionFinalize( reason: '경매 확정기가 준비되지 않았습니다.', }; } - return ctx.auctionFinalizer.finalize(command.auctionId); + return ctx.auctionFinalizer.finalize(command.auctionId, ctx.commandDb); } async function handleAuctionBid( @@ -695,14 +668,14 @@ async function handleAuctionBid( reason: '경매 입찰기가 준비되지 않았습니다.', }; } - return ctx.auctionBidder.bid(command); + return ctx.auctionBidder.bid(command, ctx.commandDb); } async function handleChangePermission( ctx: CommandHandlerContext, command: Extract ): Promise { - const { world, hooks } = ctx; + const { world } = ctx; const general = world.getGeneralById(command.generalId); if (!general) { return { @@ -728,8 +701,6 @@ async function handleChangePermission( }); } } - - await flushWorld(world, hooks); return { type: 'changePermission', ok: true, generalId: command.generalId }; } @@ -737,7 +708,7 @@ async function handleKick( ctx: CommandHandlerContext, command: Extract ): Promise { - const { world, hooks } = ctx; + const { world } = ctx; const general = world.getGeneralById(command.generalId); if (!general) { return { type: 'kick', ok: false, generalId: command.generalId, reason: '장수 정보를 찾을 수 없습니다.' }; @@ -761,8 +732,6 @@ async function handleKick( nationId: 0, officerLevel: 0, }); - - await flushWorld(world, hooks); return { type: 'kick', ok: true, generalId: command.generalId }; } @@ -770,7 +739,7 @@ async function handleAppoint( ctx: CommandHandlerContext, command: Extract ): Promise { - const { world, hooks } = ctx; + const { world } = ctx; const general = world.getGeneralById(command.generalId); if (!general) { return { type: 'appoint', ok: false, generalId: command.generalId, reason: '장수 정보를 찾을 수 없습니다.' }; @@ -825,8 +794,6 @@ async function handleAppoint( }); } } - - await flushWorld(world, hooks); return { type: 'appoint', ok: true, generalId: command.generalId }; } @@ -834,7 +801,7 @@ async function handleTournamentRefund( ctx: CommandHandlerContext, command: Extract ): Promise { - const { world, hooks } = ctx; + const { world } = ctx; if (!command.refunds || command.refunds.length === 0) { return { type: 'tournamentRefund', @@ -866,8 +833,6 @@ async function handleTournamentRefund( processed += 1; totalRefund += refund.amount; } - - await flushWorld(world, hooks); return { type: 'tournamentRefund', ok: true, @@ -882,7 +847,7 @@ async function handleTournamentBettingPayout( ctx: CommandHandlerContext, command: Extract ): Promise { - const { world, hooks } = ctx; + const { world } = ctx; if (!command.payouts || command.payouts.length === 0) { return { type: 'tournamentBettingPayout', @@ -935,8 +900,6 @@ async function handleTournamentBettingPayout( nextMeta[betwingoldKey] = currentBetwingold + delta.betwingold; world.updateGeneral(generalId, { meta: nextMeta }); } - - await flushWorld(world, hooks); return { type: 'tournamentBettingPayout', ok: true, @@ -960,7 +923,7 @@ async function handleTournamentReward( reason: '보상 처리기가 준비되지 않았습니다.', }; } - return ctx.tournamentRewardFinalizer.finalize(command); + return ctx.tournamentRewardFinalizer.finalize(command, ctx.commandDb); } // 설문 보상은 API에서 전달된 RNG 결과를 재검증한 뒤 월드에 반영한다. @@ -968,7 +931,7 @@ async function handleVoteReward( ctx: CommandHandlerContext, command: Extract ): Promise { - const { world, hooks } = ctx; + const { world } = ctx; const general = world.getGeneralById(command.generalId); if (!general) { return { @@ -1153,7 +1116,6 @@ async function handleVoteReward( } world.updateGeneral(command.generalId, patch); - await flushWorld(world, hooks); return { type: 'voteReward', ok: true, @@ -1166,54 +1128,81 @@ async function handleVoteReward( export const createTurnDaemonCommandHandler = (options: { world: InMemoryTurnWorld; - hooks?: TurnDaemonHooks; auctionFinalizer?: AuctionFinalizer; auctionBidder?: AuctionBidder; tournamentRewardFinalizer?: TournamentRewardFinalizer; }): TurnDaemonCommandHandler => { - const ctx = { + const ctx: CommandHandlerContext = { world: options.world, - hooks: options.hooks, auctionFinalizer: options.auctionFinalizer, auctionBidder: options.auctionBidder, tournamentRewardFinalizer: options.tournamentRewardFinalizer, }; - type HandlerMap = Partial Promise>>; + type HandlerMap = Partial< + Record Promise> + >; const handlers: HandlerMap = { troopJoin: (command) => handleTroopJoin(ctx, command as Extract), troopExit: (command) => handleTroopExit(ctx, command as Extract), - dieOnPrestart: (command) => handleDieOnPrestart(ctx, command as Extract), - buildNationCandidate: (command) => handleBuildNationCandidate(ctx, command as Extract), - instantRetreat: (command) => handleInstantRetreat(ctx, command as Extract), + dieOnPrestart: (command) => + handleDieOnPrestart(ctx, command as Extract), + buildNationCandidate: (command) => + handleBuildNationCandidate(ctx, command as Extract), + instantRetreat: (command) => + handleInstantRetreat(ctx, command as Extract), vacation: (command) => handleVacation(ctx, command as Extract), - setMySetting: (command) => handleSetMySetting(ctx, command as Extract), + setMySetting: (command) => + handleSetMySetting(ctx, command as Extract), dropItem: (command) => handleDropItem(ctx, command as Extract), - auctionFinalize: (command) => handleAuctionFinalize(ctx, command as Extract), + auctionFinalize: (command) => + handleAuctionFinalize(ctx, command as Extract), auctionBid: (command) => handleAuctionBid(ctx, command as Extract), - changePermission: (command) => handleChangePermission(ctx, command as Extract), + changePermission: (command) => + handleChangePermission(ctx, command as Extract), kick: (command) => handleKick(ctx, command as Extract), appoint: (command) => handleAppoint(ctx, command as Extract), - tournamentRefund: (command) => handleTournamentRefund(ctx, command as Extract), - tournamentBettingPayout: (command) => handleTournamentBettingPayout(ctx, command as Extract), - tournamentReward: (command) => handleTournamentReward(ctx, command as Extract), + tournamentRefund: (command) => + handleTournamentRefund(ctx, command as Extract), + tournamentBettingPayout: (command) => + handleTournamentBettingPayout( + ctx, + command as Extract + ), + tournamentReward: (command) => + handleTournamentReward(ctx, command as Extract), voteReward: (command) => handleVoteReward(ctx, command as Extract), - setNationMeta: (command) => handleSetNationMeta(ctx, command as Extract), - adjustGeneralResources: (command) => handleAdjustGeneralResources(ctx, command as Extract), - adjustGeneralMeta: (command) => handleAdjustGeneralMeta(ctx, command as Extract), + setNationMeta: (command) => + handleSetNationMeta(ctx, command as Extract), + adjustGeneralResources: (command) => + handleAdjustGeneralResources( + ctx, + command as Extract + ), + adjustGeneralMeta: (command) => + handleAdjustGeneralMeta(ctx, command as Extract), tournamentMatchResult: (command) => handleTournamentMatchResult(ctx, command as Extract), - patchGeneral: (command) => handlePatchGeneral(ctx, command as Extract), + patchGeneral: (command) => + handlePatchGeneral(ctx, command as Extract), }; return { - handle: async (command): Promise => { + handle: async ( + command, + executionContext?: TurnDaemonCommandExecutionContext + ): Promise => { const handler = handlers[command.type]; if (!handler) { return null; } - return handler(command); + ctx.commandDb = executionContext?.db; + try { + return await handler(command); + } finally { + ctx.commandDb = undefined; + } }, }; }; diff --git a/app/game-engine/test/databaseCommandQueue.integration.test.ts b/app/game-engine/test/databaseCommandQueue.integration.test.ts new file mode 100644 index 0000000..9912dd2 --- /dev/null +++ b/app/game-engine/test/databaseCommandQueue.integration.test.ts @@ -0,0 +1,95 @@ +import { afterAll, beforeAll, describe, expect, it } from 'vitest'; +import { createGamePostgresConnector, GamePrisma, type GamePrismaClient } from '@sammo-ts/infra'; + +import { DatabaseTurnDaemonCommandQueue } from '../src/lifecycle/databaseCommandQueue.js'; + +const databaseUrl = process.env.INPUT_EVENT_DATABASE_URL; +const integration = describe.skipIf(!databaseUrl); + +integration('database command queue', () => { + let close: (() => Promise) | undefined; + let db: GamePrismaClient; + + beforeAll(async () => { + const connector = createGamePostgresConnector({ url: databaseUrl! }); + await connector.connect(); + db = connector.prisma; + close = () => connector.disconnect(); + await db.inputEvent.deleteMany({ + where: { requestId: { startsWith: 'integration:engine:' } }, + }); + }); + + afterAll(async () => { + await db.inputEvent.deleteMany({ + where: { requestId: { startsWith: 'integration:engine:' } }, + }); + await close?.(); + }); + + it('claims one durable event only once across concurrent consumers and persists its result', async () => { + const requestId = 'integration:engine:claim-once'; + await db.inputEvent.create({ + data: { + requestId, + target: 'ENGINE', + eventType: 'vacation', + payload: { type: 'vacation', requestId, generalId: 7 } as GamePrisma.InputJsonValue, + }, + }); + + const first = new DatabaseTurnDaemonCommandQueue(db); + const second = new DatabaseTurnDaemonCommandQueue(db); + const [firstCommands, secondCommands] = await Promise.all([first.drain(), second.drain()]); + const commands = firstCommands.concat(secondCommands); + + expect(commands).toEqual([{ type: 'vacation', requestId, generalId: 7 }]); + await first.publishCommandResult(requestId, { type: 'vacation', ok: true, generalId: 7 }); + + const stored = await db.inputEvent.findUniqueOrThrow({ where: { requestId } }); + expect(stored).toMatchObject({ + status: 'SUCCEEDED', + attempts: 1, + result: { type: 'vacation', ok: true, generalId: 7 }, + }); + }); + + it('recovers only an expired processing lease', async () => { + const expiredId = 'integration:engine:expired'; + const activeId = 'integration:engine:active'; + await db.inputEvent.createMany({ + data: [ + { + requestId: expiredId, + target: 'ENGINE', + eventType: 'vacation', + payload: { type: 'vacation', requestId: expiredId, generalId: 8 } as GamePrisma.InputJsonValue, + status: 'PROCESSING', + processingAt: new Date(Date.now() - 120_000), + lockedBy: 'dead-worker', + leaseUntil: new Date(Date.now() - 60_000), + }, + { + requestId: activeId, + target: 'ENGINE', + eventType: 'vacation', + payload: { type: 'vacation', requestId: activeId, generalId: 9 } as GamePrisma.InputJsonValue, + status: 'PROCESSING', + processingAt: new Date(), + lockedBy: 'active-worker', + leaseUntil: new Date(Date.now() + 60_000), + }, + ], + }); + + const queue = new DatabaseTurnDaemonCommandQueue(db); + await queue.initialize(); + const commands = await queue.drain(); + + expect(commands).toEqual([{ type: 'vacation', requestId: expiredId, generalId: 8 }]); + expect(await db.inputEvent.findUniqueOrThrow({ where: { requestId: activeId } })).toMatchObject({ + status: 'PROCESSING', + lockedBy: 'active-worker', + }); + }); +}); diff --git a/app/game-engine/test/inputEventAtomicity.test.ts b/app/game-engine/test/inputEventAtomicity.test.ts new file mode 100644 index 0000000..711675b --- /dev/null +++ b/app/game-engine/test/inputEventAtomicity.test.ts @@ -0,0 +1,285 @@ +import { describe, expect, it, vi } from 'vitest'; + +import { + InMemoryControlQueue, + ManualClock, + TurnDaemonLifecycle, + type TurnDaemonCommandResult, + type TurnProcessor, + type TurnStateStore, +} from '../src/index.js'; +import { InMemoryReservedTurnStore } from '../src/turn/reservedTurnStore.js'; + +const createStateStore = (): TurnStateStore => ({ + loadLastTurnTime: async () => new Date('2026-01-01T00:00:00.000Z'), + loadNextGeneralTurnTime: async () => null, + saveLastTurnTime: async () => {}, + loadCheckpoint: async () => undefined, + saveCheckpoint: async () => {}, +}); + +const processor: TurnProcessor = { + run: async () => { + throw new Error('scheduled turn must not run in this test'); + }, +}; + +describe('input event atomicity', () => { + it('keeps reserved-turn dirty state when persistence fails', async () => { + let failCreate = true; + const prisma = { + generalTurn: { + findMany: vi.fn(async () => []), + deleteMany: vi.fn(async () => ({ count: 0 })), + createMany: vi.fn(async () => { + if (failCreate) { + throw new Error('injected write failure'); + } + return { count: 1 }; + }), + }, + nationTurn: { + findMany: vi.fn(async () => []), + deleteMany: vi.fn(async () => ({ count: 0 })), + createMany: vi.fn(async () => ({ count: 0 })), + }, + }; + const store = new InMemoryReservedTurnStore(prisma, { + maxGeneralTurns: 1, + maxNationTurns: 1, + }); + store.shiftGeneralTurns(7, -1); + + await expect(store.flushChanges()).rejects.toThrow('injected write failure'); + expect(store.peekDirtyState()).toEqual({ generalIds: [7], nationKeys: [] }); + + failCreate = false; + await store.flushChanges(); + expect(store.peekDirtyState()).toEqual({ generalIds: [], nationKeys: [] }); + }); + + it('dispatches registry mutations that the old lifecycle switch dropped, then commits before responding', async () => { + const queue = new InMemoryControlQueue(); + const order: string[] = []; + const result: TurnDaemonCommandResult = { + type: 'auctionBid', + ok: true, + auctionId: 3, + closeAt: '2026-01-01T00:10:00.000Z', + }; + let resolveResponse: (() => void) | undefined; + const responded = new Promise((resolve) => { + resolveResponse = resolve; + }); + const lifecycle = new TurnDaemonLifecycle( + { + clock: new ManualClock(new Date('2026-01-01T00:00:00.000Z').getTime()), + controlQueue: queue, + getNextTickTime: () => new Date('2026-01-01T01:00:00.000Z'), + stateStore: createStateStore(), + processor, + commandHandler: { + handle: async (command) => { + order.push(`handle:${command.type}`); + return result; + }, + }, + hooks: { + commitCommand: async (requestId, committedResult) => { + order.push(`commit:${requestId}:${committedResult.type}`); + }, + }, + commandResponder: { + publishStatus: async () => {}, + publishCommandResult: async (requestId, response) => { + order.push(`respond:${requestId}:${response.type}`); + resolveResponse?.(); + }, + }, + }, + { + profile: 'test', + defaultBudget: { budgetMs: 100, maxGenerals: 1, catchUpCap: 1 }, + } + ); + + queue.enqueue({ + type: 'auctionBid', + requestId: 'event-1', + auctionId: 3, + generalId: 7, + amount: 1000, + }); + const loop = lifecycle.start(); + await responded; + + expect(order).toEqual(['handle:auctionBid', 'commit:event-1:auctionBid', 'respond:event-1:auctionBid']); + + await lifecycle.stop('done'); + await loop; + }); + + it('runs a mutation inside the database-owned execution boundary before responding', async () => { + const queue = new InMemoryControlQueue(); + const order: string[] = []; + let resolveResponse: (() => void) | undefined; + const responded = new Promise((resolve) => { + resolveResponse = resolve; + }); + const lifecycle = new TurnDaemonLifecycle( + { + clock: new ManualClock(new Date('2026-01-01T00:00:00.000Z').getTime()), + controlQueue: queue, + getNextTickTime: () => new Date('2026-01-01T01:00:00.000Z'), + stateStore: createStateStore(), + processor, + commandHandler: { + handle: async (command) => { + order.push(`handle:${command.type}`); + return { type: 'vacation', ok: true, generalId: 7 }; + }, + }, + hooks: { + executeCommand: async (requestId, execute) => { + order.push(`transaction:${requestId}:begin`); + const result = await execute({}); + order.push(`transaction:${requestId}:commit`); + return result; + }, + commitCommand: async () => { + order.push('legacy-commit'); + }, + }, + commandResponder: { + publishStatus: async () => {}, + publishCommandResult: async () => { + order.push('respond'); + resolveResponse?.(); + }, + }, + }, + { + profile: 'test', + defaultBudget: { budgetMs: 100, maxGenerals: 1, catchUpCap: 1 }, + } + ); + + queue.enqueue({ type: 'vacation', requestId: 'event-uow', generalId: 7 }); + const loop = lifecycle.start(); + await responded; + + expect(order).toEqual([ + 'transaction:event-uow:begin', + 'handle:vacation', + 'transaction:event-uow:commit', + 'respond', + ]); + + await lifecycle.stop('done'); + await loop; + }); + + it('pauses without publishing success when the atomic command commit fails', async () => { + const queue = new InMemoryControlQueue(); + let resolveError: (() => void) | undefined; + const errorObserved = new Promise((resolve) => { + resolveError = resolve; + }); + const publishCommandResult = vi.fn(async () => {}); + const lifecycle = new TurnDaemonLifecycle( + { + clock: new ManualClock(new Date('2026-01-01T00:00:00.000Z').getTime()), + controlQueue: queue, + getNextTickTime: () => new Date('2026-01-01T01:00:00.000Z'), + stateStore: createStateStore(), + processor, + commandHandler: { + handle: async () => ({ type: 'vacation', ok: true, generalId: 7 }), + }, + hooks: { + commitCommand: async () => { + throw new Error('injected commit failure'); + }, + onRunError: async () => { + resolveError?.(); + }, + }, + commandResponder: { + publishStatus: async () => {}, + publishCommandResult, + }, + }, + { + profile: 'test', + defaultBudget: { budgetMs: 100, maxGenerals: 1, catchUpCap: 1 }, + } + ); + + queue.enqueue({ type: 'vacation', requestId: 'event-2', generalId: 7 }); + const loop = lifecycle.start(); + await errorObserved; + + expect(lifecycle.getStatus()).toMatchObject({ + state: 'paused', + paused: true, + lastError: 'injected commit failure', + }); + expect(publishCommandResult).not.toHaveBeenCalled(); + + await lifecycle.stop('done'); + await loop; + }); + + it('pauses without committing or acknowledging a handler that throws after changing memory', async () => { + const queue = new InMemoryControlQueue(); + let resolveError: (() => void) | undefined; + const errorObserved = new Promise((resolve) => { + resolveError = resolve; + }); + const commitCommand = vi.fn(async () => {}); + const publishCommandResult = vi.fn(async () => {}); + const lifecycle = new TurnDaemonLifecycle( + { + clock: new ManualClock(new Date('2026-01-01T00:00:00.000Z').getTime()), + controlQueue: queue, + getNextTickTime: () => new Date('2026-01-01T01:00:00.000Z'), + stateStore: createStateStore(), + processor, + commandHandler: { + handle: async () => { + throw new Error('injected handler failure'); + }, + }, + hooks: { + commitCommand, + onRunError: async () => { + resolveError?.(); + }, + }, + commandResponder: { + publishStatus: async () => {}, + publishCommandResult, + }, + }, + { + profile: 'test', + defaultBudget: { budgetMs: 100, maxGenerals: 1, catchUpCap: 1 }, + } + ); + + queue.enqueue({ type: 'vacation', requestId: 'event-3', generalId: 7 }); + const loop = lifecycle.start(); + await errorObserved; + + expect(lifecycle.getStatus()).toMatchObject({ + state: 'paused', + paused: true, + lastError: 'injected handler failure', + }); + expect(commitCommand).not.toHaveBeenCalled(); + expect(publishCommandResult).not.toHaveBeenCalled(); + + await lifecycle.stop('done'); + await loop; + }); +}); diff --git a/docs/architecture/runtime.md b/docs/architecture/runtime.md index 8fcf9c5..ea760a9 100644 --- a/docs/architecture/runtime.md +++ b/docs/architecture/runtime.md @@ -82,32 +82,37 @@ Gateway runs a lightweight cron loop (setInterval) that: ## Current Implementation Status - Turn daemon lifecycle + in-memory state live in `app/game-engine` with DB flush hooks. -- Redis transport for daemon control is implemented and wired into the daemon. -- API server already exposes turn-daemon commands (run/pause/resume/status) via tRPC, communicating via Redis Streams. -- API server writes reserved turns and messages directly to the DB; daemon focuses on world state/logs. +- 모든 tRPC mutation은 PostgreSQL `input_event` 경계에서 request ID와 처리 + 상태를 기록한다. +- 월드 mutation과 daemon control은 PostgreSQL inbox로 전달된다. Redis는 + 더 이상 이 명령들의 영속 큐가 아니며 realtime/battle-sim 전송에만 남아 + 있다. +- API가 직접 변경하는 예약 턴·메시지 등의 DB 쓰기는 API input event 완료와 + 같은 transaction에서 commit된다. +- 엔진 명령은 도메인 DB 쓰기, in-memory world flush, 예약 턴, 로그, 결과 + 저장과 inbox 완료를 하나의 transaction으로 commit한다. ## Turn Daemon and API Server Behavior (Outline) - Turn daemon responsibilities: scheduling, turn resolution, state persistence - API server responsibilities: query/command intake, validation, response shaping - Concurrency model between daemon and API server -- Communication channel: Redis Stream or Redis pub/sub +- Durable command channel: PostgreSQL `input_event` - Client updates: SSE between API server and frontend where appropriate -## Redis Communication Recommendation (Draft) +## Durable Input Event Split -Use Redis Streams for daemon control and mutation requests, and Redis pub/sub -for transient fan-out events. Streams provide durability, backpressure, and -replay while pub/sub keeps live updates simple and low-latency. +PostgreSQL is the source of truth for accepted input. Redis pub/sub remains a +best-effort notification/fan-out path and must not decide whether a gameplay +mutation was committed. ### Recommended Split -- Redis Streams: - - API server -> daemon: mutation requests, turn-run commands. - - Daemon -> API server: run status events, job results, error reports. - - Use consumer groups for daemon workers and API server listeners. - - Require `requestId` for correlation and idempotency. - - Ack on success; move failed items to a dead-letter stream after retry. +- PostgreSQL `input_event`: + - API mutation acceptance, idempotency, attempts and audit state. + - API server -> daemon mutation/control payloads and durable results. + - `FOR UPDATE SKIP LOCKED` claim plus worker lease for concurrent consumers. + - engine result and gameplay state commit in the same transaction. - Redis pub/sub: - Daemon -> API server: low-stakes live update signals (run started/ended). - API server -> frontend: SSE fan-out triggered by pub/sub updates. @@ -115,11 +120,11 @@ replay while pub/sub keeps live updates simple and low-latency. ### Operational Notes -- Stream keys should be namespaced per server+scenario profile. -- Use bounded stream length (`MAXLEN`) to cap storage. -- API server should guard against duplicate processing by `requestId`. -- When the daemon is busy, API queues new mutations to stream and responds - with an accepted status to clients. +- 각 profile은 별도 game DB schema/connection을 사용한다. +- HTTP `Idempotency-Key`(없으면 Fastify request ID)와 tRPC path를 합친 값이 + API input event key다. +- 처리 중 lease가 만료된 engine event만 `PENDING`으로 회수한다. +- 완료 event 보존/정리 기간과 `FAILED` 재처리 운영 정책은 별도로 정해야 한다. Detailed lifecycle and control flow are defined in `docs/architecture/turn-daemon-lifecycle.md`. @@ -167,13 +172,14 @@ also supports local ID/password login for users who cannot use Kakao. immediately even if requests remain queued. - While the daemon is resolving a turn, the API server queues incoming requests. -Note: the current implementation does not yet process API mutation requests -between turns; only control commands are handled by the in-process queue. +현재 구현은 control 명령과 registry의 모든 world mutation을 같은 DB queue로 +읽어 턴 사이에 처리한다. ### Daemon Control Contract (Draft) -API server commands are delivered to the daemon over the control channel -(Redis Stream or in-process). The daemon replies with status and run events. +API server commands are inserted into PostgreSQL `input_event`; the daemon +stores status/results on the same row. An in-process queue remains available +for tests and internal signals. ```ts export type RunReason = 'schedule' | 'manual' | 'poke'; @@ -193,19 +199,17 @@ export type DaemonEvent = ### API Server Flow -- The API server validates queries/commands and writes them to Redis Streams - or Redis pub/sub. +- The API server validates commands and records every tRPC mutation in + PostgreSQL. - After a request is processed, the API server returns the result to clients. - Read-only queries may access the DBMS directly. - The API server may use SSE to stream live updates to the frontend. ### Queue and Rate Limits -- API server requests are delivered to the daemon via Redis Streams or - Redis pub/sub. -- Redis Stream mutation requests are rate-limited per user. - - Each user can have up to 30 pending mutation requests. - - Additional requests are rejected once the limit is exceeded. +- Engine-owned requests are delivered through the PostgreSQL inbox. +- Per-user pending limits and event retention remain operational follow-up + work; idempotency and exclusive claim are implemented. ### In-Memory and DBMS Flush diff --git a/docs/architecture/todo.md b/docs/architecture/todo.md index 2e36adf..f3864b0 100644 --- a/docs/architecture/todo.md +++ b/docs/architecture/todo.md @@ -8,14 +8,14 @@ Move items into the main docs once they are finalized. - [AI suggestion] Make profile turn execution a durable single-writer contract: add a DB-backed lease with fencing epoch/world version, and reject stale daemon commits after a restart or split-brain. -- [AI suggestion] Replace the current destructive dirty-state drain and - multi-query flush with bounded atomic batches. Persist entity deltas, logs, - reserved turns, checkpoint, and outbox events in one transaction, then clear - dirty state only after commit. -- [AI suggestion] Introduce durable command inbox/outbox records with unique - request IDs, idempotent results, and per-profile ordering. Keep Redis as a - wake-up/fan-out layer, or add consumer groups, ACK/pending recovery, DLQ, and - stream trimming if Redis Streams remain the command source. +- [AI suggestion] Bound very large world flush batches and add a transactional + realtime outbox. Entity deltas, logs, reserved turns, checkpoint and input + results are atomic now, but realtime publication remains post-commit + best-effort. +- [AI suggestion] Define `input_event` retention, failed-event inspection/DLQ, + manual replay authorization and per-user pending limits. Durable request IDs, + idempotent results, ordering, exclusive claim and expired-lease recovery are + implemented. - [AI suggestion] Add ownership authorization to all turn reservation procedures (`auth user -> owned general -> current officer role -> nation`) and add revision/optimistic concurrency to reservation queue updates. diff --git a/docs/architecture/turn-daemon-lifecycle.md b/docs/architecture/turn-daemon-lifecycle.md index 683ab1a..6273adc 100644 --- a/docs/architecture/turn-daemon-lifecycle.md +++ b/docs/architecture/turn-daemon-lifecycle.md @@ -8,14 +8,15 @@ from the API server. - One daemon process per server+scenario profile. - API server is the only ingress for user/admin requests. -- The daemon owns world-state mutations; the API server mutates reserved turns and messages. +- The daemon owns world-state mutations; the API server mutates reserved turns + and messages inside the common API input-event transaction. ## Current Implementation Status -- The daemon loop handles scheduled runs plus Redis-based control queue commands +- The daemon loop handles scheduled runs plus PostgreSQL inbox control commands (run/pause/resume/shutdown/getStatus). - API mutation requests (troopJoin, vacation, etc.) are drained and handled by the daemon loop between turns. -- `getStatus` is fully supported and responds via Redis. +- `getStatus` and command results are stored durably on their input-event row. - API server currently writes reserved turns and messages directly to the DB. ## Responsibilities @@ -47,8 +48,7 @@ Idle -> Running -> Flushing -> Idle The daemon interleaves API request handling between scheduled turn executions. API requests are drained until the next turn time is reached; once the turn starts, incoming requests are queued and processed after the run. -Current implementation does not yet drain API requests between turns; this -section is the intended target behavior. +The current implementation follows this interleaving model. ```ts while (!stopping) { @@ -101,7 +101,7 @@ Minimum checkpoint data: ## API Server Interaction -- API server enqueues mutations and pokes the daemon. +- API server inserts mutations and pokes into the PostgreSQL inbox. - Read-only queries can read from DBMS or a read model. - During `running`, incoming requests are queued and processed after the run. @@ -171,7 +171,8 @@ Admin controls should toggle `paused`, trigger manual run, and request catch-up. ### Daemon Control Contract (Draft) -API server to daemon control messages (Redis Stream or in-process channel): +API server to daemon control payloads (PostgreSQL inbox or in-process test +channel): ```ts export type RunReason = 'schedule' | 'manual' | 'poke'; @@ -192,9 +193,33 @@ export type DaemonEvent = ## Recovery and Shutdown - On startup, load `game_env.turntime` and catch up to `now`. +- Claim pending input events with `FOR UPDATE SKIP LOCKED`. A worker ID and + expiry timestamp prevent another live consumer from claiming the same event; + only expired processing leases return to `PENDING`. +- A handler exception or transaction failure pauses the daemon without + acknowledging the event. Restart/recovery reloads world state before retry. - On shutdown, stop accepting new triggers, finish the current run, flush, then exit cleanly. +## Input Event Commit Boundary + +For engine-owned mutations the lifecycle opens one database-owned unit of work: + +```text +claim input_event + -> execute command against in-memory world + -> specialized domain DB writes (auction/tournament) + -> persist world/reserved turns/logs + -> store command result and mark input_event SUCCEEDED + -> commit + -> acknowledge in-memory dirty sets + -> respond +``` + +If any step before commit fails, PostgreSQL rolls back and dirty sets are not +cleared. Realtime publication happens after commit and cannot roll gameplay +state back. + ## Testing with a Controlled Clock - Turn daemon tests should use a controllable `Clock` implementation (예: `ManualClock`) to diff --git a/packages/common/src/turnDaemon/types.ts b/packages/common/src/turnDaemon/types.ts index c531a75..dd6a3a5 100644 --- a/packages/common/src/turnDaemon/types.ts +++ b/packages/common/src/turnDaemon/types.ts @@ -43,13 +43,14 @@ export interface TurnDaemonStatus { export type TurnDaemonCommand = | { type: 'run'; + requestId?: string; reason: RunReason; targetTime?: string; budget?: TurnRunBudget; } - | { type: 'pause'; reason?: string } - | { type: 'resume'; reason?: string } - | { type: 'shutdown'; reason?: string } + | { type: 'pause'; requestId?: string; reason?: string } + | { type: 'resume'; requestId?: string; reason?: string } + | { type: 'shutdown'; requestId?: string; reason?: string } | { type: 'getStatus'; requestId?: string } | { type: 'troopJoin'; requestId?: string; generalId: number; troopId: number } | { type: 'troopExit'; requestId?: string; generalId: number } @@ -186,6 +187,12 @@ export type TurnDaemonCommand = }; export type TurnDaemonCommandResult = + | { + type: 'commandRejected'; + ok: false; + commandType: TurnDaemonCommand['type']; + reason: string; + } | { type: 'auctionFinalize'; ok: true; diff --git a/packages/infra/prisma/game.prisma b/packages/infra/prisma/game.prisma index da7ef21..c718f89 100644 --- a/packages/infra/prisma/game.prisma +++ b/packages/infra/prisma/game.prisma @@ -44,6 +44,39 @@ enum DiplomacyLetterState { REPLACED } +enum InputEventStatus { + PENDING + PROCESSING + SUCCEEDED + FAILED +} + +enum InputEventTarget { + API + ENGINE +} + +model InputEvent { + sequence BigInt @id @default(autoincrement()) + requestId String @unique @map("request_id") + target InputEventTarget + eventType String @map("event_type") + payload Json @default(dbgenerated("'{}'::jsonb")) + actorUserId String? @map("actor_user_id") + status InputEventStatus @default(PENDING) + result Json? + error String? + attempts Int @default(0) + lockedBy String? @map("locked_by") + leaseUntil DateTime? @map("lease_until") + createdAt DateTime @default(now()) @map("created_at") + processingAt DateTime? @map("processing_at") + completedAt DateTime? @map("completed_at") + + @@index([target, status, sequence]) + @@map("input_event") +} + model WorldState { id Int @id @default(autoincrement()) scenarioCode String @map("scenario_code") diff --git a/packages/infra/prisma/migrations/20260725000000_add_input_event/migration.sql b/packages/infra/prisma/migrations/20260725000000_add_input_event/migration.sql new file mode 100644 index 0000000..b54b546 --- /dev/null +++ b/packages/infra/prisma/migrations/20260725000000_add_input_event/migration.sql @@ -0,0 +1,25 @@ +CREATE TYPE "InputEventStatus" AS ENUM ('PENDING', 'PROCESSING', 'SUCCEEDED', 'FAILED'); +CREATE TYPE "InputEventTarget" AS ENUM ('API', 'ENGINE'); + +CREATE TABLE "input_event" ( + "sequence" BIGSERIAL NOT NULL, + "request_id" TEXT NOT NULL, + "target" "InputEventTarget" NOT NULL, + "event_type" TEXT NOT NULL, + "payload" JSONB NOT NULL DEFAULT '{}'::jsonb, + "actor_user_id" TEXT, + "status" "InputEventStatus" NOT NULL DEFAULT 'PENDING', + "result" JSONB, + "error" TEXT, + "attempts" INTEGER NOT NULL DEFAULT 0, + "locked_by" TEXT, + "lease_until" TIMESTAMP(3), + "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "processing_at" TIMESTAMP(3), + "completed_at" TIMESTAMP(3), + + CONSTRAINT "input_event_pkey" PRIMARY KEY ("sequence") +); + +CREATE UNIQUE INDEX "input_event_request_id_key" ON "input_event"("request_id"); +CREATE INDEX "input_event_target_status_sequence_idx" ON "input_event"("target", "status", "sequence"); diff --git a/packages/infra/src/db.ts b/packages/infra/src/db.ts index 3e55252..d8aa337 100644 --- a/packages/infra/src/db.ts +++ b/packages/infra/src/db.ts @@ -25,4 +25,5 @@ export interface DatabaseClient { inheritanceLog: GamePrisma.InheritanceLogDelegate; inheritanceResult: GamePrisma.InheritanceResultDelegate; inheritanceUserState: GamePrisma.InheritanceUserStateDelegate; + inputEvent: GamePrisma.InputEventDelegate; }