test(integration-tests): stop Python reaper flake on a slow removal

server/csharp/MetaObjects.IntegrationTests/ReaperKillTests.cs

@@ -16,7 +16,11 @@ namespace MetaObjects.IntegrationTests;
 public sealed class ReaperKillTests
 {
     private static readonly TimeSpan ChildReady = TimeSpan.FromSeconds(60);
-    private static readonly TimeSpan RemovalBound = TimeSpan.FromSeconds(45);
+    // Ryuk waits 10s, then Docker has to finish the container's exit, and it cannot until the overlay
+    // unmount has flushed the host filesystem's dirty pages. A busy disk makes removal slow, not wrong.
+    // Measured with a 256 MiB backlog: 34-38s at 12 MB/s and 67-76s at 6 MB/s, against under 11s at
+    // 6 MB/s with no backlog. Every removal succeeded.
+    private static readonly TimeSpan RemovalBound = TimeSpan.FromSeconds(120);
 
     // The killed process: it starts a real helper container, reports the port and waits. This is
     // the test assembly's entry point, which GenerateProgramFile=false leaves free.
@@ -49,7 +53,7 @@ public async Task Killed_process_removes_its_container()
             {
                 await Task.Delay(250);
             }
-            Assert.False(await ExistsAsync(container), $"{container} outlived its killed process");
+            Assert.False(await ExistsAsync(container), $"{container} outlived its killed process ({await StatusAsync(container)})");
         }
         finally
         {
@@ -72,6 +76,9 @@ private static async Task<string> AwaitPortAsync(Process child)
     private static async Task<bool> ExistsAsync(string container) =>
         await DockerAsync("ps", "-a", "--filter", $"name=^{container}$", "--format", "{{.Names}}") != "";
 
+    private static Task<string> StatusAsync(string container) =>
+        DockerAsync("ps", "-a", "--filter", $"name=^{container}$", "--format", "{{.Status}}");
+
     private static async Task<string> DockerAsync(params string[] args)
     {
         var start = new ProcessStartInfo("docker") { RedirectStandardOutput = true, RedirectStandardError = true };

server/java/integration-tests-kotlin/src/test/kotlin/com/metaobjects/integration/kotlin/ReaperKillTest.kt

@@ -38,7 +38,7 @@ class ReaperKillTest {
 
             val deadline = System.currentTimeMillis() + REMOVAL_BOUND_MS
             while (exists(container) && System.currentTimeMillis() < deadline) Thread.sleep(250)
-            assertFalse(exists(container), "$container outlived its killed JVM")
+            assertFalse(exists(container), "$container outlived its killed JVM (${status(container)})")
         } finally {
             child.destroyForcibly()
             if (container.isNotEmpty()) docker("rm", "-f", "-v", container)
@@ -58,6 +58,9 @@ class ReaperKillTest {
     private fun exists(container: String): Boolean =
         docker("ps", "-a", "--filter", "name=^$container$", "--format", "{{.Names}}").isNotEmpty()
 
+    private fun status(container: String): String =
+        docker("ps", "-a", "--filter", "name=^$container$", "--format", "{{.Status}}")
+
     private fun docker(vararg args: String): String {
         val p = ProcessBuilder("docker", *args).redirectErrorStream(true).start()
         if (!p.waitFor(30, TimeUnit.SECONDS)) {
@@ -72,7 +75,11 @@ class ReaperKillTest {
     companion object {
         private val READY = Regex("^READY (\\d+)$", RegexOption.MULTILINE)
         private const val CHILD_READY_MS = 60_000L
-        private const val REMOVAL_BOUND_MS = 45_000L
+        // Ryuk waits 10s, then Docker has to finish the container's exit, and it cannot until the
+        // overlay unmount has flushed the host filesystem's dirty pages. A busy disk makes removal
+        // slow, not wrong. Measured with a 256 MiB backlog: 34-38s at 12 MB/s and 67-76s at 6 MB/s,
+        // against under 11s at 6 MB/s with no backlog. Every removal succeeded.
+        private const val REMOVAL_BOUND_MS = 120_000L
 
         /** The killed JVM: it starts a real helper container, reports the port and waits. */
         @JvmStatic

server/java/integration-tests/src/test/java/com/metaobjects/integration/ReaperKillTest.java

@@ -26,7 +26,11 @@
 public final class ReaperKillTest {
     private static final Pattern READY = Pattern.compile("^READY (\\d+)$", Pattern.MULTILINE);
     private static final long CHILD_READY_MS = 60_000;
-    private static final long REMOVAL_BOUND_MS = 45_000;
+    // Ryuk waits 10s, then Docker has to finish the container's exit, and it cannot until the overlay
+    // unmount has flushed the host filesystem's dirty pages. A busy disk makes removal slow, not wrong.
+    // Measured with a 256 MiB backlog: 34-38s at 12 MB/s and 67-76s at 6 MB/s, against under 11s at
+    // 6 MB/s with no backlog. Every removal succeeded.
+    private static final long REMOVAL_BOUND_MS = 120_000;
 
     /** The killed JVM: it starts a real helper container, reports the port and waits. */
     public static void main(String[] args) throws Exception {
@@ -56,7 +60,7 @@ void killedJvmRemovesItsContainer() throws Exception {
 
             long deadline = System.currentTimeMillis() + REMOVAL_BOUND_MS;
             while (exists(container) && System.currentTimeMillis() < deadline) Thread.sleep(250);
-            assertFalse(exists(container), container + " outlived its killed JVM");
+            assertFalse(exists(container), container + " outlived its killed JVM (" + status(container) + ")");
         } finally {
             child.destroyForcibly();
             if (!container.isEmpty()) docker("rm", "-f", "-v", container);
@@ -78,6 +82,10 @@ private static boolean exists(String container) throws IOException, InterruptedE
         return !docker("ps", "-a", "--filter", "name=^" + container + "$", "--format", "{{.Names}}").isEmpty();
     }
 
+    private static String status(String container) throws IOException, InterruptedException {
+        return docker("ps", "-a", "--filter", "name=^" + container + "$", "--format", "{{.Status}}");
+    }
+
     private static String docker(String... args) throws IOException, InterruptedException {
         String[] command = new String[args.length + 1];
         command[0] = "docker";

server/python/pyproject.toml

@@ -54,10 +54,10 @@ dev = ["pytest>=8", "pytest-xdist>=3.6", "mypy>=1.10", "ruff>=0.6", "pydantic>=2
 # On-demand cross-language conformance suites — persistence + api-contract.
 # pg8000 is pure-python so this extra installs cleanly without native deps;
 # the persistence harness shells out to the docker CLI directly (matches the
-# TS/Java pattern); testcontainers supplies only Ryuk, which removes a killed
-# run's containers (tests/integration/reaper.py); the api-contract harness drives a FastAPI Author API in-process via the
-# starlette TestClient (httpx).
-integration = ["pg8000>=1.31", "PyMySQL>=1.1", "fastapi>=0.110", "httpx>=0.27", "testcontainers>=4.15"]
+# TS/Java pattern), and tests/integration/reaper.py starts Ryuk the same way to remove a
+# killed run's containers; the api-contract harness drives a FastAPI Author API in-process
+# via the starlette TestClient (httpx).
+integration = ["pg8000>=1.31", "PyMySQL>=1.1", "fastapi>=0.110", "httpx>=0.27"]
 # MySQL for the ObjectManager (`metaobjects.runtime.MySqlDriver`). PyMySQL is pure Python,
 # so like pg8000 it installs without native build dependencies. SQLite needs nothing: the
 # driver wraps the standard library's `sqlite3`.

server/python/tests/integration/mysql_container.py

@@ -1,7 +1,7 @@
 """A throwaway MySQL 8.4 server for the MySQL conformance lanes.
 
-Shells out to the ``docker`` CLI exactly like :mod:`postgres_container`, with the
-testcontainers package supplying only Ryuk (see :mod:`reaper`). One server is started for a whole pytest session (see ``conftest.py``); every scenario starts
+Shells out to the ``docker`` CLI exactly like :mod:`postgres_container`, with Ryuk
+removing it when a killed run never reaches teardown (see :mod:`reaper`). One server is started for a whole pytest session (see ``conftest.py``); every scenario starts
 from an EMPTY schema via :func:`reset_database`, which is what the Java and Kotlin lanes do.
 
 Set ``METAOBJECTS_TEST_MYSQL_URL`` (``mysql://user:password@host:port/database``) to run

server/python/tests/integration/postgres_container.py

@@ -9,8 +9,8 @@
      gets a pristine empty database, so isolation is identical to the per-
      container path.
   2. Per-container (local dev): with no env var, ``docker run`` a fresh container,
-     exactly as before. The testcontainers package supplies only Ryuk, which removes
-     the container when a killed run never reaches its teardown (see :mod:`reaper`).
+     exactly as before. Ryuk removes the container when a killed run never reaches its
+     teardown (see :mod:`reaper`).
 """
 from __future__ import annotations
 

server/python/tests/integration/reaper.py

@@ -2,25 +2,107 @@
 
 A fixture's teardown cannot run in a killed process: SIGKILL, the kernel OOM killer and earlyoom
 all skip it, and a plain SIGTERM ends Python without running it. Ryuk closes that gap from outside
-the process. It runs in its own container and holds one TCP connection per session. When that
-connection closes, for any reason, it waits ten seconds and removes everything carrying the
-session's label. The operating system closes the socket of a killed process, so no code in the
-dying process has to run.
+the process. It runs in its own container and holds one TCP connection from each client. When the
+last connection closes, for any reason, it waits ten seconds and removes everything carrying the
+session labels it was given. The operating system closes the socket of a killed process, so no
+code in the dying process has to run.
 
-Each process gets its own session, so an xdist worker's death reaps that worker's containers and
-no other run's. ``TESTCONTAINERS_RYUK_DISABLED=true`` turns the reaper off.
+Each process starts a private Ryuk here, so an xdist worker's death reaps that worker's containers
+and no other run's. ``TESTCONTAINERS_RYUK_DISABLED=true`` turns the reaper off.
+
+This is not the testcontainers package's ``Reaper``, for two reasons. That class defaults to Ryuk
+0.8.1, which tries each removal once, drops the error and exits, so a single failed ``rm -f``
+leaves the container running for good while its log says it was removed. It cannot run a newer
+image, because it waits for a log line 0.10 and later no longer print. It also never reads Ryuk's
+ACK, so it cannot tell a registered session from a connection that went nowhere. Ryuk 0.14.0
+retries a failed removal, and :func:`_register` waits for the ACK.
+
+A Ryuk that no client reaches stops by itself after a minute, and ``--rm`` removes it.
 """
 from __future__ import annotations
 
+import os
+import socket
+import subprocess
+import time
+import uuid
+
+RYUK_IMAGE = "testcontainers/ryuk:0.14.0"
+SESSION_LABEL = "org.testcontainers.session-id"
+DOCKER_SOCKET_ENV = "TESTCONTAINERS_DOCKER_SOCKET_OVERRIDE"
+DEFAULT_DOCKER_SOCKET = "/var/run/docker.sock"
+#: Long enough to pull the image once.
+DOCKER_RUN_TIMEOUT_S = 180
+REGISTER_DEADLINE_S = 15
+REGISTER_ATTEMPT_TIMEOUT_S = 5
+REGISTER_RETRY_DELAY_S = 0.2
+
+_label_args: list[str] | None = None
+#: Held here for the life of the process: losing this connection ends the session.
+_ryuk_connection: socket.socket | None = None
+#: The container this process started, so a test can read its log.
+ryuk_name: str | None = None
+
+
+def docker_socket() -> str:
+    """The daemon socket Ryuk is handed, which is the path as the daemon sees it."""
+    return os.environ.get(DOCKER_SOCKET_ENV, DEFAULT_DOCKER_SOCKET)
+
+
+def _register(port: int, label: str) -> socket.socket | None:
+    """One registration attempt. Docker publishes the port before Ryuk listens, so a refused or
+    reset connection only means "not yet". Only the ACK proves Ryuk holds the label."""
+    try:
+        conn = socket.create_connection(("127.0.0.1", port), timeout=REGISTER_ATTEMPT_TIMEOUT_S)
+    except OSError:
+        return None
+    try:
+        conn.sendall(f"label={label}\n".encode())
+        with conn.makefile("rb") as replies:
+            if replies.readline() == b"ACK\n":
+                return conn
+    except OSError:
+        pass
+    conn.close()
+    return None
+
+
+def _start_reaper() -> list[str]:
+    global _ryuk_connection, ryuk_name
+    # Imported here because postgres_container imports this module.
+    from .postgres_container import published_port
+
+    session = uuid.uuid4().hex
+    name = f"metaobjects-test-ryuk-{session[:8]}"
+    label = f"{SESSION_LABEL}={session}"
+    run = subprocess.run(
+        ["docker", "run", "-d", "--rm", "--name", name,
+         "-v", f"{docker_socket()}:{DEFAULT_DOCKER_SOCKET}",
+         "-p", "127.0.0.1::8080",
+         RYUK_IMAGE],
+        capture_output=True, text=True, timeout=DOCKER_RUN_TIMEOUT_S,
+    )
+    failure = f"docker run failed: {run.stderr}"
+    if run.returncode == 0:
+        failure = f"no ACK within {REGISTER_DEADLINE_S}s"
+        port = published_port(name, 8080)
+        deadline = time.monotonic() + REGISTER_DEADLINE_S
+        while time.monotonic() < deadline:
+            _ryuk_connection = _register(port, label)
+            if _ryuk_connection is not None:
+                ryuk_name = name
+                return ["--label", label]
+            time.sleep(REGISTER_RETRY_DELAY_S)
+    subprocess.run(["docker", "rm", "-f", name], capture_output=True, timeout=30)
+    raise RuntimeError(f"could not start Ryuk ({RYUK_IMAGE}): {failure}")
+
 
 def reaper_label_args() -> list[str]:
-    """Flags for ``docker run``. The first call starts Ryuk, and must finish before the container
-    starts, so a process killed at any later moment is covered."""
-    # Imported here because the sidecar path never starts a container and should not load docker.
-    from testcontainers.core.config import testcontainers_config
-    from testcontainers.core.container import Reaper
-    from testcontainers.core.labels import LABEL_SESSION_ID, SESSION_ID
-
-    if not testcontainers_config.ryuk_disabled:
-        Reaper.get_instance()
-    return ["--label", f"{LABEL_SESSION_ID}={SESSION_ID}"]
+    """Flags for ``docker run``. The first call starts this process's Ryuk, and must finish before
+    the container starts, so a process killed at any later moment is covered."""
+    global _label_args
+    if os.environ.get("TESTCONTAINERS_RYUK_DISABLED", "").lower() == "true":
+        return []
+    if _label_args is None:
+        _label_args = _start_reaper()
+    return _label_args