feat(clone): background clone jobs that survive a reload and a server restart
Both clone POSTs take background: true: every check answers synchronously,
then 202 {job}. git runs in a detached runner (scripts/clone-runner.js, its
own process group, stderr to a file — no SIGPIPE when the server restarts);
the server tails it for progress and runs the after-step (workspace / extra
folder) once when it exits. Jobs persist in <data dir>/runtime/clone-jobs.json
(0600, atomic, throttled progress writes); on start a live runner is adopted
(counted against the clone limit), an ended one finished, a queued one
requeued (interrupted if it carried a typed login), the rest interrupted.
GET /clone/jobs, DELETE /clone/jobs/:id, POST /clone/jobs/:id/cancel (queued:
dequeued; running: SIGTERM to the runner's group -> cancelled, no after-step)
and /retry (new job from the stored request, never a credential). The
held-open path is unchanged; worktree.gitCommand builds the runner's argv
and git-identity.cloneFailure the shared error classification.
This commit is contained in:
@@ -0,0 +1,527 @@
|
||||
/**
|
||||
* @file Tests for background clones (`server/lib/clone-jobs.js` and the
|
||||
* `background: true` path of `POST /api/lanes/clone` and
|
||||
* `POST /api/lanes/:id/sources/clone`): checks still answer synchronously,
|
||||
* a job is answered 202 and ends `done` with the workspace / extra folder
|
||||
* made, a failure keeps the route's own code (ENOREPO, ECREDENTIALS with
|
||||
* auth), a typed login never reaches the job file, retry and dismiss, cancel
|
||||
* of a queued clone (never starts) and of a running one (git is killed, no
|
||||
* workspace made), and a restart: a live detached runner adopted (and still
|
||||
* cancellable), one that ended finished with its after-step run once, a dead
|
||||
* one `interrupted`, a queued job requeued (or interrupted when it carried a
|
||||
* typed login), retention; and that in server mode a job is only ever its
|
||||
* owner's.
|
||||
* @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
|
||||
*/
|
||||
|
||||
const { describe, it, before, after } = require("node:test");
|
||||
const assert = require("node:assert/strict");
|
||||
const fs = require("node:fs");
|
||||
const os = require("node:os");
|
||||
const path = require("node:path");
|
||||
const http = require("node:http");
|
||||
const { execFileSync } = require("node:child_process");
|
||||
|
||||
const ROOT = fs.mkdtempSync(path.join(os.tmpdir(), "clone-jobs-"));
|
||||
process.env.DASHBOARD_DB_PATH = path.join(ROOT, "t.db");
|
||||
process.env.DASHBOARD_DATA_DIR = path.join(ROOT, "data");
|
||||
process.env.CLAUDE_HOME = path.join(ROOT, ".claude");
|
||||
process.env.DASHBOARD_REMOTE_SYNC_MS = "0";
|
||||
process.env.DASHBOARD_LIVENESS_PROBE = "0";
|
||||
process.env.GIT_CONFIG_NOSYSTEM = "1";
|
||||
process.env.HOME = ROOT;
|
||||
|
||||
const { createApp } = require("../index");
|
||||
const cloneJobs = require("../lib/clone-jobs");
|
||||
const queue = require("../lib/clone-queue");
|
||||
|
||||
let app;
|
||||
let remote;
|
||||
let base;
|
||||
const WORK = path.join(ROOT, "work");
|
||||
|
||||
function request(method, p, body) {
|
||||
return fetch(`${base}${p}`, {
|
||||
method,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: body === undefined ? undefined : JSON.stringify(body),
|
||||
}).then(async (r) => ({ status: r.status, body: await r.json().catch(() => null) }));
|
||||
}
|
||||
|
||||
async function waitFor(fn, ms = 20_000) {
|
||||
const end = Date.now() + ms;
|
||||
for (;;) {
|
||||
const v = await fn();
|
||||
if (v) return v;
|
||||
if (Date.now() > end) throw new Error("timed out");
|
||||
await new Promise((r) => setTimeout(r, 50));
|
||||
}
|
||||
}
|
||||
|
||||
async function jobById(id) {
|
||||
const r = await request("GET", "/api/lanes/clone/jobs");
|
||||
return r.body.jobs.find((j) => j.id === id);
|
||||
}
|
||||
|
||||
const finished = (id) =>
|
||||
waitFor(async () => {
|
||||
const j = await jobById(id);
|
||||
return j && !["queued", "running"].includes(j.state) ? j : null;
|
||||
});
|
||||
|
||||
before(async () => {
|
||||
const src = path.join(ROOT, "src");
|
||||
const bare = path.join(ROOT, "bare.git");
|
||||
execFileSync("git", ["init", "-q", "-b", "main", src]);
|
||||
fs.writeFileSync(path.join(src, "README.md"), "hi\n");
|
||||
execFileSync("git", ["-C", src, "add", "."]);
|
||||
execFileSync("git", [
|
||||
"-C",
|
||||
src,
|
||||
"-c",
|
||||
"user.name=t",
|
||||
"-c",
|
||||
"user.email=t@t",
|
||||
"commit",
|
||||
"-qm",
|
||||
"i",
|
||||
]);
|
||||
execFileSync("git", ["clone", "-q", "--bare", src, bare]);
|
||||
|
||||
remote = http.createServer((_req, res) => {
|
||||
res.writeHead(401, { "WWW-Authenticate": 'Basic realm="t"' });
|
||||
res.end();
|
||||
});
|
||||
await new Promise((r) => remote.listen(0, "127.0.0.1", r));
|
||||
fs.writeFileSync(
|
||||
path.join(ROOT, ".gitconfig"),
|
||||
[
|
||||
`[url "file://${bare}"]`,
|
||||
"\tinsteadOf = https://forge.test/o/r.git",
|
||||
`[url "file://${path.join(ROOT, "missing.git")}"]`,
|
||||
"\tinsteadOf = git@forge.test:o/missing.git",
|
||||
`[url "http://127.0.0.1:${remote.address().port}/"]`,
|
||||
"\tinsteadOf = https://private.test/",
|
||||
"",
|
||||
].join("\n")
|
||||
);
|
||||
fs.mkdirSync(WORK);
|
||||
app = http.createServer(createApp());
|
||||
await new Promise((r) => app.listen(0, "127.0.0.1", r));
|
||||
base = `http://127.0.0.1:${app.address().port}`;
|
||||
});
|
||||
after(async () => {
|
||||
await new Promise((r) => app.close(r));
|
||||
await new Promise((r) => remote.close(r));
|
||||
});
|
||||
|
||||
describe("background clone", () => {
|
||||
it("still refuses bad input synchronously", async () => {
|
||||
const bad = await request("POST", "/api/lanes/clone", {
|
||||
url: "https://forge.test/o/r.git",
|
||||
parent: WORK,
|
||||
background: "yes",
|
||||
});
|
||||
assert.equal(bad.status, 400);
|
||||
assert.equal(bad.body.error.code, "EBADFIELD");
|
||||
const url = await request("POST", "/api/lanes/clone", {
|
||||
url: "file:///etc",
|
||||
parent: WORK,
|
||||
background: true,
|
||||
});
|
||||
assert.equal(url.body.error.code, "EBADURL");
|
||||
fs.mkdirSync(path.join(WORK, "taken"));
|
||||
const taken = await request("POST", "/api/lanes/clone", {
|
||||
url: "https://forge.test/o/r.git",
|
||||
parent: WORK,
|
||||
name: "taken",
|
||||
background: true,
|
||||
});
|
||||
assert.equal(taken.status, 409);
|
||||
assert.equal(taken.body.error.code, "EEXISTS");
|
||||
const tool = await request("POST", "/api/lanes/clone", {
|
||||
url: "https://forge.test/o/r.git",
|
||||
parent: WORK,
|
||||
name: "t2",
|
||||
tool: "nope",
|
||||
background: true,
|
||||
});
|
||||
assert.equal(tool.body.error.code, "EBADTOOL");
|
||||
const jobs = await request("GET", "/api/lanes/clone/jobs");
|
||||
assert.deepEqual(jobs.body.jobs, []);
|
||||
});
|
||||
|
||||
it("answers 202 with a job that ends done with its workspace", async () => {
|
||||
const r = await request("POST", "/api/lanes/clone", {
|
||||
url: "https://forge.test/o/r.git",
|
||||
parent: WORK,
|
||||
name: "bg",
|
||||
title: "Bg",
|
||||
background: true,
|
||||
});
|
||||
assert.equal(r.status, 202, JSON.stringify(r.body));
|
||||
const { job } = r.body;
|
||||
assert.match(job.id, /^[A-Za-z0-9_-]{16,}$/);
|
||||
assert.equal(job.kind, "workspace");
|
||||
assert.ok(["queued", "running"].includes(job.state));
|
||||
assert.equal(job.target, path.join(WORK, "bg"));
|
||||
assert.equal(job.laneId, null);
|
||||
assert.equal(job.finishedAt, null);
|
||||
assert.ok("phase" in job && "position" in job);
|
||||
assert.equal(job.owner, undefined);
|
||||
assert.deepEqual(job.request, {
|
||||
url: "https://forge.test/o/r.git",
|
||||
parent: WORK,
|
||||
name: "bg",
|
||||
title: "Bg",
|
||||
});
|
||||
const done = await finished(job.id);
|
||||
assert.equal(done.state, "done", JSON.stringify(done));
|
||||
assert.equal(done.result.created, true);
|
||||
assert.equal(done.result.path, path.join(WORK, "bg"));
|
||||
assert.ok(done.finishedAt);
|
||||
const lane = await request("GET", `/api/lanes/${done.result.laneId}`);
|
||||
assert.equal(lane.body.lane?.title ?? lane.body.title, "Bg");
|
||||
assert.equal(fs.readFileSync(path.join(WORK, "bg", "README.md"), "utf8"), "hi\n");
|
||||
|
||||
const file = cloneJobs.filePath();
|
||||
assert.equal(fs.statSync(file).mode & 0o777, 0o600);
|
||||
const disk = JSON.parse(fs.readFileSync(file, "utf8")).jobs.find((j) => j.id === job.id);
|
||||
assert.equal(disk.state, "done");
|
||||
|
||||
const busy = await request("POST", `/api/lanes/clone/jobs/${job.id}/retry`, {});
|
||||
assert.equal(busy.status, 409);
|
||||
assert.equal(busy.body.error.code, "EBUSY");
|
||||
const dismissed = await request("DELETE", `/api/lanes/clone/jobs/${job.id}`);
|
||||
assert.deepEqual(dismissed.body, { ok: true });
|
||||
assert.equal(await jobById(job.id), undefined);
|
||||
const again = await request("DELETE", `/api/lanes/clone/jobs/${job.id}`);
|
||||
assert.equal(again.status, 404);
|
||||
assert.equal(again.body.error.code, "ENOTFOUND");
|
||||
});
|
||||
|
||||
it("records the route's own error codes, and never a typed login", async () => {
|
||||
const missing = await request("POST", "/api/lanes/clone", {
|
||||
url: "git@forge.test:o/missing.git",
|
||||
parent: WORK,
|
||||
name: "missing",
|
||||
background: true,
|
||||
});
|
||||
assert.equal(missing.status, 202);
|
||||
const m = await finished(missing.body.job.id);
|
||||
assert.equal(m.state, "failed");
|
||||
assert.equal(m.error.code, "ENOREPO");
|
||||
assert.equal(m.result, null);
|
||||
|
||||
const login = await request("POST", "/api/lanes/clone", {
|
||||
url: "https://private.test/o/r.git",
|
||||
parent: WORK,
|
||||
name: "private",
|
||||
credentials: { username: "u", password: "s3cret-typed" },
|
||||
background: true,
|
||||
});
|
||||
assert.equal(login.status, 202);
|
||||
assert.equal(login.body.job.request.credentials, undefined);
|
||||
const l = await finished(login.body.job.id);
|
||||
assert.equal(l.error.code, "ECREDENTIALS");
|
||||
assert.deepEqual(l.error.auth, { protocol: "https", host: "private.test" });
|
||||
assert.ok(!fs.readFileSync(cloneJobs.filePath(), "utf8").includes("s3cret-typed"));
|
||||
|
||||
// Retry: a new job from the old request (+ a login typed now); the old goes.
|
||||
const retry = await request("POST", `/api/lanes/clone/jobs/${l.id}/retry`, {
|
||||
credentials: { username: "u", password: "again-typed" },
|
||||
});
|
||||
assert.equal(retry.status, 202, JSON.stringify(retry.body));
|
||||
assert.notEqual(retry.body.job.id, l.id);
|
||||
assert.equal(retry.body.job.name, "private");
|
||||
assert.equal(await jobById(l.id), undefined);
|
||||
const r2 = await finished(retry.body.job.id);
|
||||
assert.equal(r2.error.code, "ECREDENTIALS");
|
||||
assert.ok(!fs.readFileSync(cloneJobs.filePath(), "utf8").includes("again-typed"));
|
||||
const bad = await request("POST", `/api/lanes/clone/jobs/${r2.id}/retry`, {
|
||||
credentials: "nope",
|
||||
});
|
||||
assert.equal(bad.body.error.code, "EBADCRED");
|
||||
assert.ok(await jobById(r2.id), "a refused retry keeps the old job");
|
||||
const none = await request("POST", "/api/lanes/clone/jobs/nope-nope-nope-nope/retry", {});
|
||||
assert.equal(none.status, 404);
|
||||
});
|
||||
|
||||
it("clones an extra folder beside the workspace", async () => {
|
||||
const home = path.join(WORK, "app");
|
||||
fs.mkdirSync(home);
|
||||
const made = await request("POST", "/api/lanes/ensure", { cwd: home });
|
||||
const id = made.body.lane.id;
|
||||
const r = await request("POST", `/api/lanes/${id}/sources/clone`, {
|
||||
url: "https://forge.test/o/r.git",
|
||||
name: "lib",
|
||||
background: true,
|
||||
});
|
||||
assert.equal(r.status, 202, JSON.stringify(r.body));
|
||||
assert.equal(r.body.job.kind, "source");
|
||||
assert.equal(r.body.job.laneId, id);
|
||||
const done = await finished(r.body.job.id);
|
||||
assert.equal(done.state, "done", JSON.stringify(done));
|
||||
assert.equal(done.result.laneId, id);
|
||||
const ctx = await request("GET", `/api/lanes/${id}/sources`);
|
||||
const folders = (ctx.body.sources || []).filter((s) => s.kind === "folder");
|
||||
assert.deepEqual(
|
||||
folders.map((f) => fs.realpathSync(f.path)),
|
||||
[fs.realpathSync(path.join(WORK, "lib"))]
|
||||
);
|
||||
});
|
||||
|
||||
it("cancels a queued clone before it starts", async () => {
|
||||
const saved = process.env.AGENTHUB_CLONE_CONCURRENCY;
|
||||
process.env.AGENTHUB_CLONE_CONCURRENCY = "1";
|
||||
let open;
|
||||
const blocker = queue.withCloneSlot(() => new Promise((r) => (open = r)));
|
||||
try {
|
||||
const r = await request("POST", "/api/lanes/clone", {
|
||||
url: "https://forge.test/o/r.git",
|
||||
parent: WORK,
|
||||
name: "queued",
|
||||
background: true,
|
||||
});
|
||||
assert.equal(r.status, 202);
|
||||
assert.equal(r.body.job.state, "queued");
|
||||
assert.equal(r.body.job.position, 1);
|
||||
const c = await request("POST", `/api/lanes/clone/jobs/${r.body.job.id}/cancel`);
|
||||
assert.equal(c.status, 200, JSON.stringify(c.body));
|
||||
assert.equal(c.body.job.state, "cancelled");
|
||||
assert.equal(c.body.job.error.code, "ECANCELLED");
|
||||
assert.equal(queue.cloneQueueState().waiting, 0);
|
||||
const twice = await request("POST", `/api/lanes/clone/jobs/${r.body.job.id}/cancel`);
|
||||
assert.equal(twice.status, 409);
|
||||
assert.equal(twice.body.error.code, "EDONE");
|
||||
open();
|
||||
await blocker;
|
||||
await new Promise((res) => setTimeout(res, 200));
|
||||
assert.equal((await jobById(r.body.job.id)).state, "cancelled");
|
||||
assert.ok(!fs.existsSync(path.join(WORK, "queued")));
|
||||
} finally {
|
||||
open?.();
|
||||
if (saved === undefined) delete process.env.AGENTHUB_CLONE_CONCURRENCY;
|
||||
else process.env.AGENTHUB_CLONE_CONCURRENCY = saved;
|
||||
}
|
||||
});
|
||||
|
||||
it("cancels a running clone: git is killed, the slot frees, no workspace", async () => {
|
||||
// An ssh "transport" that never answers keeps git clone running (a real
|
||||
// ssh exits once git is gone; this stand-in lingers for its 15 s).
|
||||
process.env.GIT_SSH_COMMAND = "exec sleep 15 2>/dev/null #";
|
||||
try {
|
||||
const r = await request("POST", "/api/lanes/clone", {
|
||||
url: "ssh://slow.test/o/slow.git",
|
||||
parent: WORK,
|
||||
name: "slow",
|
||||
background: true,
|
||||
});
|
||||
assert.equal(r.status, 202);
|
||||
const id = r.body.job.id;
|
||||
await waitFor(async () => (await jobById(id))?.state === "running");
|
||||
await new Promise((res) => setTimeout(res, 300));
|
||||
const c = await request("POST", `/api/lanes/clone/jobs/${id}/cancel`);
|
||||
assert.equal(c.body.job?.state, "cancelled", JSON.stringify([c.body, await jobById(id)]));
|
||||
await waitFor(() => queue.cloneQueueState().running === 0, 10_000);
|
||||
const j = await jobById(id);
|
||||
assert.equal(j.state, "cancelled");
|
||||
assert.equal(j.result, null);
|
||||
assert.ok(!fs.existsSync(path.join(WORK, "slow")));
|
||||
const lanes = await request("GET", "/api/lanes");
|
||||
const list = lanes.body.lanes || lanes.body;
|
||||
assert.ok(!list.some((l) => l.cwd.endsWith(`${path.sep}slow`)));
|
||||
const retry = await request("POST", `/api/lanes/clone/jobs/${id}/retry`, {});
|
||||
assert.equal(retry.status, 202, "a cancelled clone can be retried");
|
||||
const c2 = await request("POST", `/api/lanes/clone/jobs/${retry.body.job.id}/cancel`);
|
||||
assert.equal(c2.body.job.state, "cancelled");
|
||||
await waitFor(() => queue.cloneQueueState().running === 0, 10_000);
|
||||
} finally {
|
||||
delete process.env.GIT_SSH_COMMAND;
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("clone-jobs store", () => {
|
||||
it("answers a job only to its owner in server mode", () => {
|
||||
const savedMode = process.env.AGENTHUB_MODE;
|
||||
process.env.AGENTHUB_MODE = "server";
|
||||
try {
|
||||
const mk = (owner) =>
|
||||
cloneJobs.create({
|
||||
owner,
|
||||
kind: "workspace",
|
||||
url: "https://forge.test/o/r.git",
|
||||
name: `o${owner}`,
|
||||
parent: WORK,
|
||||
target: path.join(WORK, `o${owner}`),
|
||||
request: { url: "u", parent: WORK, name: "n", credentials: { password: "x" } },
|
||||
});
|
||||
const a = mk(101);
|
||||
const b = mk(102);
|
||||
assert.equal(a.request.credentials, undefined);
|
||||
assert.deepEqual(
|
||||
cloneJobs.list(101).map((j) => j.id),
|
||||
[a.id]
|
||||
);
|
||||
assert.equal(cloneJobs.get(b.id, 101), null);
|
||||
assert.throws(() => cloneJobs.remove(b.id, 101), { code: "ENOTFOUND" });
|
||||
assert.throws(() => cloneJobs.cancel(b.id, 101), { code: "ENOTFOUND" });
|
||||
assert.throws(() => cloneJobs.remove(a.id, 101), { code: "EBUSY" });
|
||||
cloneJobs.cancel(a.id, 101);
|
||||
cloneJobs.cancel(b.id, 102);
|
||||
cloneJobs.remove(a.id, 101);
|
||||
cloneJobs.remove(b.id, 102);
|
||||
} finally {
|
||||
if (savedMode === undefined) delete process.env.AGENTHUB_MODE;
|
||||
else process.env.AGENTHUB_MODE = savedMode;
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("a restart", () => {
|
||||
const H = 60 * 60 * 1000;
|
||||
const ago = (ms) => new Date(Date.now() - ms).toISOString();
|
||||
const row = (id, state, extra = {}) => ({
|
||||
id,
|
||||
owner: null,
|
||||
kind: "workspace",
|
||||
state,
|
||||
url: "https://forge.test/o/r.git",
|
||||
name: id,
|
||||
parent: WORK,
|
||||
target: path.join(WORK, id),
|
||||
laneId: null,
|
||||
error: null,
|
||||
result: null,
|
||||
createdAt: ago(H),
|
||||
finishedAt: null,
|
||||
request: { url: "https://forge.test/o/r.git", parent: WORK, name: id },
|
||||
pid: null,
|
||||
once: null,
|
||||
...extra,
|
||||
});
|
||||
/** Writes the job file and reloads it, as a new server process would. */
|
||||
const restart = (rows) => {
|
||||
cloneJobs._reset();
|
||||
fs.mkdirSync(path.dirname(cloneJobs.filePath()), { recursive: true });
|
||||
fs.writeFileSync(cloneJobs.filePath(), JSON.stringify({ jobs: rows }));
|
||||
cloneJobs.resume();
|
||||
};
|
||||
const lanesAt = async (dir) => {
|
||||
const r = await request("GET", "/api/lanes");
|
||||
const list = r.body.lanes || r.body;
|
||||
return list.filter((l) => fs.realpathSync(l.cwd) === fs.realpathSync(dir));
|
||||
};
|
||||
|
||||
it("interrupts a dead runner, requeues a queued clone, drops expired jobs", async () => {
|
||||
const dead = 2 ** 22 + 12345; // past pid_max on a default kernel: never alive
|
||||
restart([
|
||||
row("was-running-aaaaaaaa", "running", { pid: dead }),
|
||||
row("queued-login-aaaaaaa", "queued", {
|
||||
once: { used: true, remember: false, member: false, host: "forge.test" },
|
||||
}),
|
||||
row("requeued-aaaaaaaaaaa", "queued"),
|
||||
row("done-fresh-aaaaaaaaa", "done", { finishedAt: ago(2 * H) }),
|
||||
row("done-stale-aaaaaaaaa", "done", { finishedAt: ago(25 * H) }),
|
||||
row("failed-fresh-aaaaaaa", "failed", { finishedAt: ago(6 * 24 * H) }),
|
||||
row("failed-stale-aaaaaaa", "failed", { finishedAt: ago(8 * 24 * H) }),
|
||||
row("cancel-stale-aaaaaaa", "cancelled", { finishedAt: ago(8 * 24 * H) }),
|
||||
]);
|
||||
const ids = cloneJobs.list(null).map((j) => j.id);
|
||||
assert.deepEqual([...ids].sort(), [
|
||||
"done-fresh-aaaaaaaaa",
|
||||
"failed-fresh-aaaaaaa",
|
||||
"queued-login-aaaaaaa",
|
||||
"requeued-aaaaaaaaaaa",
|
||||
"was-running-aaaaaaaa",
|
||||
]);
|
||||
for (const id of ["was-running-aaaaaaaa", "queued-login-aaaaaaa"]) {
|
||||
const j = cloneJobs.get(id, null);
|
||||
assert.equal(j.state, "interrupted", id);
|
||||
assert.deepEqual(j.error, {
|
||||
code: "EINTERRUPTED",
|
||||
message: "AgentHub restarted while this clone ran",
|
||||
});
|
||||
assert.ok(j.finishedAt);
|
||||
}
|
||||
const disk = JSON.parse(fs.readFileSync(cloneJobs.filePath(), "utf8")).jobs;
|
||||
assert.equal(disk.find((j) => j.id === "was-running-aaaaaaaa").state, "interrupted");
|
||||
// The queued one (no typed login) simply runs now, workspace included.
|
||||
const done = await finished("requeued-aaaaaaaaaaa");
|
||||
assert.equal(done.state, "done", JSON.stringify(done));
|
||||
assert.equal((await lanesAt(path.join(WORK, "requeued-aaaaaaaaaaa"))).length, 1);
|
||||
});
|
||||
|
||||
it("finishes a runner that ended while no server ran: the after-step runs once", async () => {
|
||||
const id = "ended-ok-aaaaaaaaaaa";
|
||||
const target = path.join(WORK, id);
|
||||
execFileSync("git", ["clone", "-q", path.join(ROOT, "bare.git"), target]);
|
||||
fs.mkdirSync(cloneJobs.jobDir(id), { recursive: true });
|
||||
fs.writeFileSync(path.join(cloneJobs.jobDir(id), "exit.json"), '{"code":0,"signal":null}');
|
||||
const bad = "ended-bad-aaaaaaaaaa";
|
||||
fs.mkdirSync(cloneJobs.jobDir(bad), { recursive: true });
|
||||
fs.writeFileSync(path.join(cloneJobs.jobDir(bad), "exit.json"), '{"code":128,"signal":null}');
|
||||
fs.writeFileSync(
|
||||
path.join(cloneJobs.jobDir(bad), "stderr.log"),
|
||||
"Cloning into 'x'...\nfatal: repository 'https://forge.test/o/nope.git/' not found\n"
|
||||
);
|
||||
restart([
|
||||
row(id, "running", { pid: 2 ** 22 + 1 }),
|
||||
row(bad, "running", { pid: 2 ** 22 + 2, url: "https://forge.test/o/nope.git" }),
|
||||
]);
|
||||
const ok = await finished(id);
|
||||
assert.equal(ok.state, "done", JSON.stringify(ok));
|
||||
assert.equal(ok.result.created, true);
|
||||
const failed = await finished(bad);
|
||||
assert.equal(failed.error.code, "ENOREPO");
|
||||
assert.ok(!fs.existsSync(cloneJobs.jobDir(id)), "the runner's files are cleaned up");
|
||||
// A second reload finds nothing left to do: still one workspace.
|
||||
restart(JSON.parse(fs.readFileSync(cloneJobs.filePath(), "utf8")).jobs);
|
||||
await new Promise((r) => setTimeout(r, 100));
|
||||
assert.equal((await lanesAt(target)).length, 1);
|
||||
assert.equal(cloneJobs.get(id, null).result.laneId, ok.result.laneId);
|
||||
});
|
||||
|
||||
it("adopts a runner that outlived the server, and can still cancel it", async () => {
|
||||
process.env.GIT_SSH_COMMAND = "exec sleep 15 2>/dev/null #";
|
||||
try {
|
||||
const r = await request("POST", "/api/lanes/clone", {
|
||||
url: "ssh://slow.test/o/slow.git",
|
||||
parent: WORK,
|
||||
name: "adopted",
|
||||
background: true,
|
||||
});
|
||||
assert.equal(r.status, 202);
|
||||
const id = r.body.job.id;
|
||||
const pid = await waitFor(async () => (await jobById(id))?.pid);
|
||||
// Its own session and group: a server restart leaves it running.
|
||||
const pgid = Number(
|
||||
execFileSync("ps", ["-o", "pgid=", "-p", String(pid)], { encoding: "utf8" })
|
||||
);
|
||||
assert.equal(pgid, pid);
|
||||
const before = queue.cloneQueueState().running;
|
||||
// A new server process: the old one's tail stops, the file is re-read.
|
||||
cloneJobs._reset();
|
||||
cloneJobs.resume();
|
||||
const adopted = cloneJobs.get(id, null);
|
||||
assert.equal(adopted.state, "running");
|
||||
assert.equal(queue.cloneQueueState().running, before + 1, "counted against the limit");
|
||||
const c = await request("POST", `/api/lanes/clone/jobs/${id}/cancel`);
|
||||
assert.equal(c.body.job.state, "cancelled");
|
||||
await waitFor(() => queue.cloneQueueState().running === before, 10_000);
|
||||
await waitFor(() => {
|
||||
try {
|
||||
process.kill(pid, 0);
|
||||
return false;
|
||||
} catch {
|
||||
return true;
|
||||
}
|
||||
}, 10_000);
|
||||
assert.equal((await jobById(id)).state, "cancelled");
|
||||
assert.ok(!fs.existsSync(path.join(WORK, "adopted")));
|
||||
} finally {
|
||||
delete process.env.GIT_SSH_COMMAND;
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -500,6 +500,14 @@ function startBackgroundServices() {
|
||||
// One-time legacy-session backfill (a no-op once its marker file exists).
|
||||
autoImportLegacySessions();
|
||||
|
||||
// Background clones a previous process started: adopt a runner that is
|
||||
// still cloning, finish one that ended, requeue or interrupt the rest.
|
||||
try {
|
||||
require("./lib/clone-jobs").resume();
|
||||
} catch (err) {
|
||||
console.warn(`[clone-jobs] resume failed: ${err.message}`);
|
||||
}
|
||||
|
||||
// Before any tmux call: without a UTF-8 locale every attach prints `_` for
|
||||
// non-ASCII text (a macOS LaunchAgent sets none).
|
||||
require("./lib/shell-path").ensureUtf8Locale();
|
||||
|
||||
@@ -0,0 +1,651 @@
|
||||
/**
|
||||
* @file Background clones (`POST /api/lanes/clone` with `background: true`):
|
||||
* a clone that outlives the request that started it AND the server. Each job
|
||||
* is kept in memory and in `<data dir>/runtime/clone-jobs.json` (0600, tmp +
|
||||
* rename): on every state change at once, on git's progress at most every
|
||||
* `WRITE_EVERY_MS`. Its git runs in a detached runner
|
||||
* (`scripts/clone-runner.js`, pid recorded on the job) with stderr appended
|
||||
* to `<data dir>/runtime/clone-jobs/<id>/stderr.log` — the server tails it
|
||||
* for progress and finishes the job when the runner writes `exit.json`.
|
||||
* An update restarts the server but not the runner (`KillMode=process`), so
|
||||
* the first read in a new process reconciles the file: a running job whose
|
||||
* runner is alive is ADOPTED (counted against clone-queue's limit, tailed,
|
||||
* finished when it exits), one whose `exit.json` exists is finished now, and
|
||||
* any other is `interrupted` (`EINTERRUPTED`); a queued job is queued again,
|
||||
* unless it carried a typed login (gone with the old process: interrupted,
|
||||
* so the user types it again). Finishing runs the after-step the routes
|
||||
* register (`setAfterStep`: the workspace, or the extra folder) exactly once,
|
||||
* and a cancelled job never runs it. `cancel` takes a queued job out of
|
||||
* clone-queue or SIGTERMs the runner's process group. Finished jobs are
|
||||
* dropped after 24 h (done) or 7 days (failed/interrupted/cancelled). A job
|
||||
* is answered only to its owner (`scope.ownerFor`; local mode: every job),
|
||||
* and `request` — what a retry re-sends — never holds a credential: a typed
|
||||
* login is used for its one run and never written here.
|
||||
* @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
|
||||
*/
|
||||
|
||||
"use strict";
|
||||
|
||||
const fs = require("node:fs");
|
||||
const path = require("node:path");
|
||||
const crypto = require("node:crypto");
|
||||
const { spawn } = require("node:child_process");
|
||||
const { getDataDir } = require("./claude-home");
|
||||
|
||||
const KEEP_DONE_MS = 24 * 60 * 60 * 1000;
|
||||
const KEEP_FAILED_MS = 7 * 24 * 60 * 60 * 1000;
|
||||
const WRITE_EVERY_MS = 2000;
|
||||
const POLL_MS = 500;
|
||||
const TAIL_MAX = 64 * 1024;
|
||||
const ACTIVE = new Set(["queued", "running"]);
|
||||
const REQUEST_KEYS = ["url", "parent", "name", "tool", "title", "partial"];
|
||||
const RUNNER = path.join(__dirname, "..", "..", "scripts", "clone-runner.js");
|
||||
|
||||
/** id → job record (the CloneJob fields plus `owner`, `once`). */
|
||||
let jobs = null;
|
||||
let timer = null;
|
||||
/** Live progress pollers, so a simulated restart (`_reset`) stops them. */
|
||||
const watchers = new Set();
|
||||
/** `(job) => {status, json, laneId?, created?}` — the routes' after-step. */
|
||||
let afterStep = null;
|
||||
|
||||
const runtimeDir = () => path.join(getDataDir(), "runtime");
|
||||
const filePath = () => path.join(runtimeDir(), "clone-jobs.json");
|
||||
/** A job's files; its id is server-made base64url, never from a request. */
|
||||
const jobDir = (id) => path.join(runtimeDir(), "clone-jobs", id);
|
||||
|
||||
const serverMode = () => process.env.AGENTHUB_MODE === "server";
|
||||
const queue = () => require("./clone-queue");
|
||||
|
||||
function expired(job, now) {
|
||||
if (!job.finishedAt) return false;
|
||||
const age = now - Date.parse(job.finishedAt);
|
||||
return job.state === "done" ? age > KEEP_DONE_MS : age > KEEP_FAILED_MS;
|
||||
}
|
||||
|
||||
/** What goes to disk (`view` drops `owner` and `once`); never a callback. */
|
||||
function record(job) {
|
||||
return {
|
||||
id: job.id,
|
||||
owner: job.owner,
|
||||
kind: job.kind,
|
||||
state: job.state,
|
||||
url: job.url,
|
||||
name: job.name,
|
||||
parent: job.parent,
|
||||
target: job.target,
|
||||
laneId: job.laneId,
|
||||
phase: job.phase,
|
||||
label: job.label,
|
||||
percent: job.percent,
|
||||
detail: job.detail,
|
||||
error: job.error,
|
||||
result: job.result,
|
||||
createdAt: job.createdAt,
|
||||
finishedAt: job.finishedAt,
|
||||
request: job.request,
|
||||
pid: job.pid ?? null,
|
||||
// Whether a login was typed for this run — never the login itself.
|
||||
once: job.once ?? null,
|
||||
};
|
||||
}
|
||||
|
||||
/** The in-memory-only parts of a job: cancel signal, queue hooks, mirror. */
|
||||
function arm(job) {
|
||||
const controller = new AbortController();
|
||||
Object.defineProperties(job, {
|
||||
// A progressToken's clone-queue record, kept in step (route-set).
|
||||
mirror: { enumerable: false, writable: true, value: null },
|
||||
// An adopted runner's clone-queue slot.
|
||||
release: { enumerable: false, writable: true, value: null },
|
||||
finishing: { enumerable: false, writable: true, value: false },
|
||||
signal: { enumerable: false, value: controller.signal },
|
||||
abort: { enumerable: false, value: () => controller.abort() },
|
||||
onChange: {
|
||||
enumerable: false,
|
||||
value: (what) => {
|
||||
if (job.mirror) {
|
||||
for (const k of ["state", "phase", "label", "percent", "detail"]) {
|
||||
job.mirror[k] = job[k];
|
||||
}
|
||||
}
|
||||
if (what === "state") writeNow();
|
||||
else writeSoon();
|
||||
},
|
||||
},
|
||||
});
|
||||
return job;
|
||||
}
|
||||
|
||||
function interrupt(job) {
|
||||
Object.assign(job, {
|
||||
state: "interrupted",
|
||||
error: { code: "EINTERRUPTED", message: "AgentHub restarted while this clone ran" },
|
||||
finishedAt: new Date().toISOString(),
|
||||
});
|
||||
}
|
||||
|
||||
function load() {
|
||||
if (jobs) return jobs;
|
||||
jobs = new Map();
|
||||
let list = [];
|
||||
try {
|
||||
const parsed = JSON.parse(fs.readFileSync(filePath(), "utf8"));
|
||||
if (Array.isArray(parsed?.jobs)) list = parsed.jobs;
|
||||
} catch {
|
||||
// No file yet, or an unreadable one: start empty.
|
||||
}
|
||||
const now = Date.now();
|
||||
let changed = false;
|
||||
const later = [];
|
||||
for (const j of list) {
|
||||
if (!j || typeof j.id !== "string" || !/^[A-Za-z0-9_-]{16,64}$/.test(j.id)) continue;
|
||||
if (expired(j, now)) {
|
||||
rmJobDir(j.id);
|
||||
changed = true;
|
||||
continue;
|
||||
}
|
||||
arm(j);
|
||||
jobs.set(j.id, j);
|
||||
if (j.state === "running") {
|
||||
if (readExit(j.id)) later.push(() => finishFromDisk(j));
|
||||
else if (runnerAlive(j.pid)) adopt(j);
|
||||
else {
|
||||
interrupt(j);
|
||||
rmJobDir(j.id);
|
||||
changed = true;
|
||||
}
|
||||
} else if (j.state === "queued") {
|
||||
if (j.once?.used) {
|
||||
interrupt(j);
|
||||
changed = true;
|
||||
} else later.push(() => start(j, null));
|
||||
}
|
||||
}
|
||||
if (changed) writeNow();
|
||||
for (const fn of later) setImmediate(fn);
|
||||
return jobs;
|
||||
}
|
||||
|
||||
function writeNow() {
|
||||
if (timer) {
|
||||
clearTimeout(timer);
|
||||
timer = null;
|
||||
}
|
||||
const file = filePath();
|
||||
try {
|
||||
fs.mkdirSync(path.dirname(file), { recursive: true, mode: 0o700 });
|
||||
const tmp = `${file}.${process.pid}.tmp`;
|
||||
const body = JSON.stringify({ jobs: [...(jobs || new Map()).values()].map(record) });
|
||||
fs.writeFileSync(tmp, body, { mode: 0o600 });
|
||||
fs.chmodSync(tmp, 0o600);
|
||||
fs.renameSync(tmp, file);
|
||||
} catch (err) {
|
||||
console.warn(`[clone-jobs] could not save ${file}: ${err.message}`);
|
||||
}
|
||||
}
|
||||
|
||||
/** Progress is cheap in memory; on disk at most every WRITE_EVERY_MS. */
|
||||
function writeSoon() {
|
||||
if (timer) return;
|
||||
timer = setTimeout(writeNow, WRITE_EVERY_MS);
|
||||
timer.unref?.();
|
||||
}
|
||||
|
||||
function rmJobDir(id) {
|
||||
try {
|
||||
fs.rmSync(jobDir(id), { recursive: true, force: true });
|
||||
} catch {
|
||||
/* AgentHub's own scratch: best effort */
|
||||
}
|
||||
}
|
||||
|
||||
function prune() {
|
||||
const now = Date.now();
|
||||
let changed = false;
|
||||
for (const [id, j] of load()) {
|
||||
if (expired(j, now)) {
|
||||
jobs.delete(id);
|
||||
rmJobDir(id);
|
||||
changed = true;
|
||||
}
|
||||
}
|
||||
if (changed) writeNow();
|
||||
}
|
||||
|
||||
// ── The runner ─────────────────────────────────────────────────────────────
|
||||
|
||||
function readExit(id) {
|
||||
try {
|
||||
return JSON.parse(fs.readFileSync(path.join(jobDir(id), "exit.json"), "utf8"));
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/** Whether `pid` is still OUR runner (not a reused pid, not a zombie). */
|
||||
function runnerAlive(pid) {
|
||||
if (!Number.isInteger(pid) || pid <= 0) return false;
|
||||
try {
|
||||
process.kill(pid, 0);
|
||||
} catch (err) {
|
||||
if (err.code !== "EPERM") return false;
|
||||
}
|
||||
if (process.platform !== "linux") return true;
|
||||
try {
|
||||
return fs.readFileSync(`/proc/${pid}/cmdline`, "utf8").includes("clone-runner.js");
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/** SIGTERMs the runner's process group: git dies, the runner records it. */
|
||||
function killRunner(job) {
|
||||
if (!job.pid) return;
|
||||
try {
|
||||
process.kill(process.platform === "win32" ? job.pid : -job.pid, "SIGTERM");
|
||||
} catch {
|
||||
/* already gone */
|
||||
}
|
||||
}
|
||||
|
||||
function spawnRunner(job, spec, env) {
|
||||
const dir = jobDir(job.id);
|
||||
fs.rmSync(dir, { recursive: true, force: true });
|
||||
fs.mkdirSync(dir, { recursive: true, mode: 0o700 });
|
||||
const specPath = path.join(dir, "spec.json");
|
||||
fs.writeFileSync(specPath, JSON.stringify(spec), { mode: 0o600, flag: "wx" });
|
||||
const child = spawn(process.execPath, [RUNNER, specPath], {
|
||||
cwd: dir,
|
||||
env,
|
||||
detached: true,
|
||||
stdio: "ignore",
|
||||
windowsHide: true,
|
||||
});
|
||||
child.on("error", () => {
|
||||
/* the poller sees no runner and fails the job */
|
||||
});
|
||||
child.unref();
|
||||
job.pid = child.pid ?? null;
|
||||
writeNow();
|
||||
}
|
||||
|
||||
/**
|
||||
* Tails a runner's stderr.log into `onProgress` until `exit.json` appears
|
||||
* (or the runner is gone without one: `lost`). Resolves
|
||||
* `{ok, timedOut, lost, stderr}` — `stderr` is the last TAIL_MAX of it.
|
||||
*/
|
||||
function watch(job, onProgress) {
|
||||
return new Promise((resolve) => {
|
||||
const dir = jobDir(job.id);
|
||||
let offset = 0;
|
||||
let tail = "";
|
||||
let misses = 0;
|
||||
const read = () => {
|
||||
let fd;
|
||||
try {
|
||||
fd = fs.openSync(path.join(dir, "stderr.log"), "r");
|
||||
const size = fs.fstatSync(fd).size;
|
||||
if (size <= offset) return;
|
||||
const from = Math.max(offset, size - TAIL_MAX);
|
||||
const buf = Buffer.alloc(size - from);
|
||||
fs.readSync(fd, buf, 0, buf.length, from);
|
||||
offset = size;
|
||||
const chunk = buf.toString("utf8");
|
||||
tail = (tail + chunk).slice(-TAIL_MAX);
|
||||
try {
|
||||
onProgress?.(chunk);
|
||||
} catch {
|
||||
/* a progress reader never fails the clone */
|
||||
}
|
||||
} catch {
|
||||
/* no log yet */
|
||||
} finally {
|
||||
if (fd !== undefined) fs.closeSync(fd);
|
||||
}
|
||||
};
|
||||
const w = {};
|
||||
const stop = (out) => {
|
||||
clearInterval(w.iv);
|
||||
watchers.delete(w);
|
||||
resolve(out);
|
||||
};
|
||||
w.iv = setInterval(() => {
|
||||
read();
|
||||
const exit = readExit(job.id);
|
||||
if (exit) {
|
||||
read();
|
||||
return stop({
|
||||
ok: exit.code === 0 && !exit.signal,
|
||||
timedOut: Boolean(exit.timedOut),
|
||||
lost: false,
|
||||
stderr: tail,
|
||||
});
|
||||
}
|
||||
if (runnerAlive(job.pid)) misses = 0;
|
||||
else if (++misses >= 2) stop({ ok: false, timedOut: false, lost: true, stderr: tail });
|
||||
}, POLL_MS);
|
||||
w.iv.unref?.();
|
||||
watchers.add(w);
|
||||
});
|
||||
}
|
||||
|
||||
const failureMessage = (out) =>
|
||||
out.lost
|
||||
? "the clone stopped unexpectedly"
|
||||
: out.timedOut
|
||||
? "git clone timed out"
|
||||
: "git clone failed";
|
||||
|
||||
/**
|
||||
* `cloneRepo`'s `launch`: the same git command, run by a detached runner
|
||||
* instead of as this process's child. Rejects like `worktree.git()`.
|
||||
*/
|
||||
function launcher(job) {
|
||||
return async (cwd, args, opts) => {
|
||||
const { cmd, argv, env } = require("./worktree").gitCommand(cwd, args, {
|
||||
env: opts.env,
|
||||
background: true,
|
||||
});
|
||||
spawnRunner(job, { cmd, argv, cwd, timeoutMs: opts.timeout || 0 }, env);
|
||||
const out = await watch(job, opts.onStderr);
|
||||
if (out.ok) return { stdout: "", stderr: out.stderr };
|
||||
throw Object.assign(new Error(failureMessage(out)), { git: { stderr: out.stderr } });
|
||||
};
|
||||
}
|
||||
|
||||
// ── Lifecycle ──────────────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* A new `queued` job. `request` is reduced to `REQUEST_KEYS` — never a
|
||||
* credential; `once` records only whether a login was typed. The record is
|
||||
* also clone-queue's job: `withCloneSlot` marks it running and `report`
|
||||
* fills its progress, each through `onChange`.
|
||||
*/
|
||||
function create({ owner, kind, url, name, parent, target, laneId = null, request, once = null }) {
|
||||
load();
|
||||
const req = {};
|
||||
for (const k of REQUEST_KEYS) if (request?.[k] !== undefined) req[k] = request[k];
|
||||
const job = arm({
|
||||
id: crypto.randomBytes(16).toString("base64url"),
|
||||
owner: owner ?? null,
|
||||
kind,
|
||||
state: "queued",
|
||||
url,
|
||||
name,
|
||||
parent,
|
||||
target,
|
||||
laneId,
|
||||
phase: null,
|
||||
label: null,
|
||||
percent: null,
|
||||
detail: null,
|
||||
error: null,
|
||||
result: null,
|
||||
createdAt: new Date().toISOString(),
|
||||
finishedAt: null,
|
||||
request: req,
|
||||
pid: null,
|
||||
once: once
|
||||
? {
|
||||
used: true,
|
||||
remember: once.remember === true,
|
||||
member: once.member === true,
|
||||
host: once.host ?? null,
|
||||
}
|
||||
: null,
|
||||
});
|
||||
jobs.set(job.id, job);
|
||||
writeNow();
|
||||
return job;
|
||||
}
|
||||
|
||||
/**
|
||||
* Runs a queued job: clone-queue, then the clone through a runner (`once`:
|
||||
* the typed login, this process only), then the after-step.
|
||||
*/
|
||||
function start(job, once = null) {
|
||||
void (async () => {
|
||||
let outcome;
|
||||
try {
|
||||
await require("./git-identity").cloneRepo({
|
||||
job,
|
||||
url: job.url,
|
||||
parent: job.parent,
|
||||
name: job.name,
|
||||
once,
|
||||
partial: job.request?.partial === true,
|
||||
signal: job.signal,
|
||||
launch: launcher(job),
|
||||
});
|
||||
outcome = { ok: true };
|
||||
} catch (err) {
|
||||
outcome = { ok: false, err };
|
||||
}
|
||||
finish(job, outcome);
|
||||
})();
|
||||
return job;
|
||||
}
|
||||
|
||||
/** A runner this process did not start: counted, tailed, finished on exit. */
|
||||
function adopt(job) {
|
||||
job.release = queue().occupy();
|
||||
watch(job, (chunk) => queue().report(job, chunk)).then((out) => finishFromDisk(job, out));
|
||||
}
|
||||
|
||||
/**
|
||||
* Finishes a runner's job from its files alone (no `cloneRepo` closure: the
|
||||
* process that launched it is gone). A typed login a member's store kept for
|
||||
* this run only is taken out again, as `cloneRepo` would have.
|
||||
*/
|
||||
function finishFromDisk(job, out = null) {
|
||||
if (!out) {
|
||||
const exit = readExit(job.id) || {};
|
||||
let stderr = "";
|
||||
try {
|
||||
stderr = fs.readFileSync(path.join(jobDir(job.id), "stderr.log"), "utf8").slice(-TAIL_MAX);
|
||||
} catch {
|
||||
/* none */
|
||||
}
|
||||
out = {
|
||||
ok: exit.code === 0 && !exit.signal,
|
||||
timedOut: Boolean(exit.timedOut),
|
||||
lost: false,
|
||||
stderr,
|
||||
};
|
||||
}
|
||||
const identity = require("./git-identity");
|
||||
if (job.once?.used && job.once.member && !(out.ok && job.once.remember) && job.once.host) {
|
||||
try {
|
||||
identity.removeHttps(identity.accountForDir(job.parent), "https", job.once.host);
|
||||
} catch {
|
||||
/* best effort: the store is the member's own */
|
||||
}
|
||||
}
|
||||
finish(
|
||||
job,
|
||||
out.ok
|
||||
? { ok: true }
|
||||
: { ok: false, err: identity.cloneFailure(job.url, out.stderr, failureMessage(out)) }
|
||||
);
|
||||
}
|
||||
|
||||
/** The error body a failed job records: the route's own code and message. */
|
||||
function errorBody(err) {
|
||||
return {
|
||||
code: err?.code || "ECLONE",
|
||||
message: err?.message || "clone failed",
|
||||
...(err?.code === "ECREDENTIALS" ? { auth: { protocol: err.protocol, host: err.host } } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* The end of a job, once: the slot back, the runner's files gone, then
|
||||
* nothing (cancelled — its half-made folder dropped), the failure, or the
|
||||
* after-step and its outcome.
|
||||
*/
|
||||
function finish(job, { ok, err }) {
|
||||
if (job.finishing) return;
|
||||
job.finishing = true;
|
||||
job.release?.();
|
||||
job.release = null;
|
||||
rmJobDir(job.id);
|
||||
if (cancelled(job)) {
|
||||
dropCancelledTarget(job.target);
|
||||
return;
|
||||
}
|
||||
if (!ok) return fail(job, errorBody(err));
|
||||
let out;
|
||||
try {
|
||||
out = afterStep
|
||||
? afterStep(job)
|
||||
: { status: 500, json: { error: { code: "ECLONE", message: "no after-step registered" } } };
|
||||
} catch (e) {
|
||||
out = { status: 500, json: { error: { code: e.code || "ECLONE", message: e.message } } };
|
||||
}
|
||||
if (out.status < 300) {
|
||||
succeed(job, {
|
||||
laneId: out.laneId ?? null,
|
||||
path: job.target,
|
||||
...(out.created !== undefined ? { created: out.created } : {}),
|
||||
});
|
||||
} else fail(job, out.json?.error);
|
||||
}
|
||||
|
||||
/**
|
||||
* A cancelled clone of a HOST-account folder whose git left the target
|
||||
* behind (it was checked absent before the job): that exact path, never
|
||||
* through a link. A member's folder is git's own to clean (sudo relays the
|
||||
* SIGTERM) — the dashboard never deletes inside a member's home.
|
||||
*/
|
||||
function dropCancelledTarget(target) {
|
||||
try {
|
||||
if (require("./run-as").accountForPath(target)) return;
|
||||
if (!fs.lstatSync(target).isDirectory()) return;
|
||||
fs.rmSync(target, { recursive: true, force: true });
|
||||
} catch {
|
||||
/* nothing there, or not ours to remove */
|
||||
}
|
||||
}
|
||||
|
||||
function settle(job, fields) {
|
||||
if (job.state === "cancelled") return;
|
||||
Object.assign(job, fields, { finishedAt: new Date().toISOString() });
|
||||
if (job.mirror) queue().finish(job.mirror, job.state === "done");
|
||||
if (jobs?.get(job.id) === job) writeNow();
|
||||
}
|
||||
|
||||
/** The clone and its after-step worked: `result = {laneId, path, created?}`. */
|
||||
function succeed(job, result) {
|
||||
settle(job, { state: "done", error: null, result });
|
||||
}
|
||||
|
||||
/** It failed: `error = {code, message, auth?}` — what the route would answer. */
|
||||
function fail(job, error) {
|
||||
const e = { code: error?.code || "ECLONE", message: error?.message || "clone failed" };
|
||||
if (error?.auth) e.auth = { protocol: error.auth.protocol, host: error.auth.host };
|
||||
settle(job, { state: "failed", error: e, result: null });
|
||||
}
|
||||
|
||||
/**
|
||||
* Cancels one of the caller's jobs: `ENOTFOUND` when not theirs, `EDONE` once
|
||||
* it finished. Queued: out of the queue, never started. Running: the
|
||||
* runner's group is SIGTERMed (and a borrowed-login probe aborted); the slot
|
||||
* frees when git exits.
|
||||
*/
|
||||
function cancel(id, owner) {
|
||||
const job = get(id, owner);
|
||||
if (!job) throw Object.assign(new Error("no such clone"), { code: "ENOTFOUND" });
|
||||
if (!ACTIVE.has(job.state)) {
|
||||
throw Object.assign(new Error("the clone already finished"), { code: "EDONE" });
|
||||
}
|
||||
settle(job, {
|
||||
state: "cancelled",
|
||||
error: { code: "ECANCELLED", message: "Clone cancelled" },
|
||||
result: null,
|
||||
});
|
||||
if (!queue().dequeue(job)) {
|
||||
job.abort();
|
||||
killRunner(job);
|
||||
}
|
||||
return job;
|
||||
}
|
||||
|
||||
/** Whether a cancel already settled this job. */
|
||||
const cancelled = (job) => job.state === "cancelled";
|
||||
|
||||
const mine = (job, owner) => !serverMode() || job.owner === (owner ?? null);
|
||||
|
||||
/** The CloneJob the API answers. */
|
||||
function view(job) {
|
||||
const { owner: _owner, once: _once, ...out } = record(job);
|
||||
return {
|
||||
...out,
|
||||
position: job.state === "queued" ? queue().position(job) : null,
|
||||
};
|
||||
}
|
||||
|
||||
/** The caller's jobs, newest first. */
|
||||
function list(owner) {
|
||||
prune();
|
||||
return [...jobs.values()]
|
||||
.filter((j) => mine(j, owner))
|
||||
.sort((a, b) => (a.createdAt < b.createdAt ? 1 : a.createdAt > b.createdAt ? -1 : 0));
|
||||
}
|
||||
|
||||
/** One of the caller's jobs, or null. */
|
||||
function get(id, owner) {
|
||||
prune();
|
||||
const job = typeof id === "string" ? jobs.get(id) : null;
|
||||
return job && mine(job, owner) ? job : null;
|
||||
}
|
||||
|
||||
/** Drops a finished job: `ENOTFOUND` when not the caller's, `EBUSY` while it runs. */
|
||||
function remove(id, owner) {
|
||||
const job = get(id, owner);
|
||||
if (!job) throw Object.assign(new Error("no such clone"), { code: "ENOTFOUND" });
|
||||
if (ACTIVE.has(job.state)) {
|
||||
throw Object.assign(new Error("the clone is still running"), { code: "EBUSY" });
|
||||
}
|
||||
jobs.delete(id);
|
||||
rmJobDir(id);
|
||||
writeNow();
|
||||
}
|
||||
|
||||
/** The routes' after-step: `(job) => {status, json, laneId?, created?}`. */
|
||||
function setAfterStep(fn) {
|
||||
afterStep = fn;
|
||||
}
|
||||
|
||||
/** Server start: reconcile the file now (adopt, finish, interrupt, requeue). */
|
||||
function resume() {
|
||||
load();
|
||||
}
|
||||
|
||||
/** Tests: forget the in-memory state and stop tailing, as a restart would. */
|
||||
function _reset() {
|
||||
if (timer) clearTimeout(timer);
|
||||
timer = null;
|
||||
for (const w of watchers) clearInterval(w.iv);
|
||||
watchers.clear();
|
||||
jobs = null;
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
create,
|
||||
start,
|
||||
view,
|
||||
list,
|
||||
get,
|
||||
remove,
|
||||
cancel,
|
||||
cancelled,
|
||||
setAfterStep,
|
||||
resume,
|
||||
filePath,
|
||||
jobDir,
|
||||
ACTIVE,
|
||||
KEEP_DONE_MS,
|
||||
KEEP_FAILED_MS,
|
||||
_reset,
|
||||
};
|
||||
@@ -9,6 +9,8 @@
|
||||
* least 1). A clone the browser named with a token (`track`) reports its
|
||||
* place in the queue, then git's own `--progress` lines (`report`), to
|
||||
* `GET /api/lanes/clone/progress/:token` — only to the user who started it.
|
||||
* A background clone (`server/lib/clone-jobs.js`) passes its own job record,
|
||||
* whose `onChange` hears each state change and progress line.
|
||||
* @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
|
||||
*/
|
||||
|
||||
@@ -46,15 +48,18 @@ function next() {
|
||||
while (running < concurrency() && waiting.length) {
|
||||
running++;
|
||||
const { start, job } = waiting.shift();
|
||||
if (job) job.state = "running";
|
||||
if (job) {
|
||||
job.state = "running";
|
||||
job.onChange?.("state");
|
||||
}
|
||||
start();
|
||||
}
|
||||
}
|
||||
|
||||
/** Runs `fn` once a clone slot is free; resolves or rejects with it. */
|
||||
async function withCloneSlot(fn, job = null) {
|
||||
await new Promise((start) => {
|
||||
waiting.push({ start, job });
|
||||
await new Promise((start, reject) => {
|
||||
waiting.push({ start, reject, job });
|
||||
next();
|
||||
});
|
||||
try {
|
||||
@@ -136,6 +141,7 @@ function report(job, chunk) {
|
||||
.replace(/,\s*done\.?$/, "")
|
||||
.replace(/^,\s*/, "")
|
||||
.trim() || null;
|
||||
job.onChange?.("progress");
|
||||
return;
|
||||
}
|
||||
}
|
||||
@@ -149,10 +155,9 @@ function progress(token, owner) {
|
||||
sweep();
|
||||
const job = jobs.get(token);
|
||||
if (!job || job.owner !== owner) return null;
|
||||
const position = waiting.findIndex((w) => w.job === job);
|
||||
return {
|
||||
state: job.state,
|
||||
position: position >= 0 ? position + 1 : null,
|
||||
position: position(job),
|
||||
...cloneQueueState(),
|
||||
phase: job.phase,
|
||||
label: job.label,
|
||||
@@ -161,6 +166,40 @@ function progress(token, owner) {
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Counts a clone this process did not start — a background clone's runner
|
||||
* adopted after a restart (clone-jobs.js) — against the limit, at once.
|
||||
* Returns its release.
|
||||
*/
|
||||
function occupy() {
|
||||
running++;
|
||||
let released = false;
|
||||
return () => {
|
||||
if (released) return;
|
||||
released = true;
|
||||
running--;
|
||||
next();
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Takes a still-waiting clone out of the queue: its `withCloneSlot` rejects
|
||||
* with `ECANCELLED` and it never starts. False when it is not waiting.
|
||||
*/
|
||||
function dequeue(job) {
|
||||
const i = waiting.findIndex((w) => w.job === job);
|
||||
if (i < 0) return false;
|
||||
const [w] = waiting.splice(i, 1);
|
||||
w.reject(Object.assign(new Error("Clone cancelled"), { code: "ECANCELLED" }));
|
||||
return true;
|
||||
}
|
||||
|
||||
/** A queued clone's 1-based place among the waiting ones, else null. */
|
||||
function position(job) {
|
||||
const i = waiting.findIndex((w) => w.job === job);
|
||||
return i >= 0 ? i + 1 : null;
|
||||
}
|
||||
|
||||
/** A progress-free tail of git's stderr for an error message. */
|
||||
function stripProgress(stderr) {
|
||||
return String(stderr || "")
|
||||
@@ -178,5 +217,8 @@ module.exports = {
|
||||
finish,
|
||||
report,
|
||||
progress,
|
||||
position,
|
||||
dequeue,
|
||||
occupy,
|
||||
stripProgress,
|
||||
};
|
||||
|
||||
+46
-20
@@ -938,6 +938,32 @@ function sshFailure(stderr) {
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* The error a failed `git clone` of `url` answers, from git's stderr:
|
||||
* `ECREDENTIALS` (with `{protocol, host}`), `ENOREPO`, `ESSHAUTH`, else
|
||||
* `ECLONE` with git's own words. Shared by the held-open clone and a
|
||||
* background one finished by a later server (clone-jobs.js).
|
||||
*/
|
||||
function cloneFailure(url, stderr, fallback = "git clone failed") {
|
||||
const remote = remoteOf(url);
|
||||
if (remote.scheme === "https" && httpsFailure(stderr)) {
|
||||
return Object.assign(err("ECREDENTIALS", "this repository needs a login"), {
|
||||
protocol: "https",
|
||||
host: remote.host,
|
||||
});
|
||||
}
|
||||
if (notFound(stderr)) {
|
||||
return err("ENOREPO", `no repository at ${url} (or this login cannot see it)`);
|
||||
}
|
||||
if (remote.scheme === "ssh" && sshFailure(stderr)) {
|
||||
return Object.assign(err("ESSHAUTH", "the SSH login was refused"), { host: remote.host });
|
||||
}
|
||||
return err(
|
||||
"ECLONE",
|
||||
(require("./clone-queue").stripProgress(stderr) || fallback).trim().slice(-800)
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Clone `url` into `<parent>/<name>` as the account owning `parent`, with
|
||||
* that account's own identity. `once` is a login typed for this clone
|
||||
@@ -962,7 +988,13 @@ function cloneRepo({ job = null, ...opts }) {
|
||||
);
|
||||
}
|
||||
|
||||
async function cloneNow({ url, parent, name, once = null, partial = false }, threads, onProgress) {
|
||||
// `signal` (a background clone's cancel, clone-jobs.js) SIGTERMs whatever git
|
||||
// call is running: a borrowed-login probe or the clone itself.
|
||||
async function cloneNow(
|
||||
{ url, parent, name, once = null, partial = false, signal = null, launch = null },
|
||||
threads,
|
||||
onProgress
|
||||
) {
|
||||
const { git } = require("./worktree");
|
||||
const account = accountForDir(parent);
|
||||
const remote = remoteOf(url);
|
||||
@@ -996,6 +1028,7 @@ async function cloneNow({ url, parent, name, once = null, partial = false }, thr
|
||||
borrowed.length === 1 ||
|
||||
(await git(parent, ["-c", sshFor(b.keyFile), "ls-remote", "--", b.url, "HEAD"], {
|
||||
timeout: 60_000,
|
||||
signal,
|
||||
}).then(
|
||||
() => true,
|
||||
() => false
|
||||
@@ -1055,33 +1088,25 @@ async function cloneNow({ url, parent, name, once = null, partial = false }, thr
|
||||
|
||||
let ok = false;
|
||||
try {
|
||||
if (signal?.aborted) throw err("ECANCELLED", "Clone cancelled");
|
||||
const filter = partial ? ["--filter=blob:none"] : [];
|
||||
// --progress: git prints it to a pipe too, for the UI's progress line.
|
||||
await git(parent, [...pre, "clone", ...persist, "--progress", ...filter, "--", url, name], {
|
||||
// `launch` (a background clone, clone-jobs.js) runs this same command in
|
||||
// a detached runner instead, and rejects like git() does.
|
||||
const args = [...pre, "clone", ...persist, "--progress", ...filter, "--", url, name];
|
||||
const gitOpts = {
|
||||
timeout: CLONE_TIMEOUT_MS,
|
||||
env,
|
||||
background: true,
|
||||
onStderr: onProgress,
|
||||
});
|
||||
signal,
|
||||
};
|
||||
await (launch ? launch(parent, args, gitOpts) : git(parent, args, gitOpts));
|
||||
ok = true;
|
||||
} catch (e) {
|
||||
const stderr = e.git?.stderr;
|
||||
if (remote.scheme === "https" && httpsFailure(stderr)) {
|
||||
throw Object.assign(err("ECREDENTIALS", "this repository needs a login"), {
|
||||
protocol: "https",
|
||||
host: remote.host,
|
||||
});
|
||||
}
|
||||
if (notFound(stderr)) {
|
||||
throw err("ENOREPO", `no repository at ${url} (or this login cannot see it)`);
|
||||
}
|
||||
if (remote.scheme === "ssh" && sshFailure(stderr)) {
|
||||
throw Object.assign(err("ESSHAUTH", "the SSH login was refused"), { host: remote.host });
|
||||
}
|
||||
throw err(
|
||||
"ECLONE",
|
||||
(require("./clone-queue").stripProgress(stderr) || e.message).trim().slice(-800)
|
||||
);
|
||||
// SIGTERMed by a cancel: git (sudo forwards it) removes its own half-made folder.
|
||||
if (signal?.aborted) throw err("ECANCELLED", "Clone cancelled");
|
||||
throw cloneFailure(url, e.git?.stderr, e.message);
|
||||
} finally {
|
||||
// A login typed without "remember" — or one that did not work — leaves
|
||||
// the member's store as it was.
|
||||
@@ -1122,6 +1147,7 @@ module.exports = {
|
||||
remoteOf,
|
||||
defaultName,
|
||||
cloneRepo,
|
||||
cloneFailure,
|
||||
getAuthor,
|
||||
defaultEmail,
|
||||
setAuthor,
|
||||
|
||||
+42
-27
@@ -40,34 +40,12 @@ function lowPriority() {
|
||||
}
|
||||
|
||||
/**
|
||||
* Promisified git wrapper. On failure, throws an Error with err.git = { args, code, stderr }.
|
||||
* Treats zero exit code as success even if stderr has hints.
|
||||
*
|
||||
* CRITICAL: Scrubs git hook environment variables (GIT_DIR, GIT_INDEX_FILE, etc.)
|
||||
* that leak from parent processes. Without this, git operations on a worktree (where
|
||||
* .git is a file, not a directory) fail with ".git/index: index file open failed:
|
||||
* Not a directory" when run from within a git hook or from a shell that inherited
|
||||
* these variables. This module's whole job is to run git safely against repos other
|
||||
* than the one enclosing the current working directory.
|
||||
* The command a git call runs — `{cmd, argv, env, gitCmd}` — exactly as
|
||||
* `git()` builds it (scrubbed env, a member's `sudo -n -u ah-<name> git`,
|
||||
* nice/ionice for `background`), without running it. A background clone's
|
||||
* detached runner (`scripts/clone-runner.js`) executes this very command.
|
||||
*/
|
||||
async function git(
|
||||
cwd,
|
||||
args,
|
||||
{
|
||||
timeout = 0,
|
||||
maxBuffer = 8 * 1024 * 1024,
|
||||
literal = false,
|
||||
env: extraEnv = {},
|
||||
// "buffer" for a binary read (an image at a revision).
|
||||
encoding = "utf8",
|
||||
// stdin (a patch for `apply -`): never a file the dashboard writes.
|
||||
input = null,
|
||||
// true: run under nice/ionice (a clone — see lowPriority above).
|
||||
background = false,
|
||||
// Called with each stderr chunk as it arrives (a clone's --progress).
|
||||
onStderr = null,
|
||||
} = {}
|
||||
) {
|
||||
function gitCommand(cwd, args, { literal = false, env: extraEnv = {}, background = false } = {}) {
|
||||
// Build a clean environment: copy process.env but scrub git hook variables
|
||||
// that could point to the outer repo's git directory or index.
|
||||
const env = { ...process.env };
|
||||
@@ -118,6 +96,41 @@ async function git(
|
||||
argv = [...rest, cmd, ...argv];
|
||||
cmd = first;
|
||||
}
|
||||
return { cmd, argv, env, gitCmd };
|
||||
}
|
||||
|
||||
/**
|
||||
* Promisified git wrapper. On failure, throws an Error with err.git = { args, code, stderr }.
|
||||
* Treats zero exit code as success even if stderr has hints.
|
||||
*
|
||||
* CRITICAL: Scrubs git hook environment variables (GIT_DIR, GIT_INDEX_FILE, etc.)
|
||||
* that leak from parent processes. Without this, git operations on a worktree (where
|
||||
* .git is a file, not a directory) fail with ".git/index: index file open failed:
|
||||
* Not a directory" when run from within a git hook or from a shell that inherited
|
||||
* these variables. This module's whole job is to run git safely against repos other
|
||||
* than the one enclosing the current working directory.
|
||||
*/
|
||||
async function git(
|
||||
cwd,
|
||||
args,
|
||||
{
|
||||
timeout = 0,
|
||||
maxBuffer = 8 * 1024 * 1024,
|
||||
literal = false,
|
||||
env: extraEnv = {},
|
||||
// "buffer" for a binary read (an image at a revision).
|
||||
encoding = "utf8",
|
||||
// stdin (a patch for `apply -`): never a file the dashboard writes.
|
||||
input = null,
|
||||
// true: run under nice/ionice (a clone — see lowPriority above).
|
||||
background = false,
|
||||
// Called with each stderr chunk as it arrives (a clone's --progress).
|
||||
onStderr = null,
|
||||
// An AbortSignal: aborting SIGTERMs the child (a cancelled clone).
|
||||
signal = null,
|
||||
} = {}
|
||||
) {
|
||||
const { cmd, argv, env, gitCmd } = gitCommand(cwd, args, { literal, env: extraEnv, background });
|
||||
|
||||
try {
|
||||
const run =
|
||||
@@ -142,6 +155,7 @@ async function git(
|
||||
// open forever; GIT_TERMINAL_PROMPT=0 only covers the prompting case.
|
||||
timeout,
|
||||
encoding,
|
||||
...(signal ? { signal } : {}),
|
||||
});
|
||||
// promisify(execFile) hands back its child; the stdin runner has none.
|
||||
if (onStderr && pending.child?.stderr) {
|
||||
@@ -1006,6 +1020,7 @@ async function hasNoRemotes(dir) {
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
gitCommand,
|
||||
LANES_ROOT,
|
||||
git,
|
||||
isGitRepo,
|
||||
|
||||
@@ -316,10 +316,83 @@ const paths = {
|
||||
description:
|
||||
"The caller's name for this clone, to poll `/api/lanes/clone/progress/{token}`.",
|
||||
},
|
||||
background: {
|
||||
type: "boolean",
|
||||
description:
|
||||
"Run every check now, then clone as a background job that outlives the request and a server restart (a detached runner): 202 `{job}`. 400 EBADFIELD when not a boolean.",
|
||||
},
|
||||
},
|
||||
["url", "parent"]
|
||||
),
|
||||
responses: { 201: { description: "{lane, created: true, path}" } },
|
||||
responses: {
|
||||
201: { description: "{lane, created: true, path}" },
|
||||
202: {
|
||||
description: "`background: true`: {job} — a CloneJob (see `/api/lanes/clone/jobs`).",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
"/api/lanes/clone/jobs": {
|
||||
get: {
|
||||
tags: ["Git identity"],
|
||||
summary: "The caller's background clone jobs, newest first",
|
||||
description:
|
||||
"CloneJob: {id, kind: workspace|source, state: queued|running|done|failed|interrupted|cancelled, url, name, parent, target, laneId, position (1-based while queued), phase, label, percent, detail, error: {code, message, auth?} | null, result: {laneId, path, created?} | null, createdAt, finishedAt, request: {url, parent, name, tool?, title?, partial?}, pid}. `request` never holds a credential. Kept in `<data dir>/runtime/clone-jobs.json`; a job whose runner still runs after a restart is adopted, a dead one becomes `interrupted` (EINTERRUPTED). Done jobs are dropped after 24 h, the rest after 7 days. Only the caller's own (local mode: all).",
|
||||
operationId: "listCloneJobs",
|
||||
responses: { 200: { description: "{jobs: CloneJob[]}" } },
|
||||
},
|
||||
},
|
||||
"/api/lanes/clone/jobs/{id}": {
|
||||
delete: {
|
||||
tags: ["Git identity"],
|
||||
summary: "Dismiss a finished background clone job",
|
||||
operationId: "dismissCloneJob",
|
||||
parameters: [{ name: "id", in: "path", required: true, schema: { type: "string" } }],
|
||||
responses: {
|
||||
200: { description: "{ok: true}" },
|
||||
404: { description: "ENOTFOUND — not the caller's job" },
|
||||
409: { description: "EBUSY — still queued or running" },
|
||||
},
|
||||
},
|
||||
},
|
||||
"/api/lanes/clone/jobs/{id}/cancel": {
|
||||
post: {
|
||||
tags: ["Git identity"],
|
||||
summary: "Cancel a queued or running background clone",
|
||||
description:
|
||||
"Queued: taken out of the queue, never started. Running: the runner's process group is SIGTERMed (git removes its half-made folder). The job becomes `cancelled` with ECANCELLED and never makes its workspace / extra folder. Same-origin.",
|
||||
operationId: "cancelCloneJob",
|
||||
parameters: [{ name: "id", in: "path", required: true, schema: { type: "string" } }],
|
||||
responses: {
|
||||
200: { description: "{job}" },
|
||||
404: { description: "ENOTFOUND — not the caller's job" },
|
||||
409: { description: "EDONE — it already finished" },
|
||||
},
|
||||
},
|
||||
},
|
||||
"/api/lanes/clone/jobs/{id}/retry": {
|
||||
post: {
|
||||
tags: ["Git identity"],
|
||||
summary: "Retry a failed, interrupted or cancelled background clone",
|
||||
description:
|
||||
"A NEW job from the old one's `request` (and kind / workspace), through every check of a fresh clone POST; the old job is dropped once the new one exists. `credentials` as on the clone POST, for this run only. Same-origin.",
|
||||
operationId: "retryCloneJob",
|
||||
parameters: [{ name: "id", in: "path", required: true, schema: { type: "string" } }],
|
||||
requestBody: json({
|
||||
credentials: {
|
||||
type: "object",
|
||||
properties: {
|
||||
username: { type: "string" },
|
||||
password: { type: "string" },
|
||||
remember: { type: "boolean" },
|
||||
},
|
||||
},
|
||||
}),
|
||||
responses: {
|
||||
202: { description: "{job} — the new CloneJob" },
|
||||
404: { description: "ENOTFOUND — not the caller's job (ENOLANE: its workspace is gone)" },
|
||||
409: { description: "EBUSY — queued, running or done" },
|
||||
},
|
||||
},
|
||||
},
|
||||
"/api/lanes/clone/progress/{token}": {
|
||||
|
||||
@@ -660,11 +660,17 @@ const paths = {
|
||||
additionalProperties: true,
|
||||
description: "A typed HTTPS login for this one clone.",
|
||||
},
|
||||
background: {
|
||||
type: "boolean",
|
||||
description:
|
||||
'Run as a background clone job (see `/api/lanes/clone/jobs`): the checks answer at once, then 202 `{job}` with `kind: "source"`.',
|
||||
},
|
||||
},
|
||||
}),
|
||||
},
|
||||
responses: {
|
||||
201: { description: "The context view plus `path` (the clone).", ...json(obj) },
|
||||
202: { description: "`background: true`: {job} — a CloneJob." },
|
||||
400: {
|
||||
description:
|
||||
"EUNSUPPORTED (tool takes no extra folders), a URL/name refusal, or a source check (the clone stays, `path` names it).",
|
||||
|
||||
+350
-85
@@ -738,13 +738,32 @@ router.post("/browse/mkdir", sameOriginGuard, (req, res) => {
|
||||
* `EGITAUTH`, never 401, which the client reads as "signed out".
|
||||
*/
|
||||
/**
|
||||
* The clone itself, shared by Add workspace → Clone and a workspace's Clone
|
||||
* extra folder: every check above, then `git clone` as the folder's account.
|
||||
* Answers the new folder's path, or null once it has sent the refusal.
|
||||
* `before(target, ownerId)` runs after the checks and BEFORE the clone, so a
|
||||
* refusal there never leaves minutes of cloning behind.
|
||||
* Status + error body a failed clone answers — the same codes whether the
|
||||
* request was held open or the clone ran as a background job.
|
||||
*/
|
||||
async function cloneInto(req, res, body, before) {
|
||||
function cloneFailure(err) {
|
||||
const status = ["EBADCRED", "EBADHOST", "EBADTOOL", "EDISABLED"].includes(err.code)
|
||||
? 400
|
||||
: ["ECREDENTIALS", "ESSHAUTH", "ENOREPO", "EOWNERFOLDER", "EMEMBERGONE"].includes(err.code)
|
||||
? 409
|
||||
: 500;
|
||||
return {
|
||||
status,
|
||||
error: {
|
||||
code: err.code || "ECLONE",
|
||||
message: err.message,
|
||||
...(err.code === "ECREDENTIALS" ? { auth: { protocol: err.protocol, host: err.host } } : {}),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Every check before a clone, shared by Add workspace → Clone and a
|
||||
* workspace's Clone extra folder, then `before(target, ownerId)` — so a
|
||||
* refusal there never leaves minutes of cloning behind. Answers the clone's
|
||||
* context, or null once it has sent the refusal.
|
||||
*/
|
||||
async function prepareClone(req, res, body, before) {
|
||||
const identity = require("../lib/git-identity");
|
||||
let url;
|
||||
try {
|
||||
@@ -783,6 +802,10 @@ async function cloneInto(req, res, body, before) {
|
||||
res.status(400).json({ error: { code: "EBADFIELD", message: "partial: boolean" } });
|
||||
return null;
|
||||
}
|
||||
if (body.background !== undefined && typeof body.background !== "boolean") {
|
||||
res.status(400).json({ error: { code: "EBADFIELD", message: "background: boolean" } });
|
||||
return null;
|
||||
}
|
||||
const once = body.credentials;
|
||||
if (once != null && (typeof once !== "object" || Array.isArray(once))) {
|
||||
res.status(400).json({ error: { code: "EBADCRED", message: "credentials: object" } });
|
||||
@@ -832,38 +855,141 @@ async function cloneInto(req, res, body, before) {
|
||||
}
|
||||
try {
|
||||
await before(target, ownerId);
|
||||
await identity.cloneRepo({
|
||||
job,
|
||||
url,
|
||||
parent: dir,
|
||||
name,
|
||||
once: once
|
||||
? { username: once.username, password: once.password, remember: once.remember === true }
|
||||
: null,
|
||||
partial: body.partial === true,
|
||||
});
|
||||
queue.finish(job, true);
|
||||
} catch (err) {
|
||||
queue.finish(job, false);
|
||||
const status = ["EBADCRED", "EBADHOST", "EBADTOOL", "EDISABLED"].includes(err.code)
|
||||
? 400
|
||||
: ["ECREDENTIALS", "ESSHAUTH", "ENOREPO", "EOWNERFOLDER", "EMEMBERGONE"].includes(err.code)
|
||||
? 409
|
||||
: 500;
|
||||
res.status(status).json({
|
||||
error: {
|
||||
code: err.code || "ECLONE",
|
||||
message: err.message,
|
||||
...(err.code === "ECREDENTIALS"
|
||||
? { auth: { protocol: err.protocol, host: err.host } }
|
||||
: {}),
|
||||
},
|
||||
});
|
||||
const f = cloneFailure(err);
|
||||
res.status(f.status).json({ error: f.error });
|
||||
return null;
|
||||
}
|
||||
return { target, name, ownerId };
|
||||
return {
|
||||
url,
|
||||
name,
|
||||
dir,
|
||||
target,
|
||||
ownerId,
|
||||
token: job,
|
||||
once: once
|
||||
? { username: once.username, password: once.password, remember: once.remember === true }
|
||||
: null,
|
||||
partial: body.partial === true,
|
||||
};
|
||||
}
|
||||
|
||||
/** `git clone` as the folder's account, the request held open. */
|
||||
async function runClone(ctx) {
|
||||
const queue = require("../lib/clone-queue");
|
||||
try {
|
||||
await require("../lib/git-identity").cloneRepo({
|
||||
job: ctx.token,
|
||||
url: ctx.url,
|
||||
parent: ctx.dir,
|
||||
name: ctx.name,
|
||||
once: ctx.once,
|
||||
partial: ctx.partial,
|
||||
});
|
||||
queue.finish(ctx.token, true);
|
||||
} catch (err) {
|
||||
queue.finish(ctx.token, false);
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
/** The held-open clone: its context, or null once it has sent the refusal. */
|
||||
async function cloneInto(req, res, body, before) {
|
||||
const ctx = await prepareClone(req, res, body, before);
|
||||
if (!ctx) return null;
|
||||
try {
|
||||
await runClone(ctx);
|
||||
} catch (err) {
|
||||
const f = cloneFailure(err);
|
||||
res.status(f.status).json({ error: f.error });
|
||||
return null;
|
||||
}
|
||||
return ctx;
|
||||
}
|
||||
|
||||
/**
|
||||
* Answers 202 `{ job }` and hands the clone to `server/lib/clone-jobs.js`:
|
||||
* a detached runner runs git, and when it ends the job's after-step
|
||||
* (`afterStepForJob` below — the workspace, or the extra folder, exactly as
|
||||
* the held-open route) runs once. A cancelled job never runs it.
|
||||
*/
|
||||
function startBackground(res, ctx, { kind, laneId, request }) {
|
||||
const cloneJobs = require("../lib/clone-jobs");
|
||||
let account = null;
|
||||
try {
|
||||
account = require("../lib/run-as").accountForPath(ctx.target);
|
||||
} catch {
|
||||
/* checked by prepareClone already */
|
||||
}
|
||||
const job = cloneJobs.create({
|
||||
owner: ctx.ownerId,
|
||||
kind,
|
||||
url: ctx.url,
|
||||
name: ctx.name,
|
||||
parent: ctx.dir,
|
||||
target: ctx.target,
|
||||
laneId,
|
||||
request,
|
||||
once: ctx.once
|
||||
? {
|
||||
remember: ctx.once.remember,
|
||||
member: Boolean(account),
|
||||
host: (() => {
|
||||
try {
|
||||
return require("../lib/git-identity").remoteOf(ctx.url).host;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
})(),
|
||||
}
|
||||
: null,
|
||||
});
|
||||
if (ctx.token) job.mirror = ctx.token;
|
||||
// Started before the answer: its first, synchronous step queues it, so the
|
||||
// 202 already carries its place in the queue.
|
||||
cloneJobs.start(job, ctx.once);
|
||||
res.status(202).json({ job: cloneJobs.view(job) });
|
||||
return job;
|
||||
}
|
||||
|
||||
/** A request-shaped owner for a job finished without its request (scope checks). */
|
||||
function reqForOwner(owner) {
|
||||
if (owner == null) return { method: "POST" };
|
||||
const row = db.prepare("SELECT * FROM users WHERE id = ?").get(owner);
|
||||
if (!row || row.disabled) return null;
|
||||
return { method: "POST", user: require("../lib/auth").publicUser(row) };
|
||||
}
|
||||
|
||||
/**
|
||||
* A background clone's after-step, from the job alone — it may run in a
|
||||
* later server process than the one the request reached.
|
||||
*/
|
||||
function afterStepForJob(job) {
|
||||
const req = reqForOwner(job.owner);
|
||||
if (!req) {
|
||||
return {
|
||||
status: 409,
|
||||
json: {
|
||||
error: { code: "ENOTFOUND", message: "the account that started this clone is gone" },
|
||||
},
|
||||
};
|
||||
}
|
||||
if (job.kind === "source") {
|
||||
const lane = lanesLib.getLane(job.laneId);
|
||||
if (!lane) {
|
||||
return { status: 404, json: { error: { code: "ENOLANE", message: "workspace not found" } } };
|
||||
}
|
||||
return finishSourceClone(req, lane, job.target);
|
||||
}
|
||||
return finishWorkspaceClone(
|
||||
req,
|
||||
{ tool: job.request?.tool, title: job.request?.title },
|
||||
{ target: job.target, name: job.name, ownerId: job.owner }
|
||||
);
|
||||
}
|
||||
require("../lib/clone-jobs").setAfterStep(afterStepForJob);
|
||||
|
||||
/**
|
||||
* A clone's place in the machine-wide queue, then git's own progress
|
||||
* (`server/lib/clone-queue.js`), by the `progressToken` its POST carried —
|
||||
@@ -877,17 +1003,97 @@ router.get("/clone/progress/:token", (req, res) => {
|
||||
res.json(view);
|
||||
});
|
||||
|
||||
router.post("/clone", sameOriginGuard, async (req, res) => {
|
||||
const body = req.body || {};
|
||||
// Before the clone: a workspace this user may not own there, or a tool
|
||||
// that is disabled, is refused up front — not after minutes of cloning,
|
||||
// leaving a clone with no workspace on it.
|
||||
const done = await cloneInto(req, res, body, (target, ownerId) => {
|
||||
lanesLib.validateTool(body.tool || "claude");
|
||||
require("../lib/run-as").assertOwnerFits(target, ownerId);
|
||||
});
|
||||
if (!done) return;
|
||||
const { target, name, ownerId } = done;
|
||||
/**
|
||||
* Background clones (`background: true` on either clone POST,
|
||||
* `server/lib/clone-jobs.js`): the caller's own jobs, newest first — a job id
|
||||
* is only ever looked up among them (404 otherwise).
|
||||
*/
|
||||
router.get("/clone/jobs", (req, res) => {
|
||||
const cloneJobs = require("../lib/clone-jobs");
|
||||
res.json({ jobs: cloneJobs.list(scope.ownerFor(req)).map(cloneJobs.view) });
|
||||
});
|
||||
|
||||
const cloneJobError = (res, err) => {
|
||||
const status = err.code === "ENOTFOUND" ? 404 : 409;
|
||||
return res.status(status).json({ error: { code: err.code, message: err.message } });
|
||||
};
|
||||
|
||||
router.delete("/clone/jobs/:id", sameOriginGuard, (req, res) => {
|
||||
try {
|
||||
require("../lib/clone-jobs").remove(req.params.id, scope.ownerFor(req));
|
||||
return res.json({ ok: true });
|
||||
} catch (err) {
|
||||
return cloneJobError(res, err);
|
||||
}
|
||||
});
|
||||
|
||||
router.post("/clone/jobs/:id/cancel", sameOriginGuard, (req, res) => {
|
||||
const cloneJobs = require("../lib/clone-jobs");
|
||||
try {
|
||||
const job = cloneJobs.cancel(req.params.id, scope.ownerFor(req));
|
||||
return res.json({ job: cloneJobs.view(job) });
|
||||
} catch (err) {
|
||||
return cloneJobError(res, err);
|
||||
}
|
||||
});
|
||||
|
||||
/** Jobs a retry is starting from, so a double click starts one clone. */
|
||||
const retrying = new Set();
|
||||
|
||||
/**
|
||||
* A failed, interrupted or cancelled job again: a NEW job from its `request`
|
||||
* (plus `credentials` typed now), through every check of a fresh POST; the
|
||||
* old one is dropped once the new one exists.
|
||||
*/
|
||||
router.post("/clone/jobs/:id/retry", sameOriginGuard, async (req, res) => {
|
||||
const cloneJobs = require("../lib/clone-jobs");
|
||||
const owner = scope.ownerFor(req);
|
||||
const old = cloneJobs.get(req.params.id, owner);
|
||||
if (!old) {
|
||||
return res.status(404).json({ error: { code: "ENOTFOUND", message: "no such clone" } });
|
||||
}
|
||||
if (cloneJobs.ACTIVE.has(old.state) || old.state === "done" || retrying.has(old.id)) {
|
||||
return res
|
||||
.status(409)
|
||||
.json({ error: { code: "EBUSY", message: "only a failed clone can be retried" } });
|
||||
}
|
||||
const body = { ...old.request, background: true };
|
||||
const typed = req.body?.credentials;
|
||||
if (typed !== undefined) body.credentials = typed;
|
||||
retrying.add(old.id);
|
||||
try {
|
||||
let job;
|
||||
if (old.kind === "source") {
|
||||
const lane = lanesLib.getLane(old.laneId);
|
||||
if (!lane || !scope.canSeeLane(req, lane.id)) {
|
||||
return res.status(404).json({ error: { code: "ENOLANE", message: "workspace not found" } });
|
||||
}
|
||||
// scopeGuard's live-role check for a write to a workspace, which this
|
||||
// path (no lane id in it) does not get.
|
||||
try {
|
||||
require("../lib/run-as").assertOwnerFits(lane.cwd, lane.owner_id);
|
||||
} catch (err) {
|
||||
return res.status(409).json({ error: { code: "EOWNERFOLDER", message: err.message } });
|
||||
}
|
||||
job = await startSourceClone(req, res, lane, body);
|
||||
} else {
|
||||
job = await startWorkspaceClone(req, res, body);
|
||||
}
|
||||
if (job) {
|
||||
try {
|
||||
cloneJobs.remove(old.id, owner);
|
||||
} catch {
|
||||
/* dismissed meanwhile */
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
retrying.delete(old.id);
|
||||
}
|
||||
});
|
||||
|
||||
/** After a workspace clone: the folder becomes (or re-attaches) a workspace. */
|
||||
function finishWorkspaceClone(req, body, ctx) {
|
||||
const { target, name, ownerId } = ctx;
|
||||
const cwd = lanesLib.canonicalCwd(target);
|
||||
// A workspace can outlive its folder (deleted from the File Explorer or a
|
||||
// shell): cloning into that path again re-attaches the user's own record
|
||||
@@ -896,7 +1102,12 @@ router.post("/clone", sameOriginGuard, async (req, res) => {
|
||||
const found = lanesLib.resolveLaneByCwd(cwd, body.tool || "claude");
|
||||
if (found && strip(found.cwd) === strip(cwd) && scope.canSeeLane(req, found.id)) {
|
||||
broadcastLane(found.id);
|
||||
return res.status(200).json({ lane: payload(found), created: false, path: target });
|
||||
return {
|
||||
status: 200,
|
||||
laneId: found.id,
|
||||
created: false,
|
||||
json: { lane: payload(found), created: false, path: target },
|
||||
};
|
||||
}
|
||||
try {
|
||||
const lane = lanesLib.createLane({
|
||||
@@ -906,16 +1117,106 @@ router.post("/clone", sameOriginGuard, async (req, res) => {
|
||||
...(body.tool ? { tool: body.tool } : {}),
|
||||
});
|
||||
broadcastLane(lane.id);
|
||||
return res.status(201).json({ lane: payload(lane), created: true, path: target });
|
||||
return {
|
||||
status: 201,
|
||||
laneId: lane.id,
|
||||
created: true,
|
||||
json: { lane: payload(lane), created: true, path: target },
|
||||
};
|
||||
} catch (err) {
|
||||
// The clone stays: it is the user's now, and Repo mode can adopt it.
|
||||
const status = ["EBADCWD", "EBADTOOL", "EDISABLED"].includes(err.code) ? 400 : 500;
|
||||
return res
|
||||
.status(status)
|
||||
.json({ error: { code: err.code || "ELANE", message: err.message }, path: target });
|
||||
return {
|
||||
status,
|
||||
json: { error: { code: err.code || "ELANE", message: err.message }, path: target },
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/** Add workspace → Clone; the background job when `background: true`. */
|
||||
async function startWorkspaceClone(req, res, body) {
|
||||
// Before the clone: a workspace this user may not own there, or a tool
|
||||
// that is disabled, is refused up front — not after minutes of cloning,
|
||||
// leaving a clone with no workspace on it.
|
||||
const before = (target, ownerId) => {
|
||||
lanesLib.validateTool(body.tool || "claude");
|
||||
require("../lib/run-as").assertOwnerFits(target, ownerId);
|
||||
};
|
||||
if (body.background === true) {
|
||||
const ctx = await prepareClone(req, res, body, before);
|
||||
if (!ctx) return null;
|
||||
const request = { url: ctx.url, parent: ctx.dir, name: ctx.name };
|
||||
for (const k of ["tool", "title", "partial"]) if (body[k] !== undefined) request[k] = body[k];
|
||||
return startBackground(res, ctx, { kind: "workspace", laneId: null, request });
|
||||
}
|
||||
const done = await cloneInto(req, res, body, before);
|
||||
if (!done) return null;
|
||||
const out = finishWorkspaceClone(req, body, done);
|
||||
res.status(out.status).json(out.json);
|
||||
return null;
|
||||
}
|
||||
|
||||
router.post("/clone", sameOriginGuard, async (req, res) => {
|
||||
await startWorkspaceClone(req, res, req.body || {});
|
||||
});
|
||||
|
||||
/** After an extra-folder clone: `lane_sources`' own folder checks, then add it. */
|
||||
function finishSourceClone(req, lane, target) {
|
||||
const sources = require("../lib/lane-sources");
|
||||
try {
|
||||
const allowed = (p) => scope.folderAllowed(req, lanesLib.canonicalCwd(path.resolve(p)));
|
||||
sources.addFolder(lane.id, target, { allowed });
|
||||
broadcastLane(lane.id);
|
||||
return {
|
||||
status: 201,
|
||||
laneId: lane.id,
|
||||
json: {
|
||||
path: target,
|
||||
instructions: lane.instructions || "",
|
||||
supported: true,
|
||||
maxFileBytes: sources.MAX_FILE_BYTES,
|
||||
maxInstructions: sources.MAX_INSTRUCTIONS,
|
||||
sources: sources.listSources(lane.id),
|
||||
},
|
||||
};
|
||||
} catch (err) {
|
||||
// The clone stays where it is; it can still be added by its path.
|
||||
return {
|
||||
status: err.code === "EFOLDER" ? 403 : 400,
|
||||
json: { error: { code: err.code || "ESOURCE", message: err.message }, path: target },
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/** A workspace's Clone extra folder; the background job when `background: true`. */
|
||||
async function startSourceClone(req, res, lane, body) {
|
||||
const sources = require("../lib/lane-sources");
|
||||
if (!sources.supports(lane.tool)) {
|
||||
res
|
||||
.status(400)
|
||||
.json({ error: { code: "EUNSUPPORTED", message: "this tool takes no extra folders" } });
|
||||
return null;
|
||||
}
|
||||
const cloneBody = {
|
||||
url: body.url,
|
||||
name: body.name,
|
||||
credentials: body.credentials,
|
||||
parent: path.dirname(lane.cwd),
|
||||
...(body.background !== undefined ? { background: body.background } : {}),
|
||||
};
|
||||
if (body.background === true) {
|
||||
const ctx = await prepareClone(req, res, cloneBody, () => {});
|
||||
if (!ctx) return null;
|
||||
const request = { url: ctx.url, parent: ctx.dir, name: ctx.name };
|
||||
return startBackground(res, ctx, { kind: "source", laneId: lane.id, request });
|
||||
}
|
||||
const done = await cloneInto(req, res, cloneBody, () => {});
|
||||
if (!done) return null;
|
||||
const out = finishSourceClone(req, lane, done.target);
|
||||
res.status(out.status).json(out.json);
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Clone a related repo as one of a workspace's extra folders: the same clone
|
||||
* as above (URL allowlist, the folder account's own identity, `ECREDENTIALS`
|
||||
@@ -928,43 +1229,7 @@ router.post("/:id/sources/clone", sameOriginGuard, async (req, res) => {
|
||||
if (!lane) {
|
||||
return res.status(404).json({ error: { code: "ENOLANE", message: "workspace not found" } });
|
||||
}
|
||||
const sources = require("../lib/lane-sources");
|
||||
if (!sources.supports(lane.tool)) {
|
||||
return res
|
||||
.status(400)
|
||||
.json({ error: { code: "EUNSUPPORTED", message: "this tool takes no extra folders" } });
|
||||
}
|
||||
const body = req.body || {};
|
||||
const done = await cloneInto(
|
||||
req,
|
||||
res,
|
||||
{
|
||||
url: body.url,
|
||||
name: body.name,
|
||||
credentials: body.credentials,
|
||||
parent: path.dirname(lane.cwd),
|
||||
},
|
||||
() => {}
|
||||
);
|
||||
if (!done) return;
|
||||
try {
|
||||
const allowed = (p) => scope.folderAllowed(req, lanesLib.canonicalCwd(path.resolve(p)));
|
||||
sources.addFolder(lane.id, done.target, { allowed });
|
||||
broadcastLane(lane.id);
|
||||
return res.status(201).json({
|
||||
path: done.target,
|
||||
instructions: lane.instructions || "",
|
||||
supported: true,
|
||||
maxFileBytes: sources.MAX_FILE_BYTES,
|
||||
maxInstructions: sources.MAX_INSTRUCTIONS,
|
||||
sources: sources.listSources(lane.id),
|
||||
});
|
||||
} catch (err) {
|
||||
// The clone stays where it is; it can still be added by its path.
|
||||
return res
|
||||
.status(err.code === "EFOLDER" ? 403 : 400)
|
||||
.json({ error: { code: err.code || "ESOURCE", message: err.message }, path: done.target });
|
||||
}
|
||||
await startSourceClone(req, res, lane, req.body || {});
|
||||
});
|
||||
|
||||
function isSymlink(p) {
|
||||
|
||||
Reference in New Issue
Block a user