접속 집계 교착으로 인한 턴 정지를 방지
This commit is contained in:
@@ -1,7 +1,7 @@
|
|||||||
import { createHash, randomUUID } from 'node:crypto';
|
import { createHash, randomUUID } from 'node:crypto';
|
||||||
|
|
||||||
import { asRecord, resolveAccessLimitLevel, resolveAccessRefreshLimit } from '@sammo-ts/common';
|
import { asRecord, resolveAccessLimitLevel, resolveAccessRefreshLimit } from '@sammo-ts/common';
|
||||||
import { GamePrisma } from '@sammo-ts/infra';
|
import { acquireGameSchemaAdvisoryXactLock, GENERAL_ACCESS_PERSISTENCE_LOCK, GamePrisma } from '@sammo-ts/infra';
|
||||||
import type { GameSessionTokenPayload } from '@sammo-ts/common/auth/gameToken';
|
import type { GameSessionTokenPayload } from '@sammo-ts/common/auth/gameToken';
|
||||||
|
|
||||||
import type { DatabaseClient } from '../context.js';
|
import type { DatabaseClient } from '../context.js';
|
||||||
@@ -163,7 +163,7 @@ const markBatchProcessed = async (db: Pick<DatabaseClient, '$executeRaw'>, batch
|
|||||||
};
|
};
|
||||||
|
|
||||||
export const flushDeferredGeneralAccessBatch = async (
|
export const flushDeferredGeneralAccessBatch = async (
|
||||||
db: Pick<DatabaseClient, '$queryRaw' | '$executeRaw' | 'worldState'>,
|
db: Pick<DatabaseClient, '$transaction' | '$queryRaw' | '$executeRaw' | 'worldState'>,
|
||||||
batchId: string,
|
batchId: string,
|
||||||
entries: readonly DeferredGeneralAccessEntry[]
|
entries: readonly DeferredGeneralAccessEntry[]
|
||||||
): Promise<{ states: DeferredGeneralAccessFlushRow[]; refreshLimit: number }> => {
|
): Promise<{ states: DeferredGeneralAccessFlushRow[]; refreshLimit: number }> => {
|
||||||
@@ -171,6 +171,9 @@ export const flushDeferredGeneralAccessBatch = async (
|
|||||||
await markBatchProcessed(db, batchId);
|
await markBatchProcessed(db, batchId);
|
||||||
return { states: [], refreshLimit: 0 };
|
return { states: [], refreshLimit: 0 };
|
||||||
}
|
}
|
||||||
|
if (!db.$transaction) {
|
||||||
|
throw new Error('Deferred general access persistence requires transaction support.');
|
||||||
|
}
|
||||||
const worldState = await db.worldState.findFirst({
|
const worldState = await db.worldState.findFirst({
|
||||||
orderBy: { id: 'asc' },
|
orderBy: { id: 'asc' },
|
||||||
select: {
|
select: {
|
||||||
@@ -210,7 +213,11 @@ export const flushDeferredGeneralAccessBatch = async (
|
|||||||
}))
|
}))
|
||||||
);
|
);
|
||||||
const periodKey = worldState.currentYear * 12 + worldState.currentMonth - 1;
|
const periodKey = worldState.currentYear * 12 + worldState.currentMonth - 1;
|
||||||
const states = await db.$queryRaw<DeferredGeneralAccessFlushRow[]>(GamePrisma.sql`
|
const states = await db.$transaction(async (transaction) => {
|
||||||
|
// The turn flush and synchronous access path must enter through the same
|
||||||
|
// schema-scoped lock before either traffic or access rows are mutated.
|
||||||
|
await acquireGameSchemaAdvisoryXactLock(transaction, GENERAL_ACCESS_PERSISTENCE_LOCK);
|
||||||
|
return transaction.$queryRaw<DeferredGeneralAccessFlushRow[]>(GamePrisma.sql`
|
||||||
WITH inserted_batch AS (
|
WITH inserted_batch AS (
|
||||||
INSERT INTO "general_access_batch" ("id")
|
INSERT INTO "general_access_batch" ("id")
|
||||||
VALUES (${batchId})
|
VALUES (${batchId})
|
||||||
@@ -353,6 +360,7 @@ export const flushDeferredGeneralAccessBatch = async (
|
|||||||
END,
|
END,
|
||||||
resolved."weight"
|
resolved."weight"
|
||||||
FROM resolved
|
FROM resolved
|
||||||
|
ORDER BY resolved."general_id"
|
||||||
ON CONFLICT ("general_id") DO UPDATE SET
|
ON CONFLICT ("general_id") DO UPDATE SET
|
||||||
"user_id" = EXCLUDED."user_id",
|
"user_id" = EXCLUDED."user_id",
|
||||||
"last_refresh" = GREATEST("general_access_log"."last_refresh", EXCLUDED."last_refresh"),
|
"last_refresh" = GREATEST("general_access_log"."last_refresh", EXCLUDED."last_refresh"),
|
||||||
@@ -394,7 +402,8 @@ export const flushDeferredGeneralAccessBatch = async (
|
|||||||
LEFT JOIN access_updates ON access_updates."general_id" = actor."id"
|
LEFT JOIN access_updates ON access_updates."general_id" = actor."id"
|
||||||
LEFT JOIN "general_access_log" AS access_log ON access_log."general_id" = actor."id"
|
LEFT JOIN "general_access_log" AS access_log ON access_log."general_id" = actor."id"
|
||||||
ORDER BY actor."id"
|
ORDER BY actor."id"
|
||||||
`);
|
`);
|
||||||
|
});
|
||||||
return {
|
return {
|
||||||
states,
|
states,
|
||||||
refreshLimit: resolveAccessRefreshLimit(worldState.tickSeconds, meta.refreshLimit),
|
refreshLimit: resolveAccessRefreshLimit(worldState.tickSeconds, meta.refreshLimit),
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
import { asRecord, resolveAccessLimitLevel, resolveAccessRefreshLimit, type AccessLimitLevel } from '@sammo-ts/common';
|
import { asRecord, resolveAccessLimitLevel, resolveAccessRefreshLimit, type AccessLimitLevel } from '@sammo-ts/common';
|
||||||
import { GamePrisma } from '@sammo-ts/infra';
|
import { acquireGameSchemaAdvisoryXactLock, GENERAL_ACCESS_PERSISTENCE_LOCK, GamePrisma } from '@sammo-ts/infra';
|
||||||
|
|
||||||
import type { GameApiContext } from '../context.js';
|
import type { GameApiContext } from '../context.js';
|
||||||
|
|
||||||
@@ -228,6 +228,7 @@ export const upsertGeneralAccess = async (
|
|||||||
throw new Error('Traffic access persistence requires transaction support.');
|
throw new Error('Traffic access persistence requires transaction support.');
|
||||||
}
|
}
|
||||||
await db.$transaction(async (transaction) => {
|
await db.$transaction(async (transaction) => {
|
||||||
|
await acquireGameSchemaAdvisoryXactLock(transaction, GENERAL_ACCESS_PERSISTENCE_LOCK);
|
||||||
const periodKey = input.year * 12 + input.month - 1;
|
const periodKey = input.year * 12 + input.month - 1;
|
||||||
const periodRows = await transaction.$queryRaw<Array<{ id: number }>>(
|
const periodRows = await transaction.$queryRaw<Array<{ id: number }>>(
|
||||||
GamePrisma.sql`
|
GamePrisma.sql`
|
||||||
|
|||||||
@@ -114,6 +114,7 @@ liveIntegration('deferred general access with PostgreSQL and Redis', () => {
|
|||||||
},
|
},
|
||||||
}),
|
}),
|
||||||
},
|
},
|
||||||
|
$transaction: db.$transaction.bind(db),
|
||||||
$queryRaw: db.$queryRaw.bind(db),
|
$queryRaw: db.$queryRaw.bind(db),
|
||||||
$executeRaw: db.$executeRaw.bind(db),
|
$executeRaw: db.$executeRaw.bind(db),
|
||||||
};
|
};
|
||||||
|
|||||||
@@ -88,6 +88,9 @@ describe('deferred general access', () => {
|
|||||||
$queryRaw: queryRaw,
|
$queryRaw: queryRaw,
|
||||||
$executeRaw: vi.fn(async () => 1),
|
$executeRaw: vi.fn(async () => 1),
|
||||||
};
|
};
|
||||||
|
Object.assign(db, {
|
||||||
|
$transaction: vi.fn(async (run: (transaction: typeof db) => Promise<unknown>) => run(db)),
|
||||||
|
});
|
||||||
|
|
||||||
await expect(
|
await expect(
|
||||||
flushDeferredGeneralAccessBatch(db as never, 'batch-1', [
|
flushDeferredGeneralAccessBatch(db as never, 'batch-1', [
|
||||||
@@ -101,12 +104,14 @@ describe('deferred general access', () => {
|
|||||||
).resolves.toMatchObject({ refreshLimit: 50, states: [{ refreshScore: 2 }] });
|
).resolves.toMatchObject({ refreshLimit: 50, states: [{ refreshScore: 2 }] });
|
||||||
|
|
||||||
expect(queryRaw).toHaveBeenCalledTimes(1);
|
expect(queryRaw).toHaveBeenCalledTimes(1);
|
||||||
|
expect(db.$executeRaw).toHaveBeenCalledTimes(1);
|
||||||
const statement = queryRaw.mock.calls[0]?.[0] as { sql: string };
|
const statement = queryRaw.mock.calls[0]?.[0] as { sql: string };
|
||||||
expect(statement.sql).toContain('INSERT INTO "general_access_batch"');
|
expect(statement.sql).toContain('INSERT INTO "general_access_batch"');
|
||||||
expect(statement.sql).toContain('jsonb_to_recordset');
|
expect(statement.sql).toContain('jsonb_to_recordset');
|
||||||
expect(statement.sql).toContain('INSERT INTO "traffic_period"');
|
expect(statement.sql).toContain('INSERT INTO "traffic_period"');
|
||||||
expect(statement.sql).toContain('INSERT INTO "traffic_period_general"');
|
expect(statement.sql).toContain('INSERT INTO "traffic_period_general"');
|
||||||
expect(statement.sql).toContain('INSERT INTO "general_access_log"');
|
expect(statement.sql).toContain('INSERT INTO "general_access_log"');
|
||||||
|
expect(statement.sql).toContain('ORDER BY resolved."general_id"');
|
||||||
expect(statement.sql).not.toContain('read_model_revision');
|
expect(statement.sql).not.toContain('read_model_revision');
|
||||||
expect(statement.sql).not.toContain('read_model_outbox');
|
expect(statement.sql).not.toContain('read_model_outbox');
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -194,7 +194,11 @@ describe('general access tracking', () => {
|
|||||||
});
|
});
|
||||||
expect(transaction).toHaveBeenCalledTimes(1);
|
expect(transaction).toHaveBeenCalledTimes(1);
|
||||||
expect(queryRaw).toHaveBeenCalledTimes(1);
|
expect(queryRaw).toHaveBeenCalledTimes(1);
|
||||||
expect(executeRaw).toHaveBeenCalledTimes(2);
|
expect(executeRaw).toHaveBeenCalledTimes(3);
|
||||||
|
|
||||||
|
const lockStatement = executeRaw.mock.calls[0]![0] as { sql: string; values: unknown[] };
|
||||||
|
expect(lockStatement.sql).toContain('pg_advisory_xact_lock');
|
||||||
|
expect(lockStatement.values).toContain('general-access:persistence');
|
||||||
|
|
||||||
const periodStatement = queryRaw.mock.calls[0]![0] as { sql: string; values: unknown[] };
|
const periodStatement = queryRaw.mock.calls[0]![0] as { sql: string; values: unknown[] };
|
||||||
expect(periodStatement.sql).toContain('INSERT INTO traffic_period');
|
expect(periodStatement.sql).toContain('INSERT INTO traffic_period');
|
||||||
@@ -208,12 +212,12 @@ describe('general access tracking', () => {
|
|||||||
expect(periodStatement.values).toContain(now);
|
expect(periodStatement.values).toContain(now);
|
||||||
expect(periodStatement.values).toContainEqual(new Date('2026-07-26T03:00:00.000Z'));
|
expect(periodStatement.values).toContainEqual(new Date('2026-07-26T03:00:00.000Z'));
|
||||||
|
|
||||||
const memberStatement = executeRaw.mock.calls[0]![0] as { sql: string; values: unknown[] };
|
const memberStatement = executeRaw.mock.calls[1]![0] as { sql: string; values: unknown[] };
|
||||||
expect(memberStatement.sql).toContain('INSERT INTO traffic_period_general');
|
expect(memberStatement.sql).toContain('INSERT INTO traffic_period_general');
|
||||||
expect(memberStatement.sql).toContain('ON CONFLICT (period_id, general_id) DO NOTHING');
|
expect(memberStatement.sql).toContain('ON CONFLICT (period_id, general_id) DO NOTHING');
|
||||||
expect(memberStatement.sql).toContain('SET online = traffic_period.online');
|
expect(memberStatement.sql).toContain('SET online = traffic_period.online');
|
||||||
|
|
||||||
const accessStatement = executeRaw.mock.calls[1]![0] as { sql: string; values: unknown[] };
|
const accessStatement = executeRaw.mock.calls[2]![0] as { sql: string; values: unknown[] };
|
||||||
expect(accessStatement.sql).toContain('ON CONFLICT (general_id) DO UPDATE');
|
expect(accessStatement.sql).toContain('ON CONFLICT (general_id) DO UPDATE');
|
||||||
expect(accessStatement.sql).toContain('general_access_log.refresh + EXCLUDED.refresh');
|
expect(accessStatement.sql).toContain('general_access_log.refresh + EXCLUDED.refresh');
|
||||||
expect(accessStatement.values).toContain(7);
|
expect(accessStatement.values).toContain(7);
|
||||||
@@ -230,9 +234,9 @@ describe('general access tracking', () => {
|
|||||||
await expect(recordGeneralAccessWeight(accessContext(db), 0, now)).resolves.toBe(true);
|
await expect(recordGeneralAccessWeight(accessContext(db), 0, now)).resolves.toBe(true);
|
||||||
expect(transaction).toHaveBeenCalledTimes(1);
|
expect(transaction).toHaveBeenCalledTimes(1);
|
||||||
expect((queryRaw.mock.calls[0]![0] as { values: unknown[] }).values).toContain(0);
|
expect((queryRaw.mock.calls[0]![0] as { values: unknown[] }).values).toContain(0);
|
||||||
expect((executeRaw.mock.calls[0]![0] as { values: unknown[] }).values).toContain(0);
|
|
||||||
expect((executeRaw.mock.calls[1]![0] as { values: unknown[] }).values).toContain(0);
|
expect((executeRaw.mock.calls[1]![0] as { values: unknown[] }).values).toContain(0);
|
||||||
expect((executeRaw.mock.calls[1]![0] as { values: unknown[] }).values).toContain(now);
|
expect((executeRaw.mock.calls[2]![0] as { values: unknown[] }).values).toContain(0);
|
||||||
|
expect((executeRaw.mock.calls[2]![0] as { values: unknown[] }).values).toContain(now);
|
||||||
});
|
});
|
||||||
|
|
||||||
it('blocks above the strict limit and lazily clears a score from before the own turn', async () => {
|
it('blocks above the strict limit and lazily clears a score from before the own turn', async () => {
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
import {
|
import {
|
||||||
acquireGameSchemaAdvisoryXactLock,
|
acquireGameSchemaAdvisoryXactLock,
|
||||||
createGamePostgresConnector,
|
createGamePostgresConnector,
|
||||||
|
GENERAL_ACCESS_PERSISTENCE_LOCK,
|
||||||
GamePrisma,
|
GamePrisma,
|
||||||
writeReadModelChangeJournal,
|
writeReadModelChangeJournal,
|
||||||
enqueuePrivateMessageWebPush,
|
enqueuePrivateMessageWebPush,
|
||||||
@@ -1322,6 +1323,20 @@ export const createDatabaseTurnHooks = async (
|
|||||||
const beforeLifecycleLogs = pendingInheritanceLogs.filter((entry) => entry.phase !== 'after_lifecycle');
|
const beforeLifecycleLogs = pendingInheritanceLogs.filter((entry) => entry.phase !== 'after_lifecycle');
|
||||||
const afterLifecycleLogs = pendingInheritanceLogs.filter((entry) => entry.phase === 'after_lifecycle');
|
const afterLifecycleLogs = pendingInheritanceLogs.filter((entry) => entry.phase === 'after_lifecycle');
|
||||||
|
|
||||||
|
const writesGeneralAccess =
|
||||||
|
accessScoreResetGeneralIds.length > 0 ||
|
||||||
|
lifecycleEvents.length > 0 ||
|
||||||
|
deletedGenerals.length > 0 ||
|
||||||
|
generals.some(
|
||||||
|
(general) =>
|
||||||
|
typeof general.refreshScoreTotal === 'number' && Number.isFinite(general.refreshScoreTotal)
|
||||||
|
);
|
||||||
|
if (writesGeneralAccess) {
|
||||||
|
// API access writers acquire this before traffic/access rows.
|
||||||
|
// Match that order before lifecycle and monthly score writes.
|
||||||
|
await acquireGameSchemaAdvisoryXactLock(prisma, GENERAL_ACCESS_PERSISTENCE_LOCK);
|
||||||
|
}
|
||||||
|
|
||||||
await persistInheritancePointAdjustments(beforeLifecycleAdjustments);
|
await persistInheritancePointAdjustments(beforeLifecycleAdjustments);
|
||||||
await persistInheritanceLogs(beforeLifecycleLogs);
|
await persistInheritanceLogs(beforeLifecycleLogs);
|
||||||
await persistGeneralLifecycleEvents(
|
await persistGeneralLifecycleEvents(
|
||||||
|
|||||||
@@ -6,6 +6,9 @@ interface TryLockRow {
|
|||||||
acquired: boolean;
|
acquired: boolean;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** Serializes score/traffic writers whose table lock order otherwise differs by entry point. */
|
||||||
|
export const GENERAL_ACCESS_PERSISTENCE_LOCK = 'general-access:persistence';
|
||||||
|
|
||||||
const lockKeySql = (logicalKey: string): GamePrisma.Sql =>
|
const lockKeySql = (logicalKey: string): GamePrisma.Sql =>
|
||||||
GamePrisma.sql`hashtextextended(current_schema() || chr(31) || ${logicalKey}, 0)`;
|
GamePrisma.sql`hashtextextended(current_schema() || chr(31) || ${logicalKey}, 0)`;
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user