Runner: Build-Auftraege mit Warteschlange

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

runner/src/index.js

@@ -1,9 +1,14 @@
 import { loadConfig } from './config.js';
 import { createApiServer } from './server.js';
+import { createDocker } from './docker.js';
+import { createJobs } from './jobs.js';
 
 // Jobs/Dev/Proxy kommen in späteren Tasks; bis dahin nur Stubs, damit /status antwortet.
 const config = loadConfig(process.env);
-const jobs = { submit: () => { throw Object.assign(new Error('Noch nicht verfügbar'), { status: 503 }); }, get: () => null, queueLength: () => 0 };
+const realJobs = createJobs({ config, docker: createDocker() });
+// Die API antwortet auf POST /jobs/build mit { runId }.
+const jobs = { submit: (job) => ({ runId: realJobs.submit(job) }), get: realJobs.get, queueLength: realJobs.queueLength };
+setInterval(() => realJobs.pruneCache().catch((e) => console.error('pruneCache', e)), 24 * 3600e3).unref();
 const dev = { ensure: async () => { throw Object.assign(new Error('Noch nicht verfügbar'), { status: 503 }); }, stop: async () => {}, list: () => [] };
 
 createApiServer({ config, jobs, dev }).listen(config.apiPort, () => {

runner/src/jobs.js

@@ -0,0 +1,281 @@
+// Build-Aufträge: Einzel-Slot-Warteschlange, Klonen, Erkennung, Installation/Build im Container, atomare Veröffentlichung.
+import crypto from 'node:crypto';
+import { spawn } from 'node:child_process';
+import * as nodeFs from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join, dirname, sep } from 'node:path';
+import { buildRunArgs } from './docker.js';
+import { detectFromDir } from './detect.js';
+import { readMemAvailable } from './server.js';
+
+const LOG_LINES = 300;
+const MAX_LINE = 2000;
+const KEEP_MS = 60 * 60e3;
+const MEM_RETRY_MS = 30e3;
+const DAY = 24 * 3600e3;
+const UID = 1000;
+
+// ---- Dateisystem-Helfer (überschreibbar für Tests) ----
+async function chownR(p) {
+  const st = await nodeFs.lstat(p);
+  await nodeFs.lchown(p, UID, UID);
+  if (st.isDirectory()) for (const e of await nodeFs.readdir(p)) await chownR(join(p, e));
+}
+
+// Kopiert nur Verzeichnisse und reguläre Dateien (keine Symlinks, nie .git), Rechte 755/644 (R7).
+async function copyTree(fs, src, dst, { skipGit = false } = {}) {
+  await fs.mkdir(dst, { recursive: true });
+  await fs.chmod(dst, 0o755);
+  for (const e of await fs.readdir(src, { withFileTypes: true })) {
+    if (skipGit && e.name === '.git') continue;
+    const s = join(src, e.name);
+    const d = join(dst, e.name);
+    if (e.isDirectory()) await copyTree(fs, s, d);
+    else if (e.isFile()) { await fs.copyFile(s, d); await fs.chmod(d, 0o644); }
+  }
+}
+
+async function dirSize(fs, p) {
+  const st = await fs.lstat(p);
+  if (!st.isDirectory()) return st.size;
+  let sum = 0;
+  for (const e of await fs.readdir(p)) sum += await dirSize(fs, join(p, e));
+  return sum;
+}
+
+const defaultFs = { ...nodeFs, chownR, copyTree, dirSize: (p) => dirSize(nodeFs, p) };
+
+// ---- git (R11) ----
+export function createGit(config) {
+  return function git({ repo, branch, dir, onLine }) {
+    const b64 = Buffer.from(`x-access-token:${config.giteaToken}`).toString('base64');
+    const url = `${String(config.giteaUrl).replace(/\/+$/, '')}/${repo}.git`;
+    const args = ['-c', `http.extraHeader=Authorization: Basic ${b64}`, 'clone', '--depth', '1', '--branch', branch, '--', url, dir];
+    const env = { PATH: process.env.PATH || '/usr/bin:/bin', GIT_TERMINAL_PROMPT: '0', GIT_CONFIG_NOSYSTEM: '1', HOME: tmpdir() };
+    return new Promise((resolve, reject) => {
+      const child = spawn('git', args, { env, stdio: ['ignore', 'pipe', 'pipe'] });
+      const feed = (s) => { let buf = ''; s.setEncoding('utf8'); s.on('data', (d) => { const parts = (buf + d).split(/\r?\n|\r/); buf = parts.pop(); parts.forEach((l) => l && onLine?.(l)); }); s.on('end', () => buf && onLine?.(buf)); };
+      feed(child.stdout); feed(child.stderr);
+      child.on('error', reject);
+      child.on('close', (code) => (code === 0 ? resolve() : reject(new Error(`git clone fehlgeschlagen (Exit-Code ${code})`))));
+    });
+  };
+}
+
+const withCorepack = (pm, cmd) => (pm === 'pnpm' || pm === 'yarn' ? ['corepack', ...cmd] : cmd); // R9
+const imageFor = (pm) => (pm === 'bun' ? 'oven/bun:1' : 'node:20');
+const cacheName = (repo) => repo.replace('/', '__');
+
+export function createJobs({ config, docker, fs: fsOver = {}, git, mem = readMemAvailable, now = Date.now, setTimer = (fn, ms) => setTimeout(fn, ms).unref?.() ?? 0 }) {
+  const fs = { ...defaultFs, ...fsOver };
+  const gitFn = git || createGit(config);
+  const runs = new Map();
+  const queue = [];
+  let active = false;
+  let retry = false;
+  const activeCaches = new Set();
+
+  // ---- Warteschlange mit genau einem aktiven Slot und Speicherwächter ----
+  function pump() {
+    if (active || queue.length === 0) return;
+    if (mem() < config.minFreeBytes) {
+      if (!retry) { retry = true; setTimer(() => { retry = false; pump(); }, MEM_RETRY_MS); }
+      return;
+    }
+    active = true;
+    const task = queue.shift();
+    Promise.resolve().then(task).finally(() => { active = false; pump(); });
+  }
+  const enqueue = (fn) => new Promise((resolve, reject) => {
+    queue.push(() => fn().then(resolve, reject));
+    pump();
+  });
+
+  function purge() {
+    const t = now();
+    for (const [id, r] of runs) if (r.finishedAt && t - r.finishedAt > KEEP_MS) runs.delete(id);
+  }
+
+  // ---- Log: Ringpuffer, Token/Variablenwerte maskiert ----
+  function masker(job) {
+    const secrets = [];
+    const tok = config.giteaToken;
+    if (tok) secrets.push(tok, Buffer.from(`x-access-token:${tok}`).toString('base64'), Buffer.from(`${tok}:`).toString('base64'));
+    for (const v of Object.values(job?.env || {})) if (typeof v === 'string' && v.length >= 4) secrets.push(v);
+    secrets.sort((a, b) => b.length - a.length);
+    return (line) => {
+      let s = String(line).slice(0, MAX_LINE);
+      for (const x of secrets) s = s.split(x).join('•••');
+      return s.replace(/(Authorization:\s*)(Basic|Bearer)\s+\S+/gi, '$1$2 •••').replace(/(https?:\/\/)[^/\s@]+@/g, '$1•••@');
+    };
+  }
+  const makeLog = (log, mask) => (line) => { log.push(mask(line)); if (log.length > LOG_LINES) log.splice(0, log.length - LOG_LINES); };
+
+  // ---- Container-Schritt ----
+  async function containerStep({ job, id, step, cmd, pm, workDir, cacheDir, env, onLine, deadline }) {
+    await fs.mkdir(cacheDir, { recursive: true });
+    await fs.chownR(workDir);
+    await fs.chownR(cacheDir); // R7
+    const remaining = Math.max(1000, deadline - now());
+    const args = buildRunArgs({ image: imageFor(pm), name: `preview-build-${id}-${step}`, workDir, cacheDir, workRoot: config.workRoot, cacheRoot: config.cacheRoot, cmd: withCorepack(pm, cmd), memory: '1g' });
+    return docker.run(args, { env, timeoutMs: remaining, onLine: (l) => onLine(l) });
+  }
+  const failText = (what, code, limit) => (code === 124 ? `${what}: Zeitlimit von ${Math.round(limit / 60e3)} Minuten überschritten` : `${what} fehlgeschlagen (Exit-Code ${code})`);
+
+  // Klonen + Erkennung + Installation (R1); Aufrufer hält den Slot.
+  async function prepare(job, { onLine, workDir, id = crypto.randomUUID(), deadline = now() + config.buildTimeoutMs }) {
+    const cacheDir = join(config.cacheRoot, cacheName(job.repo));
+    activeCaches.add(cacheName(job.repo));
+    try {
+      await fs.rm(workDir, { recursive: true, force: true });
+      await fs.mkdir(dirname(workDir), { recursive: true });
+      await gitFn({ repo: job.repo, branch: job.branch, dir: workDir, onLine });
+      const detected = detectFromDir(workDir, { mode: job.mode, index: job.index });
+      if (detected.error || detected.kind === 'static' || !detected.install?.length) return { workDir, detected };
+      const r = await containerStep({ job, id, step: 'install', cmd: detected.install, pm: detected.pm, workDir, cacheDir, onLine, deadline });
+      if (r.code !== 0) return { workDir, detected, error: failText('Installation', r.code, config.buildTimeoutMs) };
+      return { workDir, detected };
+    } finally {
+      activeCaches.delete(cacheName(job.repo));
+    }
+  }
+
+  // ---- Atomare Veröffentlichung (R11) ----
+  async function publish(job, srcDir, { skipGit }) {
+    const target = join(config.previewsRoot, job.sub);
+    const fresh = `${target}.new`;
+    const old = `${target}.old`;
+    await fs.rm(fresh, { recursive: true, force: true });
+    await fs.rm(old, { recursive: true, force: true });
+    try {
+      await fs.mkdir(config.previewsRoot, { recursive: true });
+      await fs.copyTree(fs, srcDir, fresh, { skipGit });
+      let hadOld = true;
+      try { await fs.rename(target, old); } catch (e) { if (e.code !== 'ENOENT') throw e; hadOld = false; }
+      try { await fs.rename(fresh, target); } catch (e) {
+        if (hadOld) await fs.rename(old, target).catch(() => {});
+        throw e;
+      }
+      await fs.rm(old, { recursive: true, force: true });
+    } catch (e) {
+      await fs.rm(fresh, { recursive: true, force: true }).catch(() => {});
+      throw e;
+    }
+  }
+
+  async function findOutput(workDir, dirs) {
+    const realWork = await fs.realpath(workDir);
+    for (const d of dirs) {
+      const p = join(workDir, d);
+      try {
+        const st = await fs.lstat(p);
+        if (!st.isDirectory()) continue;
+        const real = await fs.realpath(p);
+        if (real === realWork || !real.startsWith(realWork + sep)) continue;
+        return p;
+      } catch { /* nicht vorhanden */ }
+    }
+    return null;
+  }
+
+  // ---- Ablauf eines Build-Laufs ----
+  async function execute(run, job) {
+    const onLine = makeLog(run.log, masker(job));
+    const workDir = join(config.workRoot, 'build', run.runId);
+    const deadline = now() + config.buildTimeoutMs;
+    run.state = 'running';
+    try {
+      const prep = await prepare(job, { onLine, workDir, id: run.runId, deadline });
+      const d = prep.detected;
+      run.detected = d;
+      if (d.error) throw new Error(d.error);
+      if (prep.error) throw new Error(prep.error);
+      if (d.kind === 'static') {
+        if (d.hint) onLine(d.hint);
+        await publish(job, workDir, { skipGit: true });
+        run.result = 'static';
+      } else if (d.kind === 'dev') {
+        run.result = 'dev';
+      } else {
+        let problem = null;
+        const cacheDir = join(config.cacheRoot, cacheName(job.repo));
+        const r = await containerStep({ job, id: run.runId, step: 'build', cmd: d.buildCmd, pm: d.pm, workDir, cacheDir, env: job.env, onLine, deadline });
+        if (r.code !== 0) problem = failText('Build', r.code, config.buildTimeoutMs);
+        let out = null;
+        if (!problem) {
+          out = await findOutput(workDir, d.outputDirs);
+          if (!out) problem = `Kein Ausgabeordner gefunden (gesucht: ${d.outputDirs.join(', ')})`;
+        }
+        if (problem) {
+          if (job.mode === 'auto' && d.devCmd) { onLine(`${problem} – Dev-Server wird verwendet`); run.result = 'dev'; }
+          else throw new Error(problem);
+        } else {
+          await publish(job, out, { skipGit: false });
+          run.result = 'build';
+        }
+      }
+      run.state = 'done';
+    } catch (e) {
+      run.state = 'failed';
+      run.result = undefined;
+      run.error = masker(job)(e?.message || 'Unbekannter Fehler');
+      onLine(run.error);
+    } finally {
+      await fs.rm(workDir, { recursive: true, force: true }).catch(() => {});
+      run.finishedAt = now();
+    }
+  }
+
+  function submit(job) {
+    purge();
+    const runId = crypto.randomUUID();
+    const run = { runId, state: 'queued', detected: null, sha: job.sha, log: [], error: null, result: null, finishedAt: 0 };
+    runs.set(runId, run);
+    enqueue(() => execute(run, job));
+    return runId;
+  }
+
+  function get(runId) {
+    purge();
+    const r = runs.get(runId);
+    if (!r) return null;
+    const { finishedAt, ...pub } = r; // eslint-disable-line no-unused-vars
+    return { ...pub, log: [...r.log] };
+  }
+
+  // Für Dev (Task 5): gleicher Slot, Klon + Installation in workDir.
+  function prepareWorkspace(job, { onLine = () => {}, workDir } = {}) {
+    const log = (l) => onLine(masker(job)(l));
+    return enqueue(async () => {
+      const r = await prepare(job, { onLine: log, workDir });
+      if (r.error) throw new Error(r.error);
+      return { workDir: r.workDir, detected: r.detected };
+    });
+  }
+
+  async function pruneCache() {
+    const root = config.cacheRoot;
+    const maxAge = config.cacheMaxAgeMs ?? 14 * DAY;
+    const maxBytes = config.cacheMaxBytes ?? 5 * 1024 ** 3;
+    let names = [];
+    try { names = await fs.readdir(root); } catch { return; }
+    const items = [];
+    for (const n of names) {
+      if (activeCaches.has(n)) continue;
+      try { const p = join(root, n); const st = await fs.lstat(p); items.push({ p, mtime: st.mtimeMs, size: await fs.dirSize(p) }); } catch { /* weg */ }
+    }
+    let kept = [];
+    for (const it of items) {
+      if (now() - it.mtime > maxAge) await fs.rm(it.p, { recursive: true, force: true }); else kept.push(it);
+    }
+    kept.sort((a, b) => a.mtime - b.mtime);
+    let total = kept.reduce((s, i) => s + i.size, 0);
+    for (const it of kept) {
+      if (total <= maxBytes) break;
+      await fs.rm(it.p, { recursive: true, force: true });
+      total -= it.size;
+    }
+  }
+
+  return { submit, get, queueLength: () => queue.length + (active ? 1 : 0), prepareWorkspace, pruneCache };
+}