diff --git a/lib/durable_server.ex b/lib/durable_server.ex
index 8d62431..5154e38 100644
--- a/lib/durable_server.ex
+++ b/lib/durable_server.ex
@@ -896,6 +896,8 @@ defmodule DurableServer do
@durable :durable
@max_crash_reason_length 500
@max_sync_retries 5
+ # Matches ObjectStore.try_claim/3, the first-ever claim.
+ @lock_claim_retries 2
@bootstrap_continue {@durable, :bootstrap}
defguardp is_callback_action(action)
@@ -3991,7 +3993,14 @@ defmodule DurableServer do
%DurableServer{key: key, object_store: store} = state,
data
) do
- case StorageBackend.put_object(store, storage_key(state), data, etag: state.etag) do
+ # Retries transient failures like the first-ever claim does: object storage
+ # throttles rapid writes to one key (Tigris 503 SlowDown), and a restart
+ # claims the key right after the previous owner's final writes. A retry
+ # whose earlier attempt landed resolves as this boot's own claim.
+ case StorageBackend.put_object(store, storage_key(state), data,
+ etag: state.etag,
+ max_retries: @lock_claim_retries
+ ) do
# obj still matched our etag, we got the claim
{:ok, %{etag: new_etag}} ->
Logger.info(
diff --git a/test/durable_server/object_store_claim_test.exs b/test/durable_server/object_store_claim_test.exs
index c0d9929..97dce48 100644
--- a/test/durable_server/object_store_claim_test.exs
+++ b/test/durable_server/object_store_claim_test.exs
@@ -51,6 +51,44 @@ defmodule DurableServer.ObjectStoreClaimTest do
end
end
+ describe "conditional put retries" do
+ test "retries a throttled put" do
+ store =
+ object_store([
+ %Req.Response{status: 503, body: "SlowDown"},
+ %Req.Response{status: 200, headers: %{"etag" => ["next-etag"]}}
+ ])
+
+ assert {:ok, %{etag: "next-etag"}} =
+ ObjectStore.put_object(store, "server/one", "state",
+ etag: "current-etag",
+ max_retries: 2
+ )
+
+ for _attempt <- 1..2 do
+ assert_receive {:request, %Req.Request{method: :put} = request}
+ assert Req.Request.get_header(request, "if-match") == ["current-etag"]
+ end
+ end
+
+ test "does not retry a lost etag race" do
+ store =
+ object_store([
+ %Req.Response{status: 412},
+ %Req.Response{status: 200, headers: %{"etag" => ["never-sent"]}}
+ ])
+
+ assert {:error, :conflict} =
+ ObjectStore.put_object(store, "server/one", "state",
+ etag: "current-etag",
+ max_retries: 2
+ )
+
+ assert_receive {:request, %Req.Request{method: :put}}
+ refute_receive {:request, _request}
+ end
+ end
+
describe "initial claim recovery" do
test "adopts the committed claim when a lost response is followed by a conflict" do
attempted = stored_state(%{value: 1})
diff --git a/test/durable_server_test.exs b/test/durable_server_test.exs
index 8c709c3..ad0d861 100644
--- a/test/durable_server_test.exs
+++ b/test/durable_server_test.exs
@@ -1370,6 +1370,91 @@ defmodule DurableServerTest do
assert message =~ "allowed keys are: [:auto_sync, :sync_every_ms, :meta, :permanent]"
end
+ test "a restart retries a throttled lock claim" do
+ throttles = start_supervised!({Agent, fn -> 0 end}, id: :lock_claim_throttles)
+
+ # Object storage answers the next `throttles` conditional PUTs with 503
+ # SlowDown, as Tigris does to rapid writes to one key.
+ adapter = fn request ->
+ throttle? =
+ request.method == :put and Req.Request.get_header(request, "if-match") != [] and
+ Agent.get_and_update(throttles, fn left -> {left > 0, max(left - 1, 0)} end)
+
+ if throttle? do
+ {request, %Req.Response{status: 503, body: "SlowDown"}}
+ else
+ Req.Steps.run_finch(request)
+ end
+ end
+
+ {supervisor_name, _supervisor_pid, _prefix} =
+ start_test_supervisor(
+ object_store:
+ test_object_store_opts(
+ req_opts: [adapter: adapter, retry_delay: 0, retry_log_level: false]
+ )
+ )
+
+ key = "throttled-claim-#{DurableServer.UUID.uuid4()}"
+ child = {TestServer, key: key, initial_state: %{}}
+
+ assert {:ok, {pid, _meta}} = DurableServer.Supervisor.start_child(supervisor_name, child)
+ assert :ok = GenServer.stop(pid, :shutdown)
+ Agent.update(throttles, fn _left -> 1 end)
+
+ log =
+ ExUnit.CaptureLog.capture_log(fn ->
+ assert {:ok, {restarted, _meta}} =
+ DurableServer.Supervisor.start_child(supervisor_name, child)
+
+ assert restarted != pid
+ end)
+
+ assert log =~ "won the lock for #{inspect(key)}"
+ assert Agent.get(throttles, & &1) == 0
+ end
+
+ test "a retried lock claim whose first attempt landed is this boot's claim" do
+ lost = start_supervised!({Agent, fn -> 0 end}, id: :lost_lock_claim_responses)
+
+ # The next `lost` conditional PUTs reach storage, but their success
+ # responses are lost and the client sees 503.
+ adapter = fn request ->
+ lose? =
+ request.method == :put and Req.Request.get_header(request, "if-match") != [] and
+ Agent.get_and_update(lost, fn left -> {left > 0, max(left - 1, 0)} end)
+
+ {request, response} = Req.Steps.run_finch(request)
+
+ if lose? and response.status in 200..299 do
+ {request, %Req.Response{status: 503, body: "SlowDown"}}
+ else
+ {request, response}
+ end
+ end
+
+ {supervisor_name, _supervisor_pid, _prefix} =
+ start_test_supervisor(
+ object_store:
+ test_object_store_opts(
+ req_opts: [adapter: adapter, retry_delay: 0, retry_log_level: false]
+ )
+ )
+
+ key = "lost-claim-#{DurableServer.UUID.uuid4()}"
+ child = {TestServer, key: key, initial_state: %{}}
+
+ assert {:ok, {pid, _meta}} = DurableServer.Supervisor.start_child(supervisor_name, child)
+ assert :ok = GenServer.stop(pid, :shutdown)
+ Agent.update(lost, fn _left -> 1 end)
+
+ assert {:ok, {restarted, _meta}} =
+ DurableServer.Supervisor.start_child(supervisor_name, child)
+
+ assert restarted != pid
+ assert Agent.get(lost, & &1) == 0
+ end
+
test "returns startup sync failure to caller and logs the error" do
{supervisor_name, _supervisor_pid, _prefix} =
start_test_supervisor(