%%%------------------------------------------------------------------- %%% @doc Gossip protocol for CRDT state replication. %%% %%% Implements a gossip-based protocol for eventually-consistent state %%% synchronization across nodes. Uses push-pull-push anti-entropy: %%% %%% - Push: Periodically sends local state changes to random peers %%% - Pull: Requests state from peers when needed %%% - Anti-entropy: Full state synchronization to repair divergence %%% %%% Configuration Parameters: %%% - push_interval: 1000ms (how often to push to peers) %%% - anti_entropy_interval: 30000ms (how often to run anti-entropy) %%% - fanout: 3 (number of peers to contact per round) %%% %%% @end %%%------------------------------------------------------------------- -module(macula_gossip). -behaviour(gen_server). -include_lib("kernel/include/logger.hrl"). %% API -export([ start_link/1, stop/1, %% State management put/4, get/2, delete/2, get_all/1, %% Gossip control push_state/2, pull_state/2, anti_entropy/1, force_sync/1, %% Peer management add_peer/2, remove_peer/2, get_peers/1, %% Metrics get_stats/1 ]). %% gen_server callbacks -export([ init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2 ]). %% Gossip message handlers (called by gateway/connection) -export([ handle_gossip_push/2, handle_gossip_pull/2, handle_gossip_pull_reply/2, handle_gossip_sync/2, handle_gossip_sync_reply/2 ]). -define(DEFAULT_PUSH_INTERVAL, 1000). % 1 second -define(DEFAULT_ANTI_ENTROPY_INTERVAL, 30000). % 30 seconds -define(DEFAULT_FANOUT, 3). -define(MAX_STATES_PER_MESSAGE, 100). -record(state, { node_id :: binary(), realm :: binary(), %% Local CRDT states: #{Key => {Type, CrdtState, VectorClock}} states :: #{binary() => {atom(), term(), map()}}, %% Pending pulls: #{RequestId => {From, Timeout}} pending_pulls :: #{binary() => {pid(), reference()}}, %% Known peers for gossip peers :: [binary()], %% Send function for peer communication send_fn :: fun((binary(), map()) -> ok | {error, term()}), %% Configuration push_interval :: pos_integer(), anti_entropy_interval :: pos_integer(), fanout :: pos_integer(), %% Timers push_timer :: reference() | undefined, anti_entropy_timer :: reference() | undefined, %% Stats push_count :: non_neg_integer(), pull_count :: non_neg_integer(), merge_count :: non_neg_integer(), conflict_count :: non_neg_integer() }). -type gossip_state() :: #state{}. %%%=================================================================== %%% API Functions %%%=================================================================== %% @doc Start the gossip server. -spec start_link(map()) -> {ok, pid()} | {error, term()}. start_link(Config) -> gen_server:start_link(?MODULE, Config, []). %% @doc Stop the gossip server. -spec stop(pid()) -> ok. stop(Pid) -> gen_server:stop(Pid). %% @doc Store a CRDT value. %% Type is the CRDT type: lww_register, or_set, gcounter, pncounter -spec put(pid(), binary(), atom(), term()) -> ok. put(Pid, Key, Type, Value) -> gen_server:call(Pid, {put, Key, Type, Value}). %% @doc Get a CRDT value. -spec get(pid(), binary()) -> {ok, {atom(), term()}} | {error, not_found}. get(Pid, Key) -> gen_server:call(Pid, {get, Key}). %% @doc Delete a CRDT value. -spec delete(pid(), binary()) -> ok. delete(Pid, Key) -> gen_server:call(Pid, {delete, Key}). %% @doc Get all stored CRDT states. -spec get_all(pid()) -> #{binary() => {atom(), term()}}. get_all(Pid) -> gen_server:call(Pid, get_all). %% @doc Push local state to a specific peer. -spec push_state(pid(), binary()) -> ok | {error, term()}. push_state(Pid, PeerNodeId) -> gen_server:call(Pid, {push_state, PeerNodeId}). %% @doc Pull state from a specific peer. -spec pull_state(pid(), binary()) -> ok | {error, term()}. pull_state(Pid, PeerNodeId) -> gen_server:call(Pid, {pull_state, PeerNodeId}). %% @doc Trigger anti-entropy synchronization. -spec anti_entropy(pid()) -> ok. anti_entropy(Pid) -> gen_server:cast(Pid, anti_entropy). %% @doc Force synchronization with all peers. -spec force_sync(pid()) -> ok. force_sync(Pid) -> gen_server:cast(Pid, force_sync). %% @doc Add a peer to the gossip list. -spec add_peer(pid(), binary()) -> ok. add_peer(Pid, PeerNodeId) -> gen_server:call(Pid, {add_peer, PeerNodeId}). %% @doc Remove a peer from the gossip list. -spec remove_peer(pid(), binary()) -> ok. remove_peer(Pid, PeerNodeId) -> gen_server:call(Pid, {remove_peer, PeerNodeId}). %% @doc Get the list of known peers. -spec get_peers(pid()) -> [binary()]. get_peers(Pid) -> gen_server:call(Pid, get_peers). %% @doc Get gossip statistics. -spec get_stats(pid()) -> map(). get_stats(Pid) -> gen_server:call(Pid, get_stats). %%%=================================================================== %%% Gossip Message Handlers %%%=================================================================== %% @doc Handle incoming gossip_push message. -spec handle_gossip_push(pid(), map()) -> ok. handle_gossip_push(Pid, Msg) -> gen_server:cast(Pid, {gossip_push, Msg}). %% @doc Handle incoming gossip_pull message. -spec handle_gossip_pull(pid(), map()) -> ok. handle_gossip_pull(Pid, Msg) -> gen_server:cast(Pid, {gossip_pull, Msg}). %% @doc Handle incoming gossip_pull_reply message. -spec handle_gossip_pull_reply(pid(), map()) -> ok. handle_gossip_pull_reply(Pid, Msg) -> gen_server:cast(Pid, {gossip_pull_reply, Msg}). %% @doc Handle incoming gossip_sync message. -spec handle_gossip_sync(pid(), map()) -> ok. handle_gossip_sync(Pid, Msg) -> gen_server:cast(Pid, {gossip_sync, Msg}). %% @doc Handle incoming gossip_sync_reply message. -spec handle_gossip_sync_reply(pid(), map()) -> ok. handle_gossip_sync_reply(Pid, Msg) -> gen_server:cast(Pid, {gossip_sync_reply, Msg}). %%%=================================================================== %%% gen_server callbacks %%%=================================================================== init(Config) -> NodeId = maps:get(node_id, Config, macula_id:node_id()), Realm = maps:get(realm, Config, <<"default">>), %% Default send function (can be overridden for testing) DefaultSendFn = fun(_PeerNodeId, _Msg) -> ok end, SendFn = maps:get(send_fn, Config, DefaultSendFn), %% Configuration PushInterval = maps:get(push_interval, Config, ?DEFAULT_PUSH_INTERVAL), AntiEntropyInterval = maps:get(anti_entropy_interval, Config, ?DEFAULT_ANTI_ENTROPY_INTERVAL), Fanout = maps:get(fanout, Config, ?DEFAULT_FANOUT), %% Initial peers Peers = maps:get(peers, Config, []), State = #state{ node_id = NodeId, realm = Realm, states = #{}, pending_pulls = #{}, peers = Peers, send_fn = SendFn, push_interval = PushInterval, anti_entropy_interval = AntiEntropyInterval, fanout = Fanout, push_timer = undefined, anti_entropy_timer = undefined, push_count = 0, pull_count = 0, merge_count = 0, conflict_count = 0 }, %% Start timers State1 = start_push_timer(State), State2 = start_anti_entropy_timer(State1), ?LOG_INFO("Gossip server started for node ~s in realm ~s", [NodeId, Realm]), {ok, State2}. handle_call({put, Key, Type, Value}, _From, State) -> {Reply, NewState} = do_put(Key, Type, Value, State), {reply, Reply, NewState}; handle_call({get, Key}, _From, State) -> Reply = do_get(Key, State), {reply, Reply, State}; handle_call({delete, Key}, _From, State) -> NewState = do_delete(Key, State), {reply, ok, NewState}; handle_call(get_all, _From, #state{states = States} = State) -> Result = maps:map(fun(_K, {Type, CrdtState, _VClock}) -> {Type, CrdtState} end, States), {reply, Result, State}; handle_call({push_state, PeerNodeId}, _From, State) -> Result = do_push_state(PeerNodeId, State), {reply, Result, State}; handle_call({pull_state, PeerNodeId}, From, State) -> NewState = do_pull_state(PeerNodeId, From, State), {noreply, NewState}; handle_call({add_peer, PeerNodeId}, _From, #state{peers = Peers} = State) -> NewPeers = case lists:member(PeerNodeId, Peers) of true -> Peers; false -> [PeerNodeId | Peers] end, {reply, ok, State#state{peers = NewPeers}}; handle_call({remove_peer, PeerNodeId}, _From, #state{peers = Peers} = State) -> NewPeers = lists:delete(PeerNodeId, Peers), {reply, ok, State#state{peers = NewPeers}}; handle_call(get_peers, _From, #state{peers = Peers} = State) -> {reply, Peers, State}; handle_call(get_stats, _From, State) -> Stats = #{ node_id => State#state.node_id, realm => State#state.realm, state_count => maps:size(State#state.states), peer_count => length(State#state.peers), push_count => State#state.push_count, pull_count => State#state.pull_count, merge_count => State#state.merge_count, conflict_count => State#state.conflict_count }, {reply, Stats, State}. handle_cast(anti_entropy, State) -> NewState = do_anti_entropy(State), {noreply, NewState}; handle_cast(force_sync, State) -> NewState = do_force_sync(State), {noreply, NewState}; handle_cast({gossip_push, Msg}, State) -> NewState = handle_incoming_push(Msg, State), {noreply, NewState}; handle_cast({gossip_pull, Msg}, State) -> NewState = handle_incoming_pull(Msg, State), {noreply, NewState}; handle_cast({gossip_pull_reply, Msg}, State) -> NewState = handle_incoming_pull_reply(Msg, State), {noreply, NewState}; handle_cast({gossip_sync, Msg}, State) -> NewState = handle_incoming_sync(Msg, State), {noreply, NewState}; handle_cast({gossip_sync_reply, Msg}, State) -> NewState = handle_incoming_sync_reply(Msg, State), {noreply, NewState}. handle_info(push_timer, State) -> NewState = do_periodic_push(State), NewState1 = start_push_timer(NewState), {noreply, NewState1}; handle_info(anti_entropy_timer, State) -> NewState = do_anti_entropy(State), NewState1 = start_anti_entropy_timer(NewState), {noreply, NewState1}; handle_info({pull_timeout, RequestId}, #state{pending_pulls = Pending} = State) -> case maps:get(RequestId, Pending, undefined) of undefined -> {noreply, State}; {From, _TimerRef} -> gen_server:reply(From, {error, timeout}), NewPending = maps:remove(RequestId, Pending), {noreply, State#state{pending_pulls = NewPending}} end; handle_info(_Info, State) -> {noreply, State}. terminate(_Reason, #state{push_timer = PushTimer, anti_entropy_timer = AETimer}) -> cancel_timer(PushTimer), cancel_timer(AETimer), ok. %%%=================================================================== %%% Internal Functions - State Management %%%=================================================================== -spec do_put(binary(), atom(), term(), gossip_state()) -> {ok, gossip_state()}. do_put(Key, Type, Value, #state{node_id = NodeId, states = States} = State) -> VClock = increment_vclock(NodeId, get_vclock(Key, States)), NewStates = maps:put(Key, {Type, Value, VClock}, States), {ok, State#state{states = NewStates}}. -spec do_get(binary(), gossip_state()) -> {ok, {atom(), term()}} | {error, not_found}. do_get(Key, #state{states = States}) -> case maps:find(Key, States) of {ok, {Type, CrdtState, _VClock}} -> {ok, {Type, CrdtState}}; error -> {error, not_found} end. -spec do_delete(binary(), gossip_state()) -> gossip_state(). do_delete(Key, #state{states = States} = State) -> NewStates = maps:remove(Key, States), State#state{states = NewStates}. %%%=================================================================== %%% Internal Functions - Gossip Operations %%%=================================================================== -spec do_push_state(binary(), gossip_state()) -> ok | {error, term()}. do_push_state(PeerNodeId, #state{node_id = NodeId, states = States, send_fn = SendFn}) -> %% Build push messages for all states StateList = maps:fold( fun(Key, {Type, CrdtState, VClock}, Acc) -> Msg = #{ <<"node_id">> => NodeId, <<"state_type">> => Type, <<"state_key">> => Key, <<"state">> => CrdtState, <<"vector_clock">> => VClock }, [Msg | Acc] end, [], States ), %% Send each state (could batch in future) lists:foreach( fun(Msg) -> SendFn(PeerNodeId, {gossip_push, Msg}) end, StateList ), ok. -spec do_pull_state(binary(), {pid(), term()}, gossip_state()) -> gossip_state(). do_pull_state(PeerNodeId, From, #state{node_id = NodeId, send_fn = SendFn, pending_pulls = Pending, pull_count = Count} = State) -> RequestId = macula_id:message_id(), Msg = #{ <<"node_id">> => NodeId, <<"request_id">> => RequestId, <<"state_keys">> => [] % Empty = all states }, SendFn(PeerNodeId, {gossip_pull, Msg}), %% Set timeout for response TimerRef = erlang:send_after(5000, self(), {pull_timeout, RequestId}), NewPending = maps:put(RequestId, {From, TimerRef}, Pending), State#state{pending_pulls = NewPending, pull_count = Count + 1}. -spec do_periodic_push(gossip_state()) -> gossip_state(). do_periodic_push(#state{peers = [], push_count = Count} = State) -> State#state{push_count = Count}; do_periodic_push(#state{peers = Peers, fanout = Fanout, push_count = Count} = State) -> %% Select random peers up to fanout SelectedPeers = select_random_peers(Peers, Fanout), lists:foreach( fun(Peer) -> do_push_state(Peer, State) end, SelectedPeers ), State#state{push_count = Count + length(SelectedPeers)}. -spec do_anti_entropy(gossip_state()) -> gossip_state(). do_anti_entropy(#state{peers = []} = State) -> State; do_anti_entropy(#state{peers = Peers, fanout = Fanout, node_id = NodeId, states = States, send_fn = SendFn} = State) -> %% Build digest of local state (key -> vector clock) Digest = maps:map(fun(_Key, {Type, _CrdtState, VClock}) -> {Type, VClock} end, States), %% Send sync request to random peers SelectedPeers = select_random_peers(Peers, Fanout), Msg = #{ <<"node_id">> => NodeId, <<"digest">> => Digest }, lists:foreach( fun(Peer) -> SendFn(Peer, {gossip_sync, Msg}) end, SelectedPeers ), State. -spec do_force_sync(gossip_state()) -> gossip_state(). do_force_sync(#state{peers = []} = State) -> State; do_force_sync(#state{peers = Peers} = State) -> %% Push to all peers lists:foreach( fun(Peer) -> do_push_state(Peer, State) end, Peers ), State. %%%=================================================================== %%% Internal Functions - Incoming Message Handlers %%%=================================================================== -spec handle_incoming_push(map(), gossip_state()) -> gossip_state(). handle_incoming_push(Msg, #state{node_id = LocalNodeId, states = States, merge_count = MergeCount, conflict_count = ConflictCount} = State) -> SenderNodeId = maps:get(<<"node_id">>, Msg, undefined), Key = maps:get(<<"state_key">>, Msg, undefined), Type = maps:get(<<"state_type">>, Msg, undefined), RemoteState = maps:get(<<"state">>, Msg, undefined), RemoteVClock = maps:get(<<"vector_clock">>, Msg, #{}), %% Skip if from self or invalid case SenderNodeId of LocalNodeId -> State; undefined -> State; _ when Key =:= undefined; Type =:= undefined; RemoteState =:= undefined -> State; _ -> {NewStates, NewMergeCount, NewConflictCount} = merge_state(Key, Type, RemoteState, RemoteVClock, States, MergeCount, ConflictCount), State#state{ states = NewStates, merge_count = NewMergeCount, conflict_count = NewConflictCount } end. -spec handle_incoming_pull(map(), gossip_state()) -> gossip_state(). handle_incoming_pull(Msg, #state{node_id = NodeId, states = States, send_fn = SendFn} = State) -> SenderNodeId = maps:get(<<"node_id">>, Msg, undefined), RequestedKeys = maps:get(<<"state_keys">>, Msg, []), %% Build response with requested states (or all if empty) StateList = case RequestedKeys of [] -> %% All states maps:fold( fun(Key, {Type, CrdtState, VClock}, Acc) -> [#{key => Key, type => Type, state => CrdtState, vector_clock => VClock} | Acc] end, [], States ); Keys -> %% Only requested keys lists:filtermap( fun(Key) -> case maps:find(Key, States) of {ok, {Type, CrdtState, VClock}} -> {true, #{key => Key, type => Type, state => CrdtState, vector_clock => VClock}}; error -> false end end, Keys ) end, %% Send reply Reply = #{ <<"node_id">> => NodeId, <<"request_id">> => maps:get(<<"request_id">>, Msg, <<>>), <<"states">> => StateList }, SendFn(SenderNodeId, {gossip_pull_reply, Reply}), State. -spec handle_incoming_pull_reply(map(), gossip_state()) -> gossip_state(). handle_incoming_pull_reply(Msg, #state{pending_pulls = Pending, states = States, merge_count = MergeCount, conflict_count = ConflictCount} = State) -> RequestId = maps:get(<<"request_id">>, Msg, <<>>), ReceivedStates = maps:get(<<"states">>, Msg, []), %% Cancel pending timer and reply NewPending = case maps:get(RequestId, Pending, undefined) of undefined -> Pending; {From, TimerRef} -> erlang:cancel_timer(TimerRef), gen_server:reply(From, ok), maps:remove(RequestId, Pending) end, %% Merge received states {NewStates, NewMergeCount, NewConflictCount} = lists:foldl( fun(#{key := Key, type := Type, state := RemoteState, vector_clock := VClock}, {StatesAcc, MC, CC}) -> merge_state(Key, Type, RemoteState, VClock, StatesAcc, MC, CC) end, {States, MergeCount, ConflictCount}, ReceivedStates ), State#state{ pending_pulls = NewPending, states = NewStates, merge_count = NewMergeCount, conflict_count = NewConflictCount }. -spec handle_incoming_sync(map(), gossip_state()) -> gossip_state(). handle_incoming_sync(Msg, #state{node_id = NodeId, states = States, send_fn = SendFn} = State) -> SenderNodeId = maps:get(<<"node_id">>, Msg, undefined), RemoteDigest = maps:get(<<"digest">>, Msg, #{}), %% Find states we have that are newer than remote, or that remote doesn't have StatesToSend = maps:fold( fun(Key, {Type, CrdtState, LocalVClock}, Acc) -> case maps:find(Key, RemoteDigest) of {ok, {_RemoteType, RemoteVClock}} -> %% Check if our clock is newer case vclock_dominates(LocalVClock, RemoteVClock) of true -> [#{key => Key, type => Type, state => CrdtState, vector_clock => LocalVClock} | Acc]; false -> Acc end; error -> %% Remote doesn't have this key [#{key => Key, type => Type, state => CrdtState, vector_clock => LocalVClock} | Acc] end end, [], States ), %% Find keys remote has that we don't MissingKeys = maps:fold( fun(Key, _Value, Acc) -> case maps:is_key(Key, States) of true -> Acc; false -> [Key | Acc] end end, [], RemoteDigest ), %% Send sync reply Reply = #{ <<"node_id">> => NodeId, <<"states">> => StatesToSend, <<"missing">> => MissingKeys }, SendFn(SenderNodeId, {gossip_sync_reply, Reply}), State. -spec handle_incoming_sync_reply(map(), gossip_state()) -> gossip_state(). handle_incoming_sync_reply(Msg, #state{states = States, merge_count = MergeCount, conflict_count = ConflictCount} = State) -> ReceivedStates = maps:get(<<"states">>, Msg, []), MissingKeys = maps:get(<<"missing">>, Msg, []), SenderNodeId = maps:get(<<"node_id">>, Msg, undefined), %% Merge received states {NewStates, NewMergeCount, NewConflictCount} = lists:foldl( fun(#{key := Key, type := Type, state := RemoteState, vector_clock := VClock}, {StatesAcc, MC, CC}) -> merge_state(Key, Type, RemoteState, VClock, StatesAcc, MC, CC) end, {States, MergeCount, ConflictCount}, ReceivedStates ), %% Push missing keys to sender case MissingKeys of [] -> ok; _ -> lists:foreach( fun(Key) -> case maps:find(Key, NewStates) of {ok, {Type, CrdtState, VClock}} -> PushMsg = #{ <<"node_id">> => State#state.node_id, <<"state_type">> => Type, <<"state_key">> => Key, <<"state">> => CrdtState, <<"vector_clock">> => VClock }, (State#state.send_fn)(SenderNodeId, {gossip_push, PushMsg}); error -> ok end end, MissingKeys ) end, State#state{ states = NewStates, merge_count = NewMergeCount, conflict_count = NewConflictCount }. %%%=================================================================== %%% Internal Functions - CRDT Merging %%%=================================================================== -spec merge_state(binary(), atom(), term(), map(), map(), non_neg_integer(), non_neg_integer()) -> {map(), non_neg_integer(), non_neg_integer()}. merge_state(Key, Type, RemoteState, RemoteVClock, States, MergeCount, ConflictCount) -> case maps:find(Key, States) of {ok, {LocalType, LocalState, LocalVClock}} when LocalType =:= Type -> %% Both have the state - merge case vclock_compare(LocalVClock, RemoteVClock) of equal -> %% Same version, no change needed {States, MergeCount, ConflictCount}; dominates -> %% Local is newer, keep local {States, MergeCount, ConflictCount}; dominated -> %% Remote is newer, take remote NewStates = maps:put(Key, {Type, RemoteState, RemoteVClock}, States), {NewStates, MergeCount + 1, ConflictCount}; concurrent -> %% Concurrent updates - merge CRDTs MergedState = merge_crdt(Type, LocalState, RemoteState), MergedVClock = vclock_merge(LocalVClock, RemoteVClock), NewStates = maps:put(Key, {Type, MergedState, MergedVClock}, States), {NewStates, MergeCount + 1, ConflictCount + 1} end; {ok, {_DifferentType, _LocalState, _LocalVClock}} -> %% Type mismatch - log and keep remote (last write wins at type level) ?LOG_WARNING("Type mismatch for key ~s: local has different type, taking remote ~p", [Key, Type]), NewStates = maps:put(Key, {Type, RemoteState, RemoteVClock}, States), {NewStates, MergeCount + 1, ConflictCount + 1}; error -> %% Don't have this key locally, add it NewStates = maps:put(Key, {Type, RemoteState, RemoteVClock}, States), {NewStates, MergeCount + 1, ConflictCount} end. -spec merge_crdt(atom(), term(), term()) -> term(). merge_crdt(lww_register, Local, Remote) -> macula_crdt:lww_merge(Local, Remote); merge_crdt(or_set, Local, Remote) -> macula_crdt:or_merge(Local, Remote); merge_crdt(gcounter, Local, Remote) -> macula_crdt:gcounter_merge(Local, Remote); merge_crdt(pncounter, Local, Remote) -> macula_crdt:pncounter_merge(Local, Remote); merge_crdt(_UnknownType, _Local, Remote) -> %% Unknown type - take remote Remote. %%%=================================================================== %%% Internal Functions - Vector Clocks %%%=================================================================== -spec get_vclock(binary(), map()) -> map(). get_vclock(Key, States) -> case maps:find(Key, States) of {ok, {_Type, _State, VClock}} -> VClock; error -> #{} end. -spec increment_vclock(binary(), map()) -> map(). increment_vclock(NodeId, VClock) -> Current = maps:get(NodeId, VClock, 0), maps:put(NodeId, Current + 1, VClock). -spec vclock_compare(map(), map()) -> equal | dominates | dominated | concurrent. vclock_compare(VClock1, VClock2) when VClock1 =:= VClock2 -> equal; vclock_compare(VClock1, VClock2) -> Dom1 = vclock_dominates(VClock1, VClock2), Dom2 = vclock_dominates(VClock2, VClock1), case {Dom1, Dom2} of {true, false} -> dominates; {false, true} -> dominated; {false, false} -> concurrent; {true, true} -> equal % Shouldn't happen, but handle it end. -spec vclock_dominates(map(), map()) -> boolean(). vclock_dominates(VClock1, VClock2) -> %% VClock1 dominates VClock2 if all entries in VClock2 are <= corresponding entries in VClock1 %% and at least one entry is strictly greater AllKeys = lists:usort(maps:keys(VClock1) ++ maps:keys(VClock2)), {AllGE, SomeGT} = lists:foldl( fun(Key, {AllGEAcc, SomeGTAcc}) -> V1 = maps:get(Key, VClock1, 0), V2 = maps:get(Key, VClock2, 0), {AllGEAcc andalso V1 >= V2, SomeGTAcc orelse V1 > V2} end, {true, false}, AllKeys ), AllGE andalso SomeGT. -spec vclock_merge(map(), map()) -> map(). vclock_merge(VClock1, VClock2) -> maps:fold( fun(Key, V2, Acc) -> V1 = maps:get(Key, Acc, 0), maps:put(Key, max(V1, V2), Acc) end, VClock1, VClock2 ). %%%=================================================================== %%% Internal Functions - Timer Management %%%=================================================================== -spec start_push_timer(gossip_state()) -> gossip_state(). start_push_timer(#state{push_interval = Interval} = State) -> TimerRef = erlang:send_after(Interval, self(), push_timer), State#state{push_timer = TimerRef}. -spec start_anti_entropy_timer(gossip_state()) -> gossip_state(). start_anti_entropy_timer(#state{anti_entropy_interval = Interval} = State) -> TimerRef = erlang:send_after(Interval, self(), anti_entropy_timer), State#state{anti_entropy_timer = TimerRef}. -spec cancel_timer(reference() | undefined) -> ok. cancel_timer(undefined) -> ok; cancel_timer(TimerRef) -> erlang:cancel_timer(TimerRef), ok. %%%=================================================================== %%% Internal Functions - Peer Selection %%%=================================================================== -spec select_random_peers([binary()], pos_integer()) -> [binary()]. select_random_peers(Peers, MaxCount) when length(Peers) =< MaxCount -> Peers; select_random_peers(Peers, MaxCount) -> %% Shuffle and take first MaxCount Shuffled = [X || {_, X} <- lists:sort([{rand:uniform(), P} || P <- Peers])], lists:sublist(Shuffled, MaxCount).