backend/tests/test_media_io_gate.py
Rama master · réplica de lectura del repositorio original.
"""Process-wide gate for heavy media subprocesses (remux, extract, sync).
Background: on 2026-09-04 a boot-time wanted search ran four foreign-track
remuxes of 7 GB files in parallel (81.8 GB rewritten in one hour) while the
automation queue read whole movies for ffsubsync. The array sat at load 14
with the XFS flush thread blocked for 46 minutes and ffprobe calls timed out.
One gate, one setting (``media_io_max_parallel``), every heavy launcher.
"""
from __future__ import annotations
import threading
import time
from unittest.mock import MagicMock, patch
import pytest
from flask import Flask
from services.media_io_gate import (
MediaGateBusyError,
MediaIOGate,
media_io_gate,
)
# ---------------------------------------------------------------------------
# The primitive
# ---------------------------------------------------------------------------
def test_limit_one_serialises_two_holders():
gate = MediaIOGate(limit=1)
order: list[str] = []
def first():
with gate.slot("a"):
order.append("a-in")
time.sleep(0.2)
order.append("a-out")
def second():
with gate.slot("b"):
order.append("b-in")
t1 = threading.Thread(target=first)
t1.start()
time.sleep(0.05)
t2 = threading.Thread(target=second)
t2.start()
t1.join()
t2.join()
assert order == ["a-in", "a-out", "b-in"]
def test_limit_two_lets_two_run_together():
gate = MediaIOGate(limit=2)
both_inside = threading.Barrier(2, timeout=2)
def worker():
with gate.slot("x"):
both_inside.wait()
threads = [threading.Thread(target=worker) for _ in range(2)]
for t in threads:
t.start()
for t in threads:
t.join(timeout=3)
assert not any(t.is_alive() for t in threads)
assert gate.in_use == 0
def test_acquire_timeout_returns_false_and_slot_raises_busy():
gate = MediaIOGate(limit=1)
assert gate.acquire(timeout=1) is True
try:
assert gate.acquire(timeout=0.05) is False
with pytest.raises(MediaGateBusyError), gate.slot("late", timeout=0.05):
pass
finally:
gate.release()
assert gate.in_use == 0
def test_slot_releases_on_exception():
gate = MediaIOGate(limit=1)
with pytest.raises(RuntimeError), gate.slot("boom"):
raise RuntimeError("boom")
assert gate.in_use == 0
assert gate.acquire(timeout=0.1) is True
gate.release()
def test_release_without_acquire_is_an_error():
gate = MediaIOGate(limit=1)
with pytest.raises(ValueError):
gate.release()
def test_raising_limit_wakes_a_waiter():
gate = MediaIOGate(limit=1)
gate.acquire(timeout=1)
got_in = threading.Event()
def waiter():
with gate.slot("w", timeout=5):
got_in.set()
t = threading.Thread(target=waiter)
t.start()
time.sleep(0.05)
assert not got_in.is_set()
gate.set_limit(2)
assert got_in.wait(timeout=2)
t.join(timeout=2)
gate.release()
assert gate.in_use == 0
def test_lowering_limit_does_not_evict_but_caps_new_acquires():
gate = MediaIOGate(limit=2)
gate.acquire(timeout=1)
gate.acquire(timeout=1)
gate.set_limit(1)
assert gate.in_use == 2
assert gate.acquire(timeout=0.05) is False
gate.release()
# still at the cap: one holder remains, limit is one
assert gate.acquire(timeout=0.05) is False
gate.release()
assert gate.acquire(timeout=0.1) is True
gate.release()
@pytest.mark.parametrize("bad", [0, -1])
def test_limit_below_one_is_clamped(bad):
gate = MediaIOGate(limit=bad)
assert gate.limit == 1
gate.set_limit(bad)
assert gate.limit == 1
def test_context_manager_protocol_matches_the_old_sync_lock():
"""``with gate:`` plus ``acquire(timeout=)``/``release()`` keep the sync
engines and the sync preview working unchanged."""
gate = MediaIOGate(limit=1)
with gate:
assert gate.in_use == 1
assert gate.in_use == 0
# ---------------------------------------------------------------------------
# Settings binding
# ---------------------------------------------------------------------------
def test_limit_follows_settings_on_acquire():
gate = MediaIOGate(limit=1)
settings = MagicMock(media_io_max_parallel=3)
with patch("services.media_io_gate.peek_settings", return_value=settings):
assert gate.acquire(timeout=0.1)
assert gate.limit == 3
gate.release()
def test_no_settings_yet_keeps_last_limit():
gate = MediaIOGate(limit=2)
with patch("services.media_io_gate.peek_settings", return_value=None):
assert gate.acquire(timeout=0.1)
assert gate.limit == 2
gate.release()
def test_cap_workers_never_exceeds_the_gate():
gate = MediaIOGate(limit=1)
with patch("services.media_io_gate.peek_settings", return_value=None):
assert gate.cap_workers(4) == 1
gate.set_limit(3)
assert gate.cap_workers(4) == 3
assert gate.cap_workers(2) == 2
assert gate.cap_workers(0) == 1
# ---------------------------------------------------------------------------
# Wait policy: an HTTP request never queues for long behind a remux
# ---------------------------------------------------------------------------
@pytest.fixture
def app():
return Flask("media-io-gate-test")
def test_default_wait_is_short_inside_a_request_and_long_outside(app):
gate = MediaIOGate(limit=1)
assert gate.default_wait_s() == gate.BACKGROUND_WAIT_S
with app.test_request_context("/api/v1/health"):
assert gate.default_wait_s() == gate.REQUEST_WAIT_S
assert gate.REQUEST_WAIT_S < gate.BACKGROUND_WAIT_S
def test_request_context_slot_fails_fast_when_busy(app, monkeypatch):
gate = MediaIOGate(limit=1)
monkeypatch.setattr(gate, "REQUEST_WAIT_S", 0.05)
gate.acquire(timeout=1)
try:
with app.test_request_context("/api/v1/health"):
started = time.monotonic()
with pytest.raises(MediaGateBusyError), gate.slot("ui"):
pass
assert time.monotonic() - started < 1
finally:
gate.release()
# ---------------------------------------------------------------------------
# Wiring: every heavy launcher goes through the module-level gate
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _pin_module_gate_limit():
"""The module-level gate reads settings on acquire; pin it so the wiring
tests do not depend on whatever the test app configured."""
with patch("services.media_io_gate.peek_settings", return_value=None):
media_io_gate.set_limit(1)
yield
def _fake_run_recording_gate(seen: list[int]):
def _run(*args, **kwargs):
seen.append(media_io_gate.in_use)
return MagicMock(returncode=0, stdout="", stderr="")
return _run
def test_mkvmerge_remux_runs_inside_the_gate():
from remux import _remux_mkvmerge
seen: list[int] = []
with patch("remux.subprocess.run", side_effect=_fake_run_recording_gate(seen)):
_remux_mkvmerge("/tmp/in.mkv", [3], "/tmp/out.mkv")
assert seen == [1]
assert media_io_gate.in_use == 0
def test_ffmpeg_remux_runs_inside_the_gate():
from remux import _remux_ffmpeg
seen: list[int] = []
with (
patch("remux._which", return_value=True),
patch("remux.subprocess.run", side_effect=_fake_run_recording_gate(seen)),
):
_remux_ffmpeg("/tmp/in.mkv", [3], "/tmp/out.mkv")
assert seen == [1]
assert media_io_gate.in_use == 0
def test_remux_reports_a_busy_gate_as_remux_error():
from remux import RemuxError, _remux_mkvmerge
media_io_gate.acquire(timeout=1)
try:
with (
patch.object(media_io_gate, "default_wait_s", return_value=0.05),
pytest.raises(RemuxError, match="busy"),
):
_remux_mkvmerge("/tmp/in.mkv", [3], "/tmp/out.mkv")
finally:
media_io_gate.release()
def test_extract_subtitle_stream_runs_inside_the_gate(tmp_path):
from ass_probe import extract_subtitle_stream
seen: list[int] = []
out = tmp_path / "ep.de.srt"
def _run(*args, **kwargs):
seen.append(media_io_gate.in_use)
# ffmpeg writes to the tempfile named last in argv
with open(args[0][-1], "w", encoding="utf-8") as fh:
fh.write("1\n00:00:01,000 --> 00:00:02,000\nhi\n")
return MagicMock(returncode=0, stdout="", stderr="")
with (
patch("ass_probe.subprocess.run", side_effect=_run),
patch("ass_probe.get_settings", return_value=MagicMock(ffmpeg_timeout=5)),
):
extract_subtitle_stream(str(tmp_path / "ep.mkv"), {"sub_index": 0}, str(out))
assert seen == [1]
assert media_io_gate.in_use == 0
assert out.exists()
def test_extract_reports_a_busy_gate_as_runtime_error(tmp_path):
from ass_probe import extract_subtitle_stream
media_io_gate.acquire(timeout=1)
try:
with (
patch.object(media_io_gate, "default_wait_s", return_value=0.05),
patch("ass_probe.get_settings", return_value=MagicMock(ffmpeg_timeout=5)),
pytest.raises(RuntimeError, match="busy"),
):
extract_subtitle_stream(
str(tmp_path / "ep.mkv"), {"sub_index": 0}, str(tmp_path / "o.srt")
)
finally:
media_io_gate.release()
assert not list(tmp_path.iterdir()), "no temp file may survive a refused extraction"
def test_sync_lock_is_the_media_gate():
from services.sync_engines.concurrency import sync_subprocess_lock
assert sync_subprocess_lock is media_io_gate
def test_scanner_batch_probe_workers_are_capped_by_the_gate():
from services import wanted_item_scanner
settings = MagicMock(scan_metadata_max_workers=4, media_io_max_parallel=1)
with (
patch("services.wanted_item_scanner.get_settings", return_value=settings),
patch("services.media_io_gate.peek_settings", return_value=settings),
patch("services.wanted_item_scanner.get_media_streams", return_value={"streams": []}),
patch("services.wanted_item_scanner.ThreadPoolExecutor") as pool,
patch("services.wanted_item_scanner.as_completed", return_value=[]),
):
wanted_item_scanner.batch_probe(["/m/a.mkv", "/m/b.mkv"])
pool.assert_called_once_with(max_workers=1)
def test_route_batch_probe_workers_are_capped_by_the_gate():
from routes.wanted import batch_probe as route
settings = MagicMock(scan_metadata_max_workers=4, media_io_max_parallel=2)
with (
patch("routes.wanted.batch_probe.get_settings", return_value=settings),
patch("services.media_io_gate.peek_settings", return_value=settings),
patch("routes.wanted.batch_probe.ThreadPoolExecutor") as pool,
patch("routes.wanted.batch_probe._init_batch_probe_state"),
patch("routes.wanted.batch_probe._finalize_batch_probe"),
patch("routes.wanted.batch_probe.as_completed", return_value=[]),
):
route._run_batch_probe([], MagicMock())
pool.assert_called_once_with(max_workers=2)
def test_waiter_notices_a_raised_setting_without_a_release():
"""Codex review finding: a thread already asleep in the wait loop must see
a raised ``media_io_max_parallel`` even if nobody releases or touches the
gate afterwards."""
gate = MediaIOGate(limit=1)
settings = MagicMock(media_io_max_parallel=1)
with patch("services.media_io_gate.peek_settings", return_value=settings):
gate.acquire(timeout=1)
got_in = threading.Event()
def waiter():
with gate.slot("w", timeout=10):
got_in.set()
t = threading.Thread(target=waiter)
t.start()
time.sleep(0.1)
assert not got_in.is_set()
settings.media_io_max_parallel = 2 # nobody calls set_limit, nobody releases
assert got_in.wait(timeout=3)
t.join(timeout=2)
gate.release()
assert gate.in_use == 0
def test_non_blocking_acquire_with_timeout_is_rejected_like_a_semaphore():
gate = MediaIOGate(limit=1)
with pytest.raises(ValueError):
gate.acquire(blocking=False, timeout=1)
assert gate.acquire(blocking=False) is True
assert gate.acquire(blocking=False) is False
gate.release()
def test_request_wait_is_bounded_well_below_the_gunicorn_thread_budget():
"""Four request threads; a holder that keeps the gate for minutes must not
be able to park them for long."""
assert MediaIOGate.REQUEST_WAIT_S <= 10
def test_probe_batches_run_each_probe_inside_the_gate():
"""Codex review finding: sizing each pool by the limit still lets two
concurrent batches exceed it. Every batch probe holds a slot."""
from services import wanted_item_scanner
settings = MagicMock(scan_metadata_max_workers=2, media_io_max_parallel=1)
seen: list[int] = []
def _probe(path, use_cache):
seen.append(media_io_gate.in_use)
return {"streams": []}
with (
patch("services.wanted_item_scanner.get_settings", return_value=settings),
patch("services.media_io_gate.peek_settings", return_value=settings),
patch("services.wanted_item_scanner.get_media_streams", side_effect=_probe),
):
out = wanted_item_scanner.batch_probe(["/m/a.mkv", "/m/b.mkv", "/m/c.mkv"])
assert len(out) == 3
assert seen == [1, 1, 1]
assert media_io_gate.in_use == 0
def test_route_batch_probe_submits_gated_probes():
from routes.wanted import batch_probe as route
settings = MagicMock(scan_metadata_max_workers=2, media_io_max_parallel=1)
submitted: list = []
class _Pool:
def __init__(self, max_workers):
self.max_workers = max_workers
def __enter__(self):
return self
def __exit__(self, *a):
return False
def submit(self, fn, *args):
submitted.append((fn, args))
return MagicMock()
with (
patch("routes.wanted.batch_probe.get_settings", return_value=settings),
patch("services.media_io_gate.peek_settings", return_value=settings),
patch("routes.wanted.batch_probe.ThreadPoolExecutor", _Pool),
patch("routes.wanted.batch_probe._init_batch_probe_state"),
patch("routes.wanted.batch_probe._finalize_batch_probe"),
patch("routes.wanted.batch_probe.as_completed", return_value=[]),
):
route._run_batch_probe([{"file_path": "/m/a.mkv"}], MagicMock())
assert len(submitted) == 1
fn, args = submitted[0]
assert fn is route._gated_probe
assert args == ("/m/a.mkv",)
def test_sync_preview_waits_less_than_the_request_budget():
from services import sync_preview
assert sync_preview._LOCK_WAIT_TIMEOUT <= MediaIOGate.REQUEST_WAIT_S
def test_refresh_publishes_the_configured_limit_to_the_gauges():
"""rc.8 on RC read ``sublarr_media_io_gate_limit 0``: the gauges are only
set when the gate module is first imported, which happens lazily on the
first heavy subprocess. Startup must publish the configured limit."""
gate = MediaIOGate(limit=1)
in_use, limit = MagicMock(), MagicMock()
gate.wire_gauges(in_use, limit)
settings = MagicMock(media_io_max_parallel=2)
with patch("services.media_io_gate.peek_settings", return_value=settings):
gate.refresh_from_settings()
assert gate.limit == 2
limit.set.assert_called_with(2)
in_use.set.assert_called_with(0)
def test_create_app_refreshes_the_gate_from_settings(temp_db):
from app import create_app
with patch.object(media_io_gate, "refresh_from_settings") as refresh:
create_app(testing=True)
refresh.assert_called_once_with()
def test_setting_exists_with_a_conservative_default_and_is_in_the_media_view():
from config_settings import UISettings
from config_views import MediaServerSettings
assert UISettings.model_fields["media_io_max_parallel"].default == 1
assert "media_io_max_parallel" in MediaServerSettings._fields
def test_a_stopped_job_stops_waiting_for_the_gate():
"""Prod 2026-09-24/25: wanted_search sat up to 3600 s behind ffsubsync and
was recorded ``timeout_abandoned`` — it "did NOT stop when asked"."""
from services.scheduler import cancellation
gate = MediaIOGate(limit=1)
assert gate.acquire(timeout=1) is True
stop = threading.Event()
stop.set()
token = cancellation.activate(stop)
try:
started = time.monotonic()
with pytest.raises(MediaGateBusyError), gate.slot("remux", timeout=30):
pass
assert time.monotonic() - started < 5
finally:
cancellation.deactivate(token)
gate.release()
def test_the_with_form_still_waits_through_a_stop_request():
"""The bare ``with`` form cannot tell a refusal from a slot; a stop must
not make it run without one and then release someone else's slot."""
from services.scheduler import cancellation
gate = MediaIOGate(limit=1)
stop = threading.Event()
stop.set()
token = cancellation.activate(stop)
try:
assert gate.acquire(timeout=1) is True
releaser = threading.Timer(0.3, gate.release)
releaser.start()
with gate:
assert gate.in_use == 1
releaser.join()
finally:
cancellation.deactivate(token)
assert gate.in_use == 0