Publish curated Unraid-AppOps-Gateway source from aaf7d11821a86be65c91fead97a6a4758c4fbacd

backend/tests/test_projectbrain_delivery_revalidation.py

@@ -5,6 +5,7 @@ import io
 import json
 import subprocess
 import sys
+import threading
 import time
 import uuid
 from concurrent.futures import ThreadPoolExecutor
@@ -186,7 +187,11 @@ def test_revalidate_rejects_stale_fingerprint_without_queueing(controller, monke
     )
 
     assert code == 409
-    assert result == {"status": "blocked", "reason": "projectbrain_delivery_preflight_changed"}
+    assert result == {
+        "status": "blocked",
+        "reason": "projectbrain_delivery_preflight_changed",
+        "noMutation": True,
+    }
     assert queued == []
 
 
@@ -876,6 +881,29 @@ def test_concurrent_enqueue_creates_one_additive_attempt_and_preserves_original(
         preflight, hashlib.sha256(b"pb-revalidate-concurrent-2026").hexdigest(),
     )
     project = fixture["project"]
+    marker = fixture["inbox"] / f"{fixture['semantic']}.revalidation.json"
+    writer_opened = threading.Event()
+    allow_writer_to_continue = threading.Event()
+    gate_lock = threading.Lock()
+    gate_used = False
+    real_open = controller.os.open
+
+    def gated_open(path, flags, mode=0o777):
+        nonlocal gate_used
+        descriptor = real_open(path, flags, mode)
+        name = Path(path).name
+        if name == marker.name or name.startswith(f".{marker.name}."):
+            with gate_lock:
+                pause = not gate_used
+                gate_used = True
+            if pause:
+                writer_opened.set()
+                if not allow_writer_to_continue.wait(timeout=5):
+                    controller.os.close(descriptor)
+                    raise TimeoutError("test writer gate timed out")
+        return descriptor
+
+    monkeypatch.setattr(controller.os, "open", gated_open)
 
     def enqueue():
         return controller._enqueue_terminal_revalidation(
@@ -887,14 +915,173 @@ def test_concurrent_enqueue_creates_one_additive_attempt_and_preserves_original(
         )
 
     with ThreadPoolExecutor(max_workers=2) as workers:
-        results = list(workers.map(lambda _index: enqueue(), range(2)))
+        first = workers.submit(enqueue)
+        assert writer_opened.wait(timeout=5)
+        try:
+            assert not marker.exists(), "an incomplete revalidation marker became visible"
+            second = workers.submit(enqueue)
+            assert second.result(timeout=5) == ("awaiting_validation", fixture["semantic"])
+        finally:
+            allow_writer_to_continue.set()
+        assert first.result(timeout=5) == ("awaiting_validation", fixture["semantic"])
 
-    assert results == [("awaiting_validation", fixture["semantic"])] * 2
     assert fixture["original_path"].read_bytes() == fixture["original_bytes"]
     attempt_files = list(fixture["inbox"].glob("*.revalidation.json"))
     assert len(attempt_files) == 1
     record = controller._read_inbox_record(attempt_files[0])
     assert record["projectbrain_delivery_revalidation"] == binding
+    assert not list(fixture["inbox"].glob(".*.tmp"))
+
+
+def test_new_inbox_record_ignores_stale_temporary_file(controller, tmp_path):
+    marker = tmp_path / "stale.json"
+    record = {"record": "complete"}
+    stale = marker.with_name(
+        f".{marker.name}.{controller.os.getpid()}.{controller.threading.get_ident()}.tmp"
+    )
+    stale_bytes = b"stale temporary file from an earlier process\n"
+    stale.write_bytes(stale_bytes)
+
+    controller._write_new_inbox_record(marker, record)
+
+    assert json.loads(marker.read_text(encoding="utf-8")) == record
+    if controller.os.name != "nt":
+        assert marker.stat().st_mode & 0o777 == 0o600
+    assert stale.read_bytes() == stale_bytes
+    assert set(tmp_path.iterdir()) == {marker, stale}
+
+
+def test_new_inbox_record_collision_preserves_existing_bytes_and_cleans_temp(
+    controller, tmp_path,
+):
+    marker = tmp_path / "existing.json"
+    original = b'{"existing":true}\n'
+    marker.write_bytes(original)
+
+    with pytest.raises(FileExistsError):
+        controller._write_new_inbox_record(marker, {"replacement": True})
+
+    assert marker.read_bytes() == original
+    assert not list(tmp_path.glob(".*.tmp"))
+
+
+def test_new_inbox_record_write_failure_cleans_temporary_file(
+    controller, monkeypatch, tmp_path,
+):
+    marker = tmp_path / "failed.json"
+    real_fdopen = controller.os.fdopen
+
+    class FailingWriter:
+        def __init__(self, handle):
+            self.handle = handle
+
+        def __enter__(self):
+            self.handle.__enter__()
+            return self
+
+        def __exit__(self, exc_type, exc, traceback):
+            return self.handle.__exit__(exc_type, exc, traceback)
+
+        def write(self, _payload):
+            raise OSError("simulated inbox write failure")
+
+    monkeypatch.setattr(
+        controller.os,
+        "fdopen",
+        lambda descriptor, mode: FailingWriter(real_fdopen(descriptor, mode)),
+    )
+
+    with pytest.raises(OSError, match="simulated inbox write failure"):
+        controller._write_new_inbox_record(marker, {"record": True})
+
+    assert not marker.exists()
+    assert not list(tmp_path.glob(".*.tmp"))
+
+
+@pytest.mark.parametrize("failure_point", ["directory_fsync", "wakeup"])
+def test_publish_followup_failure_reports_unknown_and_same_key_replay_resolves(
+    controller, monkeypatch, tmp_path, failure_point,
+):
+    fixture = _snapshot_fixture(controller, monkeypatch, tmp_path)
+    monkeypatch.setattr(controller, "_projectbrain_delivery_mutation_lock", _null_context)
+    preflight = controller._projectbrain_delivery_snapshot(fixture["revision"])
+    idempotency_key = "pb-revalidate-publish-failure-2026"
+    payload = {
+        "revision": fixture["revision"],
+        "expectedPreflightSha256": preflight["preflightSha256"],
+        "idempotencyKey": idempotency_key,
+    }
+    real_fsync = controller._fsync_inbox_root
+    real_wakeup = controller._webhook_wakeup
+
+    if failure_point == "directory_fsync":
+        def fail_fsync():
+            raise OSError("simulated directory fsync failure")
+
+        monkeypatch.setattr(controller, "_fsync_inbox_root", fail_fsync)
+    else:
+        def fail_wakeup():
+            raise RuntimeError("simulated worker wakeup failure")
+
+        monkeypatch.setattr(controller, "_webhook_wakeup", SimpleNamespace(set=fail_wakeup))
+
+    code, result = invoke(controller, "revalidate", payload)
+
+    assert code == 503
+    assert result == {
+        "status": "unknown",
+        "reason": "projectbrain_delivery_enqueue_outcome_unknown",
+        "noMutation": False,
+    }
+    assert fixture["original_path"].read_bytes() == fixture["original_bytes"]
+    marker = fixture["inbox"] / f"{fixture['semantic']}.revalidation.json"
+    record = controller._read_inbox_record(marker)
+    assert record is not None
+    assert record["schema"] == controller._WEBHOOK_REVALIDATION_SCHEMA
+    assert record["status"] == "awaiting_validation"
+    assert record["projectbrain_delivery_revalidation"]["idempotencyKeySha256"] == hashlib.sha256(
+        idempotency_key.encode("ascii"),
+    ).hexdigest()
+    assert not list(fixture["inbox"].glob(".*.tmp"))
+
+    if failure_point == "directory_fsync":
+        monkeypatch.setattr(controller, "_fsync_inbox_root", real_fsync)
+    else:
+        monkeypatch.setattr(controller, "_webhook_wakeup", real_wakeup)
+    replay_code, replay = invoke(controller, "revalidate", payload)
+
+    assert replay_code == 202
+    assert replay["status"] == "awaiting_validation"
+    assert replay["delivery"] == fixture["semantic"]
+    assert fixture["original_path"].read_bytes() == fixture["original_bytes"]
+
+
+def test_revalidation_preflight_refusal_reports_no_mutation(
+    controller, monkeypatch, tmp_path,
+):
+    fixture = _snapshot_fixture(controller, monkeypatch, tmp_path)
+    monkeypatch.setattr(controller, "_projectbrain_delivery_mutation_lock", _null_context)
+
+    def refuse_snapshot(_revision):
+        raise controller.ProjectBrainDeliveryRefused("projectbrain_delivery_preflight_changed")
+
+    monkeypatch.setattr(controller, "_projectbrain_delivery_snapshot", refuse_snapshot)
+    payload = {
+        "revision": fixture["revision"],
+        "expectedPreflightSha256": "a" * 64,
+        "idempotencyKey": "pb-revalidate-preflight-refusal-2026",
+    }
+
+    code, result = invoke(controller, "revalidate", payload)
+
+    assert code == 409
+    assert result == {
+        "status": "blocked",
+        "reason": "projectbrain_delivery_preflight_changed",
+        "noMutation": True,
+    }
+    assert fixture["original_path"].read_bytes() == fixture["original_bytes"]
+    assert not (fixture["inbox"] / f"{fixture['semantic']}.revalidation.json").exists()
 
 
 def test_generic_terminal_revalidation_cannot_occupy_projectbrain_sibling(