diff --git a/INTERNALS.md b/INTERNALS.md index dcd3e6b3..ef1f5407 100644 --- a/INTERNALS.md +++ b/INTERNALS.md @@ -50,11 +50,11 @@ the following happens: * Starts monitoring the remote scope node process. * Adds the remote node to the list of the subcluster nodes. * The scope process that receives the `ack_sync` message: - * Stores the received data of the remote node. - * If it's an unknown node, it: + * If it's an unknown node, it first: * Starts monitoring the remote scope node process. - * Sends it its local data with another ack message. * Adds the remote node to the list of the subcluster nodes. + * Sends it its local data with another ack message, before a conflict resolution can change that data. + * Stores the received data of the remote node. #### Leaving diff --git a/src/syn_gen_scope.erl b/src/syn_gen_scope.erl index bf81bf09..01238220 100644 --- a/src/syn_gen_scope.erl +++ b/src/syn_gen_scope.erl @@ -234,22 +234,26 @@ handle_info({'3.0', ack_sync, RemoteScopePid, Data}, #state{ RemoteScopeNode = node(RemoteScopePid), ?LOG_NOTICE(#{handler => HandlerLogName, scope => Scope, event => ack_sync, from => RemoteScopeNode, entries => length(Data)}), - %% save remote data - Handler:save_remote_data(RemoteScopeNode, Data, State), %% is this a new node? case maps:is_key(RemoteScopeNode, NodesMap) of true -> - %% already known + %% already known, save remote data + Handler:save_remote_data(RemoteScopeNode, Data, State), {noreply, State}; false -> %% monitor _MRef = monitor(process, RemoteScopePid), - %% send local to remote (ordered to maintain FIFO with broadcasts) - {ok, LocalData} = Handler:get_local_data(State), - send_to_node(RemoteScopeNode, {'3.0', ack_sync, self(), LocalData}, State), + State1 = State#state{nodes_map = NodesMap#{RemoteScopeNode => RemoteScopePid}}, + %% add to subcluster and send local data to remote before saving remote data, + %% so that the snapshot keeps the entries a conflict resolution may remove + %% and the broadcasts of the resolution reach the remote node after it (FIFO) + {ok, LocalData} = Handler:get_local_data(State1), + send_to_node(RemoteScopeNode, {'3.0', ack_sync, self(), LocalData}, State1), + %% save remote data + Handler:save_remote_data(RemoteScopeNode, Data, State1), %% return - {noreply, State#state{nodes_map = NodesMap#{RemoteScopeNode => RemoteScopePid}}} + {noreply, State1} end; handle_info({'DOWN', MRef, process, Pid, Reason}, #state{ diff --git a/test/syn_registry_SUITE.erl b/test/syn_registry_SUITE.erl index 04d2889c..e2accfd6 100644 --- a/test/syn_registry_SUITE.erl +++ b/test/syn_registry_SUITE.erl @@ -52,6 +52,8 @@ three_nodes_ack_sync_reconciles_large_snapshot/1, three_nodes_stale_snapshot_does_not_undo_remote_register/1, three_nodes_stale_snapshot_does_not_undo_remote_update/1, + three_nodes_ack_sync_from_unknown_node_keeps_remote/1, + three_nodes_ack_sync_from_unknown_node_resolution_reaches_peer/1, three_nodes_resolver_keeps_neither_while_process_dies/1 ]). -export([ @@ -114,6 +116,8 @@ groups() -> three_nodes_ack_sync_reconciles_large_snapshot, three_nodes_stale_snapshot_does_not_undo_remote_register, three_nodes_stale_snapshot_does_not_undo_remote_update, + three_nodes_ack_sync_from_unknown_node_keeps_remote, + three_nodes_ack_sync_from_unknown_node_resolution_reaches_peer, three_nodes_resolver_keeps_neither_while_process_dies ]}, {four_nodes_registry, [shuffle], [ @@ -1939,6 +1943,127 @@ three_nodes_stale_snapshot_does_not_undo_remote_update(Config) -> {Pid, new_meta} = syn:lookup(scope_stale, "proc"), {Pid, new_meta} = rpc:call(SlaveNode1, syn, lookup, [scope_stale, "proc"]). +three_nodes_ack_sync_from_unknown_node_keeps_remote(Config) -> + %% The local node sees both slaves, which cannot see each other. The name is registered + %% on slave 1 first and on slave 2 later, so the local node holds the process on slave 2. + %% The resolver keeps the process on slave 1. + %% + %% When the slaves connect, slave 2 handles the ack_sync of slave 1 while slave 1 is not + %% in its nodes_map yet, and slave 1 handles the snapshot of slave 2 as a known node, as + %% in a join where slave 1 handles the discover of slave 2 before its own nodeup. + %% Slave 2 keeps the remote process and broadcasts the unregistration of its own process. + %% Slave 1 must get the conflicting entry in the snapshot of slave 2 and resolve the + %% conflict too: only slave 1 sends its process to the local node. + + %% get slaves + SlaveNode1 = proplists:get_value(syn_slave_1, Config), + SlaveNode2 = proplists:get_value(syn_slave_2, Config), + + %% start syn on nodes + ok = syn:start(), + ok = rpc:call(SlaveNode1, syn, start, []), + ok = rpc:call(SlaveNode2, syn, start, []), + + %% add custom handler for resolution + syn:set_event_handler(syn_test_event_handler_resolution), + rpc:call(SlaveNode1, syn, set_event_handler, [syn_test_event_handler_resolution]), + rpc:call(SlaveNode2, syn, set_event_handler, [syn_test_event_handler_resolution]), + + %% conflict in a partial netsplit, the process to keep is registered first + TestPid = self(), + LocalNode = node(), + Name = "proc-conflict-ack-sync-keep-remote", + {PidOn1, PidOn2} = register_conflict_in_partial_netsplit( + SlaveNode1, SlaveNode2, Name, {recipient, TestPid, keepthis}, {recipient, TestPid, "meta-2"} + ), + syn_test_suite_helper:assert_received_messages([ + {on_process_registered, SlaveNode1, scope_all, Name, PidOn1, keepthis, normal}, + {on_process_registered, LocalNode, scope_all, Name, PidOn1, keepthis, normal}, + {on_process_registered, SlaveNode2, scope_all, Name, PidOn2, "meta-2", normal}, + {on_process_unregistered, LocalNode, scope_all, Name, PidOn1, keepthis, normal}, + {on_process_registered, LocalNode, scope_all, Name, PidOn2, "meta-2", normal} + ]), + + %% re-join, slave 2 handles the ack_sync of slave 1 first + join_with_ack_sync_first(SlaveNode2, SlaveNode1, Name), + + %% check callbacks + syn_test_suite_helper:assert_received_messages([ + {on_process_unregistered, SlaveNode2, scope_all, Name, PidOn2, "meta-2", syn_conflict_resolution}, + {on_process_registered, SlaveNode2, scope_all, Name, PidOn1, keepthis, syn_conflict_resolution}, + {on_process_unregistered, LocalNode, scope_all, Name, PidOn2, "meta-2", syn_conflict_resolution}, + {on_process_registered, LocalNode, scope_all, Name, PidOn1, keepthis, syn_conflict_resolution} + ]), + + %% retrieve + syn_test_suite_helper:assert_wait( + {PidOn1, {recipient, TestPid, keepthis}}, + fun() -> rpc:call(SlaveNode1, syn, lookup, [scope_all, Name]) end + ), + syn_test_suite_helper:assert_wait( + {PidOn1, {recipient, TestPid, keepthis}}, + fun() -> rpc:call(SlaveNode2, syn, lookup, [scope_all, Name]) end + ), + syn_test_suite_helper:assert_wait( + {PidOn1, {recipient, TestPid, keepthis}}, + fun() -> syn:lookup(scope_all, Name) end + ). + +three_nodes_ack_sync_from_unknown_node_resolution_reaches_peer(Config) -> + %% Slave 1 adds a scope in which slave 2 already holds a name, and registers the same + %% name before slave 2 answers its discover. Slave 1 then handles the ack_sync of slave 2 + %% while slave 2 is not in its nodes_map, keeps its local process, the later registration, + %% and broadcasts it with a new time. The slaves stay connected, so no nodeup follows and + %% slave 1 handles no other ack_sync: slave 2 gets the new time only through that + %% broadcast. Both slaves must end with the same process and the same time. + + %% get slaves + SlaveNode1 = proplists:get_value(syn_slave_1, Config), + SlaveNode2 = proplists:get_value(syn_slave_2, Config), + + %% start syn on slaves + ok = rpc:call(SlaveNode1, syn, start, []), + ok = rpc:call(SlaveNode2, syn, start, []), + + %% register on slave 2, then suspend its scope process + Name = "proc-conflict-ack-sync-resolution", + ok = rpc:call(SlaveNode2, syn, add_node_to_scopes, [[scope_bc]]), + PidOn2 = syn_test_suite_helper:start_process(SlaveNode2), + ok = rpc:call(SlaveNode2, syn, register, [scope_bc, Name, PidOn2, "meta-2"]), + ScopePid2 = rpc:call(SlaveNode2, erlang, whereis, [syn_registry_scope_bc]), + ok = sys:suspend(ScopePid2), + + %% add the scope on slave 1, its discover waits in the queue of slave 2 + ok = rpc:call(SlaveNode1, syn, add_node_to_scopes, [[scope_bc]]), + wait_for_message_queue_len(ScopePid2, 1), + + %% register on slave 1 later, then let slave 2 answer the discover + PidOn1 = syn_test_suite_helper:start_process(SlaveNode1), + ok = rpc:call(SlaveNode1, syn, register, [scope_bc, Name, PidOn1, "meta-1"]), + ok = sys:resume(ScopePid2), + + %% retrieve + syn_test_suite_helper:assert_wait( + {PidOn1, "meta-1"}, + fun() -> rpc:call(SlaveNode1, syn, lookup, [scope_bc, Name]) end + ), + syn_test_suite_helper:assert_wait( + {PidOn1, "meta-1"}, + fun() -> rpc:call(SlaveNode2, syn, lookup, [scope_bc, Name]) end + ), + + %% same entry on both slaves, time included + syn_test_suite_helper:assert_wait( + ok, + fun() -> + Entries = [get_registry_entry(Node, scope_bc, Name) || Node <- [SlaveNode1, SlaveNode2]], + case lists:usort(Entries) of + [{Name, PidOn1, "meta-1", _Time}] -> ok; + _ -> Entries + end + end + ). + three_nodes_resolver_keeps_neither_while_process_dies(Config) -> %% the resolver keeps neither process, and the process on slave 1 dies after slave 1 sent its list %% but before it reads the list of slave 2: slave 1 must not keep the process of slave 2, @@ -2122,6 +2247,64 @@ get_remote_snapshot(Node, ScopeProcessName) -> {ok, Snapshot} = rpc:call(Node, syn_registry, get_local_data, [State]), {RemoteScopePid, Snapshot}. +%% Entry of Name in the registry table of Node, in the format of an ack_sync snapshot. +get_registry_entry(Node, Scope, Name) -> + TableByName = rpc:call(Node, syn_backbone, get_table_name, [syn_registry_by_name, Scope]), + case rpc:call(Node, ets, lookup, [TableByName, Name]) of + [{Name, Pid, Meta, Time, _MRef, _Node}] -> {Name, Pid, Meta, Time}; + [] -> undefined + end. + +%% Add scope_all on the local node and both slaves, disconnect slave 1 from slave 2, +%% then register Name on slave 1 and later on slave 2. The local node sees both +%% registrations and holds the later one. +register_conflict_in_partial_netsplit(SlaveNode1, SlaveNode2, Name, Meta1, Meta2) -> + %% add scopes + ok = syn:add_node_to_scopes([scope_all]), + ok = rpc:call(SlaveNode1, syn, add_node_to_scopes, [[scope_all]]), + ok = rpc:call(SlaveNode2, syn, add_node_to_scopes, [[scope_all]]), + syn_test_suite_helper:assert_scope_subcluster(registry, node(), scope_all, [SlaveNode1, SlaveNode2]), + syn_test_suite_helper:assert_scope_subcluster(registry, SlaveNode1, scope_all, [node(), SlaveNode2]), + syn_test_suite_helper:assert_scope_subcluster(registry, SlaveNode2, scope_all, [node(), SlaveNode1]), + + %% partial netsplit (1 cannot see 2) + rpc:call(SlaveNode1, syn_test_suite_helper, disconnect_node, [SlaveNode2]), + syn_test_suite_helper:assert_scope_subcluster(registry, SlaveNode1, scope_all, [node()]), + syn_test_suite_helper:assert_scope_subcluster(registry, SlaveNode2, scope_all, [node()]), + + %% register on slave 1, then on slave 2 + PidOn1 = syn_test_suite_helper:start_process(SlaveNode1), + PidOn2 = syn_test_suite_helper:start_process(SlaveNode2), + ok = rpc:call(SlaveNode1, syn, register, [scope_all, Name, PidOn1, Meta1]), + syn_test_suite_helper:assert_wait( + {PidOn1, Meta1}, + fun() -> syn:lookup(scope_all, Name) end + ), + ok = rpc:call(SlaveNode2, syn, register, [scope_all, Name, PidOn2, Meta2]), + syn_test_suite_helper:assert_wait( + {PidOn2, Meta2}, + fun() -> syn:lookup(scope_all, Name) end + ), + {PidOn1, PidOn2}. + +%% Put the snapshot of PeerNode, which holds Name, in the queue of the suspended scope_all +%% registry process of Node, connect the two nodes, then resume the process. Node handles +%% that ack_sync while PeerNode is not in its nodes_map, before the nodeup and the +%% discover of the join. Before the resume, PeerNode handles a discover of Node, so it +%% handles the snapshot of Node as a known node. +join_with_ack_sync_first(Node, PeerNode, Name) -> + ScopePid = rpc:call(Node, erlang, whereis, [syn_registry_scope_all]), + {PeerScopePid, PeerSnapshot} = get_remote_snapshot(PeerNode, syn_registry_scope_all), + true = lists:keymember(Name, 1, PeerSnapshot), + ok = sys:suspend(ScopePid), + ScopePid ! {'3.0', ack_sync, PeerScopePid, PeerSnapshot}, + wait_for_message_queue_len(ScopePid, 1), + true = rpc:call(PeerNode, syn_test_suite_helper, connect_node, [Node]), + syn_test_suite_helper:assert_cluster(Node, [node(), PeerNode]), + PeerScopePid ! {'3.0', discover, ScopePid}, + syn_test_suite_helper:assert_scope_subcluster(registry, PeerNode, scope_all, [node(), Node]), + ok = sys:resume(ScopePid). + wait_for_message_queue_len(Pid, Len) -> syn_test_suite_helper:assert_wait( {message_queue_len, Len},