gate: decode both sides sequentially; lock: heartbeat + age-based staleness

Two bugs the library run exposed on real files:

* An input seek is not a faithful decode. On an H.264 source with open GOPs the frames after the
  landing point decode differently from a sequential decode - the SAME file compared with itself,
  once seeked and once decoded in order, scored min 0.25 / mean 0.82, and the gate condemned good
  encodes because of it. Both sides are now decoded sequentially with the window cut by the trim
  filter, which is also the only decode guaranteed to match what the encoder itself saw.
* Staleness of the run lock. Killing a background session kills the wrapper but not the python
  underneath, so a live run's lock survives and the next start is correctly refused (it looked like
  a pid-reuse bug; it was not). The lock now carries a heartbeat, staleness is decided by age as
  well as pid, and a takeover that cannot delete a file the OS still holds overwrites it in place.

scripts/library_run.py

@@ -275,19 +275,25 @@ def ssim_vs_source(src_local, out_local, info, workdir, tag, dur=SSIM_WINDOW):
     ref = os.path.join(workdir, "ssim_ref.yuv")
     out = os.path.join(workdir, "ssim_out.yuv")
     counts = []
-    # Seek a little BEFORE the window and cut it with the trim filter on ABSOLUTE timestamps
-    # (`-copyts`): an input seek snaps to a keyframe, and it snapped to a *different* keyframe for
-    # the source and the encode (measured: two 1500-frame decodes whose start times differed by
-    # ~10 s -> SSIM mean 0.45 on a perfectly good encode). Trimming on PTS makes both sides cover
-    # exactly the same interval no matter where the seek landed.
-    seek = max(0.0, start - 20.0)
+    # Decode WITHOUT an input seek, and cut the window with the trim filter on absolute PTS.
+    #
+    # Two traps live here, both measured on real files in this library:
+    #   1. An input seek snaps to a keyframe, and it snaps to a DIFFERENT keyframe for the source
+    #      and for the encode (two 1500-frame decodes whose start times differed by ~10 s), so the
+    #      frames met the wrong partners and a good encode scored mean 0.45.
+    #   2. Worse, on an H.264 source with open GOPs an input seek decodes the frames after the
+    #      landing point differently from a sequential decode: the SAME file compared with itself,
+    #      once seeked and once decoded in order, scored min 0.25 / mean 0.82. A seek-based
+    #      reference therefore condemns perfectly good encodes.
+    # Sequential decoding costs a pass over the file up to the window at 480x270 - small next to the
+    # encode it is judging, and it is the only decode that is guaranteed to match what the encoder
+    # itself saw.
     vf = ("trim=start=%.3f:end=%.3f,setpts=PTS-STARTPTS,scale=%s:flags=bilinear,format=yuv420p"
           % (start, start + dur, cmp_scale))
     for path, dstp in ((src_local, ref), (out_local, out)):
-        r = run([FFMPEG, "-y", "-v", "error", "-copyts", "-ss", "%.2f" % seek,
-                 "-t", "%.2f" % (dur + 40), "-i", path,
+        r = run([FFMPEG, "-y", "-v", "error", "-i", path,
                  "-map", "0:v:0", "-an", "-vf", vf,
-                 "-fps_mode", "passthrough", "-f", "rawvideo", dstp], timeout=3600)
+                 "-fps_mode", "passthrough", "-f", "rawvideo", dstp], timeout=7200)
         if r.returncode != 0 or not os.path.exists(dstp) or not os.path.getsize(dstp):
             return None, None, 0, None
         counts.append(os.path.getsize(dstp) // (480 * 270 * 3 // 2))

scripts/transcode.py

@@ -30,6 +30,7 @@ import os
 import shutil
 import subprocess
 import sys
+import threading
 import time
 
 FFMPEG = os.environ.get("TRANSCODE_FFMPEG", "ffmpeg")
@@ -308,11 +309,29 @@ def out_path(src, args):
 
 class Lock:
     """One encoder at a time. Two concurrent NVENC jobs starve each other and drop frames -
-    the exact defect this tool exists to prevent - so cron runs must never overlap."""
+    the exact defect this tool exists to prevent - so cron runs must never overlap.
+
+    A lock carries a HEARTBEAT: a daemon thread touches the file every minute, so a live run's lock
+    is always fresh. Staleness is therefore decided by age first and pid second - killing a run
+    leaves its lock behind and the recorded pid can be REUSED by an unrelated process, which made a
+    pid-only check refuse to start a perfectly fine run (measured: a stale lock naming pid 29848,
+    which by then belonged to something else).
+    """
+
+    HEARTBEAT_S = 60
+    STALE_S = 300
 
     def __init__(self, path):
         self.path = path
         self.fd = None
+        self._stop = None
+
+    def _beat(self):
+        while not self._stop.wait(self.HEARTBEAT_S):
+            try:
+                os.utime(self.path, None)
+            except OSError:
+                return
 
     def __enter__(self):
         os.makedirs(os.path.dirname(os.path.abspath(self.path)) or ".", exist_ok=True)
@@ -324,17 +343,28 @@ class Lock:
                 pid = int(open(self.path).read().strip() or 0)
             except Exception:
                 pid = 0
-            if pid and _alive(pid):
-                raise SystemExit("another transcode is already running (pid %d, %s) - refusing "
-                                 "to start a second encoder" % (pid, self.path))
-            log("stale lock (pid %s gone) - taking it over" % pid)
-            os.remove(self.path)
-            self.fd = os.open(self.path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
+            age = time.time() - os.path.getmtime(self.path)
+            if age < self.STALE_S and pid and _alive(pid):
+                raise SystemExit("another transcode is already running (pid %d, %s, %.0f s old) - "
+                                 "refusing to start a second encoder" % (pid, self.path, age))
+            log("stale lock (pid %s, %.0f s old) - taking it over" % (pid, age))
+            try:
+                os.remove(self.path)
+            except OSError:
+                # A dead process can leave its handle on the file for a while (the remove fails
+                # with 'Device or resource busy' on Windows). Overwrite in place instead of
+                # crashing: the content is ours from here and the heartbeat keeps it fresh.
+                log("could not remove the stale lock file - overwriting it in place")
+            self.fd = os.open(self.path, os.O_CREAT | os.O_WRONLY | os.O_TRUNC)
             os.write(self.fd, str(os.getpid()).encode())
+        self._stop = threading.Event()
+        threading.Thread(target=self._beat, daemon=True).start()
         return self
 
     def __exit__(self, *a):
         try:
+            if self._stop is not None:
+                self._stop.set()
             if self.fd is not None:
                 os.close(self.fd)
             os.remove(self.path)