Return 410 code for queries beyond GC point (#4119)

* Return 410 code for queries beyond GC point

* Ensure GC env initialized in EUnit tests

* Hide test instrumentation from dialyzer

* Add release note

apps/aecore/src/aec_chain.erl

@@ -115,9 +115,14 @@ get_account_at_hash(PubKey, Hash) ->
     end.
 
 get_account_at_height(PubKey, Height) ->
-    case aec_chain_state:get_key_block_hash_at_height(Height) of
-        error -> {error, chain_too_short};
-        {ok, Hash} -> get_account_at_hash(PubKey, Hash)
+    case aec_db_gc:state_at_height_still_reachable(Height) of
+        true ->
+            case aec_chain_state:get_key_block_hash_at_height(Height) of
+                error -> {error, chain_too_short};
+                {ok, Hash} -> get_account_at_hash(PubKey, Hash)
+            end;
+        false ->
+            {error, garbage_collected}
     end.
 
 -spec all_accounts_balances_at_hash(binary()) ->
@@ -315,16 +320,27 @@ get_contract_with_code(PubKey) ->
                                {'ok', aect_call:call()} |
                                {'error', atom()}.
 get_contract_call(ContractId, CallId, BlockHash) ->
-    case get_block_state_partial(BlockHash, [calls]) of
-        error -> {error, no_state_trees};
-        {ok, Trees} ->
-            CallTree = aec_trees:calls(Trees),
-            case aect_call_state_tree:lookup_call(ContractId, CallId, CallTree) of
-                none -> {error, call_not_found};
-                {value, Call} -> {ok, Call}
-            end
+    case state_reachable_by_blockhash(BlockHash) of
+        true ->
+            case get_block_state_partial(BlockHash, [calls]) of
+                error -> {error, no_state_trees};
+                {ok, Trees} ->
+                    CallTree = aec_trees:calls(Trees),
+                    case aect_call_state_tree:lookup_call(ContractId, CallId, CallTree) of
+                        none -> {error, call_not_found};
+                        {value, Call} -> {ok, Call}
+                    end
+            end;
+        false ->
+            {error, garbage_collected}
     end.
 
+state_reachable_by_blockhash(BlockHash) ->
+    %% For now, let's assume that at least the block header exists
+    {ok, Header} = get_header(BlockHash),
+    Height = aec_headers:height(Header),
+    aec_db_gc:state_at_height_still_reachable(Height).
+
 %%%===================================================================
 %%% Generalized Accounts
 %%%===================================================================
@@ -777,6 +793,7 @@ get_key_header_by_height(Height) when is_integer(Height), Height >= 0 ->
 
 -spec sum_tokens_at_height(aec_blocks:height()) ->
                                   {error, 'chain_too_short'}
+                                | {error, 'garbage_collected'}
                                 | {ok, #{ 'accounts'         => non_neg_integer()
                                         , 'contracts'        => non_neg_integer()
                                         , 'contract_oracles' => non_neg_integer()
@@ -791,7 +808,12 @@ get_key_header_by_height(Height) when is_integer(Height), Height >= 0 ->
 sum_tokens_at_height(Height) ->
     %% Wrap in transaction for speed.
     %% TODO: This could be done dirty
-    aec_db:ensure_transaction(fun() -> int_sum_tokens_at_height(Height) end).
+    case aec_db_gc:state_at_height_still_reachable(Height) of
+        true ->
+            aec_db:ensure_transaction(fun() -> int_sum_tokens_at_height(Height) end);
+        false ->
+            {error, garbage_collected}
+    end.
 
 int_sum_tokens_at_height(Height) ->
     case aec_chain_state:get_key_block_hash_at_height(Height) of

apps/aecore/src/aec_conductor.erl

@@ -1403,13 +1403,13 @@ handle_successfully_added_block(Block, Hash, true, PrevKeyHeader, Events, State,
             (BlockType == key) andalso
                 aec_metrics:try_update(
                   [ae,epoch,aecore,blocks,key,info], info_value(NewTopBlock)),
+            [ maybe_garbage_collect(NewTopBlock) || BlockType == key ],
             IsLeader = is_leader(NewTopBlock, PrevKeyHeader, ConsensusModule),
             case IsLeader of
                 true ->
                     ok; %% Don't spend time when we are the leader.
                 false ->
-                    aec_tx_pool:garbage_collect(),
-                    [ maybe_garbage_collect(NewTopBlock) || BlockType == key ]
+                    aec_tx_pool:garbage_collect()
             end,
             {ok, setup_loop(State2, true, IsLeader, Origin)}
     end.

apps/aecore/src/aec_db.erl

@@ -144,8 +144,10 @@
 
 -export([backend_mode/0]).
 
+-ifdef(TEST).
 -export([ install_test_env/0
         , uninstall_test_env/0 ]).
+-endif.
 
 -include("blocks.hrl").
 -include("aec_db.hrl").
@@ -1535,11 +1537,14 @@ initialize_db(ram) ->
 %% == Test setup env
 %% Ensures that e.g. persistent terms are present, logging which ones had to be added.
 
+-ifdef(TEST).
 install_test_env() ->
     ensure_backend_module(),
+    aec_db_gc:install_test_env(),
     ok.
 
 uninstall_test_env() ->
+    aec_db_gc:cleanup(),
     remove_added_pts().
 
 ensure_backend_module() ->
@@ -1555,7 +1560,6 @@ ensure_backend_module() ->
 get_test_backend_module() ->
     Str = os:getenv("AETERNITY_TESTCONFIG_DB_BACKEND", "mnesia"),
     list_to_existing_atom(Str).
-
 note_added_pt(Key) ->
     Var = ?PT_ADDED_PTS,
     Set = persistent_term:get(Var, ordsets:new()),
@@ -1565,6 +1569,7 @@ remove_added_pts() ->
     Keys = persistent_term:get(?PT_ADDED_PTS, ordsets:new()),
     [persistent_term:erase(K) || K <- Keys],
     persistent_term:erase(?PT_ADDED_PTS).
+-endif.
 
 %% == End Test setup env
 

apps/aecore/src/aec_db_gc.erl

@@ -26,9 +26,11 @@
 -behaviour(gen_server).
 
 %% API
--export([start_link/0]).
+-export([start_link/0,
+         cleanup/0]).
 
 -export([ height_of_last_gc/0
+        , state_at_height_still_reachable/1
         , info/0
         , info/1
         ]).
@@ -46,6 +48,10 @@
 
 -export([config/0]).
 
+-ifdef(TEST).
+-export([install_test_env/0]).
+-endif.
+
 -type tree_name() :: aec_trees:tree_name().
 
 -type pid_ref() :: {pid(), reference()}.
@@ -100,15 +106,37 @@ start_link() ->
     #{<<"enabled">> := _, <<"trees">> := _, <<"history">> := _} = Config = config(),
     gen_server:start_link({local, ?MODULE}, ?MODULE, Config, []).
 
+cleanup() ->
+    erase_cached_enabled_status().
+
 -spec maybe_garbage_collect(aec_headers:header()) -> ok | nop.
 maybe_garbage_collect(Header) ->
-    gen_server:call(
-      ?MODULE, {maybe_garbage_collect, aec_headers:height(Header), Header}).
+    case get_cached_enabled_status() of
+        true ->
+            gen_server:call(
+              ?MODULE, {maybe_garbage_collect, aec_headers:height(Header), Header});
+        false ->
+            nop
+    end.
 
 %% If GC is disabled, or there hasn't yet been a GC, this function returns 0.
 -spec height_of_last_gc() -> non_neg_integer().
 height_of_last_gc() ->
-    aec_db:ensure_activity(async_dirty, fun aec_db:read_last_gc_switch/0).
+    case get_cached_enabled_status() of
+        true ->
+            aec_db:ensure_activity(async_dirty, fun aec_db:read_last_gc_switch/0);
+        false ->
+            0
+    end.
+
+state_at_height_still_reachable(Height) ->
+    case get_cached_enabled_status() of
+        true ->
+            #{last_gc := LastGC} = info([last_gc]),
+            LastGC =< Height;
+        false ->
+            true
+    end.
 
 info() ->
     info(info_keys()).
@@ -124,12 +152,13 @@ info(Keys) when is_list(Keys) ->
 info_keys() ->
     [enabled, history, last_gc, active_sweeps, during_sync, trees].
 
-%% called from aec_db on startup
-%% maybe_swap_nodes() ->
-%%     maybe_swap_nodes(?GCED_TABLE_NAME, ?TABLE_NAME).
+-ifdef(TEST).
+install_test_env() ->
+    cache_enabled_status(false).
+-endif.
 
 %%%===================================================================
-%%% gen_statem callbacks
+%%% gen_server callbacks
 %%%===================================================================
 
 %% Change of configuration parameters requires restart of the node.
@@ -140,6 +169,7 @@ init(#{ <<"enabled">>     := Enabled
       , <<"minimum_height">> := MinHeight
       } = Cfg) when is_integer(History), History > 0 ->
     lager:debug("Cfg = ~p", [Cfg]),
+    cache_enabled_status(Enabled),
     LastSwitch = case Enabled of
                      true ->
                          aec_events:subscribe(chain_sync),
@@ -185,6 +215,7 @@ handle_call({maybe_garbage_collect, TopHeight, Header}, _From,
                 history = History, min_height = MinHeight,
                 trees = Trees, scanners = [], last_switch = Last} = St)
   when (Synced orelse DuringSync), TopHeight >= MinHeight, TopHeight > Last + History ->
+    lager:debug("WILL collect. St = ~p", [lager:pr(St, ?MODULE)]),
     %% Double-check that the GC hasn't been requested on a microblock.
     %% This would be a bug, since aec_conductor should only ask for keyblocks.
     case aec_headers:type(Header) of
@@ -200,6 +231,7 @@ handle_call({maybe_garbage_collect, TopHeight, Header}, _From,
             {reply, nop, St}
     end;
 handle_call({maybe_garbage_collect, _, _}, _From, St) ->
+    lager:debug("Won't collect. St = ~p", [lager:pr(St, ?MODULE)]),
     {reply, nop, St};
 handle_call({info, Keys}, _, St) ->
     {reply, info_(Keys, St), St}.
@@ -210,6 +242,12 @@ handle_cast({scanning_failed, ErrHeight}, St) ->
 
 handle_cast({scan_complete, Name}, #st{scanners = Scanners} = St) ->
     Scanners1 = lists:keydelete(Name, #scanner.tree, Scanners),
+    case Scanners1 of
+        [] ->
+            aec_events:publish(gc, scans_complete);
+        _ ->
+            ok
+    end,
     {noreply, St#st{scanners = Scanners1}};
 
 handle_cast(_, St) ->
@@ -249,6 +287,7 @@ perform_switch(Trees, Height) ->
                    switch_tables(Trees),
                    aec_db:write_last_gc_switch(Height)
            end),
+    aec_events:publish(gc, {gc_switch, Height}),
     [start_scanner(T, Height) || T <- Trees].
 
 clear_secondary_tables(Trees) ->
@@ -378,6 +417,15 @@ signal_switching_failed_and_reply(St, Reply) ->
             {reply, Reply, St}
     end.
 
+cache_enabled_status(Bool) when is_boolean(Bool) ->
+    persistent_term:put({?MODULE, gc_enabled}, Bool).
+
+get_cached_enabled_status() ->
+    persistent_term:get({?MODULE, gc_enabled}).
+
+erase_cached_enabled_status() ->
+    persistent_term:erase({?MODULE, gc_enabled}).
+
 config() ->
     Trees = get_trees(),
     %% In the LC below, we rely on the schema to provide default values.

apps/aecore/src/aec_events.erl

@@ -33,6 +33,7 @@
                | peers
                | metric
                | chain_sync
+               | gc
                | oracle_query_tx_created
                | oracle_response_tx_created
                | {tx_event, any()}.

apps/aecore/src/aecore_app.erl

@@ -33,6 +33,7 @@ prep_stop(State) ->
 
 stop(_State) ->
     lager:info("Stopping aecore app", []),
+    aec_db_gc:cleanup(),
     aec_db:cleanup(),
     ok.
 

apps/aecore/test/aec_db_gc_SUITE.erl

@@ -135,7 +135,7 @@ main_test(_Config) ->
 
     true = has_key(N1, ?DUMMY_HASH, Primary1),
 
-    %% Mining of another keyblock starts second GC phase - storing cache to mnesia table and restart
+    %% Mining of another keyblock starts second GC phase
     mine_until_height(N1, N2, ?GC_HISTORY + 2),
 
     Primary2 = primary(N1),
@@ -248,7 +248,9 @@ new_pubkey() ->
     PubKey.
 
 calls_test(_Config) ->
+    aecore_suite_utils:use_swagger(oas3),   % GC-related return codes only in OAS3 for now
     [N1] = [aecore_suite_utils:node_name(X) || X <- [dev1]],
+    ok = aecore_suite_utils:subscribe(N1, gc),
     H0 = aec_headers:height(rpc:call(N1, aec_chain, top_header, [])),
     ct:log("Current height is ~p", [H0]),
     %% create a new account and seed it with tokens
@@ -333,16 +335,26 @@ calls_test(_Config) ->
     Spend(OwnerPubkey, OwnerPrivkey),
     Spend(OwnerPubkey, OwnerPrivkey),
     %% mine beyond the GC
-    aecore_suite_utils:mine_key_blocks(N1, ?GC_HISTORY + 1),
+    ct:log("Mining beyond the GC point"),
+    ct:log("GC server state: ~p", [rpc:call(N1, sys, get_state, [aec_db_gc])]),
+    {ok, Mined1} = aecore_suite_utils:mine_key_blocks(N1, ?GC_HISTORY + 1),
+    ct:log("Blocks mined: ~p", [length(Mined1)]),
+    _GcSwitch1 = await_gc_switch(),
+    await_scans_complete(),
     H1 = aec_headers:height(rpc:call(N1, aec_chain, top_header, [])),
     ct:log("Current height is ~p", [H1]),
-    {ok, 200, _} = aehttp_integration_SUITE:get_contract_call_object(ContractCreateTxHash),
-    {ok, 200, _} = aehttp_integration_SUITE:get_contract_call_object(ContractCallTxHash),
+    ct:log("Last GC switch: ~p", [rpc:call(N1, aec_db_gc, info, [[last_gc]])]),
+    {ok, Mined2} = aecore_suite_utils:mine_key_blocks(N1, ?GC_HISTORY),
+    ct:log("Mined ~p keyblocks", [length(Mined2)]),
+    await_gc_switch(),
+    ct:log("Second GC switch (and clearing tables)", []),
+    {ok, 410, _} = aehttp_integration_SUITE:get_contract_call_object(ContractCreateTxHash),
+    {ok, 410, _} = aehttp_integration_SUITE:get_contract_call_object(ContractCallTxHash),
     {ok, 200, _} =
         aehttp_integration_SUITE:get_accounts_by_pubkey_sut(OwnerAddress),
-    %% this should fail:
-    {ok, 200, _} =
+    {ok, 410, _} =
         aehttp_integration_SUITE:get_accounts_by_pubkey_and_height_sut(OwnerAddress, ContractCreateHeight),
+    aecore_suite_utils:unsubscribe(N1, gc),
     ok.
 
 latest_sophia_abi() ->
@@ -359,3 +371,21 @@ sign_and_post_tx(EncodedUnsignedTx, SenderPrivkey, Node) ->
     {Res, aeser_api_encoder:encode(tx_hash, aetx_sign:hash(SignedTx))}.
 
 
+await_scans_complete() ->
+    receive
+        {gproc_ps_event, gc, #{info := scans_complete}} ->
+            ok;
+        OtherMsg ->
+            ct:log("Got OTHER: ~p", [OtherMsg]),
+            error({unexpected_msg, OtherMsg})
+    after 10000 ->
+            error({timeout, waiting_for_scans_complete})
+    end.
+
+await_gc_switch() ->
+    receive
+        {gproc_ps_event, gc, #{info := {gc_switch, AtHeight}}} ->
+            ct:log("Got GC switch notification for height ~p", [AtHeight])
+    after 10000 ->
+            error({timeout, waiting_for_gc_switch})
+    end.

apps/aecore/test/aec_test_utils.erl

@@ -332,6 +332,7 @@ start_chain_db(ram) ->
     ok = aec_db:initialize_db(ram),
     Tabs = [Tab || {Tab, _} <- aec_db:tables(ram)],
     ok = mnesia:wait_for_tables(Tabs, 5000),
+    aec_db_gc:install_test_env(),
     ok;
 
 start_chain_db(disc) ->
@@ -344,8 +345,10 @@ start_chain_db(disc) ->
 stop_chain_db() ->
     stop_chain_db(persistent_term:get({?MODULE, db_mode})).
 stop_chain_db(ram) ->
+    aec_db_gc:cleanup(),
     application:stop(mnesia);
 stop_chain_db({disc, Persist}) ->
+    aec_db_gc:cleanup(),
     application:stop(mnesia),
     application:set_env(aecore, persist, Persist),
     ok = mnesia:delete_schema([node()]).