%% ----------------------------------------------------------------------------- %% %% Copyright (c) 2017 Christopher S. Meiklejohn. All Rights Reserved. %% Copyright (c) 2022 Alejandro M. Ramallo. All Rights Reserved. %% %% This file is provided to you 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. %% %% ----------------------------------------------------------------------------- %% ----------------------------------------------------------------------------- %% @doc This module represents the cluster membership view for this node. %% %% When a node joins the cluster it is added to the set. Conversely when a node %% leaves the cluster it is removed from the set. A node that crashes or gets %% disconnected will remain in the set so that Partisan can try to re-connect %% with the node when it restarts or becomes reachable again. %% %% == Implementation == %% The set is implemented as a CRDT set of `partisan:node_spec()' objects. More %% specifically a state_orset. %% %% Notice that because the set stores `partisan:node_spec()' objects and not %% `node()', %% the set can have multiple `partisan:node_spec()' objects for the same node. %% %% This can occur when the set contains one or more %% stale specifications. %% %% == Stale Specifications == %% A stale specification exists due to the following reasons: %% %% %% %% @end %% ----------------------------------------------------------------------------- -module(partisan_membership_set). -include("partisan.hrl"). -include("partisan_logger.hrl"). -ifdef(TEST). -include_lib("eunit/include/eunit.hrl"). -endif. -opaque t() :: state_orset:state_orset(). -export_type([t/0]). %% API -export([add/3]). -export([compare/2]). -export([decode/1]). -export([encode/1]). -export([equal/2]). -export([merge/2]). -export([new/0]). -export([remove/3]). -export([to_list/1]). -export([to_peer_list/1]). %% ============================================================================= %% API %% ============================================================================= %% ----------------------------------------------------------------------------- %% @doc %% @end %% ----------------------------------------------------------------------------- -spec new() -> t(). new() -> state_orset:new(). %% ----------------------------------------------------------------------------- %% @doc %% @end %% ----------------------------------------------------------------------------- -spec add(partisan:node_spec(), Actor :: partisan:actor(), t()) -> t(). add(#{name := _} = NodeSpec, Actor, T0) -> {ok, T} = state_orset:mutate({add, NodeSpec}, Actor, T0), T. %% ----------------------------------------------------------------------------- %% @doc %% @end %% ----------------------------------------------------------------------------- -spec remove(partisan:node_spec(), Actor :: partisan:actor(), t()) -> t(). remove(#{name := _} = NodeSpec, Actor, T) -> {ok, T1} = state_orset:mutate({rmv, NodeSpec}, Actor, T), T1. %% ----------------------------------------------------------------------------- %% @doc Returns the tuple `{Joiners, Leavers}' where `Joiners' is the list of %% node specifications that are elements of `List' but are not in the %% membership set, and `Leavers' are the node specifications for the current %% members that are not elements in `List'. %% @end %% ----------------------------------------------------------------------------- -spec compare([partisan:node_spec()], t()) -> {Joiners :: [partisan:node_spec()], Leavers :: [partisan:node_spec()]}. compare([], _) -> {[], []}; compare(List, T) when is_list(List) -> Set = sets:from_list(List), Members = state_orset:query(T), Intersection = sets:intersection(Set, Members), Joiners = sets:to_list(sets:subtract(Set, Intersection)), Leavers = sets:to_list(sets:subtract(Members, Intersection)), {Joiners, Leavers}. %% ----------------------------------------------------------------------------- %% @doc %% @end %% ----------------------------------------------------------------------------- -spec merge(t(), t()) -> t(). merge(T1, T2) -> state_orset:merge(T1, T2). %% ----------------------------------------------------------------------------- %% @doc %% @end %% ----------------------------------------------------------------------------- -spec equal(t(), t()) -> boolean(). equal(T1, T2) -> state_orset:equal(T1, T2). %% ----------------------------------------------------------------------------- %% @doc %% @end %% ----------------------------------------------------------------------------- -spec to_list(t()) -> [partisan:node_spec()]. to_list(T) -> lists:sort(sets:to_list(state_orset:query(T))). %% ----------------------------------------------------------------------------- %% @doc Returns a list of node specifications but omitting the specification %% for the local node. %% No sorting is applied, so the sorting is undefined. %% @end %% ----------------------------------------------------------------------------- -spec to_peer_list(t()) -> [partisan:node_spec()]. to_peer_list(T) -> Myself = partisan:node(), Peers = sets:fold( fun (#{name := Node} = Spec, Acc) when Node =/= Myself -> [Spec | Acc]; (_, Acc) -> Acc end, [], state_orset:query(T) ), lists:reverse(Peers). %% ----------------------------------------------------------------------------- %% @doc %% @end %% ----------------------------------------------------------------------------- -spec encode(t()) -> binary(). encode(T) -> Opts = partisan_config:get('$membership_encoding_opts', [compressed]), erlang:term_to_binary(T, Opts). %% ----------------------------------------------------------------------------- %% @doc %% @end %% ----------------------------------------------------------------------------- -spec decode(binary()) -> t(). decode(Binary) -> erlang:binary_to_term(Binary). %% ============================================================================= %% EUNIT TESTS %% ============================================================================= -ifdef(TEST). node_spec(Nodename) -> node_spec(Nodename, {127,0,0,1}). node_spec(Nodename, IP) -> #{ name => Nodename, listen_addrs => [#{ip => IP, port => 62823}], channels => #{ ?DEFAULT_CHANNEL => #{ parallelism => 1, monotonic => false } } }. add_remove_test() -> Nodename = 'node1@127.0.0.1', Node = node_spec(Nodename), A = add(Node, a, new()), ?assertEqual( [Node], to_list(A) ), A1 = remove(Node, a, A), ?assertEqual( [], to_list(A1) ). one_side_updates_test() -> Nodename = 'node1@127.0.0.1', Node1 = node_spec(Nodename, {127, 0, 0, 1}), A = B = new(), A1 = add(Node1, a, A), ?assertEqual( [Node1], to_list(A1) ), ?assertEqual( [Node1], to_list(merge(A1, B)) ). concurrent_updates_test() -> Nodename = 'node1@127.0.0.1', Node1 = node_spec(Nodename, {127, 0, 0, 1}), Node2 = node_spec(Nodename, {192, 168, 0, 1}), A = B = new(), A1 = add(Node1, a, A), B1 = add(Node2, b, B), ?assertEqual( [Node1], to_list(A1) ), ?assertEqual( [Node2], to_list(B1) ), ?assertEqual( [Node1, Node2], to_list(merge(A1, B1)) ). compare_test() -> Node1 = node_spec('node1@127.0.0.1', {192, 168, 0, 1}), Node2 = node_spec('node2@127.0.0.1', {192, 168, 0, 2}), Node3 = node_spec('node3@127.0.0.1', {192, 168, 0, 3}), S0 = new(), S1 = add(Node1, a, S0), S2 = add(Node2, a, S1), ?assertEqual( {[], []}, compare([], S2) ), ?assertEqual( {[], []}, compare([Node1, Node2], S2) ), ?assertEqual( {[Node3], []}, compare([Node1, Node2, Node3], S2) ), ?assertEqual( {[Node3], [Node2]}, compare([Node1, Node3], S2) ). concurrent_remove_update_test() -> Nodename = 'node1@127.0.0.1', Node1 = node_spec(Nodename, {127, 0, 0, 1}), Node2 = node_spec(Nodename, {192, 168, 0, 1}), A = add(Node1, a, new()), A1 = remove(Node1, a, A), B = A, %% not A1 B1 = add(Node2, b, B), ?assertEqual( [Node1], to_list(A) ), ?assertEqual( [], to_list(A1) ), ?assertEqual( [Node1, Node2], to_list(B1) ), %% Replicate to A ?assertEqual( [Node2], to_list(merge(A1, B1)) ). %% This fails on previous version of riak_dt_map assoc_test() -> Nodename = 'node1@127.0.0.1', Node = node_spec(Nodename), A = add(Node, a, new()), B = add(Node, b, new()), B2 = remove(Node, b, B), C = A, C3 = remove(Node, c, C), ?assertEqual( merge(A, merge(B2, C3)), merge(merge(A, B2), C3) ), ?assertEqual( to_list(merge(merge(A, C3), B2)), to_list(merge(merge(A, B2), C3)) ), ?assertEqual( merge(merge(A, C3), B2), merge(merge(A, B2), C3) ). clock_test() -> Nodename = 'node1@127.0.0.1', Node1 = node_spec(Nodename), Node2 = node_spec(Nodename, {192, 168, 0, 1}), A = add(Node1, a, new()), B = A, B2 = add(Node2, b, B), A2 = remove(Node1, a, A), A4 = add(Node2, a, A2), AB = merge(A4, B2), %% LWW ?assertEqual([Node2], to_list(AB)). remfield_test() -> Name = 'node1@127.0.0.1', Node1 = node_spec(Name), Node2 = node_spec(Name, {192, 168, 0, 1}), A = add(Node1, a, new()), B = A, A2 = remove(Node1, a, A), A4 = add(Node2, a, A2), AB = merge(A4, B), ?assertEqual([Node2], to_list(AB)). %% Bug found by EQC, not dropping dots in merge when an element is %% present in both Maos leads to removed items remaining after merge. present_but_removed_test() -> Name = 'node1@127.0.0.1', Node1 = node_spec(Name), Node2 = node_spec(Name, {192, 168, 0, 1}), %% Add Z to A A = add(Node1, a, new()), %% Replicate it to C so A has 'Z'->{a, 1} C = A, %% Remove Z from A A2 = remove(Node1, a, A), %% Add Z to B, a new replica B = add(Node2, b, new()), %% Replicate B to A, so now A has a Z, the one with a Dot of %% {b,1} and clock of [{a, 1}, {b, 1}] A3 = merge(B, A2), %% Remove the 'Z' from B replica B2 = remove(Node2, b, B), %% Both C and A have a 'Z', but when they merge, there should be %% no 'Z' as C's has been removed by A and A's has been removed by %% C. Merged = lists:foldl(fun(Set, Acc) -> merge(Set, Acc) end, %% the order matters, the two replicas that %% have 'Z' need to merge first to provoke %% the bug. You end up with 'Z' with two %% dots, when really it should be removed. A3, [C, B2]), ?assertEqual([], to_list(Merged)). %% A bug EQC found where dropping the dots in merge was not enough if %% you then store the value with an empty clock (derp). no_dots_left_test() -> Name = 'node1@127.0.0.1', Node1 = node_spec(Name), Node2 = node_spec(Name, {192, 168, 0, 1}), A = add(Node1, a, new()), B = add(Node2, b, new()), C = A, %% replicate A to empty C A2 = remove(Node1, a, A), %% replicate B to A, now A has B's 'Z' A3 = merge(A2, B), %% Remove B's 'Z' B2 = remove(Node2, b, B), %% Replicate C to B, now B has A's old 'Z' B3 = merge(B2, C), %% Merge everytyhing, without the fix You end up with 'Z' present, %% with no dots Merged = lists:foldl(fun(Set, Acc) -> merge(Set, Acc) end, A3, [B3, C]), ?assertEqual([], to_list(Merged)). equals_test() -> Name1 = 'node1@127.0.0.1', Node1 = node_spec(Name1), A = add(Node1, a, new()), B = add(Node1, b, new()), ?assert(not equal(A, B)), C = merge(A, B), D = merge(B, A), ?assert(equal(C, D)), ?assert(equal(A, A)). -endif.