diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b25d52d..552bf40 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/lib/durable_server/object_store.ex b/lib/durable_server/object_store.ex index f53a05f..7309c1c 100644 --- a/lib/durable_server/object_store.ex +++ b/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. diff --git a/lib/durable_server/supervisor.ex b/lib/durable_server/supervisor.ex index 105a4ef..2553a92 100644 --- a/lib/durable_server/supervisor.ex +++ b/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} diff --git a/test/durable_server/object_store_readiness_test.exs b/test/durable_server/object_store_readiness_test.exs new file mode 100644 index 0000000..f00d574 --- /dev/null +++ b/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 diff --git a/test/durable_server/supervisor_backend_spec_test.exs b/test/durable_server/supervisor_backend_spec_test.exs index 52d17fa..435406a 100644 --- a/test/durable_server/supervisor_backend_spec_test.exs +++ b/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")