Prioritize replacement of published legacy cache objects

.gitignore

@@ -1,3 +1,5 @@
 .env
 __pycache__/
 *.pyc
+/pretranscode
+/tests/test_core

Dockerfile

@@ -1,12 +1,21 @@
-FROM ubuntu:24.04
-
+FROM ubuntu:24.04 AS build
 ENV DEBIAN_FRONTEND=noninteractive
 RUN apt-get update \
- && apt-get install -y --no-install-recommends ca-certificates ffmpeg procps python3 util-linux \
+ && apt-get install -y --no-install-recommends \
+      build-essential pkg-config libcurl4-openssl-dev libjson-c-dev libxxhash-dev \
  && rm -rf /var/lib/apt/lists/*
+WORKDIR /src
+COPY Makefile pretranscode.c ./
+COPY tests ./tests
+RUN make check CFLAGS='-O2 -std=c17 -Wall -Wextra -Wpedantic -Werror -Wconversion -Wshadow -Wstrict-prototypes -Wmissing-prototypes -Wformat=2 -fstack-protector-strong'
 
-COPY pretranscode.py /app/pretranscode
-RUN chmod 0755 /app/pretranscode
-
-ENTRYPOINT ["nice","-n","19","ionice","-c","3","python3","/app/pretranscode"]
+FROM ubuntu:24.04
+ENV DEBIAN_FRONTEND=noninteractive
+RUN apt-get update \
+ && apt-get install -y --no-install-recommends \
+      ca-certificates ffmpeg libcurl4t64 libjson-c5 libxxhash0 procps \
+ && rm -rf /var/lib/apt/lists/*
+COPY --from=build /src/pretranscode /usr/local/bin/pretranscode
+USER 1000:1000
+ENTRYPOINT ["/usr/local/bin/pretranscode"]
 CMD ["run"]

Makefile

@@ -0,0 +1,23 @@
+CC ?= cc
+CPPFLAGS += $(shell pkg-config --cflags libcurl json-c libxxhash)
+CFLAGS ?= -O2 -g
+CFLAGS += -std=c17 -Wall -Wextra -Wpedantic -Werror -Wconversion -Wshadow \
+          -Wstrict-prototypes -Wmissing-prototypes -Wformat=2 -fstack-protector-strong
+LDLIBS += $(shell pkg-config --libs libcurl json-c libxxhash)
+
+.PHONY: all clean check
+
+all: pretranscode
+
+pretranscode: pretranscode.c
+	$(CC) $(CPPFLAGS) $(CFLAGS) $(LDFLAGS) -o $@ $< $(LDLIBS)
+
+tests/test_core: tests/test_core.c pretranscode.c
+	$(CC) $(CPPFLAGS) $(CFLAGS) -Wno-unused-function $(LDFLAGS) -o $@ $< $(LDLIBS)
+
+check: pretranscode tests/test_core
+	./pretranscode --version
+	./tests/test_core
+
+clean:
+	rm -f pretranscode tests/test_core

README.md

@@ -1,40 +1,57 @@
 # Pretranscode
 
-A small Jellyfin compatibility cache. It converts only known-incompatible video
-codecs to uniform 8-bit H.264 while preserving immutable originals.
+A small, single-writer Jellyfin compatibility cache.
 
-## Design
+Immutable source media is the only source of truth. The cache is a disposable,
+filesystem-derived projection and contains no manifests, inventory database, or
+migration state.
+
+## Compatibility output
+
+A cached rendition contains only:
+
+- one H.264 High, 8-bit `yuv420p` video stream;
+- one AAC-LC, 192 kbps stereo audio stream when the source has audio;
+- BT.709 SDR output for HDR10, HLG, and Dolby Vision sources.
+
+Embedded subtitles, attachments, chapters, and inherited metadata are omitted.
+Sidecar subtitle files are mirrored unchanged. This deliberately small profile
+prevents EAC3/Atmos, PGS, or stale Dolby Vision metadata from forcing another
+Jellyfin transcode.
+
+A source needs a compatibility rendition unless it is already H.264 8-bit with
+AAC stereo audio and no embedded subtitles.
+
+## Durable model
 
 - Originals are mounted read-only at `/source`.
-- Jellyfin reads a mirror tree at `/data/library` containing symlinks.
-- Each symlink points either to `/data/original` or a validated file under
-  `/data/cache`.
-- Completed cache files replace symlinks atomically. Partial files are never
-  exposed.
-- Cache eviction restores the original symlink before deleting cached content.
-- Jellyfin integration uses only its supported HTTP API.
+- Jellyfin reads `/data/library`, a symlink mirror under `/view`.
+- A link targets either `/data/original` or an immutable cache object.
+- Cache names encode hashes of the logical path, source generation, complete
+  encoding profile, and output bytes.
+- FFmpeg writes only a `.partial` file.
+- Publication is validation, `fsync`, an atomic output rename, then an atomic
+  symlink rename.
+- Open clients retain their old inode during publication.
+- A profile change builds a new object while the published old-profile object
+  remains available. The old object is collected only after replacement.
+- LRU updates file access time only; published media modification time and bytes
+  never change.
+- Scheduling, retries, priorities, and cooldowns exist only in process memory.
+- One controller owns the cache directory through `flock()`.
 
-There is one worker process. x264 uses available CPU cores; `nice 19`, idle I/O
-priority, and minimal Docker CPU shares make normal services take precedence.
-
-## Requirements
-
-- Linux, Docker Compose, and ext4 tools
-- An existing Jellyfin server and API key
-- A shared Docker network between Jellyfin and Pretranscode
-- Jellyfin must have read-only access to the originals and cache
+There is no cache migration command. Restarting with a changed profile is enough.
 
 ## Installation
 
-1. Copy `.env.example` to `.env` and set all host paths and the Jellyfin API key.
-2. Provision the claimed cache allocation:
+1. Copy `.env.example` to `.env` and set the host paths and Jellyfin API key.
+2. Provision the cache filesystem:
 
    ```bash
    sudo scripts/provision-cache.sh .env
    ```
 
-3. Mount the same paths in Jellyfin while keeping its existing logical library
-   path:
+3. Mount the same paths in Jellyfin:
 
    ```yaml
    volumes:
@@ -43,72 +60,55 @@ priority, and minimal Docker CPU shares make normal services take precedence.
      - ${PRETRANSCODE_CACHE_PATH}:/data/cache:ro
    ```
 
-4. Start Pretranscode:
+4. Start the controller:
 
    ```bash
    docker compose --env-file .env up -d --build
    ```
 
 For Coolify, import `compose.yaml` as a Custom Docker Compose resource and enter
-the `.env.example` values through Coolify's environment-variable UI.
+`.env.example` values through the environment-variable UI.
 
-## Codec policy
+## Operation
 
-Cached video is H.264 High Profile, 8-bit `yuv420p`, CRF 18, capped at 15 Mbps.
-Resolution and frame rate are preserved. Existing audio, subtitles, chapters,
-and Matroska attachments are copied. Ten-bit SDR is converted with
-error-diffusion dithering; sources carrying explicit HDR10 or HLG transfer
-metadata are tone-mapped to BT.709 SDR.
+The controller has one worker. FFmpeg and file hashing run with `SCHED_IDLE` and
+idle I/O priority so foreground services take precedence.
 
-The fixed source blacklist is:
+Playback locking rules:
 
-- HEVC/H.265, including Dolby Vision
-- H.264 Hi10/10-bit
-- MPEG-2
-- VC-1
+- an active movie locks that movie;
+- an active episode locks its series;
+- an encode is cancelled if its item becomes locked;
+- a series remains locked for `SERIES_COOLDOWN_SECONDS` after playback;
+- recently watched content receives in-memory queue priority.
 
-Client error reports never modify this list. Encoding settings form part of
-each cache manifest, so a global quality-policy change automatically invalidates
-and replaces older renditions without touching originals.
+Commands:
 
-## Cache resizing
+```bash
+pretranscode run
+pretranscode sync
+pretranscode active
+pretranscode trim BYTES
+pretranscode purge
+pretranscode --version
+```
 
-`CACHE_MAX_BYTES` is the worker's configurable LRU limit. After changing it and
-restarting the container:
+`purge` and `trim` refuse to operate during active playback.
 
-- A smaller value evicts the oldest unlocked entries until usage fits.
-- A larger value immediately makes the additional capacity available for
-  background population, up to the size of the mounted cache filesystem.
+## Failure behaviour
 
-The host filesystem is deliberately separate from application logic. Enlarge or
-shrink its preallocated image during infrastructure maintenance, then set the
-matching `CACHE_MAX_BYTES` value in Coolify.
+Partial outputs are never published. Startup removes stale partials and
+unreferenced objects. A missing cache mount marker, second controller, failed
+probe, failed validation, failed `fsync`, or unavailable Jellyfin session API
+causes work to stop safely rather than mutate published media.
 
-## Playback locking
+## Tests
 
-- An active movie locks that movie.
-- An active episode locks its complete series.
-- Encoding is cancelled and its partial deleted if the item becomes locked.
-- A series remains locked for `SERIES_COOLDOWN_SECONDS` after playback.
-- Recently stopped content receives queue priority.
+The image build treats compiler warnings as errors and runs the C core tests.
+The integration test creates incompatible media and verifies canonical streams,
+atomic publication, refresh retry, and stateless profile replacement:
 
-## Failure behavior
-
-Graceful stops delete partial output. After a crash, startup removes stale
-partials and orphan files. Atomic symlinks always point to either an original or
-a completed cache file. If the cache mount marker is absent, the worker refuses
-to start.
-
-## Migration
-
-Copy or recreate these resources on the destination:
-
-- Repository and `.env`
-- State directory
-- Symlink-view directory
-- Cache image/mount, if retaining completed transcodes
-
-The cache is optional and can be rebuilt. The source fingerprint in every
-manifest prevents stale cache files from replacing changed originals. No
-Coolify UUID, Docker volume UUID, container name, or Jellyfin database path is
-stored by the application.
+```bash
+podman build -t pretranscode-test .
+IMAGE=pretranscode-test tests/integration.sh
+```

compose.yaml

@@ -39,7 +39,7 @@ services:
       - no-new-privileges:true
     stop_grace_period: 30s
     healthcheck:
-      test: ["CMD-SHELL", "pgrep -f '/app/pretranscode run' >/dev/null"]
+      test: ["CMD-SHELL", "pgrep -x pretranscode >/dev/null"]
       interval: 30s
       timeout: 5s
       retries: 3

pretranscode.py

@@ -1,801 +0,0 @@
-#!/usr/bin/env python3
-"""Immutable Jellyfin H.264 cache backed by an atomic symlink view."""
-
-import hashlib
-import json
-import os
-import signal
-import subprocess
-import sys
-import time
-import urllib.error
-import urllib.parse
-import urllib.request
-from pathlib import Path, PurePosixPath
-
-VIDEO_EXTENSIONS = {".mkv", ".mp4", ".m4v", ".avi", ".mov", ".ts", ".m2ts", ".webm"}
-BLACKLISTED_CODECS = {"hevc", "h265", "mpeg2video", "vc1", "vc-1"}
-
-
-def env_int(name, default):
-    return int(os.environ.get(name, default))
-
-
-SOURCE_ROOT = Path(os.environ.get("SOURCE_ROOT", "/source"))
-VIEW_ROOT = Path(os.environ.get("VIEW_ROOT", "/view"))
-CACHE_ROOT = Path(os.environ.get("CACHE_ROOT", "/cache"))
-STATE_ROOT = Path(os.environ.get("STATE_ROOT", "/state"))
-STATE_FILE = STATE_ROOT / "state.json"
-JELLYFIN_URL = os.environ.get("JELLYFIN_URL", "http://jellyfin:8096").rstrip("/")
-JELLYFIN_LIBRARY_ROOT = os.environ.get("JELLYFIN_LIBRARY_ROOT", "/data/library").rstrip("/")
-JELLYFIN_ORIGINAL_ROOT = os.environ.get("JELLYFIN_ORIGINAL_ROOT", "/data/original").rstrip("/")
-JELLYFIN_CACHE_ROOT = os.environ.get("JELLYFIN_CACHE_ROOT", "/data/cache").rstrip("/")
-JELLYFIN_API_KEY = os.environ.get("JELLYFIN_API_KEY", "")
-CACHE_MAX_BYTES = env_int("CACHE_MAX_BYTES", 1_000_000_000_000)
-SERIES_COOLDOWN_SECONDS = env_int("SERIES_COOLDOWN_SECONDS", 3600)
-POLL_SECONDS = env_int("POLL_SECONDS", 5)
-RESCAN_SECONDS = env_int("RESCAN_SECONDS", 300)
-H264_CRF = os.environ.get("H264_CRF", "18")
-H264_PRESET = os.environ.get("H264_PRESET", "medium")
-H264_MAXRATE = os.environ.get("H264_MAXRATE", "15M")
-H264_BUFSIZE = os.environ.get("H264_BUFSIZE", "30M")
-ENCODE_PROFILE = f"h264-high-8bit-crf{H264_CRF}-{H264_PRESET}-{H264_MAXRATE}-quality-v3"
-FFMPEG = os.environ.get("FFMPEG", "/usr/bin/ffmpeg")
-FFPROBE = os.environ.get("FFPROBE", "/usr/bin/ffprobe")
-
-
-def log(message):
-    print(time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), message, flush=True)
-
-
-def atomic_json(path, value):
-    path.parent.mkdir(parents=True, exist_ok=True)
-    temporary = path.with_name(path.name + ".tmp")
-    with temporary.open("w") as handle:
-        json.dump(value, handle, sort_keys=True, separators=(",", ":"))
-        handle.flush()
-        os.fsync(handle.fileno())
-    os.replace(temporary, path)
-
-
-def load_json(path, default):
-    try:
-        with path.open() as handle:
-            return json.load(handle)
-    except (FileNotFoundError, json.JSONDecodeError, OSError):
-        return default
-
-
-def fingerprint(path):
-    stat = path.stat()
-    return f"{stat.st_size}:{stat.st_mtime_ns}"
-
-
-def relative_source(path):
-    value = str(path)
-    for root in (JELLYFIN_LIBRARY_ROOT, str(SOURCE_ROOT)):
-        prefix = root.rstrip("/") + "/"
-        if value.startswith(prefix):
-            return value[len(prefix):]
-    return None
-
-
-def item_key(relative):
-    parts = PurePosixPath(relative).parts
-    if len(parts) >= 2 and parts[0].lower() == "tv":
-        return "series:" + parts[1]
-    return "item:" + relative
-
-
-def cache_basename(relative):
-    digest = hashlib.sha256(relative.encode()).hexdigest()
-    suffix = ".mp4" if Path(relative).suffix.lower() in {".mp4", ".m4v"} else ".mkv"
-    return digest + suffix
-
-
-def source_target(relative):
-    return JELLYFIN_ORIGINAL_ROOT + "/" + PurePosixPath(relative).as_posix()
-
-
-def cache_target(relative):
-    return JELLYFIN_CACHE_ROOT + "/files/" + cache_basename(relative)
-
-
-class Manager:
-    def __init__(self):
-        STATE_ROOT.mkdir(parents=True, exist_ok=True)
-        if not (CACHE_ROOT / ".pretranscode-cache").is_file():
-            raise RuntimeError(f"cache filesystem is not mounted at {CACHE_ROOT}")
-        files_root = CACHE_ROOT / "files"
-        files_root.mkdir(parents=True, exist_ok=True)
-        for partial in files_root.glob("*.partial"):
-            partial.unlink(missing_ok=True)
-        for output in list(files_root.glob("*.mkv")) + list(files_root.glob("*.mp4")):
-            if not output.with_suffix(output.suffix + ".json").is_file():
-                output.unlink(missing_ok=True)
-        for manifest in files_root.glob("*.json"):
-            if not manifest.with_suffix("").is_file():
-                manifest.unlink(missing_ok=True)
-        VIEW_ROOT.mkdir(parents=True, exist_ok=True)
-        self.state = load_json(STATE_FILE, {})
-        self.state.setdefault("inventory", {})
-        self.state.setdefault("priorities", {})
-        self.state.setdefault("cooldowns", {})
-        self.state.setdefault("retry_after", {})
-        self.active_items = set()
-        self.active_keys = set()
-        self.previous_active_items = set()
-        self.previous_active_keys = set()
-        self.sessions_healthy = False
-        self.running = True
-        self.process = None
-        self.last_scan = 0
-        self.last_activity_save = 0
-        signal.signal(signal.SIGTERM, self.stop)
-        signal.signal(signal.SIGINT, self.stop)
-
-    def stop(self, *_):
-        self.running = False
-        self.cancel_process("service stopping")
-
-    def save(self):
-        atomic_json(STATE_FILE, self.state)
-
-    def api(self, path, method="GET"):
-        request = urllib.request.Request(
-            JELLYFIN_URL + path,
-            method=method,
-            headers={"X-Emby-Token": JELLYFIN_API_KEY, "Content-Length": "0"},
-        )
-        with urllib.request.urlopen(request, timeout=10) as response:
-            data = response.read()
-            return json.loads(data) if data else None
-
-    def read_sessions(self):
-        try:
-            sessions = self.api("/Sessions")
-        except Exception as error:
-            if self.sessions_healthy:
-                log(f"Jellyfin sessions unavailable; stopping cache work: {error}")
-            self.sessions_healthy = False
-            self.active_items = set()
-            self.active_keys = set()
-            return False
-
-        now = time.time()
-        current_items = set()
-        current_keys = set()
-        for session in sessions or []:
-            item = session.get("NowPlayingItem") or {}
-            path = item.get("Path")
-            relative = relative_source(path) if path else None
-            if not relative and item.get("Id"):
-                try:
-                    detail = self.api("/Items/" + item["Id"] + "?Fields=Path") or {}
-                    relative = relative_source(detail.get("Path", ""))
-                except Exception:
-                    relative = None
-            if not relative:
-                continue
-            current_items.add(relative)
-            current_keys.add(item_key(relative))
-            self.touch_cache(relative)
-
-        ended_items = self.previous_active_items - current_items
-        ended_keys = self.previous_active_keys - current_keys
-        changed = current_items != self.previous_active_items or current_keys != self.previous_active_keys
-        for relative in current_items | ended_items:
-            self.state["priorities"][relative] = now
-        for key in current_keys | ended_keys:
-            if key.startswith("series:"):
-                self.state["cooldowns"][key] = now + SERIES_COOLDOWN_SECONDS
-        if changed or (current_items and now - self.last_activity_save >= 60):
-            self.save()
-            self.last_activity_save = now
-
-        self.previous_active_items = current_items
-        self.previous_active_keys = current_keys
-        self.active_items = current_items
-        self.active_keys = current_keys
-        self.sessions_healthy = True
-        return True
-
-    def locked_keys(self):
-        now = time.time()
-        expired = [key for key, until in self.state["cooldowns"].items() if float(until) <= now]
-        if expired:
-            for key in expired:
-                self.state["cooldowns"].pop(key, None)
-            self.save()
-        return self.active_keys | {
-            key for key, until in self.state["cooldowns"].items() if float(until) > now
-        }
-
-    def is_locked(self, relative):
-        return item_key(relative) in self.locked_keys()
-
-    def cache_paths(self, relative):
-        output = CACHE_ROOT / "files" / cache_basename(relative)
-        manifest = output.with_suffix(output.suffix + ".json")
-        return output, manifest
-
-    def cache_valid(self, relative):
-        source = SOURCE_ROOT / relative
-        output, manifest_path = self.cache_paths(relative)
-        if not source.is_file() or not output.is_file():
-            return False
-        manifest = load_json(manifest_path, {})
-        try:
-            return (
-                manifest.get("source_fingerprint") == fingerprint(source)
-                and manifest.get("encode_profile") == ENCODE_PROFILE
-            )
-        except OSError:
-            return False
-
-    def touch_cache(self, relative):
-        if not self.cache_valid(relative):
-            return
-        output, manifest = self.cache_paths(relative)
-        now = time.time()
-        for path in (output, manifest):
-            try:
-                os.utime(path, (now, now), follow_symlinks=False)
-            except OSError:
-                pass
-
-    def ensure_link(self, relative, target, locked=False):
-        link = VIEW_ROOT / relative
-        link.parent.mkdir(parents=True, exist_ok=True)
-        try:
-            if link.is_symlink() and os.readlink(link) == target:
-                return False
-        except OSError:
-            pass
-        if locked and (link.exists() or link.is_symlink()):
-            return False
-        temporary = link.with_name(link.name + ".pretranscode-tmp")
-        try:
-            temporary.unlink(missing_ok=True)
-            os.symlink(target, temporary)
-            os.replace(temporary, link)
-            return True
-        finally:
-            temporary.unlink(missing_ok=True)
-
-    def swap_to_original(self, relative):
-        return self.ensure_link(relative, source_target(relative), locked=self.is_locked(relative))
-
-    def swap_to_cache(self, relative):
-        if self.is_locked(relative) or not self.cache_valid(relative):
-            return False
-        output, manifest_path = self.cache_paths(relative)
-        manifest = load_json(manifest_path, {})
-        changed = self.ensure_link(relative, cache_target(relative), locked=False)
-        if changed or manifest.get("refresh_version") != 2:
-            self.refresh_item(relative)
-            manifest["refresh_version"] = 2
-            atomic_json(manifest_path, manifest)
-        return changed
-
-    def refresh_item(self, relative):
-        item_id = (self.state["inventory"].get(relative) or {}).get("item_id")
-        if not item_id:
-            return
-        query = urllib.parse.urlencode({
-            "metadataRefreshMode": "FullRefresh",
-            "imageRefreshMode": "None",
-            "replaceAllMetadata": "false",
-            "replaceAllImages": "false",
-        })
-        try:
-            self.api(f"/Items/{item_id}/Refresh?{query}", method="POST")
-        except Exception as error:
-            log(f"Unable to refresh Jellyfin item {relative}: {error}")
-
-    def probe(self, path):
-        command = [
-            FFPROBE, "-v", "error", "-select_streams", "v:0",
-            "-show_entries", "stream=codec_name,bits_per_raw_sample,pix_fmt,profile,color_transfer,color_primaries,color_space",
-            "-show_entries", "format=duration", "-of", "json", str(path),
-        ]
-        result = subprocess.run(command, capture_output=True, text=True, timeout=120)
-        if result.returncode:
-            raise RuntimeError(result.stderr.strip() or "ffprobe failed")
-        data = json.loads(result.stdout)
-        stream = (data.get("streams") or [{}])[0]
-        raw_depth = stream.get("bits_per_raw_sample")
-        try:
-            depth = int(raw_depth or 0)
-        except ValueError:
-            depth = 0
-        pixel = (stream.get("pix_fmt") or "").lower()
-        profile = (stream.get("profile") or "").lower()
-        if not depth:
-            if "12" in pixel or "12" in profile:
-                depth = 12
-            elif "10" in pixel or "10" in profile:
-                depth = 10
-            else:
-                depth = 8
-        return {
-            "codec": (stream.get("codec_name") or "").lower(),
-            "bit_depth": depth,
-            "profile": profile,
-            "color_transfer": (stream.get("color_transfer") or "").lower(),
-            "color_primaries": (stream.get("color_primaries") or "").lower(),
-            "color_space": (stream.get("color_space") or "").lower(),
-            "duration": float((data.get("format") or {}).get("duration") or 0),
-        }
-
-    def jellyfin_items(self):
-        values = {}
-        start = 0
-        limit = 1000
-        try:
-            while True:
-                query = urllib.parse.urlencode({
-                    "Recursive": "true",
-                    "IncludeItemTypes": "Movie,Episode",
-                    "Fields": "Path,MediaStreams",
-                    "StartIndex": start,
-                    "Limit": limit,
-                })
-                page = self.api("/Items?" + query) or {}
-                items = page.get("Items") or []
-                for item in items:
-                    relative = relative_source(item.get("Path", ""))
-                    if not relative:
-                        continue
-                    video = next(
-                        (stream for stream in item.get("MediaStreams", [])
-                         if (stream.get("Type") or "").lower() == "video"),
-                        {},
-                    )
-                    values[relative] = {
-                        "item_id": item.get("Id"),
-                        "codec": (video.get("Codec") or "").lower(),
-                        "bit_depth": int(video.get("BitDepth") or 8),
-                        "profile": (video.get("Profile") or "").lower(),
-                        "color_transfer": (video.get("ColorTransfer") or "").lower(),
-                        "color_primaries": (video.get("ColorPrimaries") or "").lower(),
-                        "color_space": (video.get("ColorSpace") or "").lower(),
-                        "duration": float(item.get("RunTimeTicks") or 0) / 10_000_000,
-                    }
-                start += len(items)
-                if not items or start >= int(page.get("TotalRecordCount") or start):
-                    break
-        except Exception as error:
-            log(f"Unable to read Jellyfin inventory: {error}")
-        return values
-
-    def scan_inventory(self):
-        catalog = self.jellyfin_items()
-        inventory = self.state["inventory"]
-        seen = set()
-        added = False
-        for root, directories, files in os.walk(SOURCE_ROOT):
-            directories.sort()
-            files.sort()
-            for filename in files:
-                if Path(filename).suffix.lower() not in VIDEO_EXTENSIONS:
-                    continue
-                source = Path(root) / filename
-                relative = source.relative_to(SOURCE_ROOT).as_posix()
-                seen.add(relative)
-                try:
-                    current_fingerprint = fingerprint(source)
-                except OSError:
-                    continue
-                old = inventory.get(relative) or {}
-                if old.get("fingerprint") == current_fingerprint:
-                    if relative in catalog:
-                        for key in ("item_id", "color_transfer", "color_primaries", "color_space"):
-                            if catalog[relative].get(key):
-                                old[key] = catalog[relative][key]
-                    continue
-                details = catalog.get(relative)
-                view = VIEW_ROOT / relative
-                view_uses_cache = view.is_symlink() and os.readlink(view) == cache_target(relative)
-                if details and not old and not view_uses_cache:
-                    details = dict(details)
-                else:
-                    try:
-                        details = self.probe(source)
-                    except Exception as error:
-                        log(f"Probe failed for {relative}: {error}")
-                        self.state["retry_after"][relative] = time.time() + 21600
-                        continue
-                    if relative in catalog:
-                        details["item_id"] = catalog[relative].get("item_id")
-                details["fingerprint"] = current_fingerprint
-                details["size"] = source.stat().st_size
-                details["mtime"] = source.stat().st_mtime
-                inventory[relative] = details
-                self.state["retry_after"].pop(relative, None)
-                added = True
-
-        removed = set(inventory) - seen
-        for relative in removed:
-            if self.is_locked(relative):
-                continue
-            self.remove_cache(relative)
-            inventory.pop(relative, None)
-            self.state["priorities"].pop(relative, None)
-            self.state["retry_after"].pop(relative, None)
-            added = True
-        if added:
-            self.save()
-        return added
-
-    def sync_view(self):
-        changed = False
-        source_files = set()
-        for root, directories, files in os.walk(SOURCE_ROOT):
-            relative_root = Path(root).relative_to(SOURCE_ROOT)
-            (VIEW_ROOT / relative_root).mkdir(parents=True, exist_ok=True)
-            directories.sort()
-            files.sort()
-            for filename in files:
-                relative = (relative_root / filename).as_posix()
-                source_files.add(relative)
-                is_video = Path(filename).suffix.lower() in VIDEO_EXTENSIONS
-                valid_cache = is_video and self.cache_valid(relative)
-                if valid_cache and not self.is_locked(relative):
-                    changed |= self.swap_to_cache(relative)
-                else:
-                    changed |= self.ensure_link(
-                        relative, source_target(relative), locked=is_video and self.is_locked(relative)
-                    )
-
-        for root, directories, files in os.walk(VIEW_ROOT, topdown=False):
-            for filename in files:
-                path = Path(root) / filename
-                relative = path.relative_to(VIEW_ROOT).as_posix()
-                if relative.endswith(".pretranscode-tmp"):
-                    path.unlink(missing_ok=True)
-                elif relative not in source_files and path.is_symlink() and not self.is_locked(relative):
-                    path.unlink(missing_ok=True)
-                    changed = True
-            for directory in directories:
-                path = Path(root) / directory
-                try:
-                    path.rmdir()
-                except OSError:
-                    pass
-        return changed
-
-    def incompatible(self, info):
-        codec = (info.get("codec") or "").lower()
-        depth = int(info.get("bit_depth") or 8)
-        profile = (info.get("profile") or "").lower()
-        return codec in BLACKLISTED_CODECS or (codec == "h264" and (depth > 8 or "10" in profile))
-
-    def is_hdr(self, info):
-        return (info.get("color_transfer") or "").lower() in {"smpte2084", "arib-std-b67"}
-
-    def candidate_list(self):
-        now = time.time()
-        values = []
-        for relative, info in self.state["inventory"].items():
-            if not self.incompatible(info) or self.cache_valid(relative) or self.is_locked(relative):
-                continue
-            if float(self.state["retry_after"].get(relative, 0)) > now:
-                continue
-            priority = self.state["priorities"].get(relative)
-            values.append((0 if priority is not None else 1, -float(priority or info.get("mtime", 0)), relative))
-        values.sort()
-        return values
-
-    def cache_entries(self):
-        entries = []
-        for manifest_path in (CACHE_ROOT / "files").glob("*.json"):
-            manifest = load_json(manifest_path, {})
-            relative = manifest.get("relative")
-            if not relative:
-                continue
-            output = manifest_path.with_suffix("")
-            if not output.is_file():
-                manifest_path.unlink(missing_ok=True)
-                continue
-            entries.append((output.stat().st_mtime, relative, output, manifest_path))
-        return sorted(entries)
-
-    def remove_cache(self, relative, force=False):
-        if not force and self.is_locked(relative):
-            return False
-        output, manifest = self.cache_paths(relative)
-        link = VIEW_ROOT / relative
-        if link.is_symlink():
-            try:
-                if os.readlink(link) == cache_target(relative):
-                    self.ensure_link(relative, source_target(relative), locked=False)
-                    self.refresh_item(relative)
-            except OSError:
-                pass
-        output.unlink(missing_ok=True)
-        manifest.unlink(missing_ok=True)
-        return True
-
-    def cache_usage(self):
-        return sum(
-            path.stat().st_size
-            for path in (CACHE_ROOT / "files").iterdir()
-            if path.is_file()
-        )
-
-    def cache_available(self):
-        stat = os.statvfs(CACHE_ROOT)
-        filesystem_free = stat.f_bavail * stat.f_frsize
-        configured_free = max(0, CACHE_MAX_BYTES - self.cache_usage())
-        return min(filesystem_free, configured_free)
-
-    def purge_incompatible_cache_profiles(self):
-        for _, relative, _, _ in self.cache_entries():
-            if self.cache_valid(relative) or self.is_locked(relative):
-                continue
-            log(f"Dropping obsolete cache rendition for {relative}")
-            self.remove_cache(relative)
-
-    def enforce_cache_limit(self):
-        usage = self.cache_usage()
-        for _, relative, output, manifest in self.cache_entries():
-            if usage <= CACHE_MAX_BYTES:
-                break
-            if self.is_locked(relative):
-                continue
-            size = output.stat().st_size + manifest.stat().st_size
-            log(f"Evicting {relative} after cache resize")
-            if self.remove_cache(relative):
-                usage -= size
-
-    def ensure_space(self, required, may_evict):
-        if self.cache_available() >= required:
-            return True
-        if not may_evict:
-            return False
-        for _, relative, _, _ in self.cache_entries():
-            if self.is_locked(relative):
-                continue
-            log(f"Evicting {relative}")
-            self.remove_cache(relative)
-            if self.cache_available() >= required:
-                return True
-        return False
-
-    def cancel_process(self, reason):
-        if not self.process or self.process.poll() is not None:
-            return
-        log(f"Stopping encode: {reason}")
-        try:
-            os.killpg(self.process.pid, signal.SIGTERM)
-            self.process.wait(timeout=10)
-        except (ProcessLookupError, subprocess.TimeoutExpired):
-            try:
-                os.killpg(self.process.pid, signal.SIGKILL)
-            except ProcessLookupError:
-                pass
-        self.process = None
-
-    def ffmpeg_command(self, source, partial, info):
-        hdr = self.is_hdr(info)
-        if hdr:
-            transfer = (info.get("color_transfer") or "").lower()
-            if transfer == "arib-std-b67":
-                input_colors = "tin=arib-std-b67:pin=bt2020:min=bt2020nc:"
-            else:
-                input_colors = ""
-            video_filter = (
-                f"zscale={input_colors}t=linear:npl=100,format=gbrpf32le,"
-                "zscale=p=bt709,tonemap=tonemap=hable:desat=0,"
-                "zscale=t=bt709:m=bt709:r=tv:dither=error_diffusion,format=yuv420p"
-            )
-        else:
-            video_filter = "zscale=dither=error_diffusion,format=yuv420p"
-
-        common = [
-            FFMPEG, "-nostdin", "-hide_banner", "-loglevel", "warning", "-i", str(source),
-            "-map", "0:v:0", "-map", "0:a?", "-map", "0:s?", "-map_metadata", "0", "-map_chapters", "0",
-            "-vf", video_filter,
-            "-c:v", "libx264", "-preset", H264_PRESET, "-crf", H264_CRF,
-            "-maxrate", H264_MAXRATE, "-bufsize", H264_BUFSIZE,
-            "-pix_fmt", "yuv420p", "-profile:v", "high",
-            "-c:a", "copy", "-c:s", "copy",
-        ]
-        if hdr:
-            common += [
-                "-color_primaries", "bt709", "-color_trc", "bt709",
-                "-colorspace", "bt709", "-color_range", "tv",
-            ]
-        if partial.name.endswith(".mp4.partial"):
-            common += ["-movflags", "+faststart", "-f", "mp4"]
-        else:
-            common += ["-map", "0:t?", "-c:t", "copy", "-f", "matroska"]
-        return common + [str(partial)]
-
-    def validate_output(self, source, output, expected_duration):
-        details = self.probe(output)
-        if details.get("codec") != "h264" or int(details.get("bit_depth") or 0) > 8:
-            raise RuntimeError("output is not 8-bit H.264")
-        source_duration = float(expected_duration or 0)
-        output_duration = float(details.get("duration") or 0)
-        if source_duration and abs(source_duration - output_duration) > max(3, source_duration * 0.01):
-            raise RuntimeError(f"duration mismatch: source={source_duration}, output={output_duration}")
-        if output.stat().st_size < 1_000_000:
-            raise RuntimeError("output is unexpectedly small")
-
-    def encode(self, relative, priority):
-        source = SOURCE_ROOT / relative
-        if not source.is_file() or self.is_locked(relative):
-            return "skipped"
-        output, manifest_path = self.cache_paths(relative)
-        partial = output.with_name(output.name + ".partial")
-        partial.unlink(missing_ok=True)
-        info = self.state["inventory"].get(relative) or {}
-        estimate = min(max(int(source.stat().st_size * 2.2), 1_000_000_000), 100_000_000_000)
-        if not self.ensure_space(estimate, may_evict=priority):
-            return "full"
-
-        log(f"Encoding {relative}")
-        command = self.ffmpeg_command(source, partial, info)
-        log_path = STATE_ROOT / "ffmpeg.log"
-        with log_path.open("ab", buffering=0) as ffmpeg_log:
-            self.process = subprocess.Popen(
-                command, stdout=ffmpeg_log, stderr=ffmpeg_log, start_new_session=True
-            )
-            last_mirror = 0
-            while self.running and self.process.poll() is None:
-                time.sleep(POLL_SECONDS)
-                if not self.read_sessions() or self.is_locked(relative):
-                    self.cancel_process("item or series is being watched")
-                    partial.unlink(missing_ok=True)
-                    return "cancelled"
-                if time.time() - last_mirror >= 60:
-                    if self.sync_view():
-                        try:
-                            self.api("/Library/Refresh", method="POST")
-                        except Exception as error:
-                            log(f"Unable to request library refresh: {error}")
-                    last_mirror = time.time()
-            returncode = self.process.wait() if self.process else -1
-            self.process = None
-
-        if not self.running:
-            partial.unlink(missing_ok=True)
-            return "cancelled"
-        if returncode != 0:
-            partial.unlink(missing_ok=True)
-            raise RuntimeError(f"ffmpeg exited with status {returncode}")
-        if self.is_locked(relative):
-            partial.unlink(missing_ok=True)
-            return "cancelled"
-
-        self.validate_output(source, partial, info.get("duration"))
-        os.replace(partial, output)
-        atomic_json(manifest_path, {
-            "relative": relative,
-            "source_fingerprint": fingerprint(source),
-            "created_at": time.time(),
-            "encode_profile": ENCODE_PROFILE,
-        })
-        if self.is_locked(relative):
-            output.unlink(missing_ok=True)
-            manifest_path.unlink(missing_ok=True)
-            return "cancelled"
-        self.swap_to_cache(relative)
-        self.state["priorities"].pop(relative, None)
-        self.state["retry_after"].pop(relative, None)
-        self.save()
-        log(f"Completed {relative}")
-        return "completed"
-
-    def run(self):
-        if not JELLYFIN_API_KEY:
-            raise RuntimeError("JELLYFIN_API_KEY is required")
-        self.read_sessions()
-        self.purge_incompatible_cache_profiles()
-        self.enforce_cache_limit()
-        self.scan_inventory()
-        self.sync_view()
-        self.save()
-        while self.running:
-            if not self.read_sessions():
-                time.sleep(POLL_SECONDS)
-                continue
-            self.purge_incompatible_cache_profiles()
-            self.enforce_cache_limit()
-            now = time.time()
-            if now - self.last_scan >= RESCAN_SECONDS:
-                changed = self.scan_inventory()
-                changed |= self.sync_view()
-                if changed:
-                    try:
-                        self.api("/Library/Refresh", method="POST")
-                    except Exception as error:
-                        log(f"Unable to request library refresh: {error}")
-                self.last_scan = now
-
-            candidates = self.candidate_list()
-            if not candidates:
-                time.sleep(POLL_SECONDS)
-                continue
-            selected = None
-            for rank, _, relative in candidates:
-                source = SOURCE_ROOT / relative
-                estimate = min(max(int(source.stat().st_size * 2.2), 1_000_000_000), 100_000_000_000)
-                if rank == 0 or self.cache_available() >= estimate:
-                    selected = (rank == 0, relative)
-                    break
-            if not selected:
-                time.sleep(60)
-                continue
-            priority, relative = selected
-            try:
-                result = self.encode(relative, priority)
-                if result == "full":
-                    time.sleep(60)
-            except Exception as error:
-                log(f"Encode failed for {relative}: {error}")
-                self.state["retry_after"][relative] = time.time() + 21600
-                self.save()
-                output, manifest = self.cache_paths(relative)
-                output.with_name(output.name + ".partial").unlink(missing_ok=True)
-                time.sleep(POLL_SECONDS)
-
-    def sync_once(self):
-        self.scan_inventory()
-        self.sync_view()
-        self.save()
-
-    def active(self):
-        if not self.read_sessions():
-            return True
-        return bool(self.active_items)
-
-    def purge(self, force=False):
-        if not force and self.active():
-            raise RuntimeError("cache purge refused while media is playing")
-        for _, relative, _, _ in self.cache_entries():
-            log(f"Purging {relative}")
-            self.remove_cache(relative, force=force)
-        self.sync_view()
-        self.save()
-
-    def trim(self, target_bytes):
-        if self.active():
-            raise RuntimeError("cache resize refused while media is playing")
-        target = max(0, int(target_bytes) - 5_000_000_000)
-        entries = self.cache_entries()
-        used = sum(output.stat().st_size + manifest.stat().st_size for _, _, output, manifest in entries)
-        for _, relative, output, manifest in entries:
-            if used <= target:
-                break
-            size = output.stat().st_size + manifest.stat().st_size
-            if self.remove_cache(relative):
-                log(f"Trimmed {relative}")
-                used -= size
-        if used > target:
-            raise RuntimeError("unable to trim cache below target because content is locked")
-        self.sync_view()
-        self.save()
-
-
-def main():
-    manager = Manager()
-    command = sys.argv[1] if len(sys.argv) > 1 else "run"
-    if command == "run":
-        manager.run()
-    elif command == "sync":
-        manager.sync_once()
-    elif command == "active":
-        raise SystemExit(0 if not manager.active() else 1)
-    elif command == "trim" and len(sys.argv) == 3:
-        manager.trim(int(sys.argv[2]))
-    elif command == "purge":
-        manager.purge(force="--force" in sys.argv[2:])
-    else:
-        raise SystemExit(f"unknown command: {command}")
-
-
-if __name__ == "__main__":
-    main()

tests/fake_jellyfin.py

@@ -0,0 +1,88 @@
+#!/usr/bin/env python3
+import json
+import os
+import sys
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from urllib.parse import urlparse
+
+view = os.environ["TEST_VIEW"]
+state_path = os.environ["TEST_STATE"]
+codec = "hevc"
+depth = 10
+refresh_attempts = 0
+
+
+def record_state():
+    with open(state_path, "w") as handle:
+        json.dump({"codec": codec, "depth": depth, "attempts": refresh_attempts}, handle)
+
+
+def refresh_media():
+    global codec, depth
+    link = os.path.join(view, "movies/Test/Test.mkv")
+    try:
+        target = os.readlink(link)
+    except OSError:
+        target = ""
+    codec, depth = ("h264", 8) if target.startswith("/data/cache/") else ("hevc", 10)
+    record_state()
+
+
+class Handler(BaseHTTPRequestHandler):
+    def send_json(self, value, status=200):
+        body = json.dumps(value, separators=(",", ":")).encode()
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def do_GET(self):
+        path = urlparse(self.path).path
+        if path == "/Sessions":
+            self.send_json([])
+        elif path == "/Items/test-item":
+            self.send_json({"Id": "test-item", "Path": "/data/library/movies/Test/Test.mkv"})
+        elif path == "/Items":
+            self.send_json({
+                "Items": [{
+                    "Id": "test-item",
+                    "Path": "/data/library/movies/Test/Test.mkv",
+                    "RunTimeTicks": 6 * 10_000_000,
+                    "MediaStreams": [{
+                        "Type": "Video", "Codec": codec, "BitDepth": depth,
+                        "Profile": "Main 10" if depth == 10 else "High",
+                        "Width": 1280, "Height": 720,
+                    }],
+                }],
+                "TotalRecordCount": 1,
+            })
+        else:
+            self.send_json({"error": "not found"}, 404)
+
+    def do_POST(self):
+        global refresh_attempts
+        path = urlparse(self.path).path
+        if path.startswith("/Items/test-item/Refresh"):
+            refresh_attempts += 1
+            if refresh_attempts == 1:
+                record_state()
+                self.send_response(503)
+                self.end_headers()
+                return
+            refresh_media()
+            self.send_response(204)
+            self.end_headers()
+        elif path == "/Library/Refresh":
+            self.send_response(204)
+            self.end_headers()
+        else:
+            self.send_json({"error": "not found"}, 404)
+
+    def log_message(self, fmt, *args):
+        pass
+
+
+record_state()
+server = ThreadingHTTPServer(("0.0.0.0", int(sys.argv[1])), Handler)
+server.serve_forever()

tests/integration.sh

@@ -1,40 +1,139 @@
 #!/bin/bash
 set -euo pipefail
-: "${JELLYFIN_API_KEY:?}"
-IMAGE=${IMAGE:-pretranscode:test}
-NETWORK=${MEDIA_NETWORK:-media-internal}
-JELLYFIN_URL=${JELLYFIN_URL:-http://jellyfin:8096}
+IMAGE=${IMAGE:-localhost/pretranscode-c-review}
 ROOT=$(mktemp -d)
-trap 'rm -rf "$ROOT"' EXIT
-mkdir -p "$ROOT"/{source/movies/Test,view,cache,state}
+PORT=$(python3 - <<'PY'
+import socket
+s=socket.socket(); s.bind(('127.0.0.1', 0)); print(s.getsockname()[1]); s.close()
+PY
+)
+PID=
+CONTAINER=
+cleanup() {
+  [ -z "$CONTAINER" ] || podman rm -f "$CONTAINER" >/dev/null 2>&1 || true
+  [ -z "$PID" ] || kill "$PID" >/dev/null 2>&1 || true
+  rm -rf "$ROOT"
+}
+trap cleanup EXIT
+mkdir -p "$ROOT"/{source/movies/Test,view,cache/files,state}
 touch "$ROOT/cache/.pretranscode-cache"
+printf partial >"$ROOT/cache/files/failed.partial"
+printf orphan >"$ROOT/cache/files/orphan.bin"
 
-docker build -q -t "$IMAGE" . >/dev/null
-docker run --rm --entrypoint ffmpeg -v "$ROOT/source:/source" "$IMAGE" \
+cat >"$ROOT/source/movies/Test/embedded.srt" <<'EOF'
+1
+00:00:00,000 --> 00:00:05,000
+Compatibility test
+EOF
+
+podman run --rm --userns=keep-id --entrypoint ffmpeg -v "$ROOT/source:/source:Z" "$IMAGE" \
   -hide_banner -loglevel error -f lavfi -i testsrc2=size=1280x720:rate=24 \
-  -t 6 -c:v libx265 -pix_fmt yuv420p10le -an /source/movies/Test/Test.mkv
+  -f lavfi -i sine=frequency=440:sample_rate=48000 \
+  -f srt -i /source/movies/Test/embedded.srt -t 6 \
+  -map 0:v:0 -map 1:a:0 -map 2:s:0 -c:v libx265 -pix_fmt yuv420p10le \
+  -c:a eac3 -c:s srt /source/movies/Test/Test.mkv
 
-docker run --rm --entrypoint ffmpeg -v "$ROOT/source:/source" "$IMAGE" \
-  -hide_banner -loglevel error -f lavfi -i testsrc2=size=1280x720:rate=24 \
-  -t 6 -c:v libx265 -pix_fmt yuv420p10le \
-  -color_primaries bt2020 -color_trc smpte2084 -colorspace bt2020nc \
-  -x265-params 'colorprim=bt2020:transfer=smpte2084:colormatrix=bt2020nc' \
-  -an /source/movies/Test/Test.HDR10.mkv
-
-docker run --rm --network "$NETWORK" --entrypoint python3 \
+TEST_VIEW="$ROOT/view" TEST_STATE="$ROOT/jellyfin-state.json" \
+  python3 "$(dirname "$0")/fake_jellyfin.py" "$PORT" &
+PID=$!
+sleep 0.2
+CONTAINER=pretranscode-c-integration-$RANDOM
+podman run -d --userns=keep-id --name "$CONTAINER" \
+  --add-host host.containers.internal:host-gateway \
   -e SOURCE_ROOT=/source -e VIEW_ROOT=/view -e CACHE_ROOT=/cache -e STATE_ROOT=/state \
-  -e JELLYFIN_URL="$JELLYFIN_URL" -e JELLYFIN_API_KEY \
+  -e JELLYFIN_URL="http://host.containers.internal:$PORT" -e JELLYFIN_API_KEY=test \
   -e JELLYFIN_LIBRARY_ROOT=/data/library -e JELLYFIN_ORIGINAL_ROOT=/data/original \
-  -e JELLYFIN_CACHE_ROOT=/data/cache \
-  -v "$ROOT/source:/source:ro" -v "$ROOT/view:/view" -v "$ROOT/cache:/cache" -v "$ROOT/state:/state" \
-  "$IMAGE" -c 'import importlib.machinery; m=importlib.machinery.SourceFileLoader("p","/app/pretranscode").load_module(); x=m.Manager(); assert x.read_sessions(); x.scan_inventory(); x.sync_view(); assert x.encode("movies/Test/Test.mkv",False)=="completed"; assert x.encode("movies/Test/Test.HDR10.mkv",False)=="completed"'
+  -e JELLYFIN_CACHE_ROOT=/data/cache -e POLL_SECONDS=1 -e RESCAN_SECONDS=1 \
+  -e H264_PRESET=ultrafast \
+  -v "$ROOT/source:/source:ro,Z" -v "$ROOT/view:/view:Z" \
+  -v "$ROOT/cache:/cache:Z" -v "$ROOT/state:/state:Z" "$IMAGE" >/dev/null
 
-SDR_TARGET=$(readlink "$ROOT/view/movies/Test/Test.mkv")
-HDR_TARGET=$(readlink "$ROOT/view/movies/Test/Test.HDR10.mkv")
-docker run --rm --entrypoint ffprobe -v "$ROOT/cache:/cache:ro" "$IMAGE" \
-  -v error -select_streams v:0 -show_entries stream=codec_name,pix_fmt \
-  -of default=nw=1 "${SDR_TARGET/\/data\/cache/\/cache}" | grep -qx 'codec_name=h264'
-docker run --rm --entrypoint ffprobe -v "$ROOT/cache:/cache:ro" "$IMAGE" \
-  -v error -select_streams v:0 -show_entries stream=color_transfer,color_primaries,color_space \
-  -of default=nw=1 "${HDR_TARGET/\/data\/cache/\/cache}" | grep -qx 'color_transfer=bt709'
+for _ in $(seq 1 90); do
+  LINK=$(readlink "$ROOT/view/movies/Test/Test.mkv" 2>/dev/null || true)
+  case "$LINK" in
+    /data/cache/files/*) break ;;
+  esac
+  if [ "$(podman inspect -f '{{.State.Running}}' "$CONTAINER" 2>/dev/null || true)" != true ]; then
+    podman logs "$CONTAINER" >&2 || true
+    exit 1
+  fi
+  sleep 1
+done
+LINK=$(readlink "$ROOT/view/movies/Test/Test.mkv" 2>/dev/null || true)
+if [[ "$LINK" != /data/cache/files/* ]]; then
+  podman logs "$CONTAINER" >&2 || true
+  cat "$ROOT/state/ffmpeg.log" >&2 2>/dev/null || true
+  exit 1
+fi
+[[ "$LINK" == /data/cache/files/* ]]
+BASENAME=${LINK##*/}
+[[ "$BASENAME" =~ ^[0-9a-f]{32}\.[0-9a-f]{32}\.[0-9a-f]{32}\.[0-9a-f]{32}\.(mkv|mp4)$ ]]
+OUTPUT="$ROOT/cache/${LINK#/data/cache/}"
+test -s "$OUTPUT"
+! find "$ROOT/cache/files" -type f \( -name '*.json' -o -name '*.partial' \) | grep -q .
+test ! -e "$ROOT/cache/files/orphan.bin"
+podman run --rm --userns=keep-id --entrypoint ffprobe -v "$ROOT/cache:/cache:ro,Z" "$IMAGE" \
+  -v error -show_entries stream=codec_type,codec_name,pix_fmt,channels:stream_side_data=side_data_type \
+  -show_entries chapter=id -of json "/cache/${LINK#/data/cache/}" | python3 -c '
+import json,sys
+d=json.load(sys.stdin); streams=d.get("streams",[])
+video=[s for s in streams if s.get("codec_type")=="video"]
+audio=[s for s in streams if s.get("codec_type")=="audio"]
+subs=[s for s in streams if s.get("codec_type")=="subtitle"]
+assert len(video)==1 and video[0].get("codec_name")=="h264" and video[0].get("pix_fmt")=="yuv420p"
+assert len(audio)==1 and audio[0].get("codec_name")=="aac" and audio[0].get("channels")==2
+assert not subs and not d.get("chapters")
+assert not any("dovi" in x.get("side_data_type","").lower() for x in video[0].get("side_data_list",[]))
+'
+python3 - "$ROOT/jellyfin-state.json" <<'PY'
+import json, sys
+state=json.load(open(sys.argv[1]))
+assert state["codec"] == "h264" and state["depth"] == 8
+assert state["attempts"] >= 2
+PY
+
+# A profile change, including a published 1.x object, keeps the old object live
+# until its replacement exists.
+podman stop -t 10 "$CONTAINER" >/dev/null
+podman rm "$CONTAINER" >/dev/null
+CONTAINER=
+mv "$OUTPUT" "$ROOT/cache/files/legacy.mkv"
+ln -sfn /data/cache/files/legacy.mkv "$ROOT/view/movies/Test/Test.mkv"
+OLD_LINK=/data/cache/files/legacy.mkv
+podman run --rm --userns=keep-id \
+  --add-host host.containers.internal:host-gateway \
+  -e SOURCE_ROOT=/source -e VIEW_ROOT=/view -e CACHE_ROOT=/cache -e STATE_ROOT=/state \
+  -e JELLYFIN_URL="http://host.containers.internal:$PORT" -e JELLYFIN_API_KEY=test \
+  -e JELLYFIN_LIBRARY_ROOT=/data/library -e JELLYFIN_ORIGINAL_ROOT=/data/original \
+  -e JELLYFIN_CACHE_ROOT=/data/cache -e POLL_SECONDS=1 -e RESCAN_SECONDS=1 \
+  -e H264_PRESET=ultrafast -e H264_CRF=19 \
+  -v "$ROOT/source:/source:ro,Z" -v "$ROOT/view:/view:Z" \
+  -v "$ROOT/cache:/cache:Z" -v "$ROOT/state:/state:Z" "$IMAGE" sync
+[[ "$(readlink "$ROOT/view/movies/Test/Test.mkv")" == "$OLD_LINK" ]]
+test -s "$ROOT/cache/${OLD_LINK#/data/cache/}"
+
+CONTAINER=pretranscode-c-integration-$RANDOM
+podman run -d --userns=keep-id --name "$CONTAINER" \
+  --add-host host.containers.internal:host-gateway \
+  -e SOURCE_ROOT=/source -e VIEW_ROOT=/view -e CACHE_ROOT=/cache -e STATE_ROOT=/state \
+  -e JELLYFIN_URL="http://host.containers.internal:$PORT" -e JELLYFIN_API_KEY=test \
+  -e JELLYFIN_LIBRARY_ROOT=/data/library -e JELLYFIN_ORIGINAL_ROOT=/data/original \
+  -e JELLYFIN_CACHE_ROOT=/data/cache -e POLL_SECONDS=1 -e RESCAN_SECONDS=1 \
+  -e H264_PRESET=ultrafast -e H264_CRF=19 \
+  -v "$ROOT/source:/source:ro,Z" -v "$ROOT/view:/view:Z" \
+  -v "$ROOT/cache:/cache:Z" -v "$ROOT/state:/state:Z" "$IMAGE" >/dev/null
+for _ in $(seq 1 90); do
+  LINK=$(readlink "$ROOT/view/movies/Test/Test.mkv" 2>/dev/null || true)
+  [[ "$LINK" == /data/cache/files/* && "$LINK" != "$OLD_LINK" ]] && break
+  sleep 1
+done
+[[ "$LINK" == /data/cache/files/* && "$LINK" != "$OLD_LINK" ]]
+for _ in $(seq 1 10); do
+  test ! -e "$ROOT/cache/${OLD_LINK#/data/cache/}" && break
+  sleep 1
+done
+test ! -e "$ROOT/cache/${OLD_LINK#/data/cache/}"
+podman stop -t 10 "$CONTAINER" >/dev/null
+podman rm "$CONTAINER" >/dev/null
+CONTAINER=
 echo "integration test passed"

tests/test_core.c

@@ -0,0 +1,159 @@
+#define PRETRANSCODE_TEST
+#include "../pretranscode.c"
+
+#include <assert.h>
+
+static void write_file(const char *path, const char *value)
+{
+    char *parent = xstrdup(path), *slash = strrchr(parent, '/');
+    assert(slash); *slash = 0; assert(mkdirs(parent) == 0); free(parent);
+    int fd = open(path, O_WRONLY | O_CREAT | O_TRUNC | O_CLOEXEC, 0644);
+    assert(fd >= 0); assert(write(fd, value, strlen(value)) == (ssize_t)strlen(value));
+    assert(close(fd) == 0);
+}
+
+static void remove_tree(const char *root)
+{
+    char *roots[] = {(char *)root, NULL};
+    FTS *tree = fts_open(roots, FTS_PHYSICAL | FTS_NOCHDIR, NULL); assert(tree);
+    FTSENT *entry;
+    while ((entry = fts_read(tree))) {
+        if (entry->fts_info == FTS_DP) assert(rmdir(entry->fts_path) == 0);
+        else if (entry->fts_info != FTS_D) assert(unlink(entry->fts_path) == 0);
+    }
+    fts_close(tree);
+}
+
+static void test_hashes(void)
+{
+    char a[33], b[33];
+    path_hash("tv/Test/one.mkv", a); path_hash("tv/Test/one.mkv", b);
+    assert(strcmp(a, b) == 0); path_hash("tv/Test/two.mkv", b); assert(strcmp(a, b));
+    struct stat st = {.st_dev = 1, .st_ino = 2, .st_size = 3};
+    st.st_mtim.tv_sec = 4; st.st_mtim.tv_nsec = 5;
+    generation_hash(&st, a); st.st_ino++; generation_hash(&st, b); assert(strcmp(a, b));
+
+    char path[] = "/tmp/pretranscode-hash-XXXXXX";
+    int fd = mkstemp(path); assert(fd >= 0);
+    const char value[] = "immutable output";
+    assert(write(fd, value, sizeof(value) - 1) == (ssize_t)(sizeof(value) - 1)); close(fd);
+    assert(file_hash(path, a) == 0); hash_bytes(value, sizeof(value) - 1, b);
+    assert(strcmp(a, b) == 0); assert(unlink(path) == 0);
+}
+
+static void test_atomic_link_keeps_open_inode(void)
+{
+    char root[] = "/tmp/pretranscode-link-XXXXXX"; assert(mkdtemp(root));
+    char *view = xprintf("%s/view/file.mkv", root);
+    char *original = xprintf("%s/original", root), *cached = xprintf("%s/cached", root);
+    write_file(original, "original"); write_file(cached, "cached");
+    assert(atomic_link(view, original) == 1);
+    int old = open(view, O_RDONLY | O_CLOEXEC); assert(old >= 0);
+    assert(atomic_link(view, cached) == 1);
+    char data[16] = {0}; assert(read(old, data, sizeof(data)) == 8); assert(!strcmp(data, "original")); close(old);
+    memset(data, 0, sizeof(data)); int current = open(view, O_RDONLY | O_CLOEXEC); assert(current >= 0);
+    assert(read(current, data, sizeof(data)) == 6); assert(!strcmp(data, "cached")); close(current);
+    free(view); free(original); free(cached); remove_tree(root);
+}
+
+static void test_cache_name(void)
+{
+    const char *hash = "00112233445566778899aabbccddeeff";
+    char *name = xprintf("%s.%s.%s.%s.mkv", hash, hash, hash, hash);
+    struct cache cache = {0}; assert(parse_cache_name(name, &cache)); cache.named = true;
+    assert(!strcmp(cache.path_hash, hash)); assert(!strcmp(cache.extension, ".mkv"));
+    struct item item = {.extension = ".mkv"};
+    memcpy(item.path_hash, hash, HASH_HEX + 1); memcpy(item.generation_hash, hash, HASH_HEX + 1);
+    assert(cache_matches(&cache, &item));
+    name[2] = 'X'; assert(!parse_cache_name(name, &cache)); free(name);
+}
+
+static void test_lru_does_not_change_media_mtime(void)
+{
+    char root[] = "/tmp/pretranscode-atime-XXXXXX"; assert(mkdtemp(root));
+    char *view_root = xprintf("%s/view", root), *media = xprintf("%s/media.mkv", root);
+    char *view = xprintf("%s/movie.mkv", view_root); write_file(media, "cached");
+    assert(atomic_link(view, "/cache/files/media.mkv") == 1);
+    struct timespec old[2] = {{.tv_sec = 100}, {.tv_sec = 100}};
+    assert(utimensat(AT_FDCWD, media, old, 0) == 0);
+    struct manager manager = {.cfg.view_root = view_root};
+    struct cache cache = {.path = media, .target = "/cache/files/media.mkv"};
+    manager.caches.v = &cache; manager.caches.n = 1;
+    touch_target(&manager, "movie.mkv");
+    struct stat st; assert(stat(media, &st) == 0);
+    assert(st.st_mtim.tv_sec == 100); assert(st.st_atim.tv_sec > 100);
+    manager.caches.v = NULL; manager.caches.n = 0;
+    free(view_root); free(media); free(view); remove_tree(root);
+}
+
+static void test_old_profile_is_replacement_candidate(void)
+{
+    const char *hash = "00112233445566778899aabbccddeeff";
+    struct cache old = {.named = true, .target = "/cache/files/old.mkv"};
+    memcpy(old.path_hash, hash, HASH_HEX + 1); memcpy(old.generation_hash, hash, HASH_HEX + 1);
+    memcpy(old.extension, ".mkv", 5); memset(old.profile_hash, '1', HASH_HEX); old.profile_hash[HASH_HEX] = 0;
+    struct item item = {.relative = "movie.mkv", .extension = ".mkv", .fallback = &old, .mtime = 1};
+    memcpy(item.path_hash, hash, HASH_HEX + 1); memcpy(item.generation_hash, hash, HASH_HEX + 1);
+    struct manager manager = {0}; manager.items.v = &item; manager.items.n = 1;
+    memset(manager.cfg.profile_hash, '2', HASH_HEX); manager.cfg.profile_hash[HASH_HEX] = 0;
+    assert(valid_cache(&manager, &item) == NULL);
+    assert(candidate(&manager) == &item);
+}
+
+static void test_single_writer_lock(void)
+{
+    char root[] = "/tmp/pretranscode-lock-XXXXXX";
+    assert(mkdtemp(root));
+    int first = open(root, O_RDONLY | O_DIRECTORY | O_CLOEXEC);
+    int second = open(root, O_RDONLY | O_DIRECTORY | O_CLOEXEC);
+    assert(first >= 0 && second >= 0);
+    assert(flock(first, LOCK_EX | LOCK_NB) == 0);
+    assert(flock(second, LOCK_EX | LOCK_NB) < 0 && (errno == EWOULDBLOCK || errno == EAGAIN));
+    close(second); close(first); assert(rmdir(root) == 0);
+}
+
+static void test_scheduling(void)
+{
+    int pipes[2]; assert(pipe(pipes) == 0); pid_t pid = fork(); assert(pid >= 0);
+    if (!pid) {
+        close(pipes[0]); background_or_exit();
+        int values[2] = {sched_getscheduler(0),
+            IOPRIO_PRIO_CLASS((int)syscall(SYS_ioprio_get, IOPRIO_WHO_PROCESS, 0))};
+        if (write(pipes[1], values, sizeof(values)) != (ssize_t)sizeof(values))
+            _exit(1);
+        _exit(0);
+    }
+    close(pipes[1]); int values[2]; assert(read(pipes[0], values, sizeof(values)) == (ssize_t)sizeof(values));
+    close(pipes[0]); assert(waitpid(pid, NULL, 0) == pid);
+    assert(values[0] == SCHED_IDLE); assert(values[1] == IOPRIO_CLASS_IDLE);
+}
+
+static void test_output_validation(void)
+{
+    struct media source = {.codec = "hevc", .depth = 10, .width = 1920, .height = 1080,
+        .audio = 2, .subtitles = 3, .attachments = 1, .chapters = 12, .duration = 1200};
+    struct media output = {.codec = "h264", .depth = 8, .width = 1920, .height = 1080,
+        .audio = 1, .audio_compatible = true, .duration = 1200};
+    assert(valid_output(&source, &output));
+    output.dolby_vision = true; assert(!valid_output(&source, &output)); output.dolby_vision = false;
+    output.subtitles = 1; assert(!valid_output(&source, &output)); output.subtitles = 0;
+    output.audio_compatible = false; assert(!valid_output(&source, &output));
+}
+
+static void test_canonical_policy(void)
+{
+    struct media media = {.codec = "h264", .depth = 8, .audio = 1, .audio_compatible = true};
+    assert(!incompatible(&media));
+    snprintf(media.codec, sizeof(media.codec), "av1"); assert(incompatible(&media));
+    snprintf(media.codec, sizeof(media.codec), "h264"); media.audio_compatible = false; assert(incompatible(&media));
+    media.audio_compatible = true; media.subtitles = 1; assert(incompatible(&media));
+}
+
+int main(void)
+{
+    test_hashes(); test_atomic_link_keeps_open_inode(); test_cache_name();
+    test_lru_does_not_change_media_mtime(); test_old_profile_is_replacement_candidate();
+    test_single_writer_lock(); test_scheduling();
+    test_output_validation(); test_canonical_policy();
+    puts("all core tests passed"); return 0;
+}

tests/test_core.py

@@ -1,108 +0,0 @@
-import importlib.util
-import json
-import os
-import tempfile
-import unittest
-from pathlib import Path
-
-
-class CoreTests(unittest.TestCase):
-    @classmethod
-    def setUpClass(cls):
-        cls.temp = tempfile.TemporaryDirectory()
-        root = Path(cls.temp.name)
-        for name in ("source", "view", "cache", "state"):
-            (root / name).mkdir()
-        (root / "cache" / ".pretranscode-cache").touch()
-        os.environ.update({
-            "SOURCE_ROOT": str(root / "source"),
-            "VIEW_ROOT": str(root / "view"),
-            "CACHE_ROOT": str(root / "cache"),
-            "STATE_ROOT": str(root / "state"),
-            "JELLYFIN_LIBRARY_ROOT": "/library",
-            "JELLYFIN_ORIGINAL_ROOT": "/original",
-            "JELLYFIN_CACHE_ROOT": "/cached",
-        })
-        spec = importlib.util.spec_from_file_location("pretranscode", Path(__file__).parents[1] / "pretranscode.py")
-        cls.module = importlib.util.module_from_spec(spec)
-        spec.loader.exec_module(cls.module)
-        cls.root = root
-
-    @classmethod
-    def tearDownClass(cls):
-        cls.temp.cleanup()
-
-    def test_paths_are_configurable(self):
-        m = self.module
-        self.assertEqual(m.relative_source("/library/tv/Show/episode.mkv"), "tv/Show/episode.mkv")
-        self.assertEqual(m.source_target("tv/Show/episode.mkv"), "/original/tv/Show/episode.mkv")
-        self.assertTrue(m.cache_target("tv/Show/episode.mkv").startswith("/cached/files/"))
-
-    def test_series_lock_key(self):
-        m = self.module
-        self.assertEqual(m.item_key("tv/Show/Season 1/Episode.mkv"), "series:Show")
-        self.assertEqual(m.item_key("movies/Movie/file.mkv"), "item:movies/Movie/file.mkv")
-
-    def test_fixed_codec_blacklist(self):
-        manager = self.module.Manager()
-        self.assertTrue(manager.incompatible({"codec": "hevc", "bit_depth": 10}))
-        self.assertTrue(manager.incompatible({"codec": "h264", "bit_depth": 10}))
-        self.assertFalse(manager.incompatible({"codec": "h264", "bit_depth": 8}))
-        self.assertFalse(manager.incompatible({"codec": "av1", "bit_depth": 10}))
-
-    def test_quality_filters_are_global(self):
-        manager = self.module.Manager()
-        sdr = manager.ffmpeg_command(Path("show.mkv"), Path("out.mkv.partial"), {})
-        hdr = manager.ffmpeg_command(
-            Path("show.HDR10.mkv"), Path("out.mkv.partial"), {"color_transfer": "smpte2084"}
-        )
-        self.assertIn("zscale=dither=error_diffusion,format=yuv420p", sdr)
-        self.assertIn("18", sdr)
-        self.assertTrue(any("tonemap=tonemap=hable" in value for value in hdr))
-        self.assertIn("bt709", hdr)
-
-    def test_cache_requires_matching_source_fingerprint(self):
-        m = self.module
-        manager = m.Manager()
-        relative = "movies/Test/Test.mkv"
-        source = self.root / "source" / relative
-        source.parent.mkdir(parents=True, exist_ok=True)
-        source.write_bytes(b"original")
-        output, manifest = manager.cache_paths(relative)
-        output.write_bytes(b"cached")
-        manifest.write_text(json.dumps({
-            "relative": relative,
-            "source_fingerprint": m.fingerprint(source),
-            "encode_profile": m.ENCODE_PROFILE,
-        }))
-        self.assertTrue(manager.cache_valid(relative))
-        source.write_bytes(b"changed")
-        self.assertFalse(manager.cache_valid(relative))
-
-    def test_smaller_cache_limit_evicts_oldest_content(self):
-        m = self.module
-        manager = m.Manager()
-        relative = "movies/Resize/Resize.mkv"
-        source = self.root / "source" / relative
-        source.parent.mkdir(parents=True, exist_ok=True)
-        source.write_bytes(b"original")
-        output, manifest = manager.cache_paths(relative)
-        output.write_bytes(b"cached rendition")
-        manifest.write_text(json.dumps({
-            "relative": relative,
-            "source_fingerprint": m.fingerprint(source),
-            "encode_profile": m.ENCODE_PROFILE,
-        }))
-        manager.ensure_link(relative, m.cache_target(relative))
-        old_limit = m.CACHE_MAX_BYTES
-        try:
-            m.CACHE_MAX_BYTES = 1
-            manager.enforce_cache_limit()
-        finally:
-            m.CACHE_MAX_BYTES = old_limit
-        self.assertFalse(output.exists())
-        self.assertEqual(os.readlink(self.root / "view" / relative), m.source_target(relative))
-
-
-if __name__ == "__main__":
-    unittest.main()