perf: cut startup job discovery from ~20h to ~2h on NFS

src/python/tools/archive_transcriber.py

@@ -131,6 +131,10 @@ def signal_handler(signum: int, frame: Optional[FrameType]) -> None:
 
 LOGGER = logging.getLogger("archive_transcriber")
 
+# Startup job discovery is NFS-latency-bound (stat/exists round trips), not
+# CPU-bound; a wide thread pool turns hours of sequential stats into minutes.
+STARTUP_STAT_THREADS = 64
+
 
 class SegmentLike(Protocol):
     """Protocol for transcription segment objects."""
@@ -1002,41 +1006,58 @@ def discover_video_jobs(
     skipped_count = 0
     last_log_time = time.time()
 
-    for candidates in grouped.values():
-        best_path = select_best_variant(candidates)
-        if not best_path:
-            continue
-        normalized_name = normalise_variant_name(best_path)
-        ru_vtt, en_vtt, ttml_path, smil_path = build_output_artifacts(
-            best_path, normalized_name, input_root, output_root
+    def build_job(candidates: List[Path]) -> Optional[VideoJob]:
+        """Return the job for a variant group, or None if empty/already processed."""
+        try:
+            best_path = select_best_variant(candidates)
+            if not best_path:
+                return None
+            normalized_name = normalise_variant_name(best_path)
+            ru_vtt, en_vtt, ttml_path, smil_path = build_output_artifacts(
+                best_path, normalized_name, input_root, output_root
+            )
+            if should_skip(best_path, ru_vtt, en_vtt, ttml_path, smil_path, force, ttml_enabled):
+                LOGGER.debug("Skipping already processed %s", best_path)
+                return None
+        except OSError as exc:
+            LOGGER.warning("Skipping group %s: stat failed during discovery: %s", candidates[0], exc)
+            return None
+        return VideoJob(
+            video_path=best_path,
+            normalized_name=normalized_name,
+            ru_vtt=ru_vtt,
+            en_vtt=en_vtt,
+            ttml=ttml_path,
+            smil=smil_path,
         )
 
-        if should_skip(best_path, ru_vtt, en_vtt, ttml_path, smil_path, force, ttml_enabled):
-            LOGGER.debug("Skipping already processed %s", best_path)
-            skipped_count += 1
-        else:
-            jobs.append(
-                VideoJob(
-                    video_path=best_path,
-                    normalized_name=normalized_name,
-                    ru_vtt=ru_vtt,
-                    en_vtt=en_vtt,
-                    ttml=ttml_path,
-                    smil=smil_path,
-                )
-            )
+    # The per-group checks are NFS-latency-bound; run them in parallel threads
+    # (stat/exists release the GIL) instead of one round trip at a time.
+    with ThreadPoolExecutor(max_workers=STARTUP_STAT_THREADS) as executor:
+        try:
+            for job in executor.map(build_job, grouped.values(), chunksize=16):
+                if job is not None:
+                    jobs.append(job)
+                else:
+                    skipped_count += 1
 
-        processed_groups += 1
-        current_time = time.time()
-        if current_time - last_log_time >= 5:
-            LOGGER.info(
-                "Building job list... processed %d/%d groups (%d to process, %d already done)",
-                processed_groups,
-                len(grouped),
-                len(jobs),
-                skipped_count,
-            )
-            last_log_time = current_time
+                processed_groups += 1
+                current_time = time.time()
+                if current_time - last_log_time >= 5:
+                    LOGGER.info(
+                        "Building job list... processed %d/%d groups (%d to process, %d already done)",
+                        processed_groups,
+                        len(grouped),
+                        len(jobs),
+                        skipped_count,
+                    )
+                    last_log_time = current_time
+        except BaseException:
+            # Ctrl+C (or a worker error) must not wait for the queued backlog:
+            # executor.map submits every group up front, and the context
+            # manager's shutdown would otherwise drain all of them first.
+            executor.shutdown(wait=False, cancel_futures=True)
+            raise
 
     LOGGER.info(
         "Job list complete! %d videos to process (%d already done)",
@@ -1048,19 +1069,34 @@ def discover_video_jobs(
 
 
 def select_best_variant(candidates: List[Path]) -> Optional[Path]:
-    best_path: Optional[Path] = None
-    best_score = (-1, -1)
+    """Pick the highest-resolution variant, falling back to file size on ties.
 
+    Stat calls are only made when the filename resolution is ambiguous — on a
+    143k-group archive over NFS this avoids ~700k round trips at startup.
+    """
+    if not candidates:
+        return None
+    if len(candidates) == 1:
+        return candidates[0]
+
+    by_priority: Dict[int, List[Path]] = {}
     for path in candidates:
-        resolution = extract_resolution(path.name)
-        priority = resolution or 0
+        priority = extract_resolution(path.name) or 0
+        by_priority.setdefault(priority, []).append(path)
+
+    contenders = by_priority[max(by_priority)]
+    if len(contenders) == 1:
+        return contenders[0]
+
+    best_path: Optional[Path] = None
+    best_size = -1
+    for path in contenders:
         try:
             size = path.stat().st_size
         except OSError:
             size = 0
-        score = (priority, size)
-        if score > best_score:
-            best_score = score
+        if size > best_size:
+            best_size = size
             best_path = path
 
     return best_path
@@ -1120,6 +1156,35 @@ def needs_translation(job: VideoJob, ttml_enabled: bool) -> bool:
         return True
 
 
+def phase_needs(job: VideoJob, ttml_enabled: bool) -> Tuple[bool, bool]:
+    """Combined needs_transcription/needs_translation check with one stat per file.
+
+    Semantically identical to calling the two functions separately, but stats
+    each artifact exactly once (5 round trips max instead of ~11) — this check
+    runs once per video across a 143k-video archive on NFS at every startup.
+    """
+
+    def _mtime(path: Path) -> Optional[float]:
+        try:
+            return path.stat().st_mtime
+        except OSError:
+            return None
+
+    ru_mtime = _mtime(job.ru_vtt)
+    if ru_mtime is None:
+        return True, False  # needs transcription; can't translate without it
+
+    video_mtime = _mtime(job.video_path)
+    need_transcription = video_mtime is None or ru_mtime < video_mtime
+
+    required = [_mtime(job.en_vtt), _mtime(job.smil)]
+    if ttml_enabled:
+        required.append(_mtime(job.ttml))
+    need_translation = any(m is None or m < ru_mtime for m in required)
+
+    return need_transcription, need_translation
+
+
 def extract_audio(video_path: Path, sample_rate: int) -> Path:
     tmp_file = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
     tmp_file_path = Path(tmp_file.name)
@@ -2013,21 +2078,34 @@ def run_two_phase(
     transcription_jobs: List[VideoJob] = []
     translation_jobs: List[VideoJob] = []
     last_log_time = time.time()
+
+    def check_job(j: VideoJob) -> Tuple[bool, bool]:
+        return phase_needs(j, ttml_enabled)
+
     try:
-        for i, j in enumerate(all_jobs):
-            if needs_transcription(j):
-                transcription_jobs.append(j)
-            if needs_translation(j, ttml_enabled):
-                translation_jobs.append(j)
-            if time.time() - last_log_time >= 5:
-                LOGGER.info(
-                    "  Filtering: checked %d/%d (%d transcription, %d translation)",
-                    i + 1,
-                    len(all_jobs),
-                    len(transcription_jobs),
-                    len(translation_jobs),
-                )
-                last_log_time = time.time()
+        # NFS-latency-bound; parallel threads keep job order via executor.map
+        with ThreadPoolExecutor(max_workers=STARTUP_STAT_THREADS) as executor:
+            try:
+                for i, (j, (need_ts, need_tl)) in enumerate(
+                    zip(all_jobs, executor.map(check_job, all_jobs, chunksize=16))
+                ):
+                    if need_ts:
+                        transcription_jobs.append(j)
+                    if need_tl:
+                        translation_jobs.append(j)
+                    if time.time() - last_log_time >= 5:
+                        LOGGER.info(
+                            "  Filtering: checked %d/%d (%d transcription, %d translation)",
+                            i + 1,
+                            len(all_jobs),
+                            len(transcription_jobs),
+                            len(translation_jobs),
+                        )
+                        last_log_time = time.time()
+            except BaseException:
+                # Don't drain the queued backlog on interrupt (see discover_video_jobs)
+                executor.shutdown(wait=False, cancel_futures=True)
+                raise
     except KeyboardInterrupt:
         LOGGER.warning("Interrupted during filtering")
         return 130

tests/test_archive_transcriber.py

@@ -534,3 +534,45 @@ def test_sample_rate_preserved(self):
         ar_index = cmd.index("-ar")
         assert cmd[ar_index + 1] == "22050"
         print("✓ test_sample_rate_preserved passed")
+
+
+class TestPhaseNeeds:
+    """phase_needs must match needs_transcription/needs_translation exactly."""
+
+    def _job(self, tmp: Path):
+        return VideoJob(
+            video_path=tmp / "video_1080p.mp4",
+            normalized_name="video.mp4",
+            ru_vtt=tmp / "video.ru.vtt",
+            en_vtt=tmp / "video.en.vtt",
+            ttml=tmp / "video.ttml",
+            smil=tmp / "video.smil",
+        )
+
+    def test_matches_separate_checks_across_all_states(self):
+        import itertools
+        import os
+
+        phase_needs = archive_transcriber.phase_needs
+        needs_transcription = archive_transcriber.needs_transcription
+        needs_translation = archive_transcriber.needs_translation
+
+        # Every combination of artifact presence, with fresh and stale mtimes
+        names = ["video_1080p.mp4", "video.ru.vtt", "video.en.vtt", "video.ttml", "video.smil"]
+        for present in itertools.product([False, True], repeat=len(names)):
+            for stale_ru in (False, True):
+                with tempfile.TemporaryDirectory() as tmpdir:
+                    tmp = Path(tmpdir)
+                    for name, exists in zip(names, present):
+                        if exists:
+                            (tmp / name).write_text("x")
+                            os.utime(tmp / name, (2000000, 2000000))
+                    ru = tmp / "video.ru.vtt"
+                    if stale_ru and ru.exists():
+                        os.utime(ru, (1000000, 1000000))  # older than everything
+                    job = self._job(tmp)
+                    for ttml_enabled in (False, True):
+                        expected = (needs_transcription(job), needs_translation(job, ttml_enabled))
+                        assert phase_needs(job, ttml_enabled) == expected, (
+                            f"mismatch for present={present} stale_ru={stale_ru} ttml={ttml_enabled}"
+                        )