devsam/core2026
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 일괄 릴리스 실행 계획을 추가한다

Browse Source 
Gateway와 권한 있는 프로필 작업을 하나의 고정 커밋과 순서로 영속화한다. 앞 대상 성공 뒤 다음 작업을 claim하고 실패 대상은 같은 작업 ID로 재시도한다.
This commit is contained in:
Hide_D
2026-08-25 09:19:12 +00:00
parent ac48f2ae5b
commit 9613b482c1
11 changed files with 609 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/gateway-api/src/orchestrator/gatewayReleaseRepository.ts
+70 -5
View File
@@ -36,6 +36,8 @@ export interface GatewayReleaseOperationRecord {
leaseUntil?: string;
heartbeatAt?: string;
attempts: number;
bulkReleaseId?: string;
bulkOrder?: number;
createdAt: string;
updatedAt: string;
}
@@ -135,6 +137,8 @@ const mapOperation = (row: {
leaseUntil: Date | null;
heartbeatAt: Date | null;
attempts: number;
bulkReleaseId: string | null;
bulkOrder: number | null;
createdAt: Date;
updatedAt: Date;
}): GatewayReleaseOperationRecord => ({
@@ -154,6 +158,8 @@ const mapOperation = (row: {
leaseUntil: toIso(row.leaseUntil),
heartbeatAt: toIso(row.heartbeatAt),
attempts: row.attempts,
bulkReleaseId: row.bulkReleaseId ?? undefined,
bulkOrder: row.bulkOrder ?? undefined,
createdAt: row.createdAt.toISOString(),
updatedAt: row.updatedAt.toISOString(),
});
@@ -263,12 +269,40 @@ export const createGatewayReleaseRepository = (prisma: GatewayPrismaClient): Gat
if (running && !runningIsStale) {
return null;
}
const candidate =
running ??
(await tx.gatewayReleaseOperation.findFirst({
let candidate = running;
if (!candidate) {
const queuedCandidates = await tx.gatewayReleaseOperation.findMany({
where: { status: 'QUEUED' },
orderBy: { createdAt: 'asc' },
}));
orderBy: [{ createdAt: 'asc' }, { bulkOrder: 'asc' }],
take: 200,
});
for (const queuedCandidate of queuedCandidates) {
if (!queuedCandidate.bulkReleaseId || queuedCandidate.bulkOrder === null) {
candidate = queuedCandidate;
break;
}
const [blockingGatewayTargets, blockingProfileTargets] = await Promise.all([
tx.gatewayReleaseOperation.count({
where: {
bulkReleaseId: queuedCandidate.bulkReleaseId,
bulkOrder: { lt: queuedCandidate.bulkOrder },
status: { not: 'SUCCEEDED' },
},
}),
tx.gatewayOperation.count({
where: {
bulkReleaseId: queuedCandidate.bulkReleaseId,
bulkOrder: { lt: queuedCandidate.bulkOrder },
status: { not: 'SUCCEEDED' },
},
}),
]);
if (blockingGatewayTargets + blockingProfileTargets === 0) {
candidate = queuedCandidate;
break;
}
}
}
if (!candidate) {
return null;
}
@@ -421,6 +455,37 @@ export const createGatewayReleaseRepository = (prisma: GatewayPrismaClient): Gat
if (!previous || (previous.status !== 'FAILED' && previous.status !== 'CANCELLED')) {
return null;
}
if (previous.bulkReleaseId) {
const batch = await tx.gatewayBulkRelease.findUniqueOrThrow({
where: { id: previous.bulkReleaseId },
select: { resolvedCommitSha: true },
});
const operation = await tx.gatewayReleaseOperation.update({
where: { id: previous.id },
data: {
status: 'QUEUED',
sourceMode: 'COMMIT',
sourceRef: batch.resolvedCommitSha,
resolvedCommitSha: null,
requestedBy,
startedAt: null,
completedAt: null,
error: null,
leaseOwner: null,
leaseUntil: null,
heartbeatAt: null,
},
});
await tx.gatewayReleaseLog.create({
data: {
operationId: operation.id,
level: 'INFO',
phase: 'queue',
message: '일괄 업데이트의 고정 커밋으로 재시도가 등록되었습니다.',
},
});
return operation;
}
return tx.gatewayReleaseOperation.create({
data: {
type: previous.type,
app/gateway-api/src/orchestrator/profileRepository.ts
+70 -5
View File
@@ -40,6 +40,8 @@ export interface GatewayOperationRecord {
leaseUntil?: string;
heartbeatAt?: string;
attempts?: number;
bulkReleaseId?: string;
bulkOrder?: number;
createdAt: string;
updatedAt: string;
}
@@ -270,6 +272,8 @@ type GatewayOperationRow = {
leaseUntil: Date | null;
heartbeatAt: Date | null;
attempts: number;
bulkReleaseId: string | null;
bulkOrder: number | null;
createdAt: Date;
updatedAt: Date;
};
@@ -337,6 +341,8 @@ const mapOperation = (row: GatewayOperationRow): GatewayOperationRecord => ({
leaseUntil: toIso(row.leaseUntil),
heartbeatAt: toIso(row.heartbeatAt),
attempts: row.attempts,
bulkReleaseId: row.bulkReleaseId ?? undefined,
bulkOrder: row.bulkOrder ?? undefined,
createdAt: row.createdAt.toISOString(),
updatedAt: row.updatedAt.toISOString(),
});
@@ -746,15 +752,43 @@ export const createGatewayProfileRepository = (prisma: GatewayPrismaClient): Gat
})),
});
}
const candidate =
running ??
(await tx.gatewayOperation.findFirst({
let candidate = running;
if (!candidate) {
const queuedCandidates = await tx.gatewayOperation.findMany({
where: {
status: 'QUEUED',
OR: [{ scheduledAt: null }, { scheduledAt: { lte: now } }],
},
orderBy: { createdAt: 'asc' },
}));
orderBy: [{ createdAt: 'asc' }, { bulkOrder: 'asc' }],
take: 200,
});
for (const queuedCandidate of queuedCandidates) {
if (!queuedCandidate.bulkReleaseId || queuedCandidate.bulkOrder === null) {
candidate = queuedCandidate;
break;
}
const [blockingGatewayTargets, blockingProfileTargets] = await Promise.all([
tx.gatewayReleaseOperation.count({
where: {
bulkReleaseId: queuedCandidate.bulkReleaseId,
bulkOrder: { lt: queuedCandidate.bulkOrder },
status: { not: 'SUCCEEDED' },
},
}),
tx.gatewayOperation.count({
where: {
bulkReleaseId: queuedCandidate.bulkReleaseId,
bulkOrder: { lt: queuedCandidate.bulkOrder },
status: { not: 'SUCCEEDED' },
},
}),
]);
if (blockingGatewayTargets + blockingProfileTargets === 0) {
candidate = queuedCandidate;
break;
}
}
}
if (!candidate) {
return null;
}
@@ -983,6 +1017,37 @@ export const createGatewayProfileRepository = (prisma: GatewayPrismaClient): Gat
) {
throw new GatewayProfileOperationConflictError();
}
if (previous.bulkReleaseId) {
const batch = await tx.gatewayBulkRelease.findUniqueOrThrow({
where: { id: previous.bulkReleaseId },
select: { resolvedCommitSha: true },
});
const operation = await tx.gatewayOperation.update({
where: { id: previous.id },
data: {
status: 'QUEUED',
sourceMode: 'COMMIT',
sourceRef: batch.resolvedCommitSha,
resolvedCommitSha: null,
requestedBy,
startedAt: null,
completedAt: null,
error: null,
leaseOwner: null,
leaseUntil: null,
heartbeatAt: null,
},
});
await tx.gatewayOperationLog.create({
data: {
operationId: operation.id,
level: 'INFO',
phase: 'queue',
message: '일괄 업데이트의 고정 커밋으로 재시도가 등록되었습니다.',
},
});
return operation;
}
const previousPayload = previous.payload as GatewayPrisma.JsonObject;
const retrySource = buildRetryOperationSource(previous);
const operation = await tx.gatewayOperation.create({