fix: diagnostics survive recovery races end-to-end; exactly-once per recovery window (2.0.3, published)

CHANGELOG.md

@@ -6,6 +6,40 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## Unreleased
 
+## 2.0.3 — October 7, 2026
+
+### Fixed
+
+- Delivery diagnostics survive recovery races, end to end. Re-admission no longer
+  wipes `last_error`: a successful release after a durable enqueue replays the row's
+  admission-time diagnostic (the story of the failure that scheduled the retry —
+  `http_503` stays visible through the retry; only a fresh enqueue failure writes a
+  new error). Every outcome that resolves the row's immediate fate — `mark_succeeded`,
+  `mark_send_failed` (both the terminal and the retryable form), `finalize_disable`,
+  and the stored 410 obligation itself (`mark_disable_pending`) — now RELEASES the
+  enqueue claim in the same statement, so a reconciler whose enqueue raced the
+  outcome finds the token gone. A release that then matches zero rows on a
+  cleared claim classifies as BENIGN COMPLETION (`:reconciled`, never
+  `:endpoint_error`); only a still-live foreign claim is a genuine
+  `:stale_enqueue_claim`. And a successful release now PARKS the enqueue lease for
+  one horizon (30s): a concurrent recovery sweep can no longer re-admit the row in
+  the instant between the token clearing and the sweep ending — admission is
+  exactly-once per recovery window at the seam itself (the concurrent-reconciler
+  contract the suite pins, previously upheld only by SQLite's statement
+  serialization; the final coverage gate caught the true-concurrency gap). The
+  `:release_enqueue` machine primitive gains an optional `enqueue_lease_expires_at`
+  argument (default nil — direct callers see no change). Consumer-observable
+  consequence: a recovery sweep that runs inside a freshly admitted row's
+  30-second parked-lease horizon now reports that row as a contended
+  `:duplicate` where 2.0.2 re-admitted it — sub-30s recovery cadences and
+  `:duplicate`-count alerting will see the new shape. A row whose release
+  reload finds it deleted or unreadable now reports `:enqueue_reload_failed`
+  instead of a stale-claim error. First reported by a sirtify consumer review,
+  with the race family caught by the 2.0.3 release reviews reproducing it on
+  real SQLite and real PostgreSQL.
+- Test: the 5xx backoff bound is asserted exactly (4..7) — the legacy `<= 8` accepted an
+  unreachable value.
+
 ## 2.0.2 — October 6, 2026
 
 ### Fixed

documentation/livebooks/get-started.livemd

@@ -16,7 +16,7 @@ receiver. The receiver must already answer at `http://127.0.0.1:52871`;
 set `LOCAL_WEBHOOK_TESTER_PORT` to use another existing receiver. See
 the [local receiver guide](https://github.com/baselabs/ash_hooks/blob/main/ops/local/README.md).
 The notebook starts no server.
-Uses ash_hooks **2.0.2**. Its examples exercise the library locally; vendor-origin
+Uses ash_hooks **2.0.3**. Its examples exercise the library locally; vendor-origin
 delivery and a production queue require your own application integration.
 
 <!-- livebook:{"break_marks":true} -->
@@ -26,7 +26,7 @@ delivery and a production queue require your own application integration.
 ```elixir
 package =
   case System.get_env("ASH_HOOKS_NOTEBOOK_SOURCE") do
-    nil -> {:ash_hooks, "== 2.0.2"}
+    nil -> {:ash_hooks, "== 2.0.3"}
     source -> {:ash_hooks, path: Path.expand(source)}
   end
 
@@ -50,7 +50,7 @@ Mix.install(
 # and matches how SQL data layers count.
 Application.put_env(:ash, :default_string_length_count, :codepoints)
 Application.put_env(:ash_hooks, :tour_inbound_secret, :crypto.strong_rand_bytes(32))
-"2.0.2" = Application.spec(:ash_hooks, :vsn) |> to_string()
+"2.0.3" = Application.spec(:ash_hooks, :vsn) |> to_string()
 ```
 
 ```elixir

lib/ash_hooks/dispatcher.ex

@@ -599,8 +599,12 @@ defmodule AshHooks.Dispatcher do
 
   Each row is claimed with a separate enqueue token and a 30-second lease in the
   same atomic update that checks its state and due time. Successful durable
-  admission clears that token without rewriting delivery state, attempts,
-  retry time, or send lease. A crash leaves a reclaimable lease. The enqueue
+  admission releases the token and PARKS the enqueue lease for one 30-second
+  horizon (the row stays unclaimable while concurrent sweeps from the same
+  recovery window are in flight — a sweep inside that horizon sees the row as a
+  contended `:duplicate`); the row's delivery state, attempts, retry time, send
+  lease, and `last_error` diagnostic are preserved. A failed enqueue releases
+  the claim fully and records its bounded error. A crash leaves a reclaimable lease. The enqueue
   seam must remain idempotent because an external effect can occur before a
   crashed claimant clears its lease.
 
@@ -746,7 +750,11 @@ defmodule AshHooks.Dispatcher do
   defp reconcile_enqueue_result(deliv_mod, claimed, result, tenant) do
     case result do
       :ok ->
-        release_reconcile_claim(deliv_mod, claimed, nil, :reconciled, nil, tenant)
+        # A successful release PRESERVES the row's failure diagnostic by replaying
+        # its current value — re-admission is a transport event, not a delivery
+        # outcome; only the {:error, _} branch (a fresh enqueue failure) writes a
+        # new story. (The sirtify-routed last_error-wipe finding.)
+        release_reconcile_claim(deliv_mod, claimed, claimed.last_error, :reconciled, nil, tenant)
 
       {:error, reason} ->
         release_reconcile_claim(
@@ -761,7 +769,12 @@ defmodule AshHooks.Dispatcher do
   end
 
   defp release_reconcile_claim(deliv_mod, claimed, error, status, reason, tenant) do
-    case release_enqueue(deliv_mod, claimed, error, tenant) do
+    # The park decision is the explicit enqueue outcome (:reconciled admitted
+    # durably) — never the error string, which on success carries the
+    # PRESERVED diagnostic.
+    enqueue_succeeded? = match?(:reconciled, status)
+
+    case release_enqueue(deliv_mod, claimed, error, tenant, enqueue_succeeded?) do
       :ok -> reconcile_result(claimed, status, reason)
       {:error, release_reason} -> reconcile_result(claimed, :endpoint_error, release_reason)
     end
@@ -812,8 +825,7 @@ defmodule AshHooks.Dispatcher do
       |> Ash.Query.do_filter(PrimaryKey.filter(row))
       |> Ash.Query.filter(
         dispatch_source == ^row.dispatch_source and dispatch_route == ^row.dispatch_route and
-          (is_nil(enqueue_token) or is_nil(enqueue_lease_expires_at) or
-             enqueue_lease_expires_at <= ^now)
+          (is_nil(enqueue_lease_expires_at) or enqueue_lease_expires_at <= ^now)
       )
       |> recovery_state_filter(row.status, cutoff, now)
       |> Ash.bulk_update(
@@ -854,14 +866,31 @@ defmodule AshHooks.Dispatcher do
   defp recovery_state_filter(query, :disable_pending, _cutoff, _now),
     do: Ash.Query.filter(query, status == :disable_pending)
 
-  defp release_enqueue(deliv_mod, row, error, tenant) do
+  defp release_enqueue(deliv_mod, row, error, tenant, enqueue_succeeded?) do
+    # A SUCCESSFUL enqueue parks the lease for one horizon: the row stays
+    # unclaimable while CONCURRENT sweeps from the same recovery window are
+    # still in flight, so admission is exactly-once per recovery window at the
+    # seam. The park decision is the explicit enqueue outcome — never the
+    # error string, which on success carries the PRESERVED diagnostic. A
+    # failed enqueue releases fully (the next sweep must re-claim it).
+    parked =
+      if enqueue_succeeded? do
+        DateTime.add(
+          DateTime.utc_now() |> DateTime.truncate(:microsecond),
+          @enqueue_lease_seconds,
+          :second
+        )
+      else
+        nil
+      end
+
     result =
       deliv_mod
       |> Ash.Query.do_filter(PrimaryKey.filter(row))
       |> Ash.Query.filter(
         dispatch_source == ^row.dispatch_source and enqueue_token == ^row.enqueue_token
       )
-      |> Ash.bulk_update(:release_enqueue, %{error: error},
+      |> Ash.bulk_update(:release_enqueue, %{error: error, enqueue_lease_expires_at: parked},
         authorize?: false,
         return_records?: true,
         return_errors?: true,
@@ -870,8 +899,31 @@ defmodule AshHooks.Dispatcher do
       )
 
     case bulk_one(result, :stale_enqueue_claim) do
-      {:ok, _row} -> :ok
-      {:error, reason} -> {:error, reason}
+      {:ok, _row} ->
+        :ok
+
+      {:error, :stale_enqueue_claim} ->
+        # A terminal/retryable outcome between claim and release CLEARS the
+        # enqueue claim in the same statement, so the release matching zero
+        # rows here means the worker already finished the row — benign
+        # completion, never a release failure. Only a row that is STILL
+        # live-claimed is a genuine stale claim.
+        classify_stale_claim(deliv_mod, row, tenant)
+
+      {:error, reason} ->
+        {:error, reason}
+    end
+  end
+
+  defp classify_stale_claim(deliv_mod, row, tenant) do
+    case reload(deliv_mod, row, tenant) do
+      # The claim was released by an outcome (terminal, retryable, or the stored
+      # 410 obligation) — benign completion. A live token is genuine contention;
+      # a row that vanished or could not be reloaded is its own (distinguishable)
+      # failure, never silently one of the other two.
+      %{enqueue_token: nil} -> :ok
+      %{} -> {:error, :stale_enqueue_claim}
+      nil -> {:error, :enqueue_reload_failed}
     end
   end
 

lib/ash_hooks/outbound_delivery/transformers/add_delivery_actions.ex

@@ -162,10 +162,22 @@ defmodule AshHooks.OutboundDelivery.Transformers.AddDeliveryActions do
   defp build_release_enqueue do
     Builder.build_action(:update, :release_enqueue,
       accept: [],
-      arguments: [argument(:error, :string, allow_nil?: true, default: nil)],
+      arguments: [
+        argument(:error, :string, allow_nil?: true, default: nil),
+        # An optional PARKED lease: a successful release may keep the row
+        # unclaimable for one lease horizon so a CONCURRENT recovery sweep
+        # cannot re-admit it the instant the token clears (the exactly-once
+        # race the reconciler suite pins). nil releases fully.
+        argument(:enqueue_lease_expires_at, :utc_datetime_usec, allow_nil?: true, default: nil)
+      ],
       changes: [
         change(Builtins.set_attribute(:enqueue_token, nil)),
-        change(Builtins.set_attribute(:enqueue_lease_expires_at, nil)),
+        change(
+          Builtins.set_attribute(
+            :enqueue_lease_expires_at,
+            Ash.Expr.arg(:enqueue_lease_expires_at)
+          )
+        ),
         change(Builtins.set_attribute(:last_error, Ash.Expr.arg(:error)))
       ]
     )

lib/ash_hooks/outbound_delivery/transformers/add_send_actions.ex

@@ -77,13 +77,23 @@ defmodule AshHooks.OutboundDelivery.Transformers.AddSendActions do
     {:ok, clear_lease} =
       Builder.build_action_change(Builtins.set_attribute(:send_lease_expires_at, nil))
 
+    # A terminal outcome RELEASES the enqueue claim: a reconciler whose enqueue
+    # raced this worker's completion must find the token gone and no-op its
+    # release, never clobber the terminal diagnostic (the 2.0.3 review's
+    # stale-replay finding).
+    {:ok, clear_enqueue_token} =
+      Builder.build_action_change(Builtins.set_attribute(:enqueue_token, nil))
+
+    {:ok, clear_enqueue_lease} =
+      Builder.build_action_change(Builtins.set_attribute(:enqueue_lease_expires_at, nil))
+
     Builder.build_action(:update, :mark_succeeded,
       accept: [],
       arguments: [
         argument(:response_status, :integer, allow_nil?: false),
         argument(:response_snippet, :string, allow_nil?: true)
       ],
-      changes: [status, code, snippet, clear_lease]
+      changes: [status, code, snippet, clear_lease, clear_enqueue_token, clear_enqueue_lease]
     )
   end
 
@@ -123,6 +133,13 @@ defmodule AshHooks.OutboundDelivery.Transformers.AddSendActions do
     {:ok, clear_lease} =
       Builder.build_action_change(Builtins.set_attribute(:send_lease_expires_at, nil))
 
+    # A terminal/retry outcome RELEASES the enqueue claim (see mark_succeeded).
+    {:ok, clear_enqueue_token} =
+      Builder.build_action_change(Builtins.set_attribute(:enqueue_token, nil))
+
+    {:ok, clear_enqueue_lease} =
+      Builder.build_action_change(Builtins.set_attribute(:enqueue_lease_expires_at, nil))
+
     Builder.build_action(:update, :mark_send_failed,
       accept: [],
       arguments: [
@@ -132,7 +149,16 @@ defmodule AshHooks.OutboundDelivery.Transformers.AddSendActions do
         argument(:response_status, :integer, allow_nil?: true, default: nil),
         argument(:response_snippet, :string, allow_nil?: true, default: nil)
       ],
-      changes: [status, error, next, code, snippet, clear_lease]
+      changes: [
+        status,
+        error,
+        next,
+        code,
+        snippet,
+        clear_lease,
+        clear_enqueue_token,
+        clear_enqueue_lease
+      ]
     )
   end
 
@@ -151,7 +177,12 @@ defmodule AshHooks.OutboundDelivery.Transformers.AddSendActions do
         change(Builtins.set_attribute(:endpoint_snapshot, arg(:endpoint_snapshot))),
         change(Builtins.set_attribute(:last_error, "gone_410")),
         change(Builtins.set_attribute(:next_attempt_at, nil)),
-        change(Builtins.set_attribute(:send_lease_expires_at, nil))
+        change(Builtins.set_attribute(:send_lease_expires_at, nil)),
+        # The durable obligation releases the enqueue claim with it: a crash
+        # between storage and finalization must leave gone_410 as the standing
+        # diagnostic, never restorable-over by a stale reconciler release.
+        change(Builtins.set_attribute(:enqueue_token, nil)),
+        change(Builtins.set_attribute(:enqueue_lease_expires_at, nil))
       ]
     )
   end
@@ -162,7 +193,9 @@ defmodule AshHooks.OutboundDelivery.Transformers.AddSendActions do
       arguments: [argument(:error, :string, allow_nil?: false)],
       changes: [
         change(Builtins.set_attribute(:status, :dead_letter)),
-        change(Builtins.set_attribute(:last_error, arg(:error)))
+        change(Builtins.set_attribute(:last_error, arg(:error))),
+        change(Builtins.set_attribute(:enqueue_token, nil)),
+        change(Builtins.set_attribute(:enqueue_lease_expires_at, nil))
       ]
     )
   end

mix.exs

@@ -1,7 +1,7 @@
 defmodule AshHooks.MixProject do
   use Mix.Project
 
-  @version "2.0.2"
+  @version "2.0.3"
   @source_url "https://github.com/baselabs/ash_hooks"
 
   def project do

test/ash_hooks/delivery_test.exs

@@ -654,8 +654,9 @@ defmodule AshHooks.DeliveryTest do
 
       assert {:snooze, delay} = DeliveryRuntime.run(args(row), config())
 
-      # attempts == 1 after the mark: base 2 * 2^1 = 4; jitter adds [0, delay)
-      assert delay >= 4 and delay <= 8
+      # attempts == 1 after the mark: base 2 * 2^1 = 4; jitter adds [0, 4) — the
+      # bound is exactly 4..7 (8 was unreachable; tightened at the 2.0.3 release).
+      assert delay in 4..7
       assert row!(row.id).status == :failed_retryable
     end