<commit_before>%  @copyright 2010-2012 Zuse Institute Berlin
%  @end
%
%   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.
%%%-------------------------------------------------------------------
%%% File    rrepair_SUITE.erl
%%% @author <NAME> <<EMAIL>
%%% @doc    Tests for rep update module.
%%% @end
%%% Created : 2011-05-27
%%%-------------------------------------------------------------------
%% @version $Id $

-module(rrepair_SUITE).

-author('<EMAIL>').

-compile(export_all).

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

-define(REP_FACTOR, 4).
-define(DBSizeKey, rrepair_SUITE_dbsize).    %Process Dictionary Key for generated db size

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

basic_tests() ->
    [get_symmetric_keys_test,
     blobCoding,
     tester_get_key_quadrant,
     tester_mapInterval,
     tester_minKeyInInterval
    ].

repair_tests() ->
    [no_diff,        % ring is not out of sync e.g. no outdated or missing replicas
     one_node,       % sync in ring with only one node
     %mpath
     dest,           % run one sync with a specified dest node 
     simple,         % run one sync round
     multi_round     % run multiple sync rounds
	].

bloom_tests() ->    
    repair_tests() ++ 
        [parts]. % get_chunk with limited items / leads to multiple bloom filters

all() ->
    [{group, basic},
     {group, repair}
     ].

groups() ->
    [{basic,  [parallel], basic_tests()},
     {repair, [sequence], [{upd_bloom,    [sequence], bloom_tests()}, %{repeat_until_any_fail, 1000}
                           {upd_merkle,   [sequence], repair_tests()},
                           {upd_art,      [sequence], repair_tests()},
                           {regen_bloom,  [sequence], bloom_tests()},
                           {regen_merkle, [sequence], repair_tests()},
                           {regen_art,    [sequence], repair_tests()},
                           {mixed_bloom,  [sequence], bloom_tests()}, 
                           {mixed_merkle, [sequence], repair_tests()},
                           {mixed_art,    [sequence], repair_tests()}]}
    ].

suite() ->
    [
     {timetrap, {seconds, 240}}
    ].

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

init_per_suite(Config) ->
    _ = crypto:start(),    
    unittest_helper:init_per_suite(Config).

end_per_suite(Config) ->
    erlang:erase(?DBSizeKey),
    crypto:stop(),
    _ = unittest_helper:end_per_suite(Config),
    ok.

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

init_per_group(Group, Config) ->
    ct:comment(io_lib:format("BEGIN ~p", [Group])),
    case Group of
        upd_bloom -> [{ru_method, bloom}, {ftype, update}];
        upd_merkle -> [{ru_method, merkle_tree}, {ftype, update}];
        upd_art -> [{ru_method, art}, {ftype, update}];
        regen_bloom -> [{ru_method, bloom}, {ftype, regen}];
        regen_merkle -> [{ru_method, merkle_tree}, {ftype, regen}];
        regen_art -> [{ru_method, art}, {ftype, regen}];
        mixed_bloom -> [{ru_method, bloom}, {ftype, mixed}];
        mixed_merkle -> [{ru_method, merkle_tree}, {ftype, mixed}];
        mixed_art -> [{ru_method, art}, {ftype, mixed}];
        _ -> []
    end ++ Config.

end_per_group(Group, Config) ->  
    Method = proplists:get_value(ru_method, Config, undefined),
    FType = proplists:get_value(ftype, Config, undefined),
    case Method of
        undefined -> ct:comment(io_lib:format("END ~p", [Group]));
        M -> ct:comment(io_lib:format("END ~p/~p", [FType, M]))
    end,
    proplists:delete(ru_method, Config).

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

end_per_testcase(_TestCase, _Config) ->
    unittest_helper:stop_ring(),
    ok.

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

get_rep_upd_config(Method) ->
    [{rrepair_enabled, true},
     {rr_trigger, trigger_periodic},
     {rr_trigger_interval, 100000000}, %stop trigger
     {rr_recon_method, Method},
     {rr_session_ttl, 100000},
     {rr_gc_interval, 60000},
     {rr_bloom_fpr, 0.1},
	 {rr_trigger_probability, 100},
     {rr_max_items, 10000},
     {rr_art_inner_fpr, 0.01},
     {rr_art_leaf_fpr, 0.1},
     {rr_art_correction_factor, 2},
     {rr_merkle_branch_factor, 2},
     {rr_merkle_bucket_size, 25}].    

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Replica Update tests
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

no_diff(Config) ->
    Method = proplists:get_value(ru_method, Config),
    FType = proplists:get_value(ftype, Config),
    {Start, End} = start_sync(Config, 4, 1000, [{fprob, 0}, {ftype, FType}], 
                              1, 0.1, get_rep_upd_config(Method)),
    ?assert(sync_degree(Start) =:= sync_degree(End)).

one_node(Config) ->
    Method = proplists:get_value(ru_method, Config),
    FType = proplists:get_value(ftype, Config),
    {Start, End} = start_sync(Config, 1, 1, [{fprob, 50}, {ftype, FType}], 
                              1, 0.2, get_rep_upd_config(Method)),
    ?assert(sync_degree(Start) =:= sync_degree(End)).    

mpath_map({request_resolve, _, {key_upd, L}, _}) ->
    {key_upd, length(L)};
mpath_map(Msg) ->
    {element(1, Msg)}.

mpath(Config) ->
	%parameter
    NodeCount = 4,
    DataCount = 1000,
    Fpr = 0.1,
    Method = proplists:get_value(ru_method, Config),
    FType = proplists:get_value(ftype, Config),
	TraceName = erlang:list_to_atom(atom_to_list(Method)++atom_to_list(FType)),
    %build and fill ring
    build_symmetric_ring(NodeCount, Config, get_rep_upd_config(Method)),
    config:write(rr_bloom_fpr, Fpr),
    db_generator:fill_ring(random, DataCount, [{ftype, FType}, 
                                               {fprob, 50}, 
                                               {distribution, uniform}]),
    %chose node pair    
    SKey = ?RT:get_random_node_id(),
    CKey = util:randomelem(lists:delete(SKey, ?RT:get_replica_keys(SKey))),
    %server starts sync
	%trace_mpath:start(TraceName, fun mpath_map/1),
    trace_mpath:start(TraceName),
    api_dht_raw:unreliable_lookup(SKey, {send_to_group_member, rrepair, 
                                              {request_sync, Method, CKey}}),
    %waitForSyncRoundEnd(NodeKeys),
	timer:sleep(3000),
	trace_mpath:stop(),
	%TRACE
	A = trace_mpath:get_trace(TraceName),
    trace_mpath:cleanup(TraceName),
	B = [X || X = {log_send, _Time, _, 
				   {{_FIP,_FPort,_FPid}, _FName}, 
				   {{_TIP,_TPort,_TPid}, _TName}, 
				   _Msg} <- A], 
	file:write_file("TRACE_" ++ atom_to_list(TraceName) ++ ".txt", io_lib:fwrite("~.0p\n", [B])), 
	file:write_file("TRACE_HISTO_" ++ atom_to_list(TraceName) ++ ".txt", io_lib:fwrite("~.0p\n", [trace_mpath:send_histogram(A)])),
    %file:write_file("TRACE_EVAL_" ++ atom_to_list(TraceName) ++ ".txt", io_lib:fwrite("~.0p\n", [eval_admin:get_bandwidth(A)])),
    %clean up
    unittest_helper:stop_ring(),    
	ok.

simple(Config) ->
    Method = proplists:get_value(ru_method, Config),
    FType = proplists:get_value(ftype, Config),
    {Start, End} = start_sync(Config, 4, 1000, [{fprob, 10}, {ftype, FType}], 
                              1, 0.1, get_rep_upd_config(Method)),
    ?assert(sync_degree(Start) < sync_degree(End)).

multi_round(Config) ->
    Method = proplists:get_value(ru_method, Config),
    FType = proplists:get_value(ftype, Config),
    {Start, End} = start_sync(Config, 4, 1000, [{fprob, 10}, {ftype, FType}], 
                              3, 0.1, get_rep_upd_config(Method)),
    ?assert(sync_degree(Start) < sync_degree(End)). 

dest(Config) ->
    %parameter
    NodeCount = 7,
    DataCount = 1000,
    Fpr = 0.1,
    Method = proplists:get_value(ru_method, Config),
    FType = proplists:get_value(ftype, Config),
    %build and fill ring
    build_symmetric_ring(NodeCount, Config, get_rep_upd_config(Method)),
    config:write(rr_bloom_fpr, Fpr),
    db_generator:fill_ring(random, DataCount, [{ftype, FType}, 
                                               {fprob, 50}, 
                                               {distribution, uniform}]),
    %chose node pair    
    SKey = ?RT:get_random_node_id(),
    CKey = util:randomelem(lists:delete(SKey, ?RT:get_replica_keys(SKey))),
    %measure initial sync degree
    SO = count_outdated(SKey),
    SM = count_dbsize(SKey),
    CO = count_outdated(CKey),
    CM = count_dbsize(CKey),
    %server starts sync
    api_dht_raw:unreliable_lookup(SKey, {send_to_group_member, rrepair, 
                                              {request_sync, Method, CKey}}),
    %waitForSyncRoundEnd(NodeKeys),
    waitForSyncRoundEnd([SKey, CKey]),
    %measure sync degree
    SONew = count_outdated(SKey),
    SMNew = count_dbsize(SKey),
    CONew = count_outdated(CKey),
    CMNew = count_dbsize(CKey),
    ct:pal("SYNC RUN << ~p / ~p >>~nServerKey=~p~nClientKey=~p~n"
           "Server Outdated=[~p -> ~p] DBSize=[~p -> ~p] - Upd=~p ; Regen=~p~n"
           "Client Outdated=[~p -> ~p] DBSize=[~p -> ~p] - Upd=~p ; Regen=~p", 
           [Method, FType, SKey, CKey, 
            SO, SONew, SM, SMNew, SO - SONew, SMNew - SM,
            CO, CONew, CM, CMNew, CO - CONew, CMNew - CM]),
    %clean up
    unittest_helper:stop_ring(),
    ?implies(SO > 0 orelse CO > 0, SONew < SO orelse CONew < CO) andalso
        ?implies(SM =/= SMNew, SMNew > SM) andalso
        ?implies(CM =/= CMNew, CMNew > CM).

parts(Config) ->
    Method = proplists:get_value(ru_method, Config),
    FType = proplists:get_value(ftype, Config),
    OldConf = get_rep_upd_config(Method),
    Conf = lists:keyreplace(rr_max_items, 1, OldConf, {rr_max_items, 500}),    
    {Start, End} = start_sync(Config, 4, 1000, [{fprob, 100}, {ftype, FType}], 
                              1, 0.1, Conf),
    ?assert(sync_degree(Start) < sync_degree(End)).

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Basic Functions Group
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% 

get_symmetric_keys_test(Config) ->
    Conf2 = unittest_helper:start_minimal_procs(Config, [], true),
    ToTest = lists:sort(get_symmetric_keys(4)),
    ToBe = lists:sort(?RT:get_replica_keys(?MINUS_INFINITY)),
    unittest_helper:stop_minimal_procs(Conf2),
    ?equals_w_note(ToTest, ToBe, 
                   io_lib:format("GenKeys=~w~nRTKeys=~w", [ToTest, ToBe])),
    ok.

blobCoding(_) ->
    A = 180000001,
    B = 4,
    Coded = rr_recon:encodeBlob(A, B),
    {DA, DB} = rr_recon:decodeBlob(Coded),
    ?equals_w_note(A, DA, io_lib:format("A=~p ; Coded=~p ; DecodedA=~p", [A, Coded, DA])),
    ?equals_w_note(B, DB, io_lib:format("B=~p ; Coded=~p ; DecodedB=~p", [B, Coded, DB])),
    ok.

-spec prop_get_key_quadrant(?RT:key()) -> boolean().
prop_get_key_quadrant(Key) ->
    Q = rr_recon:get_key_quadrant(Key),
    QI = intervals:split(intervals:all(), 4),
    {TestStatus, TestQ} = 
        lists:foldl(fun(I, {Status, Nr} = Acc) ->
                            case intervals:in(Key, I) of
                                true when Status =:= no -> {yes, Nr};
                                false when Status =:= no -> {no, Nr + 1};<commit_msg>rrepair_SUITE: added session_ttl test to check session garbage collection<commit_after>%  @copyright 2010-2012 Zuse Institute Berlin
%  @end
%
%   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.
%%%-------------------------------------------------------------------
%%% File    rrepair_SUITE.erl
%%% @author <NAME> <<EMAIL>
%%% @doc    Tests for rep update module.
%%% @end
%%% Created : 2011-05-27
%%%-------------------------------------------------------------------
%% @version $Id $

-module(rrepair_SUITE).

-author('<EMAIL>').

-compile(export_all).

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

-define(REP_FACTOR, 4).
-define(DBSizeKey, rrepair_SUITE_dbsize).    %Process Dictionary Key for generated db size

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

basic_tests() ->
    [get_symmetric_keys_test,
     blobCoding,     
     tester_get_key_quadrant,
     tester_mapInterval,
     tester_minKeyInInterval     
    ].

repair_tests() ->
    [no_diff,        % ring is not out of sync e.g. no outdated or missing replicas
     one_node,       % sync in ring with only one node
     %mpath
     dest,           % run one sync with a specified dest node 
     simple,         % run one sync round
     multi_round     % run multiple sync rounds
	].

bloom_tests() ->    
    repair_tests() ++ 
        [parts]. % get_chunk with limited items / leads to multiple bloom filters

all() ->
    [{group, basic},
     session_ttl,
     {group, repair}     
     ].

groups() ->
    [{basic,  [parallel], basic_tests()},
     {repair, [sequence], [{upd_bloom,    [sequence], bloom_tests()}, %{repeat_until_any_fail, 1000}
                           {upd_merkle,   [sequence], repair_tests()},
                           {upd_art,      [sequence], repair_tests()},
                           {regen_bloom,  [sequence], bloom_tests()},
                           {regen_merkle, [sequence], repair_tests()},
                           {regen_art,    [sequence], repair_tests()},
                           {mixed_bloom,  [sequence], bloom_tests()}, 
                           {mixed_merkle, [sequence], repair_tests()},
                           {mixed_art,    [sequence], repair_tests()}]}
    ].

suite() ->
    [
     {timetrap, {seconds, 240}}
    ].

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

init_per_suite(Config) ->
    _ = crypto:start(),    
    unittest_helper:init_per_suite(Config).

end_per_suite(Config) ->
    erlang:erase(?DBSizeKey),
    crypto:stop(),
    _ = unittest_helper:end_per_suite(Config),
    ok.

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

init_per_group(Group, Config) ->
    ct:comment(io_lib:format("BEGIN ~p", [Group])),
    case Group of
        upd_bloom -> [{ru_method, bloom}, {ftype, update}];
        upd_merkle -> [{ru_method, merkle_tree}, {ftype, update}];
        upd_art -> [{ru_method, art}, {ftype, update}];
        regen_bloom -> [{ru_method, bloom}, {ftype, regen}];
        regen_merkle -> [{ru_method, merkle_tree}, {ftype, regen}];
        regen_art -> [{ru_method, art}, {ftype, regen}];
        mixed_bloom -> [{ru_method, bloom}, {ftype, mixed}];
        mixed_merkle -> [{ru_method, merkle_tree}, {ftype, mixed}];
        mixed_art -> [{ru_method, art}, {ftype, mixed}];
        _ -> []
    end ++ Config.

end_per_group(Group, Config) ->  
    Method = proplists:get_value(ru_method, Config, undefined),
    FType = proplists:get_value(ftype, Config, undefined),
    case Method of
        undefined -> ct:comment(io_lib:format("END ~p", [Group]));
        M -> ct:comment(io_lib:format("END ~p/~p", [FType, M]))
    end,
    proplists:delete(ru_method, Config).

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

end_per_testcase(_TestCase, _Config) ->
    unittest_helper:stop_ring(),
    ok.

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

get_rep_upd_config(Method) ->
    [{rrepair_enabled, true},
     {rr_trigger, trigger_periodic},
     {rr_trigger_interval, 100000000}, %stop trigger
     {rr_recon_method, Method},
     {rr_session_ttl, 100000},
     {rr_gc_interval, 60000},
     {rr_bloom_fpr, 0.1},
	 {rr_trigger_probability, 100},
     {rr_max_items, 10000},
     {rr_art_inner_fpr, 0.01},
     {rr_art_leaf_fpr, 0.1},
     {rr_art_correction_factor, 2},
     {rr_merkle_branch_factor, 2},
     {rr_merkle_bucket_size, 25}].    

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Replica Update tests
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

no_diff(Config) ->
    Method = proplists:get_value(ru_method, Config),
    FType = proplists:get_value(ftype, Config),
    {Start, End} = start_sync(Config, 4, 1000, [{fprob, 0}, {ftype, FType}], 
                              1, 0.1, get_rep_upd_config(Method)),
    ?assert(sync_degree(Start) =:= sync_degree(End)).

one_node(Config) ->
    Method = proplists:get_value(ru_method, Config),
    FType = proplists:get_value(ftype, Config),
    {Start, End} = start_sync(Config, 1, 1, [{fprob, 50}, {ftype, FType}], 
                              1, 0.2, get_rep_upd_config(Method)),
    ?assert(sync_degree(Start) =:= sync_degree(End)).    

mpath_map({request_resolve, _, {key_upd, L}, _}) ->
    {key_upd, length(L)};
mpath_map(Msg) ->
    {element(1, Msg)}.

mpath(Config) ->
	%parameter
    NodeCount = 4,
    DataCount = 1000,
    Fpr = 0.1,
    Method = proplists:get_value(ru_method, Config),
    FType = proplists:get_value(ftype, Config),
	TraceName = erlang:list_to_atom(atom_to_list(Method)++atom_to_list(FType)),
    %build and fill ring
    build_symmetric_ring(NodeCount, Config, get_rep_upd_config(Method)),
    config:write(rr_bloom_fpr, Fpr),
    db_generator:fill_ring(random, DataCount, [{ftype, FType}, 
                                               {fprob, 50}, 
                                               {distribution, uniform}]),
    %chose node pair    
    SKey = ?RT:get_random_node_id(),
    CKey = util:randomelem(lists:delete(SKey, ?RT:get_replica_keys(SKey))),
    %server starts sync
	%trace_mpath:start(TraceName, fun mpath_map/1),
    trace_mpath:start(TraceName),
    api_dht_raw:unreliable_lookup(SKey, {send_to_group_member, rrepair, 
                                              {request_sync, Method, CKey}}),
    %waitForSyncRoundEnd(NodeKeys),
	timer:sleep(3000),
	trace_mpath:stop(),
	%TRACE
	A = trace_mpath:get_trace(TraceName),
    trace_mpath:cleanup(TraceName),
	B = [X || X = {log_send, _Time, _, 
				   {{_FIP,_FPort,_FPid}, _FName}, 
				   {{_TIP,_TPort,_TPid}, _TName}, 
				   _Msg} <- A], 
	file:write_file("TRACE_" ++ atom_to_list(TraceName) ++ ".txt", io_lib:fwrite("~.0p\n", [B])), 
	file:write_file("TRACE_HISTO_" ++ atom_to_list(TraceName) ++ ".txt", io_lib:fwrite("~.0p\n", [trace_mpath:send_histogram(A)])),
    %file:write_file("TRACE_EVAL_" ++ atom_to_list(TraceName) ++ ".txt", io_lib:fwrite("~.0p\n", [eval_admin:get_bandwidth(A)])),  
	ok.

simple(Config) ->
    Method = proplists:get_value(ru_method, Config),
    FType = proplists:get_value(ftype, Config),
    {Start, End} = start_sync(Config, 4, 1000, [{fprob, 10}, {ftype, FType}], 
                              1, 0.1, get_rep_upd_config(Method)),
    ?assert(sync_degree(Start) < sync_degree(End)).

multi_round(Config) ->
    Method = proplists:get_value(ru_method, Config),
    FType = proplists:get_value(ftype, Config),
    {Start, End} = start_sync(Config, 4, 1000, [{fprob, 10}, {ftype, FType}], 
                              3, 0.1, get_rep_upd_config(Method)),
    ?assert(sync_degree(Start) < sync_degree(End)). 

dest(Config) ->
    %parameter
    NodeCount = 7,
    DataCount = 1000,
    Fpr = 0.1,
    Method = proplists:get_value(ru_method, Config),
    FType = proplists:get_value(ftype, Config),
    %build and fill ring
    build_symmetric_ring(NodeCount, Config, get_rep_upd_config(Method)),
    config:write(rr_bloom_fpr, Fpr),
    db_generator:fill_ring(random, DataCount, [{ftype, FType}, 
                                               {fprob, 50}, 
                                               {distribution, uniform}]),
    %chose node pair    
    SKey = ?RT:get_random_node_id(),
    CKey = util:randomelem(lists:delete(SKey, ?RT:get_replica_keys(SKey))),
    %measure initial sync degree
    SO = count_outdated(SKey),
    SM = count_dbsize(SKey),
    CO = count_outdated(CKey),
    CM = count_dbsize(CKey),
    %server starts sync
    api_dht_raw:unreliable_lookup(SKey, {send_to_group_member, rrepair, 
                                              {request_sync, Method, CKey}}),
    %waitForSyncRoundEnd(NodeKeys),
    waitForSyncRoundEnd([SKey, CKey]),
    %measure sync degree
    SONew = count_outdated(SKey),
    SMNew = count_dbsize(SKey),
    CONew = count_outdated(CKey),
    CMNew = count_dbsize(CKey),
    ct:pal("SYNC RUN << ~p / ~p >>~nServerKey=~p~nClientKey=~p~n"
           "Server Outdated=[~p -> ~p] DBSize=[~p -> ~p] - Upd=~p ; Regen=~p~n"
           "Client Outdated=[~p -> ~p] DBSize=[~p -> ~p] - Upd=~p ; Regen=~p", 
           [Method, FType, SKey, CKey, 
            SO, SONew, SM, SMNew, SO - SONew, SMNew - SM,
            CO, CONew, CM, CMNew, CO - CONew, CMNew - CM]),
    %clean up
    ?implies(SO > 0 orelse CO > 0, SONew < SO orelse CONew < CO) andalso
        ?implies(SM =/= SMNew, SMNew > SM) andalso
        ?implies(CM =/= CMNew, CMNew > CM).

parts(Config) ->
    Method = proplists:get_value(ru_method, Config),
    FType = proplists:get_value(ftype, Config),
    OldConf = get_rep_upd_config(Method),
    Conf = lists:keyreplace(rr_max_items, 1, OldConf, {rr_max_items, 500}),    
    {Start, End} = start_sync(Config, 4, 1000, [{fprob, 100}, {ftype, FType}], 
                              2, 0.2, Conf),
    ?assert(sync_degree(Start) < sync_degree(End)).

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Basic Functions Group
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% 

get_symmetric_keys_test(Config) ->
    Conf2 = unittest_helper:start_minimal_procs(Config, [], true),
    ToTest = lists:sort(get_symmetric_keys(4)),
    ToBe = lists:sort(?RT:get_replica_keys(?MINUS_INFINITY)),
    unittest_helper:stop_minimal_procs(Conf2),
    ?equals_w_note(ToTest, ToBe, 
                   io_lib:format("GenKeys=~w~nRTKeys=~w", [ToTest, ToBe])),
    ok.

blobCoding(_) ->
    A = 180000001,
    B = 4,
    Coded = rr_recon:encodeBlob(A, B),
    {DA, DB} = rr_recon:decodeBlob(Coded),
    ?equals_w_note(A, DA, io_lib:format("A=~p ; Coded=~p ; DecodedA=~p", [A, Coded, DA])),
    ?equals_w_note(B, DB, io_lib:format("B=~p ; Coded=~p ; DecodedB=~p", [B, Coded, DB])),
    ok.

session_ttl(Config) ->
    %parameter
    NodeCount = 7,
    DataCount = 1000,
    Method = bloom,
    FType = mixed,
    TTL = 2500,
    
    _R1 = get_rep_upd_config(Method),
    _R2 = lists:keyreplace(rr_session_ttl, 1, _R1, {rr_session_ttl, TTL}),
    RRConf = lists:keyreplace(rr_gc_interval, 1, _R2, {rr_gc_interval, erlang:round(TTL / 10)}),
    
    %build and fill ring
    build_symmetric_ring(NodeCount, Config, RRConf),    
    db_generator:fill_ring(random, DataCount, [{ftype, FType}, 
                                               {fprob, 50}, 
                                               {distribution, uniform}]),
    %chose node pair
    SKey = ?RT:get_random_node_id(),
    CKey = util:randomelem(lists:delete(SKey, ?RT:get_replica_keys(SKey))),
    
    api_dht_raw:unreliable_lookup(CKey, {get_pid_group, comm:this()}),    
    CName = receive {get_pid_group_response, Key} -> Key end,
    
    %server starts sync
    api_dht_raw:unreliable_lookup(SKey, {send_to_group_member, rrepair, 
                                              {request_sync, Method, CKey}}),
    api_vm:kill_node(CName),

    %check timeout
    Req = {send_to_group_member, rrepair, {get_state, comm:this(), open_sessions}},
    api_dht_raw:unreliable_lookup(SKey, Req),
    Open = receive {get_state_response, R1} -> R1 =/= 0 end,
    timer:sleep(TTL),
    api_dht_raw:unreliable_lookup(SKey, Req),
    Open2 = receive {get_state_response, R2} -> R2 =:= 0 end,
    ?equals_pattern_w_note(Open, Open2, io_lib:format("start session open = ~p - end session closed = ~p", [Open, Open2])),
    ok.

-spec prop_get_key_quadrant(?RT:key()) -> boolean().
prop_get_key_quadrant(Key) ->
    Q = rr_recon:get_key_quadrant(Key),
    QI = intervals:split(intervals:all(), 4),
    {TestStatus, TestQ} = 
        lists:foldl(fun(I, {Status, Nr} = Acc) ->
                            case intervals:in(Key, I) of
                                true when Status =:= no -> {yes, Nr};
                                false when Status =:= no -> {no, Nr + 1};
{'fraction_non_alphanumeric': 0.20267322413040892, 'fraction_numerical': 0.013961051838337365, 'mean_word_length': 2.9225264467952705, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 113, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 42, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4403481', 'n_tokens_mistral': 8912, 'n_tokens_neox': 8416, 'n_words': 1661}