anchor.diff:      ?DBG_ASSERT(?implies(RMethod =/= merkle_tree andalso RMethod =/= art,
                            not Initiator)),
       % note: RestI already is a sub-interval of the sync interval
  -    BeginSync =
  -        case intervals:is_empty(RestI) of
  -            false ->
  -                send_chunk_req(pid_groups:get_my(dht_node), self(),
  -                               RestI, get_max_items()),
  -                false;
  -            true -> true
  -        end,
  +    BeginSync = intervals:is_empty(RestI),
       NewKVList = lists:append(KVList, DBList),
       if BeginSync ->
              ToBuild = if Initiator andalso RMethod =:= art -> merkle_tree;
  
positive.code:         0 -> shutdown(sync_finished, State#rr_recon_state{kv_list = []});
          _ -> State#rr_recon_state{struct = {}, stage = resolve}
      end;
  begin_sync(_OtherSyncStruct,
             State = #rr_recon_state{method = merkle_tree, initiator = Initiator,
                                     struct = MySyncStruct,
                                     ownerPid = OwnerL, stats = Stats,
                                     params = Params,
                                     dest_recon_pid = DestReconPid,
                                     dest_rr_pid = DestRRPid}) ->
      ?TRACE("BEGIN SYNC", []),
      case Initiator of
          true ->
              MTSize = merkle_tree:size_detail(MySyncStruct),
              Stats1 = rr_recon_stats:set([{tree_size, MTSize}], Stats),
              #merkle_params{p1e = P1ETotal, num_trees = NumTrees,
                             ni_item_count = OtherItemsCount} = Params,
              MyItemCount = lists:max([0 | [merkle_tree:get_item_count(Node) || Node <- MySyncStruct]]),
              ?MERKLE_DEBUG("merkle (I) - CurrentNodes: ~B~n"
                            "Inner/Leaf/Items: ~p, EmptyLeaves: ~B",
                            [length(MySyncStruct), MTSize,
                             length([ok || L <- merkle_tree:get_leaves(MySyncStruct),
                                           merkle_tree:is_empty(L)])]),
              P1ETotal2 = calc_n_subparts_p1e(2, P1ETotal),
              P1ETotal3 = calc_n_subparts_p1e(NumTrees, P1ETotal2),
  
              {_P1E_I, _P1E_L, NextSigSizeI, NextSigSizeL, EffectiveP1E_I, EffectiveP1E_L} =
                  merkle_next_signature_sizes(Params, P1ETotal3, MyItemCount,
                                              OtherItemsCount),
  
              Req = merkle_compress_hashlist(MySyncStruct, <<>>, NextSigSizeI, NextSigSizeL),
              send(DestReconPid, {?check_nodes, comm:this(), Req, MyItemCount}),
              State#rr_recon_state{stats = Stats1,
                                   misc = [{signature_size, {NextSigSizeI, NextSigSizeL}},
                                           {p1e, {EffectiveP1E_I, EffectiveP1E_L}},
                                           {p1e_phase_x, P1ETotal2},
                                           {icount, MyItemCount},
                                           {oicount, OtherItemsCount}],
                                   kv_list = []};
          false ->
              % tell the initiator to create its struct first, and then build ours
              % (at this stage, we do not have any data in the merkle tree yet!)
              DBItems = State#rr_recon_state.kv_list,
              MerkleI = merkle_tree:get_interval(MySyncStruct),
              MerkleV = merkle_tree:get_branch_factor(MySyncStruct),
              MerkleB = merkle_tree:get_bucket_size(MySyncStruct),
              NumTrees = get_merkle_num_trees(),
              P1ETotal = get_p1e(),
              P1ETotal2 = calc_n_subparts_p1e(2, P1ETotal),
              
              % split interval first and create NumTrees merkle trees later
              {BuildTime1, ICBList} =
                  util:tc(fun() ->
                                  merkle_tree:keys_to_intervals(
                                    DBItems, intervals:split(MerkleI, NumTrees))
                          end),
              ItemCount = lists:max([0 | [Count || {_SubI, Count, _Bucket} <- ICBList]]),
              P1ETotal3 = calc_n_subparts_p1e(NumTrees, P1ETotal2),
  
              MySyncParams = #merkle_params{interval = MerkleI,
                                            branch_factor = MerkleV,
                                            bucket_size = MerkleB,
                                            num_trees = NumTrees,
                                            p1e = P1ETotal,
                                            ni_item_count = ItemCount},
              SyncParams = MySyncParams#merkle_params{reconPid = comm:this()},
              SID = rr_recon_stats:get(session_id, Stats),
              send(DestRRPid, {continue_recon, comm:make_global(OwnerL), SID,
                               {start_recon, merkle_tree, SyncParams}}),
              
              % finally create the real merkle tree containing data
              % -> this way, the initiator can create its struct in parallel!
              {BuildTime2, SyncStruct} =
                  util:tc(fun() ->
                                  [merkle_tree:get_root(
                                     merkle_tree:new(SubI, Bucket,
                                                     [{keep_bucket, true},
                                                      {branch_factor, MerkleV},
                                                      {bucket_size, MerkleB}]))
                                     || {SubI, _Count, Bucket} <- ICBList]
                          end),
              MTSize = merkle_tree:size_detail(SyncStruct),
              Stats1 = rr_recon_stats:set([{tree_size, MTSize}], Stats),
              Stats2 = rr_recon_stats:inc([{build_time, BuildTime1 + BuildTime2}], Stats1),
              ?MERKLE_DEBUG("merkle (NI) - CurrentNodes: ~B~n"
                            "Inner/Leaf/Items: ~p, EmptyLeaves: ~B",
                            [length(SyncStruct), MTSize,
                             length([ok || L <- merkle_tree:get_leaves(SyncStruct),
                                           merkle_tree:is_empty(L)])]),
              
              State#rr_recon_state{struct = SyncStruct,
                                   stats = Stats2, params = MySyncParams,
                                   misc = [{p1e, P1ETotal3},
                                           {p1e_phase_x, P1ETotal2},
                                           {icount, ItemCount}],
                                   kv_list = []}
      end;
  begin_sync(OtherSyncStruct,
             State = #rr_recon_state{method = art, initiator = Initiator,
                                     struct = MySyncStruct,
                                     ownerPid = OwnerL, stats = Stats,
                                     dest_recon_pid = DestReconPid,
                                     dest_rr_pid = DestRRPid}) ->
      ?TRACE("BEGIN SYNC", []),
      case Initiator of
          true ->
              ART = OtherSyncStruct#art_recon_struct.art,
              Stats1 = rr_recon_stats:set(
                         [{tree_size, merkle_tree:size_detail(MySyncStruct)}], Stats),
              OtherItemCount = art:get_property(ART, items_count),
              case merkle_tree:get_interval(MySyncStruct) =:= art:get_interval(ART) of
                  true ->
                      {ASyncLeafs, NComp, NSkip, NLSync} =
                          art_get_sync_leaves([merkle_tree:get_root(MySyncStruct)], ART,
                                              [], 0, 0, 0),
                      Diff = lists:append([merkle_tree:get_bucket(N) || N <- ASyncLeafs]),
                      MyItemCount = merkle_tree:get_item_count(MySyncStruct),
                      P1E = get_p1e(),
                      % TODO: correctly calculate the probabilities and select appropriate parameters beforehand
                      P1E_p1_0 = bloom_worst_case_failprob(
                                   art:get_property(ART, leaf_bf), MyItemCount),
                      P1E_p1 = if P1E_p1_0 == 1 ->
                                      log:log("~w: [ ~p:~.0p ] P1E constraint broken (phase 1 overstepped?)"
                                              " - continuing with smallest possible failure probability",
                                              [?MODULE, pid_groups:my_groupname(), self()]),
                                      1 - 1.0e-16;
                                  true -> P1E_p1_0
                               end,
                      Stats2  = rr_recon_stats:set([{p1e_phase1, P1E_p1}], Stats1),
                      % NOTE: use left-over P1E after phase 1 (ART) for phase 2 (trivial RC)
                      P1E_p2 = calc_n_subparts_p1e(1, P1E, (1 - P1E_p1)),
                      
                      Stats3 = rr_recon_stats:inc([{tree_nodesCompared, NComp},
                                                   {tree_compareSkipped, NSkip},
                                                   {tree_leavesSynced, NLSync}], Stats2),
                      
                      phase2_run_trivial_on_diff(Diff, none, [], OtherItemCount,
                                                 P1E_p2, OtherItemCount,
                                                 State#rr_recon_state{stats = Stats3});
                  false when OtherItemCount =/= 0 ->
                      % must send resolve_req message for the non-initiator to shut down
                      send(DestReconPid, {resolve_req, shutdown}),
                      shutdown(sync_finished, State#rr_recon_state{stats = Stats1,
                                                                   kv_list = []});
                  false ->
                      shutdown(sync_finished, State#rr_recon_state{stats = Stats1,
                                                                   kv_list = []})
              end;
          false ->
              SID = rr_recon_stats:get(session_id, Stats),
              send(DestRRPid, {continue_recon, comm:make_global(OwnerL), SID,
                               {start_recon, art, MySyncStruct}}),
              case art:get_property(MySyncStruct#art_recon_struct.art, items_count) of
                  0 -> shutdown(sync_finished, State#rr_recon_state{kv_list = []});
                  _ -> State#rr_recon_state{struct = {}, stage = resolve}
              end
      end.
  
  -spec shutdown(exit_reason(), state()) -> kill.
  
negative.code:                                            kv_list = NewKVList},
             begin_sync(Params, NewState#rr_recon_state{stage = reconciliation});
         true ->
             send_chunk_req(pid_groups:get_my(dht_node), self(),
                            RestI, get_max_items()),
             % keep stage (at initiator: reconciliation, at other: build_struct)
             State#rr_recon_state{kv_list = NewKVList}
      end.
  