From a7251ff76b825ffe3e9f02b9e1fa0aaad8de0356 Mon Sep 17 00:00:00 2001 From: Jay Shen Date: Tue, 1 Sep 2026 10:54:27 +0800 Subject: [PATCH] fix: serialize merge requests through fifo --- src/server/app.ts | 85 +- src/server/automation/poller.ts | 2 +- src/server/automation/queue.ts | 214 +++- src/server/automation/service.ts | 194 ++- .../db/migrations/0006-closure-merge-queue.ts | 78 ++ src/server/db/migrations/index.ts | 2 + src/server/db/schema.ts | 5 + src/server/repository/errors.ts | 1 + src/server/repository/git-repository.ts | 374 ++++-- src/server/repository/service.ts | 88 +- src/server/runs/orchestrator.ts | 5 +- src/server/runs/types.ts | 2 + src/shared/types.ts | 5 + src/web/App.tsx | 52 +- tests/acceptance/phase9.ts | 31 +- tests/closure2-merge-queue.test.ts | 1106 +++++++++++++++++ tests/e2e/phase8-ui-smoke.ts | 7 +- tests/phase2.test.ts | 47 +- tests/phase3-api.test.ts | 79 +- tests/phase6.test.ts | 52 +- 20 files changed, 2167 insertions(+), 262 deletions(-) create mode 100644 src/server/db/migrations/0006-closure-merge-queue.ts create mode 100644 tests/closure2-merge-queue.test.ts diff --git a/src/server/app.ts b/src/server/app.ts index e8cfd3c..4b7c766 100644 --- a/src/server/app.ts +++ b/src/server/app.ts @@ -303,6 +303,7 @@ export async function createApp(options: AppOptions) { app.put('/api/config', async (request, reply) => { requireAuth(request, auth); const body = readBody(request); + assertRepositoryIdentityMutable(body.repository, configuration, automation); updateConfiguration(body, configuration, secretStore); return reply.send(makeConfigResponse(configuration, secretStore)); }); @@ -318,6 +319,7 @@ export async function createApp(options: AppOptions) { app.put('/api/config/repository', async (request, reply) => { requireAuth(request, auth); const body = readBody(request); + assertRepositoryIdentityMutable(body, configuration, automation); applySecretValues(body.secrets, secretStore); if (body.repository !== undefined && typeof body.repository === 'string') { validateRepositoryUrl(body.repository); @@ -337,14 +339,13 @@ export async function createApp(options: AppOptions) { return reply.send(await indexer.sync()); }); - app.post('/api/repository/scenario-branch', async (request, reply) => { + app.post('/api/repository/scenario-branch', async (request) => { requireAuth(request, auth); - const body = readBody(request); - const initialRef = body.initialRef; - if (initialRef !== undefined && typeof initialRef !== 'string') { - throw new AppError('INVALID_REQUEST', 'initialRef 必须是字符串', 400); - } - return reply.send(await repository.ensureScenarioBranch(initialRef)); + throw new AppError( + 'SCENARIO_BRANCH_QUEUE_REQUIRED', + '场景测试分支不能在队列外创建；请提交 confirmed initialization merge-source 请求', + 409, + ); }); app.post('/api/repository/merge', async (request, reply) => { @@ -353,10 +354,28 @@ export async function createApp(options: AppOptions) { if (typeof body.sourceRef !== 'string' || body.sourceRef.trim() === '') { throw new AppError('INVALID_REQUEST', 'sourceRef 必须是非空字符串', 400); } - if (typeof body.confirmed !== 'boolean') { + if (body.confirmed !== true) { throw new AppError('MERGE_CONFIRMATION_REQUIRED', '需要明确确认后才能合并', 400); } - return reply.send(await repository.mergeSourceRef(body.sourceRef, body.confirmed)); + if (body.initialization !== undefined && typeof body.initialization !== 'boolean') { + throw new AppError('RUN_REQUEST_INVALID', 'initialization 必须是布尔值', 400); + } + if (body.request !== undefined && typeof body.request !== 'string') { + throw new AppError('RUN_REQUEST_INVALID', 'request 必须是字符串', 400); + } + const sourceRef = body.sourceRef.trim(); + const submission = await automation.submitTestRequest({ + request: + typeof body.request === 'string' && body.request.trim() !== '' + ? body.request + : '合并已确认来源并测试固定场景分支 target', + trigger: 'manual', + requestKind: 'manual-merge-source', + sourceRef, + confirmed: true, + ...(body.initialization === true ? { initialization: true } : {}), + }); + return reply.status(202).send(formatAutomationSubmission(submission)); }); app.get('/api/repository/tree', async (request, reply) => { @@ -429,11 +448,12 @@ export async function createApp(options: AppOptions) { if (typeof body.request !== 'string' || body.request.trim() === '') { throw new AppError('RUN_REQUEST_REQUIRED', 'Run 请求内容不能为空', 400); } - if (body.targetCommit !== undefined && typeof body.targetCommit !== 'string') { - throw new AppError('RUN_TARGET_INVALID', 'targetCommit 必须是字符串', 400); - } - if (typeof body.targetCommit === 'string' && body.targetCommit.trim() === '') { - throw new AppError('RUN_TARGET_INVALID', 'targetCommit 不能为空', 400); + if (body.targetCommit !== undefined || body.targetRef !== undefined) { + throw new AppError( + 'RUN_TARGET_INVALID', + '普通 Run 不能指定任意 target；请使用 merge-source 或当前场景分支 HEAD', + 400, + ); } if (body.initialization !== undefined && typeof body.initialization !== 'boolean') { throw new AppError('RUN_REQUEST_INVALID', 'initialization 必须是布尔值', 400); @@ -445,7 +465,7 @@ export async function createApp(options: AppOptions) { const submission = await automation.submitTestRequest({ request: body.request, trigger, - targetCommit: body.targetCommit as string | undefined, + requestKind: 'manual-current-head', ...(body.initialization === true ? { initialization: true } : {}), }); return reply.status(202).send(formatAutomationSubmission(submission)); @@ -838,7 +858,11 @@ function formatAutomationSubmission(submission: AutomationSubmission): Record ['queued', 'running', 'waiting_archive'].includes(item.status)) + ) { + throw new AppError( + 'REPOSITORY_CHANGE_BLOCKED', + '存在未结束的测试请求，不能更换目标仓库或场景测试分支', + 409, + ); + } +} + function updateConfiguration( body: JsonRecord, configuration: ConfigurationStore, diff --git a/src/server/automation/poller.ts b/src/server/automation/poller.ts index 69de7ad..08cd584 100644 --- a/src/server/automation/poller.ts +++ b/src/server/automation/poller.ts @@ -172,7 +172,7 @@ class DefaultGitPoller implements GitPoller { const submission = await this.options.submitter.submitTestRequest({ trigger, request: `${trigger === 'git' ? 'Git Poll' : 'Cron'} 检测到场景测试分支有待测试提交：${includedCommits.join(', ')}`, - targetRef: currentHead, + requestKind: 'automatic-head', }); this.options.state.set(LAST_REPOSITORY_KEY, config.repository); this.options.state.set(LAST_BRANCH_KEY, scenarioBranch); diff --git a/src/server/automation/queue.ts b/src/server/automation/queue.ts index ffa08a1..73f1659 100644 --- a/src/server/automation/queue.ts +++ b/src/server/automation/queue.ts @@ -7,12 +7,17 @@ import type { RunTrigger } from '../../shared/types.js'; export type TestRequestStatus = 'queued' | 'running' | 'waiting_archive' | 'completed' | 'failed' | 'interrupted'; +export type TestRequestKind = 'automatic-head' | 'manual-current-head' | 'manual-merge-source'; +export type PreparedMergeMode = 'existing-branch' | 'initial-create'; + export interface TestRequestInput { request: string; trigger: RunTrigger; - /** A SHA, branch, tag, or another ref resolved by the Run Orchestrator. */ + requestKind?: TestRequestKind; + sourceRef?: string | null; + confirmed?: boolean; + /** Rejected legacy fields retained only to produce an explicit migration error. */ targetRef?: string | null; - /** Backwards-compatible spelling used by the Run API. */ targetCommit?: string | null; initialization?: boolean; } @@ -24,7 +29,13 @@ export interface TestRequestRecord { triggerSources: RunTrigger[]; requestIds: string[]; request: string; + /** Historical v0.1.0 input. New scheduling never reads this field. */ targetRef: string | null; + requestKind: TestRequestKind; + sourceRef: string | null; + preparedMergeCommit: string | null; + preparedMergeMode: PreparedMergeMode | null; + resolvedTargetCommit: string | null; status: TestRequestStatus; runId: string | null; claimedAt: string | null; @@ -48,6 +59,8 @@ export interface QueueCompletion { export interface TestRequestQueue { enqueue(input: TestRequestInput): TestRequestRecord; claimNext(): TestRequestRecord | null; + markPrepared(queueId: number, commit: string, mode: PreparedMergeMode): TestRequestRecord; + markResolved(queueId: number, commit: string): TestRequestRecord; markStarted(queueId: number, runId: string): TestRequestRecord; requeue(queueId: number): TestRequestRecord; markWaitingArchive(queueId: number, runId: string): TestRequestRecord; @@ -70,7 +83,13 @@ export class TestRequestQueueError extends Error { } const MAX_REQUEST_LENGTH = 16_384; -const MAX_TARGET_REF_LENGTH = 255; +const MAX_SOURCE_REF_LENGTH = 255; +const SOURCE_REF_PATTERN = /^[^\s~^:?*\\[\]]{1,255}$/; +const SHA_PATTERN = /^[0-9a-f]{40}$/; +const CREDENTIAL_LIKE_SOURCE = + /(?:github_pat_|gh[opsur]_|sk-[A-Za-z0-9]{12,}|AKIA[0-9A-Z]{16}|x-access-token@)/i; +const URL_LIKE_SOURCE = + /(?:^|@)(?:www\.)?(?:github\.com|gitlab\.com|bitbucket\.org)(?:\/|$)|^[^/@\s]+@[^/\s]+\.[A-Za-z]{2,}(?:\/|$)/i; const AUTOMATIC_TRIGGERS: readonly RunTrigger[] = ['git', 'schedule']; export function createTestRequestQueue( @@ -110,6 +129,8 @@ class SqliteTestRequestQueue implements TestRequestQueue { if ( tail && + tail.request_kind === 'automatic-head' && + normalized.requestKind === 'automatic-head' && isAutomatic(tail.trigger) && isAutomatic(normalized.trigger) && tail.initialization === (normalized.initialization ? 1 : 0) @@ -126,14 +147,13 @@ class SqliteTestRequestQueue implements TestRequestQueue { this.database .prepare( `UPDATE test_request_queue - SET trigger = ?, request = ?, target_ref = ?, + SET trigger = ?, request = ?, trigger_sources_json = ?, request_ids_json = ?, updated_at = ? WHERE queue_id = ?`, ) .run( mergeTrigger(tail.trigger as RunTrigger, normalized.trigger), mergedRequest, - normalized.targetRef, JSON.stringify(sources), JSON.stringify(requestIds), timestamp, @@ -145,16 +165,20 @@ class SqliteTestRequestQueue implements TestRequestQueue { const result = this.database .prepare( `INSERT INTO test_request_queue - (request_id, trigger, request, target_ref, trigger_sources_json, request_ids_json, - status, run_id, claimed_at, waiting_archive_at, completed_at, error_message, - archive_status, progressed, created_at, updated_at, initialization) - VALUES (?, ?, ?, ?, ?, ?, 'queued', NULL, NULL, NULL, NULL, NULL, NULL, NULL, ?, ?, ?)`, + (request_id, trigger, request, target_ref, request_kind, source_ref, + prepared_merge_commit, prepared_merge_mode, resolved_target_commit, + trigger_sources_json, request_ids_json, status, run_id, claimed_at, + waiting_archive_at, completed_at, error_message, archive_status, progressed, + created_at, updated_at, initialization) + VALUES (?, ?, ?, NULL, ?, ?, NULL, NULL, NULL, ?, ?, 'queued', NULL, NULL, NULL, NULL, + NULL, NULL, NULL, ?, ?, ?)`, ) .run( requestId, normalized.trigger, normalized.request, - normalized.targetRef, + normalized.requestKind, + normalized.sourceRef, JSON.stringify([normalized.trigger]), JSON.stringify([requestId]), timestamp, @@ -194,14 +218,75 @@ class SqliteTestRequestQueue implements TestRequestQueue { return queueId === null ? null : this.require(queueId); } + markPrepared(queueId: number, commit: string, mode: PreparedMergeMode): TestRequestRecord { + const normalized = normalizeCommit(commit, 'prepared commit'); + if (!['existing-branch', 'initial-create'].includes(mode)) { + throw new TestRequestQueueError('QUEUE_REQUEST_INVALID', 'prepared merge mode 无效'); + } + const timestamp = this.now(); + const result = this.database + .prepare( + `UPDATE test_request_queue + SET prepared_merge_commit = ?, prepared_merge_mode = ?, updated_at = ? + WHERE queue_id = ? AND status = 'running' + AND request_kind = 'manual-merge-source' + AND prepared_merge_commit IS NULL AND prepared_merge_mode IS NULL`, + ) + .run(normalized, mode, timestamp, queueId); + if (result.changes !== 1) { + throw new TestRequestQueueError( + 'QUEUE_STATE_INVALID', + '队列请求无法记录 prepared merge commit', + ); + } + return this.require(queueId); + } + + markResolved(queueId: number, commit: string): TestRequestRecord { + const normalized = normalizeCommit(commit, 'resolved target commit'); + const current = this.require(queueId); + if ( + current.requestKind === 'manual-merge-source' && + current.preparedMergeCommit !== normalized + ) { + throw new TestRequestQueueError( + 'QUEUE_STATE_INVALID', + 'resolved target 必须等于 prepared merge commit', + ); + } + if (current.resolvedTargetCommit !== null) { + if (current.resolvedTargetCommit === normalized) return current; + throw new TestRequestQueueError('QUEUE_STATE_INVALID', '队列请求的 target 已固定，不能改变'); + } + const timestamp = this.now(); + const result = this.database + .prepare( + `UPDATE test_request_queue + SET resolved_target_commit = ?, updated_at = ? + WHERE queue_id = ? + AND (status = 'running' OR run_id IS NOT NULL) + AND resolved_target_commit IS NULL`, + ) + .run(normalized, timestamp, queueId); + if (result.changes !== 1) { + throw new TestRequestQueueError('QUEUE_STATE_INVALID', '队列请求无法固定 resolved target'); + } + return this.require(queueId); + } + markStarted(queueId: number, runId: string): TestRequestRecord { this.assertRunId(runId); + const current = this.require(queueId); + if (current.runId !== null) { + if (current.runId === runId && current.status === 'running') return current; + throw new TestRequestQueueError('QUEUE_STATE_INVALID', '队列请求已经关联其他 Run'); + } const timestamp = this.now(); const result = this.database .prepare( `UPDATE test_request_queue SET run_id = ?, updated_at = ? - WHERE queue_id = ? AND status = 'running'`, + WHERE queue_id = ? AND status = 'running' AND run_id IS NULL`, ) .run(runId, timestamp, queueId); if (result.changes !== 1) { @@ -234,10 +319,11 @@ class SqliteTestRequestQueue implements TestRequestQueue { const result = this.database .prepare( `UPDATE test_request_queue - SET status = 'waiting_archive', run_id = ?, waiting_archive_at = ?, updated_at = ? - WHERE queue_id = ? AND status = 'running'`, + SET status = 'waiting_archive', run_id = COALESCE(run_id, ?), + waiting_archive_at = ?, updated_at = ? + WHERE queue_id = ? AND status = 'running' AND (run_id IS NULL OR run_id = ?)`, ) - .run(runId, timestamp, timestamp, queueId); + .run(runId, timestamp, timestamp, queueId, runId); if (result.changes !== 1) { throw new TestRequestQueueError('QUEUE_STATE_INVALID', '队列请求不在 running 状态'); } @@ -249,9 +335,10 @@ class SqliteTestRequestQueue implements TestRequestQueue { const result = this.database .prepare( `UPDATE test_request_queue - SET status = 'completed', run_id = COALESCE(?, run_id), completed_at = ?, + SET status = 'completed', run_id = COALESCE(run_id, ?), completed_at = ?, error_message = ?, archive_status = ?, progressed = ?, updated_at = ? - WHERE queue_id = ? AND status IN ('running', 'waiting_archive')`, + WHERE queue_id = ? AND status IN ('running', 'waiting_archive') + AND (? IS NULL OR run_id IS NULL OR run_id = ?)`, ) .run( completion.runId ?? null, @@ -265,6 +352,8 @@ class SqliteTestRequestQueue implements TestRequestQueue { : 0, timestamp, queueId, + completion.runId ?? null, + completion.runId ?? null, ); if (result.changes !== 1) { throw new TestRequestQueueError( @@ -353,6 +442,11 @@ interface QueueRow { trigger: string; request: string; target_ref: string | null; + request_kind: TestRequestKind; + source_ref: string | null; + prepared_merge_commit: string | null; + prepared_merge_mode: PreparedMergeMode | null; + resolved_target_commit: string | null; trigger_sources_json: string; request_ids_json: string; status: TestRequestStatus; @@ -371,7 +465,8 @@ interface QueueRow { function normalizeInput(input: TestRequestInput): { request: string; trigger: RunTrigger; - targetRef: string | null; + requestKind: TestRequestKind; + sourceRef: string | null; initialization: boolean; } { if (!input || typeof input.request !== 'string' || input.request.trim() === '') { @@ -386,38 +481,76 @@ function normalizeInput(input: TestRequestInput): { if (input.initialization !== undefined && typeof input.initialization !== 'boolean') { throw new TestRequestQueueError('QUEUE_REQUEST_INVALID', 'initialization 必须是布尔值'); } - if (input.targetRef !== undefined && input.targetCommit !== undefined) { - if ( - input.targetRef !== null && - input.targetCommit !== null && - input.targetRef.trim() !== input.targetCommit.trim() - ) { - throw new TestRequestQueueError('QUEUE_REQUEST_INVALID', 'targetRef 与 targetCommit 不一致'); + for (const legacyTarget of [input.targetRef, input.targetCommit]) { + if (legacyTarget !== undefined && legacyTarget !== null && legacyTarget.trim() !== '') { + throw new TestRequestQueueError( + 'QUEUE_REQUEST_INVALID', + '普通测试请求不能指定任意 target；请使用 merge-source 入口', + ); } } - const candidate = input.targetRef ?? input.targetCommit; - let targetRef: string | null = null; - if (candidate !== undefined && candidate !== null) { - targetRef = candidate.trim(); - if ( - targetRef === '' || - targetRef.length > MAX_TARGET_REF_LENGTH || - [...targetRef].some((character) => { - const code = character.charCodeAt(0); - return code < 32 || code === 127; - }) - ) { - throw new TestRequestQueueError('QUEUE_REQUEST_INVALID', 'target ref 格式无效'); + const requestKind = + input.requestKind ?? (isAutomatic(input.trigger) ? 'automatic-head' : 'manual-current-head'); + if (!['automatic-head', 'manual-current-head', 'manual-merge-source'].includes(requestKind)) { + throw new TestRequestQueueError('QUEUE_REQUEST_INVALID', '测试请求种类无效'); + } + if (requestKind === 'automatic-head' && !isAutomatic(input.trigger)) { + throw new TestRequestQueueError('QUEUE_REQUEST_INVALID', '人工请求不能伪装为自动 HEAD 请求'); + } + if (requestKind !== 'automatic-head' && isAutomatic(input.trigger)) { + throw new TestRequestQueueError('QUEUE_REQUEST_INVALID', '自动触发只能提交 automatic-head'); + } + let sourceRef: string | null = null; + if (requestKind === 'manual-merge-source') { + if (input.confirmed !== true) { + throw new TestRequestQueueError('QUEUE_REQUEST_INVALID', 'merge-source 请求需要明确确认'); } + sourceRef = normalizeSourceRef(input.sourceRef); + } else if (input.sourceRef !== undefined && input.sourceRef !== null) { + throw new TestRequestQueueError( + 'QUEUE_REQUEST_INVALID', + '只有 merge-source 请求可以指定 sourceRef', + ); } return { request: input.request.trim(), trigger: input.trigger, - targetRef, + requestKind, + sourceRef, initialization: input.initialization === true, }; } +function normalizeSourceRef(value: string | null | undefined): string { + if (typeof value !== 'string') { + throw new TestRequestQueueError('QUEUE_REQUEST_INVALID', 'sourceRef 必须是非空字符串'); + } + const normalized = value.trim(); + if ( + normalized === '' || + normalized.length > MAX_SOURCE_REF_LENGTH || + !SOURCE_REF_PATTERN.test(normalized) || + normalized.includes('..') || + normalized.includes('@{') || + normalized.startsWith('refs/luowang/') || + normalized.startsWith('refs/remotes/') || + CREDENTIAL_LIKE_SOURCE.test(normalized) || + URL_LIKE_SOURCE.test(normalized) || + hasControlCharacters(normalized) + ) { + throw new TestRequestQueueError('QUEUE_REQUEST_INVALID', 'sourceRef 格式无效'); + } + return normalized; +} + +function normalizeCommit(value: string, label: string): string { + const normalized = value.trim().toLowerCase(); + if (!SHA_PATTERN.test(normalized)) { + throw new TestRequestQueueError('QUEUE_REQUEST_INVALID', `${label} 格式无效`); + } + return normalized; +} + function normalizeRequestId(value: string): string { const normalized = value.trim(); if (normalized === '' || normalized.length > 200 || hasControlCharacters(normalized)) { @@ -465,6 +598,11 @@ function toRecord(row: QueueRow): TestRequestRecord { requestIds: parseJson(row.request_ids_json, [row.request_id]), request: row.request, targetRef: row.target_ref, + requestKind: row.request_kind, + sourceRef: row.source_ref, + preparedMergeCommit: row.prepared_merge_commit, + preparedMergeMode: row.prepared_merge_mode, + resolvedTargetCommit: row.resolved_target_commit, status: row.status, runId: row.run_id, claimedAt: row.claimed_at, diff --git a/src/server/automation/service.ts b/src/server/automation/service.ts index 5202f83..dbb3f29 100644 --- a/src/server/automation/service.ts +++ b/src/server/automation/service.ts @@ -1,3 +1,5 @@ +import { createHash } from 'node:crypto'; + import type Database from 'better-sqlite3'; import type { Logger } from 'pino'; @@ -8,7 +10,7 @@ import type { RepositoryService } from '../repository/service.js'; import type { RunArchiver, ArchiveResult } from '../runs/archiver.js'; import type { RunOrchestrator } from '../runs/orchestrator.js'; import type { RunStore } from '../runs/store.js'; -import { RunWorkspaceStore } from '../runs/workspace.js'; +import { createRunId, RunWorkspaceStore } from '../runs/workspace.js'; import { createRunRecoveryStore, type RunRecoveryStore } from './recovery.js'; import { createTestRequestQueue, @@ -129,6 +131,7 @@ class DefaultAutomationService implements AutomationService { this.recovering = true; try { await this.options.runs.recover(); + await this.reconcileMergeRequestRefs(); for (const item of this.options.queue.listInFlight()) { await this.recoverQueueItem(item); } @@ -148,7 +151,7 @@ class DefaultAutomationService implements AutomationService { for (const item of this.options.queue.listInFlight()) { if (item.status !== 'waiting_archive' || !item.runId) continue; const result = byRunId.get(item.runId); - if (result) this.finishArchive(item.queueId, item.runId, result); + if (result) await this.finishArchive(item.queueId, item.runId, result); else await this.archiveQueueItem(item.queueId, item.runId); } await this.dispatchNext(); @@ -188,7 +191,7 @@ class DefaultAutomationService implements AutomationService { return this.submitTestRequest({ request: rerunRequest, trigger: 'manual', - targetRef: detail.targetCommit, + requestKind: 'manual-current-head', }); } @@ -226,10 +229,12 @@ class DefaultAutomationService implements AutomationService { this.activeQueueId = item.queueId; try { + const targetCommit = await this.resolveTarget(item); const runInput = { request: item.request, trigger: item.trigger, - targetCommit: item.targetRef ?? undefined, + runId: queueRunId(item), + targetCommit, ...(item.initialization ? { initialization: true as const } : {}), }; const run = await this.options.runs.start(runInput); @@ -238,7 +243,7 @@ class DefaultAutomationService implements AutomationService { void this.monitorRun(item.queueId, run.runId); return { queueId: item.queueId, run }; } catch (error) { - this.options.queue.fail(item.queueId, safeMessage(error)); + await this.failQueueItem(item.queueId, safeMessage(error)); this.activeQueueId = null; this.activeRunId = null; Promise.resolve() @@ -255,24 +260,24 @@ class DefaultAutomationService implements AutomationService { try { const detail = await this.options.runs.wait(runId); if (!detail) { - this.options.queue.fail(queueId, 'Run 完成状态无法读取'); + await this.failQueueItem(queueId, 'Run 完成状态无法读取'); return; } if (detail.status === 'completed') { this.options.queue.markWaitingArchive(queueId, runId); await this.archiveQueueItem(queueId, runId); } else if (detail.status === 'interrupted') { - this.options.queue.fail( + await this.failQueueItem( queueId, detail.errorMessage ?? 'Run 因进程重启而中断', 'interrupted', ); } else { - this.options.queue.fail(queueId, detail.errorMessage ?? 'Run 执行失败'); + await this.failQueueItem(queueId, detail.errorMessage ?? 'Run 执行失败'); } } catch (error) { try { - this.options.queue.fail(queueId, safeMessage(error)); + await this.failQueueItem(queueId, safeMessage(error)); } catch { // The queue may have been reconciled by the recovery or archive task. } @@ -291,46 +296,166 @@ class DefaultAutomationService implements AutomationService { private async recoverQueueItem(item: TestRequestRecord): Promise { if (item.status === 'waiting_archive') { - if (item.runId) await this.archiveQueueItem(item.queueId, item.runId); - else this.options.queue.fail(item.queueId, '等待归档的队列请求缺少 Run ID'); + if (item.runId) { + const detail = await this.options.runs.get(item.runId); + if (!item.resolvedTargetCommit && detail?.targetCommit) { + this.options.queue.markResolved(item.queueId, detail.targetCommit); + } + await this.archiveQueueItem(item.queueId, item.runId); + } else { + await this.failQueueItem(item.queueId, '等待归档的队列请求缺少 Run ID'); + } return; } - if (!item.runId) { - this.options.queue.requeue(item.queueId); - return; + let runId = item.runId; + if (!runId) { + const reservedRunId = queueRunId(item); + const reservedRun = await this.options.runs.get(reservedRunId); + if (!reservedRun) { + this.options.queue.requeue(item.queueId); + return; + } + this.options.queue.markStarted(item.queueId, reservedRunId); + runId = reservedRunId; + item = this.options.queue.get(item.queueId) ?? item; } - const detail = await this.options.runs.get(item.runId); + const detail = await this.options.runs.get(runId); + if (!item.resolvedTargetCommit && detail?.targetCommit) { + this.options.queue.markResolved(item.queueId, detail.targetCommit); + } if (detail?.status === 'completed') { - this.options.queue.markWaitingArchive(item.queueId, item.runId); - await this.archiveQueueItem(item.queueId, item.runId); + this.options.queue.markWaitingArchive(item.queueId, runId); + await this.archiveQueueItem(item.queueId, runId); } else if (detail?.status === 'interrupted' || !detail) { if (detail?.status === 'interrupted') { this.options.recoveryStore.record( - { - ...detail, - trigger: item.trigger, - request: item.request, - targetCommit: detail.targetCommit ?? item.targetRef, - }, + { ...detail, trigger: item.trigger, request: item.request }, { interruptedAt: detail.finishedAt ?? undefined }, ); } - this.options.queue.fail( + await this.failQueueItem( item.queueId, '进程重启时 Run 尚在 running 目录，未恢复 Agent 会话', 'interrupted', ); } else if (detail.status === 'failed') { - this.options.queue.fail(item.queueId, detail.errorMessage ?? 'Run 执行失败'); + await this.failQueueItem(item.queueId, detail.errorMessage ?? 'Run 执行失败'); + } + } + + private async resolveTarget(item: TestRequestRecord): Promise { + if (item.resolvedTargetCommit) { + if (!(await this.options.repository.isPublishedTarget(item.resolvedTargetCommit))) { + throw new AutomationServiceError( + 'AUTOMATION_REQUEST_INVALID', + '已固定的 target 不在远端场景测试分支历史中', + ); + } + return item.resolvedTargetCommit; + } + + if (item.requestKind === 'manual-merge-source') { + let prepared = item.preparedMergeCommit; + if (!prepared) { + if (!item.sourceRef) { + throw new AutomationServiceError( + 'AUTOMATION_REQUEST_INVALID', + 'merge-source 请求缺少 sourceRef', + ); + } + if (await this.options.repository.readMergeRequestRef(item.queueId)) { + throw new AutomationServiceError( + 'AUTOMATION_REQUEST_INVALID', + 'internal ref 已存在但 prepared commit 尚未持久化，拒绝猜测或重做 merge', + ); + } + const result = await this.options.repository.prepareMergeRequest( + item.sourceRef, + item.queueId, + item.initialization, + ); + prepared = result.preparedCommit; + this.options.queue.markPrepared(item.queueId, prepared, result.mode); + } + const refreshed = this.options.queue.get(item.queueId); + const published = await this.options.repository.publishPreparedMerge( + item.queueId, + prepared, + refreshed?.preparedMergeMode ?? item.preparedMergeMode, + ); + return this.options.queue.markResolved(item.queueId, published).resolvedTargetCommit!; + } + + const repository = await this.options.repository.getRepository(); + await repository.fetch(); + const head = await repository.remoteBranchHead(this.options.repository.getScenarioBranch()); + if (!head) { + throw new AutomationServiceError( + 'AUTOMATION_REQUEST_INVALID', + item.requestKind === 'automatic-head' + ? '场景测试分支尚未创建，自动请求没有可测试批次' + : '场景测试分支尚未创建；请通过 initialization merge-source 请求首次创建', + ); + } + return this.options.queue.markResolved(item.queueId, head).resolvedTargetCommit!; + } + + private async reconcileMergeRequestRefs(): Promise { + if ( + typeof this.options.repository.getRepositoryUrl !== 'function' || + typeof this.options.repository.listMergeRequestRefs !== 'function' || + this.options.repository.getRepositoryUrl().trim() === '' + ) + return; + const records = new Map(this.options.queue.list().map((item) => [item.queueId, item])); + for (const queueId of await this.options.repository.listMergeRequestRefs()) { + const item = records.get(queueId); + if (!item || !['queued', 'running', 'waiting_archive'].includes(item.status)) { + await this.cleanupMergeRequestRef(queueId); + continue; + } + if (!item.preparedMergeCommit || !item.preparedMergeMode) { + await this.failQueueItem( + queueId, + 'internal ref 已存在但 prepared commit 尚未持久化，拒绝猜测或重做 merge', + ); + } + } + } + + private async failQueueItem( + queueId: number, + message: string, + status: 'failed' | 'interrupted' = 'failed', + ): Promise { + this.options.queue.fail(queueId, message, status); + await this.cleanupTerminalRef(queueId); + } + + private async cleanupTerminalRef(queueId: number): Promise { + const item = this.options.queue.get(queueId); + if (!item || item.requestKind !== 'manual-merge-source') return; + if (!['completed', 'failed', 'interrupted'].includes(item.status)) return; + await this.cleanupMergeRequestRef(queueId); + } + + private async cleanupMergeRequestRef(queueId: number): Promise { + try { + await this.options.repository.cleanupMergeRequestRef(queueId); + } catch (error) { + this.options.logger?.warn( + { queueId, errorName: error instanceof Error ? error.name : 'UnknownError' }, + 'terminal merge request internal ref cleanup failed', + ); } } private async archiveQueueItem(queueId: number, runId: string): Promise { try { const result = await this.options.archiver.archive(runId); - this.finishArchive(queueId, runId, result); + await this.finishArchive(queueId, runId, result); } catch (error) { this.options.queue.complete(queueId, { runId, @@ -338,17 +463,32 @@ class DefaultAutomationService implements AutomationService { progressed: false, errorMessage: safeMessage(error), }); + await this.cleanupTerminalRef(queueId); } } - private finishArchive(queueId: number, runId: string, result: ArchiveResult): void { + private async finishArchive( + queueId: number, + runId: string, + result: ArchiveResult, + ): Promise { this.options.queue.complete(queueId, { runId, archiveStatus: result.status, progressed: result.progressed, errorMessage: result.errorMessage, }); + await this.cleanupTerminalRef(queueId); + } +} + +function queueRunId(item: Pick): string { + const timestamp = Date.parse(item.createdAt); + if (!Number.isFinite(timestamp)) { + throw new AutomationServiceError('AUTOMATION_REQUEST_INVALID', '队列请求创建时间无效'); } + const entropy = createHash('sha256').update(item.requestId).digest().subarray(0, 10); + return createRunId(timestamp, entropy); } function safeMessage(error: unknown): string { diff --git a/src/server/db/migrations/0006-closure-merge-queue.ts b/src/server/db/migrations/0006-closure-merge-queue.ts new file mode 100644 index 0000000..124e0ac --- /dev/null +++ b/src/server/db/migrations/0006-closure-merge-queue.ts @@ -0,0 +1,78 @@ +import type Database from 'better-sqlite3'; + +import type { Migration } from './0000-foundation.js'; + +const LEGACY_TARGET_ERROR = + '旧请求包含任意 target，未自动升级为 merge 授权；请通过 merge-source 入口重新提交'; +const MISSING_ARCHIVE_RUN_ERROR = '等待归档的旧队列请求缺少 Run ID，未重新调度'; + +export const closureMergeQueueMigration: Migration = { + version: '0006_closure_merge_queue', + apply(database: Database.Database) { + addColumn(database, 'test_request_queue', 'request_kind', 'TEXT'); + addColumn(database, 'test_request_queue', 'source_ref', 'TEXT'); + addColumn(database, 'test_request_queue', 'prepared_merge_commit', 'TEXT'); + addColumn(database, 'test_request_queue', 'prepared_merge_mode', 'TEXT'); + addColumn(database, 'test_request_queue', 'resolved_target_commit', 'TEXT'); + + database + .prepare( + `UPDATE test_request_queue + SET request_kind = CASE + WHEN trigger IN ('git', 'schedule') THEN 'automatic-head' + ELSE 'manual-current-head' + END + WHERE request_kind IS NULL`, + ) + .run(); + + database + .prepare( + `UPDATE test_request_queue + SET resolved_target_commit = COALESCE( + (SELECT target_commit FROM run_store_runs WHERE run_store_runs.run_id = test_request_queue.run_id), + (SELECT target_commit FROM interrupted_run_records WHERE interrupted_run_records.run_id = test_request_queue.run_id) + ) + WHERE run_id IS NOT NULL AND resolved_target_commit IS NULL`, + ) + .run(); + + database + .prepare( + `UPDATE test_request_queue + SET status = 'failed', completed_at = COALESCE(completed_at, updated_at), error_message = ? + WHERE status = 'waiting_archive' AND run_id IS NULL`, + ) + .run(MISSING_ARCHIVE_RUN_ERROR); + + database + .prepare( + `UPDATE test_request_queue + SET status = 'failed', completed_at = COALESCE(completed_at, updated_at), error_message = ? + WHERE status IN ('queued', 'running') + AND run_id IS NULL + AND trigger IN ('manual', 'api') + AND target_ref IS NOT NULL`, + ) + .run(LEGACY_TARGET_ERROR); + + database.exec(` + UPDATE test_request_queue + SET status = 'queued', claimed_at = NULL + WHERE status = 'running' + AND run_id IS NULL + AND NOT (trigger IN ('manual', 'api') AND target_ref IS NOT NULL); + `); + }, +}; + +function addColumn( + database: Database.Database, + table: string, + column: string, + definition: string, +): void { + const columns = database.prepare(`PRAGMA table_info(${table})`).all() as Array<{ name: string }>; + if (columns.some((item) => item.name === column)) return; + database.exec(`ALTER TABLE ${table} ADD COLUMN ${column} ${definition}`); +} diff --git a/src/server/db/migrations/index.ts b/src/server/db/migrations/index.ts index 4acd1fd..e9731eb 100644 --- a/src/server/db/migrations/index.ts +++ b/src/server/db/migrations/index.ts @@ -4,6 +4,7 @@ import { repositoryIndexMigration } from './0002-repository-index.js'; import { runArchiveMigration } from './0003-run-archive.js'; import { automationRecoveryMigration } from './0004-automation-recovery.js'; import { scenarioLifecycleMigration } from './0005-scenario-lifecycle.js'; +import { closureMergeQueueMigration } from './0006-closure-merge-queue.js'; export type { Migration } from './0000-foundation.js'; @@ -14,4 +15,5 @@ export const migrations = [ runArchiveMigration, automationRecoveryMigration, scenarioLifecycleMigration, + closureMergeQueueMigration, ]; diff --git a/src/server/db/schema.ts b/src/server/db/schema.ts index 44341fb..4bcf7b3 100644 --- a/src/server/db/schema.ts +++ b/src/server/db/schema.ts @@ -161,6 +161,11 @@ export const testRequestQueue = sqliteTable('test_request_queue', { trigger: text('trigger').notNull(), request: text('request').notNull(), targetRef: text('target_ref'), + requestKind: text('request_kind'), + sourceRef: text('source_ref'), + preparedMergeCommit: text('prepared_merge_commit'), + preparedMergeMode: text('prepared_merge_mode'), + resolvedTargetCommit: text('resolved_target_commit'), triggerSourcesJson: text('trigger_sources_json').notNull(), requestIdsJson: text('request_ids_json').notNull(), status: text('status').notNull(), diff --git a/src/server/repository/errors.ts b/src/server/repository/errors.ts index 72a65cd..1c52309 100644 --- a/src/server/repository/errors.ts +++ b/src/server/repository/errors.ts @@ -8,6 +8,7 @@ export type RepositoryErrorCode = | 'SCENARIO_BRANCH_HISTORY_BROKEN' | 'MERGE_CONFIRMATION_REQUIRED' | 'MERGE_CONFLICT' + | 'MERGE_REQUEST_STATE_INVALID' | 'PUSH_REJECTED' | 'REPORT_CONFLICT' | 'REPORT_PUBLISH_CONFLICT' diff --git a/src/server/repository/git-repository.ts b/src/server/repository/git-repository.ts index ad1ba0e..85a0a60 100644 --- a/src/server/repository/git-repository.ts +++ b/src/server/repository/git-repository.ts @@ -45,11 +45,13 @@ export interface GitTreeEntry { sha: string; } -export interface GitMergeResult { - originalHead: string; +export type PreparedMergeMode = 'existing-branch' | 'initial-create'; + +export interface PreparedMergeResult { + mode: PreparedMergeMode; sourceCommit: string; - mergeCommit: string | null; - scenarioBranchHead: string; + originalHead: string | null; + preparedCommit: string; alreadyIncluded: boolean; } @@ -216,64 +218,50 @@ export class GitRepository { } } - async createScenarioBranch(branch: string, initialRef: string): Promise { - assertBranchName(branch); - await this.fetch(); - if (await this.remoteBranchHead(branch)) { - throw new RepositoryError( - 'SCENARIO_BRANCH_REMOTE_CHANGED', - '场景测试分支已被其他请求创建，请重新读取状态', - 409, - ); - } - const source = await this.resolveRemoteRef(initialRef); - if (!source) { - throw new RepositoryError('TARGET_INVALID', `无法解析初始 Git ref：${initialRef}`, 400); - } - try { - await this.checkoutTarget(source); - await this.run(['push', 'origin', `HEAD:refs/heads/${branch}`]); - const head = await this.remoteBranchHead(branch); - if (!head) - throw new RepositoryError('PUSH_REJECTED', '场景测试分支创建后无法读取远端 HEAD', 502); - return head; - } catch (error) { - await this.cleanWorkspace(); - throw error; - } finally { - await this.cleanWorkspace(); - } - } - - async mergeNoFastForward( + async prepareMergeRequest( branch: string, sourceRef: string, - confirmed: boolean, - ): Promise { + queueId: number, + initialization: boolean, + ): Promise { assertBranchName(branch); - if (!confirmed) { + assertQueueId(queueId); + await this.fetch(); + const internalRef = mergeRequestRef(queueId); + if (await this.readInternalRef(queueId)) { throw new RepositoryError( - 'MERGE_CONFIRMATION_REQUIRED', - '合并到场景测试分支前需要操作者确认', - 400, + 'MERGE_REQUEST_STATE_INVALID', + 'merge 请求 internal ref 已存在', + 409, ); } - await this.fetch(); const originalHead = await this.remoteBranchHead(branch); + const sourceCommit = await this.resolvePublishedSourceCommit(sourceRef); if (!originalHead) { - throw new RepositoryError('SCENARIO_BRANCH_NOT_FOUND', '场景测试分支尚未创建', 409); - } - const sourceCommit = await this.resolveRemoteRef(sourceRef); - if (!sourceCommit) { - throw new RepositoryError('TARGET_INVALID', `无法解析来源 Git ref：${sourceRef}`, 400); + if (!initialization) { + throw new RepositoryError( + 'SCENARIO_BRANCH_NOT_FOUND', + '场景测试分支尚未创建；首次创建必须提交 initialization merge-source 请求', + 409, + ); + } + await this.createInternalRef(internalRef, sourceCommit, 'initial-create'); + return { + mode: 'initial-create', + sourceCommit, + originalHead: null, + preparedCommit: sourceCommit, + alreadyIncluded: false, + }; } + if (await this.isAncestor(sourceCommit, originalHead)) { - await this.cleanWorkspace(); + await this.createInternalRef(internalRef, originalHead, 'existing-branch'); return { - originalHead, + mode: 'existing-branch', sourceCommit, - mergeCommit: null, - scenarioBranchHead: originalHead, + originalHead, + preparedCommit: originalHead, alreadyIncluded: true, }; } @@ -289,6 +277,8 @@ export class GitRepository { 'merge', '--no-ff', '--no-edit', + '-m', + `luowang merge request #${queueId}`, sourceCommit, ]); } catch (error) { @@ -302,43 +292,141 @@ export class GitRepository { } throw error; } - const latestRemoteHead = await this.remoteBranchHead(branch); - if (latestRemoteHead !== originalHead) { - await this.cleanWorkspace(); - throw new RepositoryError( - 'SCENARIO_BRANCH_REMOTE_CHANGED', - '合并期间远端场景测试分支发生变化，请重新尝试', - 409, - ); - } - const mergeCommit = (await this.run(['rev-parse', 'HEAD'])).stdout.trim(); - await this.run(['push', 'origin', `HEAD:refs/heads/${branch}`]); - const scenarioBranchHead = await this.remoteBranchHead(branch); - if (!scenarioBranchHead) { - throw new RepositoryError('PUSH_REJECTED', '推送成功后无法读取场景测试分支 HEAD', 502); - } + const preparedCommit = normalizeSha((await this.run(['rev-parse', 'HEAD'])).stdout); + await this.createInternalRef(internalRef, preparedCommit, 'existing-branch'); return { - originalHead, + mode: 'existing-branch', sourceCommit, - mergeCommit, - scenarioBranchHead, + originalHead, + preparedCommit, alreadyIncluded: false, }; - } catch (error) { - await this.cleanWorkspace(); - if (error instanceof GitCommandError) { - throw new RepositoryError( - 'PUSH_REJECTED', - '场景测试分支推送被拒绝，未执行 force push', - 409, - ); - } - throw error; } finally { await this.cleanWorkspace(); } } + async publishPreparedMerge( + branch: string, + queueId: number, + preparedCommit: string, + mode: PreparedMergeMode | null, + ): Promise { + assertBranchName(branch); + assertQueueId(queueId); + const prepared = normalizeSha(preparedCommit); + await this.fetch(); + const remoteHead = await this.remoteBranchHead(branch); + if (remoteHead && (await this.isAncestor(prepared, remoteHead))) return prepared; + + const internal = await this.readInternalRef(queueId); + if (internal !== prepared) { + throw new RepositoryError( + 'MERGE_REQUEST_STATE_INVALID', + 'prepared merge commit 与 internal ref 不一致', + 409, + ); + } + if (mode !== 'initial-create' && mode !== 'existing-branch') { + throw new RepositoryError( + 'MERGE_REQUEST_STATE_INVALID', + 'prepared merge commit 缺少持久化准备模式', + 409, + ); + } + if (mode === 'initial-create' && remoteHead) { + throw new RepositoryError( + 'SCENARIO_BRANCH_REMOTE_CHANGED', + '首次创建场景测试分支时发生远端竞争，未发布 prepared commit', + 409, + ); + } + if (mode === 'existing-branch' && !remoteHead) { + throw new RepositoryError( + 'SCENARIO_BRANCH_REMOTE_CHANGED', + '准备 merge 后远端场景测试分支已被删除，未重新创建', + 409, + ); + } + + try { + // An empty expected object is a create-only compare-and-swap: despite Git's + // option name, it cannot update or overwrite an existing remote ref. + const pushArguments = + mode === 'initial-create' + ? [ + 'push', + `--force-with-lease=refs/heads/${branch}:`, + 'origin', + `${prepared}:refs/heads/${branch}`, + ] + : ['push', 'origin', `${prepared}:refs/heads/${branch}`]; + await this.run(pushArguments); + } catch (error) { + if (!(error instanceof GitCommandError)) throw error; + await this.fetch(); + const recoveredHead = await this.remoteBranchHead(branch); + if (recoveredHead && (await this.isAncestor(prepared, recoveredHead))) return prepared; + throw new RepositoryError( + 'PUSH_REJECTED', + mode === 'initial-create' + ? '首次创建场景测试分支时发生远端竞争，未发布 prepared commit' + : '场景测试分支推送被拒绝，未执行 force push', + 409, + ); + } + await this.fetch(); + const publishedHead = await this.remoteBranchHead(branch); + if (!publishedHead || !(await this.isAncestor(prepared, publishedHead))) { + throw new RepositoryError('PUSH_REJECTED', 'prepared commit 发布后无法在远端分支验证', 502); + } + return prepared; + } + + async isPublishedOnBranch(branch: string, commit: string): Promise { + assertBranchName(branch); + const normalized = normalizeSha(commit); + await this.fetch(); + const remoteHead = await this.remoteBranchHead(branch); + return remoteHead !== null && (await this.isAncestor(normalized, remoteHead)); + } + + async readInternalRef(queueId: number): Promise { + assertQueueId(queueId); + await this.ensureClone(); + try { + return normalizeSha( + (await this.run(['rev-parse', '--verify', mergeRequestRef(queueId)])).stdout, + ); + } catch (error) { + if (error instanceof GitCommandError) return null; + throw error; + } + } + + async listInternalMergeRequestIds(): Promise { + await this.ensureClone(); + const output = await this.run([ + 'for-each-ref', + '--format=%(refname)', + 'refs/luowang/merge-requests/', + ]); + return output.stdout + .split(/\r?\n/) + .map((line) => line.trim().match(/^refs\/luowang\/merge-requests\/(\d+)$/)?.[1]) + .filter((value): value is string => value !== undefined) + .map(Number) + .filter((value) => Number.isSafeInteger(value) && value > 0) + .sort((left, right) => left - right); + } + + async deleteInternalRef(queueId: number): Promise { + assertQueueId(queueId); + await this.ensureClone(); + if (!(await this.readInternalRef(queueId))) return; + await this.run(['update-ref', '-d', mergeRequestRef(queueId)]); + } + async publishRunReports( branch: string, runId: string, @@ -826,6 +914,117 @@ export class GitRepository { return changes; } + private async resolvePublishedSourceCommit(sourceRef: string): Promise { + assertRef(sourceRef); + if (sourceRef.startsWith('refs/luowang/') || sourceRef.startsWith('refs/remotes/')) { + throw new RepositoryError('TARGET_INVALID', '无法解析或验证远端来源 Git ref', 400); + } + + if (SHA_PATTERN.test(sourceRef)) { + let commit: string; + try { + commit = await this.resolveCommit(sourceRef); + } catch { + throw new RepositoryError('TARGET_INVALID', '无法解析或验证远端来源 Git ref', 400); + } + const containing = await this.run([ + 'for-each-ref', + '--contains', + commit, + '--format=%(refname)', + 'refs/remotes/origin/', + ]); + const onRemoteBranch = containing.stdout + .split(/\r?\n/) + .map((line) => line.trim()) + .some( + (ref) => ref.startsWith('refs/remotes/origin/') && ref !== 'refs/remotes/origin/HEAD', + ); + const remoteTags = await this.run(['ls-remote', '--tags', 'origin']); + const isAdvertisedTagCommit = remoteTags.stdout + .split(/\r?\n/) + .map((line) => line.trim().split(/\s+/, 2)[0]?.toLowerCase()) + .some((sha) => sha === commit); + if (!onRemoteBranch && !isAdvertisedTagCommit) { + throw new RepositoryError('TARGET_INVALID', '无法解析或验证远端来源 Git ref', 400); + } + return commit; + } + + const candidates: Array<{ + localRef: string; + remoteKind: 'heads' | 'tags'; + remoteRef: string; + }> = []; + if (sourceRef.startsWith('refs/heads/')) { + candidates.push({ + localRef: `refs/remotes/origin/${sourceRef.slice('refs/heads/'.length)}`, + remoteKind: 'heads', + remoteRef: sourceRef, + }); + } else if (sourceRef.startsWith('refs/tags/')) { + candidates.push({ localRef: sourceRef, remoteKind: 'tags', remoteRef: sourceRef }); + } else if (!sourceRef.startsWith('refs/')) { + candidates.push( + { + localRef: `refs/remotes/origin/${sourceRef}`, + remoteKind: 'heads', + remoteRef: `refs/heads/${sourceRef}`, + }, + { + localRef: `refs/tags/${sourceRef}`, + remoteKind: 'tags', + remoteRef: `refs/tags/${sourceRef}`, + }, + ); + } + for (const candidate of candidates) { + try { + const advertised = await this.run([ + 'ls-remote', + `--${candidate.remoteKind}`, + 'origin', + candidate.remoteRef, + ]); + const advertisedObject = normalizeSha(advertised.stdout.trim().split(/\s+/, 1)[0] ?? ''); + const localObject = normalizeSha( + (await this.run(['rev-parse', '--verify', '--end-of-options', candidate.localRef])) + .stdout, + ); + if (advertisedObject !== localObject) continue; + return normalizeSha( + ( + await this.run([ + 'rev-parse', + '--verify', + '--end-of-options', + `${candidate.localRef}^{commit}`, + ]) + ).stdout, + ); + } catch { + // Try the next explicitly allowed and currently advertised remote namespace. + } + } + throw new RepositoryError('TARGET_INVALID', '无法解析或验证远端来源 Git ref', 400); + } + + private async createInternalRef( + ref: string, + commit: string, + mode: PreparedMergeMode, + ): Promise { + await this.run([ + 'update-ref', + '--create-reflog', + '-m', + `luowang-${mode}`, + ref, + normalizeSha(commit), + '0000000000000000000000000000000000000000', + ]); + } + private async isGitWorktree(): Promise { try { await this.run(['rev-parse', '--git-dir']); @@ -925,6 +1124,25 @@ function sanitize(value: string, secret: string | undefined): string { return secret ? value.split(secret).join('[REDACTED]') : value; } +function normalizeSha(value: string): string { + const normalized = value.trim().split(/\r?\n/, 1)[0]?.toLowerCase() ?? ''; + if (!SHA_PATTERN.test(normalized)) { + throw new RepositoryError('TARGET_INVALID', 'Git commit SHA 格式无效', 400); + } + return normalized; +} + +function mergeRequestRef(queueId: number): string { + assertQueueId(queueId); + return `refs/luowang/merge-requests/${queueId}`; +} + +function assertQueueId(queueId: number): void { + if (!Number.isSafeInteger(queueId) || queueId <= 0) { + throw new RepositoryError('TARGET_INVALID', 'merge 请求队列 ID 无效', 400); + } +} + function assertRef(ref: string): void { if ( !REF_PATTERN.test(ref) || diff --git a/src/server/repository/service.ts b/src/server/repository/service.ts index 9a5a6eb..35418c0 100644 --- a/src/server/repository/service.ts +++ b/src/server/repository/service.ts @@ -20,7 +20,7 @@ import { } from './github.js'; import { GitRepository, - type GitMergeResult, + type PreparedMergeResult, type GitTreeEntry, type ReportFileName, type ReportPublishResult, @@ -31,13 +31,20 @@ import type { ScenarioPatchValidation } from './scenario-patch.js'; export interface RepositoryService { getStatus(): Promise; - ensureScenarioBranch(initialRef?: string): Promise<{ - created: boolean; - scenarioBranch: string; - head: string; - sourceCommit?: string; - }>; - mergeSourceRef(sourceRef: string, confirmed: boolean): Promise; + prepareMergeRequest( + sourceRef: string, + queueId: number, + initialization: boolean, + ): Promise; + publishPreparedMerge( + queueId: number, + preparedCommit: string, + mode: PreparedMergeResult['mode'] | null, + ): Promise; + isPublishedTarget(commit: string): Promise; + readMergeRequestRef(queueId: number): Promise; + listMergeRequestRefs(): Promise; + cleanupMergeRequestRef(queueId: number): Promise; checkoutTarget(target: string): Promise; assertScenarioHistory(baseCommit: string, targetCommit: string): Promise; cleanWorkspace(): Promise; @@ -144,33 +151,52 @@ class DefaultRepositoryService implements RepositoryService { }; } - async ensureScenarioBranch(initialRef?: string) { + async prepareMergeRequest( + sourceRef: string, + queueId: number, + initialization: boolean, + ): Promise { const config = this.requireRepositoryConfig(); - const repository = await this.getRepository(); - await repository.fetch(); - const existing = await repository.remoteBranchHead(config.scenarioBranch); - if (existing) { - return { created: false, scenarioBranch: config.scenarioBranch, head: existing }; - } - if (!initialRef || initialRef.trim() === '') { - throw new RepositoryError( - 'SCENARIO_BRANCH_INITIAL_REF_REQUIRED', - '场景测试分支不存在，请提供已确认的初始 branch、tag 或 SHA', - 400, - ); - } - const sourceCommit = await repository.resolveRemoteRef(initialRef.trim()); - if (!sourceCommit) { - throw new RepositoryError('TARGET_INVALID', `无法解析初始 Git ref：${initialRef}`, 400); - } - const head = await repository.createScenarioBranch(config.scenarioBranch, initialRef.trim()); - return { created: true, scenarioBranch: config.scenarioBranch, head, sourceCommit }; + return (await this.getRepository()).prepareMergeRequest( + config.scenarioBranch, + sourceRef.trim(), + queueId, + initialization, + ); } - async mergeSourceRef(sourceRef: string, confirmed: boolean): Promise { + async publishPreparedMerge( + queueId: number, + preparedCommit: string, + mode: PreparedMergeResult['mode'] | null, + ): Promise { const config = this.requireRepositoryConfig(); - const repository = await this.getRepository(); - return repository.mergeNoFastForward(config.scenarioBranch, sourceRef.trim(), confirmed); + return (await this.getRepository()).publishPreparedMerge( + config.scenarioBranch, + queueId, + preparedCommit, + mode, + ); + } + + async isPublishedTarget(commit: string): Promise { + const config = this.requireRepositoryConfig(); + return (await this.getRepository()).isPublishedOnBranch(config.scenarioBranch, commit); + } + + async readMergeRequestRef(queueId: number): Promise { + this.requireRepositoryConfig(); + return (await this.getRepository()).readInternalRef(queueId); + } + + async listMergeRequestRefs(): Promise { + this.requireRepositoryConfig(); + return (await this.getRepository()).listInternalMergeRequestIds(); + } + + async cleanupMergeRequestRef(queueId: number): Promise { + this.requireRepositoryConfig(); + await (await this.getRepository()).deleteInternalRef(queueId); } async checkoutTarget(target: string): Promise { diff --git a/src/server/runs/orchestrator.ts b/src/server/runs/orchestrator.ts index dcb5bb5..bd07529 100644 --- a/src/server/runs/orchestrator.ts +++ b/src/server/runs/orchestrator.ts @@ -175,7 +175,7 @@ class DefaultRunOrchestrator implements RunOrchestrator { this.startInProgress = true; try { - const runId = this.id(); + const runId = input.runId ?? this.id(); const startedAt = this.now().toISOString(); const workspace = await this.workspaceStore.create(runId); const state: RunState = { @@ -1444,6 +1444,9 @@ function assertRunInput(input: RunInput): void { if (!['git', 'schedule', 'manual', 'api'].includes(input.trigger)) { throw new RunOrchestratorError('RUN_REQUEST_INVALID', '测试请求来源无效'); } + if (input.runId !== undefined && !/^[0-9A-HJKMNP-TV-Z]{26}$/.test(input.runId)) { + throw new RunOrchestratorError('RUN_REQUEST_INVALID', '内部 Run ID 格式无效'); + } if (input.targetCommit !== undefined && input.targetCommit.trim() === '') { throw new RunOrchestratorError('RUN_REQUEST_INVALID', 'targetCommit 不能为空'); } diff --git a/src/server/runs/types.ts b/src/server/runs/types.ts index 3a4576b..73e8b89 100644 --- a/src/server/runs/types.ts +++ b/src/server/runs/types.ts @@ -38,6 +38,8 @@ export type RunArtifactName = export interface RunInput { request: string; trigger: RunTrigger; + /** Internal deterministic ID reserved by the FIFO coordinator. Never accepted from the API. */ + runId?: string; targetCommit?: string; initialization?: boolean; } diff --git a/src/shared/types.ts b/src/shared/types.ts index 986c30c..ddd2f74 100644 --- a/src/shared/types.ts +++ b/src/shared/types.ts @@ -419,6 +419,11 @@ export interface OperationsQueueItem { requestIds: string[]; request: string; targetRef: string | null; + requestKind: 'automatic-head' | 'manual-current-head' | 'manual-merge-source'; + sourceRef: string | null; + preparedMergeCommit: string | null; + preparedMergeMode: 'existing-branch' | 'initial-create' | null; + resolvedTargetCommit: string | null; status: 'queued' | 'running' | 'waiting_archive' | 'completed' | 'failed' | 'interrupted'; runId: string | null; claimedAt: string | null; diff --git a/src/web/App.tsx b/src/web/App.tsx index 184e093..8a6f37f 100644 --- a/src/web/App.tsx +++ b/src/web/App.tsx @@ -914,15 +914,19 @@ function RepositoryWorkspace() { const createBranch = async () => { const initialRef = window.prompt('输入用于创建场景测试分支的 branch、tag 或 SHA', 'main'); - if (!initialRef) return; + if ( + !initialRef || + !window.confirm(`确认从 ${initialRef} 首次创建场景测试分支并启动初始化 Run？`) + ) + return; setBusy('branch'); setError(''); try { - const result = await requestJson<{ head: string }>('/api/repository/scenario-branch', { + const result = await requestJson<{ queueId: number }>('/api/repository/merge', { method: 'POST', - body: JSON.stringify({ initialRef }), + body: JSON.stringify({ sourceRef: initialRef, confirmed: true, initialization: true }), }); - setMessage(`场景测试分支已准备：${result.head.slice(0, 12)}`); + setMessage(`首次创建与初始化请求已进入队列：#${result.queueId}`); await load(); } catch (cause: unknown) { setError(toUserMessage(cause, '场景测试分支创建失败')); @@ -937,15 +941,11 @@ function RepositoryWorkspace() { setBusy('merge'); setError(''); try { - const result = await requestJson<{ scenarioBranchHead: string; alreadyIncluded: boolean }>( - '/api/repository/merge', - { method: 'POST', body: JSON.stringify({ sourceRef, confirmed: true }) }, - ); - setMessage( - result.alreadyIncluded - ? '来源 ref 已经在场景测试分支历史中，不重复合并' - : `合并已发布：${result.scenarioBranchHead.slice(0, 12)}`, - ); + const result = await requestJson<{ queueId: number }>('/api/repository/merge', { + method: 'POST', + body: JSON.stringify({ sourceRef, confirmed: true }), + }); + setMessage(`merge 与固定 target 测试请求已进入队列：#${result.queueId}`); await load(); } catch (cause: unknown) { setError(toUserMessage(cause, '来源 ref 合并失败')); @@ -1026,7 +1026,7 @@ function RepositoryWorkspace() { disabled={busy !== null || !status?.configured} onClick={() => void createBranch()} > - 准备场景测试分支 + 排队首次创建并初始化 {status?.indexErrors.length ? ( @@ -1252,7 +1252,9 @@ function DashboardPanel({ onNavigate }: { onNavigate: (view: ConsoleView) => voi #{item.queueId} {item.trigger} {queueStatusLabel(item.status)} - {shortSha(item.runId ?? item.targetRef)} + + {shortSha(item.runId ?? item.resolvedTargetCommit ?? item.sourceRef)} + ))} @@ -1353,7 +1355,6 @@ function DashboardPanel({ onNavigate }: { onNavigate: (view: ConsoleView) => voi function RunRequestForm({ onSubmitted }: { onSubmitted: (message: string) => void }) { const [request, setRequest] = useState('验证当前场景测试分支的核心业务流程'); - const [targetCommit, setTargetCommit] = useState(''); const [busy, setBusy] = useState(false); const [error, setError] = useState(''); @@ -1366,10 +1367,7 @@ function RunRequestForm({ onSubmitted }: { onSubmitted: (message: string) => voi '/api/runs', { method: 'POST', - body: JSON.stringify({ - request, - ...(targetCommit.trim() ? { targetCommit: targetCommit.trim() } : {}), - }), + body: JSON.stringify({ request }), }, ); onSubmitted( @@ -1394,14 +1392,10 @@ function RunRequestForm({ onSubmitted }: { onSubmitted: (message: string) => voi onChange={(event) => setRequest(event.target.value)} /> - - setTargetCommit(event.target.value)} - placeholder="留空使用场景测试分支 HEAD" - /> - +

+ target 将在该请求轮到执行时固定为远端场景测试分支 HEAD；指定 branch、tag 或 SHA + 请使用仓库页的 merge-source 入口。 +

{error &&

{error}

}