From cca2b3d925608ed15dc7e65067dc4da2f8aeea50 Mon Sep 17 00:00:00 2001 From: hided62 Date: Mon, 24 Aug 2026 16:30:38 +0000 Subject: [PATCH] =?UTF-8?q?fix:=20API=20=EC=9E=85=EB=A0=A5=20=EC=9D=B4?= =?UTF-8?q?=EB=B2=A4=ED=8A=B8=EC=9D=98=20=EC=9B=90=EC=9E=90=EC=A0=81=20?= =?UTF-8?q?=EC=9E=AC=EC=8B=A4=ED=96=89=EC=9D=84=20=EB=B3=B4=EC=9E=A5?= =?UTF-8?q?=ED=95=9C=EB=8B=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/game-api/src/inputEventBoundary.ts | 327 ++++++++++++++--- app/game-api/src/trpc.ts | 25 +- .../test/generalAccessTracking.test.ts | 40 +- .../inputEventBoundary.integration.test.ts | 279 +++++++++++++- app/game-api/test/inputEventBoundary.test.ts | 37 ++ app/game-api/test/inputEventJournal.test.ts | 83 +++-- app/game-api/test/requestId.test.ts | 9 + .../securityTransport.integration.test.ts | 343 +++++++++++++++++- docs/architecture/api-input-event-replay.md | 117 ++++++ 9 files changed, 1141 insertions(+), 119 deletions(-) create mode 100644 app/game-api/test/inputEventBoundary.test.ts create mode 100644 docs/architecture/api-input-event-replay.md diff --git a/app/game-api/src/inputEventBoundary.ts b/app/game-api/src/inputEventBoundary.ts index 4c3c8ad1..1a0993dd 100644 --- a/app/game-api/src/inputEventBoundary.ts +++ b/app/game-api/src/inputEventBoundary.ts @@ -1,86 +1,299 @@ -import type { GamePrisma } from '@sammo-ts/infra'; +import { createHash } from 'node:crypto'; + +import { GamePrisma, type DatabaseClient as InfraDatabaseClient } from '@sammo-ts/infra'; import type { DatabaseClient } from './context.js'; +const API_INPUT_PAYLOAD_VERSION = 1 as const; +const BUSINESS_SAVEPOINT = 'api_input_event_business'; + +export interface ApiInputPayloadIdentity { + version: typeof API_INPUT_PAYLOAD_VERSION; + digest: string; +} + +interface LockedInputEvent { + target: 'API' | 'ENGINE'; + eventType: string; + payload: GamePrisma.JsonValue; + actorUserId: string | null; + status: 'PENDING' | 'PROCESSING' | 'SUCCEEDED' | 'FAILED'; + result: GamePrisma.JsonValue | null; + attempts: number; +} + +type InputEventOutcome = + { kind: 'executed'; value: T } | { kind: 'replayed'; value: T } | { kind: 'failed'; error: unknown }; + +type SavepointDatabaseClient = InfraDatabaseClient & { + $executeRawUnsafe(query: string): Promise; +}; + const asJson = (value: unknown): GamePrisma.InputJsonValue => value as GamePrisma.InputJsonValue; +const canonicalJson = (value: unknown): string => + JSON.stringify(value, (_key, entry: unknown) => { + if (typeof entry === 'bigint') { + return entry.toString(); + } + if (entry && typeof entry === 'object' && !Array.isArray(entry)) { + return Object.fromEntries( + Object.entries(entry as Record).sort(([left], [right]) => + left < right ? -1 : left > right ? 1 : 0 + ) + ); + } + return entry; + }) ?? 'null'; + +const canonicalJsonValue = (value: unknown): GamePrisma.InputJsonValue => + JSON.parse(canonicalJson(value)) as GamePrisma.InputJsonValue; + +export const createApiInputPayloadIdentity = (payload: unknown): ApiInputPayloadIdentity => ({ + version: API_INPUT_PAYLOAD_VERSION, + digest: `sha256:${createHash('sha256').update(canonicalJson(payload)).digest('hex')}`, +}); + +const isLegacyEmptyPayload = (payload: GamePrisma.JsonValue): boolean => + payload !== null && !Array.isArray(payload) && typeof payload === 'object' && Object.keys(payload).length === 0; + +const sameJson = (left: unknown, right: unknown): boolean => canonicalJson(left) === canonicalJson(right); + export class DuplicateInputEventError extends Error { constructor(readonly requestId: string) { - super(`Input event ${requestId} was already accepted.`); + super(`Input event ${requestId} conflicts with an existing request.`); this.name = 'DuplicateInputEventError'; } } +const insertPendingIfAbsent = async ( + db: DatabaseClient, + options: { + requestId: string; + eventType: string; + actorUserId: string | null; + payloadIdentity: ApiInputPayloadIdentity; + } +): Promise => { + await db.$executeRaw( + GamePrisma.sql` + INSERT INTO input_event ( + request_id, + target, + event_type, + payload, + actor_user_id, + status, + attempts, + created_at + ) + VALUES ( + ${options.requestId}, + 'API'::"InputEventTarget", + ${options.eventType}, + CAST(${JSON.stringify(options.payloadIdentity)} AS jsonb), + ${options.actorUserId}, + 'PENDING'::"InputEventStatus", + 0, + CURRENT_TIMESTAMP AT TIME ZONE 'UTC' + ) + ON CONFLICT (request_id) DO NOTHING + ` + ); +}; + +const lockInputEvent = async (db: DatabaseClient, requestId: string): Promise => { + const rows = await db.$queryRaw( + GamePrisma.sql` + SELECT + target, + event_type AS "eventType", + payload, + actor_user_id AS "actorUserId", + status, + result, + attempts + FROM input_event + WHERE request_id = ${requestId} + FOR UPDATE + ` + ); + const row = rows[0]; + if (!row) { + throw new Error(`Input event ${requestId} disappeared while being claimed.`); + } + return row; +}; + +const hasMatchingBaseIdentity = ( + row: LockedInputEvent, + options: { eventType: string; actorUserId: string | null } +): boolean => row.target === 'API' && row.eventType === options.eventType && row.actorUserId === options.actorUserId; + +const canAdoptLegacyFailedPayload = ( + row: LockedInputEvent, + options: { eventType: string; actorUserId: string | null } +): boolean => + row.status === 'FAILED' && + row.result === null && + isLegacyEmptyPayload(row.payload) && + hasMatchingBaseIdentity(row, options); + +const isMatchingIdentity = ( + row: LockedInputEvent, + options: { + eventType: string; + actorUserId: string | null; + payloadIdentity: ApiInputPayloadIdentity; + } +): boolean => hasMatchingBaseIdentity(row, options) && sameJson(row.payload, options.payloadIdentity); + +const claimInputEvent = async ( + db: DatabaseClient, + requestId: string, + payloadIdentity: ApiInputPayloadIdentity +): Promise => { + await db.inputEvent.update({ + where: { requestId }, + data: { + payload: asJson(payloadIdentity), + status: 'PROCESSING', + result: GamePrisma.DbNull, + error: null, + attempts: { increment: 1 }, + lockedBy: null, + leaseUntil: null, + processingAt: new Date(), + completedAt: null, + }, + }); +}; + +const markUnexpectedFailure = async ( + db: DatabaseClient, + options: { + requestId: string; + eventType: string; + actorUserId: string | null; + payloadIdentity: ApiInputPayloadIdentity; + error: unknown; + } +): Promise => { + if (!db.$transaction) return; + const message = options.error instanceof Error ? options.error.message : 'Unknown API input event error.'; + + try { + await db.$transaction(async (transaction) => { + await insertPendingIfAbsent(transaction, options); + const row = await lockInputEvent(transaction, options.requestId); + const identityMatches = isMatchingIdentity(row, options) || canAdoptLegacyFailedPayload(row, options); + if (!identityMatches || row.status === 'SUCCEEDED' || row.status === 'PROCESSING') { + // A retry may have committed while the failed caller was unwinding. A + // late failure recorder must never replace its durable success. + return; + } + await transaction.inputEvent.update({ + where: { requestId: options.requestId }, + data: { + payload: asJson(options.payloadIdentity), + status: 'FAILED', + result: GamePrisma.DbNull, + error: message, + attempts: { increment: 1 }, + lockedBy: null, + leaseUntil: null, + processingAt: new Date(), + completedAt: new Date(), + }, + }); + }); + } catch { + // Preserve the transaction failure that the caller actually observed. If + // the database is unavailable, the prior PENDING/FAILED state (or absence + // of a newly rolled-back row) remains safely retryable. + } +}; + export const executeInputEvent = async (options: { db: DatabaseClient; requestId: string; eventType: string; + payload: unknown; actorUserId?: string | null; execute(db: DatabaseClient): Promise; }): Promise => { - const { db, requestId, eventType, actorUserId, execute } = options; + const { db, requestId, eventType, payload, execute } = options; + const actorUserId = options.actorUserId ?? null; + const payloadIdentity = createApiInputPayloadIdentity(payload); if (!db.$transaction) { return execute(db); } - const processingAt = new Date(); + let businessStarted = false; + let outcome: InputEventOutcome; 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); - } - } + outcome = await db.$transaction(async (transaction) => { + await insertPendingIfAbsent(transaction, { requestId, eventType, actorUserId, payloadIdentity }); + const row = await lockInputEvent(transaction, requestId); + const identityMatches = isMatchingIdentity(row, { eventType, actorUserId, payloadIdentity }); - 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; + if (row.status === 'SUCCEEDED') { + if (!identityMatches) throw new DuplicateInputEventError(requestId); + return { kind: 'replayed', value: row.result as T }; + } + // A visible PROCESSING row was committed by the legacy boundary. It + // may still have an active business request and its {} payload cannot + // prove identity, so automatic reclaim would risk duplicate writes. + if (row.status === 'PROCESSING') { + throw new DuplicateInputEventError(requestId); + } + if (!identityMatches && !canAdoptLegacyFailedPayload(row, { eventType, actorUserId })) { + throw new DuplicateInputEventError(requestId); + } + + await claimInputEvent(transaction, requestId, payloadIdentity); + const savepointDb = transaction as SavepointDatabaseClient; + await savepointDb.$executeRawUnsafe(`SAVEPOINT ${BUSINESS_SAVEPOINT}`); + businessStarted = true; + try { + const value = await execute(transaction); + const durableResult = canonicalJsonValue(value); + await transaction.inputEvent.update({ + where: { requestId }, + data: { + status: 'SUCCEEDED', + result: asJson(durableResult), + error: null, + completedAt: new Date(), + }, + }); + await savepointDb.$executeRawUnsafe(`RELEASE SAVEPOINT ${BUSINESS_SAVEPOINT}`); + return { kind: 'executed', value }; + } catch (error) { + await savepointDb.$executeRawUnsafe(`ROLLBACK TO SAVEPOINT ${BUSINESS_SAVEPOINT}`); + await savepointDb.$executeRawUnsafe(`RELEASE SAVEPOINT ${BUSINESS_SAVEPOINT}`); + const message = error instanceof Error ? error.message : 'Unknown API input event error.'; + await transaction.inputEvent.update({ + where: { requestId }, + data: { + status: 'FAILED', + result: GamePrisma.DbNull, + error: message, + completedAt: new Date(), + }, + }); + return { kind: 'failed', error }; + } }); } 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(), - }, - }); + if (businessStarted && !(error instanceof DuplicateInputEventError)) { + await markUnexpectedFailure(db, { requestId, eventType, actorUserId, payloadIdentity, error }); + } throw error; } + + if (outcome.kind === 'failed') { + throw outcome.error; + } + return outcome.value; }; diff --git a/app/game-api/src/trpc.ts b/app/game-api/src/trpc.ts index 34dd502d..a67aa250 100644 --- a/app/game-api/src/trpc.ts +++ b/app/game-api/src/trpc.ts @@ -1,5 +1,6 @@ import { randomUUID } from 'node:crypto'; import { initTRPC, TRPCError } from '@trpc/server'; +import { middlewareMarker } from '@trpc/server/unstable-core-do-not-import'; import { ChangeJournal } from '@sammo-ts/common'; import { isGameAccessBlocked } from '@sammo-ts/common/auth/sanctions'; import { writeReadModelChangeJournal } from '@sammo-ts/infra'; @@ -58,19 +59,25 @@ const generalActivityMiddleware = t.middleware(async ({ ctx, type, next }) => { return result; }); -const inputEventMiddleware = t.middleware(async ({ ctx, type, path, next }) => { +export const scopeApiInputEventRequestId = (baseRequestId: string, path: string, batchIndex: number): string => + `${baseRequestId}:${path}${batchIndex === 0 ? '' : `:batch:${batchIndex}`}`; + +const inputEventMiddleware = t.middleware(async ({ ctx, type, path, batchIndex, getRawInput, next }) => { if (type !== 'mutation' || !ctx.db.$transaction) { return next(); } - const requestId = `${ctx.requestId ?? randomUUID()}:${path}`; + const requestId = scopeApiInputEventRequestId(ctx.requestId ?? randomUUID(), path, batchIndex); + const payload = await getRawInput(); const changeJournal = new ChangeJournal(); let journalPersisted = false; + let executedResult: Awaited> | undefined; try { - const result = await executeInputEvent({ + const response = await executeInputEvent({ db: ctx.db, requestId, eventType: path, + payload, actorUserId: ctx.auth?.user.id, execute: async (transaction) => { const result = await next({ @@ -85,13 +92,21 @@ const inputEventMiddleware = t.middleware(async ({ ctx, type, path, next }) => { throw result.error; } journalPersisted = Boolean(await writeReadModelChangeJournal(transaction, changeJournal.snapshot())); - return result; + executedResult = result; + return result.data; }, }); if (journalPersisted) { ctx.readModelOutbox?.wake(); } - return result; + if (executedResult) { + return executedResult; + } + return { + marker: middlewareMarker, + ok: true, + data: response, + }; } catch (error) { if (error instanceof DuplicateInputEventError) { throw new TRPCError({ diff --git a/app/game-api/test/generalAccessTracking.test.ts b/app/game-api/test/generalAccessTracking.test.ts index 15d017f0..bd9330dc 100644 --- a/app/game-api/test/generalAccessTracking.test.ts +++ b/app/game-api/test/generalAccessTracking.test.ts @@ -4,6 +4,7 @@ import type { GameSessionTokenPayload } from '@sammo-ts/common/auth/gameToken'; import { z } from 'zod'; import type { GameApiContext } from '../src/context.js'; import type { DatabaseClient } from '../src/context.js'; +import { createApiInputPayloadIdentity } from '../src/inputEventBoundary.js'; import { accessAuthedProcedure, @@ -303,13 +304,35 @@ describe('general access tracking', () => { const transactionClient = { $queryRaw: vi.fn(async (query: unknown) => { const sql = (query as { sql?: string }).sql ?? ''; + if (sql.includes('FROM input_event')) { + return [ + { + target: 'API', + eventType: 'board.writeArticle', + payload: createApiInputPayloadIdentity({ value: 'ok' }), + actorUserId: 'user-7', + status: 'PENDING', + result: null, + attempts: 0, + }, + ]; + } return sql.includes('read_model_revision') ? [{ domain: 'access.general', entityId: 7, revision: 1n, outboxId: 1n }] : [{ id: 41 }]; }), - $executeRaw: vi.fn(async () => 1), + $executeRaw: vi.fn(async (query: unknown) => { + if (((query as { sql?: string }).sql ?? '').includes('INSERT INTO input_event')) { + events.push('input-event-create'); + } + return 1; + }), + $executeRawUnsafe: vi.fn(async () => 0), inputEvent: { - update: vi.fn(async () => ({})), + update: vi.fn(async (args: { data: { status: string } }) => { + if (args.data.status === 'FAILED') events.push('input-event-failed'); + return {}; + }), }, }; const db = { @@ -338,17 +361,6 @@ describe('general access tracking', () => { }, })), }, - inputEvent: { - create: vi.fn(async () => { - events.push('input-event-create'); - return {}; - }), - update: vi.fn(async () => { - events.push('input-event-failed'); - return {}; - }), - updateMany: vi.fn(async () => ({ count: 0 })), - }, $transaction: vi.fn(async (callback: (client: typeof transactionClient) => Promise) => { transactionCount += 1; events.push(transactionCount === 1 ? 'access-transaction' : 'business-transaction'); @@ -385,8 +397,8 @@ describe('general access tracking', () => { expect(events).toEqual([ 'input-parse', 'access-transaction', - 'input-event-create', 'business-transaction', + 'input-event-create', 'resolver', 'input-event-failed', ]); diff --git a/app/game-api/test/inputEventBoundary.integration.test.ts b/app/game-api/test/inputEventBoundary.integration.test.ts index a45e8f17..d1dfb8cf 100644 --- a/app/game-api/test/inputEventBoundary.integration.test.ts +++ b/app/game-api/test/inputEventBoundary.integration.test.ts @@ -2,8 +2,12 @@ import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'; import { z } from 'zod'; import { createGamePostgresConnector, type GamePrismaClient } from '@sammo-ts/infra'; -import type { GameApiContext } from '../src/context.js'; -import { DuplicateInputEventError, executeInputEvent } from '../src/inputEventBoundary.js'; +import type { DatabaseClient, GameApiContext } from '../src/context.js'; +import { + createApiInputPayloadIdentity, + DuplicateInputEventError, + executeInputEvent, +} from '../src/inputEventBoundary.js'; import { procedure, router } from '../src/trpc.js'; import { ConflictingTurnDaemonCommandError, @@ -74,6 +78,7 @@ integration('API input event boundary', () => { db, requestId, eventType: 'test.success', + payload: { markerId }, actorUserId: 'user-7', execute: async (transaction) => { await transaction.inputEvent.create({ @@ -94,6 +99,8 @@ integration('API input event boundary', () => { expect(event).toMatchObject({ status: 'SUCCEEDED', actorUserId: 'user-7', + payload: createApiInputPayloadIdentity({ markerId }), + result: { ok: true }, attempts: 1, }); expect(event.processingAt).toBeInstanceOf(Date); @@ -168,6 +175,12 @@ integration('API input event boundary', () => { const outboxes = await db.readModelOutbox.findMany({ select: { payload: true } }); expect(outboxes.some(({ payload }) => payloadHasGeneral(payload, journalGeneralIds[1]))).toBe(false); expect(wake).not.toHaveBeenCalled(); + expect(await db.inputEvent.findUniqueOrThrow({ where: { requestId: `${requestId}:mutate` } })).toMatchObject({ + payload: createApiInputPayloadIdentity({ generalId: journalGeneralIds[1], fail: true }), + status: 'FAILED', + result: null, + attempts: 1, + }); }); it('rolls back business writes, records failure, and permits one explicit retry', async () => { @@ -178,6 +191,7 @@ integration('API input event boundary', () => { db, requestId, eventType: 'test.failure', + payload: { markerId }, execute: async (transaction) => { await transaction.inputEvent.create({ data: { @@ -194,6 +208,8 @@ integration('API input event boundary', () => { expect(await db.inputEvent.findUnique({ where: { requestId: markerId } })).toBeNull(); expect(await db.inputEvent.findUniqueOrThrow({ where: { requestId } })).toMatchObject({ status: 'FAILED', + payload: createApiInputPayloadIdentity({ markerId }), + result: null, attempts: 1, error: 'injected transaction failure', }); @@ -202,15 +218,17 @@ integration('API input event boundary', () => { db, requestId, eventType: 'test.failure', + payload: { markerId }, execute: async () => ({ ok: true }), }); expect(await db.inputEvent.findUniqueOrThrow({ where: { requestId } })).toMatchObject({ status: 'SUCCEEDED', + result: { ok: true }, attempts: 2, }); }); - it('rejects a concurrent duplicate idempotency key', async () => { + it('serializes an exact concurrent retry and replays the original result without re-executing business', async () => { const requestId = 'integration:api:duplicate'; let releaseFirst: (() => void) | undefined; let signalStarted: (() => void) | undefined; @@ -224,25 +242,270 @@ integration('API input event boundary', () => { db, requestId, eventType: 'test.duplicate', + payload: { value: 7 }, execute: async () => { signalStarted?.(); await release; - return { ok: true }; + return { ok: true, revision: 17 }; }, }); await started; + const duplicateExecute = vi.fn(async () => ({ ok: true, revision: 99 })); + const duplicate = executeInputEvent({ + db, + requestId, + eventType: 'test.duplicate', + payload: { value: 7 }, + execute: duplicateExecute, + }); + releaseFirst?.(); + await expect(Promise.all([first, duplicate])).resolves.toEqual([ + { ok: true, revision: 17 }, + { ok: true, revision: 17 }, + ]); + expect(duplicateExecute).not.toHaveBeenCalled(); + expect(await db.inputEvent.findUniqueOrThrow({ where: { requestId } })).toMatchObject({ + payload: createApiInputPayloadIdentity({ value: 7 }), + result: { ok: true, revision: 17 }, + status: 'SUCCEEDED', + attempts: 1, + }); + }); + + it('rejects request-id reuse with a changed payload, event type, or actor', async () => { + const requestId = 'integration:api:identity-conflict'; + const original = await executeInputEvent({ + db, + requestId, + eventType: 'test.identity', + payload: { value: 1, nested: { left: true, right: false } }, + actorUserId: 'user-identity', + execute: async () => ({ ok: true, revision: 4 }), + }); + expect(original).toEqual({ ok: true, revision: 4 }); + + const conflicts = [ + { + eventType: 'test.identity', + payload: { value: 2, nested: { left: true, right: false } }, + actorUserId: 'user-identity', + }, + { + eventType: 'test.other-identity', + payload: { value: 1, nested: { left: true, right: false } }, + actorUserId: 'user-identity', + }, + { + eventType: 'test.identity', + payload: { value: 1, nested: { left: true, right: false } }, + actorUserId: 'other-user', + }, + ]; + for (const conflict of conflicts) { + const conflictingExecute = vi.fn(async () => ({ ok: false })); + await expect( + executeInputEvent({ db, requestId, ...conflict, execute: conflictingExecute }) + ).rejects.toBeInstanceOf(DuplicateInputEventError); + expect(conflictingExecute).not.toHaveBeenCalled(); + } + + expect(await db.inputEvent.findUniqueOrThrow({ where: { requestId } })).toMatchObject({ + eventType: 'test.identity', + actorUserId: 'user-identity', + payload: createApiInputPayloadIdentity({ value: 1, nested: { left: true, right: false } }), + result: { ok: true, revision: 4 }, + status: 'SUCCEEDED', + attempts: 1, + }); + }); + + it('reclaims an exact PENDING row under lock and counts one execution attempt', async () => { + const requestId = 'integration:api:pending-reclaim'; + const payload = { value: 'pending' }; + await db.inputEvent.create({ + data: { + requestId, + target: 'API', + eventType: 'test.pending', + payload: { ...createApiInputPayloadIdentity(payload) }, + actorUserId: 'pending-user', + status: 'PENDING', + attempts: 3, + }, + }); + await expect( executeInputEvent({ db, requestId, - eventType: 'test.duplicate', + eventType: 'test.pending', + payload, + actorUserId: 'pending-user', + execute: async () => ({ ok: true, attempt: 4 }), + }) + ).resolves.toEqual({ ok: true, attempt: 4 }); + expect(await db.inputEvent.findUniqueOrThrow({ where: { requestId } })).toMatchObject({ + status: 'SUCCEEDED', + result: { ok: true, attempt: 4 }, + attempts: 4, + }); + }); + + it('adopts only a matching legacy FAILED placeholder and replaces it with the canonical digest', async () => { + const requestId = 'integration:api:legacy-failed'; + const payload = { value: 'legacy-retry' }; + await db.inputEvent.create({ + data: { + requestId, + target: 'API', + eventType: 'test.legacy-failed', + payload: {}, + actorUserId: 'legacy-user', + status: 'FAILED', + attempts: 2, + error: 'legacy failure', + completedAt: new Date(), + }, + }); + + await expect( + executeInputEvent({ + db, + requestId, + eventType: 'test.legacy-failed', + payload, + actorUserId: 'legacy-user', execute: async () => ({ ok: true }), }) - ).rejects.toBeInstanceOf(DuplicateInputEventError); + ).resolves.toEqual({ ok: true }); + expect(await db.inputEvent.findUniqueOrThrow({ where: { requestId } })).toMatchObject({ + payload: createApiInputPayloadIdentity(payload), + status: 'SUCCEEDED', + result: { ok: true }, + error: null, + attempts: 3, + }); + }); - releaseFirst?.(); - await first; + it('fails closed on a committed legacy PROCESSING placeholder', async () => { + const requestId = 'integration:api:legacy-processing'; + const processingAt = new Date(Date.now() - 60 * 60 * 1_000); + await db.inputEvent.create({ + data: { + requestId, + target: 'API', + eventType: 'test.legacy-processing', + payload: {}, + actorUserId: 'legacy-user', + status: 'PROCESSING', + attempts: 1, + processingAt, + }, + }); + const retryExecute = vi.fn(async () => ({ ok: true })); + + await expect( + executeInputEvent({ + db, + requestId, + eventType: 'test.legacy-processing', + payload: { value: 'cannot-prove-legacy-identity' }, + actorUserId: 'legacy-user', + execute: retryExecute, + }) + ).rejects.toBeInstanceOf(DuplicateInputEventError); + expect(retryExecute).not.toHaveBeenCalled(); + expect(await db.inputEvent.findUniqueOrThrow({ where: { requestId } })).toMatchObject({ + payload: {}, + status: 'PROCESSING', + attempts: 1, + processingAt, + }); + }); + + it('commits FAILED before a blocked exact retry and preserves the exact attempt count after success', async () => { + const requestId = 'integration:api:failure-race'; + const payload = { value: 'race' }; + let releaseFailure: (() => void) | undefined; + let signalStarted: (() => void) | undefined; + const started = new Promise((resolve) => { + signalStarted = resolve; + }); + const release = new Promise((resolve) => { + releaseFailure = resolve; + }); + const failed = executeInputEvent({ + db, + requestId, + eventType: 'test.failure-race', + payload, + execute: async () => { + signalStarted?.(); + await release; + throw new Error('first attempt failed'); + }, + }); + await started; + const retry = executeInputEvent({ + db, + requestId, + eventType: 'test.failure-race', + payload, + execute: async () => ({ ok: true, attempt: 2 }), + }); + + releaseFailure?.(); + await expect(failed).rejects.toThrow('first attempt failed'); + await expect(retry).resolves.toEqual({ ok: true, attempt: 2 }); + expect(await db.inputEvent.findUniqueOrThrow({ where: { requestId } })).toMatchObject({ + status: 'SUCCEEDED', + result: { ok: true, attempt: 2 }, + error: null, + attempts: 2, + }); + }); + + it('does not let a late unexpected-failure recorder overwrite a transaction that actually committed', async () => { + const requestId = 'integration:api:ambiguous-commit'; + const payload = { value: 'committed-before-client-error' }; + const ambiguousCommitDb = new Proxy(db, { + get(target, property, receiver) { + if (property !== '$transaction') return Reflect.get(target, property, receiver); + return async (callback: (transaction: DatabaseClient) => Promise) => { + await db.$transaction(async (transaction) => callback(transaction)); + throw new Error('injected post-commit transport failure'); + }; + }, + }) as unknown as DatabaseClient; + + await expect( + executeInputEvent({ + db: ambiguousCommitDb, + requestId, + eventType: 'test.ambiguous-commit', + payload, + execute: async () => ({ ok: true, revision: 8 }), + }) + ).rejects.toThrow('injected post-commit transport failure'); + expect(await db.inputEvent.findUniqueOrThrow({ where: { requestId } })).toMatchObject({ + status: 'SUCCEEDED', + result: { ok: true, revision: 8 }, + error: null, + attempts: 1, + }); + + const replayExecute = vi.fn(async () => ({ ok: false })); + await expect( + executeInputEvent({ + db, + requestId, + eventType: 'test.ambiguous-commit', + payload, + execute: replayExecute, + }) + ).resolves.toEqual({ ok: true, revision: 8 }); + expect(replayExecute).not.toHaveBeenCalled(); }); it('reuses the same engine child event but rejects a changed retry payload', async () => { diff --git a/app/game-api/test/inputEventBoundary.test.ts b/app/game-api/test/inputEventBoundary.test.ts new file mode 100644 index 00000000..9e992eb8 --- /dev/null +++ b/app/game-api/test/inputEventBoundary.test.ts @@ -0,0 +1,37 @@ +import { describe, expect, it } from 'vitest'; + +import { createApiInputPayloadIdentity } from '../src/inputEventBoundary.js'; + +describe('API input-event payload identity', () => { + it('hashes canonical JSON independently of object key order', () => { + expect( + createApiInputPayloadIdentity({ + second: [{ z: true, a: 1 }], + first: 'value', + }) + ).toEqual( + createApiInputPayloadIdentity({ + first: 'value', + second: [{ a: 1, z: true }], + }) + ); + }); + + it('distinguishes changed values and array order', () => { + const original = createApiInputPayloadIdentity({ value: 1, items: ['a', 'b'] }); + expect(createApiInputPayloadIdentity({ value: 2, items: ['a', 'b'] })).not.toEqual(original); + expect(createApiInputPayloadIdentity({ value: 1, items: ['b', 'a'] })).not.toEqual(original); + }); + + it('stores only a bounded digest envelope for a large or private payload', () => { + const privatePayload = { dataUrl: `data:image/png;base64,${'A'.repeat(100_000)}`, text: 'private-message' }; + const identity = createApiInputPayloadIdentity(privatePayload); + + expect(identity).toEqual({ + version: 1, + digest: expect.stringMatching(/^sha256:[0-9a-f]{64}$/u), + }); + expect(JSON.stringify(identity)).not.toContain('private-message'); + expect(JSON.stringify(identity).length).toBeLessThan(128); + }); +}); diff --git a/app/game-api/test/inputEventJournal.test.ts b/app/game-api/test/inputEventJournal.test.ts index c54b4b1d..4046fb47 100644 --- a/app/game-api/test/inputEventJournal.test.ts +++ b/app/game-api/test/inputEventJournal.test.ts @@ -2,46 +2,60 @@ import { describe, expect, it, vi } from 'vitest'; import { z } from 'zod'; import type { GameApiContext } from '../src/context.js'; +import { createApiInputPayloadIdentity } from '../src/inputEventBoundary.js'; import { procedure, router } from '../src/trpc.js'; const testRouter = router({ - mutate: procedure - .input(z.object({ fail: z.boolean().optional().default(false) })) - .mutation(({ ctx, input }) => { - (ctx as GameApiContext & { testOrder: string[] }).testOrder.push('handler'); - ctx.changeJournal?.mark('front.general', 7); - if (input.fail) throw new Error('injected rollback'); - return { ok: true }; - }), + mutate: procedure.input(z.object({ fail: z.boolean().optional().default(false) })).mutation(({ ctx, input }) => { + (ctx as GameApiContext & { testOrder: string[] }).testOrder.push('handler'); + ctx.changeJournal?.mark('front.general', 7); + if (input.fail) throw new Error('injected rollback'); + return { ok: true }; + }), }); -const createContext = () => { +const createContext = (payload: unknown = {}) => { const order: string[] = []; - const queryRaw = vi.fn(async () => { + const queryRaw = vi.fn(async (query: { sql?: string }) => { + if (query.sql?.includes('FROM input_event')) { + order.push('locked'); + return [ + { + target: 'API', + eventType: 'mutate', + payload: createApiInputPayloadIdentity(payload), + actorUserId: null, + status: 'PENDING', + result: null, + attempts: 0, + }, + ]; + } order.push('journal'); return [{ domain: 'front.general', entityId: 7, revision: 1n, outboxId: 11n }]; }); const transaction = { $queryRaw: queryRaw, + $executeRaw: vi.fn(async () => { + order.push('accepted'); + return 1; + }), + $executeRawUnsafe: vi.fn(async (statement: string) => { + if (statement.startsWith('SAVEPOINT ')) order.push('savepoint'); + else if (statement.startsWith('ROLLBACK TO ')) order.push('savepoint-rollback'); + else if (statement.startsWith('RELEASE ')) order.push('savepoint-release'); + return 0; + }), inputEvent: { - update: vi.fn(async () => { - order.push('succeeded'); + update: vi.fn(async (args: { data: { status: string } }) => { + if (args.data.status === 'PROCESSING') order.push('processing'); + else if (args.data.status === 'SUCCEEDED') order.push('succeeded'); + else if (args.data.status === 'FAILED') order.push('failed'); return {}; }), }, }; const db = { - inputEvent: { - create: vi.fn(async () => { - order.push('accepted'); - return {}; - }), - updateMany: vi.fn(async () => ({ count: 0 })), - update: vi.fn(async () => { - order.push('failed'); - return {}; - }), - }, $transaction: vi.fn(async (callback: (db: typeof transaction) => Promise) => { order.push('transaction-begin'); try { @@ -73,11 +87,15 @@ describe('API input-event change journal boundary', () => { await expect(testRouter.createCaller(fixture.context).mutate({})).resolves.toEqual({ ok: true }); expect(fixture.order).toEqual([ - 'accepted', 'transaction-begin', + 'accepted', + 'locked', + 'processing', + 'savepoint', 'handler', 'journal', 'succeeded', + 'savepoint-release', 'commit', 'wake', ]); @@ -86,14 +104,25 @@ describe('API input-event change journal boundary', () => { }); it('rolls back a handler mark without writing or scheduling an outbox row', async () => { - const fixture = createContext(); + const fixture = createContext({ fail: true }); await expect(testRouter.createCaller(fixture.context).mutate({ fail: true })).rejects.toThrow( 'injected rollback' ); - expect(fixture.order).toEqual(['accepted', 'transaction-begin', 'handler', 'rollback', 'failed']); - expect(fixture.queryRaw).not.toHaveBeenCalled(); + expect(fixture.order).toEqual([ + 'transaction-begin', + 'accepted', + 'locked', + 'processing', + 'savepoint', + 'handler', + 'savepoint-rollback', + 'savepoint-release', + 'failed', + 'commit', + ]); + expect(fixture.queryRaw).toHaveBeenCalledTimes(1); expect(fixture.redisPublish).not.toHaveBeenCalled(); expect(fixture.wake).not.toHaveBeenCalled(); }); diff --git a/app/game-api/test/requestId.test.ts b/app/game-api/test/requestId.test.ts index be0b67e6..59928971 100644 --- a/app/game-api/test/requestId.test.ts +++ b/app/game-api/test/requestId.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from 'vitest'; import { scopeHttpIdempotencyKey } from '../src/requestId.js'; +import { scopeApiInputEventRequestId } from '../src/trpc.js'; describe('HTTP idempotency request IDs', () => { it('is stable for one principal and isolated across users and profiles', () => { @@ -24,4 +25,12 @@ describe('HTTP idempotency request IDs', () => { expect(scoped).toMatch(/^http:[0-9a-f]{64}$/u); expect(scoped).toHaveLength(69); }); + + it('keeps the first call compatible and isolates later calls in a same-path batch', () => { + expect(scopeApiInputEventRequestId('http:base', 'messages.send', 0)).toBe('http:base:messages.send'); + expect(scopeApiInputEventRequestId('http:base', 'messages.send', 1)).toBe('http:base:messages.send:batch:1'); + expect(scopeApiInputEventRequestId('http:base', 'messages.send', 2)).not.toBe( + scopeApiInputEventRequestId('http:base', 'messages.send', 1) + ); + }); }); diff --git a/app/game-api/test/securityTransport.integration.test.ts b/app/game-api/test/securityTransport.integration.test.ts index 5f6820c2..f67efda8 100644 --- a/app/game-api/test/securityTransport.integration.test.ts +++ b/app/game-api/test/securityTransport.integration.test.ts @@ -16,6 +16,7 @@ import { } from '@sammo-ts/infra'; import { RedisAccessTokenStore } from '../src/auth/accessTokenStore.js'; +import { createApiInputPayloadIdentity } from '../src/inputEventBoundary.js'; import { scopeHttpIdempotencyKey } from '../src/requestId.js'; import { createGameApiServer } from '../src/server.js'; import { WebPushOutboxWorker } from '../src/services/webPushOutboxWorker.js'; @@ -46,6 +47,7 @@ const matrixApiEventTypes = [ 'messages.send', 'turns.reserved.setGeneral', 'turns.reserved.setNation', + 'turns.reserved.setNationBulk', ] as const; const fixtureActorUserIds = [userId, noGeneralUserId, sameNationUserId, foreignUserId, ordinaryUserId]; const secret = 'security-http-e2e-secret'; @@ -74,6 +76,7 @@ let disconnectDb: (() => Promise) | null = null; let redis: RedisConnector | null = null; let accessTokenStore: RedisAccessTokenStore; let createdFixtureWorld = false; +let reservationWorldId = fixtureWorldId; let gatewayStatusServer: HttpServer | null = null; let receivedGatewayWebPushEvents: Array<{ internalToken: string | null; body: unknown }> = []; @@ -285,6 +288,7 @@ const readReservedMutationState = async () => ({ where: { OR: [ { domain: 'reserved.general', entityId: { in: fixtureGeneralIds } }, + { domain: 'general.content', entityId: { in: fixtureGeneralIds } }, { domain: 'dashboard.global', entityId: 0 }, ], }, @@ -420,7 +424,12 @@ const readRealtimeRedisState = async (): Promise> const expectApiInputEvent = async ( idempotencyKey: string, procedure: string, - expected: { actorUserId: string; status: 'FAILED' | 'SUCCEEDED' } | null + expected: { + actorUserId: string; + status: 'FAILED' | 'SUCCEEDED'; + payload?: unknown; + result?: unknown; + } | null ): Promise => { const events = await db.inputEvent.findMany({ // The HTTP boundary hashes the raw client key together with profile and @@ -455,10 +464,16 @@ const expectApiInputEvent = async ( requestId, target: 'API', eventType: procedure, - payload: {}, + payload: + expected.payload === undefined + ? { + version: 1, + digest: expect.stringMatching(/^sha256:[0-9a-f]{64}$/u), + } + : createApiInputPayloadIdentity(expected.payload), actorUserId: expected.actorUserId, status: expected.status, - result: expected.status === 'SUCCEEDED' ? { ok: true } : null, + result: expected.status === 'SUCCEEDED' ? (expected.result ?? expect.objectContaining({ ok: true })) : null, error: expected.status === 'SUCCEEDED' ? null : expect.any(String), attempts: 1, lockedBy: null, @@ -523,6 +538,33 @@ const requestReservedNation = (accessToken: string, idempotencyKey: string, targ idempotencyKey, }); +const requestReservedNationBulk = (accessToken: string, idempotencyKey: string, targetGeneralId: number) => + requestTrpc('turns.reserved.setNationBulk', { + method: 'POST', + input: { + generalId: targetGeneralId, + entries: [{ turnList: [0, 1], action: '휴식', args: {} }], + expectedRevision: 0, + }, + accessToken, + idempotencyKey, + }); + +type NationReservationKind = 'single' | 'bulk'; + +const nationReservationProcedure = (kind: NationReservationKind) => + kind === 'single' ? 'turns.reserved.setNation' : 'turns.reserved.setNationBulk'; + +const requestNationReservation = ( + kind: NationReservationKind, + accessToken: string, + idempotencyKey: string, + targetGeneralId = generalId +) => + kind === 'single' + ? requestReservedNation(accessToken, idempotencyKey, targetGeneralId) + : requestReservedNationBulk(accessToken, idempotencyKey, targetGeneralId); + const ownershipDenialCases = [ { label: 'authenticated user without a general', @@ -652,6 +694,13 @@ integration('game API security over HTTP transport', () => { }); createdFixtureWorld = true; } + const reservationWorlds = await db.worldState.findMany({ select: { id: true } }); + if (reservationWorlds.length !== 1 || !reservationWorlds[0]) { + throw new Error( + `security transport fixture requires exactly one world row, got ${reservationWorlds.length}` + ); + } + reservationWorldId = reservationWorlds[0].id; await db.trafficPeriodGeneral.deleteMany({ where: { generalId: { in: fixtureGeneralIds } } }); await db.trafficPeriod.deleteMany({ where: { worldStateId: fixtureWorldId } }); await db.readModelOutbox.deleteMany(); @@ -683,6 +732,7 @@ integration('game API security over HTTP transport', () => { where: { OR: [ { domain: 'reserved.general', entityId: { in: fixtureGeneralIds } }, + { domain: 'general.content', entityId: { in: fixtureGeneralIds } }, { domain: 'dashboard.global', entityId: 0 }, ], }, @@ -704,6 +754,14 @@ integration('game API security over HTTP transport', () => { beforeEach(async () => { await deleteMatrixInputEvents(); + await db.general.update({ + where: { id: generalId }, + data: { npcState: 0, meta: {}, penalty: {} }, + }); + await db.worldState.update({ + where: { id: reservationWorldId }, + data: { meta: {} }, + }); await db.trafficPeriodGeneral.deleteMany({ where: { generalId: { in: fixtureGeneralIds } } }); await db.trafficPeriod.deleteMany({ where: { worldStateId: fixtureWorldId } }); await db.generalAccessLog.deleteMany({ where: { generalId: { in: fixtureGeneralIds } } }); @@ -715,6 +773,7 @@ integration('game API security over HTTP transport', () => { where: { OR: [ { domain: 'reserved.general', entityId: { in: fixtureGeneralIds } }, + { domain: 'general.content', entityId: { in: fixtureGeneralIds } }, { domain: 'dashboard.global', entityId: 0 }, ], }, @@ -1051,6 +1110,258 @@ integration('game API security over HTTP transport', () => { }); }); + it.each( + (['single', 'bulk'] as const).flatMap((kind) => + [ + { label: 'zero', value: 0 }, + { label: 'false', value: false }, + { label: 'null', value: null }, + ].map((penalty) => ({ kind, ...penalty })) + ) + )( + 'rejects $kind nation reservation when noChiefTurnInput is $label without committing queue or journal', + async ({ kind, label, value }) => { + const procedure = nationReservationProcedure(kind); + const idempotencyKey = `${mutationRequestPrefix}nation-penalty-${kind}-${label}`; + const accessToken = await createAccessToken(`matrix-nation-penalty-${kind}-${label}`, {}); + await db.general.update({ + where: { id: generalId }, + data: { + meta: { killturn: 3, marker: 'penalty-preserved' }, + penalty: { noChiefTurnInput: value }, + }, + }); + await db.worldState.update({ + where: { id: reservationWorldId }, + data: { meta: { killturn: 12 } }, + }); + + const result = await requestNationReservation(kind, accessToken, idempotencyKey); + + expect(result.response.status).toBe(412); + expect(result.body).toMatchObject({ + error: { + message: '수뇌 턴 입력 불가능', + data: { code: 'PRECONDITION_FAILED' }, + }, + }); + expect(await db.nationTurn.count({ where: { nationId: ownerNationId, officerLevel: 12 } })).toBe(0); + expect( + await db.nationTurnRevision.findUnique({ + where: { nationId_officerLevel: { nationId: ownerNationId, officerLevel: 12 } }, + }) + ).toBeNull(); + expect(await db.general.findUniqueOrThrow({ where: { id: generalId } })).toMatchObject({ + meta: { killturn: 3, marker: 'penalty-preserved' }, + penalty: { noChiefTurnInput: value }, + }); + expect(await db.readModelRevision.count()).toBe(0); + expect(await db.readModelOutbox.count()).toBe(0); + await expectApiInputEvent(idempotencyKey, procedure, { + actorUserId: userId, + status: 'FAILED', + }); + } + ); + + it.each([ + { kind: 'single' as const, label: 'single' }, + { kind: 'bulk' as const, label: 'bulk' }, + ])('commits $label nation queue, JSONB killturn refill, and read-model journal together', async ({ kind }) => { + const procedure = nationReservationProcedure(kind); + const idempotencyKey = `${mutationRequestPrefix}nation-refill-${kind}`; + const accessToken = await createAccessToken(`matrix-nation-refill-${kind}`, {}); + await db.general.update({ + where: { id: generalId }, + data: { + npcState: 0, + meta: { killturn: 3, marker: 'preserved-by-jsonb-set' }, + penalty: {}, + }, + }); + await db.worldState.update({ + where: { id: reservationWorldId }, + data: { meta: { killturn: 12, marker: 'world-preserved' } }, + }); + + const result = await requestNationReservation(kind, accessToken, idempotencyKey); + + expect(result.response.status).toBe(200); + expect(result.body).toMatchObject({ result: { data: { ok: true, revision: 1 } } }); + expect( + await db.nationTurn.findMany({ + where: { nationId: ownerNationId, officerLevel: 12 }, + select: { turnIdx: true, actionCode: true, arg: true }, + orderBy: { turnIdx: 'asc' }, + }) + ).toEqual( + Array.from({ length: 12 }, (_, turnIdx) => ({ + turnIdx, + actionCode: '휴식', + arg: {}, + })) + ); + expect( + await db.nationTurnRevision.findUniqueOrThrow({ + where: { nationId_officerLevel: { nationId: ownerNationId, officerLevel: 12 } }, + }) + ).toMatchObject({ revision: 1, leaseOwner: null, leaseExpiresAt: null }); + expect(await db.general.findUniqueOrThrow({ where: { id: generalId } })).toMatchObject({ + npcState: 0, + meta: { killturn: 12, marker: 'preserved-by-jsonb-set' }, + }); + expect(await db.worldState.findUniqueOrThrow({ where: { id: reservationWorldId } })).toMatchObject({ + meta: { killturn: 12, marker: 'world-preserved' }, + }); + expect( + await db.readModelRevision.findMany({ + select: { domain: true, entityId: true, revision: true }, + orderBy: [{ domain: 'asc' }, { entityId: 'asc' }], + }) + ).toEqual([ + { domain: 'dashboard.global', entityId: 0, revision: 1n }, + { domain: 'general.content', entityId: generalId, revision: 1n }, + ]); + expect(await db.readModelOutbox.findMany({ select: { payload: true } })).toEqual([ + { + payload: { + version: 1, + changes: [ + ['dashboard.global', 0, '1'], + ['general.content', generalId, '1'], + ], + }, + }, + ]); + await expectApiInputEvent(idempotencyKey, procedure, { + actorUserId: userId, + status: 'SUCCEEDED', + }); + }); + + it.each([ + { + kind: 'single' as const, + label: 'already-higher user killturn', + npcState: 0, + currentKillturn: 20, + }, + { + kind: 'bulk' as const, + label: 'NPC actor', + npcState: 2, + currentKillturn: 3, + }, + ])('keeps $label unchanged while committing its nation queue', async ({ kind, npcState, currentKillturn }) => { + const procedure = nationReservationProcedure(kind); + const idempotencyKey = `${mutationRequestPrefix}nation-refill-noop-${kind}`; + const accessToken = await createAccessToken(`matrix-nation-refill-noop-${kind}`, {}); + await db.general.update({ + where: { id: generalId }, + data: { + npcState, + meta: { killturn: currentKillturn, marker: 'no-op-preserved' }, + penalty: {}, + }, + }); + await db.worldState.update({ + where: { id: reservationWorldId }, + data: { meta: { killturn: 12 } }, + }); + + const result = await requestNationReservation(kind, accessToken, idempotencyKey); + + expect(result.response.status).toBe(200); + expect(result.body).toMatchObject({ result: { data: { ok: true, revision: 1 } } }); + expect(await db.nationTurn.count({ where: { nationId: ownerNationId, officerLevel: 12 } })).toBe(12); + expect(await db.general.findUniqueOrThrow({ where: { id: generalId } })).toMatchObject({ + npcState, + meta: { killturn: currentKillturn, marker: 'no-op-preserved' }, + }); + expect(await db.readModelRevision.count()).toBe(0); + expect(await db.readModelOutbox.count()).toBe(0); + await expectApiInputEvent(idempotencyKey, procedure, { + actorUserId: userId, + status: 'SUCCEEDED', + }); + }); + + it.each([ + { kind: 'single' as const, label: 'single' }, + { kind: 'bulk' as const, label: 'bulk' }, + ])( + 'rolls back $label queue, killturn, and read-model revisions when journal persistence fails', + async ({ kind }) => { + const procedure = nationReservationProcedure(kind); + const idempotencyKey = `${mutationRequestPrefix}nation-refill-rollback-${kind}`; + const accessToken = await createAccessToken(`matrix-nation-refill-rollback-${kind}`, {}); + await db.general.update({ + where: { id: generalId }, + data: { + npcState: 0, + meta: { killturn: 3, marker: 'rollback-preserved' }, + penalty: {}, + }, + }); + await db.worldState.update({ + where: { id: reservationWorldId }, + data: { meta: { killturn: 12 } }, + }); + await db.$executeRawUnsafe(` + CREATE FUNCTION security_transport_fail_read_model_outbox() + RETURNS trigger + LANGUAGE plpgsql + AS $$ + BEGIN + RAISE EXCEPTION 'forced security transport read-model journal failure'; + END; + $$ + `); + await db.$executeRawUnsafe(` + CREATE TRIGGER security_transport_fail_read_model_outbox + BEFORE INSERT ON read_model_outbox + FOR EACH ROW + EXECUTE FUNCTION security_transport_fail_read_model_outbox() + `); + + const result = await (async () => { + try { + return await requestNationReservation(kind, accessToken, idempotencyKey); + } finally { + await db.$executeRawUnsafe( + 'DROP TRIGGER IF EXISTS security_transport_fail_read_model_outbox ON read_model_outbox' + ); + await db.$executeRawUnsafe('DROP FUNCTION IF EXISTS security_transport_fail_read_model_outbox()'); + } + })(); + + expect(result.response.status).toBe(500); + expect(result.body).toMatchObject({ error: { data: { code: 'INTERNAL_SERVER_ERROR' } } }); + expect(await db.nationTurn.count({ where: { nationId: ownerNationId, officerLevel: 12 } })).toBe(0); + expect( + await db.nationTurnRevision.findUnique({ + where: { nationId_officerLevel: { nationId: ownerNationId, officerLevel: 12 } }, + }) + ).toBeNull(); + expect(await db.general.findUniqueOrThrow({ where: { id: generalId } })).toMatchObject({ + npcState: 0, + meta: { killturn: 3, marker: 'rollback-preserved' }, + }); + expect(await db.readModelRevision.count()).toBe(0); + expect(await db.readModelOutbox.count()).toBe(0); + await expectApiInputEvent(idempotencyKey, procedure, { + actorUserId: userId, + status: 'FAILED', + }); + expect( + await db.inputEvent.findUniqueOrThrow({ + where: { requestId: resolveScopedApiRequestId(idempotencyKey, procedure, userId) }, + select: { error: true }, + }) + ).toEqual({ error: expect.stringContaining('forced security transport read-model journal failure') }); + } + ); + it('commits an owned general reservation once with an authenticated actor and durable journal', async () => { const idempotencyKey = `${mutationRequestPrefix}general-success`; const accessToken = await createAccessToken('matrix-general-success', {}); @@ -1409,8 +1720,15 @@ integration('game API security over HTTP transport', () => { expect(await readRealtimeRedisState()).toEqual(redisBefore); }, 15_000); - it('commits an owned officer nation reservation and rejects duplicate idempotency replay without a second queue mutation', async () => { + it('commits an owned officer nation reservation and replays the durable response without a second queue mutation', async () => { const idempotencyKey = `${mutationRequestPrefix}nation-success`; + const inputPayload = { + generalId, + turnIndex: 0, + action: '휴식', + args: {}, + expectedRevision: 0, + }; const accessToken = await createAccessToken('matrix-nation-success', {}); const databaseBefore = await readReservedMutationState(); const durableBefore = await readDurableSchemaStateExcludingMatrixApiJournal(); @@ -1420,6 +1738,7 @@ integration('game API security over HTTP transport', () => { const first = await requestReservedNation(accessToken, idempotencyKey, generalId); expect(first.response.status).toBe(200); expect(first.body).toMatchObject({ result: { data: { ok: true, revision: 1 } } }); + const firstResult = (first.body as { result: { data: unknown } }).result.data; expect( await db.nationTurn.findMany({ where: { nationId: ownerNationId, officerLevel: 12 }, @@ -1443,6 +1762,8 @@ integration('game API security over HTTP transport', () => { await expectApiInputEvent(idempotencyKey, 'turns.reserved.setNation', { actorUserId: userId, status: 'SUCCEEDED', + payload: inputPayload, + result: firstResult, }); const committed = await readReservedMutationState(); @@ -1551,10 +1872,14 @@ integration('game API security over HTTP transport', () => { where: { requestId: replayRequestId }, }); const replay = await requestReservedNation(accessToken, idempotencyKey, generalId); - expect(replay.response.status).toBe(409); - expect(replay.body).toMatchObject({ error: { data: { code: 'CONFLICT' } } }); - expect(await readReservedMutationState()).toEqual(committed); - expect(await readDurableSchemaStateExcludingMatrixApiJournal()).toEqual(replayDurableBefore); + expect(replay.response.status).toBe(200); + expect(replay.body).toEqual(first.body); + const replayState = await readReservedMutationState(); + expect({ ...replayState, generalAccessLogs: committed.generalAccessLogs }).toEqual(committed); + expectSingleActorActivity(replayState.generalAccessLogs); + expect( + withoutDurableTables(await readDurableSchemaStateExcludingMatrixApiJournal(), ['general_access_log']) + ).toEqual(withoutDurableTables(replayDurableBefore, ['general_access_log'])); expect(await readRealtimeRedisState()).toEqual(replayRedisBefore); expect( await db.inputEvent.findUniqueOrThrow({ @@ -1565,6 +1890,8 @@ integration('game API security over HTTP transport', () => { await expectApiInputEvent(idempotencyKey, 'turns.reserved.setNation', { actorUserId: userId, status: 'SUCCEEDED', + payload: inputPayload, + result: firstResult, }); }); diff --git a/docs/architecture/api-input-event-replay.md b/docs/architecture/api-input-event-replay.md new file mode 100644 index 00000000..955dca89 --- /dev/null +++ b/docs/architecture/api-input-event-replay.md @@ -0,0 +1,117 @@ +# API input-event 재실행·복구 계약 + +## 목적과 범위 + +game-api mutation은 HTTP `Idempotency-Key`를 profile·인증 actor와 함께 scope한 base ID, +tRPC procedure path, batch index에서 만든 operation key를 `input_event.request_id`로 +사용한다. 이 원장은 동일 요청이 네트워크 +재시도나 응답 유실 때문에 다시 도착했을 때 업무 mutation을 두 번 실행하지 않고, +이미 성공한 응답을 그대로 재생하기 위한 durable 경계다. + +이 계약은 `target = 'API'`인 tRPC mutation에만 적용한다. turn daemon의 +`target = 'ENGINE'` 처리와 gameplay 계산·RNG 순서는 바꾸지 않는다. + +## 요청 identity와 저장 경계 + +요청 identity는 다음 네 요소가 모두 같은 경우에만 일치한다. + +- scoped request ID +- tRPC procedure path인 `event_type` +- 인증된 서버-side `actor_user_id` +- HTTP JSON decoding 뒤 `getRawInput()`이 돌려준 raw tRPC input의 canonical SHA-256 digest + +raw input을 사용하므로 parser가 strip·transform하는 field도 operation identity에는 포함된다. + +`payload`에는 원문 대신 다음처럼 고정 크기 identity envelope만 저장한다. + +```json +{ + "version": 1, + "digest": "sha256:<64 hexadecimal characters>" +} +``` + +object key 순서는 digest에 영향을 주지 않지만 배열 순서와 값은 영향을 준다. +따라서 `board.uploadImage`의 data URL, 토큰, 자유 입력처럼 크거나 민감할 수 있는 +필드를 `input_event`에 한 번 더 영구 복제하지 않는다. digest만으로 원래 입력을 +복원하거나 사후 감사할 수는 없다. 입력 원문 보존이 필요한 별도 기능은 목적에 맞는 +접근 제어와 retention을 가진 저장소를 사용해야 한다. + +성공한 업무의 실제 JSON 응답은 canonical JSON으로 `result`에 저장한다. 정확히 같은 +재요청은 업무 code를 다시 호출하지 않고 이 값을 200 응답으로 재생한다. 응답 자체에 +개인정보나 큰 payload가 들어갈 수 있으므로 `input_event` DB 접근 권한과 retention은 +별도로 제한해야 한다. 이 변경은 result retention 정책을 새로 정하지 않는다. + +## transaction과 상태 전이 + +각 요청은 하나의 PostgreSQL transaction에서 해당 row를 `FOR UPDATE`로 잠근다. + +```text +없는 row -> PENDING 삽입 -> PROCESSING(attempts + 1) + -> SAVEPOINT + -> 업무 mutation + read-model journal + -> SUCCEEDED + 실제 result -> COMMIT + -> 업무 오류: ROLLBACK TO SAVEPOINT + -> FAILED + error -> COMMIT -> 오류 반환 +``` + +`PROCESSING` 표시, 업무 mutation, `SUCCEEDED`와 result 저장은 같은 transaction에 +있다. process나 DB connection이 commit 전에 사라지면 모두 rollback되어 새로 만든 +row는 없어지거나 기존 PENDING/FAILED 상태로 되돌아간다. 업무 오류는 savepoint까지만 +rollback해 업무 write를 남기지 않으면서 FAILED와 증가한 attempts를 durable하게 +남긴다. + +transaction 결과를 client가 받지 못한 ambiguous failure도 고려한다. 별도 실패 +기록기는 row lock 아래 현재 상태를 다시 확인하며, 이미 commit된 SUCCEEDED나 다른 +실행의 PROCESSING을 FAILED로 덮지 않는다. + +상태별 처리 계약은 다음과 같다. + +| 기존 상태 | identity 일치 | 처리 | +| ----------------------- | ------------- | ------------------------------------------------------------ | +| `SUCCEEDED` | 예 | 저장된 `result`를 200으로 재생, attempts 불변 | +| `SUCCEEDED` | 아니오 | `CONFLICT`(HTTP 409), 업무 미실행 | +| `PENDING` 또는 `FAILED` | 예 | row lock 아래 claim하고 attempts를 정확히 1 증가한 뒤 재실행 | +| `PENDING` 또는 `FAILED` | 아니오 | `CONFLICT`, 업무 미실행 | +| `PROCESSING` | 무관 | fail-closed `CONFLICT`, 자동 reclaim 금지 | + +구버전이 `FAILED`, `payload = {}`, `result IS NULL`로 남긴 row는 event type과 actor가 +같을 때만 현재 digest를 최초 1회 채택해 retry할 수 있다. 반면 구버전 +`PROCESSING + payload = {}`는 identity와 활성 transaction 종료 여부를 증명할 수 +없어 age나 lease를 기준으로 자동 reclaim하지 않는다. 구버전 SUCCEEDED placeholder도 +원 응답을 복원할 수 없으므로 현재 digest와 일치하는 replay로 간주하지 않는다. + +동일 HTTP batch 안에서 같은 procedure path가 여러 번 호출될 수 있다. index 0은 기존 +호환 key인 `:`를 유지하고, 이후 호출은 +`::batch:`를 사용해 서로 충돌하지 않게 한다. + +## rolling deployment 전 확인 + +새 binary를 투입하기 전에 구 binary로 들어오는 mutation을 drain하고, 각 profile +schema에서 다음 read-only query 결과가 0인지 확인한다. + +```sql +SELECT count(*) +FROM input_event +WHERE target = 'API' + AND status = 'PROCESSING'; +``` + +0이 아니면 새 binary가 해당 row를 자동 복구하도록 두지 않는다. 구 process와 traffic을 +먼저 완전히 drain한 뒤, request별 업무 commit 여부를 확인할 수 있는 offline +reconciliation 절차를 별도로 수행한다. 생성 시각이나 processing age만 보고 status를 +바꾸거나 요청을 재실행하면 오래 실행 중인 구 transaction과 중복 mutation이 생길 수 +있다. + +## 검증 위치와 남은 경계 + +- `app/game-api/test/inputEventBoundary.test.ts`: canonical digest와 원문 비저장 +- `app/game-api/test/inputEventBoundary.integration.test.ts`: 실제 PostgreSQL row + lock, replay, conflict, retry/attempts, legacy row, concurrent race와 commit ambiguity +- `app/game-api/test/securityTransport.integration.test.ts`: 실제 HTTP/tRPC 응답 replay, + durable payload identity/result와 업무 DB/Redis side effect 불변 +- `app/game-api/test/requestId.test.ts`: 동일 path batch index 분리 + +현재 frontend가 사용자 동작별 stable idempotency key를 발급·재사용하는 계약은 이 +범위에 포함되지 않는다. 따라서 client가 재시도 때 새 base request ID를 만들면 server +원장은 두 요청을 같은 operation으로 묶을 수 없다.