fix: target mode offset tie-breaker

src/preloaded/process/dev_arweave_scheduler.erl

@@ -12,10 +12,10 @@
 %%%       process ID, and the process message occupies slot 0 of its own
 %%%       schedule.</li>
 %%%   <li>Every L1 transaction whose `target' field is the process ID is a
-%%%       message in the process's schedule. Slots follow the canonical weave
-%%%       order (ascending weave offset, which is ascending block order). The
-%%%       transaction's weave `offset' -- not a scheduler-assigned nonce -- is
-%%%       its on-chain position, and it is recorded on the assignment.</li>
+%%%       message in the process's schedule. Slots follow the transaction's
+%%%       weave `offset' as their primary on-chain position, and equal offsets
+%%%       are ordered by the block's transaction list. The offset -- not a
+%%%       scheduler-assigned nonce -- is recorded on the assignment.</li>
 %%%   <li>Discovery is <em>local-first</em>: the node indexes the relevant
 %%%       blocks itself with `~copycat@1.0' and then queries its own
 %%%       `~query@1.0' GraphQL endpoint (`transactions(recipients:
@@ -421,10 +421,11 @@ no_result_cache(Opts) ->
 %%
 %% In `target' mode the recipient match is served from the node's own
 %% `~query@1.0' GraphQL endpoint, and each match is annotated with its offset --
-%% the sort key and the `offset' recorded on the assignment. No gateway is
-%% queried by default. In `all' mode there is nothing to match, so no query is
-%% run: the range's block headers are walked directly (see
-%% `enumerate_blocks/4').
+%% the sort key and the `offset' recorded on the assignment. If multiple
+%% matches share an offset, the range's block headers are walked to put only
+%% that ambiguous set back into block order. In `all' mode there is nothing to
+%% match, so no query is run: the range's block headers are walked directly
+%% (see `enumerate_blocks/4').
 %%
 %% Returns `{Extra, TXID}' pairs in the order they are to be assigned, where
 %% `Extra' is the sequencing detail recorded on each assignment. Bundled data
@@ -439,14 +440,63 @@ discover(ProcID, _Mode, From, To, Opts) ->
     maybe
         ok ?= ensure_offsets(From, To, Opts),
         {ok, IDs} ?= query_recipients(ProcID, From, To, Opts),
-        {ok,
+        target_offsets(ProcID, From, To, base_layer_offsets(IDs, Opts), Opts)
+    end.
+
+%% @doc Return target-mode matches in offset order unless an equal-offset group
+%% needs block order to break the tie.
+target_offsets(_ProcID, _From, _To, [], _Opts) -> {ok, []};
+target_offsets(ProcID, From, To, OffsetPairs, Opts) ->
+    case has_offset_ties(OffsetPairs) of
+        false ->
+            {ok, offset_assignments(OffsetPairs)};
+        true ->
+            maybe
+                {ok, Located} ?= enumerate_blocks(ProcID, From, To, Opts),
+                {ok, target_groups(OffsetPairs, Located)}
+            end
+    end.
+
+has_offset_ties(OffsetPairs) ->
+    Offsets = [Offset || {Offset, _TXID} <- OffsetPairs],
+    length(Offsets) =/= length(lists:usort(Offsets)).
+
+offset_assignments(OffsetPairs) ->
+    [
+        {#{ <<"offset">> => Offset }, TXID}
+    ||
+        {Offset, TXID} <- OffsetPairs
+    ].
+
+target_groups([], _Located) -> [];
+target_groups([{Offset, _TXID} = Pair | Rest], Located) ->
+    {Same, Next} = lists:splitwith(fun({NextOffset, _}) -> NextOffset =:= Offset end, Rest),
+    Group = [Pair | Same],
+    case Same of
+        [] -> offset_assignments(Group);
+        _ -> target_blocks(Group, Located)
+    end ++ target_groups(Next, Located).
+
+%% @doc Keep equal-offset target-mode matches in canonical block order, while
+%% preserving each transaction's weave offset as assignment metadata.
+target_blocks(OffsetPairs, Located) ->
+    Offsets =
+        maps:from_list(
             [
-                {#{ <<"offset">> => Offset }, TXID}
+                {TXID, Offset}
             ||
-                {Offset, TXID} <- base_layer_offsets(IDs, Opts)
+                {Offset, TXID} <- OffsetPairs
             ]
-        }
-    end.
+        ),
+    lists:filtermap(
+        fun({_Height, TXID}) ->
+            case maps:find(TXID, Offsets) of
+                {ok, Offset} -> {true, {#{ <<"offset">> => Offset }, TXID}};
+                error -> false
+            end
+        end,
+        Located
+    ).
 
 %% @doc Enumerate every transaction in a block range from the block headers
 %% themselves, in canonical chain order: blocks ascending by height, then each
@@ -634,11 +684,10 @@ to_transactions({error, Reason}, _Opts) ->
         }
     }.
 
-%% @doc Annotate each matched transaction with its weave offset from the
-%% local index, keeping only the base-layer (`tx@1.0') transactions and
-%% dropping bundled data items (indexed under the `ans104@1.0' codec) and any
-%% that are not yet locally indexed. The result is sorted by offset, which is
-%% the canonical weave order.
+%% @doc Annotate each matched transaction with its weave offset from the local
+%% index, keeping only the base-layer (`tx@1.0') transactions and dropping
+%% bundled data items (indexed under the `ans104@1.0' codec) and any that are
+%% not yet locally indexed.
 base_layer_offsets(IDs, Opts) ->
     Store = hb_store_arweave:store_from_opts(Opts),
     lists:keysort(
@@ -1085,6 +1134,44 @@ base_layer_offsets_test() ->
         base_layer_offsets([Late, Bundled, Early, Unindexed], Opts)
     ).
 
+%% @doc Target mode uses the query result as a candidate set, but block order
+%% as the sequencing order. Equal offsets are not enough to order data-free
+%% transactions.
+target_blocks_equal_offset_test() ->
+    Store = hb_test_utils:test_store(),
+    hb_store:start(Store),
+    Opts = #{ <<"store">> => [Store] },
+    ProcID = hb_util:human_id(crypto:strong_rand_bytes(32)),
+    First = hb_util:human_id(crypto:strong_rand_bytes(32)),
+    Second = hb_util:human_id(crypto:strong_rand_bytes(32)),
+    Other = hb_util:human_id(crypto:strong_rand_bytes(32)),
+    Later = hb_util:human_id(crypto:strong_rand_bytes(32)),
+    ?assertEqual(
+        {ok,
+            [
+                {#{ <<"offset">> => 99 }, First},
+                {#{ <<"offset">> => 100 }, Second}
+            ]},
+        target_offsets(ProcID, 10, 10, [{99, First}, {100, Second}], #{})
+    ),
+    ok = write_test_block(10, [Other, First, Second, Later], Opts),
+    ?assertEqual(
+        {ok,
+            [
+                {#{ <<"offset">> => 99 }, First},
+                {#{ <<"offset">> => 100 }, Other},
+                {#{ <<"offset">> => 100 }, Second},
+                {#{ <<"offset">> => 101 }, Later}
+            ]},
+        target_offsets(
+            ProcID,
+            10,
+            10,
+            [{99, First}, {100, Second}, {100, Other}, {101, Later}],
+            Opts
+        )
+    ).
+
 %% @doc The sequencing mode is read from the process message, and anything the
 %% device does not implement leaves the process sequenced as normal rather than
 %% wedging it: the process message cannot be corrected once it is on-chain.

src/preloaded/process/dev_arweave_swap.erl

@@ -264,7 +264,7 @@ register_interest(Base, Body, Height, Opts) ->
             <<"deposit">> := Deposit,
             <<"deadline">> := Deadline
         } ?= Order,
-        {ok, Paid} ?= amount(<<"reward">>, Body, Opts),
+        {ok, Paid} ?= hb_util:safe_int(tx_field(Body, <<"reward">>, 0, Opts)),
         true ?= Paid >= Fee,
         true ?= balance(Base, Buyer, Opts) >= Deposit,
         Until = Height + Deadline,