simplify gate graph validation
This commit is contained in:
Tianyi Cui
@@ -1,29 +1,14 @@
|
||||
import { spawnSync } from 'node:child_process'
|
||||
import { mkdtempSync, rmSync, symlinkSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import {
|
||||
executeGatePlan,
|
||||
formatGatePlanJson,
|
||||
formatGatePlanList,
|
||||
defaultConcurrency,
|
||||
formatGateResultReason,
|
||||
gateDependencyClosure,
|
||||
gatePlanForMode,
|
||||
listedGatePlan,
|
||||
parseCliRequest,
|
||||
resolvePlanConcurrency,
|
||||
gatesForMode,
|
||||
runGate,
|
||||
validateGatePlan,
|
||||
runGates,
|
||||
type Gate,
|
||||
type GatePlan,
|
||||
type GateResult,
|
||||
} from './run-gates.ts'
|
||||
|
||||
const repositoryRoot = join(import.meta.dirname, '..')
|
||||
|
||||
afterEach(() => vi.unstubAllEnvs())
|
||||
|
||||
function gate(id: string, options: Partial<Gate> = {}): Gate {
|
||||
return {
|
||||
id,
|
||||
@@ -35,17 +20,11 @@ function gate(id: string, options: Partial<Gate> = {}): Gate {
|
||||
}
|
||||
}
|
||||
|
||||
function plan(gates: Gate[]): GatePlan {
|
||||
return { mode: 'check-all', script: 'check:all', gates }
|
||||
}
|
||||
|
||||
function resultFor(subject: Gate, status: GateResult['status'] = 'passed'): GateResult {
|
||||
return {
|
||||
gate: subject,
|
||||
status,
|
||||
durationMs: 10,
|
||||
stdout: '',
|
||||
stderr: '',
|
||||
output: [],
|
||||
exitCode: status === 'passed' ? 0 : 1,
|
||||
signalCode: null,
|
||||
@@ -63,7 +42,7 @@ function withPnpmEntrypoint<T>(action: () => T): T {
|
||||
}
|
||||
}
|
||||
|
||||
describe('gate plan validation', () => {
|
||||
describe('gate graph validation', () => {
|
||||
it.each([
|
||||
'ci-primary',
|
||||
'ci-static',
|
||||
@@ -78,282 +57,54 @@ describe('gate plan validation', () => {
|
||||
'node-compat',
|
||||
'check-all',
|
||||
'doc-sync',
|
||||
] as const)('constructs a valid non-empty %s plan', (mode) => {
|
||||
const subject = withPnpmEntrypoint(() => gatePlanForMode(mode))
|
||||
expect(() => {
|
||||
validateGatePlan(subject)
|
||||
}).not.toThrow()
|
||||
] as const)('constructs and executes preflight for a valid non-empty %s graph', async (mode) => {
|
||||
const subject = withPnpmEntrypoint(() => gatesForMode(mode))
|
||||
const execute = vi.fn(async (item: Gate) => resultFor(item))
|
||||
|
||||
await expect(runGates(subject, subject.length, execute)).resolves.toHaveLength(subject.length)
|
||||
})
|
||||
|
||||
it.each([
|
||||
['empty', plan([]), /plan has no gates/],
|
||||
['duplicate ids', plan([gate('same'), gate('same')]), /duplicate gate id "same"/],
|
||||
['unsafe ids', plan([gate('unsafe id')]), /gate id "unsafe id" must contain only lowercase letters/],
|
||||
['unknown dependencies', plan([gate('subject', { needs: ['missing'] })]), /depends on unknown gate "missing"/],
|
||||
['cycles', plan([gate('first', { needs: ['second'] }), gate('second', { needs: ['first'] })]), /dependency cycle: first -> second -> first/],
|
||||
])('rejects %s before starting a child', async (_label, invalid, message) => {
|
||||
['empty', [], /gate graph has no gates/],
|
||||
['duplicate ids', [gate('same'), gate('same')], /duplicate gate id "same"/],
|
||||
['unknown dependencies', [gate('subject', { needs: ['missing'] })], /depends on unknown gate "missing"/],
|
||||
['cycles', [gate('first', { needs: ['second'] }), gate('second', { needs: ['first'] })], /dependency cycle: first -> second -> first/],
|
||||
] as const)('rejects %s before starting a child', async (_label, invalid, message) => {
|
||||
const execute = vi.fn(async (subject: Gate) => resultFor(subject))
|
||||
await expect(executeGatePlan(invalid, 1, execute)).rejects.toThrow(message)
|
||||
|
||||
await expect(runGates([...invalid], 1, execute)).rejects.toThrow(message)
|
||||
expect(execute).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('rejects an invalid plan worker bound', () => {
|
||||
expect(() => {
|
||||
validateGatePlan({ ...plan([gate('subject')]), maxWorkers: 0 })
|
||||
}).toThrow(
|
||||
'maxWorkers must be a positive integer',
|
||||
)
|
||||
})
|
||||
|
||||
it('rejects an executor request above the plan worker ceiling before starting a child', async () => {
|
||||
it('rejects an invalid worker count before starting a child', async () => {
|
||||
const execute = vi.fn(async (subject: Gate) => resultFor(subject))
|
||||
await expect(executeGatePlan({ ...plan([gate('subject')]), maxWorkers: 1 }, 2, execute)).rejects.toThrow(
|
||||
'exceeds the check-all plan ceiling 1',
|
||||
)
|
||||
|
||||
await expect(runGates([gate('subject')], 0, execute)).rejects.toThrow('max concurrency must be a positive integer')
|
||||
expect(execute).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('reports a settled failure before an unrelated gate finishes', async () => {
|
||||
const first = gate('first')
|
||||
const second = gate('second')
|
||||
const settle = new Map<string, (result: GateResult) => void>()
|
||||
const observed: string[] = []
|
||||
const execution = executeGatePlan(
|
||||
plan([first, second]),
|
||||
2,
|
||||
subject => new Promise(resolve => settle.set(subject.id, resolve)),
|
||||
result => observed.push(`${result.gate.id}:${result.status}`),
|
||||
)
|
||||
|
||||
const settleFirst = settle.get(first.id)
|
||||
const settleSecond = settle.get(second.id)
|
||||
if (settleFirst === undefined || settleSecond === undefined) throw new Error('expected both gates to start')
|
||||
settleFirst(resultFor(first, 'failed'))
|
||||
await vi.waitFor(() => {
|
||||
expect(observed).toEqual(['first:failed'])
|
||||
})
|
||||
settleSecond(resultFor(second))
|
||||
|
||||
await expect(execution).resolves.toHaveLength(2)
|
||||
expect(observed).toEqual(['first:failed', 'second:passed'])
|
||||
})
|
||||
|
||||
it('propagates dependency skips in causal order', async () => {
|
||||
const leaf = gate('leaf', { needs: ['middle'] })
|
||||
const middle = gate('middle', { needs: ['root'] })
|
||||
const rootGate = gate('root')
|
||||
it('skips dependents after their prerequisite fails', async () => {
|
||||
const dependent = gate('dependent', { needs: ['root'] })
|
||||
const root = gate('root')
|
||||
const execute = vi.fn(async (subject: Gate) => resultFor(subject, 'failed'))
|
||||
const observed: string[] = []
|
||||
|
||||
const results = await executeGatePlan(
|
||||
plan([leaf, middle, rootGate]),
|
||||
1,
|
||||
execute,
|
||||
result => observed.push(`${result.gate.id}:${result.status}`),
|
||||
)
|
||||
const results = await runGates([dependent, root], 1, execute)
|
||||
|
||||
expect(execute).toHaveBeenCalledOnce()
|
||||
expect(execute).toHaveBeenCalledWith(rootGate)
|
||||
expect(observed).toEqual(['root:failed', 'middle:skipped', 'leaf:skipped'])
|
||||
expect(results.find(result => result.gate === middle)?.error).toBe('dependency failed or skipped: root')
|
||||
expect(results.find(result => result.gate === leaf)?.error).toBe('dependency failed or skipped: middle')
|
||||
})
|
||||
|
||||
it('selects a target with its transitive dependencies in canonical plan order', () => {
|
||||
const subject = plan([
|
||||
gate('prepare'),
|
||||
gate('build', { needs: ['prepare'] }),
|
||||
gate('snapshot', { needs: ['build'], env: { DSH_EXAMPLE_MODE: { operation: 'set', value: 'lib' } } }),
|
||||
gate('unrelated'),
|
||||
])
|
||||
expect(gateDependencyClosure(subject, 'snapshot').map(item => item.id)).toEqual(['prepare', 'build', 'snapshot'])
|
||||
expect(gateDependencyClosure(subject, 'snapshot').at(-1)?.env).toEqual({
|
||||
DSH_EXAMPLE_MODE: { operation: 'set', value: 'lib' },
|
||||
})
|
||||
expect(execute).toHaveBeenCalledWith(root)
|
||||
expect(results[0]).toMatchObject({ gate: dependent, status: 'skipped', error: 'dependency failed or skipped: root' })
|
||||
})
|
||||
})
|
||||
|
||||
describe('gate plan inspection and replay', () => {
|
||||
it('parses package-script separators, list JSON, and focused runs', () => {
|
||||
expect(parseCliRequest(['check-all', '--', '--list', '--json'])).toEqual({
|
||||
mode: 'check-all', list: true, json: true,
|
||||
})
|
||||
expect(parseCliRequest(['check-all', '--only', 'snapshot'])).toEqual({
|
||||
mode: 'check-all', list: false, json: false, only: 'snapshot',
|
||||
})
|
||||
expect(() => parseCliRequest(['check-all', '--json'])).toThrow('--json requires --list')
|
||||
expect(() => parseCliRequest(['pre-push'])).toThrow('expected mode')
|
||||
})
|
||||
describe('Node 24 consumer graph', () => {
|
||||
it('owns the seven-command pool and orders restored-artifact consumers', () => {
|
||||
const subject = withPnpmEntrypoint(() => gatesForMode('ci-consumers'))
|
||||
|
||||
it('renders deterministic human and stable JSON fields without inherited environment values', () => {
|
||||
vi.stubEnv('DEEPSEEK_API_KEY', 'ambient-secret')
|
||||
const subject = plan([
|
||||
gate('prepare'),
|
||||
gate('subject', {
|
||||
needs: ['prepare'],
|
||||
allowFailure: true,
|
||||
env: {
|
||||
Z_MODE: { operation: 'set', value: 'lib' },
|
||||
ACCESS_TOKEN: { operation: 'set', value: 'scheduler-secret' },
|
||||
NODE_OPTIONS: { operation: 'append', value: '--max-old-space-size=8192' },
|
||||
},
|
||||
}),
|
||||
])
|
||||
|
||||
const json = formatGatePlanJson(subject)
|
||||
expect(formatGatePlanJson(subject)).toBe(json)
|
||||
expect(json).not.toContain('ambient-secret')
|
||||
expect(json).not.toContain('scheduler-secret')
|
||||
expect(JSON.parse(json)).toEqual({
|
||||
version: 1,
|
||||
mode: 'check-all',
|
||||
script: 'check:all',
|
||||
scope: 'complete',
|
||||
maxWorkers: null,
|
||||
gates: [
|
||||
{ id: 'prepare', label: 'prepare', command: 'run prepare', needs: [], env: {}, blocking: true },
|
||||
{
|
||||
id: 'subject',
|
||||
label: 'subject',
|
||||
command: 'run subject',
|
||||
needs: ['prepare'],
|
||||
env: {
|
||||
ACCESS_TOKEN: { operation: 'set', value: '<redacted>' },
|
||||
NODE_OPTIONS: { operation: 'append', value: '--max-old-space-size=8192' },
|
||||
Z_MODE: { operation: 'set', value: 'lib' },
|
||||
},
|
||||
blocking: false,
|
||||
},
|
||||
],
|
||||
})
|
||||
expect(formatGatePlanList(subject)).toContain('- subject [non-blocking] subject')
|
||||
expect(formatGatePlanList(subject)).toContain('needs: prepare')
|
||||
expect(formatGatePlanList(subject)).toContain('max workers: (host and gate count)')
|
||||
})
|
||||
|
||||
it('emits one clean JSON object through the documented silent package-script entry', () => {
|
||||
const result = spawnSync('pnpm', [
|
||||
'--silent',
|
||||
'run',
|
||||
'check:ci:consumers',
|
||||
'--',
|
||||
'--list',
|
||||
'--json',
|
||||
], {
|
||||
cwd: repositoryRoot,
|
||||
encoding: 'utf8',
|
||||
shell: process.platform === 'win32',
|
||||
timeout: 10_000,
|
||||
})
|
||||
if (result.error !== undefined) throw result.error
|
||||
expect(result.status, result.stderr).toBe(0)
|
||||
expect(JSON.parse(result.stdout)).toMatchObject({
|
||||
version: 1,
|
||||
mode: 'ci-consumers',
|
||||
script: 'check:ci:consumers',
|
||||
scope: 'complete',
|
||||
maxWorkers: 7,
|
||||
})
|
||||
})
|
||||
|
||||
it.skipIf(process.platform === 'win32')('executes when the script entry path is a symlink', () => {
|
||||
const temporary = mkdtempSync(join(tmpdir(), 'dsh-run-gates-entry-'))
|
||||
const entry = join(temporary, 'run-gates.ts')
|
||||
try {
|
||||
symlinkSync(join(repositoryRoot, 'scripts/run-gates.ts'), entry)
|
||||
const result = spawnSync(process.execPath, [
|
||||
'--import',
|
||||
'tsx',
|
||||
entry,
|
||||
'ci-consumers',
|
||||
'--list',
|
||||
'--json',
|
||||
], {
|
||||
cwd: repositoryRoot,
|
||||
encoding: 'utf8',
|
||||
env: { ...process.env, npm_execpath: process.env.npm_execpath ?? '/private/pnpm.cjs' },
|
||||
timeout: 10_000,
|
||||
})
|
||||
|
||||
expect(result.status, result.stderr).toBe(0)
|
||||
expect(JSON.parse(result.stdout)).toMatchObject({ mode: 'ci-consumers', maxWorkers: 7 })
|
||||
} finally {
|
||||
rmSync(temporary, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
it('prints focused-run context and replay through a real failure block', () => {
|
||||
const result = spawnSync(process.execPath, [
|
||||
'--import',
|
||||
'tsx',
|
||||
join(repositoryRoot, 'scripts/run-gates.ts'),
|
||||
'ci-lint',
|
||||
'--only',
|
||||
'duplication',
|
||||
], {
|
||||
cwd: repositoryRoot,
|
||||
encoding: 'utf8',
|
||||
env: { ...process.env, npm_execpath: join(repositoryRoot, 'scripts/missing-pnpm-entrypoint.cjs') },
|
||||
timeout: 10_000,
|
||||
})
|
||||
|
||||
expect(result.status).toBe(1)
|
||||
expect(result.stdout).toContain('partial diagnostic evidence; the complete owning mode is pnpm run check:ci:lint')
|
||||
expect(result.stderr).toContain('outcome: exit 1')
|
||||
expect(result.stderr).toContain('replay: pnpm run check:ci:lint -- --only duplication')
|
||||
})
|
||||
|
||||
it('applies append and set operations through the child spawn environment', async () => {
|
||||
vi.stubEnv('NODE_OPTIONS', '--trace-warnings')
|
||||
vi.stubEnv('INHERITED', 'kept')
|
||||
const result = await runGate(gate('subject', {
|
||||
args: ['-e', 'process.stdout.write(JSON.stringify({ nodeOptions: process.env.NODE_OPTIONS, mode: process.env.MODE, inherited: process.env.INHERITED }))'],
|
||||
env: {
|
||||
NODE_OPTIONS: { operation: 'append', value: '--max-old-space-size=8192' },
|
||||
MODE: { operation: 'set', value: 'lib' },
|
||||
},
|
||||
}))
|
||||
|
||||
expect(result.status).toBe('passed')
|
||||
expect(JSON.parse(result.stdout)).toEqual({
|
||||
nodeOptions: '--trace-warnings --max-old-space-size=8192',
|
||||
mode: 'lib',
|
||||
inherited: 'kept',
|
||||
})
|
||||
})
|
||||
|
||||
it.skipIf(process.platform === 'win32')('reports signal termination as an orthogonal real-process outcome', async () => {
|
||||
const subjectGate = gate('terminated', {
|
||||
args: ['-e', "process.kill(process.pid, 'SIGTERM')"],
|
||||
})
|
||||
const result = await runGate(subjectGate)
|
||||
|
||||
expect(result.status).toBe('failed')
|
||||
expect(result.exitCode).toBeNull()
|
||||
expect(result.signalCode).toBe('SIGTERM')
|
||||
expect(formatGateResultReason(result)).toBe('signal SIGTERM')
|
||||
})
|
||||
})
|
||||
|
||||
describe('Node 24 consumer plan', () => {
|
||||
it('owns the same seven-worker command pool and orders restored-artifact validation before dependent consumers', () => {
|
||||
const subject = withPnpmEntrypoint(() => gatePlanForMode('ci-consumers'))
|
||||
validateGatePlan(subject)
|
||||
expect(subject.maxWorkers).toBe(7)
|
||||
expect(listedGatePlan(subject).maxWorkers).toBe(7)
|
||||
expect(resolvePlanConcurrency(subject, undefined, 4)).toEqual({
|
||||
expect(defaultConcurrency('ci-consumers', subject.length, 4)).toEqual({
|
||||
workers: 7,
|
||||
source: 'ci-consumers plan default 7',
|
||||
source: 'ci-consumers gate count',
|
||||
})
|
||||
expect(resolvePlanConcurrency(subject, '4', 32)).toEqual({
|
||||
workers: 4,
|
||||
source: '$DSH_GATE_CONCURRENCY',
|
||||
})
|
||||
expect(resolvePlanConcurrency(subject, '8', 32)).toEqual({
|
||||
workers: 7,
|
||||
source: '$DSH_GATE_CONCURRENCY, ci-consumers plan cap 7',
|
||||
})
|
||||
expect(subject.gates.map(item => item.id)).toEqual([
|
||||
expect(subject.map(item => item.id)).toEqual([
|
||||
'lint-and-duplication',
|
||||
'node-compat',
|
||||
'snapshot',
|
||||
@@ -362,19 +113,25 @@ describe('Node 24 consumer plan', () => {
|
||||
'built-package-invariants',
|
||||
'built-bin-smoke',
|
||||
])
|
||||
expect(subject.gates.find(item => item.id === 'publint')?.needs).toBeUndefined()
|
||||
expect(subject.gates.find(item => item.id === 'built-package-invariants')?.needs).toEqual(['publint'])
|
||||
expect(subject.gates.find(item => item.id === 'lint-and-duplication')?.needs).toEqual(['built-package-invariants'])
|
||||
expect(subject.find(item => item.id === 'publint')?.needs).toBeUndefined()
|
||||
expect(subject.find(item => item.id === 'built-package-invariants')?.needs).toEqual(['publint'])
|
||||
expect(subject.find(item => item.id === 'lint-and-duplication')?.needs).toEqual(['built-package-invariants'])
|
||||
for (const id of ['snapshot', 'node-next-types', 'built-bin-smoke']) {
|
||||
expect(subject.gates.find(item => item.id === id)?.needs).toEqual(['built-package-invariants'])
|
||||
expect(subject.find(item => item.id === id)?.needs).toEqual(['built-package-invariants'])
|
||||
}
|
||||
expect(gateDependencyClosure(subject, 'snapshot').map(item => item.id)).toEqual([
|
||||
'snapshot',
|
||||
'publint',
|
||||
'built-package-invariants',
|
||||
])
|
||||
expect(listedGatePlan(subject).gates.find(item => item.id === 'snapshot')?.env).toEqual({
|
||||
DSH_EXAMPLE_MODE: { operation: 'set', value: 'lib' },
|
||||
})
|
||||
expect(subject.find(item => item.id === 'snapshot')?.env).toEqual({ DSH_EXAMPLE_MODE: 'lib' })
|
||||
})
|
||||
})
|
||||
|
||||
describe('gate process outcomes', () => {
|
||||
it.skipIf(process.platform === 'win32')('reports signal termination independently from exit status', async () => {
|
||||
const result = await runGate(gate('terminated', {
|
||||
args: ['-e', "process.kill(process.pid, 'SIGTERM')"],
|
||||
}))
|
||||
|
||||
expect(result.status).toBe('failed')
|
||||
expect(result.exitCode).toBeNull()
|
||||
expect(result.signalCode).toBe('SIGTERM')
|
||||
expect(formatGateResultReason(result)).toBe('signal SIGTERM')
|
||||
})
|
||||
})
|
||||
|
||||
@@ -1,46 +1,35 @@
|
||||
/**
|
||||
* Construct, inspect, and run local and CI quality-gate plans with bounded scheduling.
|
||||
* Run local and CI quality gates with bounded in-process scheduling.
|
||||
*
|
||||
* Package scripts own public aggregate names; this runner owns their validated
|
||||
* dependency graphs, scheduler environment, and replay diagnostics.
|
||||
* @see ../.agents/notes/implemented/process/2026-07-27-replayable-gate-plans.md
|
||||
* dependency graphs, scheduler environment, and process diagnostics.
|
||||
* @see ../.agents/notes/implemented/process/2026-07-06-parallel-pre-push-gates.md
|
||||
*/
|
||||
import { spawn } from 'node:child_process'
|
||||
import { availableParallelism } from 'node:os'
|
||||
import { resolve } from 'node:path'
|
||||
import { performance } from 'node:perf_hooks'
|
||||
import { parseArgs } from 'node:util'
|
||||
|
||||
const MODE_SCRIPTS = {
|
||||
'ci-primary': 'check:ci',
|
||||
'ci-static': 'check:ci:static',
|
||||
'ci-lint': 'check:ci:lint',
|
||||
'ci-coverage': 'check:ci:coverage',
|
||||
'ci-snapshot': 'check:ci:snapshot',
|
||||
'ci-artifacts': 'check:ci:artifacts',
|
||||
'ci-consumers': 'check:ci:consumers',
|
||||
'ci-windows-blocking': 'check:ci:windows-blocking',
|
||||
'ci-windows-complete': 'check:ci:windows-complete',
|
||||
'ci-windows-observational': 'check:ci:windows-observational',
|
||||
'node-compat': 'check:node-compat',
|
||||
'check-all': 'check:all',
|
||||
'doc-sync': 'doc-sync',
|
||||
} as const
|
||||
|
||||
/** A named aggregate exposed by the gate runner. */
|
||||
export type Mode = keyof typeof MODE_SCRIPTS
|
||||
|
||||
const MODES = Object.keys(MODE_SCRIPTS) as Mode[]
|
||||
export type Mode =
|
||||
| 'ci-primary'
|
||||
| 'ci-static'
|
||||
| 'ci-lint'
|
||||
| 'ci-coverage'
|
||||
| 'ci-snapshot'
|
||||
| 'ci-artifacts'
|
||||
| 'ci-consumers'
|
||||
| 'ci-windows-blocking'
|
||||
| 'ci-windows-complete'
|
||||
| 'ci-windows-observational'
|
||||
| 'node-compat'
|
||||
| 'check-all'
|
||||
| 'doc-sync'
|
||||
|
||||
type GateResultStatus = 'passed' | 'failed' | 'skipped'
|
||||
type GateState = 'pending' | 'running' | GateResultStatus
|
||||
|
||||
/** One scheduler-owned environment operation, resolved against inherited values only at spawn time. */
|
||||
export type GateEnvironmentOverride =
|
||||
| { operation: 'set'; value: string }
|
||||
| { operation: 'append'; value: string }
|
||||
|
||||
/** A command and its dependency metadata inside one gate plan. */
|
||||
/** A command and its dependency metadata inside one aggregate. */
|
||||
export interface Gate {
|
||||
id: string
|
||||
label: string
|
||||
@@ -48,27 +37,15 @@ export interface Gate {
|
||||
command: string
|
||||
args: string[]
|
||||
needs?: string[]
|
||||
env?: Record<string, GateEnvironmentOverride>
|
||||
input?: string
|
||||
verify?: (result: GateResult) => Promise<void>
|
||||
env?: Record<string, string | undefined>
|
||||
allowFailure?: boolean
|
||||
}
|
||||
|
||||
/** A complete executable aggregate and the package script that owns its diagnostics. */
|
||||
export interface GatePlan {
|
||||
mode: Mode
|
||||
script: string
|
||||
gates: Gate[]
|
||||
maxWorkers?: number
|
||||
}
|
||||
|
||||
/** The observed outcome of one gate process. */
|
||||
export interface GateResult {
|
||||
gate: Gate
|
||||
status: GateResultStatus
|
||||
durationMs: number
|
||||
stdout: string
|
||||
stderr: string
|
||||
output: GateOutputChunk[]
|
||||
exitCode: number | null
|
||||
signalCode: NodeJS.Signals | null
|
||||
@@ -85,37 +62,11 @@ interface RunningGate {
|
||||
promise: Promise<GateResult>
|
||||
}
|
||||
|
||||
/** The effective worker count and the facts that selected it. */
|
||||
export interface ResolvedConcurrency {
|
||||
interface ConcurrencyDefault {
|
||||
workers: number
|
||||
source: string
|
||||
}
|
||||
|
||||
interface RunRequest {
|
||||
mode: Mode
|
||||
list: boolean
|
||||
json: boolean
|
||||
only?: string
|
||||
}
|
||||
|
||||
interface ListedGate {
|
||||
id: string
|
||||
label: string
|
||||
command: string
|
||||
needs: string[]
|
||||
env: Record<string, GateEnvironmentOverride>
|
||||
blocking: boolean
|
||||
}
|
||||
|
||||
interface ListedPlan {
|
||||
version: 1
|
||||
mode: Mode
|
||||
script: string
|
||||
scope: 'complete'
|
||||
maxWorkers: number | null
|
||||
gates: ListedGate[]
|
||||
}
|
||||
|
||||
type GateExecutor = (gate: Gate) => Promise<GateResult>
|
||||
type ResultObserver = (result: GateResult) => void
|
||||
|
||||
@@ -125,83 +76,74 @@ if (import.meta.main) {
|
||||
}
|
||||
|
||||
async function main(args: string[]): Promise<number> {
|
||||
const request = parseCliRequest(args)
|
||||
const completePlan = gatePlanForMode(request.mode)
|
||||
validateGatePlan(completePlan)
|
||||
if (request.list) {
|
||||
console.log(request.json ? formatGatePlanJson(completePlan) : formatGatePlanList(completePlan))
|
||||
return 0
|
||||
}
|
||||
|
||||
const plan = request.only === undefined
|
||||
? completePlan
|
||||
: { ...completePlan, gates: gateDependencyClosure(completePlan, request.only) }
|
||||
validateGatePlan(plan)
|
||||
if (request.only !== undefined) console.log(formatOnlyNotice(completePlan, request.only))
|
||||
|
||||
const concurrency = resolvePlanConcurrency(plan, process.env.DSH_GATE_CONCURRENCY)
|
||||
const maxConcurrency = concurrency.workers
|
||||
const concurrencySource = concurrency.source
|
||||
const mode = parseMode(args[0])
|
||||
const gates = gatesForMode(mode)
|
||||
const concurrencyDefault = defaultConcurrency(mode, gates.length)
|
||||
const concurrencyOverride = process.env.DSH_GATE_CONCURRENCY
|
||||
const maxConcurrency = concurrencyFromEnv('DSH_GATE_CONCURRENCY', concurrencyDefault.workers)
|
||||
const concurrencySource = concurrencyOverride === undefined || concurrencyOverride === ''
|
||||
? concurrencyDefault.source
|
||||
: '$DSH_GATE_CONCURRENCY'
|
||||
const startedAt = performance.now()
|
||||
console.log(`run-gates: ${request.mode} running ${plan.gates.length} gate(s) with ${maxConcurrency} worker(s) from ${concurrencySource}.`)
|
||||
console.log(`run-gates: ${mode} running ${gates.length} gate(s) with ${maxConcurrency} worker(s) from ${concurrencySource}.`)
|
||||
|
||||
const results = await executeGatePlan(plan, maxConcurrency, runGate, (result) => {
|
||||
printResult(completePlan, result)
|
||||
})
|
||||
printSummary(completePlan, results, performance.now() - startedAt)
|
||||
const results = await runGates(gates, maxConcurrency, runGate, printResult)
|
||||
printSummary(results, performance.now() - startedAt)
|
||||
return results.some(result => result.gate.allowFailure !== true && (result.status === 'failed' || result.status === 'skipped'))
|
||||
? 1
|
||||
: 0
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse one runner invocation without constructing or starting its plan.
|
||||
* @param args - command-line arguments after the script entrypoint.
|
||||
* @returns the validated run request.
|
||||
*/
|
||||
export function parseCliRequest(args: readonly string[]): RunRequest {
|
||||
const mode = parseMode(args[0])
|
||||
const optionArgs = args[1] === '--' ? args.slice(2) : args.slice(1)
|
||||
const { values: { list, json, only } } = parseArgs({
|
||||
args: optionArgs,
|
||||
options: {
|
||||
list: { type: 'boolean', default: false },
|
||||
json: { type: 'boolean', default: false },
|
||||
only: { type: 'string' },
|
||||
},
|
||||
strict: true,
|
||||
allowPositionals: false,
|
||||
})
|
||||
if (json && !list) throw new Error('run-gates: --json requires --list.')
|
||||
if (list && only !== undefined) throw new Error('run-gates: --list and --only are mutually exclusive.')
|
||||
return { mode, list, json, ...only === undefined ? {} : { only } }
|
||||
}
|
||||
|
||||
function parseMode(raw: string | undefined): Mode {
|
||||
if (MODES.includes(raw as Mode)) return raw as Mode
|
||||
throw new Error(`run-gates: expected mode ${MODES.join(' | ')}, got ${JSON.stringify(raw)}.`)
|
||||
switch (raw) {
|
||||
case 'ci-primary':
|
||||
case 'ci-static':
|
||||
case 'ci-lint':
|
||||
case 'ci-coverage':
|
||||
case 'ci-snapshot':
|
||||
case 'ci-artifacts':
|
||||
case 'ci-consumers':
|
||||
case 'ci-windows-blocking':
|
||||
case 'ci-windows-complete':
|
||||
case 'ci-windows-observational':
|
||||
case 'node-compat':
|
||||
case 'check-all':
|
||||
case 'doc-sync':
|
||||
return raw
|
||||
default:
|
||||
throw new Error(
|
||||
`run-gates: expected mode ci-primary | ci-static | ci-lint | ci-coverage | ci-snapshot | ci-artifacts | ci-consumers | ci-windows-blocking | ci-windows-complete | ci-windows-observational | node-compat | check-all | doc-sync, got ${JSON.stringify(raw)}.`,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
function defaultConcurrency(plan: GatePlan, available: number): ResolvedConcurrency {
|
||||
if (plan.maxWorkers !== undefined) {
|
||||
return {
|
||||
workers: Math.min(plan.gates.length, plan.maxWorkers),
|
||||
source: `${plan.mode} plan default ${plan.maxWorkers}`,
|
||||
}
|
||||
}
|
||||
/**
|
||||
* Resolve the default worker count for one aggregate.
|
||||
* @param selectedMode - aggregate whose resource posture applies.
|
||||
* @param total - number of gates in the aggregate.
|
||||
* @param available - host CPU availability for ordinary modes.
|
||||
* @returns the default worker count and its diagnostic source.
|
||||
*/
|
||||
export function defaultConcurrency(
|
||||
selectedMode: Mode,
|
||||
total: number,
|
||||
available = availableParallelism(),
|
||||
): ConcurrencyDefault {
|
||||
if (selectedMode === 'ci-consumers') return { workers: total, source: 'ci-consumers gate count' }
|
||||
// Local modes cap workers: several doc gates each build a full ts.Program,
|
||||
// so an uncapped default on a large host trades wall clock for memory blowups.
|
||||
const localCap = plan.mode === 'check-all' || plan.mode === 'doc-sync'
|
||||
const localCap = selectedMode === 'check-all' || selectedMode === 'doc-sync'
|
||||
const modeLimit = localCap ? Math.min(4, available) : available
|
||||
return {
|
||||
workers: Math.min(plan.gates.length, modeLimit),
|
||||
workers: Math.min(total, modeLimit),
|
||||
source: localCap
|
||||
? `${available} available CPU(s), ${plan.mode} cap 4`
|
||||
? `${available} available CPU(s), ${selectedMode} cap 4`
|
||||
: `${available} available CPU(s)`,
|
||||
}
|
||||
}
|
||||
|
||||
function concurrencyFromValue(name: string, raw: string | undefined, fallback: number): number {
|
||||
function concurrencyFromEnv(name: string, fallback: number): number {
|
||||
const raw = process.env[name]
|
||||
if (raw === undefined || raw === '') return fallback
|
||||
const parsed = Number.parseInt(raw, 10)
|
||||
if (!Number.isSafeInteger(parsed) || parsed < 1) {
|
||||
@@ -210,33 +152,6 @@ function concurrencyFromValue(name: string, raw: string | undefined, fallback: n
|
||||
return parsed
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve a plan's default, optional environment request, and hard worker ceiling.
|
||||
* @param plan - validated complete or diagnostic plan.
|
||||
* @param override - optional `DSH_GATE_CONCURRENCY` value.
|
||||
* @param available - host CPU availability for modes without a plan-owned default.
|
||||
* @returns the effective worker count and its inspectable source.
|
||||
*/
|
||||
export function resolvePlanConcurrency(
|
||||
plan: GatePlan,
|
||||
override: string | undefined,
|
||||
available = availableParallelism(),
|
||||
): ResolvedConcurrency {
|
||||
validateGatePlan(plan)
|
||||
const defaultValue = defaultConcurrency(plan, available)
|
||||
const requested = concurrencyFromValue('DSH_GATE_CONCURRENCY', override, defaultValue.workers)
|
||||
const workers = Math.min(requested, plan.maxWorkers ?? requested)
|
||||
const requestedSource = override === undefined || override === ''
|
||||
? defaultValue.source
|
||||
: '$DSH_GATE_CONCURRENCY'
|
||||
return {
|
||||
workers,
|
||||
source: workers === requested
|
||||
? requestedSource
|
||||
: `${requestedSource}, ${plan.mode} plan cap ${String(plan.maxWorkers)}`,
|
||||
}
|
||||
}
|
||||
|
||||
function pnpmScript(id: string, script: string, options: Partial<Gate> = {}): Gate {
|
||||
return {
|
||||
id,
|
||||
@@ -266,21 +181,16 @@ function pnpmInvocation(args: string[]): Pick<Gate, 'command' | 'args'> {
|
||||
return { command: process.execPath, args: [entrypoint, ...args] }
|
||||
}
|
||||
|
||||
/**
|
||||
* Construct the complete plan for a named aggregate without executing it.
|
||||
* @param selected - aggregate mode to construct.
|
||||
* @returns the aggregate's package-script identity and gate graph.
|
||||
*/
|
||||
export function gatePlanForMode(selected: Mode): GatePlan {
|
||||
return {
|
||||
mode: selected,
|
||||
script: MODE_SCRIPTS[selected],
|
||||
gates: gatesForMode(selected),
|
||||
...selected === 'ci-consumers' ? { maxWorkers: 7 } : {},
|
||||
}
|
||||
function nodeOptions(...options: string[]): string {
|
||||
return [process.env.NODE_OPTIONS, ...options].filter(option => option !== undefined && option !== '').join(' ')
|
||||
}
|
||||
|
||||
function gatesForMode(selected: Mode): Gate[] {
|
||||
/**
|
||||
* Construct the complete gate list for a named aggregate.
|
||||
* @param selected - aggregate mode to construct.
|
||||
* @returns the aggregate's gate graph.
|
||||
*/
|
||||
export function gatesForMode(selected: Mode): Gate[] {
|
||||
switch (selected) {
|
||||
case 'ci-primary':
|
||||
return ciPrimaryGates()
|
||||
@@ -320,7 +230,7 @@ function gatesForMode(selected: Mode): Gate[] {
|
||||
...hygieneLeafGates({ artifactNeeds: ['build'] }),
|
||||
...docSyncLeafGates({
|
||||
docTypecheckNeeds: ['build'],
|
||||
docTypecheckEnv: { DSH_DOC_TYPECHECK_USE_BUILD_OUTPUT: { operation: 'set', value: '1' } },
|
||||
docTypecheckEnv: { DSH_DOC_TYPECHECK_USE_BUILD_OUTPUT: '1' },
|
||||
}),
|
||||
pnpmScript('module-graph', 'verify-module-graph', { label: 'module graph' }),
|
||||
]
|
||||
@@ -389,7 +299,7 @@ function ciStaticGates(): Gate[] {
|
||||
pnpmScript('build', 'build'),
|
||||
...docSyncLeafGates({
|
||||
docTypecheckNeeds: ['build'],
|
||||
docTypecheckEnv: { DSH_DOC_TYPECHECK_USE_BUILD_OUTPUT: { operation: 'set', value: '1' } },
|
||||
docTypecheckEnv: { DSH_DOC_TYPECHECK_USE_BUILD_OUTPUT: '1' },
|
||||
docsBuildScript: 'docs:build:mpa',
|
||||
}),
|
||||
pnpmScript('module-graph', 'verify-module-graph', { label: 'module graph' }),
|
||||
@@ -479,17 +389,17 @@ function lintGate(eslintTargets: readonly string[] = ['.']): Gate {
|
||||
'content',
|
||||
], {
|
||||
label: 'lint',
|
||||
env: { NODE_OPTIONS: { operation: 'append', value: '--max-old-space-size=8192' } },
|
||||
env: { NODE_OPTIONS: nodeOptions('--max-old-space-size=8192') },
|
||||
})
|
||||
}
|
||||
if (concurrencyArgs.length > 0) {
|
||||
return pnpmExec('lint', ['eslint', ...eslintTargets, ...concurrencyArgs], {
|
||||
label: 'lint',
|
||||
env: { NODE_OPTIONS: { operation: 'append', value: '--max-old-space-size=8192' } },
|
||||
env: { NODE_OPTIONS: nodeOptions('--max-old-space-size=8192') },
|
||||
})
|
||||
}
|
||||
return pnpmScript('lint', 'lint', {
|
||||
env: { NODE_OPTIONS: { operation: 'append', value: '--max-old-space-size=8192' } },
|
||||
env: { NODE_OPTIONS: nodeOptions('--max-old-space-size=8192') },
|
||||
})
|
||||
}
|
||||
|
||||
@@ -520,7 +430,7 @@ function coverageGate(): Gate {
|
||||
// Build-owning modes wait on `build`; a restored-artifact mode passes its validation dependency.
|
||||
function snapshotGate(needs: string[] = ['build']): Gate {
|
||||
return pnpmScript('snapshot', 'test:snapshot', {
|
||||
env: { DSH_EXAMPLE_MODE: { operation: 'set', value: 'lib' } },
|
||||
env: { DSH_EXAMPLE_MODE: 'lib' },
|
||||
needs,
|
||||
})
|
||||
}
|
||||
@@ -566,7 +476,7 @@ function hygieneLeafGates(options: { artifactNeeds?: string[] } = {}): Gate[] {
|
||||
|
||||
function docSyncLeafGates(options: {
|
||||
docTypecheckNeeds?: string[]
|
||||
docTypecheckEnv?: Record<string, GateEnvironmentOverride>
|
||||
docTypecheckEnv?: Record<string, string | undefined>
|
||||
docsBuildScript?: 'docs:build' | 'docs:build:mpa'
|
||||
} = {}): Gate[] {
|
||||
const docTypecheckOptions: Partial<Gate> = {}
|
||||
@@ -623,46 +533,32 @@ function builtBinSmokeGate(needs: string[] = ['build']): Gate {
|
||||
], {
|
||||
label: 'built-bin smoke',
|
||||
needs,
|
||||
env: { DSH_EXAMPLE_MODE: { operation: 'set', value: 'lib' } },
|
||||
env: { DSH_EXAMPLE_MODE: 'lib' },
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Reject a plan whose graph cannot be executed unambiguously.
|
||||
* @param plan - complete or diagnostic plan to validate.
|
||||
* Reject a gate list whose graph cannot be executed unambiguously.
|
||||
* @param gates - complete aggregate to validate.
|
||||
*/
|
||||
export function validateGatePlan(plan: GatePlan): void {
|
||||
const errors: string[] = []
|
||||
if (plan.gates.length === 0) errors.push('plan has no gates')
|
||||
if (plan.maxWorkers !== undefined && (!Number.isSafeInteger(plan.maxWorkers) || plan.maxWorkers < 1)) {
|
||||
errors.push(`maxWorkers must be a positive integer, got ${JSON.stringify(plan.maxWorkers)}`)
|
||||
}
|
||||
function validateGateGraph(gates: readonly Gate[]): void {
|
||||
if (gates.length === 0) throw new Error('run-gates: gate graph has no gates.')
|
||||
|
||||
const counts = new Map<string, number>()
|
||||
for (const gate of plan.gates) {
|
||||
counts.set(gate.id, (counts.get(gate.id) ?? 0) + 1)
|
||||
if (!/^[a-z0-9][a-z0-9:-]*$/.test(gate.id)) {
|
||||
errors.push(`gate id ${JSON.stringify(gate.id)} must contain only lowercase letters, digits, colons, and hyphens`)
|
||||
}
|
||||
const ids = new Set<string>()
|
||||
for (const gate of gates) {
|
||||
if (ids.has(gate.id)) throw new Error(`run-gates: duplicate gate id ${JSON.stringify(gate.id)}.`)
|
||||
ids.add(gate.id)
|
||||
}
|
||||
for (const [id, count] of counts) {
|
||||
if (count > 1) errors.push(`duplicate gate id ${JSON.stringify(id)}`)
|
||||
}
|
||||
|
||||
const ids = new Set(counts.keys())
|
||||
for (const gate of plan.gates) {
|
||||
for (const gate of gates) {
|
||||
for (const dependency of gate.needs ?? []) {
|
||||
if (!ids.has(dependency)) {
|
||||
errors.push(`gate ${JSON.stringify(gate.id)} depends on unknown gate ${JSON.stringify(dependency)}`)
|
||||
throw new Error(`run-gates: gate ${JSON.stringify(gate.id)} depends on unknown gate ${JSON.stringify(dependency)}.`)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const cycle = findDependencyCycle(plan.gates)
|
||||
if (cycle !== undefined) errors.push(`dependency cycle: ${cycle.join(' -> ')}`)
|
||||
if (errors.length > 0) {
|
||||
throw new Error(`run-gates: invalid ${plan.mode} plan:\n${errors.map(error => ` - ${error}`).join('\n')}`)
|
||||
}
|
||||
const cycle = findDependencyCycle(gates)
|
||||
if (cycle !== undefined) throw new Error(`run-gates: dependency cycle: ${cycle.join(' -> ')}.`)
|
||||
}
|
||||
|
||||
function findDependencyCycle(gates: readonly Gate[]): string[] | undefined {
|
||||
@@ -698,194 +594,31 @@ function findDependencyCycle(gates: readonly Gate[]): string[] | undefined {
|
||||
}
|
||||
|
||||
/**
|
||||
* Return one target and all of its transitive dependencies in canonical plan order.
|
||||
* @param plan - validated complete owning plan.
|
||||
* @param targetId - gate selected for diagnostic execution.
|
||||
* @returns the target's dependency closure in owning-plan order.
|
||||
*/
|
||||
export function gateDependencyClosure(plan: GatePlan, targetId: string): Gate[] {
|
||||
validateGatePlan(plan)
|
||||
const byId = new Map(plan.gates.map(gate => [gate.id, gate]))
|
||||
if (!byId.has(targetId)) {
|
||||
throw new Error(`run-gates: ${plan.mode} has no gate ${JSON.stringify(targetId)}.`)
|
||||
}
|
||||
|
||||
const selected = new Set<string>()
|
||||
const include = (id: string): void => {
|
||||
if (selected.has(id)) return
|
||||
const gate = byId.get(id)
|
||||
if (gate === undefined) throw new Error(`run-gates: missing validated dependency ${JSON.stringify(id)}.`)
|
||||
for (const dependency of gate.needs ?? []) include(dependency)
|
||||
selected.add(id)
|
||||
}
|
||||
include(targetId)
|
||||
return plan.gates.filter(gate => selected.has(gate.id))
|
||||
}
|
||||
|
||||
/**
|
||||
* Produce the stable machine-readable view used by `--list --json`.
|
||||
* @param plan - complete plan to inspect.
|
||||
* @returns the versioned environment-redacted plan view.
|
||||
*/
|
||||
export function listedGatePlan(plan: GatePlan): ListedPlan {
|
||||
validateGatePlan(plan)
|
||||
return {
|
||||
version: 1,
|
||||
mode: plan.mode,
|
||||
script: plan.script,
|
||||
scope: 'complete',
|
||||
maxWorkers: plan.maxWorkers ?? null,
|
||||
gates: plan.gates.map(listedGate),
|
||||
}
|
||||
}
|
||||
|
||||
function listedGate(gate: Gate): ListedGate {
|
||||
return {
|
||||
id: gate.id,
|
||||
label: gate.label,
|
||||
command: gate.displayCommand,
|
||||
needs: [...gate.needs ?? []],
|
||||
env: listedEnvironment(gate.env),
|
||||
blocking: gate.allowFailure !== true,
|
||||
}
|
||||
}
|
||||
|
||||
function listedEnvironment(
|
||||
environment: Readonly<Record<string, GateEnvironmentOverride>> | undefined,
|
||||
): Record<string, GateEnvironmentOverride> {
|
||||
if (environment === undefined) return {}
|
||||
return Object.fromEntries(Object.entries(environment).sort(([left], [right]) => left.localeCompare(right)).map(([name, override]) => {
|
||||
const value = sensitiveEnvironmentName(name) ? '<redacted>' : override.value
|
||||
return [name, { operation: override.operation, value }]
|
||||
}))
|
||||
}
|
||||
|
||||
function sensitiveEnvironmentName(name: string): boolean {
|
||||
return /(key|secret|token|password|credential)/i.test(name)
|
||||
}
|
||||
|
||||
/**
|
||||
* Render the deterministic human-readable view used by `--list`.
|
||||
* @param plan - complete plan to inspect.
|
||||
* @returns the formatted plan.
|
||||
*/
|
||||
export function formatGatePlanList(plan: GatePlan): string {
|
||||
const listed = listedGatePlan(plan)
|
||||
const lines = [
|
||||
`run-gates: complete ${listed.mode} plan (pnpm run ${listed.script})`,
|
||||
`max workers: ${listed.maxWorkers === null ? '(host and gate count)' : listed.maxWorkers}`,
|
||||
]
|
||||
for (const gate of listed.gates) {
|
||||
lines.push(`- ${gate.id} [${gate.blocking ? 'blocking' : 'non-blocking'}] ${gate.label}`)
|
||||
lines.push(` command: ${gate.command}`)
|
||||
lines.push(` needs: ${gate.needs.length === 0 ? '(none)' : gate.needs.join(', ')}`)
|
||||
lines.push(` env: ${Object.keys(gate.env).length === 0 ? '(none)' : JSON.stringify(gate.env)}`)
|
||||
}
|
||||
return lines.join('\n')
|
||||
}
|
||||
|
||||
/**
|
||||
* Render the stable JSON view used by `--list --json`.
|
||||
* @param plan - complete plan to inspect.
|
||||
* @returns the formatted JSON object.
|
||||
*/
|
||||
export function formatGatePlanJson(plan: GatePlan): string {
|
||||
return JSON.stringify(listedGatePlan(plan), null, 2)
|
||||
}
|
||||
|
||||
/**
|
||||
* Render the package-script command that restores a gate's scheduler context.
|
||||
* @param plan - complete owning plan.
|
||||
* @param gateId - gate to replay with its dependencies.
|
||||
* @returns a shell-independent pnpm command.
|
||||
*/
|
||||
function replayCommand(plan: GatePlan, gateId: string): string {
|
||||
validateGatePlan(plan)
|
||||
if (!plan.gates.some(gate => gate.id === gateId)) {
|
||||
throw new Error(`run-gates: ${plan.mode} has no gate ${JSON.stringify(gateId)}.`)
|
||||
}
|
||||
return `pnpm run ${plan.script} -- --only ${gateId}`
|
||||
}
|
||||
|
||||
/**
|
||||
* Explain that a focused run is diagnostic rather than the complete aggregate.
|
||||
* @param plan - complete owning plan.
|
||||
* @param gateId - selected diagnostic gate.
|
||||
* @returns the partial-evidence notice.
|
||||
*/
|
||||
function formatOnlyNotice(plan: GatePlan, gateId: string): string {
|
||||
return `run-gates: --only ${gateId} is partial diagnostic evidence; the complete owning mode is pnpm run ${plan.script}.`
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve only scheduler-declared environment operations against the spawn environment.
|
||||
* @param gate - gate whose operations to apply.
|
||||
* @param inherited - environment inherited by the runner.
|
||||
* @returns the child environment without mutating the inherited object.
|
||||
*/
|
||||
function resolveGateEnvironment(gate: Gate, inherited: NodeJS.ProcessEnv): NodeJS.ProcessEnv {
|
||||
const resolved = { ...inherited }
|
||||
for (const [name, override] of Object.entries(gate.env ?? {})) {
|
||||
switch (override.operation) {
|
||||
case 'set':
|
||||
resolved[name] = override.value
|
||||
break
|
||||
case 'append': {
|
||||
const current = resolved[name]
|
||||
resolved[name] = current === undefined || current === ''
|
||||
? override.value
|
||||
: `${current} ${override.value}`
|
||||
break
|
||||
}
|
||||
default:
|
||||
assertNever(override)
|
||||
}
|
||||
}
|
||||
return resolved
|
||||
}
|
||||
|
||||
function assertNever(value: never): never {
|
||||
throw new Error(`run-gates: unreachable value ${JSON.stringify(value)}.`)
|
||||
}
|
||||
|
||||
/**
|
||||
* Run a validated plan; invalid input rejects before the injected executor can start a child.
|
||||
* @param plan - complete or diagnostic plan to execute.
|
||||
* Validate and run one aggregate before the injected executor can start a child.
|
||||
* @param gates - complete aggregate to execute.
|
||||
* @param maxActive - maximum concurrent child count.
|
||||
* @param execute - child-process executor.
|
||||
* @param observe - result observer invoked when each gate settles.
|
||||
* @returns results in canonical plan order.
|
||||
* @returns results in aggregate order.
|
||||
*/
|
||||
export async function executeGatePlan(
|
||||
plan: GatePlan,
|
||||
export async function runGates(
|
||||
gates: Gate[],
|
||||
maxActive: number,
|
||||
execute: GateExecutor,
|
||||
observe: ResultObserver = () => {},
|
||||
): Promise<GateResult[]> {
|
||||
validateGatePlan(plan)
|
||||
validateGateGraph(gates)
|
||||
if (!Number.isSafeInteger(maxActive) || maxActive < 1) {
|
||||
throw new Error(`run-gates: max concurrency must be a positive integer, got ${JSON.stringify(maxActive)}.`)
|
||||
}
|
||||
if (plan.maxWorkers !== undefined && maxActive > plan.maxWorkers) {
|
||||
throw new Error(`run-gates: max concurrency ${maxActive} exceeds the ${plan.mode} plan ceiling ${plan.maxWorkers}.`)
|
||||
}
|
||||
return runGates(plan.gates, maxActive, execute, observe)
|
||||
}
|
||||
|
||||
async function runGates(
|
||||
allGates: Gate[],
|
||||
maxActive: number,
|
||||
execute: GateExecutor,
|
||||
observe: ResultObserver,
|
||||
): Promise<GateResult[]> {
|
||||
const states = new Map<string, GateState>(allGates.map(gate => [gate.id, 'pending']))
|
||||
const states = new Map<string, GateState>(gates.map(gate => [gate.id, 'pending']))
|
||||
const results = new Map<string, GateResult>()
|
||||
const running: RunningGate[] = []
|
||||
|
||||
for (;;) {
|
||||
let madeProgress = false
|
||||
while (running.length < maxActive) {
|
||||
const ready = allGates.find(gate => states.get(gate.id) === 'pending' && dependenciesPassed(gate, states))
|
||||
const ready = gates.find(gate => states.get(gate.id) === 'pending' && dependenciesPassed(gate, states))
|
||||
if (ready === undefined) break
|
||||
states.set(ready.id, 'running')
|
||||
running.push({ gate: ready, promise: execute(ready) })
|
||||
@@ -894,13 +627,13 @@ async function runGates(
|
||||
}
|
||||
|
||||
if (running.length === 0) {
|
||||
let pending = allGates.filter(gate => states.get(gate.id) === 'pending')
|
||||
let pending = gates.filter(gate => states.get(gate.id) === 'pending')
|
||||
while (pending.length > 0) {
|
||||
const gate = pending.find(item => (item.needs ?? []).some((id) => {
|
||||
const state = states.get(id)
|
||||
return state === 'failed' || state === 'skipped'
|
||||
}))
|
||||
if (gate === undefined) throw new Error('run-gates: validated plan stalled without a failed dependency.')
|
||||
if (gate === undefined) throw new Error('run-gates: validated graph stalled without a failed dependency.')
|
||||
const failedDeps = (gate.needs ?? []).filter((id) => {
|
||||
const state = states.get(id)
|
||||
return state === 'failed' || state === 'skipped'
|
||||
@@ -909,8 +642,6 @@ async function runGates(
|
||||
gate,
|
||||
status: 'skipped',
|
||||
durationMs: 0,
|
||||
stdout: '',
|
||||
stderr: '',
|
||||
output: [],
|
||||
exitCode: null,
|
||||
signalCode: null,
|
||||
@@ -933,7 +664,7 @@ async function runGates(
|
||||
}
|
||||
}
|
||||
|
||||
return allGates.map((gate) => {
|
||||
return gates.map((gate) => {
|
||||
const result = results.get(gate.id)
|
||||
if (result === undefined) throw new Error(`run-gates: missing result for ${gate.id}.`)
|
||||
return result
|
||||
@@ -947,12 +678,10 @@ function dependenciesPassed(gate: Gate, states: Map<string, GateState>): boolean
|
||||
/**
|
||||
* Execute one gate through the real shell-free child-process boundary.
|
||||
* @param gate - command and scheduler environment to execute.
|
||||
* @returns the complete process and verification outcome.
|
||||
* @returns the complete process outcome.
|
||||
*/
|
||||
export async function runGate(gate: Gate): Promise<GateResult> {
|
||||
const started = performance.now()
|
||||
let stdout = ''
|
||||
let stderr = ''
|
||||
const output: GateOutputChunk[] = []
|
||||
let spawnError: string | undefined
|
||||
|
||||
@@ -962,17 +691,15 @@ export async function runGate(gate: Gate): Promise<GateResult> {
|
||||
}>((resolveExit) => {
|
||||
const child = spawn(gate.command, gate.args, {
|
||||
cwd: root,
|
||||
env: resolveGateEnvironment(gate, process.env),
|
||||
env: { ...process.env, ...gate.env },
|
||||
stdio: ['pipe', 'pipe', 'pipe'],
|
||||
})
|
||||
child.stdout.setEncoding('utf8')
|
||||
child.stderr.setEncoding('utf8')
|
||||
child.stdout.on('data', (chunk: string) => {
|
||||
stdout += chunk
|
||||
output.push({ stream: 'stdout', text: chunk })
|
||||
})
|
||||
child.stderr.on('data', (chunk: string) => {
|
||||
stderr += chunk
|
||||
output.push({ stream: 'stderr', text: chunk })
|
||||
})
|
||||
child.on('error', (error) => {
|
||||
@@ -982,33 +709,20 @@ export async function runGate(gate: Gate): Promise<GateResult> {
|
||||
child.on('close', (exitCode, signalCode) => {
|
||||
resolveExit({ exitCode, signalCode })
|
||||
})
|
||||
if (gate.input !== undefined) child.stdin.end(gate.input)
|
||||
else child.stdin.end()
|
||||
child.stdin.end()
|
||||
})
|
||||
const { exitCode, signalCode } = outcome
|
||||
|
||||
let status: GateResultStatus = exitCode === 0 && signalCode === null && spawnError === undefined ? 'passed' : 'failed'
|
||||
let error = spawnError
|
||||
if (status === 'passed' && gate.verify !== undefined) {
|
||||
try {
|
||||
await gate.verify({ gate, status, durationMs: performance.now() - started, stdout, stderr, output, exitCode, signalCode })
|
||||
} catch (verifyError: unknown) {
|
||||
status = 'failed'
|
||||
error = verifyError instanceof Error ? verifyError.message : String(verifyError)
|
||||
}
|
||||
}
|
||||
|
||||
const status: GateResultStatus = exitCode === 0 && signalCode === null && spawnError === undefined ? 'passed' : 'failed'
|
||||
const result: GateResult = {
|
||||
gate,
|
||||
status,
|
||||
durationMs: performance.now() - started,
|
||||
stdout,
|
||||
stderr,
|
||||
output,
|
||||
exitCode,
|
||||
signalCode,
|
||||
}
|
||||
if (error !== undefined) result.error = error
|
||||
if (spawnError !== undefined) result.error = spawnError
|
||||
return result
|
||||
}
|
||||
|
||||
@@ -1025,7 +739,7 @@ export function formatGateResultReason(result: GateResult): string {
|
||||
return facts.length === 0 ? 'no exit code or signal' : facts.join(', ')
|
||||
}
|
||||
|
||||
function printResult(plan: GatePlan, result: GateResult): void {
|
||||
function printResult(result: GateResult): void {
|
||||
const verbose = process.env.DSH_GATE_VERBOSE === '1'
|
||||
const seconds = (result.durationMs / 1000).toFixed(2)
|
||||
if (result.status === 'passed' && !verbose) {
|
||||
@@ -1037,16 +751,13 @@ function printResult(plan: GatePlan, result: GateResult): void {
|
||||
const writeHeading = result.status === 'passed' ? console.log : console.error
|
||||
writeHeading(`\n== ${heading} ==`)
|
||||
if (result.status !== 'passed') {
|
||||
const environment = listedGate(result.gate).env
|
||||
console.error(`command: ${result.gate.displayCommand}`)
|
||||
if (Object.keys(environment).length > 0) console.error(`scheduler environment: ${JSON.stringify(environment)}`)
|
||||
console.error(`outcome: ${formatGateResultReason(result)}`)
|
||||
console.error(`replay: ${replayCommand(plan, result.gate.id)}`)
|
||||
}
|
||||
printOutput(result.output)
|
||||
}
|
||||
|
||||
function printSummary(plan: GatePlan, results: GateResult[], durationMs: number): void {
|
||||
function printSummary(results: GateResult[], durationMs: number): void {
|
||||
const passed = results.filter(result => result.status === 'passed').length
|
||||
const failed = results.filter(result => result.status === 'failed').length
|
||||
const skipped = results.filter(result => result.status === 'skipped').length
|
||||
@@ -1062,7 +773,7 @@ function printSummary(plan: GatePlan, results: GateResult[], durationMs: number)
|
||||
const reason = formatGateResultReason(result)
|
||||
const disposition = result.gate.allowFailure === true ? 'NON-BLOCKING ' : ''
|
||||
console.error(` - ${disposition}${result.status.toUpperCase()} ${result.gate.label} (${duration}s, ${reason})`)
|
||||
console.error(` replay: ${replayCommand(plan, result.gate.id)}`)
|
||||
console.error(` ${result.gate.displayCommand}`)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user