Merge pull request #602 from permaweb/fix/parallel_requests

Fix parallel requests

Author: jfrain99

src/hb_http.erl

@@ -1254,3 +1254,58 @@ index_request_test() ->
             #{}
         ),
     ?assertEqual(<<"i like dogs!">>, hb_ao:get(<<"body">>, Res, #{})).
+
+%% Test parallel requests
+parallel_request_test() ->
+    Routes = [
+        #{
+            % Routes for GraphQL requests to use a remote GraphQL API.
+            <<"template">> => <<"/graphql">>,
+            <<"parallel">> => true,
+            <<"nodes">> =>
+                [
+                    #{
+                        <<"prefix">> => <<"https://ao-search-gateway.goldsky.com">>,
+                        <<"opts">> => #{ http_client => httpc, protocol => http2 }
+                    },
+                    #{
+                        <<"prefix">> => <<"https://arweave-search.goldsky.com">>,
+                        <<"opts">> => #{ http_client => httpc, protocol => http2 }
+                    },
+                    #{
+                        <<"prefix">> => <<"https://arweave.net">>,
+                        <<"opts">> => #{ http_client => gun, protocol => http2 }
+                    }
+                ]
+            },
+            #{
+                % Routes for raw data requests to use a remote gateway.
+                <<"template">> => <<"/raw">>,
+                <<"node">> =>
+                    #{
+                        <<"prefix">> => <<"https://arweave.net">>,
+                        <<"opts">> => #{ http_client => gun, protocol => http2 }
+                    }
+            }
+    ],
+    Store = [
+        hb_test_utils:test_store(),
+        #{
+          <<"store-module">> => hb_store_gateway,
+          %% Routes need to be defined in the store, otherwise the code
+          %% will fetch the hb_opts:default_message which doesn't have
+          %% parallel property.
+          <<"routes">> => Routes
+         }
+    ],
+    hb_store:reset(Store),
+    Opts = #{ store => Store },
+    Node = hb_http_server:start_node(Opts),
+    ?assertMatch(
+        {ok, #{<<"data">> := <<"1984">>}},
+        hb_http:get(
+            Node,
+            #{<<"path">> => <<"/BOogk_XAI3bvNWnxNxwxmvOfglZt17o4MOVAdPNZ_ew">>},
+            Opts
+        )
+    ).

src/hb_http_multi.erl

@@ -230,21 +230,34 @@ admissible_response(Response, Msg, Opts) ->
     ?event(debug_multi,
         {executing_admissible_message, {message, Base}, {req, Req}}
     ),
-    case hb_ao:resolve(Base, Req, Opts) of
+    try hb_ao:resolve(Base, Req, Opts) of
         {ok, Res} when is_atom(Res) or is_binary(Res) ->
             ?event(debug_multi, {admissible_result, {result, Res}}),
             hb_util:atom(Res) == true;
         {error, Reason} ->
             ?event(debug_multi, {admissible_error, {reason, Reason}}),
             false
+    catch 
+        Class:Reason:Stacktrace ->
+            ?event(error, 
+                {admissible_response, 
+                    {class, Class}, 
+                    {reason, Reason}, 
+                    {stacktrace, Stacktrace}
+                }
+            ),
+            false
     end.
 
 %% @doc Collect the necessary number of responses, and stop workers if
 %% configured to do so.
+parallel_responses(Res,  [], Ref, _Awaiting, _StopAfter, _Admissible, _Statuses, _Opts) ->
+    empty_inbox(Ref),
+    Res;
 parallel_responses(Res, Procs, Ref, 0, false, _Admissible, _Statuses, _Opts) ->
     lists:foreach(fun(P) -> P ! no_reply end, Procs),
     empty_inbox(Ref),
-    {ok, Res};
+    Res;
 parallel_responses(Res, Procs, Ref, 0, true, _Admissible, _Statuses, _Opts) ->
     lists:foreach(fun(P) -> exit(P, kill) end, Procs),
     empty_inbox(Ref),
@@ -255,7 +268,7 @@ parallel_responses(Res, Procs, Ref, Awaiting, StopAfter, Admissible, Statuses, O
             case is_admissible(Status, NewRes, Admissible, Statuses, Opts) of
                 true ->
                     parallel_responses(
-                        [NewRes | Res],
+                        [{Status, NewRes} | Res],
                         lists:delete(Pid, Procs),
                         Ref,
                         Awaiting - 1,