feat: 예약 초기화 중간 버전 업데이트를 허용한다
미래 예약 RESET을 보존하면서 DB 유지 DEPLOY 한 건을 별도 queue lane에 등록한다. 예약 시각까지 시작되지 않은 중간 배포는 RESET claim 전에 자동 취소하고 관리자 화면에 이 경계를 안내한다.
This commit is contained in:
@@ -25,7 +25,11 @@ import {
|
||||
} from './adminCapabilities.js';
|
||||
import type { GatewayApiContext } from './context.js';
|
||||
import { resolveLocalAccountProfilePolicy } from './auth/localAccountPolicy.js';
|
||||
import { GATEWAY_BUILD_STATUSES, GATEWAY_PROFILE_STATUSES } from './orchestrator/profileRepository.js';
|
||||
import {
|
||||
GATEWAY_BUILD_STATUSES,
|
||||
GATEWAY_PROFILE_STATUSES,
|
||||
GatewayProfileOperationConflictError,
|
||||
} from './orchestrator/profileRepository.js';
|
||||
import { readProfileReleaseSource } from './orchestrator/profileReleaseSource.js';
|
||||
import {
|
||||
orderGatewayProfiles,
|
||||
@@ -469,6 +473,9 @@ const zRuntimeSettings = z
|
||||
const isUniqueConstraintError = (error: unknown): boolean =>
|
||||
Boolean(error && typeof error === 'object' && 'code' in error && error.code === 'P2002');
|
||||
|
||||
const isProfileOperationConflictError = (error: unknown): boolean =>
|
||||
error instanceof GatewayProfileOperationConflictError || isUniqueConstraintError(error);
|
||||
|
||||
const zInstallOptions = z.object({
|
||||
scenarioId: z.number().int().min(0),
|
||||
turnTermMinutes: z
|
||||
@@ -1343,7 +1350,7 @@ export const adminRouter = router({
|
||||
});
|
||||
return operation;
|
||||
} catch (error) {
|
||||
if (!isUniqueConstraintError(error)) {
|
||||
if (!isProfileOperationConflictError(error)) {
|
||||
throw error;
|
||||
}
|
||||
throw new TRPCError({
|
||||
@@ -1399,7 +1406,7 @@ export const adminRouter = router({
|
||||
requestedBy: adminAuth.user.id,
|
||||
});
|
||||
} catch (error) {
|
||||
if (!isUniqueConstraintError(error)) {
|
||||
if (!isProfileOperationConflictError(error)) {
|
||||
throw new TRPCError({
|
||||
code: 'BAD_REQUEST',
|
||||
message: 'The active Gateway release commit cannot be resolved for game cancellation.',
|
||||
@@ -1460,7 +1467,7 @@ export const adminRouter = router({
|
||||
requestedBy: adminAuth.user.id,
|
||||
});
|
||||
} catch (error) {
|
||||
if (!isUniqueConstraintError(error)) {
|
||||
if (!isProfileOperationConflictError(error)) {
|
||||
throw error;
|
||||
}
|
||||
throw new TRPCError({
|
||||
@@ -1505,7 +1512,7 @@ export const adminRouter = router({
|
||||
});
|
||||
return operation;
|
||||
} catch (error) {
|
||||
if (!isUniqueConstraintError(error)) {
|
||||
if (!isProfileOperationConflictError(error)) {
|
||||
throw error;
|
||||
}
|
||||
throw new TRPCError({
|
||||
@@ -1575,7 +1582,7 @@ export const adminRouter = router({
|
||||
if (error instanceof TRPCError) {
|
||||
throw error;
|
||||
}
|
||||
if (!isUniqueConstraintError(error)) {
|
||||
if (!isProfileOperationConflictError(error)) {
|
||||
throw error;
|
||||
}
|
||||
throw new TRPCError({
|
||||
@@ -2114,7 +2121,7 @@ export const adminRouter = router({
|
||||
});
|
||||
return { ok: true, operationId: operation.id, action: actionRecord };
|
||||
} catch (error) {
|
||||
if (!isUniqueConstraintError(error)) {
|
||||
if (!isProfileOperationConflictError(error)) {
|
||||
throw error;
|
||||
}
|
||||
throw new TRPCError({
|
||||
@@ -2197,7 +2204,7 @@ export const adminRouter = router({
|
||||
message: 'Profile install operation did not complete in time.',
|
||||
});
|
||||
} catch (error) {
|
||||
if (!isUniqueConstraintError(error)) {
|
||||
if (!isProfileOperationConflictError(error)) {
|
||||
throw error;
|
||||
}
|
||||
throw new TRPCError({
|
||||
|
||||
@@ -3,6 +3,13 @@ import type { GatewayPrisma, GatewayPrismaClient } from '@sammo-ts/infra';
|
||||
|
||||
export const CONTROL_PLANE_OPERATION_CLAIM_LOCK = 'gateway_control_plane_operation_claim';
|
||||
|
||||
export class GatewayProfileOperationConflictError extends Error {
|
||||
constructor() {
|
||||
super('This profile already has an incompatible queued or running operation.');
|
||||
this.name = 'GatewayProfileOperationConflictError';
|
||||
}
|
||||
}
|
||||
|
||||
export { GATEWAY_PROFILE_STATUSES, type GatewayProfileStatus };
|
||||
|
||||
export const GATEWAY_BUILD_STATUSES = ['IDLE', 'QUEUED', 'RUNNING', 'FAILED', 'SUCCEEDED'] as const;
|
||||
@@ -352,6 +359,24 @@ const mapOperationLog = (row: {
|
||||
createdAt: row.createdAt.toISOString(),
|
||||
});
|
||||
|
||||
const canQueueAlongsideActiveOperations = (
|
||||
input: Pick<GatewayOperationCreateInput, 'type' | 'scheduledAt'>,
|
||||
activeOperations: GatewayOperationRow[],
|
||||
now: Date
|
||||
): boolean => {
|
||||
if (activeOperations.length === 0) return true;
|
||||
if (input.type !== 'DEPLOY' || input.scheduledAt || activeOperations.length !== 1) return false;
|
||||
|
||||
const [reservedReset] = activeOperations;
|
||||
return Boolean(
|
||||
reservedReset &&
|
||||
reservedReset.type === 'RESET' &&
|
||||
reservedReset.status === 'QUEUED' &&
|
||||
reservedReset.scheduledAt &&
|
||||
reservedReset.scheduledAt.getTime() > now.getTime()
|
||||
);
|
||||
};
|
||||
|
||||
export const createGatewayProfileRepository = (prisma: GatewayPrismaClient): GatewayProfileRepository => ({
|
||||
async listProfiles(): Promise<GatewayProfileRecord[]> {
|
||||
const rows = await prisma.gatewayProfile.findMany({
|
||||
@@ -620,6 +645,21 @@ export const createGatewayProfileRepository = (prisma: GatewayPrismaClient): Gat
|
||||
},
|
||||
async createOperation(input: GatewayOperationCreateInput): Promise<GatewayOperationRecord> {
|
||||
const row = await prisma.$transaction(async (tx) => {
|
||||
await tx.$queryRaw<Array<{ lock_result: string }>>`
|
||||
SELECT pg_advisory_xact_lock(
|
||||
hashtextextended(${CONTROL_PLANE_OPERATION_CLAIM_LOCK}, 0)
|
||||
)::text AS lock_result
|
||||
`;
|
||||
const activeOperations = await tx.gatewayOperation.findMany({
|
||||
where: {
|
||||
profileName: input.profileName,
|
||||
status: { in: ['QUEUED', 'RUNNING'] },
|
||||
},
|
||||
orderBy: [{ createdAt: 'asc' }, { id: 'asc' }],
|
||||
});
|
||||
if (!canQueueAlongsideActiveOperations(input, activeOperations, new Date())) {
|
||||
throw new GatewayProfileOperationConflictError();
|
||||
}
|
||||
const operation = await tx.gatewayOperation.create({
|
||||
data: {
|
||||
profileName: input.profileName,
|
||||
@@ -675,6 +715,37 @@ export const createGatewayProfileRepository = (prisma: GatewayPrismaClient): Gat
|
||||
if (running && !runningIsStale) {
|
||||
return null;
|
||||
}
|
||||
const expiredInterimDeploys = await tx.$queryRaw<Array<{ id: string }>>`
|
||||
UPDATE "gateway_operation" AS deploy
|
||||
SET "status" = 'CANCELLED',
|
||||
"completed_at" = ${now},
|
||||
"lease_owner" = NULL,
|
||||
"lease_until" = NULL,
|
||||
"heartbeat_at" = NULL,
|
||||
"updated_at" = ${now}
|
||||
WHERE deploy."status" = 'QUEUED'
|
||||
AND deploy."type" = 'DEPLOY'
|
||||
AND EXISTS (
|
||||
SELECT 1
|
||||
FROM "gateway_operation" AS reset
|
||||
WHERE reset."profile_name" = deploy."profile_name"
|
||||
AND reset."status" = 'QUEUED'
|
||||
AND reset."type" = 'RESET'
|
||||
AND reset."scheduled_at" IS NOT NULL
|
||||
AND reset."scheduled_at" <= ${now}
|
||||
)
|
||||
RETURNING deploy."id"
|
||||
`;
|
||||
if (expiredInterimDeploys.length) {
|
||||
await tx.gatewayOperationLog.createMany({
|
||||
data: expiredInterimDeploys.map(({ id }) => ({
|
||||
operationId: id,
|
||||
level: 'INFO',
|
||||
phase: 'cancel',
|
||||
message: '예약 시나리오 초기화 시각이 되어 실행 전 중간 버전 업데이트를 자동 취소했습니다.',
|
||||
})),
|
||||
});
|
||||
}
|
||||
const candidate =
|
||||
running ??
|
||||
(await tx.gatewayOperation.findFirst({
|
||||
@@ -887,10 +958,31 @@ export const createGatewayProfileRepository = (prisma: GatewayPrismaClient): Gat
|
||||
},
|
||||
async retryOperation(id: string, requestedBy: string): Promise<GatewayOperationRecord | null> {
|
||||
const row = await prisma.$transaction(async (tx) => {
|
||||
await tx.$queryRaw<Array<{ lock_result: string }>>`
|
||||
SELECT pg_advisory_xact_lock(
|
||||
hashtextextended(${CONTROL_PLANE_OPERATION_CLAIM_LOCK}, 0)
|
||||
)::text AS lock_result
|
||||
`;
|
||||
const previous = await tx.gatewayOperation.findUnique({ where: { id } });
|
||||
if (!previous || (previous.status !== 'FAILED' && previous.status !== 'CANCELLED')) {
|
||||
return null;
|
||||
}
|
||||
const activeOperations = await tx.gatewayOperation.findMany({
|
||||
where: {
|
||||
profileName: previous.profileName,
|
||||
status: { in: ['QUEUED', 'RUNNING'] },
|
||||
},
|
||||
orderBy: [{ createdAt: 'asc' }, { id: 'asc' }],
|
||||
});
|
||||
if (
|
||||
!canQueueAlongsideActiveOperations(
|
||||
{ type: previous.type, scheduledAt: undefined },
|
||||
activeOperations,
|
||||
new Date()
|
||||
)
|
||||
) {
|
||||
throw new GatewayProfileOperationConflictError();
|
||||
}
|
||||
const previousPayload = previous.payload as GatewayPrisma.JsonObject;
|
||||
const retrySource = buildRetryOperationSource(previous);
|
||||
const operation = await tx.gatewayOperation.create({
|
||||
|
||||
@@ -52,7 +52,7 @@ describeDatabase('gateway operation lease and profile serialization', () => {
|
||||
await connector.disconnect();
|
||||
});
|
||||
|
||||
it('allows only one queued or running operation for a profile', async () => {
|
||||
it('allows only one incompatible queued or running operation for a profile', async () => {
|
||||
const results = await Promise.allSettled([
|
||||
repository.createOperation({
|
||||
profileName,
|
||||
@@ -73,6 +73,118 @@ describeDatabase('gateway operation lease and profile serialization', () => {
|
||||
await expect(repository.listOperations({ profileName })).resolves.toHaveLength(1);
|
||||
});
|
||||
|
||||
it('keeps a future scheduled RESET while one interim DEPLOY runs and completes first', async () => {
|
||||
const scheduledAt = new Date('2099-01-01T00:00:00.000Z');
|
||||
const scheduledReset = await repository.createOperation({
|
||||
profileName,
|
||||
type: 'RESET',
|
||||
sourceMode: 'BRANCH',
|
||||
sourceRef: 'main',
|
||||
scheduledAt: scheduledAt.toISOString(),
|
||||
requestedBy: 'reset-admin',
|
||||
});
|
||||
const interimDeploy = await repository.createOperation({
|
||||
profileName,
|
||||
type: 'DEPLOY',
|
||||
sourceMode: 'BRANCH',
|
||||
sourceRef: 'main',
|
||||
requestedBy: 'deploy-admin',
|
||||
});
|
||||
|
||||
await expect(
|
||||
repository.createOperation({
|
||||
profileName,
|
||||
type: 'STOP',
|
||||
requestedBy: 'runtime-admin',
|
||||
})
|
||||
).rejects.toThrow('incompatible queued or running operation');
|
||||
await expect(repository.listOperations({ profileName })).resolves.toHaveLength(2);
|
||||
|
||||
const beforeReset = new Date('2098-12-31T23:00:00.000Z');
|
||||
await expect(
|
||||
repository.claimNextOperation(beforeReset, { ownerId: 'worker-a', durationMs: 1_000 })
|
||||
).resolves.toMatchObject({ id: interimDeploy.id, type: 'DEPLOY', status: 'RUNNING' });
|
||||
await repository.completeOperation(interimDeploy.id, 'SUCCEEDED', { error: null }, 'worker-a');
|
||||
await expect(repository.getOperation(scheduledReset.id)).resolves.toMatchObject({
|
||||
status: 'QUEUED',
|
||||
scheduledAt: scheduledAt.toISOString(),
|
||||
});
|
||||
await expect(
|
||||
repository.claimNextOperation(beforeReset, { ownerId: 'worker-b', durationMs: 1_000 })
|
||||
).resolves.toBeNull();
|
||||
await expect(
|
||||
repository.claimNextOperation(scheduledAt, { ownerId: 'worker-b', durationMs: 1_000 })
|
||||
).resolves.toMatchObject({ id: scheduledReset.id, type: 'RESET', status: 'RUNNING' });
|
||||
});
|
||||
|
||||
it('accepts only one of two concurrent interim DEPLOY requests beside a scheduled RESET', async () => {
|
||||
await repository.createOperation({
|
||||
profileName,
|
||||
type: 'RESET',
|
||||
sourceMode: 'BRANCH',
|
||||
sourceRef: 'main',
|
||||
scheduledAt: new Date('2099-01-01T00:00:00.000Z').toISOString(),
|
||||
requestedBy: 'reset-admin',
|
||||
});
|
||||
|
||||
const results = await Promise.allSettled([
|
||||
repository.createOperation({
|
||||
profileName,
|
||||
type: 'DEPLOY',
|
||||
sourceMode: 'BRANCH',
|
||||
sourceRef: 'main',
|
||||
requestedBy: 'deploy-admin-a',
|
||||
}),
|
||||
repository.createOperation({
|
||||
profileName,
|
||||
type: 'DEPLOY',
|
||||
sourceMode: 'BRANCH',
|
||||
sourceRef: 'main',
|
||||
requestedBy: 'deploy-admin-b',
|
||||
}),
|
||||
]);
|
||||
|
||||
expect(results.filter((result) => result.status === 'fulfilled')).toHaveLength(1);
|
||||
expect(results.filter((result) => result.status === 'rejected')).toHaveLength(1);
|
||||
await expect(repository.listOperations({ profileName })).resolves.toHaveLength(2);
|
||||
});
|
||||
|
||||
it('auto-cancels an interim DEPLOY that is still queued when the scheduled RESET becomes due', async () => {
|
||||
const scheduledAt = new Date('2099-01-01T00:00:00.000Z');
|
||||
const scheduledReset = await repository.createOperation({
|
||||
profileName,
|
||||
type: 'RESET',
|
||||
sourceMode: 'COMMIT',
|
||||
sourceRef: 'a'.repeat(40),
|
||||
scheduledAt: scheduledAt.toISOString(),
|
||||
requestedBy: 'reset-admin',
|
||||
});
|
||||
const interimDeploy = await repository.createOperation({
|
||||
profileName,
|
||||
type: 'DEPLOY',
|
||||
sourceMode: 'COMMIT',
|
||||
sourceRef: 'b'.repeat(40),
|
||||
requestedBy: 'deploy-admin',
|
||||
});
|
||||
|
||||
await expect(
|
||||
repository.claimNextOperation(scheduledAt, { ownerId: 'worker-a', durationMs: 1_000 })
|
||||
).resolves.toMatchObject({ id: scheduledReset.id, type: 'RESET', status: 'RUNNING' });
|
||||
await expect(repository.getOperation(interimDeploy.id)).resolves.toMatchObject({
|
||||
status: 'CANCELLED',
|
||||
completedAt: scheduledAt.toISOString(),
|
||||
});
|
||||
await expect(repository.listOperationLogs(interimDeploy.id)).resolves.toEqual([
|
||||
expect.objectContaining({
|
||||
phase: 'queue',
|
||||
}),
|
||||
expect.objectContaining({
|
||||
phase: 'cancel',
|
||||
message: expect.stringContaining('중간 버전 업데이트를 자동 취소'),
|
||||
}),
|
||||
]);
|
||||
});
|
||||
|
||||
it('stores durable cursor logs for profile operations', async () => {
|
||||
const operation = await repository.createOperation({
|
||||
profileName,
|
||||
|
||||
@@ -38,7 +38,7 @@ describe('readReleaseManifest', () => {
|
||||
|
||||
await expect(readReleaseManifest(workspaceRoot)).resolves.toMatchObject({
|
||||
controllerProtocol: RELEASE_CONTROLLER_PROTOCOL,
|
||||
gatewaySchemaHead: '20260823010000_add_web_push_notifications',
|
||||
gatewaySchemaHead: '20260824090000_allow_interim_profile_deploy',
|
||||
gameSchemaHead: '20260824080000_vote_utc_wall_timestamps',
|
||||
});
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user