feat(compose): added ryuk cleanup functionality for compose projects

docs/features/docker_compose.md

@@ -39,6 +39,43 @@ compose = DockerCompose(
 )
 ```
 
+## Cleanup after process termination
+
+Normal context-manager exit stops the environment using Docker Compose. To also
+clean up when the Python process is abruptly terminated, opt in to Ryuk:
+
+```python
+with DockerCompose("path/to/compose/directory", ryuk=True) as compose:
+    # Run tests against the services.
+    pass
+```
+
+`ryuk=True` generates a unique project name for this `DockerCompose` object and
+registers its `com.docker.compose.project` label with the process's shared Ryuk
+container before running `compose up`. All commands use the generated project
+name, including when the same object is stopped and restarted. Separate objects
+use separate projects, even if they use the same Compose files.
+
+The generated name overrides `COMPOSE_PROJECT_NAME` and the Compose file's
+top-level `name`. Hard-coded container names, published ports, and explicitly
+named resources are not made unique; use a Compose file suitable for isolated,
+disposable environments. External networks and volumes should be managed
+separately and must not carry the generated project's cleanup label.
+
+With `keep_volumes=True`, leaving the context preserves the environment's volumes
+for reuse by the same object. Those resources **remain eligible for Ryuk cleanup**
+after the owning process loses its connection. This option does not guarantee
+persistence beyond the process's lifetime when `ryuk=True`.
+
+`TESTCONTAINERS_RYUK_DISABLED=true` disables Ryuk startup and registration, even
+with `ryuk=True`; the generated project name remains unchanged. When Ryuk is
+enabled, startup or filter-registration errors prevent `compose up` from running.
+Ryuk must have access to the same Docker daemon as the Compose command.
+
+Without `ryuk=True`, project naming and cleanup behavior are unchanged. Continue
+using a context manager or calling `stop()` for normal cleanup; Ryuk is a fallback
+for abrupt termination. Stopping one environment does not stop the shared Ryuk.
+
 ## Accessing Services
 
 You can access service information and interact with containers:

src/testcontainers/compose/compose.py

@@ -11,9 +11,11 @@
 from subprocess import run as subprocess_run
 from types import TracebackType
 from typing import Any, Callable, Literal, Optional, TypeVar, Union, cast
+from uuid import uuid4
 
 from typing_extensions import Self
 
+from testcontainers.core.config import testcontainers_config
 from testcontainers.core.docker_client import DockerClient, get_docker_host_hostname, is_podman
 from testcontainers.core.exceptions import ContainerIsNotRunning, NoSuchPortExposed
 from testcontainers.core.inspect import ContainerInspectInfo, _ignore_properties
@@ -225,6 +227,10 @@ class DockerCompose:
             Whether to suppress output when pulling images.
         quiet_build:
             Whether to suppress output when building images.
+        ryuk:
+            Run in an isolated project and register it for Ryuk cleanup before startup.
+            Defaults to False. TESTCONTAINERS_RYUK_DISABLED disables registration,
+            but the project remains isolated. Retained volumes are eligible for Ryuk cleanup.
 
     Example:
 
@@ -260,10 +266,14 @@ class DockerCompose:
     profiles: Optional[list[str]] = None
     quiet_pull: bool = False
     quiet_build: bool = False
+    ryuk: bool = False
+    _project_name: Optional[str] = field(default=None, init=False, repr=False)
     _wait_strategies: Optional[dict[str, Any]] = field(default=None, init=False, repr=False)
     _docker_client: Optional[DockerClient] = field(default=None, init=False, repr=False)
 
     def __post_init__(self) -> None:
+        if self.ryuk:
+            self._project_name = f"testcontainers-{uuid4().hex}"
         if isinstance(self.compose_file_name, str):
             self.compose_file_name = [self.compose_file_name]
         if isinstance(self.env_file, str):
@@ -295,6 +305,8 @@ def docker_compose_command(self) -> list[str]:
     def compose_command_property(self) -> list[str]:
         binary = self.docker_command_path or _default_compose_binary()
         docker_compose_cmd = [binary, "compose"]
+        if self._project_name is not None:
+            docker_compose_cmd += ["--project-name", self._project_name]
         if self.compose_file_name:
             for file in self.compose_file_name:
                 docker_compose_cmd += ["-f", file]
@@ -319,6 +331,12 @@ def start(self) -> None:
         """
         Starts the docker compose environment.
         """
+        if self._project_name is not None and not testcontainers_config.ryuk_disabled:
+            # container imports wait strategies, which in turn import compose.
+            from testcontainers.core.container import Reaper
+
+            Reaper.get_instance().register_labels_filter({"com.docker.compose.project": self._project_name})
+
         base_cmd = self.compose_command_property or []
 
         # pull means running a separate command before starting

src/testcontainers/core/container.py

@@ -7,8 +7,11 @@
 from dataclasses import dataclass
 from os import PathLike
 from socket import socket
+from threading import Lock
+from time import monotonic
 from types import TracebackType
 from typing import TYPE_CHECKING, Any, Optional, TypedDict, Union
+from urllib.parse import urlencode
 
 import docker.errors
 from docker import version
@@ -465,6 +468,47 @@ class Reaper:
     _instance: "Optional[Reaper]" = None
     _container: Optional[DockerContainer] = None
     _socket: Optional[socket] = None
+    _ACK_TIMEOUT = 10.0
+
+    def __init__(self) -> None:
+        self._filter_lock = Lock()
+        self._registration_failed = False
+
+    def register_labels_filter(self, labels: dict[str, str]) -> None:
+        """Register an additional cleanup filter and wait for Ryuk to acknowledge it."""
+        if not labels:
+            raise ValueError("A Ryuk cleanup filter must contain at least one label")
+
+        with self._filter_lock:
+            rs = Reaper._socket
+            if rs is None or self._registration_failed:
+                raise ConnectionError("Ryuk connection is unavailable for filter registration")
+
+            message = urlencode([("label", f"{key}={value}") for key, value in labels.items()])
+            previous_timeout = rs.gettimeout()
+            deadline = monotonic() + self._ACK_TIMEOUT
+            try:
+                rs.settimeout(self._ACK_TIMEOUT)
+                rs.sendall((message + "\n").encode())
+                response = b""
+                while len(response) < len(b"ACK\n"):
+                    remaining = deadline - monotonic()
+                    if remaining <= 0:
+                        raise TimeoutError("Timed out waiting for Ryuk to acknowledge a cleanup filter")
+                    rs.settimeout(remaining)
+                    chunk = rs.recv(len(b"ACK\n") - len(response))
+                    if not chunk:
+                        raise ConnectionError("Ryuk disconnected before acknowledging a cleanup filter")
+                    response += chunk
+                if response != b"ACK\n":
+                    raise ConnectionError(f"Unexpected acknowledgement from Ryuk: {response!r}")
+            except OSError:
+                # A late ACK must not be mistaken for the next filter's ACK. Keep
+                # the socket open so existing environments are not reaped early.
+                self._registration_failed = True
+                raise
+            finally:
+                rs.settimeout(previous_timeout)
 
     @classmethod
     def get_instance(cls) -> "Reaper":
@@ -536,11 +580,14 @@ def _create_instance(cls) -> "Reaper":
         if last_connection_exception:
             raise last_connection_exception
 
-        rs = Reaper._socket
-        assert rs is not None
-        rs.send(f"label={LABEL_SESSION_ID}={SESSION_ID}\r\n".encode())
+        instance = Reaper()
+        try:
+            instance.register_labels_filter({LABEL_SESSION_ID: SESSION_ID})
+        except OSError:
+            Reaper.delete_instance()
+            raise
 
-        Reaper._instance = Reaper()
+        Reaper._instance = instance
         atexit.register(Reaper.delete_instance)
 
         return Reaper._instance

tests/core/compose_fixtures/ryuk/compose.yaml

@@ -0,0 +1,21 @@
+name: testcontainers-ryuk-fixture
+services:
+  service:
+    image: alpine:3.20
+    init: true
+    command: ["sleep", "300"]
+    volumes:
+      - data:/data
+      - external_data:/external
+    networks:
+      - default
+      - external_network
+volumes:
+  data: {}
+  external_data:
+    external: true
+    name: ${TC_RYUK_EXTERNAL_VOLUME}
+networks:
+  external_network:
+    external: true
+    name: ${TC_RYUK_EXTERNAL_NETWORK}

tests/core/compose_fixtures/ryuk/start_compose.py

@@ -0,0 +1,53 @@
+"""Compose-only child process used by the abrupt-termination regression tests."""
+
+import json
+import sys
+from pathlib import Path
+from time import sleep
+
+from testcontainers.compose import DockerCompose
+from testcontainers.core.container import Reaper
+from testcontainers.core.labels import SESSION_ID
+from testcontainers.core.waiting_utils import WaitStrategy
+
+
+class FailedReadiness(WaitStrategy):
+    def wait_until_ready(self, container) -> None:
+        raise TimeoutError("Simulated failure after Compose created resources")
+
+
+def main() -> None:
+    ready = Path(sys.argv[1])
+    scenario = sys.argv[2]
+    compose = DockerCompose(Path(__file__).parent, ryuk=True, keep_volumes=True)
+    ready.with_suffix(".state.json").write_text(
+        json.dumps({"project": compose._project_name, "ryuk": f"testcontainers-ryuk-{SESSION_ID}"})
+    )
+    if scenario == "partial":
+        compose.waiting_for({"service": FailedReadiness()})
+        try:
+            compose.start()
+        except TimeoutError:
+            pass
+        else:
+            raise AssertionError("Readiness should have failed")
+    else:
+        compose.start()
+    if scenario == "retained":
+        compose.exec_in_container(["sh", "-c", "echo retained > /data/value"])
+        compose.stop(down=False)
+        with compose:
+            assert compose.exec_in_container(["cat", "/data/value"])[0].strip() == "retained"
+
+    container = compose.get_container(include_all=True)
+    assert Reaper._container is not None
+    state = {"project": container.Project, "ryuk": Reaper._container.get_container_id()}
+    temporary = ready.with_suffix(".tmp")
+    temporary.write_text(json.dumps(state))
+    temporary.replace(ready)
+    while True:
+        sleep(1)
+
+
+if __name__ == "__main__":
+    main()