OverviewFilesCommitsIssuesDocsReleasesPeers

stremio-vsembed commits / f425000

Ingest queue survives a restart (0.12.0)
Job state was in-memory only, so a restart lost every intent: a copy mid-flight never resumed, and because a partial writes to <dest>.part while the row keys off the finished file, nothing surfaced the loss. Rolling forward makes that worse -- it queues work nobody is waiting on, so nobody notices it vanished. _set_job_state persists state AND the job arguments, since a key alone cannot be re-run. _resume_ingests runs before the server binds. Two deliberate asymmetries: - A running job is NOT rehydrated as "running". _start_mp4_job refuses a key already marked running, so pre-setting it would silently prevent the very restart this exists to perform. The restarted job sets its own state. - Terminal states ARE rehydrated, or the next request for an already-copied title starts a duplicate ingest of something on disk. A failed job is not retried on boot -- a failure that recurs every start would launch a browser each time, and the user can still ask for it by playing the row. Nothing is deleted to recover: an orphaned .part is reused, because ffmpeg runs with -y and writes the same path. Persistence is never load-bearing -- a missing, corrupt or unwritable queue file degrades to in-memory behaviour rather than refusing to boot. 13 new tests, 130 pass. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
f425000f2f3e086f68f1d6ca95596c6d0b100577
by plankatron <fleeky@gmail.com> · 2026-08-11T15:03:51.000Z

Changed files (2) +214 -3

addon.py +73 -3
……@@ -533,7 +533,7 @@ TV_GENRES = [
533533
534534MANIFEST = {
535535 "id": "org.tabertron.vsembed",
536 "version": "0.11.0",
536 "version": "0.12.0",
537537 "name": "Tabertron VsEmbed",
538538 "description": "Self-hosted HLS extractor with TMDB browse + search, played via vidcore",
539539 "types": ["movie", "series"],
……@@ -1065,6 +1065,72 @@ _mp4_jobs: dict[str, str] = {} # key -> "running" | "done" | "failed: .
10651065_mp4_jobs_guard = threading.Lock()
10661066_mp4_sem = threading.Semaphore(2) # concurrent remuxes
10671067
1068# Job state was in-memory only, so a restart lost every intent: a copy that was
1069# mid-flight never resumed, and since a partial writes to `<dest>.part` while the
1070# row keys off the finished file, nothing surfaced the loss. Rolling forward
1071# makes that worse — it queues work nobody is waiting on, so nobody notices.
1072#
1073# Nothing is deleted to recover: an orphaned .part is simply reused, because
1074# ffmpeg runs with -y and writes the same path.
1075INGEST_QUEUE_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)),
1076 "ingest_queue.json")
1077
1078
1079def _load_job_queue() -> dict:
1080 """Persisted job states. Never raises — this is a convenience, and a bad
1081 file must not be a reason to refuse to boot."""
1082 try:
1083 with open(INGEST_QUEUE_FILE) as f:
1084 data = json.load(f)
1085 return data if isinstance(data, dict) else {}
1086 except Exception:
1087 return {}
1088
1089
1090def _set_job_state(key: str, state: str, args: Optional[tuple] = None) -> None:
1091 """Record a job's state in memory and on disk.
1092
1093 `args` is what makes a resume possible: a key alone cannot be re-run.
1094 """
1095 _mp4_jobs[key] = state
1096 try:
1097 queue = _load_job_queue()
1098 entry = queue.get(key) or {}
1099 entry["state"] = state
1100 if args is not None:
1101 entry["args"] = list(args)
1102 entry["ts"] = time.time()
1103 queue[key] = entry
1104 with open(INGEST_QUEUE_FILE, "w") as f:
1105 json.dump(queue, f)
1106 except Exception as exc: # persistence is never load-bearing
1107 log.warning("ingest queue save failed: %s", exc)
1108
1109
1110def _resume_ingests() -> None:
1111 """Rehydrate job state at startup and restart anything left mid-flight.
1112
1113 Finished jobs are rehydrated too, not just running ones: without that, the
1114 next request for an already-copied title starts a duplicate ingest.
1115
1116 A FAILED job is not retried. A failure that recurs every boot would launch a
1117 browser on each one; the user can still ask for it by playing the row.
1118 """
1119 for key, entry in _load_job_queue().items():
1120 state = entry.get("state") or ""
1121 if state != "running":
1122 _mp4_jobs[key] = state
1123 continue
1124 # Deliberately NOT rehydrated as "running": _start_mp4_job refuses a key
1125 # already marked running, so pre-setting it here would silently prevent
1126 # the very restart this function exists to perform. The job sets it.
1127 args = entry.get("args")
1128 if not args or len(args) != 5:
1129 log.info("ingest resume: %s has no arguments — skipped", key)
1130 continue
1131 log.info("ingest resume: %s", key)
1132 _start_mp4_job(*args)
1133
10681134
10691135def _mp4_key(kind: str, ident: str, season: int, episode: int, height: int) -> str:
10701136 return f"{kind}:{ident}:{season}:{episode}:{height}"
……@@ -1323,7 +1389,9 @@ def _start_mp4_job(kind: str, ident: str, season: int, episode: int, height: int
13231389 state = _mp4_jobs.get(key)
13241390 if state == "running":
13251391 return state
1326 _mp4_jobs[key] = "running"
1392 # The arguments are persisted alongside the state so a restart mid-remux
1393 # can pick the job up again — a key on its own cannot be re-run.
1394 _set_job_state(key, "running", args=(kind, ident, season, episode, height))
13271395
13281396 def run() -> None:
13291397 acquired = _mp4_sem.acquire(timeout=MP4_REMUX_TIMEOUT)
……@@ -1337,7 +1405,7 @@ def _start_mp4_job(kind: str, ident: str, season: int, episode: int, height: int
13371405 if acquired:
13381406 _mp4_sem.release()
13391407 with _mp4_jobs_guard:
1340 _mp4_jobs[key] = state
1408 _set_job_state(key, state)
13411409
13421410 threading.Thread(target=run, name=f"mp4-{key}", daemon=True).start()
13431411 return "running"
……@@ -1853,4 +1921,6 @@ if __name__ == "__main__":
18531921 parser.add_argument("--host", default="0.0.0.0")
18541922 parser.add_argument("--port", type=int, default=7220)
18551923 args = parser.parse_args()
1924 # Pick up anything a previous process left mid-flight before serving.
1925 _resume_ingests()
18561926 app.run(host=args.host, port=args.port, debug=False, threaded=True)
test_ingest_queue.py +141 -0
……@@ -0,0 +1,141 @@
1"""The ingest queue survives a restart.
2
3_mp4_jobs was in-memory only, so restarting the addon lost every job intent: a
4copy that was mid-flight simply never resumed, and because a partial writes to
5`<dest>.part` while the row keys off the finished file, nothing surfaced the
6loss. The rolling library makes that worse -- it queues work the user never
7explicitly asked for, so nobody is waiting to notice it vanished.
8
9Nothing is deleted to recover. An orphaned .part is reused: ffmpeg runs with -y
10and writes the same path, so a resumed ingest overwrites it in place.
11
12Run: .venv/bin/python3 -m pytest test_ingest_queue.py -q
13"""
14
15import json
16import os
17
18import pytest
19
20import addon
21
22
23@pytest.fixture(autouse=True)
24def queue_file(tmp_path, monkeypatch):
25 monkeypatch.setattr(addon, "INGEST_QUEUE_FILE",
26 os.path.join(str(tmp_path), "ingest_queue.json"))
27 monkeypatch.setattr(addon, "_mp4_jobs", {})
28 monkeypatch.setattr(addon, "LIBRARY_ROOT", str(tmp_path / "lib"))
29 yield
30
31
32def _load():
33 with open(addon.INGEST_QUEUE_FILE) as f:
34 return json.load(f)
35
36
37# --- persistence ------------------------------------------------------------
38
39def test_a_job_state_is_written_to_disk():
40 addon._set_job_state("series:tmdb:1:1:1:720", "running")
41 assert _load()["series:tmdb:1:1:1:720"]["state"] == "running"
42
43
44def test_a_state_change_is_persisted():
45 addon._set_job_state("k", "running")
46 addon._set_job_state("k", "done")
47 assert _load()["k"]["state"] == "done"
48
49
50def test_the_job_arguments_are_persisted_so_it_can_be_restarted():
51 """A key alone cannot be re-run -- the resume path needs the arguments."""
52 addon._set_job_state("k", "running", args=("series", "tmdb:1", 1, 1, 720))
53 assert _load()["k"]["args"] == ["series", "tmdb:1", 1, 1, 720]
54
55
56def test_a_missing_queue_file_is_not_an_error():
57 assert addon._load_job_queue() == {}
58
59
60def test_a_corrupt_queue_file_is_not_an_error():
61 """State is a convenience, never a reason to refuse to boot."""
62 with open(addon.INGEST_QUEUE_FILE, "w") as f:
63 f.write("{not json")
64 assert addon._load_job_queue() == {}
65
66
67def test_an_unwritable_queue_file_does_not_break_the_job(monkeypatch):
68 monkeypatch.setattr(addon, "INGEST_QUEUE_FILE", "/nonexistent/dir/q.json")
69 addon._set_job_state("k", "running") # must not raise
70 assert addon._mp4_jobs["k"] == "running"
71
72
73# --- resume -----------------------------------------------------------------
74
75@pytest.fixture
76def resumed(monkeypatch):
77 started = []
78 monkeypatch.setattr(addon, "_start_mp4_job",
79 lambda *a, **k: started.append(tuple(a)) or "running")
80 return started
81
82
83def test_a_job_interrupted_mid_flight_is_restarted(resumed):
84 addon._set_job_state("k", "running", args=("series", "tmdb:1", 1, 1, 720))
85 addon._resume_ingests()
86 assert resumed == [("series", "tmdb:1", 1, 1, 720)]
87
88
89def test_a_finished_job_is_not_restarted(resumed):
90 addon._set_job_state("k", "done", args=("series", "tmdb:1", 1, 1, 720))
91 addon._resume_ingests()
92 assert resumed == []
93
94
95def test_a_failed_job_is_not_restarted(resumed):
96 """A failure that recurs every boot would launch a browser each time. The
97 user can still ask for it explicitly by playing the row."""
98 addon._set_job_state("k", "failed: could not resolve title",
99 args=("series", "tmdb:1", 1, 1, 720))
100 addon._resume_ingests()
101 assert resumed == []
102
103
104def test_a_job_with_no_arguments_is_skipped_rather_than_guessed(resumed):
105 addon._set_job_state("k", "running")
106 addon._resume_ingests()
107 assert resumed == []
108
109
110def test_resuming_survives_a_corrupt_file(resumed):
111 with open(addon.INGEST_QUEUE_FILE, "w") as f:
112 f.write("nonsense")
113 addon._resume_ingests() # must not raise
114 assert resumed == []
115
116
117def test_a_resume_rehydrates_finished_state_and_restarts_live_ones(resumed):
118 """A finished job must be remembered, or the next request for it starts a
119 duplicate ingest of something already on disk.
120
121 A running one is deliberately NOT rehydrated as "running" -- _start_mp4_job
122 refuses a key already marked running, so pre-setting it would silently
123 prevent the restart. The restarted job sets its own state.
124 """
125 addon._set_job_state("done-one", "done", args=("series", "tmdb:1", 1, 1, 720))
126 addon._set_job_state("live-one", "running", args=("series", "tmdb:1", 1, 2, 720))
127 addon._mp4_jobs.clear() # simulate a fresh process
128 addon._resume_ingests()
129 assert addon._mp4_jobs["done-one"] == "done"
130 assert resumed == [("series", "tmdb:1", 1, 2, 720)]
131 assert "live-one" not in addon._mp4_jobs
132
133
134# --- the real job path writes state ----------------------------------------
135
136def test_start_mp4_job_records_its_arguments(monkeypatch):
137 """So that a restart mid-remux can pick it up again."""
138 monkeypatch.setattr(addon, "_run_ingest", lambda *a, **k: "done")
139 addon._start_mp4_job("series", "tmdb:1", 1, 1, 720)
140 entry = _load()[addon._mp4_key("series", "tmdb:1", 1, 1, 720)]
141 assert entry["args"] == ["series", "tmdb:1", 1, 1, 720]