| … | … | @@ -0,0 +1,86 @@ |
| 1 | """The ingest queue survives more than one job finishing at once. |
| 2 | |
| 3 | Found by a flaky test, but it is not a test problem. _set_job_state truncated |
| 4 | the queue file and then wrote it, so a concurrent reader could catch it empty -- |
| 5 | "Expecting value: line 1 column 1 (char 0)". _load_job_queue catches everything |
| 6 | and returns {} by design, so a torn read does not fail loudly: it reports an |
| 7 | empty queue, and the next write persists that. Every job on record, gone. |
| 8 | |
| 9 | Two jobs finishing together was enough to hit it. A season finishing episode |
| 10 | after episode makes it routine, which is what made this worth fixing rather |
| 11 | than retrying. |
| 12 | |
| 13 | Run: .venv/bin/python3 -m pytest test_queue_atomicity.py -q |
| 14 | """ |
| 15 | |
| 16 | import json |
| 17 | import os |
| 18 | import threading |
| 19 | |
| 20 | import pytest |
| 21 | |
| 22 | import addon |
| 23 | |
| 24 | |
| 25 | @pytest.fixture(autouse=True) |
| 26 | def queue_file(tmp_path, monkeypatch): |
| 27 | monkeypatch.setattr(addon, "INGEST_QUEUE_FILE", |
| 28 | os.path.join(str(tmp_path), "queue.json")) |
| 29 | monkeypatch.setattr(addon, "_mp4_jobs", {}) |
| 30 | yield |
| 31 | |
| 32 | |
| 33 | def test_concurrent_writers_do_not_lose_each_others_entries(): |
| 34 | """Read-modify-write without a lock: two threads both load the queue, both |
| 35 | add their own key, and the second write drops the first.""" |
| 36 | def write(n): |
| 37 | addon._set_job_state(f"key{n}", "done", args=("series", f"tmdb:{n}", 1, 1, 720)) |
| 38 | |
| 39 | threads = [threading.Thread(target=write, args=(n,)) for n in range(24)] |
| 40 | for t in threads: |
| 41 | t.start() |
| 42 | for t in threads: |
| 43 | t.join(10) |
| 44 | assert len(addon._load_job_queue()) == 24 |
| 45 | |
| 46 | |
| 47 | def test_a_reader_never_sees_a_half_written_queue(): |
| 48 | """The failure the flaky test kept hitting, reproduced deliberately.""" |
| 49 | addon._set_job_state("seed", "done", args=("series", "tmdb:1", 1, 1, 720)) |
| 50 | errors = [] |
| 51 | stop = threading.Event() |
| 52 | |
| 53 | def reader(): |
| 54 | while not stop.is_set(): |
| 55 | try: |
| 56 | with open(addon.INGEST_QUEUE_FILE) as f: |
| 57 | json.load(f) |
| 58 | except FileNotFoundError: |
| 59 | pass # the rename window, not a torn read |
| 60 | except Exception as exc: |
| 61 | errors.append(exc) |
| 62 | |
| 63 | def writer(): |
| 64 | for n in range(150): |
| 65 | addon._set_job_state(f"k{n}", "running", args=("series", "x", 1, n, 720)) |
| 66 | |
| 67 | r = threading.Thread(target=reader, daemon=True) |
| 68 | r.start() |
| 69 | w = threading.Thread(target=writer) |
| 70 | w.start() |
| 71 | w.join(30) |
| 72 | stop.set() |
| 73 | r.join(5) |
| 74 | assert errors == [], f"reader saw a partial file: {errors[:3]}" |
| 75 | |
| 76 | |
| 77 | def test_the_queue_is_still_readable_afterwards(): |
| 78 | addon._set_job_state("k", "done", args=("series", "tmdb:1", 1, 1, 720)) |
| 79 | assert addon._load_job_queue()["k"]["args"] == ["series", "tmdb:1", 1, 1, 720] |
| 80 | |
| 81 | |
| 82 | def test_no_temp_file_is_left_behind(): |
| 83 | addon._set_job_state("k", "done", args=("series", "tmdb:1", 1, 1, 720)) |
| 84 | leftovers = [f for f in os.listdir(os.path.dirname(addon.INGEST_QUEUE_FILE)) |
| 85 | if f.endswith(".tmp")] |
| 86 | assert leftovers == [] |