fix: fix scheduler and process flow regressions

scripts/build-preloaded-store.escript

@@ -90,6 +90,8 @@ hb_opts_compile_opts(Ebin) ->
     end.
 
 drop_outdir([{outdir, _} | Rest]) -> drop_outdir(Rest);
+drop_outdir([{d, Name, Value} | Rest]) when is_list(Name) ->
+    [{d, list_to_atom(Name), Value} | drop_outdir(Rest)];
 drop_outdir([Opt | Rest]) -> [Opt | drop_outdir(Rest)];
 drop_outdir([]) -> [].
 

scripts/hyper-token.lua

@@ -119,11 +119,25 @@ function count_common(a, b)
     if type(a) ~= "table" then a = { a } end
     if type(b) ~= "table" then b = { b } end
 
+    -- local count = 0
+    -- for _, v in ipairs(a) do
+    --     for _, w in ipairs(b) do
+    --         if v == w then
+    --             count = count + 1
+    --         end
+    --     end
+    -- end
+
+    local seen = {}
     local count = 0
     for _, v in ipairs(a) do
-        for _, w in ipairs(b) do
-            if v == w then
-                count = count + 1
+        if not seen[v] then
+            seen[v] = true
+            for _, w in ipairs(b) do
+                if v == w then
+                    count = count + 1
+                    break
+                end
             end
         end
     end
@@ -895,4 +909,4 @@ function compute(base, assignment)
         ao.event({ "Process initialized.", { slot = assignment.slot } })
         return "ok", base
     end
-end
+end

src/core/http/hb_client_remote.erl

@@ -98,7 +98,13 @@ upload(Msg, Opts) ->
             end,
             hb_message:commitment_devices(Msg, Opts)
         ),
-    {ok, UploadResults}.
+    case lists:filter(fun upload_failed/1, UploadResults) of
+        [] -> {ok, UploadResults};
+        Errors -> {error, Errors}
+    end.
+upload_failed({error, _}) -> true;
+upload_failed({failure, _}) -> true;
+upload_failed(_) -> false.
 upload(Msg, Opts, <<"httpsig@1.0">>) ->
     case hb_opts:get(bundler_httpsig, not_found, Opts) of
         not_found ->
@@ -107,13 +113,17 @@ upload(Msg, Opts, <<"httpsig@1.0">>) ->
             ?event({uploading_item, Msg}),
             hb_http:post(Bundler, <<"/tx">>, Msg, Opts)
     end;
-upload(Msg, Opts, _CommitmentDevice) ->
+upload(Msg, Opts, CommitmentDevice) ->
     ?event({uploading_item, Msg}),
     hb_ao:raw(
         <<"arweave@2.9">>,
         <<"tx">>,
-        #{},
-        Msg#{ <<"method">> => <<"POST">> },
+        Msg,
+        #{
+            <<"method">> => <<"POST">>,
+            <<"target">> => <<"base">>,
+            <<"commitment-device">> => CommitmentDevice
+        },
         Opts
     ).
 

src/core/http/hb_http.erl

@@ -638,9 +638,15 @@ encode_reply(Status, TABMReq, Message, Opts) ->
             end,
 			Opts
         ),
+    DefaultAcceptBundle =
+        case {Codec, hb_maps:get(<<"require-codec">>, TABMReq, not_found, Opts)} of
+            {<<"json@1.0">>, not_found} -> false;
+            {<<"json@1.0">>, _} -> true;
+            _ -> false
+        end,
     AcceptBundle =
         hb_util:atom(
-            hb_maps:get(<<"accept-bundle">>, TABMReq, false, Opts)
+            hb_maps:get(<<"accept-bundle">>, TABMReq, DefaultAcceptBundle, Opts)
         ),
     ?event(debug_http,
         {encoding_reply,
@@ -1054,6 +1060,16 @@ normalize_unsigned(PrimMsg, Req = #{ headers := RawHeaders }, Msg, Opts) ->
         ),
     FilterKeys = hb_opts:get(http_inbound_filter_keys, ?DEFAULT_FILTER_KEYS, Opts),
     FilteredMsg = hb_message:without_unless_signed(FilterKeys, Msg, Opts),
+    DefaultAcceptBundle =
+        case maps:get(
+            <<"require-codec">>,
+            Msg,
+            maps:get(<<"require-codec">>, PrimMsg, not_found)
+        ) of
+            <<"application/json">> -> true;
+            <<"json@1.0">> -> true;
+            _ -> maps:get(<<"accept-bundle">>, RawHeaders, false)
+        end,
     BaseMsg =
         FilteredMsg#{
             <<"method">> => Method,
@@ -1065,7 +1081,7 @@ normalize_unsigned(PrimMsg, Req = #{ headers := RawHeaders }, Msg, Opts) ->
                     maps:get(
                         <<"accept-bundle">>,
                         PrimMsg,
-                        maps:get(<<"accept-bundle">>, RawHeaders, false)
+                        DefaultAcceptBundle
                     )
                 ),
             <<"accept">> =>

src/core/resolver/hb_cache_control.erl

@@ -56,7 +56,14 @@ lookup(Base, Req, Opts) ->
                     Opts,
                     hb_opts:get(store_scope_resolved, local, Opts)
                 ),
-            case hb_cache:read_resolved(Base, Req, OutputScopedOpts) of
+            CacheRead =
+                try hb_cache:read_resolved(Base, Req, OutputScopedOpts) of
+                    ReadRes -> ReadRes
+                catch
+                    throw:{necessary_message_not_found, _, _} ->
+                        miss
+                end,
+            case CacheRead of
                 {hit, not_found} ->
                     {error, not_found};
                 {hit, {ok, Res}} ->

src/core/resolver/hb_opts.erl

@@ -531,7 +531,7 @@ raw_default_message() ->
                     }
                 ]
         },
-        <<"scheduler-default-commitment-spec">> => <<"httpsig@1.0">>,
+        <<"scheduler-default-commitment-spec">> => <<"ans104@1.0">>,
         <<"genesis-wasm-import-authorities">> =>
             [
                 <<"WjnS-s03HWsDSdMnyTdzB1eHZB2QheUWP_FVRVYxkXk">>

src/preloaded/arweave/dev_arweave.erl

@@ -47,11 +47,16 @@ tx(Base, Request, Opts) ->
 %% you should use the ~bundler@1.0 device.
 post_tx(Base, RawRequest, Opts) ->
     {ok, Request} = extract_target(Base, RawRequest, Opts),
-    case hb_maps:find(<<"commitment-device">>, Request, Opts) of
+    case hb_maps:find(<<"commitment-device">>, RawRequest, Opts) of
         {ok, Device} ->
             post_tx(Base, Request, Opts, Device);
         error ->
-            post_tx_detect_device(Base, Request, Opts)
+            case hb_maps:find(<<"commitment-device">>, Request, Opts) of
+                {ok, Device} ->
+                    post_tx(Base, Request, Opts, Device);
+                error ->
+                    post_tx_detect_device(Base, Request, Opts)
+            end
     end.
 
 %% @doc Detect the commitment device to use when posting a transaction.

src/preloaded/codec/dev_json.erl

@@ -27,9 +27,10 @@ to(Msg, Req, Opts) ->
             tabm,
             ConvOpts
         ),
+    Bundle = hb_maps:get(<<"bundle">>, Req, false, Opts),
     Loaded =
-        case hb_maps:get(<<"bundle">>, Req, false, Opts) of
-            true -> hb_cache:ensure_all_loaded(Restructured, Opts);
+        case Bundle of
+            true -> load_available_links(hb_link:decode_all_links(Restructured), Opts);
             false -> Restructured
         end,
     JSONStructured =
@@ -38,12 +39,37 @@ to(Msg, Req, Opts) ->
             tabm,
             #{
                 <<"device">> => <<"structured@1.0">>,
+                <<"bundle">> => Bundle,
                 <<"encode-types">> => [<<"atom">>]
             },
             ConvOpts
         ),
     {ok, hb_json:encode(JSONStructured)}.
 
+%% @doc Eager-load resolvable links for bundled JSON responses, while leaving
+%% missing lazy links in place so optional response fields do not fail encoding.
+load_available_links(Msg, Opts) ->
+    load_available_links([], Msg, Opts).
+
+load_available_links(_Ref, Link, Opts) when ?IS_LINK(Link) ->
+    try hb_cache:ensure_loaded(Link, Opts) of
+        Loaded -> load_available_links([], Loaded, Opts)
+    catch
+        throw:{necessary_message_not_found, _, _} -> Link
+    end;
+load_available_links(Ref, Msg, Opts) when is_map(Msg) ->
+    maps:map(
+        fun(K, V) -> load_available_links([K|Ref], V, Opts) end,
+        Msg
+    );
+load_available_links(Ref, Msg, Opts) when is_list(Msg) ->
+    lists:map(
+        fun({N, V}) -> load_available_links([N|Ref], V, Opts) end,
+        hb_util:number(Msg)
+    );
+load_available_links(_Ref, Msg, _Opts) ->
+    Msg.
+
 %% @doc Decode a JSON string to a message.
 from(Map, _Req, _Opts) when is_map(Map) -> {ok, Map};
 from(JSON, Req, Opts) ->

src/preloaded/codec/dev_json_iface.erl

@@ -208,14 +208,25 @@ prepare_tags(Msg, Opts) ->
                 {ok, OriginalTags} ->
                     Res = hb_util:message_to_ordered_list(OriginalTags),
                     ?event({using_original_tags, Res}),
-                    Res;
+                    case complete_tags(Res) of
+                        true -> Res;
+                        false -> prepare_header_case_tags(Msg, Opts)
+                    end;
                 error -> 
                     prepare_header_case_tags(Msg, Opts)
             end;
         _ ->
             prepare_header_case_tags(Msg, Opts)
     end.
 
+complete_tags(Tags) ->
+    lists:all(
+        fun(#{ <<"name">> := _, <<"value">> := _ }) -> true;
+           (_) -> false
+        end,
+        Tags
+    ).
+
 %% @doc Convert a message without an `original-tags' field into a list of
 %% key-value pairs, with the keys in HTTP header-case.
 prepare_header_case_tags(TABM, Opts) ->

src/preloaded/codec/lib_arweave_common.erl

@@ -535,7 +535,9 @@ original_tags_to_tags(TagMap) ->
     ?event({ordered_tagmap, {explicit, OrderedList}, {input, {explicit, TagMap}}}),
     lists:map(
         fun(#{ <<"name">> := Key, <<"value">> := Value }) ->
-            {Key, Value}
+            {Key, Value};
+           (#{ <<"name">> := Key }) ->
+            {Key, <<>>}
         end,
         OrderedList
     ).

src/preloaded/process/dev_process.erl

@@ -217,7 +217,7 @@ compute(Base, Req, Opts) ->
                             {result, Result}
                         }
                     ),
-                    {ok, without_snapshot(Result, Opts)};
+                    {ok, compute_response(Result, Req, Opts)};
                 {error, not_found} ->
                     {ok, Loaded} = ensure_loaded(ProcBase, Req, Opts),
                     ?event(compute,
@@ -259,7 +259,7 @@ compute_to_slot(ProcID, Base, Req, TargetSlot, Opts) ->
                 Opts
             ),
             store_result(true, ProcID, TargetSlot, Base, Req, Opts),
-            {ok, without_snapshot(lib_process:as_process(Base, Opts), Opts)};
+            {ok, compute_response(lib_process:as_process(Base, Opts), Req, Opts)};
         CurrentSlot when CurrentSlot < TargetSlot ->
             % Compute the next state transition.
             NextSlot = CurrentSlot + 1,
@@ -646,7 +646,7 @@ now(RawBase, Req, Opts) ->
             LatestKnown = dev_process_cache:latest(ProcessID, [], Opts),
             case LatestKnown of
                 {ok, LatestSlot, RawLatestMsg} ->
-                    LatestMsg = without_snapshot(RawLatestMsg, Opts),
+                    LatestMsg = compute_response(RawLatestMsg, Req, Opts),
                     ?event(compute_cache,
                         {serving_latest_cached_state,
                             {proc_id, ProcessID},
@@ -791,3 +791,17 @@ ensure_loaded(Base, Req, Opts) ->
 %% @doc Remove the `snapshot' key from a message and return it.
 without_snapshot(Msg, Opts) ->
     hb_ao:set(Msg, <<"snapshot">>, unset, Opts).
+
+%% @doc Format a compute response for the caller with its result payload inline.
+compute_response(Msg, _Req, Opts) ->
+    with_loaded_results(without_snapshot(Msg, Opts), Opts).
+
+with_loaded_results(Msg, Opts) ->
+    Decoded = hb_link:decode_all_links(Msg),
+    case hb_maps:get(<<"results">>, Decoded, not_found, Opts) of
+        not_found -> Msg;
+        Results ->
+            (maps:remove(<<"results+link">>, Msg))#{
+                <<"results">> => hb_cache:ensure_all_loaded(Results, Opts)
+            }
+    end.

src/preloaded/process/dev_push.erl

@@ -44,7 +44,7 @@ push(Base, Req, Opts) ->
         no_slot ->
             case schedule_initial_message(Process, Req, Opts) of
                 {ok, Assignment} ->
-                    case find_type(hb_ao:get(<<"body">>, Assignment, Opts), Opts) of
+                    case find_type(Req, Opts) of
                         <<"Process">> ->
                             ?event(push,
                                 {initializing_process,
@@ -459,12 +459,15 @@ push_downstream_local(TargetID, NextSlotOnProc, Origin, Opts) ->
             {origin, Origin}
         }
     ),
+    ResultDepth =
+        decrement_result_depth(
+            hb_maps:get(<<"result-depth">>, Origin, 1, Opts)
+        ),
     BaseReq =
         #{
             <<"path">> => <<"push">>,
             <<"slot">> => NextSlotOnProc,
-            <<"result-depth">> =>
-                hb_maps:get(<<"result-depth">>, Origin, 1, Opts) - 1
+            <<"result-depth">> => ResultDepth
         },
     Req =
         case parse_max_depth(hb_maps:get(<<"max-depth">>, Origin, undefined, Opts)) of
@@ -492,6 +495,15 @@ parse_max_depth(Bin) when is_binary(Bin) ->
     end;
 parse_max_depth(_) -> undefined.
 
+decrement_result_depth(Depth) when is_integer(Depth), Depth > 0 -> Depth - 1;
+decrement_result_depth(Depth) when is_binary(Depth) ->
+    try hb_util:int(Depth) of
+        N -> decrement_result_depth(N)
+    catch
+        _:_ -> 0
+    end;
+decrement_result_depth(_) -> 0.
+
 %% @doc Augment the message with from-* keys, if it doesn't already have them.
 normalize_message(MsgToPush, Opts) ->
     hb_ao:set(
@@ -990,7 +1002,7 @@ test_push_as_identity() ->
 
 test_multi_process_push() ->
     {Sender, _Receiver, MsgSlot, Opts} = setup_two_process_message(),
-    %% Install a catch-all `Pong' handler on the Sender so the Receiver's
+    %% Install a `Pong' handler on the Sender so the Receiver's
     %% reply (the helper's `reply_script' fires on `Action = "Ping"' and
     %% sends back `Action = "Reply"') is observable as `GOT PONG' in the
     %% Sender's `now/results/data'.
@@ -999,7 +1011,9 @@ test_multi_process_push() ->
             Sender,
             <<
                 "Handlers.add(\"Pong\",\n"
-                "   function (test) return true end,\n"
+                "   function (test)\n"
+                "       return (test.Action or test.action) == \"Reply\"\n"
+                "   end,\n"
                 "   function(m)\n"
                 "       print(\"GOT PONG\")\n"
                 "   end\n"
@@ -1585,9 +1599,11 @@ test_nested_push_prompts_encoding_change() ->
 ping_pong_script(Limit) ->
     <<
         "Handlers.add(\"Ping\",\n"
-        "   function (test) return true end,\n"
+        "   function (test)\n"
+        "       return (test.Action or test.action) == \"Ping\"\n"
+        "   end,\n"
         "   function(m)\n"
-        "       C = tonumber(m.Count)\n"
+        "       C = tonumber(m.Count or m.count)\n"
         "       if C <= ", (integer_to_binary(Limit))/binary, " then\n"
         "           Send({ Target = ao.id, Action = \"Ping\", Count = C + 1 })\n"
         "           print(\"Ping\", C + 1)\n"
@@ -1603,11 +1619,14 @@ reply_script() ->
     <<
         """
         Handlers.add("Reply",
-           { Action = "Ping" },
            function(m)
+               return (m.Action or m.action) == "Ping"
+           end,
+           function(m)
+               local from = m.From or m.from
                print("Replying to...")
-               print(m.From)
-               Send({ Target = m.From, Action = "Reply", Message = "Pong!" })
+               print(from)
+               Send({ Target = from, Action = "Reply", Message = "Pong!" })
                print("Done.")
            end
         )