Recover object-store readiness failures

.github/workflows/ci.yml

@@ -73,5 +73,8 @@ jobs:
       - name: Compile with warnings as errors
         run: mix compile --warnings-as-errors
 
+      - name: Start EPMD for distributed tests
+        run: epmd -daemon
+
       - name: Test with warnings as errors
         run: mix test --warnings-as-errors

lib/durable_server/object_store.ex

@@ -116,6 +116,7 @@ defmodule DurableServer.ObjectStore do
     case create_bucket(client, client.bucket) do
       {:error, %{status: 409}} -> :ok
       {:ok, %__MODULE__{}} -> :ok
+      {:error, reason} -> {:error, reason}
     end
   end
 
@@ -137,7 +138,8 @@ defmodule DurableServer.ObjectStore do
            params: %{
              "location-constraint" => "us-east-1"
            },
-           retry: :transient
+           retry: &retry_bucket_create?/2,
+           max_retries: 2
          ) do
       {:ok, %{status: status}} when status >= 200 and status < 300 ->
         {:ok, new(client, bucket: bucket_name)}
@@ -150,6 +152,22 @@ defmodule DurableServer.ObjectStore do
     end
   end
 
+  # A transport failure can hide a successful bucket create. Retrying is safe:
+  # the create is idempotent and an already-created bucket returns 409.
+  defp retry_bucket_create?(_request, %Req.TransportError{}), do: true
+
+  defp retry_bucket_create?(_request, %Req.Response{status: status})
+       when status in [408, 429, 500, 502, 503, 504],
+       do: true
+
+  defp retry_bucket_create?(
+         _request,
+         %Req.HTTPError{protocol: :http2, reason: :unprocessed}
+       ),
+       do: true
+
+  defp retry_bucket_create?(_request, _response_or_exception), do: false
+
   @doc """
   Lists all buckets in the account.
   Returns {:ok, buckets} on success, or {:error, reason} if listing fails.

lib/durable_server/supervisor.ex

@@ -3431,10 +3431,7 @@ defmodule DurableServer.Supervisor do
     task_sup = Keyword.get(opts, :task_supervisor, DurableServer.TaskSupervisor)
 
     backend_resources = build_backend_resources(opts, finch, task_sup)
-
-    Enum.each(backend_resources.ensure_ready_backends, fn backend ->
-      :ok = StorageBackend.ensure_ready(backend)
-    end)
+    ensure_backends_ready!(backend_resources.ensure_ready_backends)
 
     storage_backend = backend_resources.storage_backend
     heartbeat_backend = backend_resources.heartbeat_backend
@@ -3757,6 +3754,22 @@ defmodule DurableServer.Supervisor do
     Supervisor.init(children, strategy: :one_for_all)
   end
 
+  defp ensure_backends_ready!(backends) do
+    Enum.each(backends, fn %StorageBackend{adapter: adapter} = backend ->
+      case StorageBackend.ensure_ready(backend) do
+        :ok ->
+          :ok
+
+        {:error, reason} ->
+          exit({:backend_unavailable, adapter, reason})
+
+        other ->
+          raise ArgumentError,
+                "#{inspect(adapter)}.ensure_ready/1 returned invalid result: #{inspect(other)}"
+      end
+    end)
+  end
+
   @doc false
   def __supervisor_presence_group__(sup_name) when is_atom(sup_name) do
     {__MODULE__, :presence, sup_name}

test/durable_server/object_store_readiness_test.exs

@@ -0,0 +1,56 @@
+defmodule DurableServer.ObjectStoreReadinessTest do
+  use ExUnit.Case, async: true
+
+  alias DurableServer.ObjectStore
+
+  test "returns a transport error with a binary reason from a bucket readiness check" do
+    reason = "request process exited before returning a response"
+    store = object_store(%Req.TransportError{reason: reason})
+
+    assert {:error, %Req.TransportError{reason: ^reason}} =
+             ObjectStore.ensure_bucket_exists(store)
+  end
+
+  test "retries an ambiguous create and accepts an existing bucket" do
+    reason = "request process exited before returning a response"
+
+    store =
+      object_store([
+        %Req.TransportError{reason: reason},
+        %Req.Response{status: 409}
+      ])
+
+    assert :ok = ObjectStore.ensure_bucket_exists(store)
+    assert_receive {:request, %Req.Request{method: :put}}
+    assert_receive {:request, %Req.Request{method: :put}}
+  end
+
+  defp object_store(responses) when is_list(responses) do
+    parent = self()
+    responses = start_supervised!({Agent, fn -> responses end})
+
+    adapter = fn request ->
+      send(parent, {:request, request})
+      response = Agent.get_and_update(responses, fn [response | rest] -> {response, rest} end)
+      {request, response}
+    end
+
+    new_object_store(adapter)
+  end
+
+  defp object_store(response) do
+    adapter = fn request -> {request, response} end
+    new_object_store(adapter)
+  end
+
+  defp new_object_store(adapter) do
+    ObjectStore.new(
+      bucket: "test-bucket",
+      access_key_id: "test-access-key",
+      secret_access_key: "test-secret-key",
+      s3_endpoint: "http://s3.test",
+      default_region: "us-east-1",
+      req_opts: [adapter: adapter, retry_delay: 0, retry_log_level: false]
+    )
+  end
+end

test/durable_server/supervisor_backend_spec_test.exs

@@ -28,7 +28,8 @@ defmodule DurableServer.SupervisorBackendSpecTest do
        %{
          state: %{
            table: :ets.new(__MODULE__, [:set, :public]),
-           name: Map.get(opts, :name)
+           name: Map.get(opts, :name),
+           ensure_ready: Map.get(opts, :ensure_ready, :ok)
          },
          defaults: %{
            heartbeat_tracking_mode: :poll,
@@ -40,7 +41,7 @@ defmodule DurableServer.SupervisorBackendSpecTest do
     end
 
     @impl true
-    def ensure_ready(_state), do: :ok
+    def ensure_ready(%{ensure_ready: result}), do: result
 
     @impl true
     def get_object(%{table: table}, key, _opts) do
@@ -110,6 +111,41 @@ defmodule DurableServer.SupervisorBackendSpecTest do
     end
   end
 
+  test "exits with an actionable startup error when a backend is unavailable" do
+    supervisor_name = unique_supervisor_name("backend_unavailable")
+    prefix = unique_prefix("backend_unavailable")
+    reason = %Req.TransportError{reason: "request process exited before returning a response"}
+
+    {pid, ref} =
+      spawn_monitor(fn ->
+        DurableServer.Supervisor.start_link(
+          name: supervisor_name,
+          prefix: prefix,
+          backend: {InMemoryBackend, name: :unavailable, ensure_ready: {:error, reason}}
+        )
+      end)
+
+    assert_receive {:DOWN, ^ref, :process, ^pid, {:backend_unavailable, InMemoryBackend, ^reason}}
+  end
+
+  test "raises when a backend violates the readiness result contract" do
+    supervisor_name = unique_supervisor_name("invalid_readiness")
+    prefix = unique_prefix("invalid_readiness")
+
+    {pid, ref} =
+      spawn_monitor(fn ->
+        DurableServer.Supervisor.start_link(
+          name: supervisor_name,
+          prefix: prefix,
+          backend: {InMemoryBackend, name: :invalid, ensure_ready: :unexpected}
+        )
+      end)
+
+    assert_receive {:DOWN, ^ref, :process, ^pid, {%ArgumentError{message: message}, _stacktrace}}
+
+    assert message =~ "ensure_ready/1 returned invalid result: :unexpected"
+  end
+
   test "rejects heartbeats beyond the configured future clock-skew tolerance" do
     supervisor_name = unique_supervisor_name("heartbeat_skew")
     prefix = unique_prefix("heartbeat_skew")