fix: harden install queue/dispatcher lifecycle and repair review-sweep regressions (#395)

Fixes 15 defects found by a max-effort multi-agent review of the last 6 merged PRs (#388, #390, #391, #392, #393, #394), all adversarially verified before fixing. Install queue + dispatcher (the serious cluster): - features.ts: finalize the installer child exactly once. A failed spawn fires both "error" and "close", and the second event released the file lock and active slot that pump() had just handed to the next queued bundle, letting two pip processes write the same venv concurrently. Outcome recording now happens before pump() so the next bundle's first progress frame cannot race the previous install's bookkeeping. - feature-status.ts: keep failed-install errors in a per-bundle map instead of the single progress slot. With the queue auto-starting the next install, the slot was overwritten within seconds and a failed install vanished without ever surfacing to GET /features. - bridge.ts: scope child lifecycle per process (stopped-children set + request generation tags) instead of an instance-wide shuttingDown flag that the next spawn reset. A stale SIGTERMed child's late close event could record a phantom crash (5 of which permanently disable the dispatcher), null out the freshly spawned child, and reject the new child's pending requests. The request-timeout kill path still counts as a real crash. - install_feature.py: the pre-write disk re-check measured ai_dir's filesystem even when budgeting the cross-filesystem copy that lands on the venv's disk; now each budget is checked against the filesystem the bytes actually land on, so ENOSPC cannot strike mid-write and leave site-packages half overwritten. Behavior regressions: - embed-subtitles: preserve pre-existing subtitle tracks (0:s?) and MKV attachments (0:t?) that the -map 0:v:0/0:a? rewrite silently dropped; data streams stay unmapped on purpose (the actual MPEG remux fix). The new subtitle maps first so the language tag hits the right stream. - usage-survey-overlay: fail closed when the settings fetch fails; the fail-open path rendered the blocking survey against an unhealthy API and soft-locked admins, the lock-out class #392 fixed. - features-store: queued bundles poll instead of each holding an SSE connection (Install All could pin 7 EventSources and exhaust the browser's 6-per-origin HTTP/1.1 limit, hanging the whole app); listenToProgress closes any prior stream and stops any poll before subscribing; installAll skips bundles already installing or queued. Contracts, tests, i18n: - openapi.yaml: add "queued" to the features status enum and document downloadBytes/installedBytes (Schemathesis conformance). - feature-lifecycle e2e: queue transcription (~0.5 GB) instead of ocr (~6 GB) and give the test a budget that covers both install drains (the stacked waits exceeded the old 900s timeout). - docker-compose.qa.yml: parameterize the host port (QA_APP_PORT) so QA_PROJECT_NAME concurrent stacks can actually bind. - compare + watermark-image: restore per-input error attribution ("Invalid first/second image", "Invalid watermark image") lost in the shared-handler migration. - ai-features-section: the "{size} on disk" suffix now goes through i18n; key added to all 21 locales. - watermark-image + content-aware-resize: migrate to the shared inputHandlerFor("image") chain like compare/vectorize/compose, fixing drift in the inline copies (no SVG sanitize, no RAW extension hint, no AVIF probe). Verified: typecheck across 9 workspaces, Biome clean on all changed files, 584 targeted unit tests and 249 integration tests green (including real-ffmpeg embed-subtitles runs). One unit test updated to the new poll-while-queued contract with a single-EventSource assertion. Claude-Session: https://claude.ai/code/session_017mR1HiHaf3a1BmUtrHX4j3
2026-08-03 07:46:42 +02:00 · 2026-07-03 13:47:15 +08:00
parent b37faed95f
commit b4375e558d
46 changed files with 383 additions and 255 deletions
@@ -51,7 +51,7 @@ def detect_arch() -> str:
    Only two archive variants are currently published to the bundle repo:
    'amd64-gpu' and 'arm64-cpu' (see deepsafe/feature-bundles). There is no
    CPU-only amd64 variant yet, so amd64 hosts always resolve to 'amd64-gpu'
-    even when no GPU is present -- this downloads working CUDA-capable
+    even when no GPU is present: this downloads working CUDA-capable
    packages, just larger than a CPU-only host strictly needs. Do not change
    this to branch on GPU presence without first publishing an 'amd64-cpu'
    archive for every bundle; requesting a key that doesn't exist in the
@@ -97,7 +97,7 @@ def estimate_extracted(compressed: int, extracted: int) -> int:
    extractedSize (0), the budget would otherwise collapse to just the
    compressed size and under-reserve for the extracted payload; fall back to a
    conservative 3x of compressed (measured extracted/compressed ratios reach
-    ~3x). This is only the early sanity bail -- the accurate guard is the
+    ~3x). This is only the early sanity bail; the accurate guard is the
    real-on-disk re-check just before the destructive venv write."""
    return extracted if extracted > 0 else compressed * 3

@@ -499,22 +499,23 @@ def main() -> None:
    # -- Disk re-check before the first destructive venv write --
    # The upfront check ran before the download and used an estimate; now the
    # payload is really on disk, so measure it and verify there's room to place
-    # it before we start writing into the venv. On the same filesystem the move
-    # is a rename (no extra space needed, just a safety floor); across
-    # filesystems it is a copy that transiently needs the payload's size again.
-    # Running here (after the local/remote branches merge) also covers the
-    # offline-import path, which skipped the upfront check entirely.
-    staging_real = dir_size(staging_dir)
-    if same_filesystem(staging_dir, venv_path):
-        recheck_needed = 1024 ** 3  # 1 GB floor for fixups / installed.json / slack
-    else:
-        recheck_needed = staging_real + 1024 ** 3
-    check_disk_space(ai_dir, recheck_needed)
+    # it before we start writing into the venv. Running here (after the
+    # local/remote branches merge) also covers the offline-import path, which
+    # skipped the upfront check entirely. Each budget is checked against the
+    # filesystem the bytes actually land on: when the venv lives on a
+    # different filesystem than staging, the site-packages payload is COPIED
+    # onto the venv's disk, so that disk (not ai_dir's) must hold it. Models
+    # stay under ai_dir either way, moving by rename.
+    disk_floor = 1024 ** 3  # 1 GB for fixups / installed.json / slack
+    staging_sp = os.path.join(staging_dir, "site-packages")
+    if not same_filesystem(staging_dir, venv_path):
+        sp_bytes = dir_size(staging_sp) if os.path.isdir(staging_sp) else 0
+        check_disk_space(venv_path, sp_bytes + disk_floor)
+    check_disk_space(ai_dir, disk_floor)

    # -- Move site-packages --
    emit_progress(92, "Installing packages...")
    site_packages_dir = get_site_packages_dir(venv_path)
-    staging_sp = os.path.join(staging_dir, "site-packages")

    try:
        if os.path.isdir(staging_sp) and site_packages_dir:
@@ -105,6 +105,8 @@ interface PendingRequest {
  reject: (err: Error) => void;
  onProgress?: ProgressCallback;
  stderrLines: string[];
+  /** Which child generation this request was written to (see startChild). */
+  generation: number;
 }

 // Crash recovery constants
@@ -133,11 +135,24 @@ export class PythonDispatcher {
  private childFailed = false;
  private gpuAvail = false;
  private pending = new Map<string, PendingRequest>();
-  private stdoutBuf = "";
  private crashes = 0;
  private lastCrashTs = 0;
  private backoffEnd = 0;
-  private shuttingDown = false;
+  /**
+   * Monotonic child counter. Each spawned child and every request written to
+   * it carry the generation current at spawn time, so a stale child's late
+   * close/error events (SIGTERM delivery can lag a replacement spawn) only
+   * ever touch their own generation's pending requests.
+   */
+  private generation = 0;
+  /**
+   * Children we SIGTERMed on purpose (shutdown/reload). Tracked per child
+   * rather than as an instance-wide flag: an instance flag reset by the next
+   * spawn would let the stale child's close event record a phantom crash and
+   * null out the fresh child. The request-timeout kill path deliberately does
+   * NOT add to this set, so a genuinely hung script still counts as a crash.
+   */
+  private stoppedChildren = new WeakSet<ChildProcess>();

  constructor(opts: { profile: "ai" | "docs" }) {
    this.profile = opts.profile;
@@ -173,9 +188,19 @@ export class PythonDispatcher {
    );
  }

+  /** Reject and drop the pending requests written to one child generation. */
+  private rejectPendingForGeneration(generation: number, message: string): void {
+    for (const [id, req] of this.pending.entries()) {
+      if (req.generation !== generation) continue;
+      req.reject(new Error(message));
+      this.pending.delete(id);
+    }
+  }
+
  private startChild(): ChildProcess | null {
    if (this.childFailed) return null;
-    this.shuttingDown = false;
+    this.generation++;
+    const gen = this.generation;

    try {
      const proc = spawn(getPythonPath(), [resolve(PYTHON_DIR, "dispatcher.py")], {
@@ -188,23 +213,23 @@ export class PythonDispatcher {
          console.error(
            `[bridge] Dispatcher stdin pipe broken (${err.code}), rejecting pending requests`,
          );
-          for (const [id, req] of this.pending.entries()) {
-            req.reject(new Error("Python dispatcher stdin closed unexpectedly"));
-            this.pending.delete(id);
+          this.rejectPendingForGeneration(gen, "Python dispatcher stdin closed unexpectedly");
+          // An intentional shutdown() ends stdin then SIGTERMs the child,
+          // which can surface here as an EPIPE/ERR_STREAM_DESTROYED. That is
+          // not a crash: counting it would let repeated legitimate restarts
+          // (shutdownDispatcher() runs after every AI bundle install) trip
+          // the crash limit and permanently disable the dispatcher. Guard
+          // mirrors the "close" handler below.
+          if (!this.stoppedChildren.has(proc)) this.recordCrash();
+          if (this.child === proc) {
+            this.child = null;
+            this.childReady = false;
          }
-          // An intentional shutdown() ends stdin then SIGTERMs the child, which
-          // can surface here as an EPIPE/ERR_STREAM_DESTROYED. That is not a
-          // crash -- counting it would let repeated legitimate restarts (e.g.
-          // shutdownDispatcher() on every AI bundle install) trip the crash
-          // limit and permanently disable the dispatcher. Guard mirrors the
-          // "close" handler below.
-          if (!this.shuttingDown) this.recordCrash();
-          this.child = null;
-          this.childReady = false;
        }
      });

      let stderrBuf = "";
+      let stdoutBuf = "";

      proc.stderr?.on("data", (chunk: Buffer) => {
        stderrBuf += chunk.toString();
@@ -218,21 +243,24 @@ export class PythonDispatcher {
          try {
            const parsed = JSON.parse(trimmed);

-            // Readiness signal
+            // Readiness signal. Ignore it from a superseded child so a stale
+            // process can't mark a not-yet-ready replacement as ready.
            if (parsed.ready === true) {
-              this.childReady = true;
-              this.gpuAvail = parsed.gpu === true;
-              this.crashes = 0;
-              console.log(`[bridge] Python dispatcher ready (GPU: ${parsed.gpu === true})`);
+              if (this.child === proc) {
+                this.childReady = true;
+                this.gpuAvail = parsed.gpu === true;
+                this.crashes = 0;
+                console.log(`[bridge] Python dispatcher ready (GPU: ${parsed.gpu === true})`);
+              }
              continue;
            }

            // Progress event - route to the currently active request
            if (typeof parsed.progress === "number" && typeof parsed.stage === "string") {
-              // Progress goes to all pending requests (only one should be active at a time
-              // since Python processes synchronously)
+              // Progress goes to this child's pending requests (only one
+              // should be active at a time since Python processes synchronously)
              for (const req of this.pending.values()) {
-                req.onProgress?.(parsed.progress, parsed.stage);
+                if (req.generation === gen) req.onProgress?.(parsed.progress, parsed.stage);
              }
            }
          } catch {
@@ -242,16 +270,16 @@ export class PythonDispatcher {
              console.log(`[python] ${trimmed}`);
            }
            for (const req of this.pending.values()) {
-              req.stderrLines.push(trimmed);
+              if (req.generation === gen) req.stderrLines.push(trimmed);
            }
          }
        }
      });

      proc.stdout?.on("data", (chunk: Buffer) => {
-        this.stdoutBuf += chunk.toString();
-        const lines = this.stdoutBuf.split("\n");
-        this.stdoutBuf = lines.pop() ?? "";
+        stdoutBuf += chunk.toString();
+        const lines = stdoutBuf.split("\n");
+        stdoutBuf = lines.pop() ?? "";

        for (const line of lines) {
          const trimmed = line.trim();
@@ -292,29 +320,27 @@ export class PythonDispatcher {
        console.error(`[bridge] Dispatcher error: ${err.message} (code: ${err.code})`);
        if (err.code === "ENOENT") {
          this.childFailed = true;
-        } else if (!this.shuttingDown) {
+        } else if (!this.stoppedChildren.has(proc)) {
          // Skip crash accounting when we initiated the teardown (shutdown()
-          // sets shuttingDown before killing the child); mirrors "close".
+          // marks the child stopped before killing it); mirrors "close".
          this.recordCrash();
        }
-        for (const [id, req] of this.pending.entries()) {
-          req.reject(new Error(extractPythonError(err)));
-          this.pending.delete(id);
+        this.rejectPendingForGeneration(gen, extractPythonError(err));
+        if (this.child === proc) {
+          this.child = null;
+          this.childReady = false;
        }
-        this.child = null;
-        this.childReady = false;
      });

      proc.on("close", (code) => {
-        for (const [id, req] of this.pending.entries()) {
-          req.reject(new Error("Python dispatcher exited unexpectedly"));
-          this.pending.delete(id);
-        }
-        if (code !== 0 && !this.shuttingDown) {
+        this.rejectPendingForGeneration(gen, "Python dispatcher exited unexpectedly");
+        if (code !== 0 && !this.stoppedChildren.has(proc)) {
          this.recordCrash();
        }
-        this.child = null;
-        this.childReady = false;
+        if (this.child === proc) {
+          this.child = null;
+          this.childReady = false;
+        }
      });

      return proc;
@@ -378,6 +404,9 @@ export class PythonDispatcher {
        reject: wrappedReject,
        onProgress: options.onProgress,
        stderrLines: [],
+        // getChild() above either reused or just spawned the child this
+        // request is written to, so the current generation is its generation.
+        generation: this.generation,
      });

      const msg: Record<string, unknown> = { id, script: scriptName.replace(".py", ""), args };
@@ -541,7 +570,7 @@ export class PythonDispatcher {
   */
  shutdown(): void {
    if (this.child && !this.child.killed) {
-      this.shuttingDown = true;
+      this.stoppedChildren.add(this.child);
      this.child.stdin?.end();
      this.child.kill("SIGTERM");
      this.child = null;
@@ -10,7 +10,7 @@ import type {
 } from "../types.js";

 /**
- * Above this pixel count, CLAHE is skipped in applyCorrections() -- see the
+ * Above this pixel count, CLAHE is skipped in applyCorrections(); see the
 * comment at its call site for why.
 */
 const MAX_CLAHE_PIXELS = 16_000_000;
@@ -230,7 +230,7 @@ export function applyCorrections(

  let result = image;
  // Tracks whether .clahe() actually ran (not just whether the toggle allowed
-  // it) -- Step 5 below applies a compensation boost keyed off this, and it
+  // it). Step 5 below applies a compensation boost keyed off this, and it
  // needs to stay correct now that CLAHE can also be skipped by image size.
  let claheApplied = false;

@@ -238,7 +238,7 @@ export function applyCorrections(
  // maxSlope must be an integer (Sharp requirement); skip for tiny images.
  // CLAHE's cost scales with total pixel count regardless of tile size (tile
  // size only bounds granularity, not the per-pixel histogram/interpolation
-  // work), so it's also skipped above MAX_CLAHE_PIXELS -- a 5504x3672 (20MP)
+  // work), so it's also skipped above MAX_CLAHE_PIXELS: a 5504x3672 (20MP)
  // real-world RAW photo measured 40+ seconds in this step alone versus ~1s
  // for every other correction combined. The other six corrections below
  // still apply at full resolution regardless of size.
@@ -3475,6 +3475,7 @@ export const ar: TranslationKeys = {
      description: "إدارة حزم نماذج AI لمعالجة الملفات المتقدمة.",
      installAll: "تثبيت الكل",
      diskUsage: "استخدام القرص: {size}",
+      sizeOnDisk: "{size} على القرص",
      installed: "مُثبَّت",
      notInstalled: "غير مُثبَّت",
      queued: "في قائمة الانتظار",
@@ -3512,6 +3512,7 @@ export const de: TranslationKeys = {
      description: "AI-Modellpakete für erweiterte Dateiverarbeitung verwalten.",
      installAll: "Alle installieren",
      diskUsage: "Speicherplatznutzung: {size}",
+      sizeOnDisk: "{size} auf der Festplatte",
      installed: "Installiert",
      notInstalled: "Nicht installiert",
      queued: "In Warteschlange",
@@ -3425,6 +3425,7 @@ export const en = {
      description: "Manage AI model bundles for advanced file processing.",
      installAll: "Install All",
      diskUsage: "Disk usage: {size}",
+      sizeOnDisk: "{size} on disk",
      installed: "Installed",
      notInstalled: "Not installed",
      queued: "Queued",
@@ -3492,6 +3492,7 @@ export const es: TranslationKeys = {
      description: "Gestiona paquetes de modelos AI para procesamiento avanzado de archivos.",
      installAll: "Instalar todo",
      diskUsage: "Uso del disco: {size}",
+      sizeOnDisk: "{size} en disco",
      installed: "Instalado",
      notInstalled: "No instalado",
      queued: "En cola",
@@ -3516,6 +3516,7 @@ export const fr: TranslationKeys = {
      description: "Gérez les paquets de modèles AI pour le traitement avancé de fichiers.",
      installAll: "Tout installer",
      diskUsage: "Utilisation du disque : {size}",
+      sizeOnDisk: "{size} sur le disque",
      installed: "Installé",
      notInstalled: "Non installé",
      queued: "En file d'attente",
@@ -3303,6 +3303,7 @@ export const hi: TranslationKeys = {
      description: "उन्नत फ़ाइल प्रोसेसिंग के लिए AI मॉडल बंडल प्रबंधित करें।",
      installAll: "सभी इंस्टॉल करें",
      diskUsage: "डिस्क उपयोग: {size}",
+      sizeOnDisk: "डिस्क पर {size}",
      installed: "इंस्टॉल किया गया",
      notInstalled: "इंस्टॉल नहीं है",
      queued: "कतार में",
@@ -3494,6 +3494,7 @@ export const id: TranslationKeys = {
      description: "Kelola bundel model AI untuk pemrosesan file tingkat lanjut.",
      installAll: "Instal Semua",
      diskUsage: "Penggunaan disk: {size}",
+      sizeOnDisk: "{size} di disk",
      installed: "Terinstal",
      notInstalled: "Belum terinstal",
      queued: "Dalam antrean",
@@ -3506,6 +3506,7 @@ export const it: TranslationKeys = {
      description: "Gestisci pacchetti di modelli di IA per l'elaborazione avanzata dei file.",
      installAll: "Installa tutto",
      diskUsage: "Utilizzo del disco: {size}",
+      sizeOnDisk: "{size} su disco",
      installed: "Installato",
      notInstalled: "Non installato",
      queued: "Code",
@@ -3445,6 +3445,7 @@ export const ja: TranslationKeys = {
      description: "高度なファイル処理のためのAIモデルバンドルを管理します。",
      installAll: "すべてインストール",
      diskUsage: "ディスク使用量: {size}",
+      sizeOnDisk: "ディスク上 {size}",
      installed: "インストール済み",
      notInstalled: "未インストール",
      queued: "待機中",
@@ -3429,6 +3429,7 @@ export const ko: TranslationKeys = {
      description: "고급 파일 처리를 위한 AI 모델 번들을 관리합니다.",
      installAll: "모두 설치",
      diskUsage: "디스크 사용량: {size}",
+      sizeOnDisk: "디스크에 {size}",
      installed: "설치됨",
      notInstalled: "설치되지 않음",
      queued: "대기 중",
@@ -3502,6 +3502,7 @@ export const nl: TranslationKeys = {
      description: "Beheer AI-modelbundels voor geavanceerde bestandsverwerking.",
      installAll: "Alles installeren",
      diskUsage: "Schijfgebruik: {size}",
+      sizeOnDisk: "{size} op schijf",
      installed: "Geïnstalleerd",
      notInstalled: "Niet geïnstalleerd",
      queued: "In wachtrij",
@@ -3507,6 +3507,7 @@ export const pl: TranslationKeys = {
      description: "Zarządzanie pakietami modeli AI do zaawansowanego przetwarzania plików.",
      installAll: "Zainstaluj wszystkie",
      diskUsage: "Wykorzystanie dysku: {size}",
+      sizeOnDisk: "{size} na dysku",
      installed: "Zainstalowane",
      notInstalled: "Niezainstalowane",
      queued: "W kolejce",
@@ -3500,6 +3500,7 @@ export const ptBR: TranslationKeys = {
      description: "Gerencie pacotes de modelos de IA para processamento avançado de arquivos.",
      installAll: "Instalar tudo",
      diskUsage: "Uso do disco: {size}",
+      sizeOnDisk: "{size} em disco",
      installed: "Instalado",
      notInstalled: "Não instalado",
      queued: "Na fila",
@@ -3497,6 +3497,7 @@ export const ru: TranslationKeys = {
      description: "Управление пакетами AI-моделей для расширенной обработки файлов.",
      installAll: "Установить все",
      diskUsage: "Использование диска: {size}",
+      sizeOnDisk: "{size} на диске",
      installed: "Установлено",
      notInstalled: "Не установлено",
      queued: "В очереди",
@@ -3492,6 +3492,7 @@ export const sv: TranslationKeys = {
      description: "Hantera AI-modellpaket för avancerad filbehandling.",
      installAll: "Installera alla",
      diskUsage: "Diskanvändning: {size}",
+      sizeOnDisk: "{size} på disk",
      installed: "Installerad",
      notInstalled: "Inte installerad",
      queued: "I kö",
@@ -3456,6 +3456,7 @@ export const th: TranslationKeys = {
      description: "จัดการชุดโมเดล AI สำหรับการประมวลผลไฟล์ขั้นสูง",
      installAll: "ติดตั้งทั้งหมด",
      diskUsage: "การใช้พื้นที่ดิสก์: {size}",
+      sizeOnDisk: "{size} บนดิสก์",
      installed: "ติดตั้งแล้ว",
      notInstalled: "ยังไม่ได้ติดตั้ง",
      queued: "อยู่ในคิว",
@@ -3500,6 +3500,7 @@ export const tr: TranslationKeys = {
      description: "Gelişmiş dosya işleme için AI model paketlerini yönetin.",
      installAll: "Tümünü Kur",
      diskUsage: "Disk kullanımı: {size}",
+      sizeOnDisk: "diskte {size}",
      installed: "Kurulu",
      notInstalled: "Kurulu değil",
      queued: "Sırada",
@@ -3497,6 +3497,7 @@ export const uk: TranslationKeys = {
      description: "Керування пакетами AI-моделей для розширеної обробки файлів.",
      installAll: "Встановити все",
      diskUsage: "Використання диска: {size}",
+      sizeOnDisk: "{size} на диску",
      installed: "Встановлено",
      notInstalled: "Не встановлено",
      queued: "У черзі",
@@ -3492,6 +3492,7 @@ export const vi: TranslationKeys = {
      description: "Quản lý các gói mô hình AI cho xử lý tệp nâng cao.",
      installAll: "Cài đặt tất cả",
      diskUsage: "Dung lượng đĩa: {size}",
+      sizeOnDisk: "{size} trên ổ đĩa",
      installed: "Đã cài đặt",
      notInstalled: "Chưa cài đặt",
      queued: "Đang chờ",
@@ -3243,6 +3243,7 @@ export const zhCN: TranslationKeys = {
      description: "管理用于高级文件处理的 AI 模型包。",
      installAll: "全部安装",
      diskUsage: "磁盘使用量：{size}",
+      sizeOnDisk: "占用磁盘 {size}",
      installed: "已安装",
      notInstalled: "未安装",
      queued: "排队中",
@@ -3241,6 +3241,7 @@ export const zhTW: TranslationKeys = {
      description: "管理用於進階檔案處理的AI模型套件。",
      installAll: "全部安裝",
      diskUsage: "磁碟使用量：{size}",
+      sizeOnDisk: "佔用磁碟 {size}",
      installed: "已安裝",
      notInstalled: "未安裝",
      queued: "等待中",