AgentHub/server/lib/clone-queue.js

/**
 * @file The machine-wide limit on `git clone`s running at once, and each
 * clone's progress. A clone's index-pack resolves deltas on every core it
 * finds, so a few big (~1 GB) clones started together pinned all of them and
 * stalled the dashboard and every pane beside it. Clones past the limit wait
 * here, in arrival order, and each one gets its share of the cores
 * (`threads`) — the clone's own timeout starts only once it runs.
 * `AGENTHUB_CLONE_CONCURRENCY` overrides the default (half the cores, at
 * 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>
 */

"use strict";

const os = require("node:os");

/** A finished clone's progress stays readable this long, then is dropped. */
const KEEP_DONE_MS = 60_000;
const TOKEN_RE = /^[A-Za-z0-9_-]{16,64}$/;

function cores() {
  return Math.max(1, (os.availableParallelism?.() ?? os.cpus().length) || 1);
}

/** How many clones may run at once. */
function concurrency() {
  const set = Number.parseInt(process.env.AGENTHUB_CLONE_CONCURRENCY, 10);
  if (Number.isInteger(set) && set > 0) return set;
  return Math.max(1, Math.floor(cores() / 2));
}

/** index-pack threads for one clone: the cores split between the running ones. */
function threads() {
  return Math.max(1, Math.floor(cores() / concurrency()));
}

let running = 0;
/** `{start, job}` in arrival order; `job` is null for an untracked clone. */
const waiting = [];
/** token → job: `{owner, state, phase, label, percent, detail, finishedAt}`. */
const jobs = new Map();

function next() {
  while (running < concurrency() && waiting.length) {
    running++;
    const { start, job } = waiting.shift();
    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, reject) => {
    waiting.push({ start, reject, job });
    next();
  });
  try {
    return await fn();
  } finally {
    running--;
    next();
  }
}

/** `{running, waiting, limit}` — for a status read or a test. */
function cloneQueueState() {
  return { running, waiting: waiting.length, limit: concurrency() };
}

function sweep(now = Date.now()) {
  for (const [token, job] of jobs) {
    if (job.finishedAt && now - job.finishedAt > KEEP_DONE_MS) jobs.delete(token);
  }
}

/**
 * A progress record for the clone the browser named `token`, owned by
 * `owner` (`scope.ownerFor`). Throws `EBADFIELD` for a malformed token or one
 * still in use.
 */
function track(token, owner) {
  sweep();
  if (typeof token !== "string" || !TOKEN_RE.test(token) || jobs.has(token)) {
    throw Object.assign(new Error("progressToken: 16–64 of [A-Za-z0-9_-], unused"), {
      code: "EBADFIELD",
    });
  }
  const job = {
    owner,
    state: "queued",
    phase: null,
    label: null,
    percent: null,
    detail: null,
    finishedAt: null,
  };
  jobs.set(token, job);
  return job;
}

/** Marks a tracked clone over: `done` or `failed`. */
function finish(job, ok) {
  if (!job) return;
  job.state = ok ? "done" : "failed";
  job.finishedAt = Date.now();
}

/** git's progress words → a phase the UI translates; anything else is `other`. */
const PHASES = [
  [/^remote: /, "remote"],
  [/^Receiving objects/, "receiving"],
  [/^Resolving deltas/, "resolving"],
  [/^Updating files/, "checkout"],
];
const LINE_RE = /^((?:remote: )?[A-Za-z][A-Za-z ]*?):\s+(\d{1,3})% \(([^)]*)\)(.*)$/;

/**
 * Feeds a chunk of `git clone --progress` stderr into `job`: the LAST
 * complete progress line in it wins (git redraws a line with `\r`).
 */
function report(job, chunk) {
  if (!job) return;
  const lines = String(chunk).split(/[\r\n]+/);
  for (let i = lines.length - 1; i >= 0; i--) {
    const m = LINE_RE.exec(lines[i].trim());
    if (!m) continue;
    job.label = m[1].replace(/^remote: /, "");
    job.phase = (PHASES.find(([re]) => re.test(m[1])) || [null, "other"])[1];
    job.percent = Math.min(100, Number(m[2]));
    // "1234/5678), 120.50 MiB | 5.20 MiB/s, done." → counts + size + speed
    job.detail =
      `${m[3]}${m[4]}`
        .replace(/,\s*done\.?$/, "")
        .replace(/^,\s*/, "")
        .trim() || null;
    job.onChange?.("progress");
    return;
  }
}

/**
 * What `GET /api/lanes/clone/progress/:token` answers, or null when there is
 * no such clone or it is not `owner`'s. A queued clone carries its 1-based
 * `position` among the waiting ones.
 */
function progress(token, owner) {
  sweep();
  const job = jobs.get(token);
  if (!job || job.owner !== owner) return null;
  return {
    state: job.state,
    position: position(job),
    ...cloneQueueState(),
    phase: job.phase,
    label: job.label,
    percent: job.percent,
    detail: job.detail,
  };
}

/**
 * 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 || "")
    .split(/[\r\n]+/)
    .filter((l) => l.trim() && !LINE_RE.test(l.trim()))
    .join("\n");
}

module.exports = {
  withCloneSlot,
  cloneQueueState,
  concurrency,
  threads,
  track,
  finish,
  report,
  progress,
  position,
  dequeue,
  occupy,
  stripProgress,
};