From 3dcd40bd1db8694b5843e034c0333f1d1e9b0924 Mon Sep 17 00:00:00 2001
From: ECHO OMEGA PRIME <echoomegaprime@gmail.com>
Date: Tue, 10 Nov 2026 20:23:03 -0600
Subject: [PATCH] test(subscriber): prove two concurrent orphan recoverers
 compensate once

---
 tests/test_dispatch_orphan_concurrency.py | 43 +++++++++++++++++++++++
 1 file changed, 43 insertions(+)
 create mode 100644 tests/test_dispatch_orphan_concurrency.py

diff --git a/tests/test_dispatch_orphan_concurrency.py b/tests/test_dispatch_orphan_concurrency.py
new file mode 100644
index 0000000..13fa345
--- /dev/null
+++ b/tests/test_dispatch_orphan_concurrency.py
@@ -0,0 +1,43 @@
+"""A concurrent orphan sweep must compensate a metered run exactly once.
+
+This regression supplements the sequential orphan recovery coverage from PR #49.
+It uses an isolated SQLite fixture; no production service or Telegram/OAuth integration.
+"""
+from concurrent.futures import ThreadPoolExecutor
+from threading import Barrier
+import sqlite3
+
+from echo_certification_forge.models import RunState
+from test_dispatch_orphaned_runs import _run, _orphan_the_old_way, _usage
+
+
+def test_two_simultaneous_recoverers_compensate_once(tmp_path, manifest):
+    store, governance, org, run_id = _run(
+        tmp_path, manifest, "probe-concurrent-recovery", max_attempts=3
+    )
+    _orphan_the_old_way(governance, org, run_id)
+    barrier = Barrier(3)
+
+    def recover(label):
+        barrier.wait(timeout=8)
+        return governance.claim_dispatch(label, organization_id=org.organization_id)
+
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        one = pool.submit(recover, "probe-recoverer-a")
+        two = pool.submit(recover, "probe-recoverer-b")
+        barrier.wait(timeout=8)
+        results = [one.result(timeout=20), two.result(timeout=20)]
+
+    assert results == [None, None]
+    assert store.get_run(run_id, org.organization_id)["state"] == RunState.INFRASTRUCTURE_FAILURE.value
+    usage = _usage(governance, org)
+    assert usage["meters"]["certification_runs"] == 0
+    assert usage["active_run_reservations"] == 0
+    with sqlite3.connect(governance.db_path) as connection:
+        compensations = connection.execute(
+            "SELECT count(*) FROM subscriber_usage_events "
+            "WHERE organization_id = ? AND meter = 'certification_runs' "
+            "AND quantity = -1 AND idempotency_key LIKE 'dispatch-crash:%'",
+            (org.organization_id,),
+        ).fetchone()[0]
+    assert compensations == 1, f"expected one compensation, got {compensations}"