<commit_before>% @copyright 2011, 2012, 2013 Zuse Institute Berlin

%   Licensed under the Apache License, Version 2.0 (the "License");
%   you may not use this file except in compliance with the License.
%   You may obtain a copy of the License at
%
%       http://www.apache.org/licenses/LICENSE-2.0
%
%   Unless required by applicable law or agreed to in writing, software
%   distributed under the License is distributed on an "AS IS" BASIS,
%   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%   See the License for the specific language governing permissions and
%   limitations under the License.

%% @author <NAME> <<EMAIL>>
%% @doc    replica reconcilication module
%% @end
%% @version $Id$
-module(rr_recon).
-author('<EMAIL>').
-vsn('$Id$').

-behaviour(gen_component).

-include("record_helpers.hrl").
-include("scalaris.hrl").

-export([init/1, on/2, start/2, check_config/0]).
-export([map_key_to_interval/2, map_key_to_quadrant/2, map_interval/2,
         quadrant_intervals/0]).
-export([get_chunk_kv/1, get_chunk_kvv/1, get_chunk_filter/1]).
%-export([compress_kv_list/4, calc_signature_size_1_to_n/3, calc_signature_size_n_pair/3]).

%export for testing
-export([find_sync_interval/2, quadrant_subints_/3]).
-export([merkle_compress_hashlist/3, merkle_decompress_hashlist/3,
         merkle_compress_cmp_result/4, merkle_decompress_cmp_result/4]).

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% debug
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

-define(TRACE(X,Y), ok).
%-define(TRACE(X,Y), log:pal("~w: [ ~.0p ] " ++ X ++ "~n", [?MODULE, self()] ++ Y)).
-define(TRACE_SEND(Pid, Msg), ?TRACE("to ~.0p: ~.0p~n", [Pid, Msg])).
-define(TRACE1(Msg, State),
        ?TRACE("~n  Msg: ~.0p~n"
               "  State: method: ~.0p;  stage: ~.0p;  initiator: ~.0p~n"
               "          destI: ~.0p~n"
               "         params: ~.0p~n",
               [Msg, State#rr_recon_state.method, State#rr_recon_state.stage,
                State#rr_recon_state.initiator, State#rr_recon_state.dest_interval,
                ?IIF(is_list(State#rr_recon_state.struct), State#rr_recon_state.struct, [])])).

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% type definitions
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-ifdef(with_export_type_support).
-export_type([method/0, request/0, merkle_cmp_result/0,
              db_chunk_kv/0, db_chunk_kvv/0]).
-endif.

-type quadrant()       :: 1..4. % 1..rep_factor()
-type method()         :: trivial | bloom | merkle_tree | art.% | iblt.
-type stage()          :: req_shared_interval | build_struct | reconciliation | resolve.

-type exit_reason()    :: empty_interval |      %interval intersection between initator and client is empty
                          recon_node_crash |    %sync partner node crashed
                          sync_finished |       %finish recon on local node
                          sync_finished_remote. %client-side shutdown by merkle-tree recon initiator

-type db_chunk_kv()    :: [{?RT:key(), db_dht:version()}].
-type db_chunk_kvv()   :: [{?RT:key(), db_dht:version(), db_dht:value()}].

-type signature_size() :: 1..160. % upper bound of 160 (SHA-1) to also limit testing

-record(trivial_recon_struct,
        {
         interval = intervals:empty()                         :: intervals:interval(),
         reconPid = undefined                                 :: comm:mypid() | undefined,
         db_chunk = ?required(trivial_recon_struct, db_chunk) :: bitstring() | gb_tree(),
         sig_size = 128                                       :: signature_size(),
         ver_size = 8                                         :: signature_size()
        }).

-record(bloom_recon_struct,
        {
         interval = intervals:empty()                       :: intervals:interval(),
         bloom    = ?required(bloom_recon_struct, bloom)    :: bloom:bloom_filter()
        }).

-record(merkle_params,
        {
         interval       = ?required(merkle_param, interval)       :: intervals:interval(),
         reconPid       = undefined                               :: comm:mypid() | undefined,
         branch_factor  = ?required(merkle_param, branch_factor)  :: pos_integer(),
         bucket_size    = ?required(merkle_param, bucket_size)    :: pos_integer()
        }).

-record(art_recon_struct,
        {
         art            = ?required(art_recon_struct, art)            :: art:art(),
         branch_factor  = ?required(art_recon_struct, branch_factor)  :: pos_integer(),
         bucket_size    = ?required(art_recon_struct, bucket_size)    :: pos_integer()
        }).

-type sync_struct() :: #trivial_recon_struct{} |
                       #bloom_recon_struct{} |
                       merkle_tree:merkle_tree() |
                       [merkle_tree:mt_node()] |
                       #art_recon_struct{}.
-type parameters() :: #trivial_recon_struct{} |
                      #bloom_recon_struct{} |
                      #merkle_params{} |
                      #art_recon_struct{}.
-type recon_dest() :: ?RT:key() | random.

-record(rr_recon_state,
        {
         ownerPid           = ?required(rr_recon_state, ownerPid)    :: comm:erl_local_pid(),
         dhtNodePid         = ?required(rr_recon_state, dhtNodePid)  :: comm:erl_local_pid(),
         dest_rr_pid        = ?required(rr_recon_state, dest_rr_pid) :: comm:mypid(), %dest rrepair pid
         dest_recon_pid     = undefined                              :: comm:mypid() | undefined, %dest recon process pid
         method             = undefined                              :: method() | undefined,
         dest_interval      = intervals:empty()                      :: intervals:interval(),
         my_sync_interval   = intervals:empty()                      :: intervals:interval(),
         params             = {}                                     :: parameters() | {}, % parameters from the other node
         struct             = {}                                     :: sync_struct() | {}, % my recon structure
         stage              = req_shared_interval                    :: stage(),
         initiator          = false                                  :: boolean(),
         misc               = []                                     :: [{atom(), term()}], % any optional parameters an algorithm wants to keep
         kv_list            = []                                     :: db_chunk_kv(),
         stats              = rr_recon_stats:new()                   :: rr_recon_stats:stats(),
         to_resolve         = {[], []}                               :: {ToSend::rr_resolve:kvv_list(), ToReq::[?RT:key()]}
         }).
-type state() :: #rr_recon_state{}.

% keep in sync with check_node/2
-define(recon_ok,         1).
-define(recon_fail_stop,  2).
-define(recon_fail_cont,  3).
-define(recon_fail_cont2, 0). % failed leaf with included hash

-type merkle_cmp_result()  :: ?recon_ok |
                              ?recon_fail_stop |
                              {merkle_tree:mt_node_key()} |
                              ?recon_fail_cont.
-type merkle_cmp_request() :: {Hash::merkle_tree:mt_node_key(), IsLeaf::boolean()}.

-type request() ::
    {start, method(), DestKey::recon_dest()} |
    {create_struct, method(), SenderI::intervals:interval()} | % from initiator
    {start_recon, bloom, #bloom_recon_struct{}} | % to initiator
    {start_recon, merkle_tree, #merkle_params{}} | % to initiator
    {start_recon, art, #art_recon_struct{}}. % to initiator

-type message() ::
    % API
    request() |
    % trivial sync messages
    {resolve_req, BinKeys::bitstring(), SigSize::signature_size()} |
    % merkle tree sync messages
    {?check_nodes, SenderPid::comm:mypid(), ToCheck::bitstring(), SigSize::signature_size()} |
    {?check_nodes, ToCheck::bitstring(), SigSize::signature_size()} |
    {?check_nodes_response, Results::bitstring(), HashKeys::bitstring(), MaxLeafCount::non_neg_integer()} |
    % dht node response
    {create_struct2, {get_state_response, MyI::intervals:interval()}} |
    {create_struct2, DestI::intervals:interval(),
     {get_chunk_response, {intervals:interval(), db_chunk_kv()}}} |
    {reconcile, {get_chunk_response, {intervals:interval(), db_chunk_kv()}}} |
    {resolve, {get_chunk_response, {intervals:interval(), db_chunk_kvv()}}} |
    % internal
    {shutdown, exit_reason()} |
    {crash, DeadPid::comm:mypid()} |
    {'DOWN', MonitorRef::reference(), process, Owner::comm:erl_local_pid(), Info::any()}
    .

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Message handling
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-spec on(message(), state()) -> state() | kill.

on({create_struct, RMethod, SenderI} = _Msg, State) ->
    ?TRACE1(_Msg, State),
    % (first) request from initiator to create a sync struct
    This = comm:reply_as(comm:this(), 2, {create_struct2, '_'}),
    comm:send_local(State#rr_recon_state.dhtNodePid, {get_state, This, my_range}),
    State#rr_recon_state{method = RMethod, initiator = false, dest_interval = SenderI};

on({create_struct2, {get_state_response, MyI}} = _Msg,
   State = #rr_recon_state{stage = req_shared_interval, initiator = false,
                           method = RMethod,            dhtNodePid = DhtPid,
                           dest_rr_pid = DestRRPid,     dest_interval = SenderI}) ->
    ?TRACE1(_Msg, State),
    % target node got sync request, asked for its interval
    % dest_interval contains the interval of the initiator
    % -> client creates recon structure based on common interval, sends it to initiator
    SyncI = find_sync_interval(MyI, SenderI),
    NewState = State#rr_recon_state{stage = build_struct,
                                    my_sync_interval = SyncI},
    case intervals:is_empty(SyncI) of
        false ->
            case RMethod of
                trivial -> fd:subscribe(DestRRPid);
                merkle_tree -> fd:subscribe(DestRRPid);
                _ -> ok
            end,
            % reduce SenderI to the sub-interval matching SyncI, i.e. a mapped SyncI
            SenderSyncI = map_interval(SenderI, SyncI),
            send_chunk_req(DhtPid, self(), SyncI, SenderSyncI, get_max_items(), create_struct),
            NewState;
        true ->
            shutdown(empty_interval, NewState)
    end;

on({create_struct2, DestI, {get_chunk_response, {RestI, DBList0}}} = _Msg,
   State = #rr_recon_state{stage = build_struct,       initiator = false}) ->
    ?TRACE1(_Msg, State),
    % create recon structure based on all elements in sync interval
    DBList = [{Key, VersionX} || {KeyX, VersionX} <- DBList0,
                                 none =/= (Key = map_key_to_interval(KeyX, DestI))],
    build_struct(DBList, DestI, RestI, State);

on({start_recon, RMethod, Params} = _Msg, State) ->
    ?TRACE1(_Msg, State),
    % initiator got other node's sync struct or parameters over sync interval
    % (mapped to the initiator's range)
    % -> create own recon structure based on sync interval and reconcile
    % note: sync interval may be an outdated sub-interval of this node's range
    %       -> pay attention when saving values to DB!
    %       (it could be outdated then even if we retrieved the current range now!)
    case RMethod of
        trivial ->
            #trivial_recon_struct{interval = MySyncI, reconPid = DestReconPid,
                                  db_chunk = DBChunk,
                                  sig_size = SigSize, ver_size = VSize} = Params,
            ?ASSERT(DestReconPid =/= undefined),
            fd:subscribe(DestReconPid),
            % convert db_chunk to a gb_tree for faster access checks
            DBChunkTree =
                decompress_kv_list(DBChunk, gb_trees:empty(), SigSize, VSize),
            Params1 = Params#trivial_recon_struct{db_chunk = DBChunkTree},
            Reconcile = resolve,
            Stage = resolve;
        bloom ->
            MySyncI = Params#bloom_recon_struct.interval,
            DestReconPid = undefined,
            Params1 = Params,
            Reconcile = reconcile,
            Stage = reconciliation;
        merkle_tree ->
            #merkle_params{interval = MySyncI, reconPid = DestReconPid} = Params,
            ?ASSERT(DestReconPid =/= undefined),
            fd:subscribe(DestReconPid),
            Params1 = Params,
            Reconcile = reconcile,
            Stage = reconciliation;
        art ->
            MySyncI = art:get_interval(Params#art_recon_struct.art),
            DestReconPid = undefined,
            Params1 = Params,
            Reconcile = reconcile,
            Stage = reconciliation
    end,
    % client only sends non-empty sync intervals or exits
    ?ASSERT(not intervals:is_empty(MySyncI)),
    
    DhtNodePid = State#rr_recon_state.dhtNodePid,
    send_chunk_req(DhtNodePid, self(), MySyncI, MySyncI, get_max_items(), Reconcile),
    State#rr_recon_state{stage = Stage, params = Params1,
                         method = RMethod, initiator = true,
                         dest_recon_pid = DestReconPid};

on({resolve, {get_chunk_response, {RestI, DBList}}} = _Msg,
   State = #rr_recon_state{stage = resolve,            initiator = true,
                           method = trivial,           dhtNodePid = DhtNodePid,
                           params = #trivial_recon_struct{db_chunk = OtherDBChunk,
                                                          sig_size = SigSize,
                                                          ver_size = VSize} = Params,
                           dest_rr_pid = DestRR_Pid,   stats = Stats,
                           ownerPid = OwnerL, to_resolve = {ToSend, ToReq},
                           dest_recon_pid = DestReconPid}) ->
    ?TRACE1(_Msg, State),

    {ToSend1, ToReq1, OtherDBChunk1} =
        get_diff(DBList, OtherDBChunk, ToSend, ToReq, SigSize, VSize),

    %if rest interval is non empty start another sync
    SID = rr_recon_stats:get(session_id, Stats),
    SyncFinished = intervals:is_empty(RestI),
    if not SyncFinished ->
           send_chunk_req(DhtNodePid, self(), RestI, RestI, get_max_items(), resolve);
       true -> ok
    end,
    ?TRACE("Reconcile Trivial Session=~p ; ToSend=~p ; ToReq=~p",
           [SID, length(ToSend1), length(ToReq1)]),
    NewStats =
        if ToSend1 =/= [] orelse ToReq1 =/= [] orelse SyncFinished ->
               send(DestRR_Pid, {request_resolve, SID,
                                 {?key_upd, ToSend1, ToReq1},
                                 [{from_my_node, 0},
                                  {feedback_request, comm:make_global(OwnerL)}]}),
               % we will get one reply from a subsequent feedback response
               rr_recon_stats:inc([{resolve_started, 1},
                                   {await_rs_fb, 1}], Stats);
           true ->
               Stats
        end,
    Params1 = Params#trivial_recon_struct{db_chunk = OtherDBChunk1},
    NewState = State#rr_recon_state{to_resolve = {[], []}},
    if SyncFinished ->
           % let the non-initiator's rr_recon process identify the remaining keys
           Req2Count = gb_trees:size(OtherDBChunk1),
           ToReq2 = util:gb_trees_foldl(
                      fun(KeyBin, _VersionShort, Acc) ->
                              <<Acc/bitstring, KeyBin/bitstring>>
                      end, <<>>, OtherDBChunk1),
           NewStats2 =
               if Req2Count > 0 ->
                      rr_recon_stats:inc([{resolve_started, 1}], NewStats);
                  true -> NewStats
               end,

           ?TRACE("resolve_req Trivial Session=~p ; ToReq=~p (~p bits)",
                  [SID, Req2Count, erlang:bit_size(ToReq2)]),
           comm:send(DestReconPid, {resolve_req, ToReq2, SigSize}),
           
           Params2 = Params1#trivial_recon_struct{db_chunk = gb_trees:empty()},
           shutdown(sync_finished,
                    NewState#rr_recon_state{stats = NewStats2, params = Params2});
       true ->
           NewState#rr_recon_state{stats = NewStats, params = Params1}
    end;

on({reconcile, {get_chunk_response, {RestI, DBList0}}} = _Msg,
   State = #rr_recon_state{stage = reconciliation,     initiator = true,
                           method = bloom,             dhtNodePid = DhtNodePid,
                           params = #bloom_recon_struct{bloom = BF},
                           dest_rr_pid = DestRR_Pid,   stats = Stats,
                           ownerPid = OwnerL}) ->
    ?TRACE1(_Msg, State),
    % no need to map keys since the other node's bloom filter was created with
    % keys mapped to our interval
    Diff = case bloom:item_count(BF) of
               0 -> DBList0;
               _ -> [X || X <- DBList0, not bloom:is_element(BF, X)]
           end,
    %if rest interval is non empty start another sync
    SID = rr_recon_stats:get(session_id, Stats),
    SyncFinished = intervals:is_empty(RestI),
    if not SyncFinished ->
           send_chunk_req(DhtNodePid, self(), RestI, RestI, get_max_items(), reconcile);
       true -> ok
    end,
    ?TRACE("Reconcile Bloom Session=~p ; Diff=~p", [SID, length(Diff)]),
    NewStats =
        case Diff of
            [_|_] ->
                send(DestRR_Pid, {request_resolve, SID,
                                  {?key_upd2, Diff, comm:make_global(OwnerL)},
                                    [{from_my_node, 0}]}),
                % we will get one reply from a subsequent ?key_upd resolve
                rr_recon_stats:inc([{resolve_started, 1}], Stats);
            [] -> Stats
        end,
    NewState = State#rr_recon_state{stats = NewStats},
    if SyncFinished -> shutdown(sync_finished, NewState);
       true         -> NewState
    end;

on({reconcile, {get_chunk_response, {RestI, DBList}}} = _Msg,
   State = #rr_recon_state{stage = reconciliation,     initiator = true,
                           method = RMethod,           params = Params})
  when RMethod =:= merkle_tree orelse RMethod =:= art->
    ?TRACE1(_Msg, State),
    % no need to map keys since the other node's sync struct was created with
    % keys mapped to our interval
    MySyncI = case RMethod of
                  merkle_tree -> Params#merkle_params.interval;
                  art         -> art:get_interval(Params#art_recon_struct.art)
              end,
    build_struct(DBList, MySyncI, RestI, State);

on({crash, _Pid} = _Msg, State) ->
    ?TRACE1(_Msg, State),
    shutdown(recon_node_crash, State);

on({shutdown, Reason}, State) ->
    shutdown(Reason, State);

on({'DOWN', _MonitorRef, process, _Owner, _Info}, _State) ->
    log:log(info, "[ ~p - ~p] shutdown due to rrepair shut down", [?MODULE, comm:this()]),
    kill;

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% trivial reconciliation sync messages
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

on({resolve_req, BinKeys, SigSize} = _Msg,
   State = #rr_recon_state{stage = resolve,           initiator = false,
                           dest_rr_pid = DestRRPid,   ownerPid = OwnerL,
                           kv_list = KVList,          stats = Stats}) ->
    ?TRACE1(_Msg, State),
    ReqSet = decompress_k_list(BinKeys, gb_sets:empty(), SigSize),
    NewStats =
        case gb_sets:is_empty(ReqSet) of
            true -> Stats;
            _ ->
                ReqKeys = [Key || {Key, _Version} <- KVList,
                                  gb_sets:is_member(compress_key(Key, SigSize),
                                                    ReqSet)],
                
                SID = rr_recon_stats:get(session_id, Stats),
                % note: the resolve request was counted at the initiator and
                %       thus from_my_node must be 0 on this node!
                send_local(OwnerL, {request_resolve, SID,
                                    {key_upd_send, DestRRPid, ReqKeys, []},
                                    [{feedback_request, comm:make_global(OwnerL)},
                                     {from_my_node, 0}]}),
                % we will get one reply from a subsequent ?key_upd resolve
                rr_recon_stats:inc([{resolve_started, 1}], Stats)
        end,
    shutdown(sync_finished, State#rr_recon_state{stats = NewStats});

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% merkle tree sync messages
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

on({?check_nodes, SenderPid, ToCheck, SigSize}, State) ->
    NewState = State#rr_recon_state{dest_recon_pid = SenderPid},
    on({?check_nodes, ToCheck, SigSize}, NewState);

on({?check_nodes, ToCheck0, SigSize},
   State = #rr_recon_state{ stage = reconciliation,    initiator = false,
                            struct = Tree,             ownerPid = OwnerL,
                            dest_rr_pid = DestNodePid, stats = Stats,
                            dest_recon_pid = DestReconPid }) ->
    ?ASSERT(comm:is_valid(DestReconPid)),
    ToCheck = merkle_decompress_hashlist(ToCheck0, [], SigSize),
    {Result, RestTree, ToResolve, MaxLeafCount} =
        check_node(ToCheck, Tree, SigSize),
    {Flags, HashKeys} =
        merkle_compress_cmp_result(Result, <<>>, <<>>, SigSize),
    ToResolveLength = length(ToResolve),
    NStats =
        if ToResolveLength > 0 ->
               SID = rr_recon_stats:get(session_id, Stats),
               ResolveCount = resolve_leaves(ToResolve, DestNodePid, SID, OwnerL),
               rr_recon_stats:inc([{tree_leavesSynced, ToResolveLength},
                                   {resolve_started, ResolveCount}], Stats);
           true -> Stats
        end,
    send(DestReconPid, {?check_nodes_response, Flags, HashKeys, MaxLeafCount}),
    State#rr_recon_state{ struct = RestTree, stats = NStats };

on({?check_nodes_response, Flags, HashKeys, OtherMaxLeafCount}, State =
       #rr_recon_state{ stage = reconciliation,        initiator = true,
                        struct = Tree,                 ownerPid = OwnerL,
                        dest_rr_pid = DestNodePid,     stats = Stats,
                        dest_recon_pid = DestReconPid, params = Params,
                        misc = [{signature_size, SigSize}] }) ->
    CmpResults = merkle_decompress_cmp_result(Flags, HashKeys, [], SigSize),
    SID = rr_recon_stats:get(session_id, Stats),
    case process_tree_cmp_result(CmpResults, Tree,
                                 Params#merkle_params.branch_factor,
                                 SigSize, Stats) of
        {[], ToResolve, NStats0, RTree, _MyMaxLeafCount} ->
            NextSigSize = SigSize,
            ok;
        {[_|_] = Req0, ToResolve, NStats0, RTree, MyMaxLeafCount} ->
            % note: we have m one-to-n comparisons, assuming the probability of
            %       a failure in a single one-to-n comparison is p, the overall
            %       p1e = 1 - (1-p)^n  <=>  p = 1 - (1 - p1e)^(1/n)
            P = 1 - math:pow(1 - get_p1e(), 1 / erlang:max(1, length(Req0))),
            NextSigSize = calc_signature_size_1_to_n(
                            erlang:max(MyMaxLeafCount, OtherMaxLeafCount),
                            P, 160),
%%             log:pal("MyMLC: ~p~n    OtherMLC: ~p~n     SigSize: ~p",
%%                     [MyMaxLeafCount, OtherMaxLeafCount, NextSigSize]),
            Req = merkle_compress_hashlist(Req0, <<>>, NextSigSize),
            send(DestReconPid, {?check_nodes, Req, NextSigSize})
    end,
    ResolveCount = resolve_leaves(ToResolve, DestNodePid, SID, OwnerL),
    NStats = rr_recon_stats:inc([{tree_leavesSynced, length(ToResolve)},
                                 {resolve_started, ResolveCount}], NStats0),
    CompLeft = rr_recon_stats:get(tree_compareLeft, NStats),
    NewState = State#rr_recon_state{stats = NStats, struct = RTree,
                                    misc = [{signature_size, NextSigSize}]},
    if CompLeft =:= 0 ->
           send(DestReconPid, {shutdown, sync_finished_remote}),
           shutdown(sync_finished, NewState);
       true -> NewState
    end.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

-spec build_struct(DBList::db_chunk_kv(), DestI::intervals:non_empty_interval(),
                   RestI::intervals:interval(), state()) -> state() | kill.
build_struct(DBList, DestI, RestI,
             State = #rr_recon_state{method = RMethod, params = Params,
                                     struct = OldSyncStruct,
                                     initiator = Initiator, stats = Stats,
                                     dhtNodePid = DhtNodePid, stage = Stage,
                                     kv_list = KVList}) ->
    ?ASSERT(not intervals:is_empty(DestI)),
    % bloom may fork more recon processes if un-synced elements remain
    BeginSync =
        case intervals:is_empty(RestI) of
            false ->
                SubSyncI = map_interval(DestI, RestI),
                case intervals:is_empty(SubSyncI) of
                    false ->
                        MySubSyncI = map_interval(RestI, DestI), % mapped to my range
                        Reconcile =
                            if Initiator andalso (Stage =:= reconciliation) -> reconcile;
                               true -> create_struct
                            end,
                        send_chunk_req(DhtNodePid, self(), MySubSyncI, SubSyncI,
                                       get_max_items(), Reconcile),
                        false;
                    true -> true
                end;
            true -> true
        end,
    NewKVList = lists:append(KVList, DBList),
    if BeginSync ->
           ToBuild = if Initiator andalso RMethod =:= art -> merkle_tree;
                        true -> RMethod
                     end,
           {BuildTime, SyncStruct} =
               util:tc(fun() -> build_recon_struct(ToBuild, OldSyncStruct, DestI,
                                                   NewKVList, Params, BeginSync)
                       end),
           Stats1 = rr_recon_stats:inc([{build_time, BuildTime}], Stats),
           NewState = State#rr_recon_state{struct = SyncStruct, stats = Stats1,
                                           kv_list = NewKVList},
           begin_sync(SyncStruct, Params, NewState#rr_recon_state{stage = reconciliation});
       true ->
           % keep stage (at initiator: reconciliation, at other: build_struct)
           State#rr_recon_state{kv_list = NewKVList}
    end.

-spec begin_sync(MySyncStruct::sync_struct(), OtherSyncStruct::parameters() | {},
                 state()) -> state() | kill.
begin_sync(MySyncStruct, _OtherSyncStruct = {},
           State = #rr_recon_state{method = trivial, initiator = false,
                                   ownerPid = OwnerL, stats = Stats,
                                   dest_rr_pid = DestRRPid}) ->
    ?TRACE("BEGIN SYNC", []),
    SID = rr_recon_stats:get(session_id, Stats),
    send(DestRRPid, {?IIF(SID =:= null, start_recon, continue_recon),
                     comm:make_global(OwnerL), SID,
                     {start_recon, trivial, MySyncStruct}}),
    State#rr_recon_state{struct = {}, stage = resolve};
begin_sync(MySyncStruct, _OtherSyncStruct = {},
           State = #rr_recon_state{method = bloom, initiator = false,
                                   ownerPid = OwnerL, stats = Stats,
                                   dest_rr_pid = DestRRPid}) ->
    ?TRACE("BEGIN SYNC", []),
    SID = rr_recon_stats:get(session_id, Stats),
    send(DestRRPid, {?IIF(SID =:= null, start_recon, continue_recon),
                     comm:make_global(OwnerL), SID,
                     {start_recon, bloom, MySyncStruct}}),
    shutdown(sync_finished, State#rr_recon_state{kv_list = []});
begin_sync(MySyncStruct, _OtherSyncStruct,
           State = #rr_recon_state{method = merkle_tree, initiator = Initiator,
                                   ownerPid = OwnerL, stats = Stats,
                                   dest_recon_pid = DestReconPid,
                                   dest_rr_pid = DestRRPid}) ->
    ?TRACE("BEGIN SYNC", []),
    Stats1 =
        rr_recon_stats:set(
          [{tree_compareLeft, ?IIF(Initiator, 1, 0)},
           {tree_size, merkle_tree:size_detail(MySyncStruct)}], Stats),
    case Initiator of
        true ->
            SigSize = 160,
            Req = merkle_compress_hashlist([merkle_tree:get_root(MySyncStruct)],
                                           <<>>, SigSize),
            send(DestReconPid, {?check_nodes, comm:this(), Req, SigSize}),
            State#rr_recon_state{stats = Stats1,
                                 misc = [{signature_size, SigSize}],
                                 kv_list = []};
        false ->
            MerkleI = merkle_tree:get_interval(MySyncStruct),
            MerkleV = merkle_tree:get_branch_factor(MySyncStruct),
            MerkleB = merkle_tree:get_bucket_size(MySyncStruct),
            MySyncParams = #merkle_params{interval = MerkleI,
                                          branch_factor = MerkleV,
                                          bucket_size = MerkleB},
            SyncParams = MySyncParams#merkle_params{reconPid = comm:this()},
            SID = rr_recon_stats:get(session_id, Stats),
            send(DestRRPid, {?IIF(SID =:= null, start_recon, continue_recon),
                             comm:make_global(OwnerL), SID,
                             {start_recon, merkle_tree, SyncParams}}),
            State#rr_recon_state{stats = Stats1, params = MySyncParams,
                                 kv_list = []}
    end;
begin_sync(MySyncStruct, OtherSyncStruct,
           State = #rr_recon_state{method = art, initiator = Initiator,
                                   ownerPid = OwnerL, stats = Stats,
                                   dest_rr_pid = DestRRPid}) ->
    ?TRACE("BEGIN SYNC", []),
    case Initiator of
        true ->
            Stats1 = art_recon(MySyncStruct, OtherSyncStruct#art_recon_struct.art, State),
            shutdown(sync_finished, State#rr_recon_state{stats = Stats1,
                                                         kv_list = []});
        false ->
            SID = rr_recon_stats:get(session_id, Stats),
            send(DestRRPid, {?IIF(SID =:= null, start_recon, continue_recon),
                             comm:make_global(OwnerL), SID,
                             {start_recon, art, MySyncStruct}}),
            shutdown(sync_finished, State#rr_recon_state{kv_list = []})
    end.

-spec shutdown(exit_reason(), state()) -> kill.
shutdown(Reason, #rr_recon_state{ownerPid = OwnerL, stats = Stats,
                                 initiator = Initiator, dest_rr_pid = DestRR,
                                 dest_recon_pid = DestRC}) ->
    ?TRACE("SHUTDOWN Session=~p Reason=~p",
           [rr_recon_stats:get(session_id, Stats), Reason]),
    Status = exit_reason_to_rc_status(Reason),
    NewStats = rr_recon_stats:set([{status, Status}], Stats),
    send_local(OwnerL, {recon_progress_report, comm:this(), Initiator, DestRR,
                        DestRC, NewStats}),
    kill.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% KV-List compression
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

%% @doc Calculates the minimum number of bits needed to have a hash collision
%%      probability of P1E, given we compare N hashes pairwise with each other.
-spec calc_signature_size_n_pair(N::non_neg_integer(), P1E::float(),
                                 MaxSize::signature_size())
        -> SigSize::signature_size().
calc_signature_size_n_pair(0, P1E, _MaxSize) when P1E > 0 ->
    1;
calc_signature_size_n_pair(N, P1E, MaxSize) when P1E > 0 ->
    erlang:min(MaxSize, erlang:max(1, util:ceil(util:log2(N * (N - 1) / P1E) - 1))).

%% @doc Transforms a list of key and version tuples (with unique keys), into a
%%      compact binary representation for transfer.
-spec compress_kv_list(KVList::[{?RT:key(), db_dht:version()}], Bin,
                       SigSize::signature_size(), VSize::signature_size())
        -> Bin when is_subtype(Bin, bitstring()).
compress_kv_list(KVList, Bin, SigSize, VSize) ->
    compress_kv_list_(KVList, Bin, SigSize, VSize, util:pow(2, VSize)).<commit_msg>rrepair, recon: fix signature size calculation for 1-element DB lists<commit_after>% @copyright 2011, 2012, 2013 Zuse Institute Berlin

%   Licensed under the Apache License, Version 2.0 (the "License");
%   you may not use this file except in compliance with the License.
%   You may obtain a copy of the License at
%
%       http://www.apache.org/licenses/LICENSE-2.0
%
%   Unless required by applicable law or agreed to in writing, software
%   distributed under the License is distributed on an "AS IS" BASIS,
%   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%   See the License for the specific language governing permissions and
%   limitations under the License.

%% @author <NAME> <<EMAIL>>
%% @doc    replica reconcilication module
%% @end
%% @version $Id$
-module(rr_recon).
-author('<EMAIL>').
-vsn('$Id$').

-behaviour(gen_component).

-include("record_helpers.hrl").
-include("scalaris.hrl").

-export([init/1, on/2, start/2, check_config/0]).
-export([map_key_to_interval/2, map_key_to_quadrant/2, map_interval/2,
         quadrant_intervals/0]).
-export([get_chunk_kv/1, get_chunk_kvv/1, get_chunk_filter/1]).
%-export([compress_kv_list/4, calc_signature_size_1_to_n/3, calc_signature_size_n_pair/3]).

%export for testing
-export([find_sync_interval/2, quadrant_subints_/3]).
-export([merkle_compress_hashlist/3, merkle_decompress_hashlist/3,
         merkle_compress_cmp_result/4, merkle_decompress_cmp_result/4]).

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% debug
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

-define(TRACE(X,Y), ok).
%-define(TRACE(X,Y), log:pal("~w: [ ~.0p ] " ++ X ++ "~n", [?MODULE, self()] ++ Y)).
-define(TRACE_SEND(Pid, Msg), ?TRACE("to ~.0p: ~.0p~n", [Pid, Msg])).
-define(TRACE1(Msg, State),
        ?TRACE("~n  Msg: ~.0p~n"
               "  State: method: ~.0p;  stage: ~.0p;  initiator: ~.0p~n"
               "          destI: ~.0p~n"
               "         params: ~.0p~n",
               [Msg, State#rr_recon_state.method, State#rr_recon_state.stage,
                State#rr_recon_state.initiator, State#rr_recon_state.dest_interval,
                ?IIF(is_list(State#rr_recon_state.struct), State#rr_recon_state.struct, [])])).

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% type definitions
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-ifdef(with_export_type_support).
-export_type([method/0, request/0, merkle_cmp_result/0,
              db_chunk_kv/0, db_chunk_kvv/0]).
-endif.

-type quadrant()       :: 1..4. % 1..rep_factor()
-type method()         :: trivial | bloom | merkle_tree | art.% | iblt.
-type stage()          :: req_shared_interval | build_struct | reconciliation | resolve.

-type exit_reason()    :: empty_interval |      %interval intersection between initator and client is empty
                          recon_node_crash |    %sync partner node crashed
                          sync_finished |       %finish recon on local node
                          sync_finished_remote. %client-side shutdown by merkle-tree recon initiator

-type db_chunk_kv()    :: [{?RT:key(), db_dht:version()}].
-type db_chunk_kvv()   :: [{?RT:key(), db_dht:version(), db_dht:value()}].

-type signature_size() :: 1..160. % upper bound of 160 (SHA-1) to also limit testing

-record(trivial_recon_struct,
        {
         interval = intervals:empty()                         :: intervals:interval(),
         reconPid = undefined                                 :: comm:mypid() | undefined,
         db_chunk = ?required(trivial_recon_struct, db_chunk) :: bitstring() | gb_tree(),
         sig_size = 128                                       :: signature_size(),
         ver_size = 8                                         :: signature_size()
        }).

-record(bloom_recon_struct,
        {
         interval = intervals:empty()                       :: intervals:interval(),
         bloom    = ?required(bloom_recon_struct, bloom)    :: bloom:bloom_filter()
        }).

-record(merkle_params,
        {
         interval       = ?required(merkle_param, interval)       :: intervals:interval(),
         reconPid       = undefined                               :: comm:mypid() | undefined,
         branch_factor  = ?required(merkle_param, branch_factor)  :: pos_integer(),
         bucket_size    = ?required(merkle_param, bucket_size)    :: pos_integer()
        }).

-record(art_recon_struct,
        {
         art            = ?required(art_recon_struct, art)            :: art:art(),
         branch_factor  = ?required(art_recon_struct, branch_factor)  :: pos_integer(),
         bucket_size    = ?required(art_recon_struct, bucket_size)    :: pos_integer()
        }).

-type sync_struct() :: #trivial_recon_struct{} |
                       #bloom_recon_struct{} |
                       merkle_tree:merkle_tree() |
                       [merkle_tree:mt_node()] |
                       #art_recon_struct{}.
-type parameters() :: #trivial_recon_struct{} |
                      #bloom_recon_struct{} |
                      #merkle_params{} |
                      #art_recon_struct{}.
-type recon_dest() :: ?RT:key() | random.

-record(rr_recon_state,
        {
         ownerPid           = ?required(rr_recon_state, ownerPid)    :: comm:erl_local_pid(),
         dhtNodePid         = ?required(rr_recon_state, dhtNodePid)  :: comm:erl_local_pid(),
         dest_rr_pid        = ?required(rr_recon_state, dest_rr_pid) :: comm:mypid(), %dest rrepair pid
         dest_recon_pid     = undefined                              :: comm:mypid() | undefined, %dest recon process pid
         method             = undefined                              :: method() | undefined,
         dest_interval      = intervals:empty()                      :: intervals:interval(),
         my_sync_interval   = intervals:empty()                      :: intervals:interval(),
         params             = {}                                     :: parameters() | {}, % parameters from the other node
         struct             = {}                                     :: sync_struct() | {}, % my recon structure
         stage              = req_shared_interval                    :: stage(),
         initiator          = false                                  :: boolean(),
         misc               = []                                     :: [{atom(), term()}], % any optional parameters an algorithm wants to keep
         kv_list            = []                                     :: db_chunk_kv(),
         stats              = rr_recon_stats:new()                   :: rr_recon_stats:stats(),
         to_resolve         = {[], []}                               :: {ToSend::rr_resolve:kvv_list(), ToReq::[?RT:key()]}
         }).
-type state() :: #rr_recon_state{}.

% keep in sync with check_node/2
-define(recon_ok,         1).
-define(recon_fail_stop,  2).
-define(recon_fail_cont,  3).
-define(recon_fail_cont2, 0). % failed leaf with included hash

-type merkle_cmp_result()  :: ?recon_ok |
                              ?recon_fail_stop |
                              {merkle_tree:mt_node_key()} |
                              ?recon_fail_cont.
-type merkle_cmp_request() :: {Hash::merkle_tree:mt_node_key(), IsLeaf::boolean()}.

-type request() ::
    {start, method(), DestKey::recon_dest()} |
    {create_struct, method(), SenderI::intervals:interval()} | % from initiator
    {start_recon, bloom, #bloom_recon_struct{}} | % to initiator
    {start_recon, merkle_tree, #merkle_params{}} | % to initiator
    {start_recon, art, #art_recon_struct{}}. % to initiator

-type message() ::
    % API
    request() |
    % trivial sync messages
    {resolve_req, BinKeys::bitstring(), SigSize::signature_size()} |
    % merkle tree sync messages
    {?check_nodes, SenderPid::comm:mypid(), ToCheck::bitstring(), SigSize::signature_size()} |
    {?check_nodes, ToCheck::bitstring(), SigSize::signature_size()} |
    {?check_nodes_response, Results::bitstring(), HashKeys::bitstring(), MaxLeafCount::non_neg_integer()} |
    % dht node response
    {create_struct2, {get_state_response, MyI::intervals:interval()}} |
    {create_struct2, DestI::intervals:interval(),
     {get_chunk_response, {intervals:interval(), db_chunk_kv()}}} |
    {reconcile, {get_chunk_response, {intervals:interval(), db_chunk_kv()}}} |
    {resolve, {get_chunk_response, {intervals:interval(), db_chunk_kvv()}}} |
    % internal
    {shutdown, exit_reason()} |
    {crash, DeadPid::comm:mypid()} |
    {'DOWN', MonitorRef::reference(), process, Owner::comm:erl_local_pid(), Info::any()}
    .

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Message handling
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
-spec on(message(), state()) -> state() | kill.

on({create_struct, RMethod, SenderI} = _Msg, State) ->
    ?TRACE1(_Msg, State),
    % (first) request from initiator to create a sync struct
    This = comm:reply_as(comm:this(), 2, {create_struct2, '_'}),
    comm:send_local(State#rr_recon_state.dhtNodePid, {get_state, This, my_range}),
    State#rr_recon_state{method = RMethod, initiator = false, dest_interval = SenderI};

on({create_struct2, {get_state_response, MyI}} = _Msg,
   State = #rr_recon_state{stage = req_shared_interval, initiator = false,
                           method = RMethod,            dhtNodePid = DhtPid,
                           dest_rr_pid = DestRRPid,     dest_interval = SenderI}) ->
    ?TRACE1(_Msg, State),
    % target node got sync request, asked for its interval
    % dest_interval contains the interval of the initiator
    % -> client creates recon structure based on common interval, sends it to initiator
    SyncI = find_sync_interval(MyI, SenderI),
    NewState = State#rr_recon_state{stage = build_struct,
                                    my_sync_interval = SyncI},
    case intervals:is_empty(SyncI) of
        false ->
            case RMethod of
                trivial -> fd:subscribe(DestRRPid);
                merkle_tree -> fd:subscribe(DestRRPid);
                _ -> ok
            end,
            % reduce SenderI to the sub-interval matching SyncI, i.e. a mapped SyncI
            SenderSyncI = map_interval(SenderI, SyncI),
            send_chunk_req(DhtPid, self(), SyncI, SenderSyncI, get_max_items(), create_struct),
            NewState;
        true ->
            shutdown(empty_interval, NewState)
    end;

on({create_struct2, DestI, {get_chunk_response, {RestI, DBList0}}} = _Msg,
   State = #rr_recon_state{stage = build_struct,       initiator = false}) ->
    ?TRACE1(_Msg, State),
    % create recon structure based on all elements in sync interval
    DBList = [{Key, VersionX} || {KeyX, VersionX} <- DBList0,
                                 none =/= (Key = map_key_to_interval(KeyX, DestI))],
    build_struct(DBList, DestI, RestI, State);

on({start_recon, RMethod, Params} = _Msg, State) ->
    ?TRACE1(_Msg, State),
    % initiator got other node's sync struct or parameters over sync interval
    % (mapped to the initiator's range)
    % -> create own recon structure based on sync interval and reconcile
    % note: sync interval may be an outdated sub-interval of this node's range
    %       -> pay attention when saving values to DB!
    %       (it could be outdated then even if we retrieved the current range now!)
    case RMethod of
        trivial ->
            #trivial_recon_struct{interval = MySyncI, reconPid = DestReconPid,
                                  db_chunk = DBChunk,
                                  sig_size = SigSize, ver_size = VSize} = Params,
            ?ASSERT(DestReconPid =/= undefined),
            fd:subscribe(DestReconPid),
            % convert db_chunk to a gb_tree for faster access checks
            DBChunkTree =
                decompress_kv_list(DBChunk, gb_trees:empty(), SigSize, VSize),
            Params1 = Params#trivial_recon_struct{db_chunk = DBChunkTree},
            Reconcile = resolve,
            Stage = resolve;
        bloom ->
            MySyncI = Params#bloom_recon_struct.interval,
            DestReconPid = undefined,
            Params1 = Params,
            Reconcile = reconcile,
            Stage = reconciliation;
        merkle_tree ->
            #merkle_params{interval = MySyncI, reconPid = DestReconPid} = Params,
            ?ASSERT(DestReconPid =/= undefined),
            fd:subscribe(DestReconPid),
            Params1 = Params,
            Reconcile = reconcile,
            Stage = reconciliation;
        art ->
            MySyncI = art:get_interval(Params#art_recon_struct.art),
            DestReconPid = undefined,
            Params1 = Params,
            Reconcile = reconcile,
            Stage = reconciliation
    end,
    % client only sends non-empty sync intervals or exits
    ?ASSERT(not intervals:is_empty(MySyncI)),
    
    DhtNodePid = State#rr_recon_state.dhtNodePid,
    send_chunk_req(DhtNodePid, self(), MySyncI, MySyncI, get_max_items(), Reconcile),
    State#rr_recon_state{stage = Stage, params = Params1,
                         method = RMethod, initiator = true,
                         dest_recon_pid = DestReconPid};

on({resolve, {get_chunk_response, {RestI, DBList}}} = _Msg,
   State = #rr_recon_state{stage = resolve,            initiator = true,
                           method = trivial,           dhtNodePid = DhtNodePid,
                           params = #trivial_recon_struct{db_chunk = OtherDBChunk,
                                                          sig_size = SigSize,
                                                          ver_size = VSize} = Params,
                           dest_rr_pid = DestRR_Pid,   stats = Stats,
                           ownerPid = OwnerL, to_resolve = {ToSend, ToReq},
                           dest_recon_pid = DestReconPid}) ->
    ?TRACE1(_Msg, State),

    {ToSend1, ToReq1, OtherDBChunk1} =
        get_diff(DBList, OtherDBChunk, ToSend, ToReq, SigSize, VSize),

    %if rest interval is non empty start another sync
    SID = rr_recon_stats:get(session_id, Stats),
    SyncFinished = intervals:is_empty(RestI),
    if not SyncFinished ->
           send_chunk_req(DhtNodePid, self(), RestI, RestI, get_max_items(), resolve);
       true -> ok
    end,
    ?TRACE("Reconcile Trivial Session=~p ; ToSend=~p ; ToReq=~p",
           [SID, length(ToSend1), length(ToReq1)]),
    NewStats =
        if ToSend1 =/= [] orelse ToReq1 =/= [] orelse SyncFinished ->
               send(DestRR_Pid, {request_resolve, SID,
                                 {?key_upd, ToSend1, ToReq1},
                                 [{from_my_node, 0},
                                  {feedback_request, comm:make_global(OwnerL)}]}),
               % we will get one reply from a subsequent feedback response
               rr_recon_stats:inc([{resolve_started, 1},
                                   {await_rs_fb, 1}], Stats);
           true ->
               Stats
        end,
    Params1 = Params#trivial_recon_struct{db_chunk = OtherDBChunk1},
    NewState = State#rr_recon_state{to_resolve = {[], []}},
    if SyncFinished ->
           % let the non-initiator's rr_recon process identify the remaining keys
           Req2Count = gb_trees:size(OtherDBChunk1),
           ToReq2 = util:gb_trees_foldl(
                      fun(KeyBin, _VersionShort, Acc) ->
                              <<Acc/bitstring, KeyBin/bitstring>>
                      end, <<>>, OtherDBChunk1),
           NewStats2 =
               if Req2Count > 0 ->
                      rr_recon_stats:inc([{resolve_started, 1}], NewStats);
                  true -> NewStats
               end,

           ?TRACE("resolve_req Trivial Session=~p ; ToReq=~p (~p bits)",
                  [SID, Req2Count, erlang:bit_size(ToReq2)]),
           comm:send(DestReconPid, {resolve_req, ToReq2, SigSize}),
           
           Params2 = Params1#trivial_recon_struct{db_chunk = gb_trees:empty()},
           shutdown(sync_finished,
                    NewState#rr_recon_state{stats = NewStats2, params = Params2});
       true ->
           NewState#rr_recon_state{stats = NewStats, params = Params1}
    end;

on({reconcile, {get_chunk_response, {RestI, DBList0}}} = _Msg,
   State = #rr_recon_state{stage = reconciliation,     initiator = true,
                           method = bloom,             dhtNodePid = DhtNodePid,
                           params = #bloom_recon_struct{bloom = BF},
                           dest_rr_pid = DestRR_Pid,   stats = Stats,
                           ownerPid = OwnerL}) ->
    ?TRACE1(_Msg, State),
    % no need to map keys since the other node's bloom filter was created with
    % keys mapped to our interval
    Diff = case bloom:item_count(BF) of
               0 -> DBList0;
               _ -> [X || X <- DBList0, not bloom:is_element(BF, X)]
           end,
    %if rest interval is non empty start another sync
    SID = rr_recon_stats:get(session_id, Stats),
    SyncFinished = intervals:is_empty(RestI),
    if not SyncFinished ->
           send_chunk_req(DhtNodePid, self(), RestI, RestI, get_max_items(), reconcile);
       true -> ok
    end,
    ?TRACE("Reconcile Bloom Session=~p ; Diff=~p", [SID, length(Diff)]),
    NewStats =
        case Diff of
            [_|_] ->
                send(DestRR_Pid, {request_resolve, SID,
                                  {?key_upd2, Diff, comm:make_global(OwnerL)},
                                    [{from_my_node, 0}]}),
                % we will get one reply from a subsequent ?key_upd resolve
                rr_recon_stats:inc([{resolve_started, 1}], Stats);
            [] -> Stats
        end,
    NewState = State#rr_recon_state{stats = NewStats},
    if SyncFinished -> shutdown(sync_finished, NewState);
       true         -> NewState
    end;

on({reconcile, {get_chunk_response, {RestI, DBList}}} = _Msg,
   State = #rr_recon_state{stage = reconciliation,     initiator = true,
                           method = RMethod,           params = Params})
  when RMethod =:= merkle_tree orelse RMethod =:= art->
    ?TRACE1(_Msg, State),
    % no need to map keys since the other node's sync struct was created with
    % keys mapped to our interval
    MySyncI = case RMethod of
                  merkle_tree -> Params#merkle_params.interval;
                  art         -> art:get_interval(Params#art_recon_struct.art)
              end,
    build_struct(DBList, MySyncI, RestI, State);

on({crash, _Pid} = _Msg, State) ->
    ?TRACE1(_Msg, State),
    shutdown(recon_node_crash, State);

on({shutdown, Reason}, State) ->
    shutdown(Reason, State);

on({'DOWN', _MonitorRef, process, _Owner, _Info}, _State) ->
    log:log(info, "[ ~p - ~p] shutdown due to rrepair shut down", [?MODULE, comm:this()]),
    kill;

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% trivial reconciliation sync messages
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

on({resolve_req, BinKeys, SigSize} = _Msg,
   State = #rr_recon_state{stage = resolve,           initiator = false,
                           dest_rr_pid = DestRRPid,   ownerPid = OwnerL,
                           kv_list = KVList,          stats = Stats}) ->
    ?TRACE1(_Msg, State),
    ReqSet = decompress_k_list(BinKeys, gb_sets:empty(), SigSize),
    NewStats =
        case gb_sets:is_empty(ReqSet) of
            true -> Stats;
            _ ->
                ReqKeys = [Key || {Key, _Version} <- KVList,
                                  gb_sets:is_member(compress_key(Key, SigSize),
                                                    ReqSet)],
                
                SID = rr_recon_stats:get(session_id, Stats),
                % note: the resolve request was counted at the initiator and
                %       thus from_my_node must be 0 on this node!
                send_local(OwnerL, {request_resolve, SID,
                                    {key_upd_send, DestRRPid, ReqKeys, []},
                                    [{feedback_request, comm:make_global(OwnerL)},
                                     {from_my_node, 0}]}),
                % we will get one reply from a subsequent ?key_upd resolve
                rr_recon_stats:inc([{resolve_started, 1}], Stats)
        end,
    shutdown(sync_finished, State#rr_recon_state{stats = NewStats});

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% merkle tree sync messages
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

on({?check_nodes, SenderPid, ToCheck, SigSize}, State) ->
    NewState = State#rr_recon_state{dest_recon_pid = SenderPid},
    on({?check_nodes, ToCheck, SigSize}, NewState);

on({?check_nodes, ToCheck0, SigSize},
   State = #rr_recon_state{ stage = reconciliation,    initiator = false,
                            struct = Tree,             ownerPid = OwnerL,
                            dest_rr_pid = DestNodePid, stats = Stats,
                            dest_recon_pid = DestReconPid }) ->
    ?ASSERT(comm:is_valid(DestReconPid)),
    ToCheck = merkle_decompress_hashlist(ToCheck0, [], SigSize),
    {Result, RestTree, ToResolve, MaxLeafCount} =
        check_node(ToCheck, Tree, SigSize),
    {Flags, HashKeys} =
        merkle_compress_cmp_result(Result, <<>>, <<>>, SigSize),
    ToResolveLength = length(ToResolve),
    NStats =
        if ToResolveLength > 0 ->
               SID = rr_recon_stats:get(session_id, Stats),
               ResolveCount = resolve_leaves(ToResolve, DestNodePid, SID, OwnerL),
               rr_recon_stats:inc([{tree_leavesSynced, ToResolveLength},
                                   {resolve_started, ResolveCount}], Stats);
           true -> Stats
        end,
    send(DestReconPid, {?check_nodes_response, Flags, HashKeys, MaxLeafCount}),
    State#rr_recon_state{ struct = RestTree, stats = NStats };

on({?check_nodes_response, Flags, HashKeys, OtherMaxLeafCount}, State =
       #rr_recon_state{ stage = reconciliation,        initiator = true,
                        struct = Tree,                 ownerPid = OwnerL,
                        dest_rr_pid = DestNodePid,     stats = Stats,
                        dest_recon_pid = DestReconPid, params = Params,
                        misc = [{signature_size, SigSize}] }) ->
    CmpResults = merkle_decompress_cmp_result(Flags, HashKeys, [], SigSize),
    SID = rr_recon_stats:get(session_id, Stats),
    case process_tree_cmp_result(CmpResults, Tree,
                                 Params#merkle_params.branch_factor,
                                 SigSize, Stats) of
        {[], ToResolve, NStats0, RTree, _MyMaxLeafCount} ->
            NextSigSize = SigSize,
            ok;
        {[_|_] = Req0, ToResolve, NStats0, RTree, MyMaxLeafCount} ->
            % note: we have m one-to-n comparisons, assuming the probability of
            %       a failure in a single one-to-n comparison is p, the overall
            %       p1e = 1 - (1-p)^n  <=>  p = 1 - (1 - p1e)^(1/n)
            P = 1 - math:pow(1 - get_p1e(), 1 / erlang:max(1, length(Req0))),
            NextSigSize = calc_signature_size_1_to_n(
                            erlang:max(MyMaxLeafCount, OtherMaxLeafCount),
                            P, 160),
%%             log:pal("MyMLC: ~p~n    OtherMLC: ~p~n     SigSize: ~p",
%%                     [MyMaxLeafCount, OtherMaxLeafCount, NextSigSize]),
            Req = merkle_compress_hashlist(Req0, <<>>, NextSigSize),
            send(DestReconPid, {?check_nodes, Req, NextSigSize})
    end,
    ResolveCount = resolve_leaves(ToResolve, DestNodePid, SID, OwnerL),
    NStats = rr_recon_stats:inc([{tree_leavesSynced, length(ToResolve)},
                                 {resolve_started, ResolveCount}], NStats0),
    CompLeft = rr_recon_stats:get(tree_compareLeft, NStats),
    NewState = State#rr_recon_state{stats = NStats, struct = RTree,
                                    misc = [{signature_size, NextSigSize}]},
    if CompLeft =:= 0 ->
           send(DestReconPid, {shutdown, sync_finished_remote}),
           shutdown(sync_finished, NewState);
       true -> NewState
    end.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

-spec build_struct(DBList::db_chunk_kv(), DestI::intervals:non_empty_interval(),
                   RestI::intervals:interval(), state()) -> state() | kill.
build_struct(DBList, DestI, RestI,
             State = #rr_recon_state{method = RMethod, params = Params,
                                     struct = OldSyncStruct,
                                     initiator = Initiator, stats = Stats,
                                     dhtNodePid = DhtNodePid, stage = Stage,
                                     kv_list = KVList}) ->
    ?ASSERT(not intervals:is_empty(DestI)),
    % bloom may fork more recon processes if un-synced elements remain
    BeginSync =
        case intervals:is_empty(RestI) of
            false ->
                SubSyncI = map_interval(DestI, RestI),
                case intervals:is_empty(SubSyncI) of
                    false ->
                        MySubSyncI = map_interval(RestI, DestI), % mapped to my range
                        Reconcile =
                            if Initiator andalso (Stage =:= reconciliation) -> reconcile;
                               true -> create_struct
                            end,
                        send_chunk_req(DhtNodePid, self(), MySubSyncI, SubSyncI,
                                       get_max_items(), Reconcile),
                        false;
                    true -> true
                end;
            true -> true
        end,
    NewKVList = lists:append(KVList, DBList),
    if BeginSync ->
           ToBuild = if Initiator andalso RMethod =:= art -> merkle_tree;
                        true -> RMethod
                     end,
           {BuildTime, SyncStruct} =
               util:tc(fun() -> build_recon_struct(ToBuild, OldSyncStruct, DestI,
                                                   NewKVList, Params, BeginSync)
                       end),
           Stats1 = rr_recon_stats:inc([{build_time, BuildTime}], Stats),
           NewState = State#rr_recon_state{struct = SyncStruct, stats = Stats1,
                                           kv_list = NewKVList},
           begin_sync(SyncStruct, Params, NewState#rr_recon_state{stage = reconciliation});
       true ->
           % keep stage (at initiator: reconciliation, at other: build_struct)
           State#rr_recon_state{kv_list = NewKVList}
    end.

-spec begin_sync(MySyncStruct::sync_struct(), OtherSyncStruct::parameters() | {},
                 state()) -> state() | kill.
begin_sync(MySyncStruct, _OtherSyncStruct = {},
           State = #rr_recon_state{method = trivial, initiator = false,
                                   ownerPid = OwnerL, stats = Stats,
                                   dest_rr_pid = DestRRPid}) ->
    ?TRACE("BEGIN SYNC", []),
    SID = rr_recon_stats:get(session_id, Stats),
    send(DestRRPid, {?IIF(SID =:= null, start_recon, continue_recon),
                     comm:make_global(OwnerL), SID,
                     {start_recon, trivial, MySyncStruct}}),
    State#rr_recon_state{struct = {}, stage = resolve};
begin_sync(MySyncStruct, _OtherSyncStruct = {},
           State = #rr_recon_state{method = bloom, initiator = false,
                                   ownerPid = OwnerL, stats = Stats,
                                   dest_rr_pid = DestRRPid}) ->
    ?TRACE("BEGIN SYNC", []),
    SID = rr_recon_stats:get(session_id, Stats),
    send(DestRRPid, {?IIF(SID =:= null, start_recon, continue_recon),
                     comm:make_global(OwnerL), SID,
                     {start_recon, bloom, MySyncStruct}}),
    shutdown(sync_finished, State#rr_recon_state{kv_list = []});
begin_sync(MySyncStruct, _OtherSyncStruct,
           State = #rr_recon_state{method = merkle_tree, initiator = Initiator,
                                   ownerPid = OwnerL, stats = Stats,
                                   dest_recon_pid = DestReconPid,
                                   dest_rr_pid = DestRRPid}) ->
    ?TRACE("BEGIN SYNC", []),
    Stats1 =
        rr_recon_stats:set(
          [{tree_compareLeft, ?IIF(Initiator, 1, 0)},
           {tree_size, merkle_tree:size_detail(MySyncStruct)}], Stats),
    case Initiator of
        true ->
            SigSize = 160,
            Req = merkle_compress_hashlist([merkle_tree:get_root(MySyncStruct)],
                                           <<>>, SigSize),
            send(DestReconPid, {?check_nodes, comm:this(), Req, SigSize}),
            State#rr_recon_state{stats = Stats1,
                                 misc = [{signature_size, SigSize}],
                                 kv_list = []};
        false ->
            MerkleI = merkle_tree:get_interval(MySyncStruct),
            MerkleV = merkle_tree:get_branch_factor(MySyncStruct),
            MerkleB = merkle_tree:get_bucket_size(MySyncStruct),
            MySyncParams = #merkle_params{interval = MerkleI,
                                          branch_factor = MerkleV,
                                          bucket_size = MerkleB},
            SyncParams = MySyncParams#merkle_params{reconPid = comm:this()},
            SID = rr_recon_stats:get(session_id, Stats),
            send(DestRRPid, {?IIF(SID =:= null, start_recon, continue_recon),
                             comm:make_global(OwnerL), SID,
                             {start_recon, merkle_tree, SyncParams}}),
            State#rr_recon_state{stats = Stats1, params = MySyncParams,
                                 kv_list = []}
    end;
begin_sync(MySyncStruct, OtherSyncStruct,
           State = #rr_recon_state{method = art, initiator = Initiator,
                                   ownerPid = OwnerL, stats = Stats,
                                   dest_rr_pid = DestRRPid}) ->
    ?TRACE("BEGIN SYNC", []),
    case Initiator of
        true ->
            Stats1 = art_recon(MySyncStruct, OtherSyncStruct#art_recon_struct.art, State),
            shutdown(sync_finished, State#rr_recon_state{stats = Stats1,
                                                         kv_list = []});
        false ->
            SID = rr_recon_stats:get(session_id, Stats),
            send(DestRRPid, {?IIF(SID =:= null, start_recon, continue_recon),
                             comm:make_global(OwnerL), SID,
                             {start_recon, art, MySyncStruct}}),
            shutdown(sync_finished, State#rr_recon_state{kv_list = []})
    end.

-spec shutdown(exit_reason(), state()) -> kill.
shutdown(Reason, #rr_recon_state{ownerPid = OwnerL, stats = Stats,
                                 initiator = Initiator, dest_rr_pid = DestRR,
                                 dest_recon_pid = DestRC}) ->
    ?TRACE("SHUTDOWN Session=~p Reason=~p",
           [rr_recon_stats:get(session_id, Stats), Reason]),
    Status = exit_reason_to_rc_status(Reason),
    NewStats = rr_recon_stats:set([{status, Status}], Stats),
    send_local(OwnerL, {recon_progress_report, comm:this(), Initiator, DestRR,
                        DestRC, NewStats}),
    kill.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% KV-List compression
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

%% @doc Calculates the minimum number of bits needed to have a hash collision
%%      probability of P1E, given we compare N hashes pairwise with each other.
-spec calc_signature_size_n_pair(N::non_neg_integer(), P1E::float(),
                                 MaxSize::signature_size())
        -> SigSize::signature_size().
calc_signature_size_n_pair(0, P1E, _MaxSize) when P1E > 0 ->
    1;
calc_signature_size_n_pair(1, P1E, _MaxSize) when P1E > 0 ->
    1;
calc_signature_size_n_pair(N, P1E, MaxSize) when P1E > 0 ->
    erlang:min(MaxSize, erlang:max(1, util:ceil(util:log2(N * (N - 1) / P1E) - 1))).

%% @doc Transforms a list of key and version tuples (with unique keys), into a
%%      compact binary representation for transfer.
-spec compress_kv_list(KVList::[{?RT:key(), db_dht:version()}], Bin,
                       SigSize::signature_size(), VSize::signature_size())
        -> Bin when is_subtype(Bin, bitstring()).
compress_kv_list(KVList, Bin, SigSize, VSize) ->
    compress_kv_list_(KVList, Bin, SigSize, VSize, util:pow(2, VSize)).
{'fraction_non_alphanumeric': 0.1397794975738591, 'fraction_numerical': 0.005896443707388981, 'mean_word_length': 2.0761418922110435, 'pattern_counts': {'":': 0, '<': 43, '<?xml version=': 0, '>': 213, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 154, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '206227', 'n_tokens_mistral': 19584, 'n_tokens_neox': 18875, 'n_words': 3989}