diff --git a/app/game-engine/src/index.ts b/app/game-engine/src/index.ts index 01096d0..a8c5e98 100644 --- a/app/game-engine/src/index.ts +++ b/app/game-engine/src/index.ts @@ -6,6 +6,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/databaseTurnDaemonLease.js'; export * from './lifecycle/inMemoryControlQueue.js'; export * from './lifecycle/turnDaemonLifecycle.js'; export * from './lifecycle/getNextTickTime.js'; diff --git a/app/game-engine/src/lifecycle/databaseTurnDaemonLease.ts b/app/game-engine/src/lifecycle/databaseTurnDaemonLease.ts new file mode 100644 index 0000000..97faec6 --- /dev/null +++ b/app/game-engine/src/lifecycle/databaseTurnDaemonLease.ts @@ -0,0 +1,224 @@ +import { randomUUID } from 'node:crypto'; + +import { createGamePostgresConnector, GamePrisma, type GamePrismaClient } from '@sammo-ts/infra'; + +export interface TurnDaemonLeaseToken { + profile: string; + ownerId: string; + fencingEpoch: bigint; +} + +export interface DatabaseTurnDaemonLeaseOptions { + profile: string; + ownerId?: string; + leaseDurationMs?: number; + heartbeat?: boolean; +} + +export class TurnDaemonLeaseUnavailableError extends Error { + constructor(profile: string) { + super(`Another turn daemon holds the active lease for profile "${profile}".`); + this.name = 'TurnDaemonLeaseUnavailableError'; + } +} + +export class TurnDaemonLeaseLostError extends Error { + constructor(profile: string) { + super(`Turn daemon lease was lost for profile "${profile}".`); + this.name = 'TurnDaemonLeaseLostError'; + } +} + +type LeaseRow = { + profile: string; + owner_id: string; + fencing_epoch: bigint; +}; + +const normalizeLeaseDuration = (value?: number): number => + Math.max(1_000, Math.floor(value ?? 30_000)); + +export class DatabaseTurnDaemonLease { + private readonly db: GamePrismaClient; + private readonly disconnect: () => Promise; + private readonly profile: string; + private readonly ownerId: string; + private readonly leaseDurationMs: number; + private readonly heartbeatEnabled: boolean; + private token: TurnDaemonLeaseToken | null = null; + private heartbeatTimer: NodeJS.Timeout | null = null; + private lost = false; + + private constructor( + db: GamePrismaClient, + disconnect: () => Promise, + options: DatabaseTurnDaemonLeaseOptions + ) { + this.db = db; + this.disconnect = disconnect; + this.profile = options.profile; + this.ownerId = options.ownerId ?? randomUUID(); + this.leaseDurationMs = normalizeLeaseDuration(options.leaseDurationMs); + this.heartbeatEnabled = options.heartbeat ?? true; + } + + static async connect( + databaseUrl: string, + options: DatabaseTurnDaemonLeaseOptions + ): Promise { + const connector = createGamePostgresConnector({ url: databaseUrl }); + await connector.connect(); + return new DatabaseTurnDaemonLease(connector.prisma, () => connector.disconnect(), options); + } + + async acquire(): Promise { + const rows = await this.db.$queryRaw(GamePrisma.sql` + INSERT INTO "turn_daemon_lease" ( + "profile", + "owner_id", + "lease_until", + "fencing_epoch", + "heartbeat_at" + ) + VALUES ( + ${this.profile}, + ${this.ownerId}, + CURRENT_TIMESTAMP + (${this.leaseDurationMs} * INTERVAL '1 millisecond'), + 1, + CURRENT_TIMESTAMP + ) + ON CONFLICT ("profile") DO UPDATE + SET + "owner_id" = EXCLUDED."owner_id", + "lease_until" = EXCLUDED."lease_until", + "fencing_epoch" = CASE + WHEN "turn_daemon_lease"."owner_id" = EXCLUDED."owner_id" + THEN "turn_daemon_lease"."fencing_epoch" + ELSE "turn_daemon_lease"."fencing_epoch" + 1 + END, + "heartbeat_at" = CURRENT_TIMESTAMP + WHERE + "turn_daemon_lease"."owner_id" = EXCLUDED."owner_id" + OR "turn_daemon_lease"."lease_until" <= CURRENT_TIMESTAMP + RETURNING "profile", "owner_id", "fencing_epoch" + `); + const row = rows[0]; + if (!row) { + return null; + } + this.token = { + profile: row.profile, + ownerId: row.owner_id, + fencingEpoch: BigInt(row.fencing_epoch), + }; + this.lost = false; + if (this.heartbeatEnabled) { + this.startHeartbeat(); + } + return this.token; + } + + getToken(): TurnDaemonLeaseToken | null { + return this.token ? { ...this.token } : null; + } + + isLost(): boolean { + return this.lost; + } + + async renew(): Promise { + const token = this.token; + if (!token || this.lost) { + return false; + } + const rows = await this.db.$queryRaw(GamePrisma.sql` + UPDATE "turn_daemon_lease" + SET + "lease_until" = CURRENT_TIMESTAMP + (${this.leaseDurationMs} * INTERVAL '1 millisecond'), + "heartbeat_at" = CURRENT_TIMESTAMP + WHERE + "profile" = ${token.profile} + AND "owner_id" = ${token.ownerId} + AND "fencing_epoch" = ${token.fencingEpoch} + AND "lease_until" > CURRENT_TIMESTAMP + RETURNING "profile", "owner_id", "fencing_epoch" + `); + if (rows.length === 0) { + this.markLost(); + return false; + } + return true; + } + + async assertActive(transaction?: GamePrisma.TransactionClient): Promise { + const token = this.token; + if (!token || this.lost) { + throw new TurnDaemonLeaseLostError(this.profile); + } + const db = transaction ?? this.db; + const rows = await db.$queryRaw(GamePrisma.sql` + SELECT "profile", "owner_id", "fencing_epoch" + FROM "turn_daemon_lease" + WHERE + "profile" = ${token.profile} + AND "owner_id" = ${token.ownerId} + AND "fencing_epoch" = ${token.fencingEpoch} + AND "lease_until" > CURRENT_TIMESTAMP + FOR UPDATE + `); + if (rows.length === 0) { + this.markLost(); + throw new TurnDaemonLeaseLostError(this.profile); + } + } + + async release(): Promise { + this.stopHeartbeat(); + const token = this.token; + this.token = null; + if (!token || this.lost) { + return; + } + await this.db.$executeRaw(GamePrisma.sql` + UPDATE "turn_daemon_lease" + SET "lease_until" = CURRENT_TIMESTAMP, "heartbeat_at" = CURRENT_TIMESTAMP + WHERE + "profile" = ${token.profile} + AND "owner_id" = ${token.ownerId} + AND "fencing_epoch" = ${token.fencingEpoch} + `); + } + + async close(): Promise { + try { + await this.release(); + } finally { + await this.disconnect(); + } + } + + private startHeartbeat(): void { + if (this.heartbeatTimer) { + return; + } + const intervalMs = Math.max(250, Math.floor(this.leaseDurationMs / 3)); + this.heartbeatTimer = setInterval(() => { + void this.renew().catch(() => { + this.markLost(); + }); + }, intervalMs); + this.heartbeatTimer.unref(); + } + + private stopHeartbeat(): void { + if (this.heartbeatTimer) { + clearInterval(this.heartbeatTimer); + this.heartbeatTimer = null; + } + } + + private markLost(): void { + this.lost = true; + this.stopHeartbeat(); + } +} diff --git a/app/game-engine/src/turn/databaseHooks.ts b/app/game-engine/src/turn/databaseHooks.ts index 21e1c80..585a541 100644 --- a/app/game-engine/src/turn/databaseHooks.ts +++ b/app/game-engine/src/turn/databaseHooks.ts @@ -29,6 +29,7 @@ import type { InMemoryReservedTurnStore } from './reservedTurnStore.js'; import { buildDiplomacyMeta } from '@sammo-ts/logic'; import { ensureItemInventory, withSerializedItemInventory } from '@sammo-ts/logic/items/index.js'; import { persistGeneralLifecycleEvents } from './generalTurnLifecyclePersistence.js'; +import type { DatabaseTurnDaemonLease } from '../lifecycle/databaseTurnDaemonLease.js'; export interface DatabaseTurnHooks { hooks: TurnDaemonHooks; @@ -313,7 +314,10 @@ const buildLogCreateData = ( export const createDatabaseTurnHooks = async ( databaseUrl: string, world: InMemoryTurnWorld, - options?: { reservedTurns?: InMemoryReservedTurnStore } + options?: { + reservedTurns?: InMemoryReservedTurnStore; + turnDaemonLease?: DatabaseTurnDaemonLease; + } ): Promise => { // 턴 처리 결과를 DB에 반영하는 훅을 만든다. const connector = createGamePostgresConnector({ url: databaseUrl }); @@ -354,6 +358,10 @@ export const createDatabaseTurnHooks = async ( meta: asJson(state.meta), }; const persist = async (prisma: GamePrisma.TransactionClient): Promise => { + // Lock and validate the fencing row in the same transaction as every + // world mutation. A stale daemon can finish calculating, but it can + // never commit after another owner has advanced the epoch. + await options?.turnDaemonLease?.assertActive(prisma); await prisma.worldState.update({ where: { id: state.id }, data: worldStateUpdate, diff --git a/app/game-engine/src/turn/turnDaemon.ts b/app/game-engine/src/turn/turnDaemon.ts index 5b6cfeb..399f072 100644 --- a/app/game-engine/src/turn/turnDaemon.ts +++ b/app/game-engine/src/turn/turnDaemon.ts @@ -34,6 +34,10 @@ import { createTournamentRewardFinalizer } from '../tournament/finalizer.js'; import { createTournamentAutoStartHandler } from './tournamentAutoStart.js'; import { createYearbookHandler } from './yearbookHandler.js'; import { createMonthlyEventHandler, type MonthlyEventActionHandler } from './monthlyEventHandler.js'; +import { + DatabaseTurnDaemonLease, + TurnDaemonLeaseUnavailableError, +} from '../lifecycle/databaseTurnDaemonLease.js'; export interface TurnDaemonRuntimeOptions { profile: string; @@ -55,6 +59,9 @@ export interface TurnDaemonRuntimeOptions { adminActionIntervalMs?: number; redisUrl?: string; commandStreamStartId?: string; + leaseDurationMs?: number; + leaseOwnerId?: string; + enableLeaseHeartbeat?: boolean; } export interface TurnDaemonRuntime { @@ -90,6 +97,19 @@ const resolveRedisConfig = (redisUrl?: string, env: NodeJS.ProcessEnv = process. export const createTurnDaemonRuntime = async (options: TurnDaemonRuntimeOptions): Promise => { // DB에서 월드를 읽고 턴 데몬을 구동할 런타임을 만든다. + const databaseFlushEnabled = options.enableDatabaseFlush ?? true; + const turnDaemonLease = databaseFlushEnabled + ? await DatabaseTurnDaemonLease.connect(options.databaseUrl, { + profile: options.profileName ?? options.profile, + ownerId: options.leaseOwnerId, + leaseDurationMs: options.leaseDurationMs, + heartbeat: options.enableLeaseHeartbeat, + }) + : null; + if (turnDaemonLease && !(await turnDaemonLease.acquire())) { + await turnDaemonLease.close(); + throw new TurnDaemonLeaseUnavailableError(options.profileName ?? options.profile); + } const { state, snapshot } = await loadTurnWorldFromDatabase({ databaseUrl: options.databaseUrl, mapOptions: options.mapOptions, @@ -286,9 +306,10 @@ export const createTurnDaemonRuntime = async (options: TurnDaemonRuntimeOptions) if (gatewayGate) { pauseGate = gatewayGate.shouldPause; } - if (options.enableDatabaseFlush ?? true) { + if (databaseFlushEnabled) { const dbHooks = await createDatabaseTurnHooks(options.databaseUrl, world, { reservedTurns: reservedTurnStoreHandle?.store, + turnDaemonLease: turnDaemonLease ?? undefined, }); auctionBidder = await createAuctionBidder({ databaseUrl: options.databaseUrl, @@ -325,6 +346,7 @@ export const createTurnDaemonRuntime = async (options: TurnDaemonRuntimeOptions) } await gatewayGate?.close(); await adminActionConsumer?.stop(); + await turnDaemonLease?.close(); }; } else if (reservedTurnStoreHandle) { hooks = { @@ -426,7 +448,7 @@ export const createTurnDaemonRuntime = async (options: TurnDaemonRuntimeOptions) stateStore, processor, hooks, - pauseGate, + pauseGate: async () => turnDaemonLease?.isLost() || ((await pauseGate?.()) ?? false), commandHandler, commandResponder: options.controlQueue ? undefined : (databaseCommandQueue ?? undefined), }, diff --git a/app/game-engine/test/turnDaemonLease.integration.test.ts b/app/game-engine/test/turnDaemonLease.integration.test.ts new file mode 100644 index 0000000..16c7be1 --- /dev/null +++ b/app/game-engine/test/turnDaemonLease.integration.test.ts @@ -0,0 +1,121 @@ +import { afterAll, beforeAll, describe, expect, it } from 'vitest'; + +import { createGamePostgresConnector, type GamePrisma, type GamePrismaClient } from '@sammo-ts/infra'; + +import { + DatabaseTurnDaemonLease, + TurnDaemonLeaseLostError, +} from '../src/lifecycle/databaseTurnDaemonLease.js'; + +const databaseUrl = process.env.TURN_DAEMON_LEASE_DATABASE_URL ?? process.env.INPUT_EVENT_DATABASE_URL; +const integration = describe.skipIf(!databaseUrl); +const profilePrefix = 'integration:turn-lease:'; + +integration('database turn daemon lease and fencing', () => { + let db: GamePrismaClient; + let disconnect: (() => Promise) | undefined; + const leases: DatabaseTurnDaemonLease[] = []; + + beforeAll(async () => { + const connector = createGamePostgresConnector({ url: databaseUrl! }); + await connector.connect(); + db = connector.prisma; + disconnect = () => connector.disconnect(); + await db.turnDaemonLease.deleteMany({ + where: { profile: { startsWith: profilePrefix } }, + }); + }); + + afterAll(async () => { + await Promise.allSettled(leases.map((lease) => lease.close())); + await db.turnDaemonLease.deleteMany({ + where: { profile: { startsWith: profilePrefix } }, + }); + await disconnect?.(); + }); + + const createLease = async (profile: string, ownerId: string) => { + const lease = await DatabaseTurnDaemonLease.connect(databaseUrl!, { + profile, + ownerId, + leaseDurationMs: 60_000, + heartbeat: false, + }); + leases.push(lease); + return lease; + }; + + it('allows only one owner to acquire an active profile lease', async () => { + const profile = `${profilePrefix}exclusive`; + const first = await createLease(profile, 'owner-a'); + const second = await createLease(profile, 'owner-b'); + + const [firstToken, secondToken] = await Promise.all([first.acquire(), second.acquire()]); + + expect([firstToken, secondToken].filter(Boolean)).toHaveLength(1); + const row = await db.turnDaemonLease.findUniqueOrThrow({ where: { profile } }); + expect(row.fencingEpoch).toBe(1n); + expect(row.ownerId).toBe(firstToken ? 'owner-a' : 'owner-b'); + }); + + it('increments the epoch on expiry takeover and fences the stale owner', async () => { + const profile = `${profilePrefix}takeover`; + const first = await createLease(profile, 'owner-a'); + const second = await createLease(profile, 'owner-b'); + + const firstToken = await first.acquire(); + expect(firstToken?.fencingEpoch).toBe(1n); + await db.turnDaemonLease.update({ + where: { profile }, + data: { leaseUntil: new Date(Date.now() - 1_000) }, + }); + + const secondToken = await second.acquire(); + expect(secondToken).toMatchObject({ + profile, + ownerId: 'owner-b', + fencingEpoch: 2n, + }); + await expect(first.assertActive()).rejects.toBeInstanceOf(TurnDaemonLeaseLostError); + await expect(second.assertActive()).resolves.toBeUndefined(); + }); + + it('rolls back a stale fenced transaction without writing its completion marker', async () => { + const profile = `${profilePrefix}rollback`; + const requestId = `${profilePrefix}stale-write`; + const first = await createLease(profile, 'owner-a'); + const second = await createLease(profile, 'owner-b'); + await first.acquire(); + await db.turnDaemonLease.update({ + where: { profile }, + data: { leaseUntil: new Date(Date.now() - 1_000) }, + }); + await second.acquire(); + + await expect( + db.$transaction(async (transaction) => { + await first.assertActive(transaction); + await transaction.inputEvent.create({ + data: { + requestId, + target: 'ENGINE', + eventType: 'fenced-test', + payload: {} as GamePrisma.InputJsonValue, + }, + }); + }) + ).rejects.toBeInstanceOf(TurnDaemonLeaseLostError); + expect(await db.inputEvent.findUnique({ where: { requestId } })).toBeNull(); + }); + + it('permits a clean successor after release while fencing a resumed old token', async () => { + const profile = `${profilePrefix}release`; + const first = await createLease(profile, 'owner-a'); + const second = await createLease(profile, 'owner-b'); + + expect((await first.acquire())?.fencingEpoch).toBe(1n); + await first.release(); + expect((await second.acquire())?.fencingEpoch).toBe(2n); + await expect(first.assertActive()).rejects.toBeInstanceOf(TurnDaemonLeaseLostError); + }); +}); diff --git a/packages/infra/prisma/game.prisma b/packages/infra/prisma/game.prisma index e0d46d0..08fb247 100644 --- a/packages/infra/prisma/game.prisma +++ b/packages/infra/prisma/game.prisma @@ -77,6 +77,16 @@ model InputEvent { @@map("input_event") } +model TurnDaemonLease { + profile String @id + ownerId String @map("owner_id") + leaseUntil DateTime @map("lease_until") + fencingEpoch BigInt @default(1) @map("fencing_epoch") + heartbeatAt DateTime @default(now()) @map("heartbeat_at") + + @@map("turn_daemon_lease") +} + model WorldState { id Int @id @default(autoincrement()) scenarioCode String @map("scenario_code") diff --git a/packages/infra/prisma/migrations/20260725002000_add_turn_daemon_lease/migration.sql b/packages/infra/prisma/migrations/20260725002000_add_turn_daemon_lease/migration.sql new file mode 100644 index 0000000..98e155b --- /dev/null +++ b/packages/infra/prisma/migrations/20260725002000_add_turn_daemon_lease/migration.sql @@ -0,0 +1,9 @@ +CREATE TABLE "turn_daemon_lease" ( + "profile" TEXT NOT NULL, + "owner_id" TEXT NOT NULL, + "lease_until" TIMESTAMP(3) NOT NULL, + "fencing_epoch" BIGINT NOT NULL DEFAULT 1, + "heartbeat_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "turn_daemon_lease_pkey" PRIMARY KEY ("profile") +); diff --git a/packages/infra/src/db.ts b/packages/infra/src/db.ts index 3573074..c131074 100644 --- a/packages/infra/src/db.ts +++ b/packages/infra/src/db.ts @@ -27,4 +27,5 @@ export interface DatabaseClient { inheritanceResult: GamePrisma.InheritanceResultDelegate; inheritanceUserState: GamePrisma.InheritanceUserStateDelegate; inputEvent: GamePrisma.InputEventDelegate; + turnDaemonLease: GamePrisma.TurnDaemonLeaseDelegate; }