Merge same-mechanism limit layers into one launcher invocation

The triple-stack scenarios declare memory and CPU layers on the same
container and nested cgroup. Composing every layer linearly nested a
second docker run inside the first, applied the address-space rlimit to
the docker client instead of the payload, and had two cgroup scripts
writing the same path, so the scenario could never reach a measurement.

Layers sharing an enforcement context are now grouped onto one launcher.
Also corrects the Windows job object CPU rate to a share of total machine
capacity, makes the Kubernetes probe check controller delegation as
documented, and stops discarding dependency install errors under a
non-root user.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BEZUSri3baWU58Ge1kAVnT

Author: claude

tests/e2e/launchers/docker.py

@@ -9,6 +9,14 @@
 Also used, with ``privileged=True, cgroupns_private=True``, as the outermost
 layer of the triple-stack scenario -- the nested-cgroup layer's launcher
 (:mod:`tests.e2e.launchers.nested_cgroup`) supplies the container's CMD.
+
+``memory_bytes`` and a CPU axis (``cpu_millicores`` or ``cpuset_millicores``)
+may be supplied *together*: the triple-stack scenarios declare a separate
+``DOCKER_MEMORY`` layer and a separate ``DOCKER_CPU_QUOTA`` layer for the same
+container, and :mod:`tests.e2e.launchers.factory` merges both onto **one**
+``DockerLauncher`` instance -- one ``docker run`` imposing both ceilings at
+once (``--memory ... --memory-swap ... --cpus ...``), never a second, nested
+``docker run`` (iter-2 fix for the "docker-in-docker" composition blocker).
 """
 
 from __future__ import annotations
@@ -37,9 +45,10 @@ def __init__(
         cgroupns_private: bool = False,
         image: str = DEFAULT_IMAGE,
     ):
-        set_count = sum(v is not None for v in (memory_bytes, cpu_millicores, cpuset_millicores))
-        if set_count != 1:
-            raise ValueError("DockerLauncher takes exactly one of memory_bytes/cpu_millicores/cpuset_millicores")
+        if cpu_millicores is not None and cpuset_millicores is not None:
+            raise ValueError("DockerLauncher takes at most one of cpu_millicores/cpuset_millicores (two different CPU mechanisms)")
+        if memory_bytes is None and cpu_millicores is None and cpuset_millicores is None:
+            raise ValueError("DockerLauncher requires at least one of memory_bytes/cpu_millicores/cpuset_millicores")
         self.memory_bytes = memory_bytes
         self.cpu_millicores = cpu_millicores
         self.cpuset_millicores = cpuset_millicores
@@ -61,9 +70,9 @@ def wrap_argv(self, inner_argv: Sequence[str]) -> list[str]:
             argv += ["--cgroupns", "private"]
         if self.memory_bytes is not None:
             argv += ["--memory", str(self.memory_bytes), "--memory-swap", str(self.memory_bytes)]
-        elif self.cpu_millicores is not None:
+        if self.cpu_millicores is not None:
             argv += ["--cpus", f"{self.cpu_millicores / 1000:g}"]
-        else:
+        elif self.cpuset_millicores is not None:
             n_cores = self.cpuset_millicores // 1000
             cpuset = f"0-{n_cores - 1}" if n_cores > 1 else "0"
             argv += ["--cpuset-cpus", cpuset]
@@ -75,8 +84,24 @@ def wrap_argv(self, inner_argv: Sequence[str]) -> list[str]:
             "-w", _CONTAINER_WORKDIR,
             "-e", f"PYTHONPATH={_CONTAINER_WORKDIR}/src",
         ]
+        if self.user is not None:
+            # Non-root: `pip install --user` needs a resolvable, writable user
+            # site-packages directory. The stock upstream image has no
+            # /etc/passwd entry (and so no $HOME) for an arbitrary numeric
+            # --user uid; Python's home-directory lookup checks $HOME first,
+            # before ever falling back to the passwd database, so setting it
+            # explicitly sidesteps the missing-passwd-entry problem entirely
+            # rather than depending on one existing (iter-2 fix -- this used
+            # to fail silently and kill the payload with an unlogged
+            # ModuleNotFoundError). /tmp is world-writable (mode 1777) in
+            # every upstream image, so it works for any uid.
+            argv += ["-e", "HOME=/tmp"]
         argv.append(self.image)
         pip_install = "pip install --user -q psutil" if self.user else "pip install -q psutil"
-        prelude = f"{pip_install} >/dev/null 2>&1"
-        argv += ["sh", "-c", f"{prelude}; exec {shlex.join(inner_argv)}"]
+        # `set -e` + `&&` (never `;` with output discarded): a failed install
+        # must abort the container and surface its own stderr in the captured
+        # output, never be swallowed by `>/dev/null 2>&1` and leave the
+        # payload to die on a bare, unexplained `ModuleNotFoundError` with no
+        # sentinel line (iter-2 fix).
+        argv += ["sh", "-c", f"set -e; {pip_install} && exec {shlex.join(inner_argv)}"]
         return argv

tests/e2e/launchers/factory.py

@@ -3,10 +3,27 @@
 The one place layer-type -> launcher wiring lives; everything else (the
 registry, the outer harness) works in terms of :class:`Layer`/:class:`Launcher`
 only.
+
+**Merging same-mechanism layers (iter-2 fix for the triple-stack composition
+blocker).** A naive one-launcher-per-layer mapping is wrong whenever a
+scenario declares two layers that are really the *same enforcement context*
+imposing two different axes -- e.g. the triple-stack scenarios' separate
+``DOCKER_MEMORY``/``DOCKER_CPU_QUOTA`` layers are one container, not two
+nested ``docker run`` invocations, and their separate
+``NESTED_CGROUP_MEMORY``/``NESTED_CGROUP_CPU`` layers are one nested cgroup,
+not two independently-created cgroups at the same path. :func:`build_launchers`
+(and :func:`build_launcher_groups`, which also reports which original layers
+each merged launcher covers) group a scenario's layers by *launcher family*
+first, merge every layer of a multi-axis family (currently docker and the
+nested-cgroup layer) onto one launcher instance, and leave every other layer
+type exactly one-to-one with its own launcher -- preserving outside-in order
+by each family's first occurrence in ``scenario.layers``.
 """
 
 from __future__ import annotations
 
+from typing import Sequence
+
 from tests.e2e.launchers.base import Launcher
 from tests.e2e.launchers.cgroup import CgroupLauncher
 from tests.e2e.launchers.docker import DockerLauncher
@@ -24,6 +41,24 @@
 _MEMORY_KWARG = "memory_bytes"
 _CPU_KWARG = "cpu_millicores"
 
+# Layer types that merge onto one launcher instance per scenario when a
+# scenario declares more than one of them -- the "same enforcement context,
+# two axes" case. Every LayerType not listed here is its own singleton family:
+# exactly one launcher per layer, as before.
+_DOCKER_TYPES = frozenset({LayerType.DOCKER_MEMORY, LayerType.DOCKER_CPU_QUOTA, LayerType.DOCKER_CPUSET})
+_NESTED_CGROUP_TYPES = frozenset({LayerType.NESTED_CGROUP_MEMORY, LayerType.NESTED_CGROUP_CPU})
+
+_FAMILY_DOCKER = "docker"
+_FAMILY_NESTED_CGROUP = "nested_cgroup"
+
+
+def _layer_family(layer: Layer) -> str:
+    if layer.type in _DOCKER_TYPES:
+        return _FAMILY_DOCKER
+    if layer.type in _NESTED_CGROUP_TYPES:
+        return _FAMILY_NESTED_CGROUP
+    return layer.type.value  # singleton family: this layer type never merges
+
 
 def build_launcher(layer: Layer) -> Launcher:
     hints = dict(layer.launcher_hints)
@@ -76,6 +111,100 @@ def build_launcher(layer: Layer) -> Launcher:
     raise ValueError(f"no launcher factory registered for layer type {layer.type}")  # pragma: no cover
 
 
+def _merge_hints(layers: Sequence[Layer]) -> dict[str, object]:
+    """Union launcher_hints across a merged group, raising on a genuine
+    conflict (two layers of the same family disagreeing on the same hint) --
+    every triple-stack docker layer declares matching privileged/cgroupns_private
+    hints, so a conflict here would mean a bad registry entry, not something
+    to silently resolve one way or the other."""
+    merged: dict[str, object] = {}
+    for layer in layers:
+        for key, value in layer.launcher_hints.items():
+            if key in merged and merged[key] != value:
+                raise ValueError(
+                    f"conflicting launcher_hint {key!r} ({merged[key]!r} vs {value!r}) "
+                    f"across layers merged into one launcher: {[l.type for l in layers]}"
+                )
+            merged[key] = value
+    return merged
+
+
+def _merge_docker_layers(layers: Sequence[Layer]) -> DockerLauncher:
+    """One DockerLauncher for every DOCKER_* layer in ``layers`` -- one
+    ``docker run`` imposing every axis declared (``--memory``/``--memory-swap``
+    together with ``--cpus`` or ``--cpuset-cpus``), never a second, nested
+    ``docker run`` for the CPU axis."""
+    kwargs: dict[str, object] = _merge_hints(layers)
+    for layer in layers:
+        if layer.type is LayerType.DOCKER_MEMORY:
+            kwargs[_MEMORY_KWARG] = layer.imposed
+        elif layer.type is LayerType.DOCKER_CPU_QUOTA:
+            kwargs[_CPU_KWARG] = layer.imposed
+        elif layer.type is LayerType.DOCKER_CPUSET:
+            kwargs["cpuset_millicores"] = layer.imposed
+    return DockerLauncher(**kwargs)
+
+
+def _merge_nested_cgroup_layers(layers: Sequence[Layer]) -> NestedCgroupLauncher:
+    """One NestedCgroupLauncher for every NESTED_CGROUP_* layer in ``layers``
+    -- one script that mkdir's the nested cgroup once, probes it once, and
+    writes both ``memory.max``/``memory.swap.max`` and ``cpu.max`` to it,
+    never two independent scripts racing to create/write the same path."""
+    kwargs: dict[str, object] = {}
+    probe_key: str | None = None
+    for layer in layers:
+        if layer.probe_key:
+            if probe_key is not None and probe_key != layer.probe_key:
+                raise ValueError(
+                    f"conflicting probe_key across merged nested-cgroup layers: {probe_key!r} vs {layer.probe_key!r}"
+                )
+            probe_key = layer.probe_key
+        if layer.type is LayerType.NESTED_CGROUP_MEMORY:
+            kwargs[_MEMORY_KWARG] = layer.imposed
+        elif layer.type is LayerType.NESTED_CGROUP_CPU:
+            kwargs[_CPU_KWARG] = layer.imposed
+    if probe_key:
+        kwargs["probe_key"] = probe_key
+    return NestedCgroupLauncher(**kwargs)
+
+
+def build_launcher_groups(scenario: Scenario) -> list[tuple[Launcher, tuple[Layer, ...]]]:
+    """Group ``scenario.layers`` by enforcement-context family, merge each
+    multi-axis family (docker, nested-cgroup) onto one launcher instance, and
+    return ``(launcher, layers_it_covers)`` pairs outermost-first -- ordered by
+    each family's *first* occurrence in ``scenario.layers``, so genuinely
+    nested contexts (container -> nested cgroup -> rlimit -> payload) still
+    compose outside-in via :func:`~tests.e2e.launchers.base.compose_argv`.
+
+    Every layer type outside the two merging families is its own singleton
+    group: one launcher per layer, exactly as before this fix.
+    """
+    groups: dict[str, list[Layer]] = {}
+    order: list[str] = []
+    for layer in scenario.layers:
+        family = _layer_family(layer)
+        if family not in groups:
+            groups[family] = []
+            order.append(family)
+        groups[family].append(layer)
+
+    result: list[tuple[Launcher, tuple[Layer, ...]]] = []
+    for family in order:
+        group = tuple(groups[family])
+        if family == _FAMILY_DOCKER:
+            result.append((_merge_docker_layers(group), group))
+        elif family == _FAMILY_NESTED_CGROUP:
+            result.append((_merge_nested_cgroup_layers(group), group))
+        else:
+            assert len(group) == 1, f"unexpected multiple layers grouped under singleton family {family!r}: {group}"
+            result.append((build_launcher(group[0]), group))
+    return result
+
+
 def build_launchers(scenario: Scenario) -> list[Launcher]:
-    """One launcher per layer, outermost-first, matching scenario.layers order."""
-    return [build_launcher(layer) for layer in scenario.layers]
+    """One launcher per *enforcement context* (same-mechanism, multi-axis
+    layers merged -- see module docstring), outermost-first. Use
+    :func:`build_launcher_groups` when the caller also needs to know which
+    original layers a given launcher covers (e.g. to fold probe results back
+    per layer)."""
+    return [launcher for launcher, _layers in build_launcher_groups(scenario)]

tests/e2e/launchers/kubernetes.py

@@ -45,23 +45,62 @@ def __init__(
         self.namespace = namespace
 
     def probe(self) -> ProbeResult:
+        """Structurally confirm cgroup v2 delegation *into the kind node*,
+        not just that the cluster is reachable (iter-2 fix: the previous
+        implementation only ran ``kubectl get nodes`` -- general cluster
+        liveness -- which can report a fully Ready cluster even when cgroup
+        delegation into the node itself is broken, e.g. a cgroup-driver
+        mismatch or a controller not delegated).
+
+        A kind "node" is itself a docker container (the container's hostname
+        matches the Kubernetes node name by kind's own convention), so its own
+        ``/sys/fs/cgroup/cgroup.controllers`` can be inspected the same way
+        :class:`~tests.e2e.launchers.podman.PodmanLauncher.probe` inspects
+        podman's host cgroup version: a cheap, structural presence check,
+        never a behavioral one.
+        """
         if shutil.which("kubectl") is None:
             return ProbeResult(False, "kubectl binary not found on PATH")
         try:
             nodes = subprocess.run(
-                ["kubectl", "get", "nodes", "-o", "json"], capture_output=True, text=True, timeout=30,
+                ["kubectl", "get", "nodes", "-o", "jsonpath={.items[0].metadata.name}"],
+                capture_output=True, text=True, timeout=30,
             )
         except (OSError, subprocess.TimeoutExpired) as exc:
             return ProbeResult(False, f"kubectl get nodes failed: {exc}")
         if nodes.returncode != 0:
             return ProbeResult(False, f"kubectl get nodes exited {nodes.returncode}: {nodes.stderr.strip()}")
-        try:
-            parsed = json.loads(nodes.stdout)
-        except json.JSONDecodeError as exc:
-            return ProbeResult(False, f"kubectl get nodes returned invalid JSON: {exc}")
-        if not parsed.get("items"):
+        node_name = nodes.stdout.strip()
+        if not node_name:
             return ProbeResult(False, "kind cluster reports zero nodes -- cgroup delegation not confirmed")
-        return ProbeResult(True, "kind cluster reachable and reports at least one Ready node")
+
+        if shutil.which("docker") is None:
+            return ProbeResult(
+                False, "docker binary not found on PATH; cannot inspect the kind node's cgroup delegation"
+            )
+        try:
+            controllers = subprocess.run(
+                ["docker", "exec", node_name, "cat", "/sys/fs/cgroup/cgroup.controllers"],
+                capture_output=True, text=True, timeout=30,
+            )
+        except (OSError, subprocess.TimeoutExpired) as exc:
+            return ProbeResult(False, f"docker exec into kind node {node_name!r} failed: {exc}")
+        if controllers.returncode != 0:
+            return ProbeResult(
+                False,
+                f"docker exec into kind node {node_name!r} exited {controllers.returncode}: "
+                f"{controllers.stderr.strip()} -- cgroup delegation not confirmed",
+            )
+        listed = controllers.stdout.split()
+        if "memory" not in listed or "cpu" not in listed:
+            return ProbeResult(
+                False,
+                f"kind node {node_name!r} cgroup.controllers is missing memory/cpu delegation: "
+                f"{controllers.stdout.strip()!r}",
+            )
+        return ProbeResult(
+            True, f"kind node {node_name!r} cgroup.controllers lists memory and cpu; delegation structurally confirmed"
+        )
 
     def _resource_limits(self) -> dict[str, str]:
         if self.memory_bytes is not None:

tests/e2e/launchers/nested_cgroup.py

@@ -20,6 +20,15 @@
 This layer's memory.swap.max=0 is set independently of the container's own
 ``--memory-swap`` -- it does NOT inherit that setting (design decision #13
 explicit requirement).
+
+``memory_bytes`` and ``cpu_millicores`` may be supplied *together*: the
+triple-stack scenarios declare a separate ``NESTED_CGROUP_MEMORY`` layer and a
+separate ``NESTED_CGROUP_CPU`` layer for the same nested cgroup, and
+:mod:`tests.e2e.launchers.factory` merges both onto **one**
+``NestedCgroupLauncher`` instance -- one ``mkdir``/probe/write script setting
+both ``memory.max``+``memory.swap.max`` and ``cpu.max`` on the same nested
+cgroup, never two independent scripts racing to create/write the same path
+(iter-2 fix for the "two nested-cgroup scripts" composition blocker).
 """
 
 from __future__ import annotations
@@ -40,8 +49,8 @@ def __init__(
         cpu_millicores: int | None = None,
         probe_key: str = "nested-cgroup-write-access",
     ):
-        if (memory_bytes is None) == (cpu_millicores is None):
-            raise ValueError("NestedCgroupLauncher takes exactly one of memory_bytes/cpu_millicores")
+        if memory_bytes is None and cpu_millicores is None:
+            raise ValueError("NestedCgroupLauncher requires at least one of memory_bytes/cpu_millicores")
         self.memory_bytes = memory_bytes
         self.cpu_millicores = cpu_millicores
         self.probe_key = probe_key
@@ -85,7 +94,7 @@ def wrap_argv(self, inner_argv: Sequence[str]) -> list[str]:
         if self.memory_bytes is not None:
             lines.append(f"echo {self.memory_bytes} > {_CHILD_CGROUP}/memory.max")
             lines.append(f"echo 0 > {_CHILD_CGROUP}/memory.swap.max")
-        else:
+        if self.cpu_millicores is not None:
             quota_us = int(self.cpu_millicores * _CFS_PERIOD_US / 1000)
             lines.append(f"echo '{quota_us} {_CFS_PERIOD_US}' > {_CHILD_CGROUP}/cpu.max")
         lines.append(f"echo $$ > {_CHILD_CGROUP}/cgroup.procs")