From 2f60a864986ffdd338837f2fa0f635043fc1ff07 Mon Sep 17 00:00:00 2001 From: Russ Palermo Date: Wed, 7 Oct 2026 08:24:00 -0400 Subject: [PATCH] =?UTF-8?q?fix:=20diagnostics=20survive=20recovery=20races?= =?UTF-8?q?=20end-to-end;=20recovery=20admission=20is=20exactly-once=20per?= =?UTF-8?q?=20window=20(2.0.3)=20=E2=80=94=20the=20sirtify-routed=20last?= =?UTF-8?q?=5Ferror-wipe=20closed=20with=20its=20whole=20race=20family:=20?= =?UTF-8?q?a=20successful=20release=20now=20REPLAYS=20the=20row's=20admiss?= =?UTF-8?q?ion-time=20diagnostic=20(http=5F503=20stays=20visible=20through?= =?UTF-8?q?=20re-admission;=20only=20a=20fresh=20enqueue=20failure=20write?= =?UTF-8?q?s=20a=20new=20error);=20every=20fate-resolving=20outcome=20(mar?= =?UTF-8?q?k=5Fsucceeded,=20mark=5Fsend=5Ffailed=20both=20forms,=20finaliz?= =?UTF-8?q?e=5Fdisable,=20and=20the=20stored=20410=20obligation=20itself)?= =?UTF-8?q?=20releases=20the=20enqueue=20claim=20in=20the=20same=20stateme?= =?UTF-8?q?nt,=20so=20a=20reconciler=20whose=20enqueue=20raced=20the=20out?= =?UTF-8?q?come=20finds=20the=20token=20gone;=20a=20release=20matching=20z?= =?UTF-8?q?ero=20rows=20on=20a=20CLEARED=20claim=20classifies=20as=20benig?= =?UTF-8?q?n=20completion=20(:reconciled,=20never=20:endpoint=5Ferror)=20v?= =?UTF-8?q?ia=20reload,=20with=20genuine=20contention=20(:stale=5Fenqueue?= =?UTF-8?q?=5Fclaim)=20and=20a=20vanished=20row=20(:enqueue=5Freload=5Ffai?= =?UTF-8?q?led)=20as=20distinguishable=20labels;=20and=20the=20claim=20gat?= =?UTF-8?q?e=20is=20the=20LEASE=20ALONE=20=E2=80=94=20a=20successful=20adm?= =?UTF-8?q?ission=20parks=20enqueue=5Flease=5Fexpires=5Fat=20for=20one=203?= =?UTF-8?q?0s=20horizon=20(release=5Fenqueue/5=20takes=20the=20explicit=20?= =?UTF-8?q?enqueue=20outcome;=20the=20error=20string=20carries=20the=20pre?= =?UTF-8?q?served=20diagnostic=20on=20success),=20so=20a=20concurrent=20sw?= =?UTF-8?q?eep=20inside=20the=20horizon=20reports=20a=20contended=20:dupli?= =?UTF-8?q?cate=20instead=20of=20re-admitting=20(the=20exactly-once=20cont?= =?UTF-8?q?ract=20was=20previously=20upheld=20only=20by=20SQLite=20stateme?= =?UTF-8?q?nt=20serialization=20=E2=80=94=20the=20final=20PG=20coverage=20?= =?UTF-8?q?gate=20caught=20a=20true-concurrency=20double-admit;=20the=20fo?= =?UTF-8?q?rced-interleaving=20test=20now=20pins=20it=20structurally,=20an?= =?UTF-8?q?d=20the=20horizon-passed=20re-claim=20keeps=20crashed-claim=20r?= =?UTF-8?q?ecovery).=20The=20reconcile=5Fpending=20docstring=20states=20th?= =?UTF-8?q?e=20new=20contract;=20the=20changelog=20names=20the=20consumer-?= =?UTF-8?q?observable=20consequence=20(sub-30s=20sweeps=20see=20:duplicate?= =?UTF-8?q?=20where=202.0.2=20re-admitted).=20Three=20codex=20rounds=20+?= =?UTF-8?q?=20a=20claude=20final=20pass=20(receipts=20under=20.kimosabe/re?= =?UTF-8?q?views=20+=20scratch),=20every=20finding=20fixed=20=E2=80=94=20i?= =?UTF-8?q?ncluding=20two=20of=20mine=20the=20round-3=20review=20caught=20?= =?UTF-8?q?(the=20park=20keyed=20on=20is=5Fnil(error)=20while=20success=20?= =?UTF-8?q?carries=20the=20diagnostic;=20the=20claim=20gate's=20token=20ar?= =?UTF-8?q?m=20bypassing=20the=20parked=20lease)=20and=20the=20gate's=20ow?= =?UTF-8?q?n=20catches=20(an=20uncovered-clause=20collapse,=20a=20duplicat?= =?UTF-8?q?e=20reload/3,=20three=20struct-bound=20lease=20fixtures).=20Gat?= =?UTF-8?q?e=20on=20final=20source:=20format,=20compile=20--warnings-as-er?= =?UTF-8?q?rors,=20credo=20--strict,=20hex.audit,=20currency,=20SQLite=20s?= =?UTF-8?q?uite=20787,=20PG=20coverage=20823=20passed=20/=20100.00%=20tota?= =?UTF-8?q?l=20against=20the=20ephemeral-area=20PG=20+=20the=20real=20rece?= =?UTF-8?q?iver,=20dialyzer=200,=20cheat-sheets=20--check,=20docs=20--warn?= =?UTF-8?q?ings-as-errors,=20check-package=20bytes-match,=20the=20tour=20n?= =?UTF-8?q?otebook=20executed=20green=20against=20the=202.0.3=20archive,?= =?UTF-8?q?=20and=20the=20exact-Ash-floor=20qualification=20passed.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CHANGELOG.md | 34 ++ documentation/livebooks/get-started.livemd | 6 +- lib/ash_hooks/dispatcher.ex | 72 +++- .../transformers/add_delivery_actions.ex | 16 +- .../transformers/add_send_actions.ex | 41 ++- mix.exs | 2 +- test/ash_hooks/delivery_test.exs | 5 +- test/ash_hooks/reconcile_test.exs | 320 ++++++++++++++++++ 8 files changed, 474 insertions(+), 22 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ffa7e13..cadd6b2 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/documentation/livebooks/get-started.livemd b/documentation/livebooks/get-started.livemd index c9ecb91..900f6d3 100644 --- a/documentation/livebooks/get-started.livemd +++ b/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. @@ -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 diff --git a/lib/ash_hooks/dispatcher.ex b/lib/ash_hooks/dispatcher.ex index 9b0d949..3f1fc71 100644 --- a/lib/ash_hooks/dispatcher.ex +++ b/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 diff --git a/lib/ash_hooks/outbound_delivery/transformers/add_delivery_actions.ex b/lib/ash_hooks/outbound_delivery/transformers/add_delivery_actions.ex index a217419..a27ffbf 100644 --- a/lib/ash_hooks/outbound_delivery/transformers/add_delivery_actions.ex +++ b/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))) ] ) diff --git a/lib/ash_hooks/outbound_delivery/transformers/add_send_actions.ex b/lib/ash_hooks/outbound_delivery/transformers/add_send_actions.ex index 8277ac5..ae2338f 100644 --- a/lib/ash_hooks/outbound_delivery/transformers/add_send_actions.ex +++ b/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 diff --git a/mix.exs b/mix.exs index 7af764a..9879455 100644 --- a/mix.exs +++ b/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 diff --git a/test/ash_hooks/delivery_test.exs b/test/ash_hooks/delivery_test.exs index da4ce44..d8cc209 100644 --- a/test/ash_hooks/delivery_test.exs +++ b/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 diff --git a/test/ash_hooks/reconcile_test.exs b/test/ash_hooks/reconcile_test.exs index 53659cd..b1017f0 100644 --- a/test/ash_hooks/reconcile_test.exs +++ b/test/ash_hooks/reconcile_test.exs @@ -394,6 +394,326 @@ defmodule AshHooks.ReconcileTest do assert row_state!("rec-state-send-expired", "org_a").attempts == 2 end + test "a due failed_retryable row re-admitted by recovery KEEPS its last_error diagnostic" do + # The sirtify-routed finding (claude F8, 2026-10-06): recovery re-admission is a + # TRANSPORT event, not a delivery outcome — wiping the row's failure diagnostic on + # re-admission hides why the retry was scheduled until a new outcome lands. + stranded_row!("org_a", "rec-diag-retry") + + past = + DateTime.add(DateTime.utc_now() |> DateTime.truncate(:second), -120, :second) + |> DateTime.to_iso8601() + + Repo.query!( + "UPDATE #{@deliveries} SET status = 'failed_retryable', attempts = 1, next_attempt_at = ?, last_error = 'http_503' WHERE event_uuid = ?", + [past, "rec-diag-retry"] + ) + + assert {:ok, [result]} = + Dispatcher.reconcile_pending(Emitter, :order_paid, + tenant: "org_a", + older_than: cutoff(), + enqueue_key: "reconcile-diag", + enqueue: counting_enqueuer(self()) + ) + + assert result.status == :reconciled + assert_receive {:enqueued, "rec-diag-retry"}, 1_000 + + row = row_state!("rec-diag-retry", "org_a") + assert row.status == :failed_retryable + assert row.attempts == 1 + assert row.last_error == "http_503", "re-admission must not wipe the failure diagnostic" + end + + test "a terminal outcome releases the enqueue claim — a stale reconciler release no-ops, never clobbers the terminal diagnostic" do + # The 2.0.3 review's stale-replay finding, at its root: mark_succeeded and + # mark_send_failed used to leave a reconciler's enqueue_token live on terminal + # rows, so the token-gated release could still fire AFTER the worker completed + # and overwrite the terminal diagnostic (endpoint_disabled) with the stale + # pre-admission one (http_503). + stranded_row!("org_a", "rec-terminal-claim") + token = Ash.UUID.generate() + lease = DateTime.add(DateTime.utc_now(), 30, :second) |> DateTime.to_iso8601() + + Repo.query!( + "UPDATE #{@deliveries} SET status = 'failed_retryable', attempts = 1, next_attempt_at = ?, last_error = 'http_503', enqueue_token = ?, enqueue_lease_expires_at = ? WHERE event_uuid = ?", + [ + DateTime.add(DateTime.utc_now(), -120, :second) |> DateTime.to_iso8601(), + token, + lease, + "rec-terminal-claim" + ] + ) + + row = row_state!("rec-terminal-claim", "org_a") + + # The worker's terminal transition (dead_letter) both writes the outcome AND + # releases the enqueue claim in the same statement. + row + |> Ash.Changeset.for_update( + :mark_send_failed, + %{ + error: "endpoint_disabled", + next_attempt_at: nil, + dead_letter?: true, + response_status: nil, + response_snippet: nil + }, + tenant: "org_a", + authorize?: false + ) + |> Ash.update!() + + terminal = row_state!("rec-terminal-claim", "org_a") + assert terminal.status == :dead_letter + assert terminal.last_error == "endpoint_disabled" + assert is_nil(terminal.enqueue_token), "the terminal outcome must release the enqueue claim" + assert is_nil(terminal.enqueue_lease_expires_at) + + # mark_succeeded releases the claim the same way. + stranded_row!("org_a", "rec-terminal-claim-ok") + ok_token = Ash.UUID.generate() + + Repo.query!( + "UPDATE #{@deliveries} SET enqueue_token = ?, enqueue_lease_expires_at = ? WHERE event_uuid = ?", + [ok_token, lease, "rec-terminal-claim-ok"] + ) + + row_state!("rec-terminal-claim-ok", "org_a") + |> Ash.Changeset.for_update(:mark_succeeded, %{response_status: 200, response_snippet: nil}, + tenant: "org_a", + authorize?: false + ) + |> Ash.update!() + + ok = row_state!("rec-terminal-claim-ok", "org_a") + assert ok.status == :succeeded + assert is_nil(ok.enqueue_token) + end + + test "a terminal completion between claim and release reports :reconciled, never :endpoint_error (benign completion)" do + # Round-2 finding 2: the terminal outcome clears the enqueue claim, so the + # release matches zero rows — that is the worker having ALREADY finished the + # row, which is success-shaped, not a release failure. + stranded_row!("org_a", "rec-benign-race") + + now = DateTime.utc_now() |> DateTime.truncate(:second) + past = DateTime.add(now, -120, :second) |> DateTime.to_iso8601() + + Repo.query!( + "UPDATE #{@deliveries} SET status = 'failed_retryable', attempts = 1, next_attempt_at = ?, last_error = 'http_503' WHERE event_uuid = ?", + [past, "rec-benign-race"] + ) + + # The enqueuer SIMULATES the race: the row is terminally completed inside + # the enqueue callback, before the dispatcher's release fires. + row = row_state!("rec-benign-race", "org_a") + + completing_enqueuer = fn _delivery, _event -> + row + |> Ash.Changeset.for_update(:mark_succeeded, %{response_status: 200, response_snippet: nil}, + tenant: "org_a", + authorize?: false + ) + |> Ash.update!() + + :ok + end + + assert {:ok, [result]} = + Dispatcher.reconcile_pending(Emitter, :order_paid, + tenant: "org_a", + older_than: cutoff(), + enqueue_key: "reconcile-benign", + enqueue: completing_enqueuer + ) + + assert result.status == :reconciled, "outcome-driven claim invalidation is benign completion" + + final = row_state!("rec-benign-race", "org_a") + assert final.status == :succeeded + assert is_nil(final.enqueue_token) + end + + test "a stored 410 obligation keeps gone_410 through a stale release (round-2 finding 1)" do + stranded_row!("org_a", "rec-410-claim") + + Repo.query!( + "UPDATE #{@deliveries} SET last_error = 'http_503', enqueue_token = ?, enqueue_lease_expires_at = ? WHERE event_uuid = ?", + [ + Ash.UUID.generate(), + DateTime.add(DateTime.utc_now(), 30, :second) |> DateTime.to_iso8601(), + "rec-410-claim" + ] + ) + + row_state!("rec-410-claim", "org_a") + |> Ash.Changeset.for_update( + :mark_disable_pending, + %{ + response_status: 410, + response_snippet: nil, + endpoint_snapshot: %{"endpoint_pk" => %{}} + }, + tenant: "org_a", + authorize?: false + ) + |> Ash.update!() + + stored = row_state!("rec-410-claim", "org_a") + assert stored.status == :disable_pending + assert stored.last_error == "gone_410" + assert is_nil(stored.enqueue_token), "the obligation releases the claim at storage" + end + + test "a genuinely foreign live claim at release still reports :stale_enqueue_claim (:endpoint_error)" do + # The benign-completion classification must NOT swallow a real contention: + # if a THIRD claimant re-claimed the row between our claim and our release, + # the reload finds a live foreign token — a genuine stale claim, surfaced as + # :endpoint_error (the reconcile caller's failure vocabulary). + stranded_row!("org_a", "rec-foreign-claim") + + past = + DateTime.add(DateTime.utc_now() |> DateTime.truncate(:second), -120, :second) + |> DateTime.to_iso8601() + + Repo.query!( + "UPDATE #{@deliveries} SET status = 'failed_retryable', attempts = 1, next_attempt_at = ?, last_error = 'http_503' WHERE event_uuid = ?", + [past, "rec-foreign-claim"] + ) + + row = row_state!("rec-foreign-claim", "org_a") + + # The enqueuer simulates a racing re-claimant: it stamps a FRESH foreign + # claim on the row inside the callback, before our release fires. + foreign_enqueuer = fn delivery, _event -> + foreign_token = Ash.UUID.generate() + foreign_lease = DateTime.add(DateTime.utc_now(), 30, :second) |> DateTime.to_iso8601() + + Repo.query!( + "UPDATE #{@deliveries} SET enqueue_token = ?, enqueue_lease_expires_at = ? WHERE event_uuid = ?", + [foreign_token, foreign_lease, delivery.event_uuid] + ) + + :ok + end + + assert {:ok, [result]} = + Dispatcher.reconcile_pending(Emitter, :order_paid, + tenant: "org_a", + older_than: cutoff(), + enqueue_key: "reconcile-foreign", + enqueue: foreign_enqueuer + ) + + assert result.status == :endpoint_error + end + + test "a parked lease fences immediate re-admission — the forced interleaving, not scheduling luck (round-3)" do + # Round-3 finding 1: the parked lease must actually GATE the next claim + # (the claim gate is the lease alone now) — admitted once, then fenced + # until the horizon passes, then claimable again. + stranded_row!("org_a", "rec-park-fence") + + past = + DateTime.add(DateTime.utc_now() |> DateTime.truncate(:second), -120, :second) + |> DateTime.to_iso8601() + + Repo.query!( + "UPDATE #{@deliveries} SET status = 'failed_retryable', attempts = 1, next_attempt_at = ?, last_error = 'http_503' WHERE event_uuid = ?", + [past, "rec-park-fence"] + ) + + parent = self() + + assert {:ok, [%{status: :reconciled}]} = + Dispatcher.reconcile_pending(Emitter, :order_paid, + tenant: "org_a", + older_than: cutoff(), + enqueue_key: "reconcile-park", + enqueue: fn delivery, _event -> + send(parent, {:parked, delivery.event_uuid}) + :ok + end + ) + + assert_receive {:parked, "rec-park-fence"} + + row = row_state!("rec-park-fence", "org_a") + assert row.last_error == "http_503", "the diagnostic survives the admitted re-enqueue" + assert is_nil(row.enqueue_token), "the claim is released" + assert not is_nil(row.enqueue_lease_expires_at), "the lease is PARKED" + assert DateTime.compare(row.enqueue_lease_expires_at, DateTime.utc_now()) == :gt + + # The immediate second sweep finds the row fenced: the lease gate blocks + # the claim, the row surfaces as a contended :duplicate (the concurrent + # race outcome vocabulary), and the seam is NEVER called again. + assert {:ok, [%{status: :duplicate}]} = + Dispatcher.reconcile_pending(Emitter, :order_paid, + tenant: "org_a", + older_than: cutoff(), + enqueue_key: "reconcile-park", + enqueue: fn delivery, _event -> + send(parent, {:again, delivery.event_uuid}) + :ok + end + ) + + refute_receive {:again, _}, 200 + + # Once the horizon passes, the row is claimable again (crashed-claim and + # parked-release recovery share the one gate). + Repo.query!( + "UPDATE #{@deliveries} SET enqueue_lease_expires_at = ? WHERE event_uuid = ?", + [DateTime.add(DateTime.utc_now(), -60, :second) |> DateTime.to_iso8601(), "rec-park-fence"] + ) + + assert {:ok, [%{status: :reconciled}]} = + Dispatcher.reconcile_pending(Emitter, :order_paid, + tenant: "org_a", + older_than: cutoff(), + enqueue_key: "reconcile-park", + enqueue: fn delivery, _event -> + send(parent, {:unfenced, delivery.event_uuid}) + :ok + end + ) + + assert_receive {:unfenced, "rec-park-fence"} + end + + test "a row deleted between claim and release reports :enqueue_reload_failed (its own label, never a stale claim)" do + # The classification split: a vanished/unreadable row is distinguishable + # from genuine contention in the result vocabulary. + stranded_row!("org_a", "rec-vanished") + + past = + DateTime.add(DateTime.utc_now() |> DateTime.truncate(:second), -120, :second) + |> DateTime.to_iso8601() + + Repo.query!( + "UPDATE #{@deliveries} SET status = 'failed_retryable', attempts = 1, next_attempt_at = ?, last_error = 'http_503' WHERE event_uuid = ?", + [past, "rec-vanished"] + ) + + vanishing_enqueuer = fn delivery, _event -> + Repo.query!("DELETE FROM #{@deliveries} WHERE event_uuid = ?", [delivery.event_uuid]) + :ok + end + + assert {:ok, [result]} = + Dispatcher.reconcile_pending(Emitter, :order_paid, + tenant: "org_a", + older_than: cutoff(), + enqueue_key: "reconcile-vanished", + enqueue: vanishing_enqueuer + ) + + assert result.status == :endpoint_error + assert inspect(result.error) =~ "enqueue_reload_failed" + end + test "the cutoff gates the claim: fresh rows are left alone" do stranded_row!("org_a", "rec-fresh") # ...but backdate only 1 second — inside the default 5-minute cutoff