S3: 9:16-Render mit Warteschlange, nachrangig zum Stream

Aus der Rohdatei entsteht das fertige Hochformat. Der Filtergraph liegt jetzt
in packages/media und wird von calibrate und vom Render-Schritt gemeinsam
benutzt - die Zahlen darin sind eingemessen, sie gehoeren an genau eine Stelle.

Eine Aufgabe nach der anderen, und waehrend gesendet wird standardmaessig
gar nicht (renderWhileLive: false). ffmpeg laeuft unter nice, mit Zeitlimit,
und die letzten Zeilen von stderr landen in der Fehlermeldung statt im Nichts.
Aus dem 90-Sekunden-Puffer werden die letzten 45 Sekunden genommen (-sseof vor
-i, sonst wird die ganze Datei dekodiert und danach verworfen).

Dabei einen eigenen Fehler aus S2 korrigiert: die Rohdatei wurde in den
Windows-Bind-Mount kopiert und dort verarbeitet - genau das, was ADR-0004
Punkt 2 verbietet. Jetzt liegt sie im Arbeits-Volume, auf die Windows-Seite
kommt nur das fertige, kleine Ergebnis.

Verifiziert an einer kuenstlichen Aufnahme mit drei getrennten Tonspuren:
1920x1080/70 s hinein, 1080x1920/45 s heraus, Naht an der richtigen Stelle,
27,9 s Rechenzeit unter nice 15. 39 Tests gruen.

Neuer Befehl `render <datei>` zum Nachrendern nach Layout-Aenderungen.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013VTGjd92LsDDWF3frd9FTB

apps/ingest/src/clips.ts

@@ -29,6 +29,7 @@ export type ClipperConfig = {
   replayDirHost?: string;
   replayDirContainer: string;
   clipsDir: string;
+  workDir: string;
 };
 
 export type ClipperDeps = {
@@ -38,6 +39,8 @@ export type ClipperDeps = {
   say: (text: string) => boolean;
   currentStreamId: () => string | null;
   broadcasterId: () => string | null;
+  /** Wird gerufen, sobald die Rohdatei im Arbeits-Volume liegt. */
+  onRawReady: (job: { clipId: string; rawPath: string; outDir: string }) => void;
 };
 
 export function createClipper(deps: ClipperDeps) {
@@ -163,17 +166,22 @@ export function createClipper(deps: ClipperDeps) {
 
     try {
       const size = await waitForStableFile(containerPath);
-      // Vom nur-lesbaren Bind-Mount in den Ablageordner. Ab hier gehoert die
-      // Datei uns, OBS darf seinen Puffer aufraeumen.
-      const dir = path.join(cfg.clipsDir, new Date().toISOString().slice(0, 10), clipId);
-      await fs.mkdir(dir, { recursive: true });
-      const target = path.join(dir, `raw${path.extname(containerPath) || '.mkv'}`);
-      await fs.copyFile(containerPath, target);
 
-      repo.updateClip(clipId, { raw_path: target, status: 'raw' });
-      repo.event('obs.replay_saved', { path: target, bytes: size }, deps.currentStreamId());
-      log.info(`Rohdatei gesichert: ${target} (${(size / 1024 / 1024).toFixed(1)} MB)`);
+      // Die Rohdatei landet im Arbeits-Volume, NICHT im Bind-Mount: sie ist
+      // mehrere hundert MB gross, und ueber die Windows-Bruecke zu lesen und
+      // zu schreiben kostet ein Vielfaches. Auf die Windows-Seite kommt
+      // spaeter nur das fertige, kleine Ergebnis. Siehe ADR-0004, Punkt 2.
+      const workDir = path.join(cfg.workDir, clipId);
+      await fs.mkdir(workDir, { recursive: true });
+      const raw = path.join(workDir, `raw${path.extname(containerPath) || '.mkv'}`);
+      await fs.copyFile(containerPath, raw);
+
+      const outDir = path.join(cfg.clipsDir, new Date().toISOString().slice(0, 10), clipId);
+      repo.updateClip(clipId, { raw_path: raw, status: 'raw' });
+      repo.event('obs.replay_saved', { path: raw, bytes: size }, deps.currentStreamId());
+      log.info(`Rohdatei gesichert: ${(size / 1024 / 1024).toFixed(1)} MB`);
       deps.say('Clip gesichert.');
+      deps.onRawReady({ clipId, rawPath: raw, outDir });
     } catch (err) {
       repo.updateClip(clipId, { status: 'failed' });
       log.error(`Rohdatei nicht uebernommen: ${(err as Error).message}`);

apps/ingest/src/index.ts

@@ -5,7 +5,7 @@
 // EventSub kommt spaeter - fuer den Clip-Weg reicht IRC plus Polling.
 // Siehe docs/projects/00-ingest.md.
 
-import { loadConfig, paths } from '#core/config.ts';
+import { loadConfig, loadLayout, paths } from '#core/config.ts';
 import { logger } from '#core/logger.ts';
 import { sleep } from '#core/retry.ts';
 import { openDb, closeDb } from '#db/index.ts';
@@ -15,6 +15,7 @@ import { helix } from '#twitch/helix.ts';
 import { createIrc } from '#twitch/irc.ts';
 import { readSecrets } from '#twitch/secrets.ts';
 import { createClipper } from './clips.ts';
+import { createRenderer } from './render.ts';
 import { createSpikeDetector } from './spike.ts';
 import { startServer } from './server.ts';
 
@@ -66,6 +67,26 @@ export async function runIngest(): Promise<void> {
     }
   );
 
+  const layout = loadLayout();
+  const profile = layout.profiles['default'];
+  if (!profile) throw new Error('config/layout.yaml hat kein Profil "default"');
+
+  const renderer = createRenderer({
+    cfg: {
+      profile,
+      tailSeconds: cfg.clipper.tailSeconds,
+      outputTrack: cfg.clipper.audio.outputTrack,
+      encoder: cfg.clipper.ffmpeg.encoder,
+      preset: cfg.clipper.ffmpeg.preset,
+      crf: cfg.clipper.ffmpeg.crf,
+      niceness: cfg.clipper.ffmpeg.niceness,
+      renderWhileLive: cfg.clipper.renderWhileLive,
+    },
+    repo,
+    workDir: paths.work,
+    isLive: () => state.streamId !== null,
+  });
+
   const clipper = createClipper({
     cfg: {
       enabled: cfg.clipper.enabled,
@@ -76,12 +97,14 @@ export async function runIngest(): Promise<void> {
       ...(process.env.OBS_REPLAY_DIR ? { replayDirHost: process.env.OBS_REPLAY_DIR } : {}),
       replayDirContainer: paths.replay,
       clipsDir: paths.clips,
+      workDir: paths.work,
     },
     repo,
     obs,
     say: (text) => irc.say(text),
     currentStreamId: () => state.streamId,
     broadcasterId: () => state.broadcasterId,
+    onRawReady: (job) => renderer.enqueue(job),
   });
 
   startServer({
@@ -92,6 +115,8 @@ export async function runIngest(): Promise<void> {
       db: repo.stats(),
       spikeArmed: spikes.isArmed(),
       clipsWaitingForFile: clipper.pendingCount(),
+      renderQueue: renderer.pending(),
+      rendering: renderer.currentJob(),
       note: state.chatAnonymous
         ? 'Chat wird anonym mitgelesen. Zum Senden und fuer Helix: Einrichtung unter /setup.'
         : undefined,

apps/ingest/src/render.ts

@@ -0,0 +1,145 @@
+// Render-Warteschlange: aus der Rohdatei wird ein 9:16-Video.
+//
+// Eine Aufgabe nach der anderen. Waehrend gesendet wird, ist Encodieren
+// standardmaessig AUS (`renderWhileLive: false`) - die Kandidatenliste wird
+// erst nach dem Stream gebraucht, und der Stream braucht die CPU vorher.
+// Siehe docs/07-mvp-plan.md, Risikotabelle.
+
+import fs from 'node:fs/promises';
+import path from 'node:path';
+import { logger } from '#core/logger.ts';
+import type { LayoutProfile } from '#core/config.ts';
+import type { Repos } from '#db/repos.ts';
+import { FfmpegError, probe, run } from '#media/ffmpeg.ts';
+import { thumbnailArgs, verticalArgs } from '#media/vertical.ts';
+
+const log = logger('render');
+
+export type RenderConfig = {
+  profile: LayoutProfile;
+  tailSeconds: number;
+  outputTrack: number;
+  encoder: string;
+  preset: string;
+  crf: number;
+  niceness: number;
+  renderWhileLive: boolean;
+};
+
+export type RenderDeps = {
+  cfg: RenderConfig;
+  repo: Repos;
+  workDir: string;
+  isLive: () => boolean;
+};
+
+type Job = { clipId: string; rawPath: string; outDir: string };
+
+export function createRenderer(deps: RenderDeps) {
+  const queue: Job[] = [];
+  let running = false;
+  let current: string | null = null;
+
+  function enqueue(job: Job): void {
+    if (queue.some((j) => j.clipId === job.clipId) || current === job.clipId) return;
+    queue.push(job);
+    log.info(`In der Warteschlange: ${job.clipId} (${queue.length} offen)`);
+    void pump();
+  }
+
+  async function pump(): Promise<void> {
+    if (running) return;
+    if (!deps.cfg.renderWhileLive && deps.isLive()) {
+      log.debug('Stream laeuft - Rendern verschoben');
+      return;
+    }
+    running = true;
+    try {
+      let job: Job | undefined;
+      while ((job = queue.shift())) {
+        if (!deps.cfg.renderWhileLive && deps.isLive()) {
+          queue.unshift(job);            // zurueck in die Schlange
+          break;
+        }
+        current = job.clipId;
+        try {
+          await render(job);
+        } catch (err) {
+          deps.repo.updateClip(job.clipId, { status: 'failed' });
+          if (err instanceof FfmpegError) {
+            log.error(`Rendern fehlgeschlagen (${job.clipId}): ${err.message}`);
+            for (const line of err.tail.split('\n').slice(-6)) log.error(`  ${line}`);
+          } else {
+            log.error(`Rendern fehlgeschlagen (${job.clipId}): ${(err as Error).message}`);
+          }
+        } finally {
+          current = null;
+        }
+      }
+    } finally {
+      running = false;
+    }
+  }
+
+  async function render(job: Job): Promise<void> {
+    const started = Date.now();
+    const info = await probe(job.rawPath);
+    log.info(`Rendere ${job.clipId}: ${info.width}x${info.height}, ` +
+      `${info.durationSeconds.toFixed(1)} s, ${info.audio.length} Tonspur(en)`);
+
+    if (info.width !== 1920 || info.height !== 1080) {
+      log.warn(`Quelle ist ${info.width}x${info.height}, das Layout ist fuer 1920x1080 ` +
+        'eingemessen - der Ausschnitt sitzt vermutlich falsch.');
+    }
+
+    // Faellt die konfigurierte Tonspur weg, nimm die erste vorhandene
+    const track = deps.cfg.outputTrack < info.audio.length ? deps.cfg.outputTrack : 0;
+    if (track !== deps.cfg.outputTrack) {
+      log.warn(`Tonspur ${deps.cfg.outputTrack} gibt es nicht, nehme Spur ${track}`);
+    }
+
+    await fs.mkdir(job.outDir, { recursive: true });
+    const vertical = path.join(job.outDir, 'vertical.mp4');
+    const thumb = path.join(job.outDir, 'thumb.jpg');
+
+    let lastLogged = 0;
+    await run('ffmpeg', verticalArgs({
+      input: job.rawPath,
+      output: vertical,
+      profile: deps.cfg.profile,
+      tailSeconds: Math.min(deps.cfg.tailSeconds, Math.max(1, Math.floor(info.durationSeconds))),
+      audioTrack: track,
+      encoder: deps.cfg.encoder,
+      preset: deps.cfg.preset,
+      crf: deps.cfg.crf,
+    }), {
+      niceness: deps.cfg.niceness,
+      timeoutMs: 20 * 60_000,
+      onProgress: (s) => {
+        if (s - lastLogged >= 10) { lastLogged = s; log.debug(`  ${s.toFixed(0)} s verarbeitet`); }
+      },
+    });
+
+    await run('ffmpeg', thumbnailArgs(vertical, thumb), { niceness: deps.cfg.niceness });
+
+    const out = await probe(vertical);
+    deps.repo.updateClip(job.clipId, {
+      vertical_path: vertical,
+      duration_s: out.durationSeconds,
+      status: 'rendered',
+    });
+    deps.repo.event('clip.rendered', {
+      clipId: job.clipId, path: vertical, seconds: out.durationSeconds,
+    });
+    log.info(`Fertig: ${vertical} (${out.durationSeconds.toFixed(1)} s, ` +
+      `${((Date.now() - started) / 1000).toFixed(0)} s Rechenzeit)`);
+  }
+
+  return {
+    enqueue,
+    /** Nach Streamende aufrufen: arbeitet ab, was liegen geblieben ist. */
+    flush: () => { void pump(); },
+    pending: () => queue.length + (current ? 1 : 0),
+    currentJob: () => current,
+  };
+}