fix(dub): purge order was hash-dependent, and the cap could evict a live marker (#1271)

main went red on my own test. Two distinct defects, both mine.

1. `targets = set(job_ids)` made iteration order depend on PYTHONHASHSEED, so
   which markers a cap-forced trim discarded was luck. That is why the test
   passed locally and failed in CI — verified: the old code passes at seeds
   0/7/42 and fails at 12345. Now a de-duplicated list in caller order.

2. The size cap could evict markers the CURRENT purge had just recorded. Those
   are the newest and the likeliest to still be held by a running job, so
   dropping one is precisely the resurrection this mechanism exists to prevent.
   A 'clear history' larger than the cap forced exactly that. The cap now never
   touches the current purge, making the real bound cap + one purge — stated
   plainly rather than implied.

Tests now pin both: identical survivors across runs, and an oversized purge
keeping all of its own markers. Verified across six hash seeds; full suite green
under 12345, the seed that reddened main.

backend/services/dub_pipeline.py

@@ -257,11 +257,19 @@ def merge_job(job_id: str, updates: dict) -> bool:
         return True
 
 
-def _expire_withdrawn(now: float) -> None:
+def _expire_withdrawn(now: float, protected: int = 0) -> None:
     """Drop withdrawal markers that are too old to still matter.
 
     Caller must hold ``_dub_jobs_lock``. Age first — that is the policy — then
-    the count cap purely so the mapping cannot grow without bound.
+    a size cap purely so the mapping cannot grow without bound.
+
+    ``protected`` is how many markers the current purge just recorded. Those sit
+    at the end (newest) and are never evicted by the size cap: they are the most
+    likely to still be held by a running job, and dropping one is exactly the
+    resurrection this whole mechanism exists to prevent. A single "clear
+    history" larger than the cap would otherwise force us to discard live
+    markers — which is what broke CI. The bound therefore is
+    ``cap + one purge``, not ``cap``.
     """
     cutoff = now - _WITHDRAWN_TTL_S
     while _withdrawn_jobs:
@@ -269,7 +277,8 @@ def _expire_withdrawn(now: float) -> None:
         if deleted_at >= cutoff:
             break
         _withdrawn_jobs.popitem(last=False)
-    while len(_withdrawn_jobs) > _WITHDRAWN_MAX:
+    floor = max(_WITHDRAWN_MAX, protected)
+    while len(_withdrawn_jobs) > floor:
         _withdrawn_jobs.popitem(last=False)
 
 
@@ -367,17 +376,20 @@ def purge_jobs(job_ids, *, delete_rows, include_inflight: bool = False) -> None:
     """
     with _dub_jobs_lock:
         delete_rows()
-        targets = set(job_ids)
+        # Deterministic order, de-duplicated. A `set` here made which markers
+        # the size cap evicts depend on PYTHONHASHSEED — the test for this very
+        # behaviour passed locally and failed in CI for that reason alone.
+        targets = list(dict.fromkeys(job_ids))
         if include_inflight:
             # "Clear history" means everything, including a job whose first row
             # hasn't been written yet — it wouldn't appear in `job_ids` at all.
-            targets |= _inflight_jobs
+            targets += [j for j in sorted(_inflight_jobs) if j not in set(targets)]
         now = time.monotonic()
         for job_id in targets:
             _dub_jobs.pop(job_id, None)
             _withdrawn_jobs.pop(job_id, None)
             _withdrawn_jobs[job_id] = now  # most-recent last
-        _expire_withdrawn(now)
+        _expire_withdrawn(now, protected=len(targets))
 
 
 def save_job(job_id: str, job: dict, filename: str = "", duration: float = 0.0, content_hash: str = "") -> None:

tests/test_dub_job_deleted_mid_ingest.py

@@ -391,7 +391,11 @@ def test_clearing_a_large_history_mid_render_does_not_evict_the_running_job(
 
     # One dub is mid-render; the user clears a history far larger than any
     # count cap would keep.
-    everything = ["rendering"] + [f"old{i}" for i in range(5000)]
+    # Deliberately larger than the size cap, so the cap is forced to choose —
+    # which is exactly the situation that evicted a live marker before.
+    everything = ["rendering"] + [
+        f"old{i}" for i in range(dub_pipeline._WITHDRAWN_MAX + 500)
+    ]
     dub_pipeline.purge_jobs(everything, delete_rows=lambda: None)
 
     dub_pipeline.save_job("rendering", {"filename": "a.mp4"})
@@ -502,3 +506,35 @@ def test_the_atomic_helpers_still_work_through_the_reentrant_path(monkeypatch):
     assert dub_pipeline.put_and_save_job("job1", {"filename": "a.mp4"}) is True
     assert dub_pipeline.merge_and_save_job("job1", {"scene_cuts": [1.0]}) is True
     assert written == ["job1", "job1"]
+
+
+def test_eviction_order_does_not_depend_on_hash_seed():
+    """A `set` of target ids made eviction order depend on PYTHONHASHSEED — so
+    which markers survived a cap-forced trim was luck. The test for the
+    behaviour above passed locally and reddened main for that reason alone.
+
+    Same input, same surviving markers, every time.
+    """
+    ids = [f"j{i}" for i in range(50)]
+
+    dub_pipeline._withdrawn_jobs.clear()
+    dub_pipeline.purge_jobs(ids, delete_rows=lambda: None)
+    first = list(dub_pipeline._withdrawn_jobs)
+
+    dub_pipeline._withdrawn_jobs.clear()
+    dub_pipeline.purge_jobs(ids, delete_rows=lambda: None)
+    assert list(dub_pipeline._withdrawn_jobs) == first == ids
+
+
+def test_a_purge_larger_than_the_cap_keeps_all_of_its_own_markers():
+    """The cap must never discard a marker the current purge just recorded:
+    those are the newest and the likeliest to still be held. The bound is
+    therefore `cap + one purge`, which is the honest guarantee."""
+    dub_pipeline._withdrawn_jobs.clear()
+    oversized = [f"x{i}" for i in range(dub_pipeline._WITHDRAWN_MAX + 750)]
+
+    dub_pipeline.purge_jobs(oversized, delete_rows=lambda: None)
+
+    assert len(dub_pipeline._withdrawn_jobs) == len(oversized)
+    for job_id in oversized:
+        assert job_id in dub_pipeline._withdrawn_jobs