From eaad2828fa3327e72e0f0c0435d3147601b1c415 Mon Sep 17 00:00:00 2001 From: hoelee Date: Sun, 11 Oct 2026 14:08:08 +0800 Subject: [PATCH] gate_policy: read-modify-write the park counters (an owner edit must survive a running engine) Measured 2026-10-11: an item parked at 2 rounds was released by the owner (park entry deleted) so it could be retried at a lower crf - and the very next failure of ANOTHER item restored the entry, from the stale in-memory snapshot of the engine that had loaded the file before the edit. The released item was then SKIPped as parked. save_policy now re-reads the file and writes only the counters this process bumped (POLICY_DIRTY), so a deleted park entry or an added accept entry survives. Regression: scripts/test_policy_merge.py (4 checks: released item stays released, external accept entry kept, counter reaches disk, and the old behaviour would have restored it). --- scripts/library_run.py | 17 +++++++++++- scripts/test_policy_merge.py | 54 ++++++++++++++++++++++++++++++++++++ 2 files changed, 70 insertions(+), 1 deletion(-) create mode 100644 scripts/test_policy_merge.py diff --git a/scripts/library_run.py b/scripts/library_run.py index f28b345..b197566 100644 --- a/scripts/library_run.py +++ b/scripts/library_run.py @@ -172,15 +172,29 @@ def load_policy(): def save_policy(p): + """Persist ONLY the counters this process bumped, merged into the file as it is on disk now. + + Measured 2026-10-11: save_policy used to write its whole in-memory snapshot, so an edit made while + an engine was running - a `park` entry deleted to release an item for a lower-crf retry, or an + `accept` entry added - was silently reverted by the next failure of any OTHER item. The released + item was then SKIPped as parked although the owner had just released it. Read-modify-write: the + on-disk file keeps winning for every key we are not deliberately updating. + """ try: + disk = load_policy() + park = disk["park"] + for k, v in POLICY_DIRTY.items(): + park[k] = v + POLICY_CACHE["accept"], POLICY_CACHE["park"] = disk["accept"], park with open(POLICY, "w", encoding="utf-8") as fh: - json.dump({"accept": sorted(p["accept"]), "park": p["park"]}, fh, + json.dump({"accept": sorted(disk["accept"]), "park": park}, fh, ensure_ascii=False, indent=1) except OSError as e: log(" ! could not save %s: %s" % (POLICY, e)) POLICY_CACHE = load_policy() +POLICY_DIRTY = {} # disp key -> rounds, ONLY for counters this process incremented def probe(p, tries=3): @@ -1248,6 +1262,7 @@ def main(): # is persisted: without it every pass re-encoded the same eight items forever. k = disp(f) POLICY_CACHE["park"][k] = POLICY_CACHE["park"].get(k, 0) + 1 + POLICY_DIRTY[k] = POLICY_CACHE["park"][k] save_policy(POLICY_CACHE) log(" -> quality round %d/%d failed for this item%s" % (POLICY_CACHE["park"][k], PARK_AFTER, diff --git a/scripts/test_policy_merge.py b/scripts/test_policy_merge.py new file mode 100644 index 0000000..c6eccd5 --- /dev/null +++ b/scripts/test_policy_merge.py @@ -0,0 +1,54 @@ +"""Regression: an owner edit to gate_policy.json made while an engine is running must survive. + +Measured 2026-10-11: the cpu lane's engine (started 13:35, loaded the policy at start) rewrote +gate_policy.json when its own item failed at ~13:5x, restoring the `park` entry the owner had deleted +at 13:47 to release that item for a lower-crf retry. The item was then SKIPped as parked. + +Checks + 1. the released entry stays released when another item's counter is bumped + 2. an externally added `accept` entry is preserved (the engine never manages that list) + 3. the bumped counter does reach disk + 4. the old write-the-whole-snapshot behaviour WOULD have restored it (so the test is not vacuous) +""" +import json +import os +import sys +import tempfile + +sys.path.insert(0, r"D:/dev/universal-video-transcode/scripts") +import library_run as LR # noqa: E402 + +tmp = tempfile.mkdtemp(prefix="gatepolicy_") +LR.POLICY = os.path.join(tmp, "gate_policy.json") +with open(LR.POLICY, "w", encoding="utf-8") as fh: + json.dump({"accept": ["itemX"], "park": {"itemA": 2, "itemB": 1}}, fh, ensure_ascii=False) + +# the engine process loads the policy at start (this is its in-memory snapshot) +LR.POLICY_CACHE.update(LR.load_policy()) +LR.POLICY_DIRTY.clear() + +# the owner, while that engine runs: release itemA, release itemY to the 0.990 tier +on_disk = json.load(open(LR.POLICY, encoding="utf-8")) +on_disk["park"].pop("itemA") +on_disk["accept"].append("itemY") +with open(LR.POLICY, "w", encoding="utf-8") as fh: + json.dump(on_disk, fh, ensure_ascii=False) + +# now the engine's itemB fails -> counter bump + save +LR.POLICY_CACHE["park"]["itemB"] = LR.POLICY_CACHE["park"].get("itemB", 0) + 1 +LR.POLICY_DIRTY["itemB"] = LR.POLICY_CACHE["park"]["itemB"] +LR.save_policy(LR.POLICY_CACHE) + +after = json.load(open(LR.POLICY, encoding="utf-8")) +old_behaviour = {"accept": ["itemX"], "park": {"itemA": 2, "itemB": 2}} # what the old code wrote + +checks = { + "released_item_stays_released": "itemA" not in after["park"], + "external_accept_entry_kept": "itemY" in after["accept"], + "counter_reaches_disk": after["park"].get("itemB") == 2, + "old_behaviour_would_have_restored_it": "itemA" in old_behaviour["park"], +} +print("on disk after the save:", json.dumps(after, ensure_ascii=False, sort_keys=True)) +for k, v in checks.items(): + print("%-40s %s" % (k, "PASS" if v else "FAIL")) +sys.exit(0 if all(checks.values()) else 1)