fix: API 입력 이벤트의 원자적 재실행을 보장한다

2026-08-24 16:30:38 +00:00
parent 604e8fdaf2
commit cca2b3d925
9 changed files with 1141 additions and 119 deletions
@@ -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<T> =
+    { kind: 'executed'; value: T } | { kind: 'replayed'; value: T } | { kind: 'failed'; error: unknown };
+
+type SavepointDatabaseClient = InfraDatabaseClient & {
+    $executeRawUnsafe(query: string): Promise<number>;
+};
+
 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<string, unknown>).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<void> => {
+    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<LockedInputEvent> => {
+    const rows = await db.$queryRaw<LockedInputEvent[]>(
+        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<void> => {
+    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<void> => {
+    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 <T>(options: {
    db: DatabaseClient;
    requestId: string;
    eventType: string;
+    payload: unknown;
    actorUserId?: string | null;
    execute(db: DatabaseClient): Promise<T>;
 }): Promise<T> => {
-    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<T>;
    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;
 };
@@ -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<ReturnType<typeof next>> | 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({