feat: 自动物化 AgentRun managed repositories
This commit is contained in:
Codex
@@ -0,0 +1,55 @@
|
||||
#!/usr/bin/env node
|
||||
const net = require("node:net");
|
||||
|
||||
const [proxyHost, proxyPortRaw, targetHost, targetPortRaw] = process.argv.slice(2);
|
||||
const proxyPort = Number.parseInt(proxyPortRaw || "", 10);
|
||||
const targetPort = Number.parseInt(targetPortRaw || "", 10);
|
||||
if (!proxyHost || !Number.isInteger(proxyPort) || !targetHost || !Number.isInteger(targetPort)) {
|
||||
console.error("agentrun managed repository proxy-connect: invalid arguments");
|
||||
process.exit(64);
|
||||
}
|
||||
|
||||
let settled = false;
|
||||
let tunnelEstablished = false;
|
||||
function finish(code, message) {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
if (message) console.error(`agentrun managed repository proxy-connect: ${message}`);
|
||||
process.exit(code);
|
||||
}
|
||||
|
||||
const socket = net.createConnection({ host: proxyHost, port: proxyPort });
|
||||
let buffer = Buffer.alloc(0);
|
||||
socket.setTimeout(15000, () => {
|
||||
socket.destroy();
|
||||
finish(65, "proxy connection timeout");
|
||||
});
|
||||
socket.on("connect", () => socket.write(`CONNECT ${targetHost}:${targetPort} HTTP/1.1\r\nHost: ${targetHost}:${targetPort}\r\nProxy-Connection: Keep-Alive\r\n\r\n`));
|
||||
socket.on("error", (error) => finish(tunnelEstablished ? 69 : 66, error?.message || String(error)));
|
||||
socket.on("close", () => finish(tunnelEstablished ? 0 : 68, tunnelEstablished ? null : "proxy closed before CONNECT completed"));
|
||||
socket.on("data", function onData(chunk) {
|
||||
buffer = Buffer.concat([buffer, chunk]);
|
||||
const headerEnd = buffer.indexOf("\r\n\r\n");
|
||||
if (headerEnd === -1 && buffer.length < 8192) return;
|
||||
if (headerEnd === -1) {
|
||||
socket.destroy();
|
||||
finish(68, "proxy response header exceeded 8192 bytes");
|
||||
return;
|
||||
}
|
||||
const statusLine = buffer.subarray(0, headerEnd).toString("latin1").split("\r\n", 1)[0] || "";
|
||||
const statusCode = Number.parseInt(statusLine.split(" ")[1] || "", 10);
|
||||
if (!statusLine.startsWith("HTTP/1.") || !Number.isInteger(statusCode) || statusCode < 200 || statusCode > 299) {
|
||||
socket.destroy();
|
||||
finish(67, `proxy CONNECT failed: ${statusLine.replace(/[^\x20-\x7e]/g, "?").slice(0, 160)}`);
|
||||
return;
|
||||
}
|
||||
socket.removeListener("data", onData);
|
||||
socket.setTimeout(0);
|
||||
tunnelEstablished = true;
|
||||
const rest = buffer.subarray(headerEnd + 4);
|
||||
if (rest.length > 0) process.stdout.write(rest);
|
||||
process.stdin.on("error", () => {});
|
||||
process.stdout.on("error", () => {});
|
||||
process.stdin.pipe(socket);
|
||||
socket.pipe(process.stdout);
|
||||
});
|
||||
@@ -0,0 +1,593 @@
|
||||
#!/usr/bin/env node
|
||||
import { createHash, randomBytes } from "node:crypto";
|
||||
import { spawn } from "node:child_process";
|
||||
import {
|
||||
existsSync,
|
||||
mkdirSync,
|
||||
readFileSync,
|
||||
renameSync,
|
||||
rmSync,
|
||||
writeFileSync,
|
||||
} from "node:fs";
|
||||
import { dirname, join, resolve, sep } from "node:path";
|
||||
|
||||
const args = process.argv.slice(2);
|
||||
const configPath = optionValue("--config");
|
||||
const mode = args.includes("--health")
|
||||
? "health"
|
||||
: args.includes("--status")
|
||||
? "status"
|
||||
: args.includes("--once")
|
||||
? "once"
|
||||
: "run";
|
||||
|
||||
if (!configPath) failUsage("--config <path> is required");
|
||||
|
||||
const config = readConfig(configPath);
|
||||
const statePath = join(config.cacheMountPath, config.stateDirectory, "status.json");
|
||||
const activeChildren = new Set();
|
||||
let stopping = false;
|
||||
|
||||
process.on("SIGTERM", beginShutdown);
|
||||
process.on("SIGINT", beginShutdown);
|
||||
|
||||
if (mode === "health" || mode === "status") {
|
||||
const view = await currentStatus(config, statePath);
|
||||
process.stdout.write(`${JSON.stringify(view)}\n`);
|
||||
process.exit(mode === "health" && view.ready !== true ? 1 : 0);
|
||||
}
|
||||
|
||||
if (mode === "once") {
|
||||
const state = await reconcileOnce(config, statePath);
|
||||
process.stdout.write(`${JSON.stringify(compactStatus(config, state))}\n`);
|
||||
process.exit(state.ready === true ? 0 : 1);
|
||||
}
|
||||
|
||||
log({ event: "managed-repository-reconciler-started", node: config.nodeId, lane: config.lane, desiredHash: config.desiredHash, repositoryCount: config.repositories.length });
|
||||
while (!stopping) {
|
||||
const state = await reconcileOnce(config, statePath);
|
||||
log({
|
||||
event: "managed-repository-reconcile-completed",
|
||||
node: config.nodeId,
|
||||
lane: config.lane,
|
||||
ready: state.ready,
|
||||
lastCycleOk: state.lastCycleOk,
|
||||
repositoryCount: state.repositories.length,
|
||||
desiredHash: config.desiredHash,
|
||||
});
|
||||
if (!stopping) await sleep(config.reconcileIntervalMs);
|
||||
}
|
||||
log({ event: "managed-repository-reconciler-stopped", node: config.nodeId, lane: config.lane });
|
||||
|
||||
function optionValue(name) {
|
||||
const index = args.indexOf(name);
|
||||
if (index < 0) return null;
|
||||
const value = args[index + 1];
|
||||
if (!value || value.startsWith("--")) failUsage(`${name} requires a value`);
|
||||
return value;
|
||||
}
|
||||
|
||||
function failUsage(message) {
|
||||
process.stderr.write(`${JSON.stringify({ ok: false, errorType: "invalid-arguments", message })}\n`);
|
||||
process.exit(64);
|
||||
}
|
||||
|
||||
function readConfig(path) {
|
||||
let raw;
|
||||
try {
|
||||
raw = JSON.parse(readFileSync(path, "utf8"));
|
||||
} catch {
|
||||
throw new Error("managed repository config is not readable JSON");
|
||||
}
|
||||
requireObject(raw, "config");
|
||||
const desiredHash = requireString(raw.desiredHash, "desiredHash");
|
||||
const { desiredHash: _omitted, ...desired } = raw;
|
||||
const computedHash = sha256(JSON.stringify(desired));
|
||||
if (computedHash !== desiredHash) throw new Error("managed repository config desiredHash does not match rendered content");
|
||||
if (raw.schemaVersion !== 1 || raw.kind !== "AgentRunManagedRepositoryReconciler") throw new Error("unsupported managed repository config envelope");
|
||||
const cacheMountPath = requireAbsolutePath(raw.cacheMountPath, "cacheMountPath");
|
||||
const stateDirectory = requireRelativePath(raw.stateDirectory, "stateDirectory");
|
||||
const repositories = requireArray(raw.repositories, "repositories").map((item, index) => {
|
||||
requireObject(item, `repositories[${index}]`);
|
||||
const key = requireSimpleId(item.key, `repositories[${index}].key`);
|
||||
const repository = requireRepository(item.repository, `repositories[${index}].repository`);
|
||||
const remote = requireString(item.remote, `repositories[${index}].remote`);
|
||||
const sourceBranch = requireGitRef(item.sourceBranch, `repositories[${index}].sourceBranch`);
|
||||
const desiredFingerprint = requireHex(item.desiredFingerprint, `repositories[${index}].desiredFingerprint`);
|
||||
const expectedFingerprint = sha256([key, repository, remote, sourceBranch].join("\0"));
|
||||
if (desiredFingerprint !== expectedFingerprint) throw new Error(`repositories[${index}].desiredFingerprint does not match repository desired state`);
|
||||
return { key, repository, remote, sourceBranch, desiredFingerprint };
|
||||
});
|
||||
if (repositories.length === 0) throw new Error("repositories must not be empty");
|
||||
ensureUnique(repositories.map((item) => item.key), "repository keys");
|
||||
ensureUnique(repositories.map((item) => item.repository), "repository paths");
|
||||
ensureUnique(repositories.map((item) => item.remote), "repository remotes");
|
||||
requireObject(raw.retry, "retry");
|
||||
requireObject(raw.freshness, "freshness");
|
||||
requireObject(raw.lifecycle, "lifecycle");
|
||||
if (raw.lifecycle.undeclaredRepositoryPolicy !== "retain" || raw.lifecycle.managedRefs !== "source-branch-only") {
|
||||
throw new Error("unsupported managed repository lifecycle policy");
|
||||
}
|
||||
return {
|
||||
schemaVersion: 1,
|
||||
kind: raw.kind,
|
||||
desiredHash,
|
||||
nodeId: requireString(raw.nodeId, "nodeId"),
|
||||
lane: requireString(raw.lane, "lane"),
|
||||
cacheMountPath,
|
||||
stateDirectory,
|
||||
reconcileIntervalMs: requirePositiveInteger(raw.reconcileIntervalMs, "reconcileIntervalMs"),
|
||||
fetchTimeoutMs: requirePositiveInteger(raw.fetchTimeoutMs, "fetchTimeoutMs"),
|
||||
shutdownGraceMs: requirePositiveInteger(raw.shutdownGraceMs, "shutdownGraceMs"),
|
||||
maxConcurrentRepositories: requirePositiveInteger(raw.maxConcurrentRepositories, "maxConcurrentRepositories"),
|
||||
retry: {
|
||||
maxAttempts: requirePositiveInteger(raw.retry.maxAttempts, "retry.maxAttempts"),
|
||||
initialDelayMs: requirePositiveInteger(raw.retry.initialDelayMs, "retry.initialDelayMs"),
|
||||
maxDelayMs: requirePositiveInteger(raw.retry.maxDelayMs, "retry.maxDelayMs"),
|
||||
},
|
||||
freshness: {
|
||||
maxAgeMs: requirePositiveInteger(raw.freshness.maxAgeMs, "freshness.maxAgeMs"),
|
||||
},
|
||||
lifecycle: {
|
||||
undeclaredRepositoryPolicy: "retain",
|
||||
managedRefs: "source-branch-only",
|
||||
},
|
||||
repositories,
|
||||
ssh: raw.ssh && typeof raw.ssh === "object" ? {
|
||||
identityFile: requireAbsolutePath(raw.ssh.identityFile, "ssh.identityFile"),
|
||||
knownHostsFile: requireAbsolutePath(raw.ssh.knownHostsFile, "ssh.knownHostsFile"),
|
||||
proxyConnectScript: requireAbsolutePath(raw.ssh.proxyConnectScript, "ssh.proxyConnectScript"),
|
||||
proxyHost: requireString(raw.ssh.proxyHost, "ssh.proxyHost"),
|
||||
proxyPort: requirePositiveInteger(raw.ssh.proxyPort, "ssh.proxyPort"),
|
||||
} : null,
|
||||
};
|
||||
}
|
||||
|
||||
async function reconcileOnce(cfg, path) {
|
||||
mkdirSync(dirname(path), { recursive: true });
|
||||
const previous = readState(path);
|
||||
const previousByFingerprint = new Map(
|
||||
Array.isArray(previous?.repositories)
|
||||
? previous.repositories.map((item) => [item?.desiredFingerprint, item])
|
||||
: [],
|
||||
);
|
||||
const results = await mapConcurrent(cfg.repositories, cfg.maxConcurrentRepositories, async (repository) => {
|
||||
const prior = previousByFingerprint.get(repository.desiredFingerprint) ?? null;
|
||||
return reconcileWithRetry(cfg, repository, prior);
|
||||
});
|
||||
const now = new Date().toISOString();
|
||||
const repositories = results.map((item) => withFreshness(item, cfg.freshness.maxAgeMs));
|
||||
const state = {
|
||||
schemaVersion: 1,
|
||||
kind: "AgentRunManagedRepositoryReconcilerStatus",
|
||||
desiredHash: cfg.desiredHash,
|
||||
nodeId: cfg.nodeId,
|
||||
lane: cfg.lane,
|
||||
observedAt: now,
|
||||
heartbeatAt: now,
|
||||
ready: repositories.every((item) => item.fresh === true && item.refPresent === true),
|
||||
lastCycleOk: repositories.every((item) => item.lastAttemptOk === true),
|
||||
lifecycle: cfg.lifecycle,
|
||||
repositories,
|
||||
valuesPrinted: false,
|
||||
};
|
||||
writeState(path, state);
|
||||
return state;
|
||||
}
|
||||
|
||||
async function reconcileWithRetry(cfg, repository, prior) {
|
||||
let failure = null;
|
||||
for (let attempt = 1; attempt <= cfg.retry.maxAttempts && !stopping; attempt += 1) {
|
||||
try {
|
||||
const result = await materializeRepository(cfg, repository);
|
||||
return {
|
||||
...repositoryIdentity(repository),
|
||||
refPresent: true,
|
||||
commit: result.commit,
|
||||
lastAttemptAt: new Date().toISOString(),
|
||||
lastAttemptOk: true,
|
||||
lastSuccessAt: new Date().toISOString(),
|
||||
attempts: attempt,
|
||||
errorCode: null,
|
||||
errorFingerprint: null,
|
||||
};
|
||||
} catch (error) {
|
||||
failure = safeError(error);
|
||||
log({ event: "managed-repository-reconcile-attempt-failed", repository: repository.repository, sourceBranch: repository.sourceBranch, attempt, errorCode: failure.code, errorFingerprint: failure.fingerprint });
|
||||
if (attempt < cfg.retry.maxAttempts && !stopping) {
|
||||
const delayMs = Math.min(cfg.retry.maxDelayMs, cfg.retry.initialDelayMs * (2 ** (attempt - 1)));
|
||||
await sleep(delayMs);
|
||||
}
|
||||
}
|
||||
}
|
||||
const inspected = await inspectRepository(cfg, repository);
|
||||
const preserveSuccess = prior
|
||||
&& prior.desiredFingerprint === repository.desiredFingerprint
|
||||
&& prior.commit === inspected.commit
|
||||
&& typeof prior.lastSuccessAt === "string";
|
||||
return {
|
||||
...repositoryIdentity(repository),
|
||||
refPresent: inspected.refPresent,
|
||||
commit: inspected.commit,
|
||||
lastAttemptAt: new Date().toISOString(),
|
||||
lastAttemptOk: false,
|
||||
lastSuccessAt: preserveSuccess ? prior.lastSuccessAt : null,
|
||||
attempts: cfg.retry.maxAttempts,
|
||||
errorCode: failure?.code ?? (stopping ? "shutdown" : "reconcile-failed"),
|
||||
errorFingerprint: failure?.fingerprint ?? null,
|
||||
};
|
||||
}
|
||||
|
||||
async function materializeRepository(cfg, repository) {
|
||||
const finalPath = repositoryPath(cfg, repository);
|
||||
mkdirSync(dirname(finalPath), { recursive: true });
|
||||
const existing = existsSync(finalPath);
|
||||
if (existing && !(await isBareRepository(finalPath, cfg.fetchTimeoutMs))) {
|
||||
throw codedError("invalid-repository-path", "declared repository path exists but is not a bare Git repository");
|
||||
}
|
||||
const workPath = existing
|
||||
? finalPath
|
||||
: join(dirname(finalPath), `.${repository.key}.managed-tmp-${process.pid}-${randomBytes(6).toString("hex")}`);
|
||||
const stageRef = `refs/agentrun-managed/stage/${repository.key}`;
|
||||
try {
|
||||
if (!existing) await git(cfg, ["init", "--bare", workPath]);
|
||||
await git(cfg, ["--git-dir", workPath, "config", "uploadpack.allowReachableSHA1InWant", "true"]);
|
||||
await git(cfg, ["--git-dir", workPath, "config", "uploadpack.allowAnySHA1InWant", "true"]);
|
||||
await git(cfg, ["--git-dir", workPath, "config", "http.uploadpack", "true"]);
|
||||
await git(cfg, ["--git-dir", workPath, "fetch", "--no-tags", repository.remote, `+refs/heads/${repository.sourceBranch}:${stageRef}`]);
|
||||
const commit = (await git(cfg, ["--git-dir", workPath, "rev-parse", "--verify", `${stageRef}^{commit}`])).stdout.trim();
|
||||
if (!/^[0-9a-f]{40}$/u.test(commit)) throw codedError("invalid-source-commit", "fetched source ref did not resolve to a commit");
|
||||
await git(cfg, ["--git-dir", workPath, "update-ref", `refs/heads/${repository.sourceBranch}`, commit]);
|
||||
await git(cfg, ["--git-dir", workPath, "update-ref", "-d", stageRef]);
|
||||
await git(cfg, ["--git-dir", workPath, "update-server-info"]);
|
||||
if (!existing) {
|
||||
await git(cfg, ["--git-dir", workPath, "symbolic-ref", "HEAD", `refs/heads/${repository.sourceBranch}`]);
|
||||
renameSync(workPath, finalPath);
|
||||
}
|
||||
return { commit };
|
||||
} catch (error) {
|
||||
if (existing) {
|
||||
try { await git(cfg, ["--git-dir", workPath, "update-ref", "-d", stageRef]); } catch {}
|
||||
} else {
|
||||
rmSync(workPath, { recursive: true, force: true });
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async function inspectRepository(cfg, repository) {
|
||||
const path = repositoryPath(cfg, repository);
|
||||
if (!existsSync(path) || !(await isBareRepository(path, Math.min(cfg.fetchTimeoutMs, 10000)))) return { refPresent: false, commit: null };
|
||||
try {
|
||||
const result = await run("git", ["--git-dir", path, "rev-parse", "--verify", `refs/heads/${repository.sourceBranch}^{commit}`], { timeoutMs: Math.min(cfg.fetchTimeoutMs, 10000), env: process.env });
|
||||
const commit = result.stdout.trim();
|
||||
return { refPresent: /^[0-9a-f]{40}$/u.test(commit), commit: /^[0-9a-f]{40}$/u.test(commit) ? commit : null };
|
||||
} catch {
|
||||
return { refPresent: false, commit: null };
|
||||
}
|
||||
}
|
||||
|
||||
async function isBareRepository(path, timeoutMs) {
|
||||
try {
|
||||
const result = await run("git", ["--git-dir", path, "rev-parse", "--is-bare-repository"], { timeoutMs, env: process.env });
|
||||
return result.stdout.trim() === "true";
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
async function git(cfg, gitArgs) {
|
||||
const env = { ...process.env, GIT_TERMINAL_PROMPT: "0" };
|
||||
if (cfg.ssh) {
|
||||
mkdirSync(dirname(cfg.ssh.knownHostsFile), { recursive: true });
|
||||
const proxyCommand = `node ${shellWord(cfg.ssh.proxyConnectScript)} ${shellWord(cfg.ssh.proxyHost)} ${cfg.ssh.proxyPort} %h %p`;
|
||||
env.GIT_SSH_COMMAND = [
|
||||
"ssh",
|
||||
"-i", shellWord(cfg.ssh.identityFile),
|
||||
"-o", "IdentitiesOnly=yes",
|
||||
"-o", "BatchMode=yes",
|
||||
"-o", "StrictHostKeyChecking=accept-new",
|
||||
"-o", `UserKnownHostsFile=${shellWord(cfg.ssh.knownHostsFile)}`,
|
||||
"-o", "ConnectTimeout=15",
|
||||
"-o", `ProxyCommand=${shellWord(proxyCommand)}`,
|
||||
].join(" ");
|
||||
}
|
||||
return run("git", gitArgs, { timeoutMs: cfg.fetchTimeoutMs, env });
|
||||
}
|
||||
|
||||
async function currentStatus(cfg, path) {
|
||||
const state = readState(path);
|
||||
if (!state || state.desiredHash !== cfg.desiredHash || !Array.isArray(state.repositories)) {
|
||||
return {
|
||||
ok: false,
|
||||
ready: false,
|
||||
desiredHash: cfg.desiredHash,
|
||||
observedDesiredHash: state?.desiredHash ?? null,
|
||||
reason: state ? "desired-state-mismatch" : "status-missing",
|
||||
repositoryCount: cfg.repositories.length,
|
||||
valuesPrinted: false,
|
||||
};
|
||||
}
|
||||
const stateByFingerprint = new Map(state.repositories.map((item) => [item?.desiredFingerprint, item]));
|
||||
const repositories = [];
|
||||
for (const desired of cfg.repositories) {
|
||||
const observed = stateByFingerprint.get(desired.desiredFingerprint) ?? null;
|
||||
const inspected = await inspectRepository(cfg, desired);
|
||||
const desiredIdentity = repositoryIdentity(desired);
|
||||
const item = withFreshness({
|
||||
...desiredIdentity,
|
||||
remoteFingerprint: observed?.remoteFingerprint ?? null,
|
||||
remoteFingerprintAligned: observed?.remoteFingerprint === desiredIdentity.remoteFingerprint,
|
||||
refPresent: inspected.refPresent,
|
||||
commit: inspected.commit,
|
||||
lastAttemptAt: observed?.lastAttemptAt ?? null,
|
||||
lastAttemptOk: observed?.lastAttemptOk === true,
|
||||
lastSuccessAt: observed?.commit === inspected.commit ? observed?.lastSuccessAt ?? null : null,
|
||||
attempts: Number.isInteger(observed?.attempts) ? observed.attempts : null,
|
||||
errorCode: observed?.errorCode ?? null,
|
||||
errorFingerprint: observed?.errorFingerprint ?? null,
|
||||
}, cfg.freshness.maxAgeMs);
|
||||
repositories.push(item);
|
||||
}
|
||||
const heartbeatAgeMs = isoAgeMs(state.heartbeatAt);
|
||||
const heartbeatFresh = heartbeatAgeMs !== null && heartbeatAgeMs <= cfg.freshness.maxAgeMs;
|
||||
const ready = heartbeatFresh && repositories.every((item) => item.refPresent === true && item.fresh === true && item.remoteFingerprintAligned === true);
|
||||
return {
|
||||
ok: ready,
|
||||
ready,
|
||||
desiredHash: cfg.desiredHash,
|
||||
observedDesiredHash: state.desiredHash,
|
||||
nodeId: cfg.nodeId,
|
||||
lane: cfg.lane,
|
||||
heartbeatAt: state.heartbeatAt ?? null,
|
||||
heartbeatAgeMs,
|
||||
heartbeatFresh,
|
||||
lifecycle: cfg.lifecycle,
|
||||
repositories,
|
||||
valuesPrinted: false,
|
||||
};
|
||||
}
|
||||
|
||||
function compactStatus(cfg, state) {
|
||||
return {
|
||||
ok: state.ready === true,
|
||||
ready: state.ready === true,
|
||||
desiredHash: cfg.desiredHash,
|
||||
nodeId: cfg.nodeId,
|
||||
lane: cfg.lane,
|
||||
heartbeatAt: state.heartbeatAt,
|
||||
lifecycle: cfg.lifecycle,
|
||||
repositories: state.repositories,
|
||||
valuesPrinted: false,
|
||||
};
|
||||
}
|
||||
|
||||
function repositoryIdentity(repository) {
|
||||
return {
|
||||
key: repository.key,
|
||||
repository: repository.repository,
|
||||
sourceBranch: repository.sourceBranch,
|
||||
desiredFingerprint: repository.desiredFingerprint,
|
||||
remoteFingerprint: `sha256:${sha256(repository.remote)}`,
|
||||
};
|
||||
}
|
||||
|
||||
function withFreshness(item, maxAgeMs) {
|
||||
const ageMs = isoAgeMs(item.lastSuccessAt);
|
||||
return {
|
||||
...item,
|
||||
lastSuccessAgeMs: ageMs,
|
||||
fresh: item.refPresent === true && ageMs !== null && ageMs <= maxAgeMs,
|
||||
};
|
||||
}
|
||||
|
||||
function repositoryPath(cfg, repository) {
|
||||
const root = resolve(cfg.cacheMountPath);
|
||||
const path = resolve(root, `${repository.repository}.git`);
|
||||
if (path !== root && !path.startsWith(`${root}${sep}`)) throw codedError("unsafe-repository-path", "repository path escapes cache root");
|
||||
return path;
|
||||
}
|
||||
|
||||
function readState(path) {
|
||||
try {
|
||||
return JSON.parse(readFileSync(path, "utf8"));
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
function writeState(path, state) {
|
||||
mkdirSync(dirname(path), { recursive: true });
|
||||
const temporary = `${path}.tmp-${process.pid}-${randomBytes(4).toString("hex")}`;
|
||||
writeFileSync(temporary, `${JSON.stringify(state)}\n`, { mode: 0o600 });
|
||||
renameSync(temporary, path);
|
||||
}
|
||||
|
||||
async function mapConcurrent(items, concurrency, worker) {
|
||||
const results = new Array(items.length);
|
||||
let index = 0;
|
||||
async function consume() {
|
||||
while (!stopping) {
|
||||
const current = index;
|
||||
index += 1;
|
||||
if (current >= items.length) return;
|
||||
results[current] = await worker(items[current], current);
|
||||
}
|
||||
}
|
||||
await Promise.all(Array.from({ length: Math.min(concurrency, items.length) }, () => consume()));
|
||||
for (let current = 0; current < results.length; current += 1) {
|
||||
if (results[current] === undefined) {
|
||||
const item = items[current];
|
||||
results[current] = {
|
||||
...repositoryIdentity(item),
|
||||
refPresent: false,
|
||||
commit: null,
|
||||
lastAttemptAt: new Date().toISOString(),
|
||||
lastAttemptOk: false,
|
||||
lastSuccessAt: null,
|
||||
attempts: 0,
|
||||
errorCode: "shutdown",
|
||||
errorFingerprint: null,
|
||||
};
|
||||
}
|
||||
}
|
||||
return results;
|
||||
}
|
||||
|
||||
function run(command, commandArgs, options) {
|
||||
return new Promise((resolvePromise, rejectPromise) => {
|
||||
const child = spawn(command, commandArgs, { env: options.env, stdio: ["ignore", "pipe", "pipe"] });
|
||||
activeChildren.add(child);
|
||||
let stdout = "";
|
||||
let stderr = "";
|
||||
let timedOut = false;
|
||||
const timer = setTimeout(() => {
|
||||
timedOut = true;
|
||||
child.kill("SIGKILL");
|
||||
}, options.timeoutMs);
|
||||
child.stdout.on("data", (chunk) => { if (stdout.length < 32768) stdout += chunk.toString("utf8"); });
|
||||
child.stderr.on("data", (chunk) => { if (stderr.length < 32768) stderr += chunk.toString("utf8"); });
|
||||
child.on("error", (error) => {
|
||||
clearTimeout(timer);
|
||||
activeChildren.delete(child);
|
||||
rejectPromise(codedError("command-start-failed", error.message));
|
||||
});
|
||||
child.on("close", (code, signal) => {
|
||||
clearTimeout(timer);
|
||||
activeChildren.delete(child);
|
||||
if (timedOut) return rejectPromise(codedError("fetch-timeout", `${command} exceeded timeout`));
|
||||
if (code !== 0) return rejectPromise(codedError("command-failed", `${command} exited ${code ?? signal}: ${stderr || stdout}`));
|
||||
resolvePromise({ stdout, stderr });
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
function beginShutdown() {
|
||||
if (stopping) return;
|
||||
stopping = true;
|
||||
const timer = setTimeout(() => {
|
||||
for (const child of activeChildren) child.kill("SIGKILL");
|
||||
}, config.shutdownGraceMs);
|
||||
timer.unref();
|
||||
}
|
||||
|
||||
function sleep(ms) {
|
||||
if (stopping) return Promise.resolve();
|
||||
return new Promise((resolvePromise) => {
|
||||
let settled = false;
|
||||
const finish = () => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
clearTimeout(timer);
|
||||
clearInterval(check);
|
||||
resolvePromise();
|
||||
};
|
||||
const timer = setTimeout(finish, ms);
|
||||
const check = setInterval(() => {
|
||||
if (stopping) finish();
|
||||
}, Math.min(ms, 100));
|
||||
});
|
||||
}
|
||||
|
||||
function safeError(error) {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
const lower = message.toLowerCase();
|
||||
const code = error?.code && typeof error.code === "string" && error.code !== "command-failed"
|
||||
? error.code
|
||||
: lower.includes("couldn't find remote ref") || lower.includes("remote ref does not exist")
|
||||
? "source-ref-missing"
|
||||
: lower.includes("repository not found") || lower.includes("not found")
|
||||
? "repository-not-found"
|
||||
: lower.includes("permission denied") || lower.includes("publickey") || lower.includes("authentication")
|
||||
? "authentication-failed"
|
||||
: lower.includes("timeout") || lower.includes("timed out")
|
||||
? "fetch-timeout"
|
||||
: "fetch-failed";
|
||||
return { code, fingerprint: sha256(message).slice(0, 16) };
|
||||
}
|
||||
|
||||
function codedError(code, message) {
|
||||
const error = new Error(message);
|
||||
error.code = code;
|
||||
return error;
|
||||
}
|
||||
|
||||
function isoAgeMs(value) {
|
||||
if (typeof value !== "string") return null;
|
||||
const time = Date.parse(value);
|
||||
return Number.isFinite(time) ? Math.max(0, Date.now() - time) : null;
|
||||
}
|
||||
|
||||
function shellWord(value) {
|
||||
return `'${String(value).replaceAll("'", `'"'"'`)}'`;
|
||||
}
|
||||
|
||||
function sha256(value) {
|
||||
return createHash("sha256").update(value).digest("hex");
|
||||
}
|
||||
|
||||
function log(value) {
|
||||
process.stdout.write(`${JSON.stringify({ ...value, valuesPrinted: false })}\n`);
|
||||
}
|
||||
|
||||
function requireObject(value, path) {
|
||||
if (!value || typeof value !== "object" || Array.isArray(value)) throw new Error(`${path} must be an object`);
|
||||
return value;
|
||||
}
|
||||
|
||||
function requireArray(value, path) {
|
||||
if (!Array.isArray(value)) throw new Error(`${path} must be an array`);
|
||||
return value;
|
||||
}
|
||||
|
||||
function requireString(value, path) {
|
||||
if (typeof value !== "string" || value.length === 0) throw new Error(`${path} must be a non-empty string`);
|
||||
return value;
|
||||
}
|
||||
|
||||
function requirePositiveInteger(value, path) {
|
||||
if (!Number.isInteger(value) || value <= 0) throw new Error(`${path} must be a positive integer`);
|
||||
return value;
|
||||
}
|
||||
|
||||
function requireAbsolutePath(value, path) {
|
||||
const text = requireString(value, path);
|
||||
if (!text.startsWith("/") || text.includes("..")) throw new Error(`${path} must be an absolute path without ..`);
|
||||
return text;
|
||||
}
|
||||
|
||||
function requireRelativePath(value, path) {
|
||||
const text = requireString(value, path);
|
||||
if (text.startsWith("/") || text.includes("..") || text === ".") throw new Error(`${path} must be a dedicated relative path without ..`);
|
||||
return text;
|
||||
}
|
||||
|
||||
function requireSimpleId(value, path) {
|
||||
const text = requireString(value, path);
|
||||
if (!/^[A-Za-z0-9._-]+$/u.test(text)) throw new Error(`${path} must use a simple id`);
|
||||
return text;
|
||||
}
|
||||
|
||||
function requireRepository(value, path) {
|
||||
const text = requireString(value, path);
|
||||
if (!/^[A-Za-z0-9._-]+\/[A-Za-z0-9._-]+$/u.test(text) || text.includes("..")) throw new Error(`${path} must be owner/name without ..`);
|
||||
return text;
|
||||
}
|
||||
|
||||
function requireGitRef(value, path) {
|
||||
const text = requireString(value, path);
|
||||
if (text.startsWith("-") || text.startsWith("/") || text.endsWith("/") || text.endsWith(".") || text.includes("..") || text.includes("@{") || /[\u0000-\u0020~^:?*[\\]/u.test(text)) {
|
||||
throw new Error(`${path} must be a safe Git ref name`);
|
||||
}
|
||||
return text;
|
||||
}
|
||||
|
||||
function requireHex(value, path) {
|
||||
const text = requireString(value, path);
|
||||
if (!/^[0-9a-f]{64}$/u.test(text)) throw new Error(`${path} must be a sha256 fingerprint`);
|
||||
return text;
|
||||
}
|
||||
|
||||
function ensureUnique(values, path) {
|
||||
if (new Set(values).size !== values.length) throw new Error(`${path} must be unique`);
|
||||
}
|
||||
Reference in New Issue
Block a user