Copycat L3 indexer with parallel processing and parent lookups

src/dev_arweave.erl

@@ -5,7 +5,7 @@
 %%% `/arweave` route in the node's configuration message.
 -module(dev_arweave).
 -export([info/0]).
--export([tx/3, raw/3, chunk/3, block/3, current/3, status/3, price/3, tx_anchor/3]).
+-export([tx/3, raw/3, chunk/3, block/3, parent/3, current/3, status/3, price/3, tx_anchor/3]).
 -export([pending/3]).
 -export([post_tx_header/2, post_tx/3, post_tx/4, post_chunk/2]).
 %%% Helper functions
@@ -173,7 +173,16 @@ head_raw(Base, Request, Opts) ->
                                 <<"tx@1.0">> -> fun head_raw_tx/4;
                                 _ -> throw({invalid_codec_device, CodecDevice})
                             end,
-                        CodecFun(TXID, StartOffset, Length, Opts);
+                        try CodecFun(TXID, StartOffset, Length, Opts)
+                        catch _:Reason:Stacktrace ->
+                            %% This can be prone to serialization error. 
+                            %% Catch and output as an error.
+                            ?event(store_error, {head_raw, 
+                                {txid, TXID},
+                                {reason, Reason},
+                                {stacktrace, Stacktrace}}),
+                            {error, Reason}
+                        end;
                 not_found ->
                     ?event(
                         arweave,
@@ -793,6 +802,57 @@ block({height, Height}, Opts) ->
             )
     end.
 
+%% @doc Look up the parent (block or bundle) that contains an item.
+parent(Base, Request, Opts) ->
+    case find_key(<<"parent">>, Base, Request, Opts) of
+        not_found ->
+            {error, not_found};
+        ID ->
+            StoreOpts = hb_store_arweave:store_from_opts(Opts),
+            try hb_store_arweave:read_parent(StoreOpts, ID, Opts) of
+                {ok, [{Height, block} | _]} ->
+                    Entry = #{
+                        <<"type">> => <<"block">>,
+                        <<"height">> => Height
+                    },
+                    {ok, #{
+                        <<"content-type">> => <<"application/json">>,
+                        <<"body">> =>
+                            hb_json:encode(#{<<"parents">> => [Entry]})
+                    }};
+                {ok, [{ParentID, bundle} | _]} ->
+                    Entry = #{
+                        <<"type">> => <<"bundle">>,
+                        <<"id">> => hb_util:encode(ParentID)
+                    },
+                    {ok, #{
+                        <<"content-type">> => <<"application/json">>,
+                        <<"body">> =>
+                            hb_json:encode(#{<<"parents">> => [Entry]})
+                    }};
+                {error, Reason} ->
+                    ?event(warning,
+                        {parent_read_error, {id, ID}, {reason, Reason}}),
+                    {error, not_found};
+                not_found ->
+                    {error, not_found}
+            catch
+                error:Reason:Stacktrace ->
+                    ?event(error, 
+                        {parent_read_error, 
+                            {id, ID},
+                            {reason, Reason},
+                            {stacktrace, Stacktrace}
+                        }),
+                    {failure, 
+                        #{
+                            <<"status">> => 500, 
+                            <<"type">> => <<"parent_read_error">>
+                         }
+                    }
+            end
+    end.
+
 %% @doc Retrieve the current block information from Arweave.
 current(_Base, _Request, Opts) ->
     request(<<"GET">>, <<"/block/current">>, Opts).
@@ -953,7 +1013,7 @@ to_message(Path = <<"/block/", _/binary>>, <<"GET">>, {ok, #{ <<"body">> := Body
             Opts
         ),
     CacheRes =
-        case hb_opts:get(arweave_index_blocks, true, Opts) of
+        case hb_opts:get(<<"arweave-index-blocks">>, true, Opts) of
             true -> dev_arweave_block_cache:write(Block, Opts);
             false -> skipped
         end,
@@ -1357,7 +1417,8 @@ index_test_tx(TXID, IndexStore, Opts) ->
             TXID,
             <<"tx@1.0">>,
             StartOffset,
-            Size
+            Size,
+            Opts
         ),
     ?assertMatch({ok, _}, hb_store_arweave:read_offset(IndexStore, TXID)),
     ok.
@@ -1551,6 +1612,40 @@ head_raw_ans104_test_parallel() ->
         hb_maps:find(<<"content-length">>, Result, Opts)
     ).
 
+%% @doc Interior Arweave offset returns bytes that are not a valid ANS-104
+%% header, so head_raw_ans104/4 throws inside do_head_raw_ans104/5. The
+%% try-catch added to head_raw/3 must convert that throw into {error, _}.
+head_raw_ans104_deserialize_throws_test_parallel() ->
+    TestStore = hb_test_utils:test_store(hb_store_volatile, <<"head-raw-throws">>),
+    IndexStore = #{
+        <<"module">> => hb_store_arweave,
+        <<"index-store">> => [TestStore]
+    },
+    Opts = #{
+        <<"store">> => [TestStore],
+        <<"arweave-index-ids">> => true,
+        <<"arweave-index-store">> => IndexStore
+    },
+    FakeID = <<"CCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCC">>,
+    %% Same interior offset as bundle_header_garbage_guard_test_parallel.
+    ProbeOffset = 376836336327208,
+    Size = 4096,
+    ok = hb_store_arweave:write_offset(
+        IndexStore, FakeID, <<"ans104@1.0">>, ProbeOffset - 1, Size, Opts
+    ),
+    ?assertMatch(
+        {error, _},
+        hb_ao:resolve(
+            #{ <<"device">> => <<"arweave@2.9">> },
+            #{
+                <<"path">> => <<"raw">>,
+                <<"raw">> => FakeID,
+                <<"method">> => <<"HEAD">>
+            },
+            Opts
+        )
+    ).
+
 get_raw_range_tx_test_parallel() ->
     DataItemID = <<"ptBC0UwDmrUTBQX3MqZ1lB57ex20ygwzkjjCrQjIx3o">>,
     Opts = setup_arweave_index_opts([DataItemID]),