From 0cf6a69826b4af00c6a611165fa3bc1ca021e9f9 Mon Sep 17 00:00:00 2001 From: Nick Juliano Date: Thu, 16 Jul 2026 14:21:43 -0400 Subject: [PATCH 1/3] fix: stabilize process uploads and scheduling --- src/core/http/hb_client_remote.erl | 26 ++++- .../process/dev_scheduler_server.erl | 88 +++++++++++++-- src/preloaded/vm/dev_lua.erl | 100 +++++++++++++++++- 3 files changed, 199 insertions(+), 15 deletions(-) diff --git a/src/core/http/hb_client_remote.erl b/src/core/http/hb_client_remote.erl index 579703c4ce..84db58c409 100644 --- a/src/core/http/hb_client_remote.erl +++ b/src/core/http/hb_client_remote.erl @@ -112,8 +112,11 @@ upload(Msg, Opts, _CommitmentDevice) -> hb_ao:raw( <<"arweave@2.9">>, <<"tx">>, - #{}, - Msg#{ <<"method">> => <<"POST">> }, + Msg, + #{ + <<"method">> => <<"POST">>, + <<"target">> => <<"base">> + }, Opts ). @@ -154,3 +157,22 @@ upload_single_layer_message_test() -> Result = upload(Committed, Opts, <<"ans104@1.0">>), ?event({upload_result, Result}), ?assertMatch({ok, _}, Result). + +%% @doc A message's AO target is transaction data, not the arweave device's +%% request/base selector. Upload it as the base so a process ID target is not +%% misinterpreted as a selector path. +upload_targeted_message_test() -> + Opts = upload_test_opts(), + Msg = #{ + <<"target">> => hb_util:human_id(<<0:256>>), + <<"data">> => <<"TEST">> + }, + Committed = + hb_message:commit( + Msg, + Opts, + <<"ans104@1.0">> + ), + Result = upload(Committed, Opts, <<"ans104@1.0">>), + ?event({upload_result, Result}), + ?assertMatch({ok, _}, Result). diff --git a/src/preloaded/process/dev_scheduler_server.erl b/src/preloaded/process/dev_scheduler_server.erl index 3a4f4ea95c..559e68bc69 100644 --- a/src/preloaded/process/dev_scheduler_server.erl +++ b/src/preloaded/process/dev_scheduler_server.erl @@ -263,17 +263,7 @@ do_assign(State, Message, ReplyPID) -> State ), ?event(writes_complete), - ?event(uploading_message), - hb_client_remote:upload(Message, Opts), - hb_client_remote:upload(Assignment, Opts), - ?event(uploads_complete), - maybe_inform_recipient( - remote_confirmation, - ReplyPID, - Message, - Assignment, - State - ) + upload_after_write(Message, Assignment, ReplyPID, State) end, case hb_opts:get(scheduling_mode, sync, Opts) of aggressive -> @@ -288,6 +278,43 @@ do_assign(State, Message, ReplyPID) -> base_state_hashpath := next_hashpath(BaseStateHashpath, Assignment, State) }. +%% @doc Upload a durably written assignment without allowing a remote upload +%% failure to roll back the scheduler's in-memory slot. In local-confirmation +%% mode the caller has already received the assignment at this point; reusing +%% the old state would let the next message overwrite that accepted slot. +upload_after_write(Message, Assignment, ReplyPID, State) -> + Opts = maps:get(opts, State), + Upload = hb_opts:get( + scheduler_upload_fun, + fun hb_client_remote:upload/2, + Opts + ), + try + ?event(uploading_message), + Upload(Message, Opts), + Upload(Assignment, Opts), + ?event(uploads_complete), + maybe_inform_recipient( + remote_confirmation, + ReplyPID, + Message, + Assignment, + State + ) + catch + Class:Reason:Stack -> + ?event(error, + {assignment_upload_failed_after_write, + {class, Class}, + {reason, Reason}, + {trace, Stack}, + {process, maps:get(id, State)}, + {slot, hb_ao:get(<<"slot">>, Assignment, Opts)} + } + ), + ok + end. + %% @doc Commit to the assignment using all of our appropriate wallets. commit_assignment(BaseAssignment, State) -> Wallets = maps:get(wallets, State), @@ -368,6 +395,45 @@ new_proc_test() -> #{ current := 2 }, dev_scheduler_server:info(dev_scheduler_registry:find(ID)) ). + +%% @doc Once local confirmation follows a durable cache write, a subsequent +%% upload failure must not reset the scheduler to its previous slot. +local_confirmation_upload_failure_preserves_slot_test() -> + Wallet = hb:wallet(), + Address = hb_util:human_id(ar_wallet:to_address(Wallet)), + Opts = #{ + <<"priv-wallet">> => Wallet, + <<"store">> => [hb_test_utils:test_store()], + <<"scheduling-mode">> => local_confirmation, + <<"scheduler-upload-fun">> => + fun(_, _) -> error(simulated_upload_failure) end + }, + Process = hb_message:commit( + #{ + <<"type">> => <<"Process">>, + <<"scheduler">> => Address, + <<"test-random-seed">> => rand:uniform(1000000) + }, + Opts + ), + ProcID = hb_message:id(Process, all, Opts), + PID = start(ProcID, Process, Opts), + First = schedule(PID, Process), + ?assertEqual(0, hb_ao:get(<<"slot">>, First, Opts)), + ?assertMatch(#{ current := 0 }, info(PID)), + Message = hb_message:commit( + #{ + <<"type">> => <<"Message">>, + <<"target">> => ProcID, + <<"data">> => <<"next">> + }, + Opts + ), + Second = schedule(PID, Message), + ?assertEqual(1, hb_ao:get(<<"slot">>, Second, Opts)), + ?assertMatch(#{ current := 1 }, info(PID)), + stop(PID), + hb_name:unregister({<<"scheduler@1.0">>, ProcID}). benchmark_test() -> diff --git a/src/preloaded/vm/dev_lua.erl b/src/preloaded/vm/dev_lua.erl index e09c8a289e..df88f2e303 100644 --- a/src/preloaded/vm/dev_lua.erl +++ b/src/preloaded/vm/dev_lua.erl @@ -148,8 +148,8 @@ load_modules([Module|Rest], Opts, Acc) when is_map(Module) -> ModuleBin = hb_ao:get_first( [ - {Module, <<"body">>}, - {Module, <<"data">>} + {{as, <<"message@1.0">>, Module}, <<"body">>}, + {{as, <<"message@1.0">>, Module}, <<"data">>} ], Module, Opts @@ -909,6 +909,102 @@ pure_lua_process_test() -> {ok, Results} = hb_ao:resolve(Process, <<"now">>, #{}), ?assertEqual(42, hb_ao:get(<<"results/output/body">>, Results, #{})). +%% @doc An inline ANS-104 process stores its Lua source under `data'. Ensure +%% module discovery reads that field literally instead of invoking Lua's +%% default handler for the absent `body' field during initialization. +inline_lua_data_process_test() -> + Opts = #{ <<"priv-wallet">> => hb:wallet() }, + Wallet = hb_opts:get(priv_wallet, hb:wallet(), Opts), + Address = hb_util:human_id(ar_wallet:to_address(Wallet)), + Script = + <<"function compute(process, message, opts)\n" + " process.count = (process.count or 0) + 1\n" + " process.results = { output = { body = process.count }, outbox = {} }\n" + " return process\n" + "end\n">>, + Process = + hb_message:commit( + #{ + <<"device">> => <<"process@1.0">>, + <<"type">> => <<"Process">>, + <<"scheduler-device">> => <<"scheduler@1.0">>, + <<"execution-device">> => <<"lua@5.3a">>, + <<"content-type">> => <<"application/lua">>, + <<"data">> => Script, + <<"name">> => <<"inline-lua-data-test">>, + <<"authority">> => [Address], + <<"scheduler-location">> => Address, + <<"test-random-seed">> => rand:uniform(1337) + }, + Opts + ), + {ok, _} = hb_cache:write(Process, Opts), + Message = generate_test_message(Process, Opts), + {ok, _} = hb_ao:resolve(Process, Message, Opts#{ <<"hashpath">> => ignore }), + {ok, Results} = hb_ao:resolve(Process, <<"now">>, Opts), + ?assertEqual(1, hb_ao:get(<<"results/output/body">>, Results, #{})). + +%% @doc Sequential pushes must advance the scheduler rather than replacing the +%% previous assignment at the same slot. This mirrors the request shape used by +%% aoconnect for `POST /~process@1.0/push'. +inline_lua_sequential_push_test() -> + Opts = #{ + <<"priv-wallet">> => hb:wallet(), + <<"cache-control">> => [<<"always">>] + }, + Wallet = hb_opts:get(priv_wallet, hb:wallet(), Opts), + Address = hb_util:human_id(ar_wallet:to_address(Wallet)), + Process = + hb_message:commit( + #{ + <<"device">> => <<"process@1.0">>, + <<"type">> => <<"Process">>, + <<"scheduler-device">> => <<"scheduler@1.0">>, + <<"push-device">> => <<"push@1.0">>, + <<"execution-device">> => <<"lua@5.3a">>, + <<"content-type">> => <<"application/lua">>, + <<"data">> => + <<"function compute(process, message, opts)\n" + " process.results = { output = { body = message.data }, outbox = {} }\n" + " return process\n" + "end\n">>, + <<"authority">> => [Address], + <<"scheduler-location">> => Address, + <<"test-random-seed">> => rand:uniform(1337) + }, + Opts + ), + ProcID = hb_message:id(Process, all, Opts), + {ok, _} = hb_cache:write(Process, Opts), + {ok, SpawnAssignment} = + hb_ao:resolve( + Process, + #{ + <<"method">> => <<"POST">>, + <<"path">> => <<"schedule">>, + <<"body">> => Process + }, + Opts + ), + ?assertEqual(0, hb_ao:get(<<"slot">>, SpawnAssignment, Opts)), + lists:foreach( + fun({ExpectedSlot, Data}) -> + Message = + hb_message:commit( + #{ + <<"target">> => ProcID, + <<"type">> => <<"Message">>, + <<"data">> => Data + }, + Opts + ), + PushReq = Message#{ <<"path">> => <<"push">> }, + {ok, PushResult} = hb_ao:resolve(Process, PushReq, Opts), + ?assertEqual(ExpectedSlot, hb_ao:get(<<"slot">>, PushResult, Opts)) + end, + [{1, <<"one">>}, {2, <<"two">>}] + ). + %% @doc Call a process whose `execution-device' is set to `lua@5.3a'. pure_lua_restore_test() -> Opts = #{ <<"process-cache-frequency">> => 1 }, From 27dc46168ce1121fecec97822375d4cdca6fc0c0 Mon Sep 17 00:00:00 2001 From: Nick Juliano Date: Fri, 17 Jul 2026 09:46:25 -0400 Subject: [PATCH 2/3] fix: make scheduler upload rollback confirmation-aware --- src/core/store/hb_store_rocksdb.erl | 20 +-- src/preloaded/process/dev_scheduler_cache.erl | 99 ++++++++--- .../process/dev_scheduler_server.erl | 67 ++------ src/preloaded/vm/dev_lua.erl | 154 ++++-------------- 4 files changed, 137 insertions(+), 203 deletions(-) diff --git a/src/core/store/hb_store_rocksdb.erl b/src/core/store/hb_store_rocksdb.erl index 43103d39bc..a96cd7173b 100644 --- a/src/core/store/hb_store_rocksdb.erl +++ b/src/core/store/hb_store_rocksdb.erl @@ -224,14 +224,9 @@ link_path(_, Key1, Key1) -> link_path(Opts, Existing, New) -> ExistingBin = convert_if_list(Existing), NewBin = convert_if_list(New), - - % Create: NewValue -> ExistingBin - case do_read(Opts, NewBin) of - not_found -> - do_write(Opts, NewBin, encode_value(link, ExistingBin)); - _ -> - ok - end. + % Create or replace: NewValue -> ExistingBin. Replacement matches the + % semantics of the FS, LMDB, and volatile stores. + do_write(Opts, NewBin, encode_value(link, ExistingBin)). %% @doc List all items registered in rocksdb store. Should be used only %% for testing/debugging, as the underlying operation is doing full traversal @@ -457,8 +452,13 @@ write_read_test_() -> ), ok = link(#{}, #{ <<"test_key">> => <<"test_key2">> }, #{}), {ok, Value} = read(#{}, #{ <<"read">> => <<"test_key">> }, #{}), - - ?assertEqual(<<"value_under_linked_key">>, Value) + ?assertEqual(<<"value_under_linked_key">>, Value), + ok = write(#{}, #{ <<"test_key3">> => <<"replacement">> }, #{}), + ok = link(#{}, #{ <<"test_key">> => <<"test_key3">> }, #{}), + ?assertEqual( + {ok, <<"replacement">>}, + read(#{}, #{ <<"read">> => <<"test_key">> }, #{}) + ) end} ]}. diff --git a/src/preloaded/process/dev_scheduler_cache.erl b/src/preloaded/process/dev_scheduler_cache.erl index 4ec202c771..5a335f3621 100644 --- a/src/preloaded/process/dev_scheduler_cache.erl +++ b/src/preloaded/process/dev_scheduler_cache.erl @@ -1,6 +1,6 @@ %%% @doc A module that provides a cache for scheduler assignments and locations. -module(dev_scheduler_cache). --export([write/2, write_spawn/2, read/3]). +-export([write/2, unlink/2, write_spawn/2, read/3]). -export([list/2, latest/2]). -include("include/hb.hrl"). -include_lib("eunit/include/eunit.hrl"). @@ -41,14 +41,8 @@ write(RawAssignment, RawOpts) -> % slot on the process to the underlying data. ok = hb_store:link( Store, - #{ - hb_path:to_binary([ - ?SCHEDULER_CACHE_PREFIX, - <<"assignments">>, - hb_util:human_id(ProcID), - hb_ao:normalize_key(Slot) - ]) => hb_message:id(Assignment, signed, Opts) - }, + #{ assignment_path(ProcID, Slot) => + hb_message:id(Assignment, signed, Opts) }, Opts ), ok; @@ -57,6 +51,34 @@ write(RawAssignment, RawOpts) -> {error, Reason} end. +%% @doc Remove a failed assignment from the usable schedule while retaining +%% its content-addressed data. Stores do not expose a destructive unlink API, +%% so replace the slot link with a deliberately dangling rollback link. +unlink(RawAssignment, RawOpts) -> + Assignment = hb_cache:ensure_all_loaded(RawAssignment, RawOpts), + Opts = opts(RawOpts), + Store = hb_opts:get(store, no_viable_store, Opts), + ProcID = hb_ao:get(<<"process">>, Assignment, Opts), + Slot = hb_ao:get(<<"slot">>, Assignment, Opts), + AssignmentID = hb_message:id(Assignment, signed, Opts), + ?event( + {unlinking_assignment, + {proc_id, ProcID}, + {slot, Slot}, + {assignment, AssignmentID} + } + ), + hb_store:link( + Store, + #{ assignment_path(ProcID, Slot) => + hb_path:to_binary([ + ?SCHEDULER_CACHE_PREFIX, + <<"rolled-back">>, + AssignmentID + ]) }, + Opts + ). + %% @doc Write the initial assignment message to the cache. write_spawn(RawInitMessage, Opts) -> InitMessage = hb_cache:ensure_all_loaded(RawInitMessage, Opts), @@ -68,12 +90,7 @@ read(ProcID, Slot, Opts) when is_integer(Slot) -> read(ProcID, Slot, RawOpts) -> Opts = opts(RawOpts), Store = hb_opts:get(store, no_viable_store, Opts), - P1 = hb_path:to_binary([ - ?SCHEDULER_CACHE_PREFIX, - <<"assignments">>, - hb_util:human_id(ProcID), - Slot - ]), + P1 = assignment_path(ProcID, Slot), ?event( {read_assignment, {proc_id, ProcID}, @@ -114,14 +131,36 @@ read(ProcID, Slot, RawOpts) -> %% @doc Get the assignments for a process. list(ProcID, RawOpts) -> Opts = opts(RawOpts), - hb_cache:list_numbered( - hb_path:to_binary([ - ?SCHEDULER_CACHE_PREFIX, - <<"assignments">>, - hb_util:human_id(ProcID) - ]), - Opts - ). + Slots = + hb_cache:list_numbered( + hb_path:to_binary([ + ?SCHEDULER_CACHE_PREFIX, + <<"assignments">>, + hb_util:human_id(ProcID) + ]), + Opts + ), + trim_unavailable_tail(ProcID, Slots, Opts). + +%% Rolled-back assignments can only occur at the unconfirmed tail because the +%% scheduler does not advance its state in remote-confirmation mode. Usually +%% this checks one slot; recursion also tolerates multiple interrupted retries. +trim_unavailable_tail(_ProcID, [], _Opts) -> []; +trim_unavailable_tail(ProcID, Slots, Opts) -> + Latest = lists:max(Slots), + case read(ProcID, Latest, Opts) of + {ok, _} -> Slots; + not_found -> + trim_unavailable_tail(ProcID, lists:delete(Latest, Slots), Opts) + end. + +assignment_path(ProcID, Slot) -> + hb_path:to_binary([ + ?SCHEDULER_CACHE_PREFIX, + <<"assignments">>, + hb_util:human_id(ProcID), + hb_ao:normalize_key(Slot) + ]). %% @doc Get the latest assignment from the cache. latest(ProcID, RawOpts) -> @@ -178,6 +217,20 @@ volatile_schedule_test() -> ?assertMatch({1, _}, latest(ProcID, Opts)), {ok, ReadAssignment} = read(ProcID, 1, Opts), ?assertEqual(ReadAssignment, hb_message:normalize_commitments(Assignment, Opts)), + ?assertEqual(ok, unlink(Assignment, Opts)), + ?assertEqual([], list(ProcID, Opts)), + ?assertEqual(not_found, read(ProcID, 1, Opts)), + ?assertEqual(ok, write(Assignment, Opts)), + ?assertMatch({ok, _}, read(ProcID, 1, Opts)), + LMDBStore = hb_test_utils:test_store(hb_store_lmdb, <<"scheduler-unlink">>), + LMDBOpts = #{ <<"store">> => [LMDBStore] }, + hb_store:start(LMDBStore), + ?assertEqual(ok, write(Assignment, LMDBOpts)), + ?assertEqual(ok, unlink(Assignment, LMDBOpts)), + ?assertEqual([], list(ProcID, LMDBOpts)), + ?assertEqual(ok, write(Assignment, LMDBOpts)), + ?assertMatch({ok, _}, read(ProcID, 1, LMDBOpts)), + hb_store:reset(LMDBStore), hb_store:stop(VolStore), hb_store:reset(VolStore), hb_store:start(VolStore), diff --git a/src/preloaded/process/dev_scheduler_server.erl b/src/preloaded/process/dev_scheduler_server.erl index 559e68bc69..05262fb2e3 100644 --- a/src/preloaded/process/dev_scheduler_server.erl +++ b/src/preloaded/process/dev_scheduler_server.erl @@ -278,21 +278,16 @@ do_assign(State, Message, ReplyPID) -> base_state_hashpath := next_hashpath(BaseStateHashpath, Assignment, State) }. -%% @doc Upload a durably written assignment without allowing a remote upload -%% failure to roll back the scheduler's in-memory slot. In local-confirmation -%% mode the caller has already received the assignment at this point; reusing -%% the old state would let the next message overwrite that accepted slot. +%% @doc Upload a durably written assignment. Local and aggressive confirmation +%% have already replied by this point, so an upload failure must not roll their +%% scheduler state back. Remote confirmation has not replied, so remove the +%% local schedule link and re-raise to keep the previous scheduler state. upload_after_write(Message, Assignment, ReplyPID, State) -> Opts = maps:get(opts, State), - Upload = hb_opts:get( - scheduler_upload_fun, - fun hb_client_remote:upload/2, - Opts - ), try ?event(uploading_message), - Upload(Message, Opts), - Upload(Assignment, Opts), + hb_client_remote:upload(Message, Opts), + hb_client_remote:upload(Assignment, Opts), ?event(uploads_complete), maybe_inform_recipient( remote_confirmation, @@ -312,7 +307,15 @@ upload_after_write(Message, Assignment, ReplyPID, State) -> {slot, hb_ao:get(<<"slot">>, Assignment, Opts)} } ), - ok + case maps:get(mode, State) of + aggressive -> ok; + local_confirmation -> ok; + remote_confirmation -> + ok = dev_scheduler_cache:unlink(Assignment, Opts), + erlang:raise(Class, Reason, Stack); + _ -> + erlang:raise(Class, Reason, Stack) + end end. %% @doc Commit to the assignment using all of our appropriate wallets. @@ -396,46 +399,6 @@ new_proc_test() -> dev_scheduler_server:info(dev_scheduler_registry:find(ID)) ). -%% @doc Once local confirmation follows a durable cache write, a subsequent -%% upload failure must not reset the scheduler to its previous slot. -local_confirmation_upload_failure_preserves_slot_test() -> - Wallet = hb:wallet(), - Address = hb_util:human_id(ar_wallet:to_address(Wallet)), - Opts = #{ - <<"priv-wallet">> => Wallet, - <<"store">> => [hb_test_utils:test_store()], - <<"scheduling-mode">> => local_confirmation, - <<"scheduler-upload-fun">> => - fun(_, _) -> error(simulated_upload_failure) end - }, - Process = hb_message:commit( - #{ - <<"type">> => <<"Process">>, - <<"scheduler">> => Address, - <<"test-random-seed">> => rand:uniform(1000000) - }, - Opts - ), - ProcID = hb_message:id(Process, all, Opts), - PID = start(ProcID, Process, Opts), - First = schedule(PID, Process), - ?assertEqual(0, hb_ao:get(<<"slot">>, First, Opts)), - ?assertMatch(#{ current := 0 }, info(PID)), - Message = hb_message:commit( - #{ - <<"type">> => <<"Message">>, - <<"target">> => ProcID, - <<"data">> => <<"next">> - }, - Opts - ), - Second = schedule(PID, Message), - ?assertEqual(1, hb_ao:get(<<"slot">>, Second, Opts)), - ?assertMatch(#{ current := 1 }, info(PID)), - stop(PID), - hb_name:unregister({<<"scheduler@1.0">>, ProcID}). - - benchmark_test() -> BenchTime = 1, Wallet = ar_wallet:new(), diff --git a/src/preloaded/vm/dev_lua.erl b/src/preloaded/vm/dev_lua.erl index df88f2e303..2a18ce0c8a 100644 --- a/src/preloaded/vm/dev_lua.erl +++ b/src/preloaded/vm/dev_lua.erl @@ -146,12 +146,10 @@ load_modules([Module|Rest], Opts, Acc) when is_map(Module) -> % We have found a message with a Lua module inside. Search for the binary % of the program in the body and the data. ModuleBin = - hb_ao:get_first( - [ - {{as, <<"message@1.0">>, Module}, <<"body">>}, - {{as, <<"message@1.0">>, Module}, <<"data">>} - ], + hb_maps:get( + <<"body">>, Module, + hb_maps:get(<<"data">>, Module, not_found, Opts), Opts ), case ModuleBin of @@ -902,109 +900,13 @@ lua_http_hook_test() -> %% @doc Call a process whose `execution-device' is set to `lua@5.3a'. pure_lua_process_test() -> - Process = generate_lua_process("test/test.lua", #{}), + Process = generate_lua_process("test/test.lua", inline, #{}), {ok, _} = hb_cache:write(Process, #{}), Message = generate_test_message(Process, #{}), {ok, _} = hb_ao:resolve(Process, Message, #{ <<"hashpath">> => ignore }), {ok, Results} = hb_ao:resolve(Process, <<"now">>, #{}), ?assertEqual(42, hb_ao:get(<<"results/output/body">>, Results, #{})). -%% @doc An inline ANS-104 process stores its Lua source under `data'. Ensure -%% module discovery reads that field literally instead of invoking Lua's -%% default handler for the absent `body' field during initialization. -inline_lua_data_process_test() -> - Opts = #{ <<"priv-wallet">> => hb:wallet() }, - Wallet = hb_opts:get(priv_wallet, hb:wallet(), Opts), - Address = hb_util:human_id(ar_wallet:to_address(Wallet)), - Script = - <<"function compute(process, message, opts)\n" - " process.count = (process.count or 0) + 1\n" - " process.results = { output = { body = process.count }, outbox = {} }\n" - " return process\n" - "end\n">>, - Process = - hb_message:commit( - #{ - <<"device">> => <<"process@1.0">>, - <<"type">> => <<"Process">>, - <<"scheduler-device">> => <<"scheduler@1.0">>, - <<"execution-device">> => <<"lua@5.3a">>, - <<"content-type">> => <<"application/lua">>, - <<"data">> => Script, - <<"name">> => <<"inline-lua-data-test">>, - <<"authority">> => [Address], - <<"scheduler-location">> => Address, - <<"test-random-seed">> => rand:uniform(1337) - }, - Opts - ), - {ok, _} = hb_cache:write(Process, Opts), - Message = generate_test_message(Process, Opts), - {ok, _} = hb_ao:resolve(Process, Message, Opts#{ <<"hashpath">> => ignore }), - {ok, Results} = hb_ao:resolve(Process, <<"now">>, Opts), - ?assertEqual(1, hb_ao:get(<<"results/output/body">>, Results, #{})). - -%% @doc Sequential pushes must advance the scheduler rather than replacing the -%% previous assignment at the same slot. This mirrors the request shape used by -%% aoconnect for `POST /~process@1.0/push'. -inline_lua_sequential_push_test() -> - Opts = #{ - <<"priv-wallet">> => hb:wallet(), - <<"cache-control">> => [<<"always">>] - }, - Wallet = hb_opts:get(priv_wallet, hb:wallet(), Opts), - Address = hb_util:human_id(ar_wallet:to_address(Wallet)), - Process = - hb_message:commit( - #{ - <<"device">> => <<"process@1.0">>, - <<"type">> => <<"Process">>, - <<"scheduler-device">> => <<"scheduler@1.0">>, - <<"push-device">> => <<"push@1.0">>, - <<"execution-device">> => <<"lua@5.3a">>, - <<"content-type">> => <<"application/lua">>, - <<"data">> => - <<"function compute(process, message, opts)\n" - " process.results = { output = { body = message.data }, outbox = {} }\n" - " return process\n" - "end\n">>, - <<"authority">> => [Address], - <<"scheduler-location">> => Address, - <<"test-random-seed">> => rand:uniform(1337) - }, - Opts - ), - ProcID = hb_message:id(Process, all, Opts), - {ok, _} = hb_cache:write(Process, Opts), - {ok, SpawnAssignment} = - hb_ao:resolve( - Process, - #{ - <<"method">> => <<"POST">>, - <<"path">> => <<"schedule">>, - <<"body">> => Process - }, - Opts - ), - ?assertEqual(0, hb_ao:get(<<"slot">>, SpawnAssignment, Opts)), - lists:foreach( - fun({ExpectedSlot, Data}) -> - Message = - hb_message:commit( - #{ - <<"target">> => ProcID, - <<"type">> => <<"Message">>, - <<"data">> => Data - }, - Opts - ), - PushReq = Message#{ <<"path">> => <<"push">> }, - {ok, PushResult} = hb_ao:resolve(Process, PushReq, Opts), - ?assertEqual(ExpectedSlot, hb_ao:get(<<"slot">>, PushResult, Opts)) - end, - [{1, <<"one">>}, {2, <<"two">>}] - ). - %% @doc Call a process whose `execution-device' is set to `lua@5.3a'. pure_lua_restore_test() -> Opts = #{ <<"process-cache-frequency">> => 1 }, @@ -1135,29 +1037,45 @@ aos_process_benchmark_test_() -> %% @doc Generate a Lua process message. generate_lua_process(File, Opts) -> + generate_lua_process(File, module, Opts). +generate_lua_process(File, Source, Opts) -> NormOpts = Opts#{ <<"priv-wallet">> => hb_opts:get(priv_wallet, hb:wallet(), Opts) }, Wallet = hb_opts:get(priv_wallet, hb:wallet(), NormOpts), Address = hb_util:human_id(ar_wallet:to_address(Wallet)), {ok, Module} = file:read_file(File), + SourceFields = + case Source of + inline -> + #{ + <<"content-type">> => <<"application/lua">>, + <<"data">> => Module + }; + module -> + #{ + <<"module">> => #{ + <<"content-type">> => <<"application/lua">>, + <<"body">> => Module + } + } + end, hb_message:commit( - #{ - <<"device">> => <<"process@1.0">>, - <<"type">> => <<"Process">>, - <<"scheduler-device">> => <<"scheduler@1.0">>, - <<"execution-device">> => <<"lua@5.3a">>, - <<"module">> => #{ - <<"content-type">> => <<"application/lua">>, - <<"body">> => Module + maps:merge( + #{ + <<"device">> => <<"process@1.0">>, + <<"type">> => <<"Process">>, + <<"scheduler-device">> => <<"scheduler@1.0">>, + <<"execution-device">> => <<"lua@5.3a">>, + <<"authority">> => [ + Address, + <<"E3FJ53E6xtAzcftBpaw2E1H4ZM9h6qy6xz9NXh5lhEQ">> + ], + <<"scheduler-location">> => + hb_util:human_id(ar_wallet:to_address(Wallet)), + <<"test-random-seed">> => rand:uniform(1337) }, - <<"authority">> => [ - Address, - <<"E3FJ53E6xtAzcftBpaw2E1H4ZM9h6qy6xz9NXh5lhEQ">> - ], - <<"scheduler-location">> => - hb_util:human_id(ar_wallet:to_address(Wallet)), - <<"test-random-seed">> => rand:uniform(1337) - }, + SourceFields + ), NormOpts ). From 0cb64c443d76c93b34a93e2dcc7677d28b93f8c8 Mon Sep 17 00:00:00 2001 From: Nick Juliano Date: Mon, 20 Jul 2026 09:34:03 -0400 Subject: [PATCH 3/3] chore: add upload-failure test to verify remote confirmation assignments are unlinked --- src/preloaded/process/dev_scheduler_cache.erl | 2 + .../process/dev_scheduler_server.erl | 59 ++++++++++++++++++- 2 files changed, 59 insertions(+), 2 deletions(-) diff --git a/src/preloaded/process/dev_scheduler_cache.erl b/src/preloaded/process/dev_scheduler_cache.erl index 5a335f3621..b1114fdcbb 100644 --- a/src/preloaded/process/dev_scheduler_cache.erl +++ b/src/preloaded/process/dev_scheduler_cache.erl @@ -117,6 +117,8 @@ read(ProcID, Slot, RawOpts) -> ?event({normalized_aos2_assignment, Norm}), {ok, Norm}; <<"ao.N.1">> -> + {ok, hb_cache:ensure_all_loaded(Assignment, Opts)}; + not_found -> {ok, hb_cache:ensure_all_loaded(Assignment, Opts)} end; {error, not_found} -> diff --git a/src/preloaded/process/dev_scheduler_server.erl b/src/preloaded/process/dev_scheduler_server.erl index 05262fb2e3..d471de46ff 100644 --- a/src/preloaded/process/dev_scheduler_server.erl +++ b/src/preloaded/process/dev_scheduler_server.erl @@ -286,8 +286,8 @@ upload_after_write(Message, Assignment, ReplyPID, State) -> Opts = maps:get(opts, State), try ?event(uploading_message), - hb_client_remote:upload(Message, Opts), - hb_client_remote:upload(Assignment, Opts), + ok = ensure_upload_succeeded(hb_client_remote:upload(Message, Opts)), + ok = ensure_upload_succeeded(hb_client_remote:upload(Assignment, Opts)), ?event(uploads_complete), maybe_inform_recipient( remote_confirmation, @@ -318,6 +318,24 @@ upload_after_write(Message, Assignment, ReplyPID, State) -> end end. +%% @doc Turn failed upload return values into exceptions so the confirmation- +%% mode-specific recovery in `upload_after_write/4' is always applied. +ensure_upload_succeeded({ok, Results}) when is_list(Results) -> + case [Result || Result <- Results, not upload_result_succeeded(Result)] of + [] -> ok; + Failed -> erlang:error({upload_failed, Failed}) + end; +ensure_upload_succeeded({error, Reason}) -> + erlang:error({upload_failed, Reason}); +ensure_upload_succeeded({failure, Reason}) -> + erlang:error({upload_failed, Reason}); +ensure_upload_succeeded(Result) -> + erlang:error({upload_failed, Result}). + +upload_result_succeeded(ok) -> true; +upload_result_succeeded({ok, _}) -> true; +upload_result_succeeded(_) -> false. + %% @doc Commit to the assignment using all of our appropriate wallets. commit_assignment(BaseAssignment, State) -> Wallets = maps:get(wallets, State), @@ -399,6 +417,43 @@ new_proc_test() -> dev_scheduler_server:info(dev_scheduler_registry:find(ID)) ). +%% @doc A returned upload error must invalidate an assignment that has not yet +%% received remote confirmation. +remote_confirmation_upload_failure_unlinks_test() -> + Wallet = hb:wallet(), + Store = hb_test_utils:test_store(hb_store_fs, <<"remote-upload-failure">>), + Opts = #{ + <<"priv-wallet">> => Wallet, + <<"store">> => [Store] + }, + hb_store:start(Store), + ProcID = hb_util:human_id(crypto:strong_rand_bytes(32)), + Message = hb_message:commit( + #{ <<"target">> => ProcID, <<"data">> => <<"test">> }, + Opts, + <<"httpsig@1.0">> + ), + Assignment = hb_message:commit( + #{ + <<"variant">> => <<"ao.N.1">>, + <<"process">> => ProcID, + <<"slot">> => 0, + <<"hash-chain">> => <<"test-hash-chain">> + }, + Opts, + <<"httpsig@1.0">> + ), + ok = dev_scheduler_cache:write(Assignment, Opts), + ?assertEqual([0], dev_scheduler_cache:list(ProcID, Opts)), + State = #{ id => ProcID, mode => remote_confirmation, opts => Opts }, + ?assertError( + {upload_failed, [{error, no_httpsig_bundler}]}, + upload_after_write(Message, Assignment, self(), State) + ), + ?assertEqual([], dev_scheduler_cache:list(ProcID, Opts)), + ?assertEqual(not_found, dev_scheduler_cache:read(ProcID, 0, Opts)), + hb_store:reset(Store). + benchmark_test() -> BenchTime = 1, Wallet = ar_wallet:new(),