diff --git a/src/agent_env/env/envs/computer/utils/composer.py b/src/agent_env/env/envs/computer/utils/composer.py index 4ab6da95d..f3863976e 100644 --- a/src/agent_env/env/envs/computer/utils/composer.py +++ b/src/agent_env/env/envs/computer/utils/composer.py @@ -677,7 +677,9 @@ def _built_disk(env: ComputerEnv, base: ComputerArtifact, arch: str, container: """The status of the disk form of ``arch`` once built from ``container``, its container's result, and pushed, or reused, or the stage that failed and why. Its key is of the container's key, the base disk and the disk recipe; the base disk is the base's own, the one published for it, or else one made on the builder from the base's - container and the boot recipe. Its build runs no program of ``arch``'s but that recipe's.""" + container and the boot recipe. Its build runs no program of ``arch``'s but that recipe's. Two composes of one disk + on this machine take turns, so the second reuses the first's: built again, its tag would move off the image the + first pinned, which a boot mounts by that name and Docker doesn't pull for a mount.""" pair, platform = f"disk/{arch}", f"linux/{arch}" with (_attempting(env, "disk", arch) as attempt, tempfile.TemporaryDirectory(prefix="agentenv-compose-") as work, contextlib.ExitStack() as opened): @@ -687,34 +689,36 @@ def _built_disk(env: ComputerEnv, base: ComputerArtifact, arch: str, container: base_disk, base_identity, recipe = *_base_disk(base, arch), _digest(context) key = attempt.key = _key(container=container["key"], base_disk=base_identity, recipe=recipe, form="disk", arch=arch, format=FORMAT) - store, ref, pinned = _stored(env, key) - if pinned is not None: - progress.at(pair, "reusing", "reusing the disk of the same key, made earlier") - return {"status": "unverified", "key": key, "image": pinned, "reused": True, "built_at": now()} - link = builds.link(platform, base.image()) - progress.at(pair, "starting its builder", f"making the disk {builder.where(link)}") - on = opened.enter_context(builds.open(link, platform, Path(work), base.image(), {})) - on.login(store.auth(container["image"])) - if base_disk is not None: - on.login(get_config().get_image_store().auth(base_disk)) - if (tools := published_pinned(DISK_TOOLS_IMAGE)) is None: - tools = f"agentenv-disk-tools:{DISK_TOOLS_IMAGE.tag()}" - on.build((context / "tools" / "Dockerfile").read_text(), tools, on.platform, {}, - context=f"{builder.CONTEXT}/tools") - dockerfile = (context / "Dockerfile").read_text() - if base_disk is None: - progress.at(pair, "making the base's disk") - base_disk = _made_base_disk(on, base, arch, f"{base_identity} {recipe}", tools, dockerfile) - progress.at(pair, "making the disk") - local = f"agentenv-computer:{key[7:23]}-disk-{arch}" - on.build(dockerfile, local, platform, {KEY_LABEL: key}, build_args={ - "ROOTFS": container["image"], "ROOTFS_CONFIG": _encoded(_config(on, container["image"], arch)), - "ESP": base_disk, "TOOLS": tools}) - attempt.stage = "push" - progress.at(pair, "pushing the disk") - on.push(local, ref, store) - return {"status": "unverified", "key": key, "image": pin_digest(ref, store.auth(ref)), "reused": False, - "builder": on.name, "built_at": now()} + with holding_locks([f"linux-disk:{key}"], lambda: progress.at( + pair, "waiting", "waiting for the compose of the same disk under way on this machine")): + store, ref, pinned = _stored(env, key) + if pinned is not None: + progress.at(pair, "reusing", "reusing the disk of the same key, made earlier") + return {"status": "unverified", "key": key, "image": pinned, "reused": True, "built_at": now()} + link = builds.link(platform, base.image()) + progress.at(pair, "starting its builder", f"making the disk {builder.where(link)}") + on = opened.enter_context(builds.open(link, platform, Path(work), base.image(), {})) + on.login(store.auth(container["image"])) + if base_disk is not None: + on.login(get_config().get_image_store().auth(base_disk)) + if (tools := published_pinned(DISK_TOOLS_IMAGE)) is None: + tools = f"agentenv-disk-tools:{DISK_TOOLS_IMAGE.tag()}" + on.build((context / "tools" / "Dockerfile").read_text(), tools, on.platform, {}, + context=f"{builder.CONTEXT}/tools") + dockerfile = (context / "Dockerfile").read_text() + if base_disk is None: + progress.at(pair, "making the base's disk") + base_disk = _made_base_disk(on, base, arch, f"{base_identity} {recipe}", tools, dockerfile) + progress.at(pair, "making the disk") + local = f"agentenv-computer:{key[7:23]}-disk-{arch}" + on.build(dockerfile, local, platform, {KEY_LABEL: key}, build_args={ + "ROOTFS": container["image"], "ROOTFS_CONFIG": _encoded(_config(on, container["image"], arch)), + "ESP": base_disk, "TOOLS": tools}) + attempt.stage = "push" + progress.at(pair, "pushing the disk") + on.push(local, ref, store) + return {"status": "unverified", "key": key, "image": pin_digest(ref, store.auth(ref)), "reused": False, + "builder": on.name, "built_at": now()} return attempt.result diff --git a/tst/integration/conftest.py b/tst/integration/conftest.py new file mode 100644 index 000000000..6f79b6565 --- /dev/null +++ b/tst/integration/conftest.py @@ -0,0 +1,32 @@ +"""The integration tier's setup shared by every test.""" + +import os +import tempfile +from pathlib import Path + +import pytest + +from agent_env.store import local_state + +#: The locks that guard this machine's Docker and image registry, by the start of their ids: a run's tests keep stores +#: and state roots of their own, but all of them, on every pytest worker, use the one Docker and registry. +MACHINE_LOCKS = ("computer-compose:", "linux-disk:", "windows-disk:", "omarchy-disk:", "disk-export:", "image-build:", + "image-pull:", "image-save:") + + +@pytest.fixture(scope="session", autouse=True) +def machine_locks_shared_by_the_runs_workers(): + """Each test's state root, where agent-env keeps its locks, is its own, so two pytest workers composing one + computer took no lock in common and both built it under the same tags: the second moved them off the image the + first had pinned, or their pushes collided. The locks guarding this machine's Docker and registry are kept in one + folder for the run, as one user's processes keep theirs, so such work takes turns and the second reuses the first's. + """ + run = os.environ.get("PYTEST_XDIST_TESTRUNUID") or str(os.getpid()) + folder = Path(tempfile.gettempdir()) / f"agent-env-machine-locks-{run}" + folder.mkdir(mode=0o700, exist_ok=True) + per_test = local_state._lock_path + patch = pytest.MonkeyPatch() + patch.setattr(local_state, "_lock_path", + lambda id: folder / per_test(id).name if id.startswith(MACHINE_LOCKS) else per_test(id)) + yield + patch.undo() diff --git a/tst/unit/env/envs/test_computer_disk.py b/tst/unit/env/envs/test_computer_disk.py index daff46e01..119de0b0a 100644 --- a/tst/unit/env/envs/test_computer_disk.py +++ b/tst/unit/env/envs/test_computer_disk.py @@ -7,6 +7,7 @@ import asyncio import base64 +import contextlib import json import threading from datetime import datetime @@ -22,7 +23,7 @@ from agent_env.env import Env from agent_env.config.runtime import Config from agent_env.env.envs.computer.env import ComputerEnv -from agent_env.env.envs.computer.utils import boot, composer, probing +from agent_env.env.envs.computer.utils import boot, builder, composer, probing from agent_env.env.envs.computer.utils.composer import Composing from agent_env.env.published_images import DISK from agent_env import providers @@ -89,6 +90,46 @@ def test_a_disk_is_reused_by_its_key_of_its_containers_key_the_base_disk_and_the assert len(docker.disk_builds) > made +def test_a_disk_is_looked_up_made_and_pushed_holding_a_lock_of_its_key_so_composes_of_it_take_turns( + disk_base, driver, docker, monkeypatch): + held, events = set(), [] + real_locks, real_stored = composer.holding_locks, composer._stored + real_build, real_push = docker.build_image, docker.push + + @contextlib.contextmanager + def holding(ids, on_wait=None): + with real_locks(ids, on_wait): + held.update(ids) + try: + yield + finally: + held.difference_update(ids) + + def stored(env, key): + events.append(("looked up", key, set(held))) + return real_stored(env, key) + + def build(dockerfile, context, tag, **kw): + events.append(("made", tag, set(held))) + return real_build(dockerfile, context, tag, **kw) + + def push(tag, ref, store): + events.append(("pushed", tag, set(held))) + return real_push(tag, ref, store) + + monkeypatch.setattr(composer, "holding_locks", holding) + monkeypatch.setattr(composer, "_stored", stored) + monkeypatch.setattr(builder, "build_image", build) + monkeypatch.setattr(builder, "push_local_image", push) + + disks = Env.get(_put(disk_base, driver).id).metadata["composed"]["disk"] + + for arch, result in disks.items(): + names = (result["key"], f"agentenv-computer:{result['key'][7:23]}-disk-{arch}") + assert [(what, f"linux-disk:{result['key']}" in locks) for what, name, locks in events if name in names] == [ + ("looked up", True), ("made", True), ("pushed", True)] + + def test_a_bases_disk_size_is_in_the_key_and_the_build_of_the_disk_made_for_it(disk_base, driver, docker): bigger = _base(disk_base.image(), id="@local/~/kit/bigger", forms={ **disk_base.forms.model_dump(exclude_none=True), "disk": {**DISK_FORM, "min_disk_gb": 32}})