fed-sx-m2: Step 8b-pure — retry-time bookkeeping + 11 tests + 2 Blockers

delivery_worker state shape gains :next_retry proplist alongside
the existing :attempts:

  [{peer, _}, {pending, _}, {attempts, [{Cid, N}]},
   {next_retry, [{Cid, NextRetryAt}]}, {dead_letter, _},
   {dispatch_fn, _}]

New pure-functional exports:
  record_failure_pure/3(Cid, Now, State)
      Bumps :attempts for Cid. On the 6th failure
      (backoff_for returns dead_letter) moves the matching
      activity from :pending to :dead_letter and clears the
      :next_retry entry. Otherwise sets next_retry to
      Now + backoff_for(NewAttempts).
  record_success_pure/2(Cid, State)
      Clears both :attempts and :next_retry for Cid.
  next_due_pure/2(Now, State)
      Returns cids whose retry time has passed (insertion
      order preserved so the worker drains in FIFO retry
      order).
  attempts_for/2, next_retry_at/2, dead_letter_list/1
      Read-side accessors.

Internal helper move_to_dead_letter/2 + take_by_cid/4 walks
:pending to find the matching activity by cid.

11/11 in next/tests/delivery_retry.sh covering:
  - fresh state: 0 attempts / undefined retry / [] dead_letter
  - record_failure bumps to 1
  - record_failure sets next_retry_at = Now + 30 (slot 1)
  - second failure: attempts=2, NextRetryAt = Now + 300 (slot 2)
  - record_success clears both
  - next_due returns due cids
  - next_due empty before due
  - 6th failure -> dead-letter; activity out of :pending
  - dead-lettered cid removed from :next_retry
  - per-cid isolation: success on one doesn't disturb another

delivery_worker.sh 17/17 unchanged (new exports are additive).

Blockers added:
  #2 — Native http-request primitive missing in bin/sx_server.ml
       (briefing assumed it existed; only http-listen exists).
       Belongs to loops/fed-prims. Step 8e wrapper waits for
       the native.
  #3 — erlang:send_after-style timer primitive missing. Needed
       for the real retry loop. Belongs to loops/erlang. 8b-pure
       captures the semantics so 8b-timer is a 1-shot wiring
       when the primitive lands.

Conformance preserved at 761/761.

next/kernel/delivery_worker.erl

@@ -3,6 +3,9 @@
 -export([new/1, pending/1, peer/1,
          enqueue_pure/3, drain_pure/1, deliver_one_pure/2,
          backoff_for/1, schedule_for/1,
+         record_failure_pure/3, record_success_pure/2,
+         next_due_pure/2, attempts_for/2, next_retry_at/2,
+         dead_letter_list/1,
          start_link/1, start_link/2, stop/1,
          enqueue/2, flush/1, pending_srv/1, set_dispatch_fn/2]).
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2]).
@@ -31,6 +34,7 @@
 %%   [{peer, PeerId},
 %%    {pending, [Activity, ...]},          %% FIFO; head delivered first
 %%    {attempts, [{Cid, AttemptCount}, ...]},
+%%    {next_retry, [{Cid, NextRetryAt}, ...]}, %% Step 8b-pure
 %%    {dead_letter, [Activity, ...]},
 %%    {dispatch_fn, fun/1 | undefined}]
 %%
@@ -43,6 +47,7 @@ new(PeerId) ->
     [{peer, PeerId},
      {pending, []},
      {attempts, []},
+     {next_retry, []},
      {dead_letter, []},
      {dispatch_fn, undefined}].
 
@@ -118,6 +123,85 @@ schedule_for(Attempts) ->
         Seconds     -> {retry_in, Seconds}
     end.
 
+%% ── Step 8b-pure: retry-time bookkeeping ───────────────────────
+%%
+%% `record_failure_pure/3(Cid, Now, State)` — call after a failed
+%% deliver_one. Bumps the per-cid attempt counter; if the new
+%% attempt is past the dead-letter threshold, moves the matching
+%% activity from :pending to :dead_letter. Otherwise records the
+%% next retry time as Now + backoff_for(NewAttempt).
+%%
+%% Real timer wiring (erlang:send_after self-cast on the worker
+%% pid) needs substrate support — Step 8b-timer when that lands.
+%%
+%% `record_success_pure/2(Cid, State)` — clears :attempts and
+%% :next_retry entries for the cid; called after a successful
+%% deliver_one.
+%%
+%% `next_due_pure/2(Now, State)` — returns the list of Cids whose
+%% NextRetryAt has passed, in insertion order.
+
+record_failure_pure(Cid, Now, State) ->
+    Attempts = field(attempts, State),
+    Current  = case find_keyed(Cid, Attempts) of
+        {ok, N} -> N;
+        _       -> 0
+    end,
+    New = Current + 1,
+    State1 = set_field(attempts, set_keyed(Cid, New, Attempts), State),
+    case backoff_for(New) of
+        dead_letter ->
+            move_to_dead_letter(Cid, State1);
+        Seconds ->
+            NextAt = Now + Seconds,
+            NR = field(next_retry, State1),
+            set_field(next_retry, set_keyed(Cid, NextAt, NR), State1)
+    end.
+
+record_success_pure(Cid, State) ->
+    A1 = del_keyed(Cid, field(attempts, State)),
+    NR1 = del_keyed(Cid, field(next_retry, State)),
+    set_field(attempts, A1, set_field(next_retry, NR1, State)).
+
+%% next_due_pure/2 — Cids whose NextRetryAt <= Now. Preserves
+%% insertion order so the worker drains them in FIFO retry order.
+
+next_due_pure(Now, State) ->
+    [Cid || {Cid, At} <- field(next_retry, State), At =< Now].
+
+attempts_for(Cid, State) ->
+    case find_keyed(Cid, field(attempts, State)) of
+        {ok, N} -> N;
+        _       -> 0
+    end.
+
+next_retry_at(Cid, State) ->
+    case find_keyed(Cid, field(next_retry, State)) of
+        {ok, At} -> At;
+        _        -> undefined
+    end.
+
+dead_letter_list(State) -> field(dead_letter, State).
+
+move_to_dead_letter(Cid, State) ->
+    Pending = field(pending, State),
+    {Match, Rest} = take_by_cid(Cid, Pending, [], []),
+    DL = field(dead_letter, State),
+    State1 = set_field(pending, Rest, State),
+    State2 = case Match of
+        none -> State1;
+        Act  -> set_field(dead_letter, DL ++ [Act], State1)
+    end,
+    NR = field(next_retry, State2),
+    set_field(next_retry, del_keyed(Cid, NR), State2).
+
+take_by_cid(_, [], Acc, _) -> {none, lists:reverse(Acc)};
+take_by_cid(Cid, [A | Rest], Acc, _) ->
+    case activity_cid(A) of
+        Cid -> {A, lists:reverse(Acc) ++ Rest};
+        _   -> take_by_cid(Cid, Rest, [A | Acc], 0)
+    end.
+
 %% ── gen_server wrapper ──────────────────────────────────────────
 
 start_link(PeerId) ->
@@ -196,3 +280,7 @@ find_keyed(K, [_ | Rest]) -> find_keyed(K, Rest).
 set_keyed(K, V, []) -> [{K, V}];
 set_keyed(K, V, [{K, _} | Rest]) -> [{K, V} | Rest];
 set_keyed(K, V, [P | Rest]) -> [P | set_keyed(K, V, Rest)].
+
+del_keyed(_, []) -> [];
+del_keyed(K, [{K, _} | Rest]) -> Rest;
+del_keyed(K, [P | Rest]) -> [P | del_keyed(K, Rest)].

next/tests/delivery_retry.sh

@@ -0,0 +1,126 @@
+#!/usr/bin/env bash
+# next/tests/delivery_retry.sh — m2 Step 8b-pure test.
+#
+# Pure-functional retry-time bookkeeping for the delivery worker.
+# record_failure bumps the attempt counter and computes the next
+# retry time per backoff_for. record_success clears state for a
+# cid. next_due returns cids whose retry time has passed.
+#
+# Real timer wiring (erlang:send_after self-cast) is Step 8b-timer
+# once substrate support lands.
+
+set -uo pipefail
+cd "$(git rev-parse --show-toplevel)"
+
+SX_SERVER="${SX_SERVER:-hosts/ocaml/_build/default/bin/sx_server.exe}"
+if [ ! -x "$SX_SERVER" ]; then
+  SX_SERVER="/root/rose-ash/hosts/ocaml/_build/default/bin/sx_server.exe"
+fi
+if [ ! -x "$SX_SERVER" ]; then
+  echo "ERROR: sx_server.exe not found." >&2
+  exit 1
+fi
+
+VERBOSE="${1:-}"
+PASS=0; FAIL=0; ERRORS=""
+TMPFILE=$(mktemp); trap "rm -f $TMPFILE" EXIT
+
+SETUP='Act1 = [{id, <<1>>}, {type, note}, {actor, alice}], Act2 = [{id, <<2>>}, {type, note}, {actor, alice}],'
+
+cat > "$TMPFILE" <<EPOCHS
+(epoch 1)
+(load "lib/erlang/tokenizer.sx")
+(load "lib/erlang/parser.sx")
+(load "lib/erlang/parser-core.sx")
+(load "lib/erlang/parser-expr.sx")
+(load "lib/erlang/parser-module.sx")
+(load "lib/erlang/transpile.sx")
+(load "lib/erlang/runtime.sx")
+(load "lib/erlang/vm/dispatcher.sx")
+(epoch 2)
+(eval "(get (erlang-load-module (file-read \"next/kernel/envelope.erl\")) :name)")
+(epoch 3)
+(eval "(get (erlang-load-module (file-read \"next/kernel/delivery_worker.erl\")) :name)")
+
+;; Fresh state: no attempts, no next_retry, no dead_letter
+(epoch 10)
+(eval "(get (erlang-eval-ast \"${SETUP} S = delivery_worker:new(bob), {delivery_worker:attempts_for(<<1>>, S), delivery_worker:next_retry_at(<<1>>, S), delivery_worker:dead_letter_list(S)} =:= {0, undefined, []}\") :name)")
+
+;; record_failure bumps the attempt counter
+(epoch 11)
+(eval "(get (erlang-eval-ast \"${SETUP} S0 = delivery_worker:enqueue_pure(bob, Act1, delivery_worker:new(bob)), S1 = delivery_worker:record_failure_pure(<<1>>, 1000, S0), delivery_worker:attempts_for(<<1>>, S1) =:= 1\") :name)")
+
+;; record_failure sets next_retry_at = Now + backoff(1) = Now + 30
+(epoch 12)
+(eval "(get (erlang-eval-ast \"${SETUP} S0 = delivery_worker:enqueue_pure(bob, Act1, delivery_worker:new(bob)), S1 = delivery_worker:record_failure_pure(<<1>>, 1000, S0), delivery_worker:next_retry_at(<<1>>, S1) =:= 1030\") :name)")
+
+;; Second failure -> attempts=2, NextRetryAt = Now+300
+(epoch 13)
+(eval "(get (erlang-eval-ast \"${SETUP} S0 = delivery_worker:enqueue_pure(bob, Act1, delivery_worker:new(bob)), S1 = delivery_worker:record_failure_pure(<<1>>, 1000, S0), S2 = delivery_worker:record_failure_pure(<<1>>, 2000, S1), {delivery_worker:attempts_for(<<1>>, S2), delivery_worker:next_retry_at(<<1>>, S2)} =:= {2, 2300}\") :name)")
+
+;; record_success clears attempts + next_retry for the cid
+(epoch 14)
+(eval "(get (erlang-eval-ast \"${SETUP} S0 = delivery_worker:enqueue_pure(bob, Act1, delivery_worker:new(bob)), S1 = delivery_worker:record_failure_pure(<<1>>, 1000, S0), S2 = delivery_worker:record_success_pure(<<1>>, S1), {delivery_worker:attempts_for(<<1>>, S2), delivery_worker:next_retry_at(<<1>>, S2)} =:= {0, undefined}\") :name)")
+
+;; next_due returns Cids whose retry time has passed
+(epoch 15)
+(eval "(get (erlang-eval-ast \"${SETUP} S0 = delivery_worker:enqueue_pure(bob, Act1, delivery_worker:new(bob)), S1 = delivery_worker:record_failure_pure(<<1>>, 1000, S0), delivery_worker:next_due_pure(1030, S1) =:= [<<1>>]\") :name)")
+
+;; next_due returns [] before retry time
+(epoch 16)
+(eval "(get (erlang-eval-ast \"${SETUP} S0 = delivery_worker:enqueue_pure(bob, Act1, delivery_worker:new(bob)), S1 = delivery_worker:record_failure_pure(<<1>>, 1000, S0), delivery_worker:next_due_pure(1020, S1) =:= []\") :name)")
+
+;; 6th failure -> dead_letter; activity moves out of :pending
+(epoch 17)
+(eval "(get (erlang-eval-ast \"${SETUP} F = fun(S) -> delivery_worker:record_failure_pure(<<1>>, 1000, S) end, S0 = delivery_worker:enqueue_pure(bob, Act1, delivery_worker:new(bob)), S6 = F(F(F(F(F(F(S0)))))), {delivery_worker:dead_letter_list(S6), delivery_worker:pending(S6)} =:= {[Act1], []}\") :name)")
+
+;; Dead-lettered cid is no longer in next_retry
+(epoch 18)
+(eval "(get (erlang-eval-ast \"${SETUP} F = fun(S) -> delivery_worker:record_failure_pure(<<1>>, 1000, S) end, S0 = delivery_worker:enqueue_pure(bob, Act1, delivery_worker:new(bob)), S6 = F(F(F(F(F(F(S0)))))), delivery_worker:next_retry_at(<<1>>, S6) =:= undefined\") :name)")
+
+;; Two cids: success on one doesn't disturb the other's retry state
+(epoch 19)
+(eval "(get (erlang-eval-ast \"${SETUP} S0 = delivery_worker:enqueue_pure(bob, Act1, delivery_worker:enqueue_pure(bob, Act2, delivery_worker:new(bob))), S1 = delivery_worker:record_failure_pure(<<1>>, 1000, S0), S2 = delivery_worker:record_failure_pure(<<2>>, 1000, S1), S3 = delivery_worker:record_success_pure(<<1>>, S2), delivery_worker:next_retry_at(<<2>>, S3) =:= 1030\") :name)")
+EPOCHS
+
+OUTPUT=$(timeout 240 "$SX_SERVER" < "$TMPFILE" 2>/dev/null)
+
+check() {
+  local epoch="$1" desc="$2" expected="$3"
+  local actual
+  actual=$(echo "$OUTPUT" | awk -v e="$epoch" '
+    $0 ~ "^\\(ok-len " e " " { getline; print; exit }
+    $0 ~ "^\\(ok " e " "     { print; exit }
+    $0 ~ "^\\(error " e " "  { print; exit }
+  ')
+  [ -z "$actual" ] && actual="<no output for epoch $epoch>"
+  if echo "$actual" | grep -qF -- "$expected"; then
+    PASS=$((PASS+1))
+    [ "$VERBOSE" = "-v" ] && echo "  ok $desc"
+  else
+    FAIL=$((FAIL+1))
+    ERRORS+="  FAIL [$desc] (epoch $epoch) expected: $expected | actual: $actual
+"
+  fi
+}
+
+check  3  "module loaded"                      "delivery_worker"
+check 10  "fresh state empty"                  "true"
+check 11  "record_failure bumps attempts"      "true"
+check 12  "record_failure sets next_retry_at"  "true"
+check 13  "second failure: slot 2 = +300"      "true"
+check 14  "record_success clears state"        "true"
+check 15  "next_due returns due cids"          "true"
+check 16  "next_due empty before due"          "true"
+check 17  "6th failure -> dead_letter"         "true"
+check 18  "dead-lettered cid out of retry"     "true"
+check 19  "success on one preserves other"     "true"
+
+TOTAL=$((PASS+FAIL))
+if [ $FAIL -eq 0 ]; then
+  echo "ok $PASS/$TOTAL next/tests/delivery_retry.sh passed"
+else
+  echo "FAIL $PASS/$TOTAL passed, $FAIL failed:"
+  echo "$ERRORS"
+fi
+[ $FAIL -eq 0 ]