From 6cb9a075d79b27e5400ea37d48a050431f1cb83e Mon Sep 17 00:00:00 2001 From: whyour Date: Tue, 18 Aug 2026 09:26:12 +0800 Subject: [PATCH] feat(ql3): reconcile legacy shadow runs on startup --- back/app.ts | 5 + .../legacyShadowStartupRecoverySource.ts | 319 ++++++++++++ ...tstrapLegacyShadowStartupReconciliation.ts | 169 ++++++ .../legacyShadowStartupReconciler.ts | 490 ++++++++++++++++++ .../compatibility/legacyExecutionBridge.ts | 4 + .../ports/legacyShadowStartupRecovery.ts | 71 +++ docs/QINGLONG_3_0_ARCHITECTURE_RFC.md | 19 +- ...-crontab-compatibility-and-shadow-write.md | 16 +- ...ed-legacy-shadow-startup-reconciliation.md | 88 ++++ docs/adr/README.md | 1 + ...LegacyShadowStartupReconciliation.test.cjs | 149 ++++++ ...legacyShadowStartupReconciliation.test.cjs | 368 +++++++++++++ 12 files changed, 1691 insertions(+), 8 deletions(-) create mode 100644 back/runtime/adapters/legacy-sequelize/legacyShadowStartupRecoverySource.ts create mode 100644 back/runtime/adapters/legacy/bootstrapLegacyShadowStartupReconciliation.ts create mode 100644 back/runtime/application/legacyShadowStartupReconciler.ts create mode 100644 back/runtime/ports/legacyShadowStartupRecovery.ts create mode 100644 docs/adr/ADR-0448-bounded-legacy-shadow-startup-reconciliation.md create mode 100644 test/back/bootstrapLegacyShadowStartupReconciliation.test.cjs create mode 100644 test/back/legacyShadowStartupReconciliation.test.cjs diff --git a/back/app.ts b/back/app.ts index 515ac93d..921ac86b 100644 --- a/back/app.ts +++ b/back/app.ts @@ -245,6 +245,11 @@ class Application { const appLoader = await import('./loaders/app'); await appLoader.default({ app: this.app }); + const { bootstrapLegacyShadowStartupReconciliation } = await import( + './runtime/adapters/legacy/bootstrapLegacyShadowStartupReconciliation' + ); + await bootstrapLegacyShadowStartupReconciliation(); + const { bootstrapDefaultManualPrimaryRuntime } = await import( './runtime/adapters/legacy/bootstrapDefaultManualPrimaryRuntime' ); diff --git a/back/runtime/adapters/legacy-sequelize/legacyShadowStartupRecoverySource.ts b/back/runtime/adapters/legacy-sequelize/legacyShadowStartupRecoverySource.ts new file mode 100644 index 00000000..d0145502 --- /dev/null +++ b/back/runtime/adapters/legacy-sequelize/legacyShadowStartupRecoverySource.ts @@ -0,0 +1,319 @@ +import { QueryTypes, type Sequelize } from 'sequelize'; +import { + RUN_ATTEMPT_TABLE, + RUN_TABLE, +} from '../../../migrations/0002-run-schema'; +import { RUNNING_INSTANCE_TABLE } from '../../../migrations/0003-running-instance-run-reference'; +import type { + ExecutionOrigin, + RunAttemptStatus, + RunStatus, +} from '../../domain/run'; +import { + MAX_LEGACY_SHADOW_STARTUP_BATCH_SIZE, + MAX_LEGACY_SHADOW_STARTUP_EVIDENCE, + type LegacyRunningInstanceEvidence, + type LegacyRunningInstanceEvidencePage, + type LegacyShadowStartupCandidate, + type LegacyShadowStartupCursor, + type LegacyShadowStartupPage, + type LegacyShadowStartupRecoverySource, +} from '../../ports/legacyShadowStartupRecovery'; + +interface RunRow { + runId: string; + legacyCronId: number | null; + origin: string; + runStatus: string; + createdAtMs: number | string; +} + +interface AttemptCountRow { + runId: string; + activeAttemptCount: number | string; +} + +interface AttemptRow { + attemptId: string; + runId: string; + status: string; + pid: number | null; + logArtifactId: string | null; + createdAtMs: number | string; + startedAtMs: number | string | null; +} + +interface RunningInstanceRow { + pid: number | null; + logPath: string | null; + startedAt: number | string; + finishedAt: number | string | null; + status: number | string; + exitCode: number | null; +} + +export type LegacyLogArtifactIdFactory = (logPath: string) => string; + +function safeInteger(value: number | string, name: string): number { + const parsed = typeof value === 'number' ? value : Number(value); + if (!Number.isSafeInteger(parsed) || parsed < 0) { + throw new TypeError(`${name} is invalid`); + } + return parsed; +} + +function secondsToMilliseconds(value: number | string, name: string): number { + const seconds = safeInteger(value, name); + const milliseconds = seconds * 1_000; + if (!Number.isSafeInteger(milliseconds)) { + throw new TypeError(`${name} is invalid`); + } + return milliseconds; +} + +function optionalPositiveInteger( + value: number | null, + name: string, +): number | undefined { + if (value === null) return undefined; + const parsed = safeInteger(value, name); + if (parsed < 1) throw new TypeError(`${name} is invalid`); + return parsed; +} + +function assertLimit(limit: number, maximum: number): void { + if (!Number.isSafeInteger(limit) || limit < 1 || limit > maximum) { + throw new RangeError(`limit must be between 1 and ${maximum}`); + } +} + +function outcome(status: number): LegacyRunningInstanceEvidence['outcome'] { + if (status === 0) return 'running'; + if (status === 1) return 'succeeded'; + if (status === 2) return 'stopped'; + if (status === 3) return 'failed'; + throw new TypeError('RunningInstance status is invalid'); +} + +export class LegacySequelizeShadowStartupRecoverySource + implements LegacyShadowStartupRecoverySource +{ + constructor( + private readonly database: Sequelize, + private readonly createLogArtifactId: LegacyLogArtifactIdFactory, + ) {} + + async listCandidates({ + origins, + cursor, + limit = MAX_LEGACY_SHADOW_STARTUP_BATCH_SIZE, + }: { + origins: readonly ExecutionOrigin[]; + cursor?: LegacyShadowStartupCursor; + limit?: number; + }): Promise { + assertLimit(limit, MAX_LEGACY_SHADOW_STARTUP_BATCH_SIZE); + const enabledOrigins = [...new Set(origins)]; + if (enabledOrigins.length === 0) { + return { candidates: [], truncated: false }; + } + if ( + cursor !== undefined && + (!Number.isSafeInteger(cursor.createdAtMs) || + cursor.createdAtMs < 0 || + cursor.runId.length === 0) + ) { + throw new TypeError('Legacy Shadow startup cursor is invalid'); + } + + const runRows = await this.database.query( + `SELECT + id AS "runId", + legacy_cron_id AS "legacyCronId", + execution_origin AS origin, + status AS "runStatus", + created_at_ms AS "createdAtMs" + FROM ${RUN_TABLE} + WHERE execution_owner = 'legacy' + AND execution_origin IN (:origins) + AND status IN ('queued', 'dispatching', 'running') + ${ + cursor === undefined + ? '' + : `AND ( + created_at_ms > :cursorCreatedAtMs OR + (created_at_ms = :cursorCreatedAtMs AND id > :cursorRunId) + )` + } + ORDER BY created_at_ms ASC, id ASC + LIMIT :fetchLimit`, + { + replacements: { + origins: enabledOrigins, + fetchLimit: limit + 1, + ...(cursor === undefined + ? {} + : { + cursorCreatedAtMs: cursor.createdAtMs, + cursorRunId: cursor.runId, + }), + }, + type: QueryTypes.SELECT, + }, + ); + const truncated = runRows.length > limit; + const boundedRuns = runRows.slice(0, limit); + if (boundedRuns.length === 0) { + return { candidates: [], truncated }; + } + + const runIds = boundedRuns.map((run) => run.runId); + const countRows = await this.database.query( + `SELECT run_id AS "runId", COUNT(*) AS "activeAttemptCount" + FROM ${RUN_ATTEMPT_TABLE} + WHERE run_id IN (:runIds) + AND status IN ('claimed', 'starting', 'running') + GROUP BY run_id`, + { + replacements: { runIds }, + type: QueryTypes.SELECT, + }, + ); + const counts = new Map( + countRows.map((row) => [ + row.runId, + safeInteger(row.activeAttemptCount, 'activeAttemptCount'), + ]), + ); + const singleAttemptRunIds = runIds.filter( + (runId) => counts.get(runId) === 1, + ); + const attemptRows = + singleAttemptRunIds.length === 0 + ? [] + : await this.database.query( + `SELECT + id AS "attemptId", + run_id AS "runId", + status, + pid, + log_artifact_id AS "logArtifactId", + created_at_ms AS "createdAtMs", + started_at_ms AS "startedAtMs" + FROM ${RUN_ATTEMPT_TABLE} + WHERE run_id IN (:runIds) + AND status IN ('claimed', 'starting', 'running')`, + { + replacements: { runIds: singleAttemptRunIds }, + type: QueryTypes.SELECT, + }, + ); + const attempts = new Map(attemptRows.map((row) => [row.runId, row])); + const candidates: LegacyShadowStartupCandidate[] = boundedRuns.map( + (run) => { + const activeAttemptCount = counts.get(run.runId) ?? 0; + const attempt = attempts.get(run.runId); + return { + runId: run.runId, + ...(run.legacyCronId === null + ? {} + : { + legacyCronId: optionalPositiveInteger( + run.legacyCronId, + 'legacyCronId', + ), + }), + origin: run.origin as ExecutionOrigin, + runStatus: run.runStatus as RunStatus, + createdAtMs: safeInteger(run.createdAtMs, 'createdAtMs'), + activeAttemptCount, + ...(attempt === undefined + ? {} + : { + attempt: { + attemptId: attempt.attemptId, + status: attempt.status as RunAttemptStatus, + ...(attempt.pid === null ? {} : { pid: attempt.pid }), + ...(attempt.logArtifactId === null + ? {} + : { logArtifactId: attempt.logArtifactId }), + createdAtMs: safeInteger( + attempt.createdAtMs, + 'attempt.createdAtMs', + ), + ...(attempt.startedAtMs === null + ? {} + : { + startedAtMs: safeInteger( + attempt.startedAtMs, + 'attempt.startedAtMs', + ), + }), + }, + }), + }; + }, + ); + const last = candidates.at(-1); + return { + candidates, + truncated, + ...(truncated && last + ? { + nextCursor: { + createdAtMs: last.createdAtMs, + runId: last.runId, + }, + } + : {}), + }; + } + + async listRunningInstanceEvidence({ + legacyCronId, + limit = MAX_LEGACY_SHADOW_STARTUP_EVIDENCE, + }: { + legacyCronId: number; + limit?: number; + }): Promise { + if (!Number.isSafeInteger(legacyCronId) || legacyCronId < 1) { + throw new RangeError('legacyCronId must be a positive safe integer'); + } + assertLimit(limit, MAX_LEGACY_SHADOW_STARTUP_EVIDENCE); + const rows = await this.database.query( + `SELECT + pid, + log_path AS "logPath", + started_at AS "startedAt", + finished_at AS "finishedAt", + status, + exit_code AS "exitCode" + FROM ${RUNNING_INSTANCE_TABLE} + WHERE cron_id = :legacyCronId + ORDER BY started_at DESC, id DESC + LIMIT :fetchLimit`, + { + replacements: { legacyCronId, fetchLimit: limit + 1 }, + type: QueryTypes.SELECT, + }, + ); + const truncated = rows.length > limit; + return { + evidence: rows.slice(0, limit).map((row) => ({ + ...(row.pid === null ? {} : { pid: row.pid }), + ...(row.logPath === null + ? {} + : { logArtifactId: this.createLogArtifactId(row.logPath) }), + startedAtMs: secondsToMilliseconds(row.startedAt, 'startedAt'), + ...(row.finishedAt === null + ? {} + : { + finishedAtMs: secondsToMilliseconds(row.finishedAt, 'finishedAt'), + }), + outcome: outcome(safeInteger(row.status, 'status')), + ...(row.exitCode === null ? {} : { exitCode: row.exitCode }), + })), + truncated, + }; + } +} diff --git a/back/runtime/adapters/legacy/bootstrapLegacyShadowStartupReconciliation.ts b/back/runtime/adapters/legacy/bootstrapLegacyShadowStartupReconciliation.ts new file mode 100644 index 00000000..0488628b --- /dev/null +++ b/back/runtime/adapters/legacy/bootstrapLegacyShadowStartupReconciliation.ts @@ -0,0 +1,169 @@ +import Logger from '../../../loaders/logger'; +import type { ExecutionOrigin } from '../../domain/run'; +import { + parseDeploymentProfile, + type DeploymentProfile, +} from '../../domain/deploymentProfile'; +import { configuredLegacyShadowOrigins } from '../../compatibility/legacyExecutionBridge'; +import { createLegacyLogArtifactId } from '../../compatibility/legacyTaskRevision'; +import type { LegacyShadowStartupSummary } from '../../application/legacyShadowStartupReconciler'; + +export interface LegacyShadowStartupBudget { + pageSize: number; + maxPages: number; +} + +export interface LegacyShadowStartupRequest extends LegacyShadowStartupBudget { + origins: readonly ExecutionOrigin[]; + profile: 'edge' | 'standalone'; +} + +export type LegacyShadowStartupAudit = + | { + state: 'disabled'; + } + | { + state: 'profile_rejected'; + profile: 'cluster-control' | 'worker'; + } + | { + state: 'reconciled' | 'incomplete'; + profile: 'edge' | 'standalone'; + origins: number; + summary: Omit & { + resumeAvailable: boolean; + }; + } + | { + state: 'failed'; + errorType: string; + }; + +export interface BootstrapLegacyShadowStartupOptions { + origins?: readonly ExecutionOrigin[]; + profile?: DeploymentProfile; + execute?: ( + request: LegacyShadowStartupRequest, + ) => Promise; + audit?: (record: LegacyShadowStartupAudit) => void | Promise; +} + +const BUDGETS: Readonly< + Record<'edge' | 'standalone', LegacyShadowStartupBudget> +> = { + edge: { pageSize: 8, maxPages: 1 }, + standalone: { pageSize: 32, maxPages: 4 }, +}; + +function auditSummary(summary: LegacyShadowStartupSummary): Omit< + LegacyShadowStartupSummary, + 'nextCursor' +> & { + resumeAvailable: boolean; +} { + const { nextCursor, ...bounded } = summary; + return { ...bounded, resumeAvailable: nextCursor !== undefined }; +} + +async function executeDefault( + request: LegacyShadowStartupRequest, +): Promise { + const [data, repositoryModule, sourceModule, writerModule, reconcilerModule] = + await Promise.all([ + import('../../../data'), + import('../legacy-sequelize/runRepository'), + import('../legacy-sequelize/legacyShadowStartupRecoverySource'), + import('../../application/legacyShadowRunWriter'), + import('../../application/legacyShadowStartupReconciler'), + ]); + const repository = new repositoryModule.LegacySequelizeRunRepository( + data.sequelize, + ); + const source = new sourceModule.LegacySequelizeShadowStartupRecoverySource( + data.sequelize, + createLegacyLogArtifactId, + ); + const writer = new writerModule.LegacyShadowRunWriter(repository); + const reconciler = new reconcilerModule.LegacyShadowStartupReconciler( + repository, + source, + writer, + ); + return new reconcilerModule.LegacyShadowStartupSupervisor(reconciler).run({ + origins: request.origins, + pageSize: request.pageSize, + maxPages: request.maxPages, + }); +} + +/** + * Runs once after Legacy startup normalization and before HTTP listen. Disabled + * and non-local Profiles never import a Repository or touch the database. + */ +export async function bootstrapLegacyShadowStartupReconciliation( + options: BootstrapLegacyShadowStartupOptions = {}, +): Promise { + const origins = [ + ...new Set(options.origins ?? configuredLegacyShadowOrigins()), + ]; + const audit = + options.audit ?? + ((record: LegacyShadowStartupAudit) => { + Logger.info(`[ql3-shadow-startup] ${JSON.stringify(record)}`); + }); + if (origins.length === 0) { + const record: LegacyShadowStartupAudit = { state: 'disabled' }; + try { + await audit(record); + } catch { + // Shadow audit output must not affect Legacy startup. + } + return record; + } + + try { + const profile = + options.profile ?? + parseDeploymentProfile(process.env.QL_DEPLOYMENT_PROFILE); + if (profile === 'cluster-control' || profile === 'worker') { + const record: LegacyShadowStartupAudit = { + state: 'profile_rejected', + profile, + }; + try { + await audit(record); + } catch { + // Shadow audit output must not affect Legacy startup. + } + return record; + } + const summary = await (options.execute ?? executeDefault)({ + origins, + profile, + ...BUDGETS[profile], + }); + const record: LegacyShadowStartupAudit = { + state: summary.remaining ? 'incomplete' : 'reconciled', + profile, + origins: origins.length, + summary: auditSummary(summary), + }; + try { + await audit(record); + } catch { + // Shadow audit output must not affect Legacy startup. + } + return record; + } catch (error) { + const record: LegacyShadowStartupAudit = { + state: 'failed', + errorType: error instanceof Error ? error.name : 'unknown', + }; + try { + await audit(record); + } catch { + // Shadow audit output must not affect Legacy startup. + } + return record; + } +} diff --git a/back/runtime/application/legacyShadowStartupReconciler.ts b/back/runtime/application/legacyShadowStartupReconciler.ts new file mode 100644 index 00000000..9e608aa9 --- /dev/null +++ b/back/runtime/application/legacyShadowStartupReconciler.ts @@ -0,0 +1,490 @@ +import { v7 as uuidV7 } from 'uuid'; +import type { RunAttemptRecord, RunRecord, RunStatus } from '../domain/run'; +import { + RUN_TRANSITIONS, + isTerminalRunAttemptStatus, + isTerminalRunStatus, +} from '../domain/runStateMachine'; +import type { + LegacyRunningInstanceEvidence, + LegacyShadowStartupAttempt, + LegacyShadowStartupCandidate, + LegacyShadowStartupCursor, + LegacyShadowStartupRecoverySource, +} from '../ports/legacyShadowStartupRecovery'; +import { MAX_LEGACY_SHADOW_STARTUP_BATCH_SIZE } from '../ports/legacyShadowStartupRecovery'; +import type { RunRepository } from '../ports/runRepository'; +import type { LegacyShadowRunWriter } from './legacyShadowRunWriter'; +import { RunCommandService } from './runCommandService'; + +export const MAX_LEGACY_SHADOW_STARTUP_PAGES = 64; + +export interface LegacyShadowStartupClock { + now(): number; +} + +export interface LegacyShadowStartupReconcileSummary { + scanned: number; + completed: number; + cancelled: number; + abandoned: number; + markedLost: number; + repaired: number; + pending: number; + ambiguous: number; + skipped: number; + failed: number; + truncated: boolean; + nextCursor?: LegacyShadowStartupCursor; +} + +export type LegacyShadowStartupStopReason = + | 'complete' + | 'page_limit' + | 'cursor_stalled'; + +export interface LegacyShadowStartupSummary + extends Omit< + LegacyShadowStartupReconcileSummary, + 'truncated' | 'nextCursor' + > { + pages: number; + stopReason: LegacyShadowStartupStopReason; + remaining: boolean; + nextCursor?: LegacyShadowStartupCursor; +} + +type EvidenceSelection = + | { status: 'matched'; evidence: LegacyRunningInstanceEvidence } + | { status: 'none' | 'ambiguous' }; + +function sameCursor( + left: LegacyShadowStartupCursor | undefined, + right: LegacyShadowStartupCursor, +): boolean { + return ( + left !== undefined && + left.createdAtMs === right.createdAtMs && + left.runId === right.runId + ); +} + +function selectEvidence( + attempt: Pick, + evidence: readonly LegacyRunningInstanceEvidence[], +): EvidenceSelection { + const byLog = + attempt.logArtifactId === undefined + ? [] + : evidence.filter( + (candidate) => candidate.logArtifactId === attempt.logArtifactId, + ); + const byPid = + attempt.pid === undefined + ? [] + : evidence.filter((candidate) => candidate.pid === attempt.pid); + + if (attempt.logArtifactId !== undefined && attempt.pid !== undefined) { + const byPidSet = new Set(byPid); + const intersection = byLog.filter((candidate) => byPidSet.has(candidate)); + if (intersection.length === 1) { + return { status: 'matched', evidence: intersection[0] }; + } + if ( + intersection.length > 1 || + byLog.length > 1 || + byPid.length > 1 || + (byLog.length > 0 && byPid.length > 0) + ) { + return { status: 'ambiguous' }; + } + if (byLog.length === 1) { + return { status: 'matched', evidence: byLog[0] }; + } + if (byPid.length === 1) { + return { status: 'matched', evidence: byPid[0] }; + } + return { status: 'none' }; + } + if (attempt.logArtifactId !== undefined) { + if (byLog.length === 1) { + return { status: 'matched', evidence: byLog[0] }; + } + return { status: byLog.length > 1 ? 'ambiguous' : 'none' }; + } + if (attempt.pid !== undefined) { + if (byPid.length === 1) { + return { status: 'matched', evidence: byPid[0] }; + } + return { status: byPid.length > 1 ? 'ambiguous' : 'none' }; + } + if (evidence.length === 1) { + return { status: 'matched', evidence: evidence[0] }; + } + return { status: evidence.length > 1 ? 'ambiguous' : 'none' }; +} + +function terminalRunTarget(attempt: RunAttemptRecord): RunStatus | undefined { + if (attempt.status === 'succeeded') return 'succeeded'; + if (attempt.status === 'failed') return 'failed'; + if (attempt.status === 'cancelled') return 'cancelled'; + if (attempt.status === 'timed_out') return 'timed_out'; + if (attempt.status === 'lost') return 'lost'; + return undefined; +} + +/** Reconciles one bounded page after Legacy startup normalization. */ +export class LegacyShadowStartupReconciler { + private readonly commands: RunCommandService; + private readonly clock: LegacyShadowStartupClock; + + constructor( + private readonly repository: RunRepository, + private readonly source: LegacyShadowStartupRecoverySource, + private readonly writer: Pick< + LegacyShadowRunWriter, + 'exited' | 'cancelled' + >, + options: { + clock?: LegacyShadowStartupClock; + createEventId?: () => string; + } = {}, + ) { + this.clock = options.clock ?? { now: Date.now }; + this.commands = new RunCommandService( + repository, + options.createEventId ?? uuidV7, + ); + } + + async reconcileBatch(options: { + origins: readonly RunRecord['executionOrigin'][]; + cursor?: LegacyShadowStartupCursor; + limit?: number; + }): Promise { + const enabledOrigins = [...new Set(options.origins)]; + const page = await this.source.listCandidates({ + origins: enabledOrigins, + ...(options.cursor === undefined ? {} : { cursor: options.cursor }), + ...(options.limit === undefined ? {} : { limit: options.limit }), + }); + const summary: LegacyShadowStartupReconcileSummary = { + scanned: page.candidates.length, + completed: 0, + cancelled: 0, + abandoned: 0, + markedLost: 0, + repaired: 0, + pending: 0, + ambiguous: 0, + skipped: 0, + failed: 0, + truncated: page.truncated, + ...(page.nextCursor === undefined ? {} : { nextCursor: page.nextCursor }), + }; + const origins = new Set(enabledOrigins); + for (const candidate of page.candidates) { + try { + await this.reconcileCandidate(candidate, origins, summary); + } catch { + summary.failed += 1; + } + } + return summary; + } + + private async reconcileCandidate( + candidate: LegacyShadowStartupCandidate, + origins: ReadonlySet, + summary: LegacyShadowStartupReconcileSummary, + ): Promise { + const run = await this.repository.findRunById(candidate.runId); + if ( + !run || + run.executionOwner !== 'legacy' || + !origins.has(run.executionOrigin) || + isTerminalRunStatus(run.status) || + run.status === 'lost' + ) { + summary.skipped += 1; + return; + } + if (candidate.activeAttemptCount === 0) { + await this.repairTerminalAttempt(run, summary); + return; + } + if (candidate.activeAttemptCount !== 1 || !candidate.attempt) { + summary.ambiguous += 1; + return; + } + const attempt = await this.repository.findAttemptById( + candidate.attempt.attemptId, + ); + if ( + !attempt || + attempt.runId !== run.id || + isTerminalRunAttemptStatus(attempt.status) + ) { + summary.skipped += 1; + return; + } + + const evidence = + candidate.legacyCronId === undefined + ? { evidence: [], truncated: false } + : await this.source.listRunningInstanceEvidence({ + legacyCronId: candidate.legacyCronId, + }); + if (evidence.truncated) { + summary.ambiguous += 1; + return; + } + const selected = selectEvidence(attempt, evidence.evidence); + if (selected.status === 'ambiguous') { + summary.ambiguous += 1; + return; + } + if ( + selected.status === 'matched' && + selected.evidence.finishedAtMs !== undefined + ) { + const atMs = this.atOrAfter(run, attempt, selected.evidence.finishedAtMs); + if (selected.evidence.outcome === 'stopped') { + await this.writer.cancelled( + { runId: run.id, attemptId: attempt.id }, + { atMs, reason: 'reconcile' }, + ); + summary.cancelled += 1; + return; + } + if ( + selected.evidence.outcome === 'succeeded' || + selected.evidence.outcome === 'failed' + ) { + const succeeded = selected.evidence.outcome === 'succeeded'; + await this.writer.exited( + { runId: run.id, attemptId: attempt.id }, + { + atMs, + exitCode: succeeded + ? 0 + : selected.evidence.exitCode === undefined || + selected.evidence.exitCode === 0 + ? 1 + : selected.evidence.exitCode, + }, + ); + summary.completed += 1; + return; + } + } + + if (run.executionOrigin === 'scheduled_system') { + summary.pending += 1; + return; + } + if (run.status === 'queued' && attempt.status === 'claimed') { + await this.abandon(run, attempt); + summary.abandoned += 1; + return; + } + if ( + (run.status === 'dispatching' || run.status === 'running') && + ['claimed', 'starting', 'running'].includes(attempt.status) + ) { + await this.markLost(run, attempt); + summary.markedLost += 1; + return; + } + summary.ambiguous += 1; + } + + private async repairTerminalAttempt( + run: RunRecord, + summary: LegacyShadowStartupReconcileSummary, + ): Promise { + const attempt = await this.repository.findLatestAttemptByRunId(run.id); + const target = attempt ? terminalRunTarget(attempt) : undefined; + if ( + !attempt || + target === undefined || + !RUN_TRANSITIONS[run.status].includes(target) + ) { + summary.ambiguous += 1; + return; + } + await this.commands.transitionRun({ + runId: run.id, + to: target, + expectedVersion: run.version, + atMs: this.atOrAfter(run, attempt), + ...(attempt.errorCode === undefined + ? {} + : { errorCode: attempt.errorCode }), + ...(attempt.errorSummary === undefined + ? {} + : { errorSummary: attempt.errorSummary }), + actor: { type: 'reconciler' }, + dedupeKey: `legacy-startup-repair:${attempt.id}:${target}`, + }); + summary.repaired += 1; + } + + private async abandon( + run: RunRecord, + attempt: RunAttemptRecord, + ): Promise { + const atMs = this.atOrAfter(run, attempt); + const attemptResult = await this.commands.transitionRunAttempt({ + runId: run.id, + attemptId: attempt.id, + to: 'cancelled', + expectedRunVersion: run.version, + atMs, + errorCode: 'LEGACY_RECONCILE_ACCEPTANCE_ABANDONED', + errorSummary: 'Legacy owner restarted before spawn was observed', + actor: { type: 'reconciler' }, + dedupeKey: `legacy-startup-abandoned-attempt:${attempt.id}`, + }); + await this.commands.transitionRun({ + runId: run.id, + to: 'cancelled', + expectedVersion: attemptResult.run.version, + atMs, + errorCode: 'LEGACY_RECONCILE_ACCEPTANCE_ABANDONED', + errorSummary: 'Legacy owner restarted before spawn was observed', + actor: { type: 'reconciler' }, + dedupeKey: `legacy-startup-abandoned-run:${attempt.id}`, + }); + } + + private async markLost( + run: RunRecord, + attempt: RunAttemptRecord, + ): Promise { + const atMs = this.atOrAfter(run, attempt); + const attemptResult = await this.commands.transitionRunAttempt({ + runId: run.id, + attemptId: attempt.id, + to: 'lost', + expectedRunVersion: run.version, + atMs, + errorCode: 'LEGACY_RECONCILE_OWNER_LOST', + errorSummary: 'Legacy worker restarted without terminal evidence', + actor: { type: 'reconciler' }, + dedupeKey: `legacy-startup-lost-attempt:${attempt.id}`, + }); + await this.commands.transitionRun({ + runId: run.id, + to: 'lost', + expectedVersion: attemptResult.run.version, + atMs, + errorCode: 'LEGACY_RECONCILE_OWNER_LOST', + errorSummary: 'Legacy worker restarted without terminal evidence', + actor: { type: 'reconciler' }, + dedupeKey: `legacy-startup-lost-run:${attempt.id}`, + }); + } + + private atOrAfter( + run: RunRecord, + attempt: RunAttemptRecord, + evidenceAtMs?: number, + ): number { + return Math.max( + this.clock.now(), + run.createdAtMs, + run.startedAtMs ?? 0, + attempt.createdAtMs, + attempt.startedAtMs ?? 0, + evidenceAtMs ?? 0, + ); + } +} + +/** Runs a complete but Profile-bounded, timer-free startup pass. */ +export class LegacyShadowStartupSupervisor { + constructor( + private readonly reconciler: Pick< + LegacyShadowStartupReconciler, + 'reconcileBatch' + >, + ) {} + + async run(options: { + origins: readonly RunRecord['executionOrigin'][]; + cursor?: LegacyShadowStartupCursor; + pageSize?: number; + maxPages?: number; + }): Promise { + const pageSize = options.pageSize ?? 8; + const maxPages = options.maxPages ?? 1; + if ( + !Number.isSafeInteger(pageSize) || + pageSize < 1 || + pageSize > MAX_LEGACY_SHADOW_STARTUP_BATCH_SIZE + ) { + throw new RangeError( + 'pageSize must be between 1 and MAX_LEGACY_SHADOW_STARTUP_BATCH_SIZE', + ); + } + if ( + !Number.isSafeInteger(maxPages) || + maxPages < 1 || + maxPages > MAX_LEGACY_SHADOW_STARTUP_PAGES + ) { + throw new RangeError( + 'maxPages must be between 1 and MAX_LEGACY_SHADOW_STARTUP_PAGES', + ); + } + + const total: LegacyShadowStartupSummary = { + pages: 0, + scanned: 0, + completed: 0, + cancelled: 0, + abandoned: 0, + markedLost: 0, + repaired: 0, + pending: 0, + ambiguous: 0, + skipped: 0, + failed: 0, + stopReason: 'complete', + remaining: false, + }; + let cursor = options.cursor; + for (let pageNumber = 0; pageNumber < maxPages; pageNumber += 1) { + const page = await this.reconciler.reconcileBatch({ + origins: options.origins, + ...(cursor === undefined ? {} : { cursor }), + limit: pageSize, + }); + total.pages += 1; + total.scanned += page.scanned; + total.completed += page.completed; + total.cancelled += page.cancelled; + total.abandoned += page.abandoned; + total.markedLost += page.markedLost; + total.repaired += page.repaired; + total.pending += page.pending; + total.ambiguous += page.ambiguous; + total.skipped += page.skipped; + total.failed += page.failed; + if (!page.truncated) return total; + if (!page.nextCursor || sameCursor(cursor, page.nextCursor)) { + total.stopReason = 'cursor_stalled'; + total.remaining = true; + return total; + } + cursor = page.nextCursor; + if (pageNumber === maxPages - 1) { + total.stopReason = 'page_limit'; + total.remaining = true; + total.nextCursor = cursor; + return total; + } + } + return total; + } +} diff --git a/back/runtime/compatibility/legacyExecutionBridge.ts b/back/runtime/compatibility/legacyExecutionBridge.ts index 1f929464..17159aa3 100644 --- a/back/runtime/compatibility/legacyExecutionBridge.ts +++ b/back/runtime/compatibility/legacyExecutionBridge.ts @@ -98,6 +98,10 @@ function readConfiguredOrigins(): ReadonlySet { return configuredOrigins; } +export function configuredLegacyShadowOrigins(): readonly ExecutionOrigin[] { + return [...readConfiguredOrigins()]; +} + async function createDefaultObserver(): Promise { const [data, repositoryModule, observerModule, writerModule, rolloutModule] = await Promise.all([ diff --git a/back/runtime/ports/legacyShadowStartupRecovery.ts b/back/runtime/ports/legacyShadowStartupRecovery.ts new file mode 100644 index 00000000..5ef87aa0 --- /dev/null +++ b/back/runtime/ports/legacyShadowStartupRecovery.ts @@ -0,0 +1,71 @@ +import type { + ExecutionOrigin, + RunAttemptStatus, + RunStatus, +} from '../domain/run'; + +export const MAX_LEGACY_SHADOW_STARTUP_BATCH_SIZE = 64; +export const MAX_LEGACY_SHADOW_STARTUP_EVIDENCE = 8; + +export interface LegacyShadowStartupCursor { + createdAtMs: number; + runId: string; +} + +export interface LegacyShadowStartupAttempt { + attemptId: string; + status: RunAttemptStatus; + pid?: number; + logArtifactId?: string; + createdAtMs: number; + startedAtMs?: number; +} + +export interface LegacyShadowStartupCandidate { + runId: string; + legacyCronId?: number; + origin: ExecutionOrigin; + runStatus: RunStatus; + createdAtMs: number; + activeAttemptCount: number; + attempt?: LegacyShadowStartupAttempt; +} + +export interface LegacyShadowStartupPage { + candidates: readonly LegacyShadowStartupCandidate[]; + truncated: boolean; + nextCursor?: LegacyShadowStartupCursor; +} + +export type LegacyRunningInstanceOutcome = + | 'running' + | 'succeeded' + | 'failed' + | 'stopped'; + +export interface LegacyRunningInstanceEvidence { + pid?: number; + logArtifactId?: string; + startedAtMs: number; + finishedAtMs?: number; + outcome: LegacyRunningInstanceOutcome; + exitCode?: number; +} + +export interface LegacyRunningInstanceEvidencePage { + evidence: readonly LegacyRunningInstanceEvidence[]; + truncated: boolean; +} + +export interface LegacyShadowStartupRecoverySource { + listCandidates(options: { + origins: readonly ExecutionOrigin[]; + cursor?: LegacyShadowStartupCursor; + limit?: number; + }): Promise; + + listRunningInstanceEvidence(options: { + legacyCronId: number; + limit?: number; + }): Promise; +} diff --git a/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md b/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md index 078d5231..e8247136 100644 --- a/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md +++ b/docs/QINGLONG_3_0_ARCHITECTURE_RFC.md @@ -11,6 +11,23 @@ 最新增量证据（2026-08-18）： +- D-356/ADR-0448（已接受）：补齐监听前、一次性且 Profile-aware 的 Legacy Shadow Startup Reconciler。它只在至少一个已审 origin 显式 + Shadow 时，于 Legacy `initData` 归一之后、manual Primary activation/HTTP listen 之前加载；默认 off 与 cluster-control/worker 均保持零 + Repository/恢复查询/写入。只读 Source 使用 `(created_at_ms, run_id)` keyset，单页硬上限 64、每 Cron RunningInstance 证据硬上限 8，原始 + log path 在 adapter 内先转为 opaque artifact ID。唯一终态证据补齐 success/failure/cancel；Node worker owner 重启后的 dispatching/running + 收敛 lost，queued/claimed 收敛 abandoned cancellation；system crond 无终态证据保持 pending，等待稳定 execution ID callback。Attempt 已提交而 + Run response loss 时，下次启动从唯一 terminal Attempt 修复 Run。edge 预算固定 `8 × 1 page`，standalone 为 `32 × 4 pages`，无 + timer/watcher/第二 SQLite authority；remaining/ambiguous/failed 留给后续差异报表、指标与 Primary gate。本切片不新增 package、生产依赖、 + schema、migration、进程、端口或部署对象。阶段门已重跑：启动恢复聚焦 `15/15`、Legacy/Shadow 扩展 `63/63`、Legacy + 身份专项 `8/8`、两者串行组合 `71/71`、完整 backend `1,434 pass / 0 fail / 2 conditional skip`、18-package clean + build/test、`build:back`、14/14 static audit 与 14/14 artifact audit 全部通过。edge/standalone 产物字节分别为：base + `2,589,998 / 2,590,076`、adopted `2,809,293 / 2,809,416`、application + `3,632,877 / 3,632,997`、application-api `3,800,430 / 3,800,574`、AI + `3,069,251 / 3,069,341`、application+AI `4,493,151 / 4,493,283`、MCP + `7,315,930 / 7,316,038`，与 D-355 完全一致。本切片只新增复用现有 SQLite authority 的只读 Legacy + Sequelize Source，不修改 schema/migration、数据库连接拓扑、容器或 Kubernetes 资源，因此不重跑物理 PostgreSQL HA/K3s 门； + 对象存储 backup/WAL/restore/PITR 与 cert-manager mTLS 轮换仍是既有发布最终化现场证据门，不因本阶段静态兼容通过而关闭。 + - D-355/ADR-0447（已接受）：完成 `once/boot/grpc` 的独立裁决，但不因枚举存在而虚构三条 owner 边界。`bootTask` 是唯一真实独立触发路径：只筛选 enabled `@boot`，固定以 `boot` origin 复用 `runSingle` 已创建的同一个 ChildProcess；显式 Shadow 时形成 legacy-owned terminal Run/Attempt/八 Event，默认关闭且失败开放。`@once` 当前只是被 system/node scheduler 排除的 schedule 标记，用户执行仍进入 manual；gRPC @@ -9082,7 +9099,7 @@ flowchart LR | PR-1 Run Schema | Incubating | Run/RunAttempt/RunEvent schema、nullable cancel request 与 Attempt deadline 字段及恢复索引、CancellationDispatch 状态/version/lease/backoff schema、Repository port、临时 Sequelize adapter、统一事件大小/分页上限、跨 adapter RunRepository contract suite（原子事务、回滚、Run/Attempt/RetryPolicy CAS、唯一错误、分页与取消恢复）；ADR-0041 的 `pg-0003-run-retry-policy`、capability v2、driver-neutral PostgreSQL Run Repository 与真实 `pg.Pool` 上的共享 Repository/rollback/SQLSTATE contract；ADR-0063/0069/0071/0073/0074/0076 的独立 Node 24 local-sqlite typed schema、十二条 reviewed migration、capability v6、共享 operation authority、readiness/RunRepository/API credential repository/receipt journal/dispatch plan/encrypted Secret envelope/Project Policy/security audit/authorized mutation/stable Identity catalog、Drizzle↔真实 catalog table/column/index/CHECK/FK lockstep、base/adopted/application edge/standalone 产物门禁；ADR-0064 的 legacy baseline/plan digest、Online Backup recovery、side-by-side target migration、staged manifest、双库栅栏 activation、source 生命周期写栅栏、target stable identity 和重启语义；ADR-0065 的独立 cutover authority、外部副作用停机 evidence、append-only journal、start/restart/stop barrier 与 unknown→manual_required 收敛；ADR-0066 的 adopted storage→Run reconciliation→receipt maintenance→domain recovery→lifecycle→admission application gate、严格有界 recovery summary 与 admission-first reverse stop；ADR-0067 的 SQLite 事实驱动 Run 候选源、256 条硬上限、截断失败关闭和唯一 Repository authority；ADR-0068 的 receipt-first Reconciler、callback token/sequence fence、exact local-process identity、Attempt/Run/双 Event 原子终态推进和最终 verifier；ADR-0069 的 local-process 单向包边界、pre-spawn journal、受审 POSIX launcher、immutable receipt、exact identity 和 Profile-aware cleanup lifecycle；ADR-0070 的独立 local-execution、spawn 前后双 transaction CAS、callback digest、exact stop 补偿与 fail-closed starting 保留；ADR-0071 的独立 local-dispatch、不可变 revision/context、Secret-first materializer、Profile Artifact admission、4/64 MiB output hard quota 和窄 application facade；ADR-0073/0074 的 Project-bound SecretRef、AES-256-GCM、外置 keyring 生命周期、双 SQLite authority CAS、application preflight、强 Principal/Policy 和 envelope+audit 原子提交；ADR-0086 的本机 Owner provisioning/challenge/claim/delivery acknowledgement/credential recovery CLI；ADR-0377 的 Local/Cluster 同构、Profile-aware、Project-scoped Artifact range read | fresh database/pepper setup、credential rotation/GC 运维编排与 Secret/Project/Role/Approval 管理 CLI/API/UI、备份/rekey、2.x/target process controller、人工 recovery、target 写后 reconciliation 与完整 cutover/rollback 演练；retry 产品策略、Artifact retention/tombstone stack、具体本机 lifecycle 和 target executable；Linux x64/arm64、PID namespace、断电与固定路由设备门禁；PostgreSQL 16/18 双连接并发与 failover integration；Task revision/context 跨方言 contract/并发压力与引用感知 retention、Keyv 数据迁移 | | PR-2 Run 状态机 | Incubating | 纯转换表、终态/时间/错误/执行器元数据规则、Run version 与 event sequence CAS、事务性 RunCommandService、回滚测试 | 重复 Worker callback/fencing、并发数据库压力测试、Primary 执行链接入 | | PR-3 Executor 端口 | Incubating | ADR-0003、ExecutionSpec/Context/Handle/Result、Executor port、LocalProcessExecutor、进程组取消/超时升级、流式背压、Legacy Cron spec builder、真实进程 contract tests、可复现 edge 基准入口 | 固定 edge/多架构设备基线、Legacy builder 与 makeCommand 差异审计、Primary 生产流量接入 | -| PR-4 Shadow Run | Incubating | origin 三态策略；默认关闭的 `QL3_SHADOW_ORIGINS`；manual、scheduled_node、boot、subscription、system 与 script 现有 ChildProcess 旁路观察；system crond 显式 origin marker、Shell execution ID、finish-only 准入、确定性 Run/Attempt 与 exact replay；`@once` 保持 manual、gRPC transport 不冒充 origin 的准入裁决；每个 worker 懒加载；Run/Attempt/Event 影子生命周期；稳定且不复制 caller 原文的 task identity/revision 与有界日志引用；同 worker 有界注册表和跨 worker 持久化候选关联；stop all/stop instance、Shell callback、乱序/迟到/歧义处理；失败开放和契约测试 | 启动后 Reconciler、差异报表、可采集指标、资源压力、回滚演练和 Primary 门禁 | +| PR-4 Shadow Run | Incubating | origin 三态策略；默认关闭的 `QL3_SHADOW_ORIGINS`；manual、scheduled_node、boot、subscription、system 与 script 现有 ChildProcess 旁路观察；system crond 显式 origin marker、Shell execution ID、finish-only 准入、确定性 Run/Attempt 与 exact replay；`@once` 保持 manual、gRPC transport 不冒充 origin 的准入裁决；每个 worker 懒加载；Run/Attempt/Event 影子生命周期；稳定且不复制 caller 原文的 task identity/revision 与有界日志引用；同 worker 有界注册表和跨 worker 持久化候选关联；stop all/stop instance、Shell callback、乱序/迟到/歧义处理；监听前一次性、Profile-aware 的 keyset Startup Reconciler，终态证据补齐、lost/abandoned/pending 分流与 terminal Attempt response-loss 修复；失败开放和契约测试 | 差异报表、可采集指标、资源压力、回滚演练和 Primary 门禁 | | PR-5 Primary LocalExecutor | Incubating（默认不激活，仅 manifest-gated manual） | runtime-owned Run 创建器；持久化先于 spawn；Run/Attempt 完整成功、失败、取消、超时与 lost 闭环；Executor handle 身份校验；spawn 后激活写失败的 stop+lost 补偿；completion rejection 安全收敛；独立 Primary 幂等查询与唯一索引竞态裁决；durable `run.cancel_requested`、stop-before-signal、首次请求幂等、晚到完成裁决与待取消有界恢复查询；最多 64 条一页的 cross-worker cancellation source；独立 CancellationDispatch Repository 原子 claim/result、lease expiry 接管、owner/token/version fencing、指数退避与结果 RunEvent；最多 64 页的单周期 cancel supervisor；显式 start/stop、无重叠、错误隔离、停止等待有上限且 timer unref 的 lifecycle runner；Linux durable handle 的 PID/boot/start ticks/process-group 复验与 TERM/KILL controller；完整有界分页且 fail-closed 的 startup Reconcile supervisor；RunningInstance nullable `run_id/attempt_id` 关联；Primary 专用组合 Repository 在同一 SQLite 事务提交前投影 Crontab/RunningInstance，失败整体回滚；有界且防穿越的 legacy log output ref；manual owner seam、真实本机装配、单 spawn/fail-closed；严格 manual-only rollout manifest loader、短期审批/gate、配置哈希审计；HTTP worker 已接轻量 lazy bootstrap，accepted 后按 receipt-first reconcile→completion receipt lifecycle→timeout intent lifecycle→cancel dispatch lifecycle→router 顺序激活，失败撤销，监听失败和 shutdown 有界停止；Primary timeout 在 spawn 前持久化绝对 deadline，有界 source/requester/supervisor 只提交 timeout 意图并复用 CancellationDispatch；代码级 edge/standalone Profile 为各 lifecycle 提供不同 cadence 与页上限，cluster-control/worker 拒绝误装本机 SQLite Primary；统一 CompletionService 原子提交 Attempt/Run/双 Event，spawn 前保存 callback token hash、终态推进 sequence，实时回调与 receipt consumer 共享入口并覆盖两个清理 crash window；manual Primary 已接入受限 POSIX launcher、`0600` direct-file stdout/stderr、父进程退出后续写、不可覆盖 receipt 生产、回执环境清除、TERM 转发等待及 live transaction 后清理；Startup Reconciler receipt-first 双检查并在确定 exited 后执行 profile 化的单次 50/100 ms publish grace；`0007` 独立 CompletionReceiptJournal 在 spawn 前登记、为升级前 active Attempt 补登记并驱动周期扫描，使终态残留继续可发现；确定无效的已知 Attempt receipt 先持久化隔离状态，再进入确定性私有分片 quarantine；终态 missing 与 quarantine 按 edge/standalone retention 有界清理；非 Journal 文件具备只读优先、固定分片/条目上限、overflow fail-closed、显式同盘隔离的 Node 24 运维 CLI；扫描具备页上限、resume cursor、timer unref、无重叠、有界 stop 和低敏计数；ENOSPC 与 launcher receipt 存储失败有代码门禁 | 部署配置写入/审批入口与用户可见状态；PostgreSQL CancellationDispatch adapter；cluster-control 生产启动拓扑；固定 edge/Linux 多架构与真实磁盘压力基线、完整 2.x API 契约和回滚演练 | | PR-7 Worker Session、Run Lease 与启动协议基础 | Incubating（默认关闭，独立入口显式 opt-in） | ADR-0012/0013/0014/0021/0057–0061/0108–0121/0231–0239/0377；有界 capability/Placement/Dispatcher；SQLite 协议孵化与 PostgreSQL v9 Session/Run Lease/credential/attestation authority；immutable revision Placement、数据库时钟 keyset candidate、认证 Worker Pull、digest-only offer recovery；versioned capability-free ExecutionSpec response、stable claim 跨重启退避、单 owner 原子 inbox 准入与 TLS 1.3 mTLS/`ql3w` HTTPS client；同一 package journal 上 revision-fenced starting/spawn/started/running/completion 状态、callback digest、tagged no-spawn 与 ambiguous recovery；PostgreSQL starting/running/start-failure/completion 数据库权威事务、精确重放与 cancellation/timeout 优先终态；batch Secret delivery 在 Attempt advisory lock 下复验 Session/Lease/revision 完整围栏并复用单 Agent，Secret-before-Artifact materializer 将同一 log ID 交给 Executor/journal/running ACK；offer-scoped `wlog-*` 私有文件 spool、Edge/Node 容量策略、append/quota/path 防护、barrier 后 output ownership、受审 POSIX Executor、truncation fact、固定内存流式 source、认证 Artifact stream、共享 immutable store port、S3-compatible SSE/checksum/条件 promotion adapter、upload-before-completion 协调，以及 Local/Cluster 同构、Profile-aware、ETag-fenced range read；用户取消 run.stop mutation 以数据库时间写 intent/Event 并在事务内复验 Project/RoleBinding fence；非执行取消 convergence lifecycle、运行期 expiry 与安全 lost retry 已接入 cluster-control 单一全局 cadence；完整 generation/version/token/Attempt fencing；独立最小权限 Worker ingress、CA/CRL 与连接 generation 热重载；offer journal、spawn barrier、receipt-first recovery；独立 `@qinglong/worker-runtime` 的本地 P-256 CSR、key/chain/trust 验证、generation + active pointer 安装和持久退避；默认关闭的 production process 已装配具体 execution graph、完整 Session heartbeat/drain/offline、direct-file bootstrap、单 Agent/单 cadence、startup reconciliation、证书 maintenance、transport fail-close/recovery 与 Edge/Node 有界预算；真实 PostgreSQL 18 + Linux Node 合约已覆盖 Run completion、credential 和 CA 双轮换且保持同一 Session；真实 K3s 合约已覆盖 TLS/credential Secret 分权、双对象 CAS、Recreate 顺序、identity generation 与单节点 PVC recovery；所有能力默认不可达且受 edge/cluster import audit 约束 | 具体 cert-manager/Vault/SPIFFE/离线 CA adapter 与模板、ingress reload controller、生产 RBAC、证书到期告警和 `ql3w` credential recovery 产品面；具体 KMS/Vault Secret provider、对象存储 credential/temporary lifecycle 与 retention/tombstone；Worker 管理 API；真实 Kubernetes 多节点 CSI/node-loss/production 360 秒 drain 与固定 edge 文件系统 suspend/时钟/断电、x64/arm64 资源门禁 | | PR-8 Project/Policy/Approval Core | Incubating（默认拒绝、无生产业务执行入口） | ADR-0028；统一六类 ActorRef 与 exact-shape 校验；`0017` ownerless default Project 和 append-only versioned RoleBinding；owner/admin/operator/viewer 固定矩阵；Project 内 mutation 幂等、expected-version CAS、双 SQLite 连接竞争门禁；archived read-only、revocation、存储损坏 fail-closed；Agent 写/Secret/Tool `require_approval`；ADR-0047 把六类 subject、role/permission matrix 与 fence 抽到 runtime-core，`pg-0004-project-policy`/capability v3 建立 ownerless PostgreSQL baseline、严格 role/state CHECK、append-only runtime 权限、SERIALIZABLE Project lock、mutation replay、双连接单 winner 和 cluster admission authorizer；ADR-0049/`pg-0005` capability v4 建立 stable IdentitySubject、append-only digest-only API credential、真实 cluster bearer authenticator、write-only durable security audit 与最小权限 runtime role，且已验证 HTTP→credential→Policy→audit→handler 纵向链路；ADR-0051 建立 `/api/v3` 认证前 peer/global 双预算、transport-peer-only、无 timer 且有界内存的 overload shield；ADR-0027 Artifact authorizer adapter；ADR-0029 `AuthenticatedPrincipal` contract、`0018` digest-only versioned challenge、CSPRNG/TTL、同事务消费 challenge + 写首 owner、精确重放与双连接竞争/崩溃回滚门禁；ADR-0030 `0019` stable identity/binding、legacy HS384 + current-session membership、logout/platform/revoke/disable、single-factor 与损坏 fail-closed 门禁；ADR-0031 `0020` digest-bound ApprovalRequest、User-only decision、Project/Role version fence、精确 expiry/重放/并发裁决及同事务 immutable dispatch；ADR-0032 `0021` execution backfill、三表原子 consume、稳定 due keyset、claim/renew/start/result fencing、pre-start takeover/post-start recovery-required、attempt budget、handler inspect/digest barrier 和 bounded dispatcher；ADR-0033/`0022` control/resolution backfill、start/renew/completion 原子联动、稳定 recovery keyset、双 resolver claim/takeover、finding/result 精确重放、自动/人工终结、迟到 completion 单 winner 和 evidence-only bounded reconciler；ADR-0034/`0023` 首个 `run.create` canonical plan、Run/Attempt/Event/receipt 同事务、幂等 collision fail-closed、renew/终态 fence、真实 SQLite handler 与 automatic evidence provider；ADR-0035/`0024` 独立 `approval.recover` 矩阵、稳定 User + 五分钟强认证、Project/RoleBinding fence、human resolution + authorization fact 原子提交、撤权竞态与回滚门禁；ADR-0036 recovery-first 单 timer lifecycle、edge/standalone 独立 cadence/页预算、跨周期 cursor、非重叠与有界 stop；ADR-0074 以新的 Node 24 SQLite v5 ownerless Project/RoleBinding/audit authority 和独立 local-secret-admin 提供强 Principal、`secret.manage`、撤权 fence、envelope+allowed audit 原子提交及不回显语义；ADR-0086 以可信 POSIX console 和 staged delivery 完成本机首 Owner 产品 ceremony | fresh database/pepper setup 与安全迁移向导；`shareStore`/Express 到 authentication core 的 production migration；credential rotation/revocation API、mTLS/Worker enrollment、恢复码；Project/Role/Approval/Secret 管理 CLI/API/UI、audit retention/query/export/alert、preview Artifact/digest/immutable plan builder、真实 MFA/hardware adapter、人工 recovery API/UI/独立 rate limit 与审计事件、handler/provider registry、lifecycle startup/shutdown/指标/admission gate；PostgreSQL action/receipt/provider/recovery-authorization 与 OPA adapter、缓存 version 失效；Tool/Package/Secret/Shell 各自的 handler/evidence contract；Secret/Run/Tool/Workflow waiting_approval 全入口装配；完整回滚演练 | diff --git a/docs/adr/ADR-0002-legacy-crontab-compatibility-and-shadow-write.md b/docs/adr/ADR-0002-legacy-crontab-compatibility-and-shadow-write.md index 493c0250..345cd01f 100644 --- a/docs/adr/ADR-0002-legacy-crontab-compatibility-and-shadow-write.md +++ b/docs/adr/ADR-0002-legacy-crontab-compatibility-and-shadow-write.md @@ -5,7 +5,7 @@ - 决策者：QingLong Maintainers - 关联 RFC：[QL-RFC-0001](../QINGLONG_3_0_ARCHITECTURE_RFC.md) - 前置决策：[ADR-0001](./ADR-0001-run-state-and-transaction-boundaries.md) -- Amended by：[ADR-0445](./ADR-0445-schedule-service-origin-shadow-run-coverage.md)、[ADR-0446](./ADR-0446-system-crond-stable-shadow-admission.md)、[ADR-0447](./ADR-0447-boot-shadow-and-non-origin-boundaries.md) +- Amended by：[ADR-0445](./ADR-0445-schedule-service-origin-shadow-run-coverage.md)、[ADR-0446](./ADR-0446-system-crond-stable-shadow-admission.md)、[ADR-0447](./ADR-0447-boot-shadow-and-non-origin-boundaries.md)、[ADR-0448](./ADR-0448-bounded-legacy-shadow-startup-reconciliation.md) ## 1. 决策摘要 @@ -239,7 +239,7 @@ Shadow Adapter 不得： Shadow 转换仍必须遵守 ADR-0001。无法合法映射时追加 compat.transition_mismatch，不能强行覆盖终态。 -当前 Alpha 切片对 `manual` 与 `scheduled_node` 直接观察 Node ChildProcess 的 spawn、error 和 exit 事件，因此不依赖 Shell callback 才能形成基本终态。下述两级关联已补充 Shell callback、stop/cancel 和乱序/迟到回调，但启动后协调与差异对账仍是进入 Primary 前的门禁，不能由进程内观察能力替代。 +当前 Alpha 切片对已审 Node worker origin 直接观察同一 ChildProcess 的 spawn、error 和 exit 事件，因此不依赖 Shell callback 才能形成基本终态。下述两级关联已补充 Shell callback、stop/cancel 和乱序/迟到回调；ADR-0448 又补充了监听前一次性启动恢复。差异对账、可采集指标与 Primary gate 仍不能由进程内观察或启动 summary 替代。 ### 9.4 `next` Alpha callback 与 stop 关联 @@ -253,7 +253,7 @@ Shadow 转换仍必须遵守 ADR-0001。无法合法映射时追加 compat.trans 6. 取消事实在 Legacy kill 前投递；同 worker 的后续 exit 排在取消之后。跨 worker 使用持久化定位器尽力关联，任何查询或写入失败都不能阻断 kill 或改变 2.x API 响应。 7. 乱序 finished 可以从 queued/claimed 补齐 dispatching、starting、running 和终态；重复终态 callback、取消后的迟到成功 callback 不覆盖终态，也不追加重复完成事件。 -这仍不是完整 Reconciler：当前没有启动后批量扫描、缺失事实修复、公开指标采集器或 Shadow/Legacy 差异报表。上述能力必须在 Primary 门禁前补齐，但 edge 默认不得因此增加常驻 watcher 或无界内存队列。 +这仍不是完整的 Shadow→Primary 门禁：ADR-0448 已提供启动后有界批量扫描、终态证据补齐、lost/abandoned 收敛和两事务 response-loss 修复；当前仍没有公开指标采集器、Shadow/Legacy 差异报表与正式 Primary gate。后续能力不得让 edge 增加常驻 watcher 或无界内存队列。 ### 9.3 Shadow 写失败 @@ -445,11 +445,13 @@ Run 使用创建时的 task revision/snapshot。更新 Crontab 只影响后续 R ### 15.1 off/shadow -Legacy 启动行为暂时保持。Shadow Reconciler 在有界批次内扫描非终态 Shadow Run： +Legacy 启动归一行为保持。HTTP worker 在归一之后、Primary activation 与 listen 之前运行一次 Shadow Reconciler： -- 根据唯一 RunningInstance/PID/log path 证据更新。 -- 无法确认时标记 Attempt lost 或写 mismatch，不把 Legacy 任务强制改状态。 -- edge Profile 默认低频、有限批次运行，禁止全表高频扫描。 +- 使用 `(created_at_ms, run_id)` keyset，只扫描 enabled origin 的 queued/dispatching/running legacy-owned Run；单页最多 64。 +- 每个 Cron 最多读取 8 条 RunningInstance，只有唯一 PID/log/实例终态证据才补齐 succeeded/failed/cancelled；冲突与截断保持 ambiguous。 +- Node worker-owned dispatching/running 在 owner 重启且无终态证据时标记 lost；queued/claimed 收敛为 abandoned cancellation。 +- scheduled_system 无终态证据时保持 pending，等待稳定 execution ID callback，不把 HTTP worker 生命周期误当成 system crond 生命周期。 +- edge 每次启动最多 `8 × 1 page`，standalone 最多 `32 × 4 pages`，不启动 timer/watcher；cluster-control/worker 拒绝本机 SQLite 装配。 ### 15.2 primary diff --git a/docs/adr/ADR-0448-bounded-legacy-shadow-startup-reconciliation.md b/docs/adr/ADR-0448-bounded-legacy-shadow-startup-reconciliation.md new file mode 100644 index 00000000..bf32b44e --- /dev/null +++ b/docs/adr/ADR-0448-bounded-legacy-shadow-startup-reconciliation.md @@ -0,0 +1,88 @@ +# ADR-0448：有界 Legacy Shadow 启动恢复 + +- 状态：Accepted +- 日期：2026-08-18 +- 关联 RFC：QL-RFC-0001 D-02、D-356、PR-4 +- 关联 ADR：ADR-0001、ADR-0002、ADR-0445、ADR-0446、ADR-0447 +- Amends：ADR-0002 的 Shadow 启动与恢复边界 + +## 上下文 + +Legacy Shadow 已能旁路观察同一 ChildProcess、关联跨 worker callback/stop，并为 system crond 提供 response-loss-safe execution ID；但 HTTP worker +重启后，原 worker 的内存注册表和 ChildProcess listener 都已消失。数据库可能留下 queued/dispatching/running 的 legacy-owned Run，而 Legacy +`initData` 又会在 HTTP 启动期间先把旧 RunningInstance 从 running 归一为 stopped。 + +因此不能照搬 Primary Reconciler：Legacy Attempt 没有可复验的 durable Executor handle，PID 可能重用，system crond 还可能独立于 HTTP worker +继续运行。仅用 PID 存活检查会伪造 owner 连续性；把所有非终态 Run 一律 lost，又会提前终结仍可通过稳定 callback 收敛的 system crond。 + +部署范围同时覆盖低性能路由设备、standalone 和 cluster 节点。启动恢复不得引入常驻 watcher、第二个 SQLite authority 或对 cluster profile 的本机 +数据库误装配。 + +## 决策 + +1. 只有 `QL3_SHADOW_ORIGINS` 至少显式启用一个已审 origin 时，HTTP worker 才加载恢复 Source、Run Repository、Writer 和 Reconciler。默认 off + 不导入重型 adapter、不查询数据库，也不创建 timer/listener。 +2. 恢复发生在 Legacy `initData` 完成 RunningInstance 状态归一之后、manual Primary activation 与 HTTP listen 之前。这样启动期间没有新的 HTTP + manual admission 与恢复扫描竞争，Primary 也不会先于 Shadow 遗留事实审计激活。 +3. 新的只读 Source 使用 `(created_at_ms, run_id)` keyset 分页，只扫描 enabled origin、legacy owner 且状态为 queued/dispatching/running 的 + Run。单页硬上限 64；每个 Cron 的 RunningInstance 证据硬上限 8，超限、重复身份或冲突一律 ambiguous，不猜测更新。 +4. Source 只把 RunningInstance 的 log path 转换为现有 36 字符 opaque log artifact ID；原始路径、command、用户名、Run ID 和 Cron ID 不进入 + audit message 或指标 label。 +5. 状态裁决如下： + - 唯一 PID/log/唯一实例证据已经 finished/error：复用 Shadow Writer 补齐 succeeded/failed； + - 唯一 stopped 且带 finished time：补齐 cancelled，reason 为 reconcile； + - queued + claimed 且没有 spawn 证据：以 reconciler actor 收敛为 cancelled，并记录 acceptance abandoned； + - dispatching/running 的 Node worker-owned origin 在重启后没有终态证据：Attempt/Run 收敛为 lost； + - scheduled_system 没有终态证据：保持 pending，等待稳定 execution ID callback，不能因 HTTP worker 重启提前终结外部 crond； + - 多 Attempt、证据截断、身份冲突或非法状态：只计 ambiguous/failed，不覆盖 Legacy UI 状态。 +6. lost/abandoned 的 Attempt 与 Run 使用两个既有原子命令事务推进。若 Attempt 已提交而 Run 响应丢失或进程退出，下次启动会从唯一 terminal Attempt + 修复 active Run；不产生第二个 Attempt，也不重放外部副作用。 +7. Reconciler 每次 HTTP 启动只运行一次：edge 为 `8 × 1 page`，standalone 为 `32 × 4 pages`。页预算耗尽返回稳定 resume cursor 和 + `remaining=true`，不在进程内排队、不自动循环；后续差异报表和正式 Primary gate 必须把 remaining/ambiguous/failed 视为未闭合证据。 +8. `cluster-control` 与 `worker` profile 拒绝本机 Legacy Shadow 恢复装配。它们未来必须使用 PostgreSQL/shared authority 的独立 Reconciler，不能复用 + Legacy SQLite。 +9. Source、写入、配置或 audit sink 失败均保持 Shadow fail-open，只输出低敏 error type/有界 summary，不阻止 2.x HTTP 服务启动。 + +## 资源与部署影响 + +- 不新增 package、生产依赖、schema、migration、表、索引、进程、线程、端口、timer、watcher 或 Kubernetes 对象。 +- edge 每次启动最多扫描 8 个 Run，每个带 Cron identity 的候选最多读取 8 条 RunningInstance；standalone 最多扫描 128 个 Run。 +- 查询使用现有 Runs/RunAttempts 索引与 keyset，不用 OFFSET、不全表加载 Attempt，也不把原始日志路径带出 adapter。 +- 默认 off 和 cluster/worker profile 都是零 Repository、零恢复查询、零写入；只有显式 Shadow 的本机 profile 承担一次性启动成本。 + +## 被拒绝的替代方案 + +### 复用 Primary Startup Reconciler + +拒绝。Primary 依赖 durable Executor handle、receipt journal 和 runtime owner；Legacy Shadow 不具备这些证据，复用会把 PID 猜测伪装成精确身份。 + +### 启动后定时全表扫描 + +拒绝。它会给路由设备增加常驻 timer、重复数据库唤醒和不可控历史扫描，也与当前一次性启动门边界不符。 + +### 所有旧 active Run 一律 lost + +拒绝。system crond 独立于 HTTP worker，并可用稳定 execution ID 在重启后补发终态 callback；提前 lost 会丢弃更强事实。 + +### 在 Source 中复用原始 log path 作为跨层 identity + +拒绝。原始路径可能暴露任务结构或用户信息；adapter 内必须先转换为现有 opaque artifact ID。 + +### 在 cluster profile 打开同一 SQLite 恢复器 + +拒绝。多节点会形成多个本机 authority，既无法看到共享事实，也可能产生冲突终态。 + +## 验证 + +- 真实 SQLite 覆盖唯一 RunningInstance 成功终态、startup-reset lost、spawn 前 abandoned、显式 stopped、system-crond pending、重复身份拒绝、 + terminal Attempt response-loss 修复和稳定 keyset 分页。 +- Bootstrap 合同覆盖默认关闭零 execute、cluster/worker 拒绝、edge/standalone 独立预算、低敏失败开放，以及 + `Legacy normalization → Shadow recovery → Primary activation → HTTP listen` 顺序。 +- 聚焦测试 `15/15`；Legacy/Shadow 扩展 `63/63`，Legacy 身份专项 `8/8`，两者串行组合 `71/71`；完整 backend + `1,434 pass / 0 fail / 2 conditional skip`；18 个 QL3 package 均完成 clean build/test，`build:back` 通过。 +- 14/14 static audit 与 14/14 artifact audit 通过。edge/standalone 的 base、adopted、application、application-api、AI、 + application+AI、MCP 产物分别为 `2,589,998 / 2,590,076`、`2,809,293 / 2,809,416`、 + `3,632,877 / 3,632,997`、`3,800,430 / 3,800,574`、`3,069,251 / 3,069,341`、 + `4,493,151 / 4,493,283`、`7,315,930 / 7,316,038` bytes，与前一阶段一致。 +- 本切片不改 schema/migration、数据库连接拓扑、容器或 Kubernetes 资源，未重跑物理 PostgreSQL HA/K3s 门；对象存储 + backup/WAL/restore/PITR 与 cert-manager mTLS 轮换继续保留为发布最终化现场证据门。 diff --git a/docs/adr/README.md b/docs/adr/README.md index 5d49f280..dfdeabe8 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -451,6 +451,7 @@ | [ADR-0445](./ADR-0445-schedule-service-origin-shadow-run-coverage.md) | ScheduleService 执行来源的 Shadow Run 覆盖 | Accepted（`scheduled_system` 后续由 ADR-0446 完成） | | [ADR-0446](./ADR-0446-system-crond-stable-shadow-admission.md) | System Crond 稳定 Shadow 准入与回调重放 | Accepted | | [ADR-0447](./ADR-0447-boot-shadow-and-non-origin-boundaries.md) | Boot Shadow 准入与 once/gRPC 非 Origin 边界 | Accepted | +| [ADR-0448](./ADR-0448-bounded-legacy-shadow-startup-reconciliation.md) | 有界 Legacy Shadow 启动恢复 | Accepted | ## 规则 diff --git a/test/back/bootstrapLegacyShadowStartupReconciliation.test.cjs b/test/back/bootstrapLegacyShadowStartupReconciliation.test.cjs new file mode 100644 index 00000000..7ae8d849 --- /dev/null +++ b/test/back/bootstrapLegacyShadowStartupReconciliation.test.cjs @@ -0,0 +1,149 @@ +require('ts-node/register/transpile-only'); + +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const path = require('node:path'); +const { test } = require('node:test'); +const { + bootstrapLegacyShadowStartupReconciliation, +} = require('../../back/runtime/adapters/legacy/bootstrapLegacyShadowStartupReconciliation'); + +function summary(overrides = {}) { + return { + pages: 1, + scanned: 0, + completed: 0, + cancelled: 0, + abandoned: 0, + markedLost: 0, + repaired: 0, + pending: 0, + ambiguous: 0, + skipped: 0, + failed: 0, + stopReason: 'complete', + remaining: false, + ...overrides, + }; +} + +test('disabled bootstrap remains fully lazy', async () => { + const calls = []; + const result = await bootstrapLegacyShadowStartupReconciliation({ + origins: [], + profile: 'edge', + async execute() { + calls.push('execute'); + throw new Error('must remain lazy'); + }, + audit(record) { + calls.push(record.state); + }, + }); + + assert.deepEqual(result, { state: 'disabled' }); + assert.deepEqual(calls, ['disabled']); +}); + +test('cluster profiles reject local Shadow recovery without loading storage', async () => { + for (const profile of ['cluster-control', 'worker']) { + const result = await bootstrapLegacyShadowStartupReconciliation({ + origins: ['manual'], + profile, + async execute() { + throw new Error('cluster profile must not load local storage'); + }, + audit() {}, + }); + assert.deepEqual(result, { state: 'profile_rejected', profile }); + } +}); + +test('applies distinct one-shot edge and standalone budgets', async () => { + const requests = []; + for (const profile of ['edge', 'standalone']) { + const result = await bootstrapLegacyShadowStartupReconciliation({ + origins: ['manual', 'scheduled_system'], + profile, + async execute(request) { + requests.push(request); + return summary(); + }, + audit() {}, + }); + assert.equal(result.state, 'reconciled'); + } + + assert.deepEqual( + requests.map(({ profile, pageSize, maxPages }) => ({ + profile, + pageSize, + maxPages, + })), + [ + { profile: 'edge', pageSize: 8, maxPages: 1 }, + { profile: 'standalone', pageSize: 32, maxPages: 4 }, + ], + ); +}); + +test('fails open with a low-sensitivity error type', async () => { + const result = await bootstrapLegacyShadowStartupReconciliation({ + origins: ['manual'], + profile: 'edge', + async execute() { + throw new RangeError('secret command must not be reported'); + }, + audit() {}, + }); + + assert.deepEqual(result, { state: 'failed', errorType: 'RangeError' }); + assert.equal(JSON.stringify(result).includes('secret command'), false); +}); + +test('redacts the resume cursor Run identity from startup audit output', async () => { + const records = []; + const result = await bootstrapLegacyShadowStartupReconciliation({ + origins: ['manual'], + profile: 'edge', + async execute() { + return summary({ + stopReason: 'page_limit', + remaining: true, + nextCursor: { createdAtMs: 10, runId: 'run-secret-identity' }, + }); + }, + audit(record) { + records.push(JSON.stringify(record)); + }, + }); + + assert.equal(result.state, 'incomplete'); + assert.equal(result.summary.resumeAvailable, true); + assert.equal('nextCursor' in result.summary, false); + assert.equal(records.join('').includes('run-secret-identity'), false); +}); + +test('HTTP startup orders Shadow recovery after Legacy normalization and before Primary/listen', () => { + const source = fs.readFileSync( + path.resolve(__dirname, '../../back/app.ts'), + 'utf8', + ); + const legacyNormalization = source.indexOf( + 'await appLoader.default({ app: this.app })', + ); + const shadowRecovery = source.indexOf( + 'await bootstrapLegacyShadowStartupReconciliation()', + ); + const primaryActivation = source.indexOf( + 'await bootstrapDefaultManualPrimaryRuntime()', + ); + const listen = source.indexOf( + 'this.httpServerService.initialize(this.app, config.port)', + ); + + assert.equal(legacyNormalization >= 0, true); + assert.equal(legacyNormalization < shadowRecovery, true); + assert.equal(shadowRecovery < primaryActivation, true); + assert.equal(primaryActivation < listen, true); +}); diff --git a/test/back/legacyShadowStartupReconciliation.test.cjs b/test/back/legacyShadowStartupReconciliation.test.cjs new file mode 100644 index 00000000..581226b2 --- /dev/null +++ b/test/back/legacyShadowStartupReconciliation.test.cjs @@ -0,0 +1,368 @@ +require('ts-node/register/transpile-only'); + +const assert = require('node:assert/strict'); +const { afterEach, test } = require('node:test'); +const { DataTypes, Sequelize } = require('sequelize'); +const { + defineSchemaMigrationModel, +} = require('../../back/data/schemaMigration'); +const { runSchemaMigration } = require('../../back/migrations/0002-run-schema'); +const { + runCancellationRequestMigration, +} = require('../../back/migrations/0004-run-cancellation-request'); +const { + runAttemptDeadlineMigration, +} = require('../../back/migrations/0006-run-attempt-deadline'); +const { runMigrations } = require('../../back/migrations/runner'); +const { + LegacySequelizeRunRepository, +} = require('../../back/runtime/adapters/legacy-sequelize/runRepository'); +const { + LegacySequelizeShadowStartupRecoverySource, +} = require('../../back/runtime/adapters/legacy-sequelize/legacyShadowStartupRecoverySource'); +const { + LegacyShadowRunWriter, +} = require('../../back/runtime/application/legacyShadowRunWriter'); +const { + LegacyShadowStartupReconciler, + LegacyShadowStartupSupervisor, +} = require('../../back/runtime/application/legacyShadowStartupReconciler'); +const { + RunCommandService, +} = require('../../back/runtime/application/runCommandService'); +const { + createLegacyLogArtifactId, +} = require('../../back/runtime/compatibility/legacyTaskRevision'); + +const databases = []; +let idSequence = 1_000; +let timeSequence = 1_750_001_000_000; + +function nextId() { + idSequence += 1; + return `019f7200-0000-7000-8000-${String(idSequence).padStart(12, '0')}`; +} + +function nextTime() { + timeSequence += 100; + return timeSequence; +} + +async function createStack() { + const database = new Sequelize({ + dialect: 'sqlite', + storage: ':memory:', + logging: false, + }); + await runMigrations({ + database, + migrationModel: defineSchemaMigrationModel(database), + migrations: [ + runSchemaMigration, + runCancellationRequestMigration, + runAttemptDeadlineMigration, + ], + logger: { info() {} }, + }); + await database.getQueryInterface().createTable('RunningInstances', { + id: { + type: DataTypes.INTEGER, + primaryKey: true, + autoIncrement: true, + }, + cron_id: { type: DataTypes.INTEGER, allowNull: false }, + pid: { type: DataTypes.INTEGER, allowNull: true }, + log_path: { type: DataTypes.STRING, allowNull: true }, + started_at: { type: DataTypes.INTEGER, allowNull: false }, + finished_at: { type: DataTypes.INTEGER, allowNull: true }, + status: { type: DataTypes.INTEGER, allowNull: false }, + exit_code: { type: DataTypes.INTEGER, allowNull: true }, + }); + databases.push(database); + const repository = new LegacySequelizeRunRepository(database); + const writer = new LegacyShadowRunWriter(repository, nextId); + const source = new LegacySequelizeShadowStartupRecoverySource( + database, + createLegacyLogArtifactId, + ); + const reconciler = new LegacyShadowStartupReconciler( + repository, + source, + writer, + { clock: { now: () => nextTime() }, createEventId: nextId }, + ); + return { database, repository, writer, source, reconciler }; +} + +async function activeShadow(writer, overrides = {}) { + const acceptedAtMs = nextTime(); + const origin = overrides.origin ?? 'manual'; + const reference = await writer.accept({ + origin, + projectId: 'default', + taskId: `legacy-cron:${overrides.legacyCronId ?? 15}`, + taskRevision: 'sha256:startup-reconciliation', + legacyCronId: overrides.legacyCronId ?? 15, + triggerType: origin, + acceptedAtMs, + }); + if (overrides.queuedOnly) return { reference, acceptedAtMs }; + await writer.spawned(reference, { + atMs: acceptedAtMs + 1, + ...(overrides.pid === undefined ? {} : { pid: overrides.pid }), + ...(overrides.logPath === undefined + ? {} + : { logArtifactId: createLegacyLogArtifactId(overrides.logPath) }), + }); + await writer.running(reference, acceptedAtMs + 2); + return { reference, acceptedAtMs }; +} + +async function insertInstance(database, values) { + await database.getQueryInterface().bulkInsert('RunningInstances', [ + { + cron_id: 15, + pid: null, + log_path: null, + started_at: 1_750_001_000, + finished_at: null, + status: 0, + exit_code: null, + ...values, + }, + ]); +} + +afterEach(async () => { + await Promise.all(databases.splice(0).map((database) => database.close())); +}); + +test('completes an active Shadow Run from unique terminal RunningInstance evidence', async () => { + const { database, repository, writer, reconciler } = await createStack(); + const logPath = 'cron/2026-08-18-01.log'; + const { reference } = await activeShadow(writer, { + pid: 4101, + logPath, + }); + await insertInstance(database, { + pid: 4101, + log_path: logPath, + finished_at: 1_750_001_010, + status: 1, + exit_code: 0, + }); + + const summary = await reconciler.reconcileBatch({ origins: ['manual'] }); + + assert.equal(summary.completed, 1); + assert.equal(summary.failed, 0); + assert.equal( + (await repository.findRunById(reference.runId)).status, + 'succeeded', + ); + assert.equal( + (await repository.findAttemptById(reference.attemptId)).status, + 'succeeded', + ); +}); + +test('maps an explicitly finished stopped instance to reconciled cancellation', async () => { + const { database, repository, writer, reconciler } = await createStack(); + const { reference } = await activeShadow(writer, { pid: 4151 }); + await insertInstance(database, { + pid: 4151, + finished_at: 1_750_001_011, + status: 2, + exit_code: 143, + }); + + const summary = await reconciler.reconcileBatch({ origins: ['manual'] }); + + assert.equal(summary.cancelled, 1); + assert.equal( + (await repository.findRunById(reference.runId)).status, + 'cancelled', + ); + assert.equal( + (await repository.findAttemptById(reference.attemptId)).status, + 'cancelled', + ); +}); + +test('pages active Shadow candidates with a stable keyset cursor', async () => { + const { writer, source } = await createStack(); + for (const legacyCronId of [21, 22, 23]) { + await activeShadow(writer, { legacyCronId, queuedOnly: true }); + } + + const first = await source.listCandidates({ origins: ['manual'], limit: 2 }); + const second = await source.listCandidates({ + origins: ['manual'], + cursor: first.nextCursor, + limit: 2, + }); + + assert.equal(first.truncated, true); + assert.equal(first.candidates.length, 2); + assert.equal(second.truncated, false); + assert.equal(second.candidates.length, 1); + assert.equal( + new Set( + [...first.candidates, ...second.candidates].map(({ runId }) => runId), + ).size, + 3, + ); +}); + +test('marks a worker-owned execution lost when startup reset has no terminal evidence', async () => { + const { database, repository, writer, reconciler } = await createStack(); + const { reference } = await activeShadow(writer, { pid: 4201 }); + await insertInstance(database, { + pid: 4201, + status: 2, + finished_at: null, + }); + + const summary = await reconciler.reconcileBatch({ origins: ['manual'] }); + + assert.equal(summary.markedLost, 1); + assert.equal((await repository.findRunById(reference.runId)).status, 'lost'); + assert.equal( + (await repository.findAttemptById(reference.attemptId)).status, + 'lost', + ); + assert.equal( + (await repository.listEvents(reference.runId)).at(-1).actorType, + 'reconciler', + ); +}); + +test('abandons an accepted execution that never produced spawn evidence', async () => { + const { repository, writer, reconciler } = await createStack(); + const { reference } = await activeShadow(writer, { queuedOnly: true }); + + const summary = await reconciler.reconcileBatch({ origins: ['manual'] }); + + assert.equal(summary.abandoned, 1); + assert.equal( + (await repository.findRunById(reference.runId)).status, + 'cancelled', + ); + assert.equal( + (await repository.findAttemptById(reference.attemptId)).status, + 'cancelled', + ); +}); + +test('keeps system-crond pending without terminal callback evidence', async () => { + const { database, repository, writer, reconciler } = await createStack(); + const { reference } = await activeShadow(writer, { + origin: 'scheduled_system', + pid: 4301, + }); + await insertInstance(database, { + pid: 4301, + status: 2, + finished_at: null, + }); + + const summary = await reconciler.reconcileBatch({ + origins: ['scheduled_system'], + }); + + assert.equal(summary.pending, 1); + assert.equal( + (await repository.findRunById(reference.runId)).status, + 'running', + ); + assert.equal( + (await repository.findAttemptById(reference.attemptId)).status, + 'running', + ); +}); + +test('refuses ambiguous duplicate identity evidence instead of guessing a terminal result', async () => { + const { database, repository, writer, reconciler } = await createStack(); + const logPath = 'cron/ambiguous.log'; + const { reference } = await activeShadow(writer, { + pid: 4351, + logPath, + }); + for (const pid of [4352, 4353]) { + await insertInstance(database, { + pid, + log_path: logPath, + finished_at: 1_750_001_020, + status: 1, + exit_code: 0, + }); + } + + const summary = await reconciler.reconcileBatch({ origins: ['manual'] }); + + assert.equal(summary.ambiguous, 1); + assert.equal( + (await repository.findRunById(reference.runId)).status, + 'running', + ); + assert.equal( + (await repository.findAttemptById(reference.attemptId)).status, + 'running', + ); +}); + +test('repairs a Run left active after its lost Attempt transaction committed', async () => { + const { repository, writer, reconciler } = await createStack(); + const { reference } = await activeShadow(writer, { pid: 4401 }); + const run = await repository.findRunById(reference.runId); + const commands = new RunCommandService(repository, nextId); + await commands.transitionRunAttempt({ + runId: reference.runId, + attemptId: reference.attemptId, + to: 'lost', + expectedRunVersion: run.version, + atMs: nextTime(), + errorCode: 'LEGACY_RECONCILE_OWNER_LOST', + errorSummary: 'simulated response loss', + actor: { type: 'reconciler' }, + }); + + const summary = await reconciler.reconcileBatch({ origins: ['manual'] }); + + assert.equal(summary.repaired, 1); + assert.equal((await repository.findRunById(reference.runId)).status, 'lost'); +}); + +test('supervisor preserves a stable cursor when its Profile budget is exhausted', async () => { + const calls = []; + const supervisor = new LegacyShadowStartupSupervisor({ + async reconcileBatch(options) { + calls.push(options); + return { + scanned: 1, + completed: 0, + cancelled: 0, + abandoned: 0, + markedLost: 1, + repaired: 0, + pending: 0, + ambiguous: 0, + skipped: 0, + failed: 0, + truncated: true, + nextCursor: { createdAtMs: 10, runId: 'run-10' }, + }; + }, + }); + + const summary = await supervisor.run({ + origins: ['manual'], + pageSize: 8, + maxPages: 1, + }); + + assert.equal(summary.stopReason, 'page_limit'); + assert.equal(summary.remaining, true); + assert.deepEqual(summary.nextCursor, { createdAtMs: 10, runId: 'run-10' }); + assert.equal(calls[0].limit, 8); +});