perf(swarm): let a runtime create its services without resolving the image

docs/reference/machines.md

@@ -862,6 +862,7 @@ that order; every environment on the runtime is reached through a target carryin
 | `fallback` | `tuple[str, ...]` | `()` | As for an ssh host: see [Falling back](#falling-back). |
 | `made` | `str` | `"typed"` | Always `typed`. |
 | `harness` | `str` | `"env"` | As for an ssh host. |
+| `resolve_image` | `bool` | `true` | Whether a service is created with its image's digest asked of the registry, as docker does; `false` passes `--no-resolve-image`: each node runs the image it holds under that name, pulling it only where it holds none. No registry round trip (about two seconds a service); for an image pinned by digest, or the same on every node. |
 
 `daemon()` returns the manager's [`Endpoint`](#endpoints), as a docker daemon's does.
 `store.node_of(via)` is how a `nodes` value is reached: `ssh:<name>` for a saved ssh runtime of

docs/reference/settings.md

@@ -214,6 +214,7 @@ entry that does not validate is skipped. A runtime an earlier Humanize wrote wit
 | `cpus` | `float` |
 | `memory`, `max_tasks` | `int` |
 | `nodes` | `{str: str}`: a node's host name to a saved ssh runtime's name or `[user@]host[:port]` |
+| `resolve_image` | `bool` (default `true`; `false`: `--no-resolve-image`) |
 | `made` | `"typed"` |
 
 `apple-container`: `image`, `workdir`, `harness` (`str`), `run_args`, `fallback` (`[str]`),

src/hmz/coganchor/machines/store.py

@@ -623,6 +623,9 @@ class SwarmRuntime:
       made: How it was made, which is :data:`TYPED`.
       harness: Where the harness of an agent working on it runs, and nowhere else: see
         :func:`affine`.
+      resolve_image: Whether a service is created with its image's digest asked of the
+        registry, docker's way; False passes `--no-resolve-image`: each node runs the image
+        it holds under that name (see :class:`~hmz.coganchor.machines.swarm.SwarmConfig`).
     """
 
     backend: ClassVar[str] = SWARM
@@ -643,6 +646,7 @@ class SwarmRuntime:
     fallback: tuple[str, ...] = ()
     made: str = TYPED
     harness: str = ENV
+    resolve_image: bool = True
 
     def __post_init__(self) -> None:
         _named(self.name)
@@ -939,11 +943,18 @@ def held(self) -> dict[str, Any]:
 )
 
 
+#: Fields written down only where they are not their default: a runtime that leaves them
+#: so is written as a Humanize without them wrote it, and reads there.
+_UNLESS_DEFAULT = frozenset({"resolve_image"})
+
+
 def _fields(runtime: Runtime) -> dict[str, Any]:
-    """Every field of one, as JSON holds it."""
+    """Every field of one, as JSON holds it, but those of `_UNLESS_DEFAULT` at default."""
     held: dict[str, Any] = {}
     for one in dataclasses.fields(runtime):
         value: object = getattr(runtime, one.name)
+        if one.name in _UNLESS_DEFAULT and value == one.default:
+            continue
         if isinstance(value, tuple):
             value = [str(each) for each in cast("tuple[object, ...]", value)]
         elif one.name in _MAPPINGS:
@@ -1117,6 +1128,10 @@ def _typed(key: str, default: object, value: object, name: str) -> object:
         if not isinstance(value, (list, tuple)):
             raise wrong
         return tuple(str(one) for one in cast("Iterable[Any]", value))
+    if isinstance(default, bool):  # a switch: true or false, never 0 or 1
+        if not isinstance(value, bool):
+            raise wrong
+        return value
     if isinstance(value, bool):
         raise wrong
     if isinstance(default, float):

src/hmz/coganchor/machines/swarm.py

@@ -115,6 +115,11 @@ class SwarmConfig(MachineConfig):
         given up on.
       starting: How long, in seconds, it may take to be running once it has a node --
         pulling its image among it.
+      resolve_image: Whether the manager asks the image's registry for its digest as it
+        creates the service, docker's way; False creates it with `--no-resolve-image`,
+        each node running the image it holds under that name and pulling it only where it
+        holds none -- no registry round trip, some two seconds each, for an image pinned
+        by digest or the same on every node.
     """
 
     image: str = "python:3.12"
@@ -133,6 +138,7 @@ class SwarmConfig(MachineConfig):
     labels: Mapping[str, str] = field(default_factory=dict[str, str], hash=False)
     placing: float = 30.0
     starting: float = 600.0
+    resolve_image: bool = True
 
     def __post_init__(self) -> None:
         """Refuses what the swarm would refuse, where it is written rather than as it starts.
@@ -459,6 +465,7 @@ def submit(self) -> str:
                     "1",
                     "--restart-condition",
                     "none",
+                    *(() if config.resolve_image else ("--no-resolve-image",)),
                     "--name",
                     name,
                     *self._labelled(),

src/hmz/runtime/flowing/environing_swarm.py

@@ -296,6 +296,7 @@ def _brought_up(self) -> tuple[Swarm, Share]:
                     nodes=_nodes(stored),
                     traced=self.traced,
                     run_args=stored.run_args if stored is not None else (),
+                    resolve_image=stored.resolve_image if stored is not None else True,
                     labels={
                         PROVIDER: self.provider,
                         ROLE: self.role,

src/hmz/tui/pick.py

@@ -8384,6 +8384,9 @@ def _more(self, typed: Mapping[str, str]) -> dict[str, Any]:
             "max_tasks": _how_many(typed.get(_TASKS, ""), "max tasks"),
             "gpu_resource": typed.get(_RESOURCE, ""),
             "nodes": _pairs(typed.get(_NODES, "")),
+            # Not on the form, and so kept as it was written rather than reset by it: a swarm
+            # told not to resolve its image would otherwise be corrected into one that does.
+            "resolve_image": self._one.resolve_image if self._one is not None else True,
         }
 
 

tests/unit/coganchor/machines/test_runtimes.py

@@ -299,6 +299,27 @@ def test_a_name_prefix_is_kept_as_it_was_written(backend: str, prefix: str) -> N
     assert new(backend, "box", name_prefix="").held()["name_prefix"] == ""
 
 
+def test_a_swarm_resolves_its_image_unless_told_not_to() -> None:
+    assert SwarmRuntime(name="sw").resolve_image is True
+    assert "resolve_image" not in SwarmRuntime(name="sw").held()  # as written before
+    unresolved = new(SWARM, "sw", resolve_image=False)
+
+    assert isinstance(unresolved, SwarmRuntime)
+    assert unresolved.resolve_image is False
+    assert unresolved.held()["resolve_image"] is False
+    written = {
+        k: v for k, v in unresolved.held().items() if k not in ("backend", "name")
+    }
+    read_back = new(SWARM, "sw", **written)
+
+    assert isinstance(read_back, SwarmRuntime)
+    assert read_back.resolve_image is False  # read back as written
+    with pytest.raises(ValueError, match="resolve_image cannot be 0"):
+        new(SWARM, "sw", resolve_image=0)
+    with pytest.raises(ValueError, match="resolve_image cannot be 'no'"):
+        new(SWARM, "sw", resolve_image="no")
+
+
 def test_a_swarm_takes_constraints_and_nodes_it_can_reach() -> None:
     runtime = SwarmRuntime(
         name="sw",

tests/unit/coganchor/machines/test_swarm.py

@@ -324,6 +324,27 @@ def test_a_service_of_one_task_is_created_and_reached_where_it_landed(
     assert handshake.asked == [anchor]
 
 
+def test_a_service_asks_the_registry_for_its_image_unless_told_not_to(
+    runs: Runs, handshake: Handshake
+) -> None:
+    a_swarm(runs, task("running"), task("running"))
+
+    SwarmConfig(workspace="/w", user="5:5").create().start()
+    SwarmConfig(workspace="/w", user="5:5", resolve_image=False).create().start()
+
+    resolved, unresolved = runs.called("service", "create")
+    assert "--no-resolve-image" not in resolved
+    assert unresolved.count("--no-resolve-image") == 1
+    assert unresolved.index("--no-resolve-image") < unresolved.index("python:3.12")
+
+    def unnamed(argv: list[str]) -> list[str]:  # each service has a name of its own
+        at = argv.index("--name")
+        return [*argv[:at], *argv[at + 2 :]]
+
+    without = [one for one in unresolved if one != "--no-resolve-image"]
+    assert unnamed(without) == unnamed(resolved)
+
+
 def test_a_task_with_no_gpus_is_handed_none_by_the_runtime(
     runs: Runs, handshake: Handshake
 ) -> None:

tests/unit/runtime/flowing/doubles_u11.py

@@ -467,6 +467,7 @@ class Runtime:
     exec_user: str = ""
     exec_env: tuple[str, ...] = ()
     exec_wrapper: tuple[str, ...] = ()
+    resolve_image: bool = True
 
     def exec_options(self) -> tuple[tuple[str, str], ...]:
         return (

tests/unit/runtime/flowing/test_environing_swarm.py

@@ -258,6 +258,26 @@ def daemon_of(via: str) -> Endpoint:
     return Config
 
 
+async def test_a_runtime_that_does_not_resolve_its_image_says_so_to_the_service(
+    manager: type[_Config],
+) -> None:
+    machine = SwarmMachine(
+        "c",
+        PurePosixPath("/srv/x"),
+        stored=runtime(
+            Runtime(name="c", cpus=8, gpu_resource="NVIDIA-GPU", resolve_image=False)
+        ),
+        role=_worker(),
+    )
+    driver = MachineEnvDriver(machine, PurePosixPath("/work"))
+
+    await driver.probe()
+
+    (made,) = manager.made
+    assert made["resolve_image"] is False
+    await driver.close()
+
+
 async def test_a_task_is_created_with_its_share_and_reached_where_it_landed(
     manager: type[_Config],
 ) -> None:
@@ -289,6 +309,7 @@ async def test_a_task_is_created_with_its_share_and_reached_where_it_landed(
     assert made["constraints"] == ("node.labels.gpu==1",)
     assert made["nodes"] == {"node-2": "tcp://node-2.lan:2376"}
     assert (made["traced"], made["run_args"]) == (True, ("--init",))
+    assert made["resolve_image"] is True  # docker's way, unless the runtime says not
     assert (made["labels"][PROVIDER], made["labels"][ROLE]) == ("c", "worker")
     assert machine.target.startswith("docker://task.1.abc@")
     assert machine.resources(gpus=True) == Resources(4, 8 << 30, 1, 16384 << 20)

tests/unit/tui/test_pick_swarm.py

@@ -0,0 +1,83 @@
+"""`hmz.tui.pick.Swarming`: a swarm corrected on its form, and what it keeps of what it was."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from typing import TYPE_CHECKING, Any, cast
+
+import pytest
+from textual.app import App
+
+import hmz.daemon
+from hmz.tui import pick
+
+if TYPE_CHECKING:
+    from hmz.coganchor.machines.store import SwarmRuntime
+
+
+class _Runtimes:
+    """The store, as far as the form asks it: none taken, and one made as it was told."""
+
+    def all(self, backend: str) -> list[Any]:
+        return []
+
+    def new(self, backend: str, name: str, **fields: Any) -> Any:
+        return SimpleNamespace(backend=backend, name=name, **fields)
+
+
+class _Host(App[None]):
+    """Puts the form up and keeps what it was answered with."""
+
+    def __init__(self, form: pick.Swarming) -> None:
+        super().__init__()
+        self.form = form
+        self.answers: list[Any] = []
+
+    def on_mount(self) -> None:
+        self.push_screen(self.form, self.answers.append)
+
+
+def _swarm(*, resolve_image: bool) -> SwarmRuntime:
+    return cast(
+        "SwarmRuntime",
+        SimpleNamespace(
+            name="gpus",
+            endpoint="local",
+            tls_dir="",
+            image="python:3.12",
+            run_args=(),
+            cpus=0.0,
+            memory=0,
+            workdir="",
+            fallback=(),
+            harness="env",
+            name_prefix="",
+            constraints=("node.labels.gpu==true",),
+            max_tasks=4,
+            gpu_resource="",
+            nodes={},
+            resolve_image=resolve_image,
+        ),
+    )
+
+
+@pytest.fixture(autouse=True)
+def runtimes(monkeypatch: pytest.MonkeyPatch) -> None:
+    here = SimpleNamespace(runtimes=_Runtimes())
+    monkeypatch.setattr(hmz.daemon, "Hmz", lambda: here)
+
+
+@pytest.mark.parametrize("resolve_image", [False, True])
+async def test_a_swarm_corrected_keeps_whether_it_resolves_its_image(
+    resolve_image: bool,
+) -> None:
+    host = _Host(pick.Swarming(_swarm(resolve_image=resolve_image)))
+    async with host.run_test() as pilot:
+        await pilot.pause()
+        host.form.action_done()
+        await pilot.pause()
+
+    [made] = host.answers
+    assert made.name == "gpus"
+    assert made.constraints == ["node.labels.gpu==true"]
+    assert made.resolve_image is resolve_image