feat(lanes): per-lane database, Redis, and .env isolation (A1+A2)
Gives each lane its own slot-derived runtime (ports, detached process lifecycle, profile-driven hooks) and its own database/Redis logical index/.env file, so two lanes running the same repo's stack at once no longer share state. Machine-level DB/Redis credentials live at ~/.ccam/secrets.env (mode 0600, never returned by any route); a hook's output is redacted of that password (raw and URL-encoded forms) before it reaches a log file or the lane_hook_output websocket broadcast. Wired into provision/up/reset/remove; reset accepts --keep-db to skip the drop/recreate/migrate/reseed block entirely.
This commit is contained in:
@@ -0,0 +1,532 @@
|
||||
/**
|
||||
* @file Lane stack lifecycle: bring a lane's services up through its profile's
|
||||
* hooks, take them down, and report what is actually running. Also owns the
|
||||
* data-isolation lifecycle (A2): `provisionLane`/`resetLaneData`/`removeLaneData`
|
||||
* seed `.env`, create/drop the lane's own database, and run `bootstrap`/
|
||||
* `migrate`/`seed` at the points Shipyard's `lane-bootstrap.sh`/`lane-up.sh`/
|
||||
* `lane-reset.sh`/`lane-remove.sh` do — CCAM's version of "ports came from A1,
|
||||
* everything else a lane needs to run its own stack comes from here."
|
||||
*
|
||||
* Two properties shape everything here.
|
||||
*
|
||||
* Services are FULLY DETACHED (see `harness_spawn` in lane-profile.js), so a
|
||||
* lane's stack outlives both the hook that started it and the dashboard itself —
|
||||
* restarting or updating CCAM must never kill work in progress.
|
||||
*
|
||||
* Liveness is COMPUTED, never stored. Whether a stack is up is not a fact CCAM
|
||||
* controls: a process dies to OOM, to a stray `kill`, to a reboot. A cached
|
||||
* "running" flag would be wrong from that moment until something noticed, so
|
||||
* `runtimeFacts` re-derives it from pid files and port probes on every read. The
|
||||
* payoff is that adopting a stack after a dashboard restart needs no code at all.
|
||||
*
|
||||
* This module writes `slot` and `ports` and nothing else on the lane row. It never
|
||||
* writes `stage`, `status` or `notes`: in CCAM those describe the AGENT's work, not
|
||||
* the stack's state, and conflating "a server is listening" with "an agent is
|
||||
* running" would corrupt lane liveness. Boot failures live in `last-error.json`
|
||||
* and surface through `runtimeFacts`.
|
||||
* @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
|
||||
*/
|
||||
|
||||
const fs = require("node:fs");
|
||||
const path = require("node:path");
|
||||
const { execFile } = require("node:child_process");
|
||||
const { promisify } = require("node:util");
|
||||
|
||||
const lanesLib = require("./lanes");
|
||||
const {
|
||||
allocateSlot,
|
||||
resolvePorts,
|
||||
slotDirs,
|
||||
portsSteppedAside,
|
||||
portBase,
|
||||
dbName,
|
||||
} = require("./lane-slots");
|
||||
const { resolveProfile, profileSearchPaths, runHook } = require("./lane-profile");
|
||||
const { isListening, listenerPids } = require("./ports");
|
||||
const { seedEnv } = require("./lane-env");
|
||||
const { ensureDatabase, dropDatabase } = require("./lane-services");
|
||||
const { readSecrets } = require("./secrets");
|
||||
|
||||
const execFileAsync = promisify(execFile);
|
||||
|
||||
/** A health hook that never returns is a failed boot, not an eternal wait. */
|
||||
const HEALTH_TIMEOUT_MS = Number(process.env.LANE_HEALTH_TIMEOUT_MS) || 180_000;
|
||||
const BOOT_TIMEOUT_MS = Number(process.env.LANE_BOOT_TIMEOUT_MS) || 900_000;
|
||||
|
||||
/** Resolve a lane's profile or throw the error a route turns into a 400. */
|
||||
function requireProfile(lane) {
|
||||
const profile = resolveProfile(lane);
|
||||
if (!profile) {
|
||||
throw Object.assign(
|
||||
new Error(`lane has no .ccam/profile — looked in: ${profileSearchPaths(lane).join(", ")}`),
|
||||
{ code: "ENOPROFILE", searched: profileSearchPaths(lane) }
|
||||
);
|
||||
}
|
||||
return profile;
|
||||
}
|
||||
|
||||
/** Direct children of a pid. Empty when pgrep is unavailable — the parent still dies. */
|
||||
async function childPids(pid) {
|
||||
try {
|
||||
const { stdout } = await execFileAsync("pgrep", ["-P", String(pid)]);
|
||||
return stdout
|
||||
.split("\n")
|
||||
.map((line) => line.trim())
|
||||
.filter((line) => /^\d+$/.test(line))
|
||||
.map(Number);
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Kill a process and every descendant, deepest first.
|
||||
*
|
||||
* Bottom-up matters: killing the parent first reparents its children to init,
|
||||
* where nothing knows to look for them. This is the failure `lane-down.sh`'s
|
||||
* `kill_tree` was written for — a uvicorn reloader or a celery prefork pool whose
|
||||
* workers survived a kill aimed at the recorded pid, kept the port bound, and made
|
||||
* the next boot fail for a reason nothing reported.
|
||||
*
|
||||
* @param {number} pid - Root of the tree.
|
||||
* @returns {Promise<number[]>} Pids signalled.
|
||||
*/
|
||||
async function killTree(pid) {
|
||||
const killed = [];
|
||||
for (const child of await childPids(pid)) {
|
||||
killed.push(...(await killTree(child)));
|
||||
}
|
||||
try {
|
||||
process.kill(pid, "SIGKILL");
|
||||
killed.push(pid);
|
||||
} catch {
|
||||
/* already gone */
|
||||
}
|
||||
return killed;
|
||||
}
|
||||
|
||||
/** True when a pid exists and we may signal it. */
|
||||
function isAlive(pid) {
|
||||
try {
|
||||
process.kill(pid, 0);
|
||||
return true;
|
||||
} catch (err) {
|
||||
// EPERM means it exists but belongs to someone else — still alive.
|
||||
return err.code === "EPERM";
|
||||
}
|
||||
}
|
||||
|
||||
/** Recorded services: every `<name>.pid` the boot hook left in the run directory. */
|
||||
function readPidFiles(runDir) {
|
||||
let entries = [];
|
||||
try {
|
||||
entries = fs.readdirSync(runDir);
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
const out = [];
|
||||
for (const entry of entries) {
|
||||
if (!entry.endsWith(".pid")) continue;
|
||||
const raw = fs.readFileSync(path.join(runDir, entry), "utf8").trim();
|
||||
if (!/^\d+$/.test(raw)) continue;
|
||||
out.push({ name: entry.slice(0, -4), pid: Number(raw), file: path.join(runDir, entry) });
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/** Persist why a boot failed, where runtimeFacts can find it. */
|
||||
function recordError(lane, error) {
|
||||
const { stateDir, errorFile } = slotDirs(lane.slot);
|
||||
fs.mkdirSync(stateDir, { recursive: true });
|
||||
fs.writeFileSync(
|
||||
errorFile,
|
||||
JSON.stringify(
|
||||
{ at: new Date().toISOString(), code: error.code || null, message: error.message },
|
||||
null,
|
||||
2
|
||||
)
|
||||
);
|
||||
}
|
||||
|
||||
/** Drop a stale failure once a boot succeeds. */
|
||||
function clearError(lane) {
|
||||
try {
|
||||
fs.rmSync(slotDirs(lane.slot).errorFile, { force: true });
|
||||
} catch {
|
||||
/* nothing recorded */
|
||||
}
|
||||
}
|
||||
|
||||
function readError(slot) {
|
||||
try {
|
||||
return JSON.parse(fs.readFileSync(slotDirs(slot).errorFile, "utf8"));
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve one `LANE_DIRS` entry against the lane's working copy, refusing
|
||||
* anything that escapes it (an absolute path, or one climbing out with `..`)
|
||||
* rather than silently touching a directory somewhere else on the machine.
|
||||
*/
|
||||
function resolveLaneDirEntry(lane, entry) {
|
||||
const target = path.resolve(lane.cwd, entry);
|
||||
const relative = path.relative(lane.cwd, target);
|
||||
if (relative.startsWith("..") || path.isAbsolute(relative)) {
|
||||
throw Object.assign(new Error(`LANE_DIRS entry escapes the lane: ${entry}`), {
|
||||
code: "EBADLANEDIR",
|
||||
entry,
|
||||
});
|
||||
}
|
||||
return target;
|
||||
}
|
||||
|
||||
/** Create the per-lane directories the profile declares. */
|
||||
function makeLaneDirs(lane, profile) {
|
||||
for (const entry of profile.laneDirs) {
|
||||
fs.mkdirSync(resolveLaneDirEntry(lane, entry), { recursive: true });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Empty and recreate the per-lane directories a `reset` should start fresh —
|
||||
* Shipyard's `rm -rf "$(lane_upload_dir "$N")"/*` generalized to every
|
||||
* declared `LANE_DIRS` entry, since a fresh feature should not inherit a
|
||||
* previous one's uploaded files.
|
||||
*/
|
||||
function clearLaneDirs(lane, profile) {
|
||||
for (const entry of profile.laneDirs) {
|
||||
const target = resolveLaneDirEntry(lane, entry);
|
||||
fs.rmSync(target, { recursive: true, force: true });
|
||||
fs.mkdirSync(target, { recursive: true });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Run a hook only when the profile declares it, throwing `errCode` on a
|
||||
* non-zero exit. Shared by every lifecycle step below (`bootstrap`, `migrate`,
|
||||
* `seed`) so "the profile never declared this" and "the hook failed" stay two
|
||||
* distinct, consistently-coded outcomes everywhere they're checked.
|
||||
*/
|
||||
async function runRequiredHook(lane, profile, name, options, errCode) {
|
||||
if (!profile.hooks.has(name)) return null;
|
||||
const result = await runHook(lane, profile, name, [], options);
|
||||
if (result.code !== 0) {
|
||||
throw Object.assign(new Error(`${name} hook exited ${result.code}`), {
|
||||
code: errCode,
|
||||
logPath: result.logPath,
|
||||
});
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
* Provision a freshly-created managed lane's data isolation: seed its `.env`,
|
||||
* run `bootstrap`/`migrate`/`seed`, and create its database. Runs once, right
|
||||
* after the worktree itself exists — the other half of isolation A1 did not
|
||||
* cover (ports and directories came from A1; database, `.env` and uploads
|
||||
* come from here). Every step degrades to nothing when the profile never
|
||||
* declares the underlying feature: no `bootstrap` hook, no `DB_PREFIX`, no
|
||||
* `ENV_FILES` are all normal, not partial failures.
|
||||
*
|
||||
* @param {object} lane - Lane row, freshly worktree-provisioned (kind "managed").
|
||||
* @param {{onLine?: Function}} [options]
|
||||
* @returns {Promise<object>} The lane row after provisioning.
|
||||
*/
|
||||
async function provisionLane(lane, options = {}) {
|
||||
const profile = requireProfile(lane);
|
||||
const { onLine } = options;
|
||||
|
||||
let current = lane;
|
||||
if (!current.slot) {
|
||||
allocateSlot(current.id);
|
||||
current = lanesLib.getLane(current.id);
|
||||
}
|
||||
|
||||
seedEnv(current, profile, readSecrets());
|
||||
await runRequiredHook(
|
||||
current,
|
||||
profile,
|
||||
"bootstrap",
|
||||
{ onLine, timeoutMs: BOOT_TIMEOUT_MS },
|
||||
"EBOOTSTRAPFAILED"
|
||||
);
|
||||
await ensureDatabase(current, profile, { onLine });
|
||||
await runRequiredHook(current, profile, "migrate", { onLine }, "EMIGRATEFAILED");
|
||||
await runRequiredHook(current, profile, "seed", { onLine }, "ESEEDFAILED");
|
||||
|
||||
return lanesLib.getLane(current.id);
|
||||
}
|
||||
|
||||
/**
|
||||
* Reset a lane's data isolation after `resetWorktree` has already put its
|
||||
* working copy back on the base branch: refresh `.env` (a `--force` refresh,
|
||||
* so it tracks whatever the base branch's source `.env` now says), re-run
|
||||
* `bootstrap` (a reset can land on a branch with different dependencies —
|
||||
* Shipyard's own comment: "deps move under node_modules/venv"), clear the
|
||||
* declared `LANE_DIRS`, and — unless `keepDb` — drop, recreate, migrate and
|
||||
* reseed the database. `keepDb` skips that whole block, not just the drop:
|
||||
* a caller who wants to keep their data wants it left alone, migrations
|
||||
* included.
|
||||
*
|
||||
* @param {object} lane - Lane row, after `resetWorktree`.
|
||||
* @param {object} profile - Resolved profile.
|
||||
* @param {{keepDb?: boolean, onLine?: Function}} [options]
|
||||
*/
|
||||
async function resetLaneData(lane, profile, options = {}) {
|
||||
const { keepDb = false, onLine } = options;
|
||||
|
||||
seedEnv(lane, profile, readSecrets(), { force: true });
|
||||
await runRequiredHook(
|
||||
lane,
|
||||
profile,
|
||||
"bootstrap",
|
||||
{ onLine, timeoutMs: BOOT_TIMEOUT_MS },
|
||||
"EBOOTSTRAPFAILED"
|
||||
);
|
||||
clearLaneDirs(lane, profile);
|
||||
|
||||
if (keepDb) return;
|
||||
if (!dbName(profile, lane.slot)) return;
|
||||
await dropDatabase(lane, profile, dbName(profile, lane.slot), { onLine });
|
||||
await ensureDatabase(lane, profile, { onLine });
|
||||
await runRequiredHook(lane, profile, "migrate", { onLine }, "EMIGRATEFAILED");
|
||||
await runRequiredHook(lane, profile, "seed", { onLine }, "ESEEDFAILED");
|
||||
}
|
||||
|
||||
/**
|
||||
* Drop a managed lane's database and its `_test` sibling before the rest of
|
||||
* removal tears down the worktree and state directory.
|
||||
*
|
||||
* A no-op for an adopted lane — its data was never CCAM's to create, so it is
|
||||
* never CCAM's to destroy either, the same invariant `assertDestroyable`
|
||||
* enforces for the worktree itself. Best-effort per database: mirrors
|
||||
* Shipyard's `dropdb --if-exists ... || true` — a failed drop is logged, not
|
||||
* thrown, because it must never block removing the lane's dashboard record.
|
||||
*
|
||||
* @param {object} lane - Lane row.
|
||||
* @param {object} profile - Resolved profile.
|
||||
* @param {{onLine?: Function}} [options]
|
||||
*/
|
||||
async function removeLaneData(lane, profile, options = {}) {
|
||||
if (lane.kind !== "managed") return;
|
||||
const name = dbName(profile, lane.slot);
|
||||
if (!name) return;
|
||||
for (const target of [name, `${name}_test`]) {
|
||||
try {
|
||||
await dropDatabase(lane, profile, target, options);
|
||||
} catch (err) {
|
||||
console.warn(`[lane-runtime] lane ${lane.id}: db-drop ${target} failed: ${err.message}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Boot a lane's stack.
|
||||
*
|
||||
* Caller MUST hold the lane lock: slot allocation reads-then-writes across the
|
||||
* `await` in port resolution, and two concurrent ups would otherwise both believe
|
||||
* they own the slot.
|
||||
*
|
||||
* Runs `boot` then `health`; deliberately NOT `bootstrap`, which belongs to
|
||||
* provisioning and reset (Shipyard's `lane-up.sh` draws the same line — installing
|
||||
* dependencies on every boot would make a routine restart minutes long).
|
||||
*
|
||||
* A failing health check leaves the processes running. They are the evidence: the
|
||||
* logs of a half-booted stack are what tells the user which service never came up,
|
||||
* and killing them to report a tidy failure destroys exactly that.
|
||||
*
|
||||
* @param {object} lane - Lane row.
|
||||
* @param {{build?: boolean, onLine?: Function}} [options]
|
||||
* @returns {Promise<object>} Runtime facts after the attempt.
|
||||
*/
|
||||
async function upLane(lane, options = {}) {
|
||||
const profile = requireProfile(lane);
|
||||
const { build = true, onLine } = options;
|
||||
|
||||
let current = lane;
|
||||
if (!current.slot) {
|
||||
allocateSlot(current.id);
|
||||
current = lanesLib.getLane(current.id);
|
||||
}
|
||||
|
||||
const ports = await resolvePorts(current, profile);
|
||||
lanesLib.setProvisioningFacts(current.id, { ports });
|
||||
current = lanesLib.getLane(current.id);
|
||||
|
||||
const dirs = slotDirs(current.slot);
|
||||
fs.mkdirSync(dirs.runDir, { recursive: true });
|
||||
fs.mkdirSync(dirs.logDir, { recursive: true });
|
||||
makeLaneDirs(current, profile);
|
||||
|
||||
// Defensive: a re-run on a live lane must not double-start services that then
|
||||
// fight over the same ports.
|
||||
await downLane(current, { profile });
|
||||
|
||||
try {
|
||||
// Repair .env on every boot (a lane whose file was hand-edited or never
|
||||
// seeded gets fixed here), then ensure the database exists — cheap when it
|
||||
// already does — migrate on every boot (schemas drift while a lane sits
|
||||
// idle), and seed only when this boot is the one that created the database.
|
||||
if (current.kind === "managed") seedEnv(current, profile, readSecrets());
|
||||
const db = await ensureDatabase(current, profile, { onLine });
|
||||
await runRequiredHook(current, profile, "migrate", { onLine }, "EMIGRATEFAILED");
|
||||
if (db.created) await runRequiredHook(current, profile, "seed", { onLine }, "ESEEDFAILED");
|
||||
|
||||
const boot = await runHook(current, profile, "boot", build ? [] : ["--no-build"], {
|
||||
onLine,
|
||||
timeoutMs: BOOT_TIMEOUT_MS,
|
||||
});
|
||||
if (boot.code !== 0) {
|
||||
throw Object.assign(new Error(`boot hook exited ${boot.code}`), {
|
||||
code: "EBOOTFAILED",
|
||||
logPath: boot.logPath,
|
||||
});
|
||||
}
|
||||
|
||||
if (profile.hooks.has("health")) {
|
||||
const health = await runHook(current, profile, "health", [], {
|
||||
onLine,
|
||||
timeoutMs: HEALTH_TIMEOUT_MS,
|
||||
});
|
||||
if (health.code !== 0) {
|
||||
throw Object.assign(new Error(`health check failed (exit ${health.code})`), {
|
||||
code: "EUNHEALTHY",
|
||||
logPath: health.logPath,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
clearError(current);
|
||||
} catch (err) {
|
||||
recordError(current, err);
|
||||
throw err;
|
||||
}
|
||||
|
||||
return runtimeFacts(lanesLib.getLane(current.id));
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop a lane's stack. Idempotent, and safe on a lane that was never up.
|
||||
*
|
||||
* Kills each recorded pid tree, then — only when there WAS something recorded —
|
||||
* sweeps any listener still holding the lane's ports. That condition is a
|
||||
* deliberate departure from Shipyard, which always sweeps: a lane whose stack is
|
||||
* already down still owns its port numbers, and if the user has since started
|
||||
* their own server on one, an unconditional sweep would kill it. Requiring a pid
|
||||
* file keeps the backstop for the case it exists to cover — a detached child that
|
||||
* outlived the parent we recorded — without ever reaching a stranger's process.
|
||||
*
|
||||
* @param {object} lane - Lane row.
|
||||
* @param {{profile?: object}} [options]
|
||||
* @returns {Promise<{killed: number[]}>}
|
||||
*/
|
||||
async function downLane(lane, options = {}) {
|
||||
if (!lane.slot) return { killed: [] };
|
||||
const { runDir } = slotDirs(lane.slot);
|
||||
const recorded = readPidFiles(runDir);
|
||||
const killed = [];
|
||||
|
||||
for (const service of recorded) {
|
||||
killed.push(...(await killTree(service.pid)));
|
||||
fs.rmSync(service.file, { force: true });
|
||||
}
|
||||
|
||||
if (recorded.length) {
|
||||
for (const port of Object.values(lane.ports || {})) {
|
||||
for (const pid of await listenerPids(port)) {
|
||||
if (!killed.includes(pid)) killed.push(...(await killTree(pid)));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return { killed };
|
||||
}
|
||||
|
||||
/**
|
||||
* What is actually running for this lane, computed fresh on every call.
|
||||
*
|
||||
* Follows the contract of `GET /api/lanes/:id/git`: a lane with no profile is a
|
||||
* normal state, reported as `{available: false}` rather than an error. Callers
|
||||
* probe ports and stat pid files here, which is why this is its own endpoint and
|
||||
* not part of the polled lane list.
|
||||
*
|
||||
* @param {object} lane - Lane row.
|
||||
* @returns {Promise<object>}
|
||||
*/
|
||||
async function runtimeFacts(lane) {
|
||||
const profile = resolveProfile(lane);
|
||||
if (!profile) return { available: false, searched: profileSearchPaths(lane) };
|
||||
if (!lane.slot) {
|
||||
return { available: true, provisioned: false, hooks: [...profile.hooks], ports: {} };
|
||||
}
|
||||
|
||||
const dirs = slotDirs(lane.slot);
|
||||
const services = readPidFiles(dirs.runDir).map((service) => ({
|
||||
name: service.name,
|
||||
pid: service.pid,
|
||||
alive: isAlive(service.pid),
|
||||
}));
|
||||
|
||||
const ports = {};
|
||||
for (const name of profile.ports) {
|
||||
const port = lane.ports?.[name] ?? null;
|
||||
ports[name] = {
|
||||
port,
|
||||
expected: portBase(profile, name) + lane.slot,
|
||||
listening: port ? await isListening(port) : false,
|
||||
};
|
||||
}
|
||||
|
||||
let logs = [];
|
||||
try {
|
||||
logs = fs.readdirSync(dirs.logDir).filter((entry) => entry.endsWith(".log"));
|
||||
} catch {
|
||||
/* never booted */
|
||||
}
|
||||
|
||||
// Names and an index, never a connection string: DATABASE_URL/REDIS_URL embed
|
||||
// the secrets.env password, and this object is exactly what GET /runtime
|
||||
// returns to a browser tab.
|
||||
const name = dbName(profile, lane.slot);
|
||||
const database = name ? { name, testName: `${name}_test` } : null;
|
||||
const redisIndex = profile.env.REDIS === "1" ? lane.slot : null;
|
||||
|
||||
return {
|
||||
available: true,
|
||||
provisioned: true,
|
||||
slot: lane.slot,
|
||||
kind: lane.kind,
|
||||
hooks: [...profile.hooks],
|
||||
profileDir: profile.dir,
|
||||
services,
|
||||
ports,
|
||||
database,
|
||||
redisIndex,
|
||||
steppedAside: portsSteppedAside(lane, profile),
|
||||
up: services.some((service) => service.alive),
|
||||
healthy:
|
||||
Object.values(ports).length > 0 && Object.values(ports).every((entry) => entry.listening),
|
||||
logs,
|
||||
logDir: dirs.logDir,
|
||||
lastError: readError(lane.slot),
|
||||
};
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
upLane,
|
||||
downLane,
|
||||
runtimeFacts,
|
||||
requireProfile,
|
||||
provisionLane,
|
||||
resetLaneData,
|
||||
removeLaneData,
|
||||
killTree,
|
||||
isAlive,
|
||||
readPidFiles,
|
||||
makeLaneDirs,
|
||||
clearLaneDirs,
|
||||
HEALTH_TIMEOUT_MS,
|
||||
BOOT_TIMEOUT_MS,
|
||||
};
|
||||
Reference in New Issue
Block a user