From bd2b623633daebb5d8769aabb21667f8834506ee Mon Sep 17 00:00:00 2001 From: SpiderFish Date: Sat, 29 Aug 2026 21:56:05 +1000 Subject: [PATCH] Replace stateful cache with atomic C controller --- .gitignore | 2 + Dockerfile | 23 +- Makefile | 23 + README.md | 164 +++-- compose.yaml | 2 +- pretranscode.c | 1333 ++++++++++++++++++++++++++++++++++++++++ pretranscode.py | 908 --------------------------- tests/__init__.py | 0 tests/fake_jellyfin.py | 88 +++ tests/integration.sh | 157 ++++- tests/test_core.c | 159 +++++ tests/test_core.py | 163 ----- 12 files changed, 1823 insertions(+), 1199 deletions(-) create mode 100644 Makefile create mode 100644 pretranscode.c delete mode 100755 pretranscode.py delete mode 100644 tests/__init__.py create mode 100755 tests/fake_jellyfin.py create mode 100644 tests/test_core.c delete mode 100644 tests/test_core.py diff --git a/.gitignore b/.gitignore index cff5543..f7c9b46 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,5 @@ .env __pycache__/ *.pyc +/pretranscode +/tests/test_core diff --git a/Dockerfile b/Dockerfile index 72fd4a0..41b467c 100644 --- a/Dockerfile +++ b/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"] diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..a124a11 --- /dev/null +++ b/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 diff --git a/README.md b/README.md index e4f8c20..5f5669d 100644 --- a/README.md +++ b/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,90 +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 filtered video is mapped as a new -stream so source-only Dolby Vision configuration data cannot leak into H.264 -outputs. +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. - -## Cache resizing - -`CACHE_MAX_BYTES` is the worker's configurable LRU limit. After changing it and -restarting the container: - -- 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. - -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. - -## Playback locking - -- 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. - -## 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. - -## Upgrading from v1.2.0 - -Version 1.2.1 uses a v4 cache profile. Before starting the new worker, stop the -old worker and migrate its manifests: +Commands: ```bash -docker compose stop pretranscode -docker compose run --rm pretranscode migrate-v4 +pretranscode run +pretranscode sync +pretranscode active +pretranscode trim BYTES +pretranscode purge +pretranscode --version ``` -The migration re-probes immutable sources, updates every valid v3 manifest, and -creates a state backup. Cached H.264 streams that inherited Dolby Vision -configuration data are repaired with a lossless stream-copy remux; video is not -re-encoded. Other cached media files are not rewritten. The command refuses to -run while Jellyfin reports active playback. +`purge` and `trim` refuse to operate during active playback. -## Migration +## Failure behaviour -Copy or recreate these resources on the destination: +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. -- Repository and `.env` -- State directory -- Symlink-view directory -- Cache image/mount, if retaining completed transcodes +## Tests -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. +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: + +```bash +podman build -t pretranscode-test . +IMAGE=pretranscode-test tests/integration.sh +``` diff --git a/compose.yaml b/compose.yaml index d0d35c7..ff18a2b 100644 --- a/compose.yaml +++ b/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 diff --git a/pretranscode.c b/pretranscode.c new file mode 100644 index 0000000..7895310 --- /dev/null +++ b/pretranscode.c @@ -0,0 +1,1333 @@ +#define _GNU_SOURCE + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +/* A single-writer, atomic Jellyfin compatibility cache. */ + +#define VERSION "2.0.0" +#define HASH_HEX 32 +#define AUDIO_BITRATE "192k" +#define HASH_CHUNK (4U * 1024U * 1024U) +#define CAPTURE_MAX (16U * 1024U * 1024U) +#define ARRAY_LEN(a) (sizeof(a) / sizeof((a)[0])) + +struct config { + char *source_root, *view_root, *cache_root, *state_root; + char *jellyfin_url, *library_root, *original_root, *jellyfin_cache_root; + char *api_key, *ffmpeg, *ffprobe; + char *crf, *preset, *maxrate, *bufsize, *profile; + char profile_hash[HASH_HEX + 1]; + uint64_t cache_max; + unsigned cooldown_seconds, poll_seconds, rescan_seconds; +}; + +struct strings { char **v; size_t n, cap; }; +struct buffer { char *v; size_t n, cap; }; + +struct timer { char *key; double value; }; +struct timers { struct timer *v; size_t n, cap; }; + +struct media { + char codec[24], profile[64], transfer[32]; + int depth, width, height, audio, subtitles, attachments, chapters; + double duration; + bool audio_compatible, dolby_vision; +}; + +struct catalog_item { char *relative, *id; struct media media; }; +struct catalog { struct catalog_item *v; size_t n, cap; }; + +struct cache { + char *name, *path, *target; + char path_hash[HASH_HEX + 1]; + char generation_hash[HASH_HEX + 1]; + char profile_hash[HASH_HEX + 1]; + char output_hash[HASH_HEX + 1]; + char extension[5]; + uint64_t size; + double used_at; + bool named, referenced, retain; +}; +struct caches { struct cache *v; size_t n, cap; }; + +struct item { + char *relative, *source, *view; + char path_hash[HASH_HEX + 1]; + char generation_hash[HASH_HEX + 1]; + const char *extension; + uint64_t size; + double mtime; + struct catalog_item *catalog; + struct cache *cache, *fallback; + char *live_target; +}; +struct items { struct item *v; size_t n, cap; }; + +struct manager { + struct config cfg; + struct strings active_items, active_keys, previous_items, previous_keys; + struct timers cooldowns, priorities, retries; + struct catalog catalog; + struct caches caches; + struct items items; + pid_t child; + int cache_lock; + bool sessions_healthy; + double last_scan; +}; + +static volatile sig_atomic_t stopping; + +static void die(const char *fmt, ...) __attribute__((format(printf, 1, 2), noreturn)); +static void logmsg(const char *fmt, ...) __attribute__((format(printf, 1, 2))); + +static void die(const char *fmt, ...) +{ + va_list ap; + va_start(ap, fmt); + fputs("pretranscode: ", stderr); + vfprintf(stderr, fmt, ap); + va_end(ap); + fputc('\n', stderr); + exit(EXIT_FAILURE); +} + +static void logmsg(const char *fmt, ...) +{ + char stamp[32]; + time_t now = time(NULL); + struct tm tm; + gmtime_r(&now, &tm); + strftime(stamp, sizeof(stamp), "%Y-%m-%dT%H:%M:%SZ", &tm); + printf("%s ", stamp); + va_list ap; + va_start(ap, fmt); + vprintf(fmt, ap); + va_end(ap); + putchar('\n'); + fflush(stdout); +} + +static void *xmalloc(size_t n) +{ + void *p = malloc(n ? n : 1); + if (!p) die("out of memory"); + return p; +} + +static void *xrealloc(void *p, size_t n) +{ + p = realloc(p, n ? n : 1); + if (!p) die("out of memory"); + return p; +} + +static char *xstrdup(const char *s) +{ + char *p = strdup(s ? s : ""); + if (!p) die("out of memory"); + return p; +} + +static char *xprintf(const char *fmt, ...) __attribute__((format(printf, 1, 2))); +static char *xprintf(const char *fmt, ...) +{ + char *s = NULL; + va_list ap; + va_start(ap, fmt); + if (vasprintf(&s, fmt, ap) < 0) s = NULL; + va_end(ap); + if (!s) die("out of memory"); + return s; +} + +static double seconds(clockid_t id) +{ + struct timespec ts; + if (clock_gettime(id, &ts) < 0) die("clock_gettime: %s", strerror(errno)); + return (double)ts.tv_sec + (double)ts.tv_nsec / 1e9; +} + +static double now(void) { return seconds(CLOCK_REALTIME); } +static double monotonic(void) { return seconds(CLOCK_MONOTONIC); } + +static bool ends(const char *s, const char *suffix) +{ + size_t a = strlen(s), b = strlen(suffix); + return a >= b && memcmp(s + a - b, suffix, b) == 0; +} + +static bool hex_hash(const char *s) +{ + for (size_t i = 0; i < HASH_HEX; i++) + if (!((s[i] >= '0' && s[i] <= '9') || (s[i] >= 'a' && s[i] <= 'f'))) + return false; + return s[HASH_HEX] == 0; +} + +static bool relative_path(const char *s) +{ + if (!s || !*s || *s == '/') return false; + for (const char *p = s; *p;) { + const char *e = strchr(p, '/'); + size_t n = e ? (size_t)(e - p) : strlen(p); + if (!n || (n == 1 && p[0] == '.') || (n == 2 && p[0] == '.' && p[1] == '.')) + return false; + if (!e) break; + p = e + 1; + } + return true; +} + +static bool video_name(const char *name) +{ + static const char *const extensions[] = { + ".mkv", ".mp4", ".m4v", ".avi", ".mov", ".ts", ".m2ts", ".webm" + }; + const char *ext = strrchr(name, '.'); + if (!ext) return false; + for (size_t i = 0; i < ARRAY_LEN(extensions); i++) + if (strcasecmp(ext, extensions[i]) == 0) return true; + return false; +} + +static const char *cache_extension(const char *relative) +{ + const char *ext = strrchr(relative, '.'); + return ext && (strcasecmp(ext, ".mp4") == 0 || strcasecmp(ext, ".m4v") == 0) + ? ".mp4" : ".mkv"; +} + +static int mkdirs(const char *path) +{ + char *copy = xstrdup(path); + for (char *p = copy + 1; *p; p++) { + if (*p != '/') continue; + *p = 0; + if (mkdir(copy, 0755) < 0 && errno != EEXIST) { + int saved = errno; free(copy); errno = saved; return -1; + } + *p = '/'; + } + int rc = mkdir(copy, 0755); + if (rc < 0 && errno == EEXIST) rc = 0; + int saved = errno; free(copy); errno = saved; return rc; +} + +static int sync_directory(const char *path) +{ + int fd = open(path, O_RDONLY | O_DIRECTORY | O_CLOEXEC); + if (fd < 0) return -1; + int rc = fsync(fd), saved = errno; + close(fd); errno = saved; return rc; +} + +static int sync_parent(const char *path) +{ + char *copy = xstrdup(path), *slash = strrchr(copy, '/'); + const char *parent = "."; + if (slash) { + if (slash == copy) slash[1] = 0; else *slash = 0; + parent = copy; + } + int rc = sync_directory(parent), saved = errno; + free(copy); errno = saved; return rc; +} + +static char *readlink_alloc(const char *path) +{ + for (size_t cap = 256;; cap *= 2) { + char *s = xmalloc(cap + 1); + ssize_t n = readlink(path, s, cap); + if (n < 0) { free(s); return NULL; } + if ((size_t)n < cap) { s[n] = 0; return s; } + free(s); + } +} + +static int atomic_link(const char *link, const char *target) +{ + char *current = readlink_alloc(link); + if (current && strcmp(current, target) == 0) { free(current); return 0; } + free(current); + char *parent = xstrdup(link), *slash = strrchr(parent, '/'); + if (!slash) { free(parent); errno = EINVAL; return -1; } + *slash = 0; + if (mkdirs(parent) < 0) { int e = errno; free(parent); errno = e; return -1; } + char *temporary = xprintf("%s/.pretranscode-link.%ld", parent, (long)getpid()); + unlink(temporary); + int rc = -1; + if (symlink(target, temporary) == 0 && rename(temporary, link) == 0 && sync_parent(link) == 0) + rc = 1; + int saved = errno; + if (rc < 0) unlink(temporary); + free(temporary); free(parent); errno = saved; return rc; +} + +static int string_cmp(const void *a, const void *b) +{ + return strcmp(*(const char *const *)a, *(const char *const *)b); +} + +static void strings_add(struct strings *s, const char *value) +{ + if (s->n == s->cap) { + s->cap = s->cap ? s->cap * 2 : 16; + s->v = xrealloc(s->v, s->cap * sizeof(*s->v)); + } + s->v[s->n++] = xstrdup(value); +} + +static void strings_sort(struct strings *s) +{ + qsort(s->v, s->n, sizeof(*s->v), string_cmp); + size_t out = 0; + for (size_t i = 0; i < s->n; i++) { + if (out && strcmp(s->v[i], s->v[out - 1]) == 0) free(s->v[i]); + else s->v[out++] = s->v[i]; + } + s->n = out; +} + +static bool strings_has(const struct strings *s, const char *value) +{ + return bsearch(&value, s->v, s->n, sizeof(*s->v), string_cmp) != NULL; +} + +static void strings_free(struct strings *s) +{ + for (size_t i = 0; i < s->n; i++) free(s->v[i]); + free(s->v); memset(s, 0, sizeof(*s)); +} + +static void strings_move(struct strings *to, struct strings *from) +{ + strings_free(to); *to = *from; memset(from, 0, sizeof(*from)); +} + +static void buffer_add(struct buffer *b, const void *data, size_t n) +{ + if (n > CAPTURE_MAX - b->n) die("subprocess output exceeded limit"); + if (b->n + n + 1 > b->cap) { + size_t cap = b->cap ? b->cap : 4096; + while (cap < b->n + n + 1) cap *= 2; + b->v = xrealloc(b->v, cap); b->cap = cap; + } + memcpy(b->v + b->n, data, n); b->n += n; b->v[b->n] = 0; +} + +static size_t curl_write(void *data, size_t size, size_t count, void *opaque) +{ + struct buffer *b = opaque; + size_t n = size * count; + if (n > CAPTURE_MAX - b->n) return 0; + buffer_add(b, data, n); return n; +} + +static const char *jstring(json_object *o, const char *key, const char *fallback) +{ + json_object *v = NULL; + return o && json_object_object_get_ex(o, key, &v) && json_object_is_type(v, json_type_string) + ? json_object_get_string(v) : fallback; +} + +static int jint(json_object *o, const char *key, int fallback) +{ + json_object *v = NULL; + return o && json_object_object_get_ex(o, key, &v) ? json_object_get_int(v) : fallback; +} + +static double jdouble(json_object *o, const char *key, double fallback) +{ + json_object *v = NULL; + return o && json_object_object_get_ex(o, key, &v) ? json_object_get_double(v) : fallback; +} + +static json_object *api(struct manager *m, const char *path, const char *method) +{ + CURL *curl = curl_easy_init(); + if (!curl) return NULL; + char *url = xprintf("%s%s", m->cfg.jellyfin_url, path); + char *token = xprintf("X-Emby-Token: %s", m->cfg.api_key); + struct curl_slist *headers = NULL; + headers = curl_slist_append(headers, token); + headers = curl_slist_append(headers, "Content-Length: 0"); + struct buffer body = {0}; + curl_easy_setopt(curl, CURLOPT_URL, url); + curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers); + curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, curl_write); + curl_easy_setopt(curl, CURLOPT_WRITEDATA, &body); + curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT, 5L); + curl_easy_setopt(curl, CURLOPT_TIMEOUT, 10L); + curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1L); + if (strcmp(method, "GET") != 0) curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, method); + CURLcode result = curl_easy_perform(curl); + long status = 0; + curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &status); + json_object *value = NULL; + if (result == CURLE_OK && status >= 200 && status < 300) + value = body.n ? json_tokener_parse(body.v) : json_object_new_object(); + free(body.v); curl_slist_free_all(headers); curl_easy_cleanup(curl); + free(token); free(url); return value; +} + +static void hash_to_hex(XXH128_hash_t hash, char out[HASH_HEX + 1]) +{ + XXH128_canonical_t canonical; + XXH128_canonicalFromHash(&canonical, hash); + for (size_t i = 0; i < sizeof(canonical.digest); i++) + snprintf(out + i * 2, 3, "%02x", canonical.digest[i]); + out[HASH_HEX] = 0; +} + +static void hash_bytes(const void *data, size_t n, char out[HASH_HEX + 1]) +{ + hash_to_hex(XXH3_128bits(data, n), out); +} + +static void path_hash(const char *relative, char out[HASH_HEX + 1]) +{ + hash_bytes(relative, strlen(relative), out); +} + +static void generation_hash(const struct stat *st, char out[HASH_HEX + 1]) +{ + char generation[192]; + int n = snprintf(generation, sizeof(generation), "%llu:%llu:%lld:%lld.%09ld", + (unsigned long long)st->st_dev, (unsigned long long)st->st_ino, + (long long)st->st_size, (long long)st->st_mtim.tv_sec, st->st_mtim.tv_nsec); + if (n < 0 || (size_t)n >= sizeof(generation)) die("source identity overflow"); + hash_bytes(generation, (size_t)n, out); +} + +static int file_hash(const char *path, char out[HASH_HEX + 1]) +{ + int fd = open(path, O_RDONLY | O_CLOEXEC | O_NOFOLLOW); + if (fd < 0) return -1; + void *data = xmalloc(HASH_CHUNK); + XXH3_state_t *state = XXH3_createState(); + if (!state || XXH3_128bits_reset(state) == XXH_ERROR) die("XXH3 initialization failed"); + int rc = 0; + while (!stopping) { + ssize_t n = read(fd, data, HASH_CHUNK); + if (!n) break; + if (n < 0) { if (errno == EINTR) continue; rc = -1; break; } + if (XXH3_128bits_update(state, data, (size_t)n) == XXH_ERROR) die("XXH3 failed"); + } + if (stopping) { errno = EINTR; rc = -1; } + if (!rc) hash_to_hex(XXH3_128bits_digest(state), out); + int saved = errno; + XXH3_freeState(state); free(data); close(fd); errno = saved; return rc; +} + +static void background_or_exit(void) +{ + struct sched_param parameter = {0}; + if (sched_setscheduler(0, SCHED_IDLE, ¶meter) < 0) { + dprintf(STDERR_FILENO, "pretranscode: SCHED_IDLE: %s\n", strerror(errno)); + _exit(126); + } + if (syscall(SYS_ioprio_set, IOPRIO_WHO_PROCESS, 0, + IOPRIO_PRIO_VALUE(IOPRIO_CLASS_IDLE, 0)) < 0) { + dprintf(STDERR_FILENO, "pretranscode: IOPRIO_CLASS_IDLE: %s\n", strerror(errno)); + _exit(126); + } +} + +static int background_file_hash(const char *path, char out[HASH_HEX + 1]) +{ + int pipes[2]; + if (pipe2(pipes, O_CLOEXEC) < 0) return -1; + pid_t pid = fork(); + if (pid < 0) { close(pipes[0]); close(pipes[1]); return -1; } + if (!pid) { + close(pipes[0]); + if (setsid() < 0) _exit(126); + background_or_exit(); + char hash[HASH_HEX + 1]; + if (file_hash(path, hash) < 0 || write(pipes[1], hash, sizeof(hash)) != (ssize_t)sizeof(hash)) + _exit(1); + _exit(0); + } + close(pipes[1]); + size_t received = 0; + while (received < HASH_HEX + 1 && !stopping) { + ssize_t n = read(pipes[0], out + received, HASH_HEX + 1 - received); + if (n > 0) received += (size_t)n; + else if (!n) break; + else if (errno != EINTR) break; + } + if (stopping) kill(-pid, SIGKILL); + int status = 0; + waitpid(pid, &status, 0); + close(pipes[0]); + return received == HASH_HEX + 1 && WIFEXITED(status) && !WEXITSTATUS(status) ? 0 : -1; +} + +static int capture(char *const argv[], unsigned timeout, struct buffer *output) +{ + int pipes[2]; + if (pipe2(pipes, O_CLOEXEC) < 0) return -1; + int flags = fcntl(pipes[0], F_GETFL); + if (flags < 0 || fcntl(pipes[0], F_SETFL, flags | O_NONBLOCK) < 0) { + close(pipes[0]); close(pipes[1]); return -1; + } + pid_t pid = fork(); + if (pid < 0) { close(pipes[0]); close(pipes[1]); return -1; } + if (!pid) { + if (setsid() < 0) _exit(126); + if (dup2(pipes[1], STDOUT_FILENO) < 0 || dup2(pipes[1], STDERR_FILENO) < 0) _exit(126); + close(pipes[0]); close(pipes[1]); + background_or_exit(); execv(argv[0], argv); _exit(127); + } + close(pipes[1]); + int status = 0; bool exited = false, eof = false; + double deadline = monotonic() + timeout; + while (!exited || !eof) { + struct pollfd pfd = {.fd = pipes[0], .events = POLLIN | POLLHUP}; + (void)poll(&pfd, 1, 100); + for (;;) { + char data[8192]; ssize_t n = read(pipes[0], data, sizeof(data)); + if (n > 0) { buffer_add(output, data, (size_t)n); continue; } + if (!n) eof = true; + break; + } + if (waitpid(pid, &status, WNOHANG) == pid) exited = true; + if ((stopping || monotonic() >= deadline) && !exited) { + kill(-pid, SIGKILL); waitpid(pid, &status, 0); exited = true; + } + } + close(pipes[0]); + return WIFEXITED(status) ? WEXITSTATUS(status) : -1; +} + +static int probe(struct manager *m, const char *path, struct media *media) +{ + char *const args[] = { + m->cfg.ffprobe, "-v", "quiet", "-show_entries", + "stream=codec_type,codec_name,bits_per_raw_sample,pix_fmt,profile,color_transfer,width,height,channels:stream_side_data=side_data_type", + "-show_entries", "format=duration", "-show_chapters", "-of", "json", (char *)path, NULL + }; + struct buffer output = {0}; + int rc = capture(args, 120, &output); + if (rc) { free(output.v); return -1; } + json_object *root = json_tokener_parse(output.v ? output.v : ""); + free(output.v); if (!root) return -1; + memset(media, 0, sizeof(*media)); media->depth = 8; media->audio_compatible = true; + json_object *streams = NULL; + bool found_video = false; + if (json_object_object_get_ex(root, "streams", &streams) && json_object_is_type(streams, json_type_array)) { + size_t count = json_object_array_length(streams); + for (size_t i = 0; i < count; i++) { + json_object *stream = json_object_array_get_idx(streams, i); + const char *type = jstring(stream, "codec_type", ""); + if (!found_video && strcmp(type, "video") == 0) { + found_video = true; + snprintf(media->codec, sizeof(media->codec), "%s", jstring(stream, "codec_name", "")); + snprintf(media->profile, sizeof(media->profile), "%s", jstring(stream, "profile", "")); + snprintf(media->transfer, sizeof(media->transfer), "%s", jstring(stream, "color_transfer", "")); + media->width = jint(stream, "width", 0); media->height = jint(stream, "height", 0); + int depth = jint(stream, "bits_per_raw_sample", 0); + const char *pixel = jstring(stream, "pix_fmt", ""); + if (!depth) depth = strstr(pixel, "12") || strstr(media->profile, "12") ? 12 + : strstr(pixel, "10") || strstr(media->profile, "10") ? 10 : 8; + media->depth = depth; + json_object *side_data = NULL; + if (json_object_object_get_ex(stream, "side_data_list", &side_data) && + json_object_is_type(side_data, json_type_array)) { + for (size_t j = 0; j < json_object_array_length(side_data); j++) { + json_object *side = json_object_array_get_idx(side_data, j); + if (strcasestr(jstring(side, "side_data_type", ""), "dovi")) + media->dolby_vision = true; + } + } + } else if (strcmp(type, "audio") == 0) { + media->audio++; + if (strcasecmp(jstring(stream, "codec_name", ""), "aac") || jint(stream, "channels", 0) > 2) + media->audio_compatible = false; + } else if (strcmp(type, "subtitle") == 0) media->subtitles++; + else if (strcmp(type, "attachment") == 0) media->attachments++; + } + } + json_object *format = NULL, *chapters = NULL; + if (json_object_object_get_ex(root, "format", &format)) media->duration = jdouble(format, "duration", 0); + if (json_object_object_get_ex(root, "chapters", &chapters) && json_object_is_type(chapters, json_type_array)) + media->chapters = (int)json_object_array_length(chapters); + json_object_put(root); return found_video ? 0 : -1; +} + +static bool incompatible(const struct media *m) +{ + return strcasecmp(m->codec, "h264") || m->depth > 8 || strstr(m->profile, "10") || + !m->audio_compatible || m->subtitles; +} + +static bool hdr(const struct media *m) +{ + return !strcasecmp(m->transfer, "smpte2084") || !strcasecmp(m->transfer, "arib-std-b67"); +} + +static char *relative_from_api(struct manager *m, const char *path) +{ + if (!path) return NULL; + const char *roots[] = {m->cfg.library_root, m->cfg.source_root}; + for (size_t i = 0; i < ARRAY_LEN(roots); i++) { + size_t n = strlen(roots[i]); + if (!strncmp(path, roots[i], n) && path[n] == '/' && relative_path(path + n + 1)) + return xstrdup(path + n + 1); + } + return NULL; +} + +static int catalog_cmp(const void *a, const void *b) +{ + return strcmp(((const struct catalog_item *)a)->relative, + ((const struct catalog_item *)b)->relative); +} + +static void catalog_add(struct catalog *c, struct catalog_item item) +{ + if (c->n == c->cap) { c->cap = c->cap ? c->cap * 2 : 256; c->v = xrealloc(c->v, c->cap * sizeof(*c->v)); } + c->v[c->n++] = item; +} + +static void catalog_free(struct catalog *c) +{ + for (size_t i = 0; i < c->n; i++) { free(c->v[i].relative); free(c->v[i].id); } + free(c->v); memset(c, 0, sizeof(*c)); +} + +static struct catalog_item *catalog_find(struct catalog *c, const char *relative) +{ + struct catalog_item key = {.relative = (char *)relative}; + return bsearch(&key, c->v, c->n, sizeof(*c->v), catalog_cmp); +} + +static bool read_catalog(struct manager *m) +{ + struct catalog next = {0}; + for (int start = 0;;) { + char *path = xprintf("/Items?Recursive=true&IncludeItemTypes=Movie%%2CEpisode&Fields=Path%%2CMediaStreams&StartIndex=%d&Limit=500", start); + json_object *page = api(m, path, "GET"); free(path); + if (!page) { catalog_free(&next); return false; } + json_object *array = NULL; + if (!json_object_object_get_ex(page, "Items", &array) || !json_object_is_type(array, json_type_array)) { + json_object_put(page); catalog_free(&next); return false; + } + size_t count = json_object_array_length(array); + for (size_t i = 0; i < count; i++) { + json_object *value = json_object_array_get_idx(array, i); + char *relative = relative_from_api(m, jstring(value, "Path", NULL)); + if (!relative) continue; + struct catalog_item item = {.relative = relative, .id = xstrdup(jstring(value, "Id", ""))}; + item.media.depth = 8; item.media.audio_compatible = true; + json_object *streams = NULL; bool video = false; + if (json_object_object_get_ex(value, "MediaStreams", &streams) && json_object_is_type(streams, json_type_array)) { + for (size_t j = 0; j < json_object_array_length(streams); j++) { + json_object *stream = json_object_array_get_idx(streams, j); + const char *type = jstring(stream, "Type", ""); + if (!video && !strcasecmp(type, "Video")) { + video = true; + snprintf(item.media.codec, sizeof(item.media.codec), "%s", jstring(stream, "Codec", "")); + snprintf(item.media.profile, sizeof(item.media.profile), "%s", jstring(stream, "Profile", "")); + snprintf(item.media.transfer, sizeof(item.media.transfer), "%s", jstring(stream, "ColorTransfer", "")); + item.media.depth = jint(stream, "BitDepth", 8); + item.media.width = jint(stream, "Width", 0); item.media.height = jint(stream, "Height", 0); + } else if (!strcasecmp(type, "Audio")) { + item.media.audio++; + if (strcasecmp(jstring(stream, "Codec", ""), "aac") || jint(stream, "Channels", 0) > 2) + item.media.audio_compatible = false; + } else if (!strcasecmp(type, "Subtitle")) item.media.subtitles++; + else if (!strcasecmp(type, "Attachment")) item.media.attachments++; + } + } + item.media.duration = jdouble(value, "RunTimeTicks", 0) / 10000000.0; + catalog_add(&next, item); + } + int total = jint(page, "TotalRecordCount", start + (int)count); + json_object_put(page); start += (int)count; + if (!count || start >= total) break; + } + qsort(next.v, next.n, sizeof(*next.v), catalog_cmp); + catalog_free(&m->catalog); m->catalog = next; return true; +} + +static bool refresh(struct manager *m, const struct item *item) +{ + if (!item || !item->catalog || !*item->catalog->id) return true; + char *path = xprintf("/Items/%s/Refresh?metadataRefreshMode=FullRefresh&imageRefreshMode=None&replaceAllMetadata=false&replaceAllImages=false", item->catalog->id); + json_object *reply = api(m, path, "POST"); free(path); + if (!reply) { logmsg("Unable to refresh Jellyfin item %s", item->relative); return false; } + json_object_put(reply); return true; +} + +static bool parse_cache_name(const char *name, struct cache *cache) +{ + size_t n = strlen(name); + const char *extension = n > 4 ? name + n - 4 : ""; + if (strcmp(extension, ".mkv") && strcmp(extension, ".mp4")) return false; + size_t stem_n = n - 4; + if (stem_n != HASH_HEX * 4 + 3) return false; + char stem[HASH_HEX * 4 + 4]; + memcpy(stem, name, stem_n); stem[stem_n] = 0; + char *parts[4], *save = NULL, *part = strtok_r(stem, ".", &save); + for (size_t i = 0; i < 4; i++) { + if (!part) return false; + parts[i] = part; part = strtok_r(NULL, ".", &save); + } + if (part || !hex_hash(parts[0]) || !hex_hash(parts[1]) || !hex_hash(parts[2]) || !hex_hash(parts[3])) + return false; + memcpy(cache->path_hash, parts[0], HASH_HEX + 1); + memcpy(cache->generation_hash, parts[1], HASH_HEX + 1); + memcpy(cache->profile_hash, parts[2], HASH_HEX + 1); + memcpy(cache->output_hash, parts[3], HASH_HEX + 1); + memcpy(cache->extension, extension, 5); + return true; +} + +static void cache_add(struct caches *c, struct cache value) +{ + if (c->n == c->cap) { c->cap = c->cap ? c->cap * 2 : 64; c->v = xrealloc(c->v, c->cap * sizeof(*c->v)); } + c->v[c->n++] = value; +} + +static void caches_free(struct caches *c) +{ + for (size_t i = 0; i < c->n; i++) { free(c->v[i].name); free(c->v[i].path); free(c->v[i].target); } + free(c->v); memset(c, 0, sizeof(*c)); +} + +static void scan_cache(struct manager *m) +{ + caches_free(&m->caches); + char *root = xprintf("%s/files", m->cfg.cache_root); + if (mkdirs(root) < 0) die("mkdir %s: %s", root, strerror(errno)); + DIR *directory = opendir(root); + if (!directory) die("opendir %s: %s", root, strerror(errno)); + bool removed = false; struct dirent *entry; + while ((entry = readdir(directory))) { + if (entry->d_name[0] == '.' || !strcmp(entry->d_name, "lost+found")) continue; + char *path = xprintf("%s/%s", root, entry->d_name); + struct stat st; + if (lstat(path, &st) < 0 || !S_ISREG(st.st_mode)) { free(path); continue; } + if (ends(entry->d_name, ".partial")) { if (unlink(path) == 0) removed = true; free(path); continue; } + struct cache cache = { + .name = xstrdup(entry->d_name), .path = path, + .target = xprintf("%s/files/%s", m->cfg.jellyfin_cache_root, entry->d_name), + .size = (uint64_t)st.st_size, + .used_at = (double)st.st_atim.tv_sec + (double)st.st_atim.tv_nsec / 1e9, + }; + cache.named = parse_cache_name(entry->d_name, &cache); + cache_add(&m->caches, cache); + } + closedir(directory); if (removed) sync_directory(root); free(root); +} + +static struct cache *cache_by_target(struct manager *m, const char *target) +{ + if (!target) return NULL; + for (size_t i = 0; i < m->caches.n; i++) + if (!strcmp(m->caches.v[i].target, target)) return &m->caches.v[i]; + return NULL; +} + +static bool cache_matches(const struct cache *cache, const struct item *item) +{ + return cache && cache->named && !strcmp(cache->path_hash, item->path_hash) && + !strcmp(cache->generation_hash, item->generation_hash) && + !strcmp(cache->extension, item->extension); +} + +static struct cache *valid_cache(struct manager *m, const struct item *item) +{ + struct cache *selected = NULL; + for (size_t i = 0; i < m->caches.n; i++) { + struct cache *cache = &m->caches.v[i]; + if (!cache_matches(cache, item) || strcmp(cache->profile_hash, m->cfg.profile_hash)) continue; + if (item->live_target && !strcmp(cache->target, item->live_target)) return cache; + if (!selected || cache->used_at > selected->used_at) selected = cache; + } + return selected; +} + +static void item_add(struct items *items, struct item item) +{ + if (items->n == items->cap) { items->cap = items->cap ? items->cap * 2 : 256; items->v = xrealloc(items->v, items->cap * sizeof(*items->v)); } + items->v[items->n++] = item; +} + +static void items_free(struct items *items) +{ + for (size_t i = 0; i < items->n; i++) { + free(items->v[i].relative); free(items->v[i].source); free(items->v[i].view); free(items->v[i].live_target); + } + free(items->v); memset(items, 0, sizeof(*items)); +} + +static struct item *item_by_path_hash(struct manager *m, const char *hash) +{ + for (size_t i = 0; i < m->items.n; i++) + if (!strcmp(m->items.v[i].path_hash, hash)) return &m->items.v[i]; + return NULL; +} + +static double timer_get(const struct timers *timers, const char *key) +{ + for (size_t i = 0; i < timers->n; i++) if (!strcmp(timers->v[i].key, key)) return timers->v[i].value; + return 0; +} + +static void timer_set(struct timers *timers, const char *key, double value) +{ + for (size_t i = 0; i < timers->n; i++) if (!strcmp(timers->v[i].key, key)) { timers->v[i].value = value; return; } + if (timers->n == timers->cap) { timers->cap = timers->cap ? timers->cap * 2 : 32; timers->v = xrealloc(timers->v, timers->cap * sizeof(*timers->v)); } + timers->v[timers->n++] = (struct timer){xstrdup(key), value}; +} + +static void timer_del(struct timers *timers, const char *key) +{ + for (size_t i = 0; i < timers->n; i++) if (!strcmp(timers->v[i].key, key)) { + free(timers->v[i].key); timers->v[i] = timers->v[--timers->n]; return; + } +} + +static void timers_prune(struct timers *timers, double before) +{ + for (size_t i = 0; i < timers->n;) + if (timers->v[i].value <= before) { + free(timers->v[i].key); timers->v[i] = timers->v[--timers->n]; + } else i++; +} + +static void timers_free(struct timers *timers) +{ + for (size_t i = 0; i < timers->n; i++) free(timers->v[i].key); + free(timers->v); memset(timers, 0, sizeof(*timers)); +} + +static char *lock_key(const char *relative) +{ + if (!strncasecmp(relative, "tv/", 3)) { + const char *slash = strchr(relative + 3, '/'); + if (slash) return xprintf("series:%.*s", (int)(slash - relative - 3), relative + 3); + } + return xprintf("item:%s", relative); +} + +static bool locked(struct manager *m, const char *relative) +{ + char *key = lock_key(relative); + bool value = strings_has(&m->active_keys, key) || timer_get(&m->cooldowns, key) > now(); + free(key); return value; +} + +static void touch_target(struct manager *m, const char *relative) +{ + char *view = xprintf("%s/%s", m->cfg.view_root, relative); + char *target = readlink_alloc(view); free(view); + struct cache *cache = cache_by_target(m, target); + free(target); + if (!cache) return; + struct timespec times[2] = {{.tv_nsec = UTIME_NOW}, {.tv_nsec = UTIME_OMIT}}; + if (utimensat(AT_FDCWD, cache->path, times, 0) == 0) cache->used_at = now(); +} + +static bool read_sessions(struct manager *m) +{ + json_object *sessions = api(m, "/Sessions", "GET"); + if (!sessions || !json_object_is_type(sessions, json_type_array)) { + if (sessions) json_object_put(sessions); + if (m->sessions_healthy) logmsg("Jellyfin sessions unavailable; pausing cache work"); + m->sessions_healthy = false; strings_free(&m->active_items); strings_free(&m->active_keys); return false; + } + struct strings items = {0}, keys = {0}; + for (size_t i = 0; i < json_object_array_length(sessions); i++) { + json_object *session = json_object_array_get_idx(sessions, i), *playing = NULL; + if (!json_object_object_get_ex(session, "NowPlayingItem", &playing)) continue; + char *relative = relative_from_api(m, jstring(playing, "Path", NULL)); + if (!relative) { + const char *id = jstring(playing, "Id", ""); + if (*id) { + char *path = xprintf("/Items/%s?Fields=Path", id); + json_object *detail = api(m, path, "GET"); + free(path); + if (detail) { + relative = relative_from_api(m, jstring(detail, "Path", NULL)); + json_object_put(detail); + } + } + } + if (!relative) continue; + strings_add(&items, relative); char *key = lock_key(relative); strings_add(&keys, key); free(key); + touch_target(m, relative); free(relative); + } + json_object_put(sessions); strings_sort(&items); strings_sort(&keys); + double time = now(); + for (size_t i = 0; i < items.n; i++) timer_set(&m->priorities, items.v[i], time); + for (size_t i = 0; i < m->previous_items.n; i++) + if (!strings_has(&items, m->previous_items.v[i])) timer_set(&m->priorities, m->previous_items.v[i], time); + for (size_t i = 0; i < keys.n; i++) if (!strncmp(keys.v[i], "series:", 7)) + timer_set(&m->cooldowns, keys.v[i], time + m->cfg.cooldown_seconds); + for (size_t i = 0; i < m->previous_keys.n; i++) + if (!strings_has(&keys, m->previous_keys.v[i]) && !strncmp(m->previous_keys.v[i], "series:", 7)) + timer_set(&m->cooldowns, m->previous_keys.v[i], time + m->cfg.cooldown_seconds); + strings_free(&m->previous_items); strings_free(&m->previous_keys); + for (size_t i = 0; i < items.n; i++) strings_add(&m->previous_items, items.v[i]); + for (size_t i = 0; i < keys.n; i++) strings_add(&m->previous_keys, keys.v[i]); + strings_sort(&m->previous_items); strings_sort(&m->previous_keys); + strings_move(&m->active_items, &items); strings_move(&m->active_keys, &keys); + m->sessions_healthy = true; return true; +} + +static bool cache_metadata_matches(const struct catalog_item *item) +{ + return item && !strcasecmp(item->media.codec, "h264") && item->media.depth <= 8 && + item->media.audio <= 1 && item->media.audio_compatible && !item->media.subtitles; +} + +static bool source_entry(FTSENT *entry) +{ + if (entry->fts_info == FTS_F) return true; + if (entry->fts_info != FTS_SL) return false; + struct stat st; return stat(entry->fts_path, &st) == 0 && S_ISREG(st.st_mode); +} + +static bool mark_references(struct manager *m) +{ + for (size_t i = 0; i < m->caches.n; i++) m->caches.v[i].referenced = false; + char *roots[] = {m->cfg.view_root, NULL}; + FTS *tree = fts_open(roots, FTS_PHYSICAL | FTS_NOCHDIR, NULL); + if (!tree) return false; + FTSENT *entry; + while ((entry = fts_read(tree))) if (entry->fts_info == FTS_SL || entry->fts_info == FTS_SLNONE) { + char *target = readlink_alloc(entry->fts_path); struct cache *cache = cache_by_target(m, target); + if (cache) + cache->referenced = true; + free(target); + } + fts_close(tree); return true; +} + +static bool reconcile(struct manager *m) +{ + if (!read_catalog(m)) { logmsg("Unable to read Jellyfin inventory"); return false; } + scan_cache(m); items_free(&m->items); + struct strings seen = {0}; + char *roots[] = {m->cfg.source_root, NULL}; + FTS *tree = fts_open(roots, FTS_PHYSICAL | FTS_NOCHDIR, NULL); + if (!tree) { strings_free(&seen); return false; } + size_t root_n = strlen(m->cfg.source_root); FTSENT *entry; bool view_changed = false; + while ((entry = fts_read(tree))) { + const char *relative = entry->fts_path + root_n; if (*relative == '/') relative++; + if (entry->fts_info == FTS_D) { + char *path = *relative ? xprintf("%s/%s", m->cfg.view_root, relative) : xstrdup(m->cfg.view_root); + if (mkdirs(path) < 0) { free(path); fts_close(tree); strings_free(&seen); return false; } + free(path); continue; + } + if (!source_entry(entry)) continue; + strings_add(&seen, relative); + char *view = xprintf("%s/%s", m->cfg.view_root, relative); + if (!video_name(entry->fts_name)) { + char *target = xprintf("%s/%s", m->cfg.original_root, relative); + int changed = atomic_link(view, target); + if (changed < 0) logmsg("Unable to link %s: %s", relative, strerror(errno)); + else if (changed > 0) view_changed = true; + free(target); free(view); continue; + } + struct stat st; + if (stat(entry->fts_path, &st) < 0 || !S_ISREG(st.st_mode)) { free(view); continue; } + struct item item = { + .relative = xstrdup(relative), .source = xstrdup(entry->fts_path), .view = view, + .extension = cache_extension(relative), .size = (uint64_t)st.st_size, + .mtime = (double)st.st_mtim.tv_sec + (double)st.st_mtim.tv_nsec / 1e9, + .catalog = catalog_find(&m->catalog, relative), .live_target = readlink_alloc(view), + }; + path_hash(relative, item.path_hash); generation_hash(&st, item.generation_hash); + item.cache = valid_cache(m, &item); + struct cache *live_cache = cache_by_target(m, item.live_target); + if (live_cache != item.cache && (cache_matches(live_cache, &item) || (live_cache && !live_cache->named))) + item.fallback = live_cache; + bool is_locked = locked(m, relative); + if (item.cache) item.cache->retain = is_locked; + const char *target; + char *original = NULL; + if (is_locked && item.live_target) target = item.live_target; + else if (item.cache) target = item.cache->target; + else if (item.fallback) target = item.fallback->target; + else { original = xprintf("%s/%s", m->cfg.original_root, relative); target = original; } + bool had_live_target = item.live_target != NULL; + bool publishing_cache = item.cache && strcmp(target, item.cache->target) == 0; + int changed = atomic_link(view, target); + if (changed < 0) { + logmsg("Unable to publish %s: %s", relative, strerror(errno)); + if (item.cache) item.cache->retain = true; + } else if (changed > 0) view_changed = true; + item_add(&m->items, item); + struct item *stored = &m->items.v[m->items.n - 1]; + if (changed > 0) { + free(stored->live_target); + stored->live_target = xstrdup(target); + if (had_live_target || publishing_cache) + (void)refresh(m, stored); + } + else if (item.cache && !is_locked && !cache_metadata_matches(item.catalog)) (void)refresh(m, stored); + free(original); + } + fts_close(tree); strings_sort(&seen); + + char *view_roots[] = {m->cfg.view_root, NULL}; + tree = fts_open(view_roots, FTS_PHYSICAL | FTS_NOCHDIR, NULL); + if (tree) { + size_t view_n = strlen(m->cfg.view_root); + while ((entry = fts_read(tree))) { + const char *relative = entry->fts_path + view_n; if (*relative == '/') relative++; + if ((entry->fts_info == FTS_SL || entry->fts_info == FTS_SLNONE) && *relative && + !strings_has(&seen, relative) && !locked(m, relative)) { + if (unlink(entry->fts_path) == 0) { + sync_parent(entry->fts_path); + view_changed = true; + } + } else if (entry->fts_info == FTS_DP && *relative) (void)rmdir(entry->fts_path); + } + fts_close(tree); + } + strings_free(&seen); + if (view_changed) { + json_object *reply = api(m, "/Library/Refresh", "POST"); + if (reply) json_object_put(reply); + } + mark_references(m); + + for (size_t i = 0; i < m->caches.n; i++) { + struct cache *cache = &m->caches.v[i]; + if (cache->referenced || cache->retain) continue; + struct item *item = *cache->path_hash ? item_by_path_hash(m, cache->path_hash) : NULL; + if (item && !cache_by_target(m, item->live_target) && !refresh(m, item)) { + cache->retain = true; + continue; + } + if (unlink(cache->path) == 0) sync_parent(cache->path); + cache->size = 0; + } + m->last_scan = now(); return true; +} + +static uint64_t cache_used(struct manager *m) +{ + uint64_t used = 0; + for (size_t i = 0; i < m->caches.n; i++) used += m->caches.v[i].size; + return used; +} + +static uint64_t cache_free(struct manager *m) +{ + struct statvfs fs; + if (statvfs(m->cfg.cache_root, &fs) < 0) return 0; + uint64_t physical = (uint64_t)fs.f_bavail * fs.f_frsize, used = cache_used(m); + uint64_t configured = used < m->cfg.cache_max ? m->cfg.cache_max - used : 0; + return physical < configured ? physical : configured; +} + +static int lru_cmp(const void *a, const void *b) +{ + const struct cache *const *x = a, *const *y = b; + return (*x)->used_at < (*y)->used_at ? -1 : (*x)->used_at > (*y)->used_at ? 1 : strcmp((*x)->name, (*y)->name); +} + +static bool evict(struct manager *m, uint64_t required, uint64_t target_used) +{ + struct cache **order = xmalloc(m->caches.n * sizeof(*order)); size_t n = 0; + for (size_t i = 0; i < m->caches.n; i++) if (m->caches.v[i].size) order[n++] = &m->caches.v[i]; + qsort(order, n, sizeof(*order), lru_cmp); + for (size_t i = 0; i < n && (cache_free(m) < required || cache_used(m) > target_used); i++) { + struct cache *cache = order[i]; + struct item *item = *cache->path_hash ? item_by_path_hash(m, cache->path_hash) : NULL; + if (cache->retain || (item && locked(m, item->relative))) continue; + if (cache->referenced && item) { + char *original = xprintf("%s/%s", m->cfg.original_root, item->relative); + if (atomic_link(item->view, original) < 0) { free(original); continue; } + free(original); cache->referenced = false; + if (!refresh(m, item)) { + cache->retain = true; + continue; + } + } else if (cache->referenced) continue; + logmsg("Evicting %s", item ? item->relative : cache->name); + if (unlink(cache->path) == 0) { sync_parent(cache->path); cache->size = 0; } + } + free(order); return cache_free(m) >= required && cache_used(m) <= target_used; +} + +static struct item *candidate(struct manager *m) +{ + struct item *best = NULL; bool best_priority = false; double best_order = 0, time = now(); + for (size_t i = 0; i < m->items.n; i++) { + struct item *item = &m->items.v[i]; + bool needs_cache = item->fallback || (item->catalog && incompatible(&item->catalog->media)); + if (item->cache || !needs_cache || locked(m, item->relative) || + timer_get(&m->retries, item->relative) > time) continue; + double priority = timer_get(&m->priorities, item->relative); + bool preferred = priority > 0; double order = preferred ? priority : item->mtime; + if (!best || preferred > best_priority || (preferred == best_priority && order > best_order)) { + best = item; best_priority = preferred; best_order = order; + } + } + return best; +} + +static void cancel_child(struct manager *m, const char *reason) +{ + if (m->child <= 0) return; + logmsg("Stopping encode: %s", reason); kill(-m->child, SIGTERM); + for (int i = 0; i < 100; i++) { + if (waitpid(m->child, NULL, WNOHANG) == m->child) { m->child = 0; return; } + struct timespec delay = {.tv_nsec = 100000000}; nanosleep(&delay, NULL); + } + kill(-m->child, SIGKILL); waitpid(m->child, NULL, 0); m->child = 0; +} + +static pid_t start_ffmpeg(struct manager *m, const struct item *item, const char *partial, const struct media *source) +{ + bool is_hdr = hdr(source); + const char *filter = !is_hdr ? "zscale=dither=error_diffusion,format=yuv420p" : + !strcasecmp(source->transfer, "arib-std-b67") ? + "zscale=tin=arib-std-b67:pin=bt2020:min=bt2020nc: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" : + "zscale=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"; + char *graph = xprintf("[0:v:0]%s[video]", filter); + char *args[64]; size_t n = 0; +#define A(x) do { if (n + 1 >= ARRAY_LEN(args)) die("ffmpeg argument overflow"); args[n++] = (char *)(x); } while (0) + A(m->cfg.ffmpeg); A("-nostdin"); A("-hide_banner"); A("-loglevel"); A("warning"); A("-i"); A(item->source); + A("-filter_complex"); A(graph); A("-map"); A("[video]"); A("-map"); A("0:a:0?"); + A("-map_metadata"); A("-1"); A("-map_chapters"); A("-1"); + A("-c:v"); A("libx264"); A("-preset"); A(m->cfg.preset); A("-crf"); A(m->cfg.crf); + A("-maxrate"); A(m->cfg.maxrate); A("-bufsize"); A(m->cfg.bufsize); A("-pix_fmt"); A("yuv420p"); A("-profile:v"); A("high"); + A("-c:a"); A("aac"); A("-ac:a"); A("2"); A("-b:a"); A(AUDIO_BITRATE); + if (is_hdr) { A("-color_primaries"); A("bt709"); A("-color_trc"); A("bt709"); A("-colorspace"); A("bt709"); A("-color_range"); A("tv"); } + if (!strcmp(item->extension, ".mp4")) { A("-movflags"); A("+faststart"); A("-f"); A("mp4"); } + else { A("-f"); A("matroska"); } + A(partial); args[n] = NULL; +#undef A + char *log = xprintf("%s/ffmpeg.log", m->cfg.state_root); + int fd = open(log, O_WRONLY | O_CREAT | O_APPEND | O_CLOEXEC, 0644); free(log); + if (fd < 0) return -1; + pid_t pid = fork(); + if (pid < 0) { close(fd); free(graph); return -1; } + if (!pid) { + if (setsid() < 0 || dup2(fd, STDOUT_FILENO) < 0 || dup2(fd, STDERR_FILENO) < 0) _exit(126); + close(fd); + background_or_exit(); execv(args[0], args); _exit(127); + } + close(fd); free(graph); return pid; +} + +static bool valid_output(const struct media *source, const struct media *output) +{ + double tolerance = source->duration * .01 > 3 ? source->duration * .01 : 3; + int expected_audio = source->audio ? 1 : 0; + return !strcasecmp(output->codec, "h264") && output->depth <= 8 && !output->dolby_vision && + (!source->duration || (output->duration >= source->duration - tolerance && output->duration <= source->duration + tolerance)) && + (!source->width || source->width == output->width) && (!source->height || source->height == output->height) && + output->audio == expected_audio && output->audio_compatible && !output->subtitles && + !output->attachments && !output->chapters; +} + +static int publish(struct manager *m, struct item *item) +{ + bool preferred = timer_get(&m->priorities, item->relative) > 0; + uint64_t estimate = (uint64_t)((long double)item->size * 2.2L); + if (estimate < 1000000000ULL) estimate = 1000000000ULL; + if (estimate > 100000000000ULL) estimate = 100000000000ULL; + if (cache_free(m) < estimate && (!preferred || !evict(m, estimate, m->cfg.cache_max))) return 2; + struct media source, output; + if (probe(m, item->source, &source) < 0) return -1; + if (!incompatible(&source)) { + char *original = xprintf("%s/%s", m->cfg.original_root, item->relative); + int changed = atomic_link(item->view, original); free(original); + if (changed < 0) return -1; + if (changed > 0) (void)refresh(m, item); + m->last_scan = 0; return changed; + } + char *root = xprintf("%s/files", m->cfg.cache_root); + char *partial = xprintf("%s/%s.%s.%s.%ld%s.partial", root, item->path_hash, + item->generation_hash, m->cfg.profile_hash, (long)getpid(), item->extension); + unlink(partial); logmsg("Encoding %s", item->relative); + m->child = start_ffmpeg(m, item, partial, &source); + if (m->child < 0) { free(root); free(partial); return -1; } + int status = 0; + while (!stopping) { + if (waitpid(m->child, &status, WNOHANG) == m->child) { m->child = 0; break; } + sleep(m->cfg.poll_seconds); + if (!read_sessions(m) || locked(m, item->relative)) { + cancel_child(m, "item or series is being watched"); unlink(partial); sync_directory(root); + free(root); free(partial); return 0; + } + } + if (stopping) { cancel_child(m, "service stopping"); unlink(partial); sync_directory(root); free(root); free(partial); return 0; } + if (!WIFEXITED(status) || WEXITSTATUS(status)) { unlink(partial); sync_directory(root); free(root); free(partial); return -1; } + struct stat st; + if (stat(partial, &st) < 0 || st.st_size < 1000000 || probe(m, partial, &output) < 0 || !valid_output(&source, &output)) { + unlink(partial); sync_directory(root); free(root); free(partial); return -1; + } + char output_hash[HASH_HEX + 1]; + if (background_file_hash(partial, output_hash) < 0) { unlink(partial); sync_directory(root); free(root); free(partial); return -1; } + int fd = open(partial, O_RDONLY | O_CLOEXEC | O_NOFOLLOW); + if (fd < 0 || fchmod(fd, 0444) < 0 || fsync(fd) < 0) { + if (fd >= 0) close(fd); + unlink(partial); sync_directory(root); free(root); free(partial); return -1; + } + close(fd); + char *name = xprintf("%s.%s.%s.%s%s", item->path_hash, item->generation_hash, + m->cfg.profile_hash, output_hash, item->extension); + char *final = xprintf("%s/%s", root, name); + if (syscall(SYS_renameat2, AT_FDCWD, partial, AT_FDCWD, final, RENAME_NOREPLACE) < 0) { + unlink(partial); free(name); free(final); free(root); free(partial); return -1; + } + if (sync_directory(root) < 0) { free(name); free(final); free(root); free(partial); return -1; } + if (locked(m, item->relative)) { unlink(final); sync_directory(root); free(name); free(final); free(root); free(partial); return 0; } + char *target = xprintf("%s/files/%s", m->cfg.jellyfin_cache_root, name); + if (atomic_link(item->view, target) < 0) { free(target); free(name); free(final); free(root); free(partial); return -1; } + (void)refresh(m, item); timer_del(&m->priorities, item->relative); timer_del(&m->retries, item->relative); + logmsg("Completed %s", item->relative); + free(target); free(name); free(final); free(root); free(partial); m->last_scan = 0; return 1; +} + +static void signal_handler(int signal_number) { (void)signal_number; stopping = 1; } + +static uint64_t number(const char *name, const char *value) +{ + char *end = NULL; errno = 0; unsigned long long n = strtoull(value, &end, 10); + if (errno || !end || *end) + die("invalid %s: %s", name, value); + return (uint64_t)n; +} + +static const char *environment_value(const char *name, const char *fallback) +{ + const char *value = getenv(name); + return value && *value ? value : fallback; +} + +static char *environment(const char *name, const char *fallback) +{ + const char *value = getenv(name); return xstrdup(value && *value ? value : fallback); +} + +static void config_load(struct config *c) +{ + memset(c, 0, sizeof(*c)); + c->source_root = environment("SOURCE_ROOT", "/source"); c->view_root = environment("VIEW_ROOT", "/view"); + c->cache_root = environment("CACHE_ROOT", "/cache"); c->state_root = environment("STATE_ROOT", "/state"); + c->jellyfin_url = environment("JELLYFIN_URL", "http://jellyfin:8096"); c->library_root = environment("JELLYFIN_LIBRARY_ROOT", "/data/library"); + c->original_root = environment("JELLYFIN_ORIGINAL_ROOT", "/data/original"); c->jellyfin_cache_root = environment("JELLYFIN_CACHE_ROOT", "/data/cache"); + c->api_key = environment("JELLYFIN_API_KEY", ""); c->ffmpeg = environment("FFMPEG", "/usr/bin/ffmpeg"); c->ffprobe = environment("FFPROBE", "/usr/bin/ffprobe"); + c->crf = environment("H264_CRF", "18"); c->preset = environment("H264_PRESET", "medium"); c->maxrate = environment("H264_MAXRATE", "15M"); c->bufsize = environment("H264_BUFSIZE", "30M"); + c->profile = xprintf("h264-high-8bit-aac-stereo-%s-nosubs-crf%s-%s-%s-%s-canonical-v1", + AUDIO_BITRATE, c->crf, c->preset, c->maxrate, c->bufsize); + hash_bytes(c->profile, strlen(c->profile), c->profile_hash); + c->cache_max = number("CACHE_MAX_BYTES", environment_value("CACHE_MAX_BYTES", "1000000000000")); + c->cooldown_seconds = (unsigned)number("SERIES_COOLDOWN_SECONDS", environment_value("SERIES_COOLDOWN_SECONDS", "3600")); + c->poll_seconds = (unsigned)number("POLL_SECONDS", environment_value("POLL_SECONDS", "5")); + c->rescan_seconds = (unsigned)number("RESCAN_SECONDS", environment_value("RESCAN_SECONDS", "300")); + if (!*c->api_key || !c->poll_seconds || !c->rescan_seconds) die("API key and positive intervals are required"); +} + +static void config_free(struct config *c) +{ + char *values[] = {c->source_root,c->view_root,c->cache_root,c->state_root,c->jellyfin_url,c->library_root,c->original_root,c->jellyfin_cache_root,c->api_key,c->ffmpeg,c->ffprobe,c->crf,c->preset,c->maxrate,c->bufsize,c->profile}; + for (size_t i = 0; i < ARRAY_LEN(values); i++) free(values[i]); +} + +static void manager_init(struct manager *m) +{ + memset(m, 0, sizeof(*m)); config_load(&m->cfg); + if (mkdirs(m->cfg.view_root) < 0 || mkdirs(m->cfg.state_root) < 0) die("unable to create working directories"); + char *marker = xprintf("%s/.pretranscode-cache", m->cfg.cache_root); struct stat st; + if (stat(marker, &st) < 0 || !S_ISREG(st.st_mode)) die("cache filesystem is not mounted at %s", m->cfg.cache_root); + free(marker); + m->cache_lock = open(m->cfg.cache_root, O_RDONLY | O_DIRECTORY | O_CLOEXEC); + if (m->cache_lock < 0 || flock(m->cache_lock, LOCK_EX | LOCK_NB) < 0) die("another controller owns the cache"); + struct sigaction action = {.sa_handler = signal_handler}; sigemptyset(&action.sa_mask); + sigaction(SIGTERM, &action, NULL); sigaction(SIGINT, &action, NULL); +} + +static void manager_free(struct manager *m) +{ + if (m->child > 0) cancel_child(m, "service stopping"); + strings_free(&m->active_items); strings_free(&m->active_keys); strings_free(&m->previous_items); strings_free(&m->previous_keys); + timers_free(&m->cooldowns); timers_free(&m->priorities); timers_free(&m->retries); + catalog_free(&m->catalog); caches_free(&m->caches); items_free(&m->items); + if (m->cache_lock >= 0) + close(m->cache_lock); + config_free(&m->cfg); +} + +static void run(struct manager *m) +{ + while (!stopping && !read_sessions(m)) sleep(m->cfg.poll_seconds); + if (stopping || !reconcile(m)) return; + (void)evict(m, 0, m->cfg.cache_max); + while (!stopping) { + if (!read_sessions(m)) { sleep(m->cfg.poll_seconds); continue; } + if (now() - m->last_scan >= m->cfg.rescan_seconds) { + if (!reconcile(m)) { sleep(m->cfg.poll_seconds); continue; } + (void)evict(m, 0, m->cfg.cache_max); + } + timers_prune(&m->cooldowns, now()); timers_prune(&m->retries, now()); + struct item *item = candidate(m); + if (!item) { sleep(m->cfg.poll_seconds); continue; } + uint64_t estimate = (uint64_t)((long double)item->size * 2.2L); + if (estimate < 1000000000ULL) estimate = 1000000000ULL; + if (estimate > 100000000000ULL) estimate = 100000000000ULL; + bool preferred = timer_get(&m->priorities, item->relative) > 0; + if (!preferred && cache_free(m) < estimate) { sleep(60); continue; } + int result = publish(m, item); + if (result < 0) { logmsg("Encode failed for %s", item->relative); timer_set(&m->retries, item->relative, now() + 21600); } + if (!stopping && m->last_scan == 0) (void)reconcile(m); + } +} + +#ifndef PRETRANSCODE_TEST +int main(int argc, char **argv) +{ + if (argc == 2 && !strcmp(argv[1], "--version")) { puts("pretranscode " VERSION); return 0; } + if (curl_global_init(CURL_GLOBAL_DEFAULT) != CURLE_OK) die("curl initialization failed"); + struct manager manager; manager_init(&manager); + const char *command = argc > 1 ? argv[1] : "run"; int result = EXIT_SUCCESS; + if (!strcmp(command, "run")) run(&manager); + else if (!strcmp(command, "sync")) { if (!read_sessions(&manager) || !reconcile(&manager)) result = EXIT_FAILURE; } + else if (!strcmp(command, "active")) result = !read_sessions(&manager) || manager.active_items.n ? EXIT_FAILURE : EXIT_SUCCESS; + else if (!strcmp(command, "purge")) { + if (!read_sessions(&manager) || manager.active_items.n || !reconcile(&manager)) result = EXIT_FAILURE; + else if (!evict(&manager, 0, 0)) result = EXIT_FAILURE; + } else if (!strcmp(command, "trim") && argc == 3) { + uint64_t target = number("trim size", argv[2]); target = target > 5000000000ULL ? target - 5000000000ULL : 0; + if (!read_sessions(&manager) || manager.active_items.n || !reconcile(&manager) || !evict(&manager, 0, target)) result = EXIT_FAILURE; + } else { fprintf(stderr, "usage: %s [run|sync|active|purge|trim BYTES|--version]\n", argv[0]); result = EXIT_FAILURE; } + manager_free(&manager); curl_global_cleanup(); return result; +} +#endif diff --git a/pretranscode.py b/pretranscode.py deleted file mode 100755 index 1896693..0000000 --- a/pretranscode.py +++ /dev/null @@ -1,908 +0,0 @@ -#!/usr/bin/env python3 -"""Immutable Jellyfin H.264 cache backed by an atomic symlink view.""" - -import hashlib -import json -import os -import shutil -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") -LEGACY_ENCODE_PROFILE = f"h264-high-8bit-crf{H264_CRF}-{H264_PRESET}-{H264_MAXRATE}-quality-v3" -ENCODE_PROFILE = f"h264-high-8bit-crf{H264_CRF}-{H264_PRESET}-{H264_MAXRATE}-quality-v4" -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 - _, manifest = self.cache_paths(relative) - try: - os.utime(manifest, None, 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:" - "stream_side_data=side_data_type,dv_profile,rpu_present_flag,el_present_flag,bl_present_flag:" - "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 - side_data = stream.get("side_data_list") or [] - 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(), - "dolby_vision": any( - "dovi" in (value.get("side_data_type") or "").lower() - for value in side_data - ), - "duration": float((data.get("format") or {}).get("duration") or 0), - } - - def source_info(self, relative): - source = SOURCE_ROOT / relative - details = self.probe(source) - previous = self.state["inventory"].get(relative) or {} - if previous.get("item_id"): - details["item_id"] = previous["item_id"] - details["fingerprint"] = fingerprint(source) - stat = source.stat() - details["size"] = stat.st_size - details["mtime"] = stat.st_mtime - self.state["inventory"][relative] = details - return details - - 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 and catalog[relative].get("item_id"): - old["item_id"] = catalog[relative]["item_id"] - continue - 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((manifest_path.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), - "-filter_complex", f"[0:v:0]{video_filter}[video]", - "-map", "[video]", "-map", "0:a?", "-map", "0:s?", "-map_metadata", "0", "-map_chapters", "0", - "-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") - if details.get("dolby_vision"): - raise RuntimeError("output incorrectly retains Dolby Vision stream metadata") - 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 video_stream_hash(self, path): - command = [ - FFMPEG, "-nostdin", "-hide_banner", "-loglevel", "error", "-i", str(path), - "-map", "0:v:0", "-c:v", "copy", "-bsf:v", "h264_mp4toannexb", - "-f", "hash", "-hash", "sha256", "-", - ] - result = subprocess.run(command, capture_output=True, text=True, timeout=3600) - if result.returncode: - raise RuntimeError(result.stderr.strip() or "unable to hash video stream") - return result.stdout.strip() - - def sanitize_cache_metadata(self, relative, output): - if output.suffix.lower() not in {".mkv", ".mp4"}: - raise RuntimeError(f"unsupported cache container for metadata repair: {output.suffix}") - stage = output.with_name(output.name + ".sanitize-video.mp4.partial") - repaired = output.with_name(output.name + ".sanitize.partial") - stage.unlink(missing_ok=True) - repaired.unlink(missing_ok=True) - original_hash = self.video_stream_hash(output) - log(f"Losslessly removing inherited Dolby Vision metadata from {relative}") - first = [ - FFMPEG, "-nostdin", "-hide_banner", "-loglevel", "warning", "-i", str(output), - "-map", "0:v:0", "-c:v", "copy", "-an", "-sn", "-dn", - "-map_metadata", "-1", "-map_chapters", "-1", "-movflags", "+faststart", - "-f", "mp4", str(stage), - ] - second = [ - FFMPEG, "-nostdin", "-hide_banner", "-loglevel", "warning", - "-i", str(stage), "-i", str(output), - "-map", "0:v:0", "-map", "1:a?", "-map", "1:s?", - "-map_metadata", "1", "-map_chapters", "1", "-c", "copy", - ] - if output.suffix.lower() == ".mkv": - second += ["-map", "1:t?", "-c:t", "copy", "-f", "matroska"] - else: - second += ["-movflags", "+faststart", "-f", "mp4"] - second.append(str(repaired)) - log_path = STATE_ROOT / "ffmpeg.log" - try: - with log_path.open("ab", buffering=0) as ffmpeg_log: - for command in (first, second): - result = subprocess.run(command, stdout=ffmpeg_log, stderr=ffmpeg_log) - if result.returncode: - raise RuntimeError(f"metadata repair ffmpeg exited with status {result.returncode}") - details = self.probe(repaired) - if details.get("dolby_vision"): - raise RuntimeError("metadata repair did not remove Dolby Vision stream metadata") - if self.video_stream_hash(repaired) != original_hash: - raise RuntimeError("metadata repair changed the encoded video stream") - os.replace(repaired, output) - finally: - stage.unlink(missing_ok=True) - repaired.unlink(missing_ok=True) - - def migrate_v4(self): - if self.active(): - raise RuntimeError("manifest migration refused while media is playing") - backup = STATE_FILE.with_name(f"state.json.pre-v4-{int(time.time())}") - if STATE_FILE.is_file(): - shutil.copy2(STATE_FILE, backup) - log(f"Backed up state to {backup}") - migrated = 0 - sanitized = 0 - skipped = 0 - for _, relative, output, manifest_path in self.cache_entries(): - manifest = load_json(manifest_path, {}) - if manifest.get("encode_profile") not in {LEGACY_ENCODE_PROFILE, ENCODE_PROFILE}: - log(f"Skipping manifest with unsupported profile: {relative}") - skipped += 1 - continue - source = SOURCE_ROOT / relative - if not source.is_file() or manifest.get("source_fingerprint") != fingerprint(source): - log(f"Skipping stale manifest: {relative}") - skipped += 1 - continue - source_details = self.source_info(relative) - output_details = self.probe(output) - if output_details.get("dolby_vision"): - self.sanitize_cache_metadata(relative, output) - manifest.pop("refresh_version", None) - sanitized += 1 - self.validate_output(source, output, source_details.get("duration")) - manifest["encode_profile"] = ENCODE_PROFILE - manifest["migration_version"] = 4 - manifest["migrated_at"] = time.time() - atomic_json(manifest_path, manifest) - migrated += 1 - self.save() - log(f"Manifest migration complete: migrated={migrated}, sanitized={sanitized}, skipped={skipped}") - - 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.source_info(relative) - 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:]) - elif command == "migrate-v4": - manager.migrate_v4() - else: - raise SystemExit(f"unknown command: {command}") - - -if __name__ == "__main__": - main() diff --git a/tests/__init__.py b/tests/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/tests/fake_jellyfin.py b/tests/fake_jellyfin.py new file mode 100755 index 0000000..7587ad0 --- /dev/null +++ b/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() diff --git a/tests/integration.sh b/tests/integration.sh index 364827a..b34a0e5 100755 --- a/tests/integration.sh +++ b/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" diff --git a/tests/test_core.c b/tests/test_core.c new file mode 100644 index 0000000..90e064b --- /dev/null +++ b/tests/test_core.c @@ -0,0 +1,159 @@ +#define PRETRANSCODE_TEST +#include "../pretranscode.c" + +#include + +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; +} diff --git a/tests/test_core.py b/tests/test_core.py deleted file mode 100644 index 3e9463e..0000000 --- a/tests/test_core.py +++ /dev/null @@ -1,163 +0,0 @@ -import importlib.util -import json -import os -import tempfile -import unittest -from pathlib import Path -from types import SimpleNamespace -from unittest import mock - - -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.assertTrue(any("zscale=dither=error_diffusion,format=yuv420p" in value for value in sdr)) - self.assertIn("-filter_complex", sdr) - self.assertIn("[video]", sdr) - self.assertNotIn("-vf", sdr) - self.assertIn("18", sdr) - self.assertTrue(any("tonemap=tonemap=hable" in value for value in hdr)) - self.assertIn("bt709", hdr) - - def test_probe_detects_dolby_vision_stream_metadata(self): - manager = self.module.Manager() - result = SimpleNamespace( - returncode=0, - stderr="", - stdout=json.dumps({ - "streams": [{ - "codec_name": "h264", - "profile": "High", - "pix_fmt": "yuv420p", - "bits_per_raw_sample": "8", - "color_transfer": "bt709", - "side_data_list": [{"side_data_type": "DOVI configuration record"}], - }], - "format": {"duration": "10.0"}, - }), - ) - with mock.patch.object(self.module.subprocess, "run", return_value=result): - details = manager.probe(Path("cached.mkv")) - self.assertTrue(details["dolby_vision"]) - - def test_validation_rejects_inherited_dolby_vision_metadata(self): - manager = self.module.Manager() - output = self.root / "cache" / "invalid.mkv" - output.write_bytes(b"x" * 1_000_001) - details = {"codec": "h264", "bit_depth": 8, "dolby_vision": True, "duration": 10.0} - with mock.patch.object(manager, "probe", return_value=details): - with self.assertRaisesRegex(RuntimeError, "Dolby Vision"): - manager.validate_output(Path("source.mkv"), output, 10.0) - - 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_lru_touches_only_manifest(self): - m = self.module - manager = m.Manager() - relative = "movies/LRU/LRU.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, - })) - os.utime(output, (100, 100)) - os.utime(manifest, (100, 100)) - manager.touch_cache(relative) - self.assertEqual(output.stat().st_mtime, 100) - self.assertGreater(manifest.stat().st_mtime, 100) - - 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()