Publish curated Unraid-AppOps-Gateway source from ca3aabdc7d315ddd47fd180b2d10b5691adb406d

PUBLIC_SOURCE_EXPORT.md

@@ -1,3 +1,3 @@
 # Public source export
 
-This source snapshot was generated from the private canonical repository at revision `b2bb670dd165f2979e421ffc34de833438b2fd85`. It contains no private Git history or operational evidence. Changes are published from the canonical repository.
+This source snapshot was generated from the private canonical repository at revision `ca3aabdc7d315ddd47fd180b2d10b5691adb406d`. It contains no private Git history or operational evidence. Changes are published from the canonical repository.

PUBLIC_SOURCE_MANIFEST.json

@@ -1,6 +1,6 @@
 {
   "schemaVersion": 1,
-  "sourceRevision": "b2bb670dd165f2979e421ffc34de833438b2fd85",
+  "sourceRevision": "ca3aabdc7d315ddd47fd180b2d10b5691adb406d",
   "files": [
     {
       "path": ".dockerignore",
@@ -1644,8 +1644,8 @@
     },
     {
       "path": "backend/tests/test_runner_fleet_specialize.py",
-      "sha256": "7cdb734b4596f2bdbd4703166792808a0183985c7165d9d8c51bdb54119f83d9",
+      "sha256": "af16899c98919cdf9238315be2dc8d400e283990e716f766fac10ead0663ec79",
-      "bytes": 145449
+      "bytes": 158831
     },
     {
       "path": "backend/tests/test_runner_fleet_v2_execution_surface.py",

backend/tests/test_runner_fleet_specialize.py

@@ -1767,11 +1767,21 @@ def owned_orphan(specialize, guard_transaction, tmp_path, monkeypatch):
     monkeypatch.setattr(specialize, "_validate_base_composition", lambda _source: "v1")
     monkeypatch.setattr(specialize, "_runner_rows", lambda: [{"id": 9}])
     monkeypatch.setattr(specialize, "_validate_legacy_registrations", lambda *_args: [{"id": 9}])
-    monkeypatch.setattr(specialize, "_validate_legacy_containers", lambda *_args: ("v1", [{"id": "fixed"}]))
+    legacy_item = {"Id": "a" * 64, "Image": "sha256:" + "b" * 64, "Created": "fixed",
+                   "Name": "/fixed", "Path": "/fixed-entrypoint", "Args": [],
+                   "Config": {"Labels": {}}, "HostConfig": {"RestartPolicy": {"Name": "unless-stopped"}},
+                   "Mounts": [], "NetworkSettings": {"Networks": {}}, "RestartCount": 0,
+                   "State": {"Running": True, "Paused": False, "Restarting": False, "Dead": False,
+                             "Pid": 123, "StartedAt": "fixed", "FinishedAt": "zero", "ExitCode": 0}}
+    monkeypatch.setattr(specialize, "_validate_legacy_containers", lambda *_args: ({"fixed": copy.deepcopy(legacy_item)}, []))
     monkeypatch.setitem(sys.modules, "deploy", SimpleNamespace(load_projects=lambda: {"fixed": "registry"}))
     monkeypatch.setattr(quiescence, "_read_receipt", lambda: copy.deepcopy(receipt))
     monkeypatch.setattr(quiescence, "_idle", lambda *_args: events.append("idle"))
-    def attest(snapshot, *, baseline, progress):
+    def attest(snapshot, *, baseline=None, progress=None):
+        if baseline is None:
+            assert snapshot == receipt["baseline"]
+            events.append("original-attested")
+            return
         assert baseline == receipt["baseline"] and progress == receipt["progress"]
         assert snapshot["rows"] == baseline["rows"]
         events.append("attested")
@@ -1855,7 +1865,13 @@ def test_orphan_recovery_refuses_unowned_or_changed_state_without_write(speciali
         monkeypatch.setattr(q, "_read_receipt", changed)
     elif mutation == "container":
         reads = []
-        monkeypatch.setattr(specialize, "_validate_legacy_containers", lambda *_args: reads.append(True) or ("v1", [{"id": str(len(reads))}]))
+        original = specialize._validate_legacy_containers
+        def changing(*args):
+            reads.append(True)
+            items, summaries = original(*args)
+            items["fixed"]["Id"] = str(len(reads))
+            return items, summaries
+        monkeypatch.setattr(specialize, "_validate_legacy_containers", changing)
     else:
         monkeypatch.setattr(specialize, "_atomic_write_owned", lambda *_args, **_kwargs: (_ for _ in ()).throw(
             specialize.RunnerFleetSpecializeError("runner_fleet_specialization_compare_and_swap_failed")))
@@ -1867,6 +1883,220 @@ def test_orphan_recovery_refuses_unowned_or_changed_state_without_write(speciali
     assert (runner.read_bytes(), host.read_bytes()) == before
 
 
+@pytest.fixture
+def completed_orphan_engine(specialize):
+    import runner_acceptance_quiescence as q
+    rows = copy.deepcopy(q._fixed_rows())
+    for row in rows:
+        row["ownerLabelDigest"] = "b" * 64 if row["ownerLabelPresent"] else None
+        row["labelDigest"] = q._digest(row["composeLabels"])
+    origins = {}
+    for project, claim in q.ORIGINS.items():
+        attempt = claim["attempt"] if claim["attempt"] is not None else 0
+        origins[project] = {"repository": "jens/workmesh", "run": claim["run"], "head": claim["head"],
+                            "status": "completed", "conclusion": "failure", "attempt": attempt,
+                            "task": claim["task"], "taskHead": claim["head"], "taskAttempt": attempt, "jobsTerminal": True}
+    project = next(iter(q.ORIGINS))
+    networks = [{"Id": f"{index + 1:064x}", "Name": f"{project}_net{index}", "Created": "fixed",
+                 "Driver": "bridge", "Containers": {}, "Labels": {"com.docker.compose.project": project}}
+                for index in range(8)]
+    volumes = [{"Name": f"{project}_vol{index}", "CreatedAt": "fixed", "Driver": "local", "Options": None,
+                "Labels": {"com.docker.compose.project": project},
+                "Mountpoint": f"{q.ROOTLESS_STORAGE}/volumes/{project}_vol{index}/_data"} for index in range(10)]
+    resources = {"networks": [{"id": item["Id"], "name": item["Name"], "created": item["Created"],
+                               "project": project, "driver": "bridge", "ownerDigest": q._digest(None)} for item in networks],
+                 "volumes": [{"name": item["Name"], "created": item["CreatedAt"], "project": project,
+                              "driver": "local", "ownerDigest": q._digest(None)} for item in volumes],
+                 "attachmentsVerified": True, "ownershipVerified": True}
+    baseline = {"rows": rows, "origins": origins, "resources": resources,
+                "daemon": {"legacyAttested": True, "engineId": "fixed-engine", "imageId": "sha256:" + "a" * 64,
+                           "outerId": "a" * 64, "created": "original", "labelsDigest": "a" * 64,
+                           "dockerRootDir": q.ROOTLESS_STORAGE, "rootless": True, "mounts": [], "networks": {}},
+                "productBinding": {"exact": "production"}, "guardBinding": {"exact": "guards"},
+                "productExclusionVerified": True, "dispatchBarrierVerified": True}
+    snapshot = copy.deepcopy(baseline)
+    snapshot["rows"] = []
+    snapshot["resources"] = {"networks": [], "volumes": [], "attachmentsVerified": True, "ownershipVerified": True}
+    snapshot["daemon"].update(outerId="c" * 64, created="recreated", labelsDigest="c" * 64)
+
+    class Driver:
+        def __init__(self):
+            self.ps = ""
+            self.effects = []
+            self.inspect_mutation = None
+        def assert_barrier(self):
+            return None
+        def _run(self, args, **kwargs):
+            assert args[:7] == ["docker", "exec", "-e", "DOCKER_HOST=tcp://127.0.0.1:2375", specialize.DIND_NAME, "docker", args[6]]
+            command = args[6:]
+            if command == ["ps", "-a", "--no-trunc", "-q"]:
+                return self.ps
+            kind = command[0]
+            assert command in (["network", "ls", "--quiet", "--no-trunc"], ["volume", "ls", "--quiet"])
+            return "\n".join(item["Id" if kind == "network" else "Name"] for item in (networks if kind == "network" else volumes))
+        def _docker(self, kind, action, *ids):
+            assert action == "inspect"
+            result = [copy.deepcopy(item) for item in (networks if kind == "network" else volumes)
+                      if item["Id" if kind == "network" else "Name"] in ids]
+            return self.inspect_mutation(result) if self.inspect_mutation else result
+    return q, baseline, snapshot, networks, volumes, Driver
+
+
+def test_completed_orphan_admission_requires_empty_whole_engine_and_preserved_resources(specialize, completed_orphan_engine):
+    q, baseline, snapshot, networks, volumes, driver_type = completed_orphan_engine
+    before = copy.deepcopy(baseline)
+    # Include unrelated resources: complete inventories cannot assume an empty
+    # resource store when the container engine is empty.
+    networks.append({"Id": "f" * 64, "Name": "unrelated", "Labels": {}})
+    for index in range(634):
+        volumes.append({"Name": f"unrelated{index}", "Labels": {}})
+    driver = driver_type()
+    specialize._attest_completed_orphan_engine(driver, snapshot, {"baseline": baseline}, q)
+    assert baseline == before and driver.effects == [] and len(volumes) == 644
+    with pytest.raises(q.QuiescenceError, match="project_inventory_changed"):
+        q.attest_snapshot(snapshot, baseline=baseline)
+
+
+@pytest.mark.parametrize("field", ["engineId", "imageId", "dockerRootDir", "rootless", "legacyAttested", "mounts", "networks"])
+def test_completed_orphan_refuses_stable_daemon_changes(specialize, completed_orphan_engine, field):
+    q, baseline, snapshot, *_rest = completed_orphan_engine
+    snapshot["daemon"][field] = "foreign"
+    with pytest.raises(specialize.RunnerFleetSpecializeError):
+        specialize._attest_completed_orphan_snapshot(snapshot, baseline, q)
+
+
+@pytest.mark.parametrize("field", ["origins", "productBinding", "guardBinding", "productExclusionVerified", "dispatchBarrierVerified", "rows", "resources"])
+def test_completed_orphan_refuses_changed_proof(specialize, completed_orphan_engine, field):
+    q, baseline, snapshot, *_rest = completed_orphan_engine
+    snapshot[field] = [{"foreign": True}] if field == "rows" else {}
+    with pytest.raises((specialize.RunnerFleetSpecializeError, q.QuiescenceError)):
+        specialize._attest_completed_orphan_snapshot(snapshot, baseline, q)
+
+
+@pytest.mark.parametrize("mutation", ["network_id", "network_name", "network_created", "attachment", "volume_name", "volume_created",
+                                     "mountpoint", "options", "driver", "owner", "project", "missing", "extra_origin", "lost_labels"])
+def test_completed_orphan_refuses_resource_drift(specialize, completed_orphan_engine, mutation):
+    q, baseline, _snapshot, networks, volumes, _driver = completed_orphan_engine
+    if mutation == "attachment":
+        networks[0]["Containers"] = {"a" * 64: {}}
+    elif mutation == "missing":
+        volumes.pop()
+    elif mutation == "extra_origin":
+        extra = copy.deepcopy(volumes[0])
+        extra["Name"] += "extra"
+        volumes.append(extra)
+    elif mutation == "lost_labels":
+        networks[0]["Labels"] = {}
+    elif mutation in {"owner", "project"}:
+        volumes[0]["Labels"][q.OWNER_LABEL if mutation == "owner" else "com.docker.compose.project"] = "foreign"
+    else:
+        item, key = {"network_id": (networks[0], "Id"), "network_name": (networks[0], "Name"),
+                     "network_created": (networks[0], "Created"), "volume_name": (volumes[0], "Name"),
+                     "volume_created": (volumes[0], "CreatedAt"), "mountpoint": (volumes[0], "Mountpoint"),
+                     "options": (volumes[0], "Options"), "driver": (volumes[0], "Driver")}[mutation]
+        item[key] = "foreign"
+    with pytest.raises(specialize.RunnerFleetSpecializeError):
+        specialize._orphan_resource_projection(networks, volumes, baseline, q)
+
+
+@pytest.mark.parametrize("mutation", ["foreign_container", "omitted_inspect", "duplicate_inspect", "duplicate_inventory", "oversize_inventory", "new_container_during_inspect"])
+def test_completed_orphan_rejects_incomplete_or_racing_whole_inventory(specialize, completed_orphan_engine, mutation):
+    q, baseline, snapshot, networks, volumes, driver_type = completed_orphan_engine
+    driver = driver_type()
+    if mutation == "foreign_container":
+        driver.ps = "f" * 64
+    elif mutation == "omitted_inspect":
+        driver.inspect_mutation = lambda items: items[:-1]
+    elif mutation == "duplicate_inspect":
+        driver.inspect_mutation = lambda items: items[:1] * len(items)
+    elif mutation == "duplicate_inventory":
+        networks.append(copy.deepcopy(networks[0]))
+    elif mutation == "oversize_inventory":
+        volumes.extend({"Name": f"unrelated{index}"} for index in range(991))
+    else:
+        def race(items):
+            driver.ps = "f" * 64
+            return items
+        driver.inspect_mutation = race
+    with pytest.raises(specialize.RunnerFleetSpecializeError):
+        specialize._attest_completed_orphan_engine(driver, snapshot, {"baseline": baseline}, q)
+    assert driver.effects == []
+
+
+@pytest.mark.parametrize("change", [None, "origin", "resource", "container", "idle"])
+def test_empty_engine_owned_orphan_recovery_rechecks_every_admission_before_cas(
+        specialize, owned_orphan, completed_orphan_engine, monkeypatch, change):
+    q, receipt, runner, host, originals, orphan, events, driver_type = owned_orphan
+    _q, baseline, snapshot, networks, _volumes, engine_type = completed_orphan_engine
+    receipt["baseline"] = copy.deepcopy(baseline)
+    engine = engine_type()
+    snapshots = []
+    def current(self):
+        self._guard_binding()
+        snapshots.append(True)
+        if len(snapshots) == 2:
+            if change == "origin":
+                snapshot["origins"][next(iter(q.ORIGINS))]["head"] = "f" * 40
+            elif change == "resource":
+                networks[0]["Created"] = "replaced"
+            elif change == "container":
+                engine.ps = "f" * 64
+        return copy.deepcopy(snapshot)
+    monkeypatch.setattr(driver_type, "snapshot", current)
+    monkeypatch.setattr(driver_type, "_run", lambda self, *args, **kwargs: engine._run(*args, **kwargs), raising=False)
+    monkeypatch.setattr(driver_type, "_docker", lambda self, *args: engine._docker(*args), raising=False)
+    idle = q._idle
+    def recheck_idle(*args):
+        if change == "idle" and events.count("idle") == 2:
+            raise q.QuiescenceError("global_ci_work_present")
+        idle(*args)
+    monkeypatch.setattr(q, "_idle", recheck_idle)
+    write = specialize._atomic_write_owned
+    writes = []
+    def cas(*args, **kwargs):
+        assert change is None and len(snapshots) == 2 and events.count("idle") == 3
+        assert events.count("original-attested") == 2
+        writes.append(True)
+        return write(*args, **kwargs)
+    monkeypatch.setattr(specialize, "_atomic_write_owned", cas)
+    before = copy.deepcopy(receipt)
+    if change is None:
+        specialize._recover_owned_orphan_guard()
+        assert runner.read_bytes() == originals[0] and writes == [True]
+        specialize._recover_owned_orphan_guard()
+        assert writes == [True]
+    else:
+        with pytest.raises(specialize.RunnerFleetSpecializeError):
+            specialize._recover_owned_orphan_guard()
+        assert runner.read_bytes() == orphan and writes == []
+    assert receipt == before and host.read_bytes() == originals[1] and engine.effects == []
+
+
+def test_orphan_legacy_identity_ignores_only_health_observations(specialize, owned_orphan):
+    q, *_rest = owned_orphan
+    containers = specialize._validate_legacy_containers("v1")
+    containers[0]["fixed"]["State"]["Health"] = {"Status": "healthy", "FailingStreak": 0, "Log": []}
+    before = specialize._orphan_legacy_identity(containers, q)
+    containers[0]["fixed"]["State"]["Health"]["Log"] = [{"Start": "new", "End": "new"}]
+    assert specialize._orphan_legacy_identity(containers, q) == before
+    for field in ("Id", "Image", "Created", "Name", "Path", "Args", "Config", "HostConfig", "Mounts", "NetworkSettings", "RestartCount"):
+        changed = copy.deepcopy(containers)
+        if field == "Mounts":
+            changed[0]["fixed"][field] = [{"Destination": "/foreign", "Type": "bind", "Source": "/foreign"}]
+        elif field == "NetworkSettings":
+            changed[0]["fixed"][field] = {"Networks": {"foreign": {"NetworkID": "f" * 64}}}
+        else:
+            changed[0]["fixed"][field] = "foreign"
+        assert specialize._orphan_legacy_identity(changed, q) != before
+    changed = copy.deepcopy(containers)
+    changed[0]["fixed"]["State"]["Health"]["Status"] = "unhealthy"
+    assert specialize._orphan_legacy_identity(changed, q) != before
+    for field in ("Running", "Paused", "Restarting", "Dead", "Pid", "StartedAt", "FinishedAt", "ExitCode"):
+        changed = copy.deepcopy(containers)
+        changed[0]["fixed"]["State"][field] = "foreign"
+        assert specialize._orphan_legacy_identity(changed, q) != before
+
+
 def test_execute_runs_only_approved_shell_then_requires_verified_postcondition(specialize, monkeypatch):
     script = (ROOT / "scripts" / "upgrade_ci_runner_fleet_specialized.sh").read_bytes()
     digest = hashlib.sha256(script).hexdigest()