fix(plex): give the sweep's boot-time cutoff cross-clock slack

Found by running the round-8 fix in the real container instead of trusting the
unit test: a directory created 2ms AFTER `_PROCESS_START` was stamped with an
mtime 2ms BEFORE it, so the sweep deleted it as a leftover. The two sides come
from different clocks — a file's mtime from the kernel's coarse clock (one timer
tick of lag), `time.time()` from the fine one — and the unit test could not see it
because it supplies an artificial cutoff.

The window is small and lands exactly where it hurts: the sweep runs at boot, so
the session at risk is one started in the first instants after startup — the case
the guard exists for. Keep-side slack of 2s; the cost is that a leftover from the
last seconds before a restart survives until the next one.

Verified in the container: a backdated leftover is swept, a directory created by
this run is kept, and a registered live session with an old mtime is kept — 1 of 3.
Test pins the skew (mutation-checked: with the grace at 0 it goes red).

backend/app/plex/stream.py

@@ -51,6 +51,15 @@ _SESSION_DIR_RE = re.compile(r"^\d+_\d+(_[^_]+_[+-]\d+\.\d{2})?$")
 # newer belongs to a session THIS run created, and since `start_session` reuses a directory path
 # for the same (rating_key, offset, tag), "stale name" is not the same as "stale directory".
 _PROCESS_START = time.time()
+# Slack on that comparison, because the two sides come from DIFFERENT clocks. A file's mtime is
+# stamped from the kernel's COARSE clock (updated once per timer tick), while `time.time()` reads
+# the fine-grained one — measured in this very container, a directory created 2ms AFTER
+# `_PROCESS_START` got an mtime 2ms BEFORE it. Without slack the sweep would classify a session
+# started in the first instants after boot as a leftover and delete it while ffmpeg writes into it
+# — and "the first instants after boot" is exactly when the sweep runs. Erring on the keep side
+# costs nothing: a genuine leftover from the last seconds before the restart simply survives until
+# the next one.
+_MTIME_GRACE_S = 2.0
 
 _lock = threading.Lock()
 _sessions: dict[str, "HlsSession"] = {}
@@ -286,15 +295,15 @@ def sweep_orphans(older_than: float | None = None) -> int:
 
     TWO guards against deleting a LIVE session, because this now runs CONCURRENTLY with request
     handling (the caller stopped awaiting it so startup isn't held up):
-    * `older_than` (default: this process's start) — a directory touched since we booted was made
-      by THIS run, so it is by definition not a leftover. This is the guard that matters, because
-      `start_session` REUSES the same path for the same (key, offset, tag): a stale directory can
-      become a live one at any moment.
+    * `older_than` (default: this process's start, minus `_MTIME_GRACE_S` of cross-clock slack) —
+      a directory touched since we booted was made by THIS run, so it is by definition not a
+      leftover. This is the guard that matters, because `start_session` REUSES the same path for
+      the same (key, offset, tag): a stale directory can become a live one at any moment.
     * the live-session check and the rmtree happen together under `_lock`, so a session can't be
       registered in the gap between them. The lock is held for one directory at a time — long
       enough to be atomic, short enough that a boot-time sweep of a big backlog doesn't stall
       playback."""
-    cutoff = _PROCESS_START if older_than is None else older_than
+    cutoff = (_PROCESS_START - _MTIME_GRACE_S) if older_than is None else older_than
     dropped = 0
     try:
         entries = list(_HLS_ROOT.iterdir())

backend/tests/test_plex_stream.py

@@ -126,10 +126,21 @@ class TestSweepOrphans:
         # The parameter exists for the tests; production calls it with no argument, so pin what
         # that means — otherwise the guard above could be right and unused.
         d = _session_dir(hls_root, "12345_0_None_+0.00")
+        monkeypatch.setattr(stream, "_MTIME_GRACE_S", 0.0)
         monkeypatch.setattr(stream, "_PROCESS_START", d.stat().st_mtime + 1)
         assert stream.sweep_orphans() == 1  # "older than boot" → swept
         assert not d.exists()
 
+    def test_a_directory_stamped_just_before_boot_survives(self, hls_root, monkeypatch):
+        # MEASURED in the container, not theory: a file's mtime comes from the kernel's COARSE
+        # clock and `time.time()` from the fine one, so a directory created 2ms AFTER
+        # `_PROCESS_START` was stamped 2ms BEFORE it. Without the grace the sweep deletes a session
+        # that started in the first instants after boot — which is exactly when the sweep runs.
+        d = _session_dir(hls_root, "12345_0_None_+0.00")
+        monkeypatch.setattr(stream, "_PROCESS_START", d.stat().st_mtime + 0.05)  # the skew
+        assert stream.sweep_orphans() == 0
+        assert d.exists()
+
     def test_a_missing_root_is_not_an_error(self, tmp_path, monkeypatch):
         monkeypatch.setattr(stream, "_HLS_ROOT", tmp_path / "nope")
         monkeypatch.setattr(stream, "_sessions", {})