Runner: Dev-Server bei Bedarf

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

runner/src/dev.js

@@ -0,0 +1,149 @@
+// Dev-Server bei Bedarf: Container mit `npm run dev` starten, bei Leerlauf stoppen, max. devMax gleichzeitig (LRU-Verdrängung).
+import net from 'node:net';
+import { join } from 'node:path';
+import * as nodeFs from 'node:fs/promises';
+import { devRunArgs } from './docker.js';
+import { readMemAvailable } from './server.js';
+
+const LOG_LINES = 300;
+const MAX_LINE = 2000;
+const POLL_MS = 1000;
+const UID = 1000;
+const PREFIX = 'preview-dev-';
+const cacheName = (repo) => repo.replace('/', '__');
+
+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));
+}
+
+// R7: Container laufen als 1000:1000, also gehören Arbeitsordner und Cache dieser Nutzer-ID.
+const defaultPrepareDirs = async (workDir, cacheDir) => {
+  await nodeFs.mkdir(cacheDir, { recursive: true });
+  await chownR(workDir);
+  await chownR(cacheDir);
+};
+
+// TCP-Verbindung zum Dev-Server (R5: Docker-DNS-Name im Netz preview-sandbox).
+export function tcpConnect(host, port, timeoutMs = 2000) {
+  return new Promise((resolve, reject) => {
+    const s = net.connect({ host, port });
+    s.setTimeout(timeoutMs);
+    s.on('connect', () => { s.destroy(); resolve(true); });
+    s.on('timeout', () => { s.destroy(); reject(new Error('Zeitüberschreitung')); });
+    s.on('error', (e) => { s.destroy(); reject(e); });
+  });
+}
+
+export function createDev({
+  config, docker, jobs, mem = readMemAvailable, now = Date.now, connect = tcpConnect,
+  setTimer = (fn, ms) => setTimeout(fn, ms).unref?.() ?? 0, prepareDirs = defaultPrepareDirs,
+}) {
+  const entries = new Map(); // sub ⇒ { sub, state, sha, lastUsed, log, port, job, release, promise, cancelled }
+  const sleep = (ms) => new Promise((r) => setTimer(r, ms));
+  const isRunning = (e) => e.state === 'online' || e.state === 'starting';
+
+  function masker(job) {
+    const secrets = [];
+    if (config.giteaToken) secrets.push(config.giteaToken);
+    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; };
+  }
+
+  // Container entfernen, Cache freigeben; Zielzustand 'sleeping' (Leerlauf/Verdrängung) oder Eintrag löschen.
+  async function shutdown(e, finalState) {
+    e.cancelled = true;
+    e.release?.(); e.release = null;
+    if (finalState === 'off') entries.delete(e.sub); else { e.state = finalState; e.port = null; }
+    try { await docker.rm(PREFIX + e.sub); } catch (err) { console.error('dev rm', e.sub, err); }
+  }
+
+  async function start(e, job) {
+    const name = PREFIX + e.sub;
+    const mask = masker(job);
+    const log = (l) => { e.log.push(mask(l)); if (e.log.length > LOG_LINES) e.log.splice(0, e.log.length - LOG_LINES); };
+    const fail = async (msg) => {
+      if (e.cancelled) return;
+      log(msg);
+      e.release?.(); e.release = null;
+      e.state = 'failed'; e.port = null;
+      try { await docker.rm(name); } catch { /* egal */ }
+    };
+    try {
+      e.release = jobs.retainCache(job.repo);
+      const workDir = join(config.workRoot, 'dev', e.sub);
+      const cacheDir = join(config.cacheRoot, cacheName(job.repo));
+      const prep = await jobs.prepareWorkspace(job, { onLine: log, workDir });
+      if (e.cancelled) return;
+      const d = prep.detected;
+      if (!d?.devCmd) return await fail('Kein dev-Skript in package.json');
+      await prepareDirs(prep.workDir ?? workDir, cacheDir);
+      if (e.cancelled) return;
+      const cmd = d.pm === 'pnpm' || d.pm === 'yarn' ? ['corepack', ...d.devCmd] : d.devCmd; // R9
+      const args = devRunArgs({
+        image: d.pm === 'bun' ? 'oven/bun:1' : 'node:20', sub: e.sub, workDir, cacheDir,
+        workRoot: config.workRoot, cacheRoot: config.cacheRoot, cmd, memory: '512m',
+      });
+      // Feste Variable zuletzt, damit Projektvariablen sie nicht überschreiben; Werte laufen per --env-file (R10).
+      const env = { ...job.env, __VITE_ADDITIONAL_SERVER_ALLOWED_HOSTS: `.${config.baseHost}` };
+      const r = await docker.run(args, { env, onLine: (l) => log(l) });
+      if (e.cancelled) { await docker.rm(name).catch(() => {}); return; }
+      if (r.code !== 0) return await fail(`Container konnte nicht gestartet werden (Exit-Code ${r.code})`);
+      e.port = d.devPort;
+      const deadline = now() + config.devReadyMs;
+      while (!e.cancelled) {
+        let ok = false;
+        try { ok = await connect(name, d.devPort); } catch { /* noch nicht bereit */ }
+        if (e.cancelled) return;
+        if (ok) { e.state = 'online'; e.lastUsed = now(); return; }
+        if (now() >= deadline) break;
+        await sleep(POLL_MS);
+      }
+      if (e.cancelled) return;
+      await docker.run(['logs', '--tail', '100', name], { onLine: (l) => log(l) }).catch(() => {});
+      await fail(`Dev-Server nach ${Math.round(config.devReadyMs / 1000)} s nicht erreichbar`);
+    } catch (err) {
+      await fail(err?.message || 'Unbekannter Fehler');
+    }
+  }
+
+  async function ensure(sub, job) {
+    const cur = entries.get(sub);
+    if (cur && cur.sha === job.sha) {
+      if (cur.state === 'online') { cur.lastUsed = now(); return { state: 'online', target: `http://${PREFIX}${sub}:${cur.port}` }; }
+      if (cur.state === 'starting') { cur.lastUsed = now(); return { state: 'starting' }; }
+    }
+    let free = 0;
+    try { free = mem(); } catch { /* Fehler = zu wenig Speicher */ }
+    if (!(free >= config.minFreeBytes)) throw Object.assign(new Error('Server ausgelastet'), { status: 503 });
+    const stops = [];
+    if (cur) stops.push(shutdown(cur, 'sleeping')); // neuer sha oder failed/sleeping: frisch starten
+    const others = [...entries.values()].filter((e) => e !== cur && isRunning(e)).sort((a, b) => a.lastUsed - b.lastUsed);
+    while (others.length >= config.devMax) stops.push(shutdown(others.shift(), 'sleeping')); // LRU
+    const e = { sub, state: 'starting', sha: job.sha, lastUsed: now(), log: [], port: null, release: null, promise: null, cancelled: false };
+    entries.set(sub, e);
+    e.promise = Promise.all(stops).then(() => start(e, job));
+    return { state: 'starting' };
+  }
+
+  const touch = (sub) => { const e = entries.get(sub); if (e) e.lastUsed = now(); };
+  async function stop(sub) { const e = entries.get(sub); if (e) await shutdown(e, 'off'); }
+  const list = () => [...entries.values()].map((e) => ({ sub: e.sub, state: e.state, lastUsed: e.lastUsed, sha: e.sha }));
+  function status(sub) {
+    const e = entries.get(sub);
+    return e ? { state: e.state, sha: e.sha, lastUsed: e.lastUsed, log: [...e.log] } : { state: 'off', log: [] };
+  }
+  async function sweep() {
+    const t = now();
+    await Promise.all([...entries.values()].filter((e) => e.state === 'online' && t - e.lastUsed > config.devIdleMs).map((e) => shutdown(e, 'sleeping')));
+  }
+  // Zustand liegt nur im Speicher: nach Runner-Neustart übrig gebliebene Dev-Container entfernen.
+  async function cleanupOrphans() {
+    for (const n of await docker.ps('webklar.preview')) if (n.startsWith(PREFIX)) await docker.rm(n);
+  }
+  const idle = () => Promise.all([...entries.values()].map((e) => e.promise)).then(() => {});
+
+  return { ensure, touch, stop, list, status, sweep, cleanupOrphans, idle };
+}

runner/src/index.js

@@ -2,14 +2,19 @@ import { loadConfig } from './config.js';
 import { createApiServer } from './server.js';
 import { createDocker } from './docker.js';
 import { createJobs } from './jobs.js';
+import { createDev } from './dev.js';
 
-// Jobs/Dev/Proxy kommen in späteren Tasks; bis dahin nur Stubs, damit /status antwortet.
+// Proxy kommt in einem späteren Task.
 const config = loadConfig(process.env);
-const realJobs = createJobs({ config, docker: createDocker() });
+const docker = createDocker();
+const realJobs = createJobs({ config, docker });
 // 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: () => [] };
+const dev = createDev({ config, docker, jobs: realJobs });
+// Zustand der Dev-Server liegt nur im Speicher: Überbleibsel entfernen, dann jede Minute Leerlauf prüfen.
+dev.cleanupOrphans().catch((e) => console.error('cleanupOrphans', e));
+setInterval(() => dev.sweep().catch((e) => console.error('sweep', e)), 60e3).unref();
 
 createApiServer({ config, jobs, dev }).listen(config.apiPort, () => {
   console.log(`Preview-Runner API auf Port ${config.apiPort}`);

runner/src/jobs.js

@@ -78,7 +78,7 @@ export function createJobs({ config, docker, fs: fsOver = {}, git, mem = readMem
   const queue = [];
   let active = false;
   let retry = false;
-  const activeCaches = new Set();
+  const activeCaches = new Map(); // Cache-Name ⇒ Anzahl Halter
 
   // ---- Warteschlange mit genau einem aktiven Slot und Speicherwächter ----
   function pump() {
@@ -144,11 +144,21 @@ export function createJobs({ config, docker, fs: fsOver = {}, git, mem = readMem
     return { workDir, detected };
   }
 
-  // Cache für die gesamte Laufzeit vor pruneCache schützen.
-  async function holdCache(repo, fn) {
+  // Cache vor pruneCache schützen; release() gibt ihn frei (idempotent). Zählend, mehrere Halter je Repo möglich.
+  function retainCache(repo) {
     const n = cacheName(repo);
-    activeCaches.add(n);
-    try { return await fn(); } finally { activeCaches.delete(n); }
+    activeCaches.set(n, (activeCaches.get(n) || 0) + 1);
+    let done = false;
+    return () => {
+      if (done) return;
+      done = true;
+      const c = (activeCaches.get(n) || 1) - 1;
+      if (c <= 0) activeCaches.delete(n); else activeCaches.set(n, c);
+    };
+  }
+  async function holdCache(repo, fn) {
+    const release = retainCache(repo);
+    try { return await fn(); } finally { release(); }
   }
 
   // ---- Atomare Veröffentlichung (R11) ----
@@ -294,5 +304,5 @@ export function createJobs({ config, docker, fs: fsOver = {}, git, mem = readMem
     }
   }
 
-  return { submit, get, queueLength: () => queue.length + (active ? 1 : 0), prepareWorkspace, pruneCache };
+  return { submit, get, queueLength: () => queue.length + (active ? 1 : 0), prepareWorkspace, retainCache, pruneCache };
 }

runner/test/dev.test.js

@@ -0,0 +1,184 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { createDev } from '../src/dev.js';
+
+const MIN = 60e3;
+const GB = 1024 ** 3;
+
+function setup({ mem = 10 * GB, connect, detected, prep } = {}) {
+  const config = { workRoot: '/w/work', cacheRoot: '/w/cache', baseHost: 'project.webklar.com', minFreeBytes: 1.5 * GB, devIdleMs: 20 * MIN, devMax: 3, devReadyMs: 180e3, giteaToken: 'tok_GEHEIM' };
+  const clock = { t: 1_000_000 };
+  const runs = []; const removed = []; const held = new Map(); const prepared = [];
+  let psResult = [];
+  const docker = {
+    run: async (args, opts) => { if (args[0] === 'run') runs.push({ args, opts }); else opts?.onLine?.('container-log-zeile', 'out'); return { code: 0 }; },
+    rm: async (n) => { removed.push(n); return { code: 0, lines: [] }; },
+    ps: async () => psResult,
+  };
+  const jobs = {
+    prepareWorkspace: async (job, o) => { prepared.push({ job, o }); if (prep) return prep(job, o); o.onLine?.('npm install ok'); return { workDir: o.workDir, detected: detected ?? { pm: 'npm', devCmd: ['npm', 'run', 'dev', '--', '--host', '0.0.0.0', '--port', '5173'], devPort: 5173 } }; },
+    retainCache: (repo) => { held.set(repo, (held.get(repo) || 0) + 1); let d = false; return () => { if (!d) { d = true; held.set(repo, held.get(repo) - 1); } }; },
+  };
+  const connects = [];
+  const dev = createDev({
+    config, docker, jobs, mem: () => (typeof mem === 'function' ? mem() : mem), now: () => clock.t,
+    connect: connect ?? (async (h, p) => { connects.push([h, p]); return true; }),
+    setTimer: (fn, ms) => { clock.t += ms; queueMicrotask(fn); return 0; },
+    prepareDirs: async () => {},
+  });
+  return { dev, clock, runs, removed, held, prepared, connects, config, setPs: (v) => { psResult = v; } };
+}
+const job = (sub, o = {}) => ({ sub, repo: `own/${sub}`, sha: 'abcdef1', branch: 'main', mode: 'dev', env: { API_KEY: 'geheim-wert' }, ...o });
+
+test('start: starting sofort, dann online mit Docker-DNS-Ziel; Container mit Fixed-Env per env-Datei', async () => {
+  const s = setup();
+  assert.deepEqual(await s.dev.ensure('a', job('a')), { state: 'starting' });
+  await s.dev.idle();
+  assert.deepEqual(s.connects[0], ['preview-dev-a', 5173]);
+  assert.equal(s.runs.length, 1);
+  const { args, opts } = s.runs[0];
+  assert.ok(args.includes('preview-dev-a') && args.includes('512m') && args.includes('-d'));
+  assert.ok(args.includes('/w/work/dev/a:/work'));
+  assert.equal(opts.env.__VITE_ADDITIONAL_SERVER_ALLOWED_HOSTS, '.project.webklar.com');
+  assert.equal(opts.env.API_KEY, 'geheim-wert');
+  assert.ok(!args.join(' ').includes('geheim-wert'));
+  assert.deepEqual(await s.dev.ensure('a', job('a')), { state: 'online', target: 'http://preview-dev-a:5173' });
+  assert.equal(s.runs.length, 1);
+});
+
+test('gleicher sub zweimal während Start ⇒ ein Container', async () => {
+  const s = setup();
+  const [r1, r2] = await Promise.all([s.dev.ensure('a', job('a')), s.dev.ensure('a', job('a'))]);
+  assert.equal(r1.state, 'starting'); assert.equal(r2.state, 'starting');
+  await s.dev.idle();
+  assert.equal(s.runs.length, 1); assert.equal(s.prepared.length, 1);
+});
+
+test('pnpm läuft über corepack, bun nutzt bun-Image', async () => {
+  const s = setup({ detected: { pm: 'pnpm', devCmd: ['pnpm', 'run', 'dev'], devPort: 4000 } });
+  await s.dev.ensure('a', job('a')); await s.dev.idle();
+  const a = s.runs[0].args;
+  assert.deepEqual(a.slice(-4), ['corepack', 'pnpm', 'run', 'dev']);
+  assert.deepEqual(s.connects[0], ['preview-dev-a', 4000]);
+  const b = setup({ detected: { pm: 'bun', devCmd: ['bun', 'run', 'dev'], devPort: 5173 } });
+  await b.dev.ensure('a', job('a')); await b.dev.idle();
+  assert.ok(b.runs[0].args.includes('oven/bun:1'));
+});
+
+test('4. Start verdrängt den am längsten ungenutzten', async () => {
+  const s = setup();
+  for (const n of ['a', 'b', 'c']) { await s.dev.ensure(n, job(n)); await s.dev.idle(); s.clock.t += MIN; }
+  s.dev.touch('a'); // b ist jetzt der Älteste
+  s.clock.t += MIN;
+  await s.dev.ensure('d', job('d')); await s.dev.idle();
+  assert.ok(s.removed.includes('preview-dev-b'));
+  assert.ok(!s.removed.includes('preview-dev-a'));
+  assert.equal(s.held.get('own/b'), 0);
+  const states = Object.fromEntries(s.dev.list().map((e) => [e.sub, e.state]));
+  assert.equal(states.d, 'online'); assert.equal(states.b, 'sleeping'); assert.equal(states.a, 'online');
+  assert.equal(s.dev.list().filter((e) => e.state === 'online').length, 3);
+});
+
+test('sweep: nach 21 min gestoppt, nach 19 min nicht', async () => {
+  const s = setup();
+  await s.dev.ensure('a', job('a')); await s.dev.idle();
+  s.clock.t += 19 * MIN; await s.dev.sweep();
+  assert.equal(s.dev.status('a').state, 'online'); assert.equal(s.removed.length, 0);
+  s.clock.t += 2 * MIN; await s.dev.sweep();
+  assert.equal(s.dev.status('a').state, 'sleeping'); assert.deepEqual(s.removed, ['preview-dev-a']);
+  assert.equal(s.held.get('own/a'), 0);
+});
+
+test('touch verlängert die Leerlaufzeit', async () => {
+  const s = setup();
+  await s.dev.ensure('a', job('a')); await s.dev.idle();
+  s.clock.t += 19 * MIN; s.dev.touch('a'); s.clock.t += 19 * MIN; await s.dev.sweep();
+  assert.equal(s.dev.status('a').state, 'online');
+});
+
+test('wenig Speicher ⇒ Fehler Server ausgelastet (503), nichts gestartet', async () => {
+  const s = setup({ mem: 1 * GB });
+  await assert.rejects(() => s.dev.ensure('a', job('a')), (e) => e.message === 'Server ausgelastet' && e.status === 503);
+  assert.equal(s.prepared.length, 0); assert.equal(s.runs.length, 0);
+});
+
+test('Speicherwächter gilt nicht für bereits laufenden sub', async () => {
+  let m = 10 * GB;
+  const s = setup({ mem: () => m });
+  await s.dev.ensure('a', job('a')); await s.dev.idle();
+  m = 1 * GB;
+  assert.equal((await s.dev.ensure('a', job('a'))).state, 'online');
+});
+
+test('neuer sha ⇒ Neustart mit neuem Code', async () => {
+  const s = setup();
+  await s.dev.ensure('a', job('a')); await s.dev.idle();
+  assert.deepEqual(await s.dev.ensure('a', job('a', { sha: '1234567' })), { state: 'starting' });
+  await s.dev.idle();
+  assert.equal(s.runs.length, 2); assert.equal(s.prepared.length, 2);
+  assert.ok(s.removed.includes('preview-dev-a'));
+  assert.equal(s.dev.status('a').sha, '1234567'); assert.equal(s.dev.status('a').state, 'online');
+  assert.equal(s.held.get('own/a'), 1);
+});
+
+test('Bereitschafts-Zeitlimit ⇒ failed, Container entfernt, Log vorhanden, Cache frei', async () => {
+  const s = setup({ connect: async () => { throw new Error('ECONNREFUSED'); } });
+  await s.dev.ensure('a', job('a')); await s.dev.idle();
+  const st = s.dev.status('a');
+  assert.equal(st.state, 'failed');
+  assert.ok(s.removed.includes('preview-dev-a'));
+  assert.ok(st.log.some((l) => l.includes('180')));
+  assert.ok(st.log.includes('container-log-zeile'));
+  assert.equal(s.held.get('own/a'), 0);
+});
+
+test('ohne dev-Skript ⇒ failed „Kein dev-Skript“', async () => {
+  const s = setup({ detected: { pm: 'npm', devCmd: null, devPort: 5173 } });
+  await s.dev.ensure('a', job('a')); await s.dev.idle();
+  assert.equal(s.dev.status('a').state, 'failed');
+  assert.ok(s.dev.status('a').log.some((l) => l.includes('Kein dev-Skript')));
+  assert.equal(s.runs.length, 0); assert.equal(s.held.get('own/a'), 0);
+});
+
+test('Vorbereitungsfehler ⇒ failed, Cache frei', async () => {
+  const s = setup({ prep: async () => { throw new Error('Installation fehlgeschlagen (Exit-Code 1)'); } });
+  await s.dev.ensure('a', job('a')); await s.dev.idle();
+  assert.equal(s.dev.status('a').state, 'failed');
+  assert.equal(s.held.get('own/a'), 0);
+});
+
+test('Cache wird gehalten solange der Container läuft, stop gibt frei', async () => {
+  const s = setup();
+  await s.dev.ensure('a', job('a')); await s.dev.idle();
+  assert.equal(s.held.get('own/a'), 1);
+  await s.dev.stop('a');
+  assert.equal(s.held.get('own/a'), 0);
+  assert.deepEqual(s.removed.filter((n) => n === 'preview-dev-a').length >= 1, true);
+  assert.equal(s.dev.status('a').state, 'off');
+  assert.deepEqual(s.dev.list(), []);
+});
+
+test('stop während des Starts: Container entfernt, nie online', async () => {
+  const s = setup();
+  await s.dev.ensure('a', job('a'));
+  await s.dev.stop('a'); await s.dev.idle();
+  assert.equal(s.dev.status('a').state, 'off');
+  assert.equal(s.held.get('own/a'), 0);
+  assert.ok(s.removed.includes('preview-dev-a'));
+});
+
+test('Protokoll maskiert Variablenwerte und ist auf 300 Zeilen begrenzt', async () => {
+  const s = setup({ prep: async (j, o) => { for (let i = 0; i < 400; i++) o.onLine(`zeile ${i} geheim-wert`); return { workDir: o.workDir, detected: { pm: 'npm', devCmd: ['npm', 'run', 'dev'], devPort: 5173 } }; } });
+  await s.dev.ensure('a', job('a')); await s.dev.idle();
+  const log = s.dev.status('a').log;
+  assert.equal(log.length, 300);
+  assert.ok(log.some((l) => l.includes('zeile 399 •••')));
+  assert.ok(!log.some((l) => l.includes('geheim-wert')));
+});
+
+test('Start: verwaiste preview-dev-*-Container werden entfernt', async () => {
+  const s = setup();
+  s.setPs(['preview-dev-x', 'preview-dev-y', 'preview-build-zzz', 'anderer']);
+  await s.dev.cleanupOrphans();
+  assert.deepEqual(s.removed, ['preview-dev-x', 'preview-dev-y']);
+});