fix(migrate): keep pre-worker earners visible to the worker after the owner migration

app/owner_migrate.py

@@ -34,6 +34,7 @@
 from pathlib import Path
 from typing import Any
 
+from app.constants import CASHPILOT_DEPLOYED
 from app.socket_policy import OWNER, OWNER_LABEL
 
 logger = logging.getLogger("cashpilot.migrate")
@@ -106,6 +107,14 @@ def owned(labels: dict[str, str] | None) -> bool:
     return (labels or {}).get(OWNER_LABEL) == OWNER
 
 
+def ours(labels: dict[str, str] | None) -> bool:
+    """A container CashPilot deployed: by the worker, or by the deploy path
+    before workers ("direct"), whose earners still run on old installs. The
+    proxy shows the worker neither kind until it carries the worker label, so
+    the migration moves both and recreates each "direct" one with it."""
+    return (labels or {}).get(OWNER_LABEL) in CASHPILOT_DEPLOYED
+
+
 def backup_name(volume: str) -> str:
     return volume + BACKUP_SUFFIX
 
@@ -185,6 +194,9 @@ def _remove_helper(self) -> None:
         with contextlib.suppress(NotFound):
             self.client.containers.get(HELPER_NAME).remove(force=True)
 
+    def _ours(self) -> list[Any]:
+        return [c for c in self.client.containers.list(all=True, filters={"label": OWNER_LABEL}) if ours(c.labels)]
+
     # -- the run ------------------------------------------------------------
 
     def run(self, compose_volumes: list[tuple[str, str]]) -> int:
@@ -289,7 +301,7 @@ def migrate_networks(self) -> None:
         names = {r["name"] for r in self.state.all("networks") if not r.get("done")}
         for net in self.client.networks.list(filters={"label": FENCE_LABEL}):
             names.add((net.attrs or {}).get("Name") or net.name)
-        for c in self.client.containers.list(all=True, filters={"label": f"{OWNER_LABEL}={OWNER}"}):
+        for c in self._ours():
             names.update(((c.attrs or {}).get("NetworkSettings") or {}).get("Networks") or {})
         for name in sorted(names - BUILTIN_NETWORKS):
             try:
@@ -317,7 +329,7 @@ def migrate_network(self, name: str) -> None:
             attrs = net.attrs or {}
             members = []
             for c in self._on_network(name):
-                if not owned(c.labels):
+                if not ours(c.labels):
                     raise LeftAlone(
                         f"{c.name} is on it and was not deployed by the worker, so the network was left as it "
                         f"was, and the worker cannot put services on it. To move it, disconnect that container "
@@ -350,7 +362,7 @@ def migrate_network(self, name: str) -> None:
             for c in self._on_network(name):
                 if c.name in known:
                     continue
-                if not owned(c.labels):
+                if not ours(c.labels):
                     # Leave it as it is, with the members an earlier run took off back on.
                     self._reconnect(net, record)
                     self.state.drop("networks", name)
@@ -465,10 +477,12 @@ def _run_plan_logged(self, plan: dict[str, Any]) -> None:
             self.fail(f"{', '.join(c['Name'].lstrip('/') for c in plan['containers'])}: {exc}")
 
     def _new_plans(self) -> list[dict[str, Any]]:
-        """Groups of worker containers that share unlabelled volumes, one plan each."""
+        """Groups of worker containers that share unlabelled volumes, one plan each.
+        A "direct" container gets a plan even with no such volume: recreating
+        it is the only way to give it the worker label."""
         from docker.errors import NotFound
 
-        workers = self.client.containers.list(all=True, filters={"label": f"{OWNER_LABEL}={OWNER}"})
+        workers = self._ours()
         # A stored plan that did not finish already holds these, stopped by it;
         # a new plan would record them as stopped and bring them back stopped.
         held = {c["Id"] for p in self.state.all("plans") if not p.get("done") for c in p["containers"]}
@@ -488,6 +502,7 @@ def _new_plans(self) -> list[dict[str, Any]]:
                     continue
                 users.setdefault(mount["Name"], set()).add(c.id)
                 attrs[c.id] = c.attrs
+        legacy = [c for c in workers if c.id not in held and not owned(c.labels)]
         # Join containers that share a volume: they stop and come back together.
         groups: list[tuple[set[str], set[str]]] = []
         for vol, cids in users.items():
@@ -497,13 +512,17 @@ def _new_plans(self) -> list[dict[str, Any]]:
                 merged[0].update(group[0])
                 merged[1].update(group[1])
             groups.append(merged)
+        for c in legacy:
+            if not any(c.id in g[0] for g in groups):
+                attrs[c.id] = c.attrs
+                groups.append(({c.id}, set()))
         groups = self._with_joiners(groups, attrs, held)
         plans = []
         for cids, vols in groups:
             strangers = sorted(
                 attrs[cid]["Name"].lstrip("/")
                 for cid in cids
-                if not owned((attrs[cid].get("Config") or {}).get("Labels"))
+                if not ours((attrs[cid].get("Config") or {}).get("Labels"))
             )
             if strangers:
                 self.warn(
@@ -603,7 +622,10 @@ def _current(self, info: dict[str, Any]) -> Any | None:
 
     def run_plan(self, plan: dict[str, Any]) -> None:
         names = [c["Name"].lstrip("/") for c in plan["containers"]]
-        logger.info("Moving %s onto labelled volumes: %s", ", ".join(names), ", ".join(plan["volumes"]))
+        if plan["volumes"]:
+            logger.info("Moving %s onto labelled volumes: %s", ", ".join(names), ", ".join(plan["volumes"]))
+        else:
+            logger.info("Recreating %s with %s=%s, as the worker deploys it", ", ".join(names), OWNER_LABEL, OWNER)
         # A container in another's namespace goes first and comes back last.
         in_plan = {c["Id"]: c["Name"].lstrip("/") for c in plan["containers"]}
         joined = [c for c in plan["containers"] if _namespace_of(c) in in_plan]
@@ -653,7 +675,7 @@ def run_plan(self, plan: dict[str, Any]) -> None:
             self._left_for_the_worker(info)
         plan["done"] = True
         self.state.put("plans", plan["key"], plan)
-        logger.info("%s back as before, on labelled volumes.", ", ".join(names))
+        logger.info("%s back as before, labelled for the worker.", ", ".join(names))
 
     @staticmethod
     def _left_for_the_worker(info: dict[str, Any]) -> None:

tests/test_owner_migrate.py

@@ -872,3 +872,36 @@ def test_a_proxied_earner_with_its_own_volume_moves_in_the_same_plan_as_its_tunn
     new_earner = d.containers.get("cashpilot-honeygain-p1")
     assert new_earner.attrs["HostConfig"]["NetworkMode"] == f"container:{new_tunnel.id}"
     assert d.data["honeygain-data"] == {"device": "id"}
+
+
+def test_earners_the_pre_worker_deploy_path_made_are_adopted_with_the_fence(install, tmp_path):
+    """deployed-by=direct counts as CashPilot's (constants.CASHPILOT_DEPLOYED).
+    Left alone, the fence would stay unlabelled and the earners would vanish
+    from the worker behind the proxy."""
+    d, myst, _ = install
+    direct = {"cashpilot.managed": "true", "cashpilot.service": "honeygain", om.OWNER_LABEL: "direct"}
+    old = d.add(Container(d, "cashpilot-honeygain", direct, networks=("cashpilot-isolated",)))
+    d.volume("earnapp-data", {"uuid": "sdk-node-1"})
+    old_vol = d.add(
+        Container(
+            d,
+            "cashpilot-earnapp",
+            {**direct, "cashpilot.service": "earnapp"},
+            [("earnapp-data", "/etc/earnapp")],
+            networks=("cashpilot-isolated",),
+            running=False,
+        )
+    )
+    m = FakeMigrator(d, om.State(tmp_path))
+    assert m.run([]) == 0 and m.warnings == 0
+    assert om.owned(d.nets["cashpilot-isolated"].attrs["Labels"])
+    hg = d.containers.get("cashpilot-honeygain")
+    assert hg.id != old.id and hg.status == "running"
+    assert hg.labels == {**direct, **OWNER, om.RECREATED_LABEL: old.id}
+    assert hg.networks() == ["cashpilot-isolated"]
+    ea = d.containers.get("cashpilot-earnapp")
+    assert ea.id != old_vol.id and ea.status == "exited" and om.owned(ea.labels)
+    assert om.owned(d.vols["earnapp-data"].attrs["Labels"]) and d.data["earnapp-data"] == {"uuid": "sdk-node-1"}
+    again = FakeMigrator(d, om.State(tmp_path))
+    assert again.run([]) == 0 and again.copies == []
+    assert d.containers.get("cashpilot-honeygain").id == hg.id