Fix reliability: container cleanup, wait strategies, and Ryuk socket (#243)

* Fix reliability: container cleanup, wait strategies, Ryuk socket, and ToxiproxyContainer

- Retry Ryuk socket on all transient errors, not just econnrefused
- Wrap Ryuk filter registration in retry loop for closed errors
- Track containers in GenServer state, clean up on terminate and wait strategy failure
- Make image cleanup opt-in via cleanup.images property
- Switch ToxiproxyContainer from PortWaitStrategy to HttpWaitStrategy
- Add retry logic with backoff to ToxiproxyContainer httpc API calls
- Fix HttpWaitStrategy response validation and retry logic
- Reduce LogWaitStrategy retry delay from 500ms to 200ms
- Set global 300s test timeout, remove per-module overrides
- Add tests for HttpWaitStrategy, terminate cleanup, and wait strategy failure

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Add property-controlled pull policy for offline support

Allow configuring the default image pull behavior via `pull.policy`
property (or `TESTCONTAINERS_PULL_POLICY` env var). Set to "missing"
or "never" to skip pulling images that already exist locally, enabling
offline usage. Per-container pull_policy set via `with_pull_policy/2`
still takes precedence over the global setting.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Fix pull policy: use merged properties for env var support, apply to Ryuk

- Fix state.properties being rebound to read_property_file() (user file
  only), losing env vars. Now uses read_property_sources() which merges
  env vars, user file, and project file with proper precedence.
- Apply pull policy to Ryuk container so it also respects the setting.
- Tag pull_policy_test.exs with :needs_registry for offline exclusion.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: jarlah

.gitignore

@@ -38,3 +38,6 @@ testcontainers_elixir-*.tar
 .mix/
 .hex/
 .nix-mix/
+
+# test runs
+test_run_*.log

lib/container.ex

@@ -29,7 +29,7 @@ defmodule Testcontainers.Container do
     hostname: nil,
     reuse: false,
     force_reuse: false,
-    pull_policy: Testcontainers.PullPolicy.always_pull()
+    pull_policy: nil
   ]
 
   @doc """

lib/container/toxiproxy_container.ex

@@ -10,7 +10,7 @@ defmodule Testcontainers.ToxiproxyContainer do
 
   alias Testcontainers.Container
   alias Testcontainers.ContainerBuilder
-  alias Testcontainers.PortWaitStrategy
+  alias Testcontainers.HttpWaitStrategy
   alias Testcontainers.ToxiproxyContainer
 
   @default_image "ghcr.io/shopify/toxiproxy"
@@ -24,6 +24,8 @@ defmodule Testcontainers.ToxiproxyContainer do
   @proxy_port_count 31
 
   @default_wait_timeout 60_000
+  @max_retries 3
+  @retry_delay_ms 500
 
   @enforce_keys [:image, :wait_timeout]
   defstruct [:image, :wait_timeout, check_image: @default_image, reuse: false]
@@ -144,7 +146,7 @@ defmodule Testcontainers.ToxiproxyContainer do
 
     headers = [{~c"content-type", ~c"application/json"}]
 
-    case :httpc.request(:post, {url, headers, ~c"application/json", body}, [], []) do
+    case httpc_request_with_retry(:post, {url, headers, ~c"application/json", body}) do
       {:ok, {{_, code, _}, _, _}} when code in [200, 201] ->
         # Return the mapped port on the host
         {:ok, Container.mapped_port(container, listen_port)}
@@ -198,7 +200,7 @@ defmodule Testcontainers.ToxiproxyContainer do
 
     url = ~c"http://#{host}:#{api_port}/proxies/#{name}"
 
-    case :httpc.request(:delete, {url, []}, [], []) do
+    case httpc_request_with_retry(:delete, {url, []}) do
       {:ok, {{_, 204, _}, _, _}} -> :ok
       {:ok, {{_, 404, _}, _, _}} -> {:error, :not_found}
       {:ok, {{_, code, _}, _, body}} -> {:error, {:http_error, code, body}}
@@ -217,7 +219,7 @@ defmodule Testcontainers.ToxiproxyContainer do
 
     url = ~c"http://#{host}:#{api_port}/reset"
 
-    case :httpc.request(:post, {url, [], ~c"application/json", "{}"}, [], []) do
+    case httpc_request_with_retry(:post, {url, [], ~c"application/json", "{}"}) do
       {:ok, {{_, 204, _}, _, _}} -> :ok
       {:ok, {{_, code, _}, _, body}} -> {:error, {:http_error, code, body}}
       {:error, reason} -> {:error, reason}
@@ -237,7 +239,7 @@ defmodule Testcontainers.ToxiproxyContainer do
 
     url = ~c"http://#{host}:#{api_port}/proxies"
 
-    case :httpc.request(:get, {url, []}, [], []) do
+    case httpc_request_with_retry(:get, {url, []}) do
       {:ok, {{_, 200, _}, _, body}} ->
         {:ok, Jason.decode!(to_string(body))}
 
@@ -254,6 +256,33 @@ defmodule Testcontainers.ToxiproxyContainer do
   """
   def proxy_port_count, do: @proxy_port_count
 
+  defp httpc_request_with_retry(method, request, retries_left \\ @max_retries) do
+    http_opts = [timeout: 5_000, connect_timeout: 5_000]
+
+    result = :httpc.request(method, request, http_opts, [])
+
+    case result do
+      {:error, reason} when retries_left > 0 ->
+        retryable =
+          case reason do
+            :socket_closed_remotely -> true
+            {:failed_connect, _} -> true
+            :econnrefused -> true
+            _ -> false
+          end
+
+        if retryable do
+          Process.sleep(@retry_delay_ms)
+          httpc_request_with_retry(method, request, retries_left - 1)
+        else
+          result
+        end
+
+      _ ->
+        result
+    end
+  end
+
   # ContainerBuilder implementation
   defimpl ContainerBuilder do
     import Container
@@ -272,10 +301,10 @@ defmodule Testcontainers.ToxiproxyContainer do
       new(config.image)
       |> with_exposed_ports(all_ports)
       |> with_waiting_strategy(
-        PortWaitStrategy.new(
-          "127.0.0.1",
+        HttpWaitStrategy.new(
+          "/version",
           ToxiproxyContainer.control_port(),
-          config.wait_timeout
+          timeout: config.wait_timeout
         )
       )
       |> with_reuse(config.reuse)

lib/docker/api.ex

@@ -64,6 +64,13 @@ defmodule Testcontainers.Docker.Api do
     end
   end
 
+  def delete_image(image, conn) when is_binary(image) do
+    case Api.Image.image_delete(conn, image, force: true) do
+      {:ok, _} -> :ok
+      {:error, _} = error -> error
+    end
+  end
+
   def create_container(%Container{} = container, conn) do
     case Api.Container.container_create(conn, container_create_request(container)) do
       {:error, %Tesla.Env{status: other}} ->

lib/testcontainers.ex

@@ -15,6 +15,7 @@ defmodule Testcontainers do
   alias Testcontainers.Connection
   alias Testcontainers.Container
   alias Testcontainers.ContainerBuilder
+  alias Testcontainers.PullPolicy
   alias Testcontainers.Util.PropertiesParser
 
   import Testcontainers.Constants
@@ -58,8 +59,7 @@ defmodule Testcontainers do
       |> Base.encode16()
 
     with {:ok, docker_hostname} <- get_docker_hostname(docker_host_url, conn),
-         {:ok} <- start_reaper(conn, session_id, properties, docker_host, docker_hostname),
-         {:ok, properties} <- PropertiesParser.read_property_file() do
+         {:ok} <- start_reaper(conn, session_id, properties, docker_host, docker_hostname) do
       Logger.info("Testcontainers initialized")
 
       {:ok,
@@ -68,7 +68,9 @@ defmodule Testcontainers do
          docker_hostname: docker_hostname,
          session_id: session_id,
          properties: properties,
-         networks: MapSet.new()
+         networks: MapSet.new(),
+         containers: MapSet.new(),
+         images: MapSet.new()
        }}
     else
       error ->
@@ -170,15 +172,70 @@ defmodule Testcontainers do
     wait_for_call({:remove_network, network_name}, name)
   end
 
+  @impl true
+  def handle_cast({:track_container, container_id, image}, state) do
+    {:noreply,
+     %{
+       state
+       | containers: MapSet.put(state.containers, container_id),
+         images: MapSet.put(state.images, image)
+     }}
+  end
+
+  def handle_cast({:track_image, image}, state) do
+    {:noreply, %{state | images: MapSet.put(state.images, image)}}
+  end
+
   @impl true
   def handle_info(_msg, state) do
     {:noreply, state}
   end
 
+  @impl true
+  def terminate(_reason, state) do
+    for container_id <- state.containers do
+      Logger.info("Terminating container #{container_id}")
+      Api.stop_container(container_id, state.conn)
+    end
+
+    for network <- state.networks do
+      Logger.info("Removing network #{network}")
+      Api.remove_network(network, state.conn)
+    end
+
+    if Map.get(state.properties, "cleanup.images", "false") == "true" do
+      for image <- state.images do
+        Logger.info("Removing image #{image}")
+        Api.delete_image(image, state.conn)
+      end
+    end
+
+    :ok
+  end
+
   @impl true
   def handle_call({:start_container, config_builder}, from, state) do
+    self_pid = self()
+
     Task.async(fn ->
-      GenServer.reply(from, start_and_wait(config_builder, state))
+      result = start_and_wait(config_builder, state)
+
+      case result do
+        {:ok, container} ->
+          GenServer.cast(self_pid, {:track_container, container.container_id, container.image})
+
+        _ ->
+          # Track the image even on failure so it gets cleaned up on terminate
+          case config_builder do
+            %Container{image: image} when is_binary(image) ->
+              GenServer.cast(self_pid, {:track_image, image})
+
+            _ ->
+              :ok
+          end
+      end
+
+      GenServer.reply(from, result)
     end)
 
     {:noreply, state}
@@ -187,7 +244,7 @@ defmodule Testcontainers do
   @impl true
   def handle_call({:stop_container, container_id}, from, state) do
     Task.async(fn -> GenServer.reply(from, Api.stop_container(container_id, state.conn)) end)
-    {:noreply, state}
+    {:noreply, %{state | containers: MapSet.delete(state.containers, container_id)}}
   end
 
   @impl true
@@ -275,20 +332,43 @@ defmodule Testcontainers do
       |> Container.with_auto_remove(true)
       |> Container.with_privileged(ryuk_privileged)
 
-    with {:ok, _} <- Api.pull_image(ryuk_config.image, conn),
+    ryuk_config = resolve_pull_policy(ryuk_config, properties)
+
+    with :ok <- maybe_pull_image(ryuk_config, conn),
          {:ok, ryuk_container_id} <- Api.create_container(ryuk_config, conn),
          :ok <- Api.start_container(ryuk_container_id, conn),
          {:ok, container} <- Api.get_container(ryuk_container_id, conn),
-         {:ok, socket} <- create_ryuk_socket(container, docker_hostname),
-         :ok <- register_ryuk_filter(session_id, socket) do
+         :ok <- connect_and_register_ryuk(container, docker_hostname, session_id) do
       {:ok}
     end
   end
 
+  defp connect_and_register_ryuk(container, docker_hostname, session_id, attempt \\ 1)
+
+  defp connect_and_register_ryuk(container, docker_hostname, session_id, attempt)
+       when attempt <= 5 do
+    with {:ok, socket} <- create_ryuk_socket(container, docker_hostname),
+         :ok <- register_ryuk_filter(session_id, socket) do
+      :ok
+    else
+      error ->
+        Logger.info(
+          "Failed to connect and register ryuk filter: #{inspect(error)}. Retrying... Attempt #{attempt}/5"
+        )
+
+        :timer.sleep(1000)
+        connect_and_register_ryuk(container, docker_hostname, session_id, attempt + 1)
+    end
+  end
+
+  defp connect_and_register_ryuk(_container, _docker_hostname, _session_id, _attempt) do
+    {:error, :ryuk_connection_failed}
+  end
+
   defp create_ryuk_socket(container, docker_hostname, reattempt_count \\ 0)
 
   defp create_ryuk_socket(%Container{} = container, docker_hostname, reattempt_count)
-       when reattempt_count < 3 do
+       when reattempt_count < 5 do
     host_port = Container.mapped_port(container, 8080)
 
     case :gen_tcp.connect(~c"#{docker_hostname}", host_port, [
@@ -300,13 +380,10 @@ defmodule Testcontainers do
       {:ok, connected} ->
         {:ok, connected}
 
-      {:error, :econnrefused} ->
-        Logger.info("Connection refused. Retrying... Attempt #{reattempt_count + 1}/3")
-        :timer.sleep(5000)
+      {:error, reason} ->
+        Logger.info("Connection failed with #{inspect(reason)}. Retrying... Attempt #{reattempt_count + 1}/5")
+        :timer.sleep(1000)
         create_ryuk_socket(container, docker_hostname, reattempt_count + 1)
-
-      {:error, error} ->
-        {:error, error}
     end
   end
 
@@ -366,13 +443,38 @@ defmodule Testcontainers do
   end
 
   defp create_and_start_container(config, config_builder, state) do
+    config = resolve_pull_policy(config, state.properties)
+
     with :ok <- maybe_pull_image(config, state.conn),
-         {:ok, id} <- Api.create_container(config, state.conn),
-         :ok <- Api.start_container(id, state.conn),
+         {:ok, id} <- Api.create_container(config, state.conn) do
+      start_and_wait_container(id, config, config_builder, state)
+    end
+  end
+
+  defp resolve_pull_policy(%{pull_policy: nil} = config, properties) do
+    pull_policy =
+      case Map.get(properties, "pull.policy", "always") do
+        "missing" -> PullPolicy.never_pull()
+        "never" -> PullPolicy.never_pull()
+        _ -> PullPolicy.always_pull()
+      end
+
+    %{config | pull_policy: pull_policy}
+  end
+
+  defp resolve_pull_policy(config, _properties), do: config
+
+  defp start_and_wait_container(id, config, config_builder, state) do
+    with :ok <- Api.start_container(id, state.conn),
          {:ok, container} <- Api.get_container(id, state.conn),
          :ok <- ContainerBuilder.after_start(config_builder, container, state.conn),
          :ok <- wait_for_container(container, config.wait_strategies || [], state.conn) do
       {:ok, container}
+    else
+      error ->
+        Logger.info("Cleaning up container #{id} after failed start")
+        Api.stop_container(id, state.conn)
+        error
     end
   end
 

lib/wait_strategy/http_wait_strategy.ex

@@ -3,8 +3,8 @@ defmodule Testcontainers.HttpWaitStrategy do
   Considers the container as ready when a http request is successful.
   """
 
-  @timeout 5000
-  @max_retries 3
+  @timeout 60_000
+  @max_retries 10
 
   @typedoc """
   The HttpWaitStrategy struct
@@ -83,7 +83,7 @@ defmodule Testcontainers.HttpWaitStrategy do
           headers: wait_strategy.headers
         )
 
-      with response <- validate_response(raw_response),
+      with {:ok, response} <- validate_response(raw_response),
            :ok <- verify_status_code(wait_strategy, response),
            :ok <- verify_match(wait_strategy, response) do
         :ok
@@ -95,7 +95,7 @@ defmodule Testcontainers.HttpWaitStrategy do
 
     # Response evaluation
 
-    defp validate_response({:ok, response}), do: response
+    defp validate_response({:ok, response}), do: {:ok, response}
     defp validate_response({:error, reason}), do: {:error, reason}
 
     defp verify_status_code(wait_strategy, %{status: status_code})
@@ -130,11 +130,12 @@ defmodule Testcontainers.HttpWaitStrategy do
         {Tesla.Middleware.BaseUrl, base_url: base_url},
         {Tesla.Middleware.Timeout, timeout: request_timeout},
         {Tesla.Middleware.Retry,
-         delay: 1,
+         delay: 500,
          max_retries: wait_strategy.max_retries,
-         max_delay: 10,
+         max_delay: 5_000,
          should_retry: fn
-           _, _env, _context -> true
+           {:ok, _response}, _env, _context -> false
+           {:error, _reason}, _env, _context -> true
          end}
       ])
     end