Fix AB-BA deadlock between DockerClientFactory and RyukResourceReaper

core/src/main/java/org/testcontainers/utility/RyukResourceReaper.java

@@ -33,6 +33,8 @@ class RyukResourceReaper extends ResourceReaper {
 
     private final AtomicBoolean started = new AtomicBoolean(false);
 
+    private volatile boolean ready = false;
+
     private final RyukContainer ryukContainer = new RyukContainer();
 
     @Override
@@ -70,66 +72,81 @@ public CreateContainerCmd register(GenericContainer<?> container, CreateContaine
     }
 
     @SneakyThrows(InterruptedException.class)
-    private synchronized void maybeStart() {
-        if (!started.compareAndSet(false, true)) {
+    private void maybeStart() {
+        if (ready) {
             return;
         }
 
-        ryukContainer.start();
-
-        CountDownLatch ryukScheduledLatch = new CountDownLatch(1);
-
-        String host = ryukContainer.getHost();
-        Integer ryukPort = ryukContainer.getFirstMappedPort();
-        Thread kiraThread = new Thread(
-            DockerClientFactory.TESTCONTAINERS_THREAD_GROUP,
-            () -> {
-                while (true) {
-                    RYUK_ACK_RATE_LIMITER.doWhenReady(() -> {
-                        int index = 0;
-                        // not set the read timeout, as Ryuk would not send anything unless a new filter is submitted, meaning that we would get a timeout exception pretty quick
-                        try (Socket clientSocket = new Socket()) {
-                            clientSocket.connect(new InetSocketAddress(host, ryukPort), 5 * 1000);
-                            ResourceReaper.FilterRegistry registry = new ResourceReaper.FilterRegistry(
-                                clientSocket.getInputStream(),
-                                clientSocket.getOutputStream()
-                            );
-
-                            synchronized (ResourceReaper.DEATH_NOTE) {
-                                while (true) {
-                                    if (ResourceReaper.DEATH_NOTE.size() <= index) {
-                                        try {
-                                            ResourceReaper.DEATH_NOTE.wait(1_000);
-                                            continue;
-                                        } catch (InterruptedException e) {
-                                            throw new RuntimeException(e);
+        DockerClientFactory.instance().client();
+
+        synchronized (this) {
+            if (ready) {
+                return;
+            }
+            if (!started.compareAndSet(false, true)) {
+                return;
+            }
+
+            ryukContainer.start();
+
+            CountDownLatch ryukScheduledLatch = new CountDownLatch(1);
+
+            String host = ryukContainer.getHost();
+            Integer ryukPort = ryukContainer.getFirstMappedPort();
+            Thread kiraThread = new Thread(
+                DockerClientFactory.TESTCONTAINERS_THREAD_GROUP,
+                () -> {
+                    while (true) {
+                        RYUK_ACK_RATE_LIMITER.doWhenReady(() -> {
+                            int index = 0;
+                            // not set the read timeout, as Ryuk would not send anything unless a new filter is submitted, meaning that we would get a timeout exception pretty quick
+                            try (Socket clientSocket = new Socket()) {
+                                clientSocket.connect(new InetSocketAddress(host, ryukPort), 5 * 1000);
+                                ResourceReaper.FilterRegistry registry = new ResourceReaper.FilterRegistry(
+                                    clientSocket.getInputStream(),
+                                    clientSocket.getOutputStream()
+                                );
+
+                                synchronized (ResourceReaper.DEATH_NOTE) {
+                                    while (true) {
+                                        if (ResourceReaper.DEATH_NOTE.size() <= index) {
+                                            try {
+                                                ResourceReaper.DEATH_NOTE.wait(1_000);
+                                                continue;
+                                            } catch (InterruptedException e) {
+                                                throw new RuntimeException(e);
+                                            }
+                                        }
+                                        List<Map.Entry<String, String>> filters = ResourceReaper.DEATH_NOTE.get(index);
+                                        boolean isAcknowledged = registry.register(filters);
+                                        if (isAcknowledged) {
+                                            log.debug("Received 'ACK' from Ryuk");
+                                            ryukScheduledLatch.countDown();
+                                            index++;
+                                        } else {
+                                            log.debug("Didn't receive 'ACK' from Ryuk. Will retry to send filters.");
                                         }
-                                    }
-                                    List<Map.Entry<String, String>> filters = ResourceReaper.DEATH_NOTE.get(index);
-                                    boolean isAcknowledged = registry.register(filters);
-                                    if (isAcknowledged) {
-                                        log.debug("Received 'ACK' from Ryuk");
-                                        ryukScheduledLatch.countDown();
-                                        index++;
-                                    } else {
-                                        log.debug("Didn't receive 'ACK' from Ryuk. Will retry to send filters.");
                                     }
                                 }
+                            } catch (IOException e) {
+                                log.warn("Can not connect to Ryuk at {}:{}", host, ryukPort, e);
                             }
-                        } catch (IOException e) {
-                            log.warn("Can not connect to Ryuk at {}:{}", host, ryukPort, e);
-                        }
-                    });
-                }
-            },
-            "testcontainers-ryuk"
-        );
-        kiraThread.setDaemon(true);
-        kiraThread.start();
-        // We need to wait before we can start any containers to make sure that we delete them
-        if (!ryukScheduledLatch.await(TestcontainersConfiguration.getInstance().getRyukTimeout(), TimeUnit.SECONDS)) {
-            log.error("Timed out waiting for Ryuk container to start. Ryuk's logs:\n{}", ryukContainer.getLogs());
-            throw new IllegalStateException(String.format("Could not connect to Ryuk at %s:%s", host, ryukPort));
+                        });
+                    }
+                },
+                "testcontainers-ryuk"
+            );
+            kiraThread.setDaemon(true);
+            kiraThread.start();
+            // We need to wait before we can start any containers to make sure that we delete them
+            if (
+                !ryukScheduledLatch.await(TestcontainersConfiguration.getInstance().getRyukTimeout(), TimeUnit.SECONDS)
+            ) {
+                log.error("Timed out waiting for Ryuk container to start. Ryuk's logs:\n{}", ryukContainer.getLogs());
+                throw new IllegalStateException(String.format("Could not connect to Ryuk at %s:%s", host, ryukPort));
+            }
+
+            ready = true;
         }
     }
 }

core/src/test/java/org/testcontainers/RyukResourceReaperDeadlockTest.java

@@ -0,0 +1,85 @@
+package org.testcontainers;
+
+import org.junit.jupiter.api.Test;
+import org.testcontainers.utility.ResourceReaper;
+
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class RyukResourceReaperDeadlockTest {
+
+    @Test
+    void shouldNotHoldReaperLockWhenCallingDockerClientFactory() {
+        ResourceReaper reaper = ResourceReaper.instance();
+        AtomicBoolean heldLock = new AtomicBoolean(true);
+
+        DockerClientFactory dockerClientFactory = DockerClientFactory.instance();
+        RuntimeException simulatedFailure = new RuntimeException("Simulated failure to intercept call") {
+            @Override
+            public synchronized Throwable fillInStackTrace() {
+                heldLock.set(Thread.holdsLock(reaper));
+                return this;
+            }
+        };
+
+        dockerClientFactory.cachedClientFailure = simulatedFailure;
+        try {
+            assertThatThrownBy(() -> reaper.registerLabelsFilterForCleanup(Collections.singletonMap("k", "v")))
+                .isSameAs(simulatedFailure);
+
+            assertThat(heldLock.get())
+                .as("RyukResourceReaper monitor must NOT be held when calling DockerClientFactory to prevent deadlocks")
+                .isFalse();
+        } finally {
+            dockerClientFactory.cachedClientFailure = null;
+        }
+    }
+
+    @Test
+    void concurrentCallersShouldNotDeadlockUnderContention() throws InterruptedException {
+        ResourceReaper reaper = ResourceReaper.instance();
+        DockerClientFactory dockerClientFactory = DockerClientFactory.instance();
+        RuntimeException simulatedFailure = new RuntimeException("Simulated client error");
+
+        dockerClientFactory.cachedClientFailure = simulatedFailure;
+        try {
+            int threadCount = 8;
+            ExecutorService executor = Executors.newFixedThreadPool(threadCount);
+            CountDownLatch startLatch = new CountDownLatch(1);
+            CountDownLatch doneLatch = new CountDownLatch(threadCount);
+
+            for (int i = 0; i < threadCount; i++) {
+                final int id = i;
+                executor.submit(() -> {
+                    try {
+                        startLatch.await(2, TimeUnit.SECONDS);
+                        if (id % 2 == 0) {
+                            reaper.registerLabelsFilterForCleanup(Collections.singletonMap("test", "val" + id));
+                        } else {
+                            reaper.getLabels();
+                        }
+                    } catch (Throwable ignored) {} finally {
+                        doneLatch.countDown();
+                    }
+                });
+            }
+
+            startLatch.countDown();
+            boolean finished = doneLatch.await(5, TimeUnit.SECONDS);
+            executor.shutdownNow();
+
+            assertThat(finished)
+                .as("Concurrent calls to RyukResourceReaper must finish within timeout and never deadlock")
+                .isTrue();
+        } finally {
+            dockerClientFactory.cachedClientFailure = null;
+        }
+    }
+}