v1.5.25 — fd-leak watchdog: restart worker before EBADF crash (#89)

Author: epheterson

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## 1.5.25 — 2026-07-05
+
+### Fixed
+- **Worker crash from file-descriptor exhaustion** (#89): Immich leaks file handles while processing some media (a reporter hit it on Sony A7IV XAVC files on an external drive), opening each source file and never closing it. His diagnostic (from v1.5.23) showed ~49,000 open descriptors, all pointing at the source files; once the fd table is that large macOS fails `spawn()` with `EBADF` and the worker crashes. The watcher now monitors the worker's live open-fd count (summed across worker processes, via libproc) and restarts the worker before the table gets dangerous (default 10,000; a healthy worker sits around 150), with a cooldown to avoid thrash. Tunable via `IMMICH_ACCEL_FD_RESTART_THRESHOLD` (0 disables) and `IMMICH_ACCEL_FD_RESTART_COOLDOWN`. This is a safety net for an upstream Immich leak, not a fix for the leak itself.
+
 ## 1.5.24 — 2026-07-03
 
 ### Immich 3.0 support

VERSION

@@ -1 +1 @@
-1.5.24
+1.5.25

immich_accelerator/__main__.py

@@ -10,6 +10,7 @@
 from __future__ import annotations
 
 import argparse
+import ctypes
 import json
 import logging
 import os
@@ -1653,6 +1654,68 @@ def kill_pid(name: str) -> bool:
     return True
 
 
+try:
+    _LIBPROC = ctypes.CDLL("libproc.dylib", use_errno=True)
+    _LIBPROC.proc_pidinfo.restype = ctypes.c_int
+    _LIBPROC.proc_pidinfo.argtypes = [
+        ctypes.c_int,
+        ctypes.c_int,
+        ctypes.c_uint64,
+        ctypes.c_void_p,
+        ctypes.c_int,
+    ]
+except OSError:
+    _LIBPROC = None
+
+_PROC_PIDLISTFDS = 1
+_PROC_FDINFO_SIZE = 8  # sizeof(struct proc_fdinfo): int32 fd + uint32 fdtype
+
+
+def _process_fd_count(pid: int) -> int | None:
+    """Live open file-descriptor count for a pid via libproc, or None.
+
+    Used by the watcher's fd-leak safety net (#89). proc_pidinfo(PROC_PIDLISTFDS)
+    with a NULL buffer returns the fd-table CAPACITY, a high-water mark that
+    never shrinks when fds close, so we must pass a real buffer and count the
+    bytes actually written (live_nfds * sizeof(proc_fdinfo)). Two syscalls, no
+    subprocess: unlike lsof it can't hang on a stalled mount, does no DNS on the
+    worker's DB/Redis sockets, and stays fast even when the table is huge (the
+    exact case we must catch).
+    """
+    if _LIBPROC is None:
+        return None
+    capacity = _LIBPROC.proc_pidinfo(pid, _PROC_PIDLISTFDS, 0, None, 0)
+    if capacity <= 0:
+        return None
+    buf = ctypes.create_string_buffer(capacity)
+    written = _LIBPROC.proc_pidinfo(pid, _PROC_PIDLISTFDS, 0, buf, capacity)
+    if written <= 0:
+        return None
+    return written // _PROC_FDINFO_SIZE
+
+
+def _worker_fd_total() -> int | None:
+    """Total open fds across all live Immich worker processes, or None.
+
+    Immich 2.7+ runs several processes titled 'immich'; the file-handle leak
+    (#89) can accumulate in any of them, not just the one tracked in worker.pid,
+    so sum every worker process rather than measuring a single pid.
+    """
+    if _LIBPROC is None:
+        return None  # can't count fds; skip the ps scan entirely
+    pids = _scan_worker_pids()
+    if not pids:
+        return None
+    total = 0
+    counted = False
+    for pid in pids:
+        count = _process_fd_count(pid)
+        if count is not None:
+            total += count
+            counted = True
+    return total if counted else None
+
+
 def cap_log(log_path: Path, max_bytes: int = LOG_MAX_BYTES) -> bool:
     """Cap a service log in place once it exceeds max_bytes, preserving the last
     LOG_KEEP_TAIL_LINES lines of context. Returns True if it rotated.
@@ -3686,7 +3749,7 @@ def cmd_start(args):
     # /build link points to our build-data dir (set up during setup).
     # Required for Immich 2.7+ plugin WASM paths stored in the shared DB.
     build_data = DATA_DIR / "build-data"
-    has_plugins = (build_data / "corePlugin" / "manifest.json").exists()
+    has_plugins = _build_has_core_plugin(build_data)
 
     if _build_link_ok():
         pass  # /build resolves correctly, both Docker and native see the same paths
@@ -3954,13 +4017,42 @@ def cmd_update(_args):
     log.info("Updated to %s. Run: python -m immich_accelerator start", running)
 
 
+def _int_env(name: str, default: int) -> int:
+    """Parse an int env var, falling back to default on a missing/bad value.
+
+    Must never raise at import: these are module-level constants, so a bad
+    value would otherwise break every subcommand, not just the watcher.
+    """
+    try:
+        return int(os.environ.get(name, str(default)))
+    except (ValueError, TypeError):
+        return default
+
+
+# fd-leak safety net (#89): Immich leaks file handles processing some media
+# (e.g. certain Sony XAVC files on an external drive), opening each source file
+# and never closing it. Once the fd table gets huge, macOS starts failing
+# spawn() with EBADF and the worker crashes. A healthy worker sits around 150
+# open fds; restart it well before the wall. 0 disables the check.
+FD_RESTART_THRESHOLD = _int_env("IMMICH_ACCEL_FD_RESTART_THRESHOLD", 10000)
+# Minimum seconds between fd-triggered restarts, so a fast leak (or a high
+# post-restart baseline) can't thrash the worker with back-to-back restarts.
+FD_RESTART_COOLDOWN = _int_env("IMMICH_ACCEL_FD_RESTART_COOLDOWN", 300)
+
+
 def cmd_watch(_args):
     """Monitor services and restart on crash. Detects Docker updates.
 
     Suitable for launchd KeepAlive — runs forever, checking every 30s.
     """
     log.info("Watching services (Ctrl+C to stop)...")
 
+    if FD_RESTART_THRESHOLD > 0 and _LIBPROC is None:
+        log.warning(
+            "fd-leak watchdog (#89) inactive: libproc unavailable, cannot read "
+            "worker fd counts on this system."
+        )
+
     # `brew services stop`/`restart` send SIGTERM to this watcher, but the
     # worker/ML/dashboard run detached (own session) and would survive — so a
     # plain stop/restart left them running (#81). Trap the signal and stop the
@@ -4012,10 +4104,15 @@ def _graceful_stop(signum, _frame):
     check_count = 0
     self_update_notified = False
     shown_worker_hints: set[str] = set()
+    # None until the fd watchdog first restarts the worker. Compared against
+    # time.monotonic() (uptime), so a 0.0 sentinel would wrongly read as "just
+    # restarted" for the first FD_RESTART_COOLDOWN seconds after boot.
+    last_fd_restart: float | None = None
     while True:
         try:
             time.sleep(30)
             config = load_config()  # reload each cycle (setup may have changed it)
+            worker_handled = False  # set by the fd watchdog to skip crash-check
 
             # Auto-apply a `brew upgrade`: if the version installed on disk no
             # longer matches the code we're running, stop the (now-stale)
@@ -4060,8 +4157,49 @@ def _graceful_stop(signum, _frame):
                     except RuntimeError:
                         log.error("  ML restart failed")
 
+            # fd-leak safety net (#89): restart the worker before a runaway
+            # open-file-descriptor count (an upstream Immich leak on some media)
+            # exhausts the table and crashes it with `spawn EBADF`. Sum fds
+            # across all worker processes (the leak may be in a sibling), and
+            # cool down between restarts so a fast leak can't thrash.
+            if FD_RESTART_THRESHOLD > 0:
+                fd_total = _worker_fd_total()
+                if fd_total is not None and fd_total >= FD_RESTART_THRESHOLD:
+                    cooled = (
+                        last_fd_restart is None
+                        or time.monotonic() - last_fd_restart >= FD_RESTART_COOLDOWN
+                    )
+                    if cooled:
+                        last_fd_restart = time.monotonic()
+                        log.warning(
+                            "Worker fd count %d exceeds %d: Immich leaks file "
+                            "handles on some media (#89). Restarting the worker "
+                            "before it exhausts the fd table and crashes.",
+                            fd_total,
+                            FD_RESTART_THRESHOLD,
+                        )
+                        kill_pid("worker")
+                        try:
+                            cmd_start(argparse.Namespace(force=True))
+                        except RuntimeError:
+                            log.error("  Worker restart after fd leak failed")
+                        # We handled the worker: skip the crash-check below so it
+                        # can't also log "Worker crashed" and double-start this
+                        # cycle. We do NOT `continue` (that would starve the
+                        # update-detection cadence); if cmd_start early-returned
+                        # without a live worker, next cycle's crash-check
+                        # restarts it.
+                        worker_handled = True
+                    else:
+                        log.debug(
+                            "Worker fd count %d high but within restart "
+                            "cooldown (%ds); not restarting yet.",
+                            fd_total,
+                            FD_RESTART_COOLDOWN,
+                        )
+
             # Check worker
-            if not read_pid("worker"):
+            if not worker_handled and not read_pid("worker"):
                 # Surface a known unrecoverable cause once (e.g. a fresh
                 # split-deploy geodata import failing) instead of looping
                 # silently — restarting won't fix it until the user acts.
@@ -4077,9 +4215,12 @@ def _graceful_stop(signum, _frame):
                 except RuntimeError:
                     log.error("  Worker restart failed, will retry in 30s")
 
-            # Every 5 min, check if Immich updated
+            # Every 5 min, check if Immich updated. Skip on a cycle where the
+            # fd watchdog just restarted the worker, so we don't tear it back
+            # down (cmd_stop + re-extract) in the same tick; check_count stays
+            # >=10 so it runs next cycle instead.
             check_count += 1
-            if check_count >= 10:
+            if check_count >= 10 and not worker_handled:
                 check_count = 0
                 try:
                     cached = config.get("version", "").lstrip("v")

tests/test_accelerator.py

@@ -1522,3 +1522,99 @@ def test_false_when_absent(self, tmp_path):
         assert _build_has_core_plugin(tmp_path) is False
         (tmp_path / "plugins").mkdir()  # empty plugins dir is not enough
         assert _build_has_core_plugin(tmp_path) is False
+
+
+class TestProcessFdCount:
+    """fd-count helpers for the #89 fd-leak watchdog (libproc based)."""
+
+    def test_counts_live_fds_from_buffered_call(self):
+        from immich_accelerator.__main__ import _process_fd_count
+        from unittest.mock import MagicMock
+
+        # First call (NULL buffer) returns the capacity high-water (1600 bytes);
+        # second call (real buffer) returns the LIVE bytes written (960 -> 120).
+        libproc = MagicMock()
+        libproc.proc_pidinfo.side_effect = [1600, 960]
+        with patch("immich_accelerator.__main__._LIBPROC", libproc):
+            assert _process_fd_count(1234) == 120  # live count, not 1600/8
+
+    def test_none_when_buffered_call_fails(self):
+        from immich_accelerator.__main__ import _process_fd_count
+        from unittest.mock import MagicMock
+
+        libproc = MagicMock()
+        libproc.proc_pidinfo.side_effect = [1600, 0]  # capacity ok, write fails
+        with patch("immich_accelerator.__main__._LIBPROC", libproc):
+            assert _process_fd_count(1234) is None
+
+    def test_none_when_libproc_unavailable(self):
+        from immich_accelerator.__main__ import _process_fd_count
+
+        with patch("immich_accelerator.__main__._LIBPROC", None):
+            assert _process_fd_count(1234) is None
+
+    def test_none_on_nonpositive_return(self):
+        from immich_accelerator.__main__ import _process_fd_count
+        from unittest.mock import MagicMock
+
+        libproc = MagicMock()
+        libproc.proc_pidinfo.return_value = 0  # dead pid / error
+        with patch("immich_accelerator.__main__._LIBPROC", libproc):
+            assert _process_fd_count(1234) is None
+
+    def test_worker_fd_total_sums_all_workers(self):
+        from immich_accelerator.__main__ import _worker_fd_total
+        from unittest.mock import MagicMock
+
+        # _LIBPROC must be truthy or _worker_fd_total short-circuits (it is None
+        # on non-macOS CI, where libproc.dylib doesn't exist).
+        with patch("immich_accelerator.__main__._LIBPROC", MagicMock()), patch(
+            "immich_accelerator.__main__._scan_worker_pids",
+            return_value=[10, 20, 30],
+        ), patch(
+            "immich_accelerator.__main__._process_fd_count",
+            side_effect=lambda p: {10: 100, 20: None, 30: 5000}.get(p),
+        ):
+            # 100 + 5000, the None (unreadable pid) is skipped
+            assert _worker_fd_total() == 5100
+
+    def test_worker_fd_total_none_when_no_workers(self):
+        from immich_accelerator.__main__ import _worker_fd_total
+        from unittest.mock import MagicMock
+
+        with patch("immich_accelerator.__main__._LIBPROC", MagicMock()), patch(
+            "immich_accelerator.__main__._scan_worker_pids", return_value=[]
+        ):
+            assert _worker_fd_total() is None
+
+    def test_worker_fd_total_none_when_libproc_unavailable(self):
+        from immich_accelerator.__main__ import _worker_fd_total
+
+        # No libproc: skip the ps scan entirely and report None.
+        with patch("immich_accelerator.__main__._LIBPROC", None), patch(
+            "immich_accelerator.__main__._scan_worker_pids"
+        ) as scan:
+            assert _worker_fd_total() is None
+            scan.assert_not_called()
+
+
+class TestIntEnv:
+    """Safe int env parsing for the fd-watchdog thresholds (#89)."""
+
+    def test_valid_value(self, monkeypatch):
+        from immich_accelerator.__main__ import _int_env
+
+        monkeypatch.setenv("IAA_TEST_INT", "42")
+        assert _int_env("IAA_TEST_INT", 10) == 42
+
+    def test_missing_falls_back(self, monkeypatch):
+        from immich_accelerator.__main__ import _int_env
+
+        monkeypatch.delenv("IAA_TEST_INT", raising=False)
+        assert _int_env("IAA_TEST_INT", 10) == 10
+
+    def test_bad_value_falls_back_not_raises(self, monkeypatch):
+        from immich_accelerator.__main__ import _int_env
+
+        monkeypatch.setenv("IAA_TEST_INT", "10k")
+        assert _int_env("IAA_TEST_INT", 10) == 10