From 4a4004cdfcea4b4e3ec36d83913fec075f142f48 Mon Sep 17 00:00:00 2001 From: Vladimir Drobyshevskiy Date: Thu, 17 Sep 2026 15:32:29 +0400 Subject: [PATCH 1/2] Reconcile remote registry snapshots with a map instead of an ordset reconcile_remote_registry_snapshot/3 tested every existing entry of the remote node against the snapshot names with ordsets:is_element/2, which is a linear scan, so the reconciliation was quadratic. It shows when an ack_sync arrives from a node whose entries are already in the table, for example when a node is discovered twice at scope start. The names now go into a map and membership is tested with is_map_key/2. The map is built with maps:from_keys/2 from OTP 24 and with maps:from_list/1 below it. The new test case asserts the cost of reconciling a 10 000 entry snapshot in reductions. --- src/syn_registry.erl | 13 +++++-- test/syn_registry_SUITE.erl | 71 +++++++++++++++++++++++++++++++++++-- 2 files changed, 80 insertions(+), 4 deletions(-) diff --git a/src/syn_registry.erl b/src/syn_registry.erl index 8b9e2e31..8ab60bec 100755 --- a/src/syn_registry.erl +++ b/src/syn_registry.erl @@ -615,10 +615,10 @@ reconcile_remote_registry_snapshot(Node, RegistryTuplesOfRemoteNode, #state{ table_by_name = TableByName, table_by_pid = TableByPid }) -> - SnapshotNames = ordsets:from_list([Name || {Name, _Pid, _Meta, _Time} <- RegistryTuplesOfRemoteNode]), + SnapshotNames = snapshot_names(RegistryTuplesOfRemoteNode), ExistingTuples = get_registry_tuples_for_node(Node, TableByName), lists:foreach(fun({Name, Pid, Meta, _Time}) -> - case ordsets:is_element(Name, SnapshotNames) of + case is_map_key(Name, SnapshotNames) of true -> ok; @@ -630,6 +630,15 @@ reconcile_remote_registry_snapshot(Node, RegistryTuplesOfRemoteNode, #state{ end end, ExistingTuples). +-spec snapshot_names([syn_registry_tuple()]) -> #{term() => true}. +- if (?OTP_RELEASE >= 24). +snapshot_names(RegistryTuples) -> + maps:from_keys([Name || {Name, _Pid, _Meta, _Time} <- RegistryTuples], true). +-else. +snapshot_names(RegistryTuples) -> + maps:from_list([{Name, true} || {Name, _Pid, _Meta, _Time} <- RegistryTuples]). +-endif. + -spec handle_registry_sync( Name :: term(), Pid :: pid(), diff --git a/test/syn_registry_SUITE.erl b/test/syn_registry_SUITE.erl index 63651203..12e06793 100644 --- a/test/syn_registry_SUITE.erl +++ b/test/syn_registry_SUITE.erl @@ -48,7 +48,8 @@ three_nodes_custom_event_handler_conflict_resolution/1, three_nodes_update/1, three_nodes_ack_sync_reconciles_registry_snapshot/1, - three_nodes_ack_sync_ordered_delivery/1 + three_nodes_ack_sync_ordered_delivery/1, + three_nodes_ack_sync_reconciles_large_snapshot/1 ]). -export([ four_nodes_concurrency/1 @@ -106,7 +107,8 @@ groups() -> three_nodes_custom_event_handler_conflict_resolution, three_nodes_update, three_nodes_ack_sync_reconciles_registry_snapshot, - three_nodes_ack_sync_ordered_delivery + three_nodes_ack_sync_ordered_delivery, + three_nodes_ack_sync_reconciles_large_snapshot ]}, {four_nodes_registry, [shuffle], [ four_nodes_concurrency @@ -1779,6 +1781,71 @@ three_nodes_ack_sync_reconciles_registry_snapshot(Config) -> {syn_remote_scope_node_up, scope_reconcile, SlaveNode1}} ]). +three_nodes_ack_sync_reconciles_large_snapshot(Config) -> + %% Reconciliation checks every entry already held for the remote node against the + %% names in the incoming snapshot. A list based membership test makes that check + %% quadratic, so a peer holding tens of thousands of names blocks the scope process + %% on every discovery, and every registration on this node waits behind it. + %% The cost is asserted in reductions so that the bound does not depend on the machine. + + SlaveNode1 = proplists:get_value(syn_slave_1, Config), + + ok = syn:start(), + ok = rpc:call(SlaveNode1, syn, start, []), + ok = syn:add_node_to_scopes([scope_reconcile_large]), + ok = rpc:call(SlaveNode1, syn, add_node_to_scopes, [[scope_reconcile_large]]), + + syn_test_suite_helper:assert_scope_subcluster(registry, node(), scope_reconcile_large, [SlaveNode1]), + + EntryCount = 10000, + %% every name is held by its own process, the way a registry of this size looks in practice. + %% Putting all the names on one pid would instead measure the per pid ets bag, which is a + %% different cost and not the one under test here. + Entries = [ + {{name, I}, spawn(SlaveNode1, syn_test_suite_helper, process_main, [])} + || I <- lists:seq(1, EntryCount) + ], + lists:foreach(fun({Name, Pid}) -> + add_to_local_table(scope_reconcile_large, Name, Pid, meta, 1, undefined) + end, Entries), + %% an entry the snapshot does not carry, to keep checking that stale entries are removed + StalePid = syn_test_suite_helper:start_process(SlaveNode1), + add_to_local_table(scope_reconcile_large, "stale-remote", StalePid, stale_meta, 1, undefined), + + %% the last entry carries a new meta, so that waiting for it means the whole snapshot was applied + {LastName, LastPid} = lists:last(Entries), + Snapshot = [{Name, Pid, meta, 2} || {Name, Pid} <- lists:droplast(Entries)] + ++ [{LastName, LastPid, applied_meta, 2}], + + RemoteScopePid = rpc:call(SlaveNode1, erlang, whereis, [syn_registry_scope_reconcile_large]), + true = is_pid(RemoteScopePid), + ScopePid = whereis(syn_registry_scope_reconcile_large), + {reductions, ReductionsBefore} = erlang:process_info(ScopePid, reductions), + + ScopePid ! {'3.0', ack_sync, RemoteScopePid, Snapshot}, + + syn_test_suite_helper:assert_wait( + {LastPid, applied_meta}, + fun() -> syn:lookup(scope_reconcile_large, LastName) end + ), + {reductions, ReductionsAfter} = erlang:process_info(ScopePid, reductions), + Reductions = ReductionsAfter - ReductionsBefore, + ct:pal("reconciled and applied a snapshot of ~p entries in ~p reductions", [EntryCount, Reductions]), + + %% the stale entry is gone and nothing else was dropped + undefined = syn:lookup(scope_reconcile_large, "stale-remote"), + EntryCount = syn:registry_count(scope_reconcile_large, SlaveNode1), + + %% linear reconciliation stays far below one thousand reductions per entry, + %% a quadratic one needs thousands of times more + case Reductions < EntryCount * 1000 of + true -> ok; + false -> ct:fail("reconciliation took ~p reductions for ~p entries", [Reductions, EntryCount]) + end, + + %% the slave node outlives this case and the group is shuffled, so the entry processes go now + lists:foreach(fun(Pid) -> exit(Pid, kill) end, [StalePid | [Pid || {_Name, Pid} <- Entries]]). + four_nodes_concurrency(Config) -> %% get slaves SlaveNode1 = proplists:get_value(syn_slave_1, Config), From e49c2977cfaef717a4dfb3ba262c14ac37e927c6 Mon Sep 17 00:00:00 2001 From: Roberto Ostinelli Date: Tue, 13 Oct 2026 20:27:09 +0200 Subject: [PATCH 2/2] Skip reconciliation when there's nothing to reconcile, and harden the large snapshot test. --- src/syn_registry.erl | 40 ++++++------- test/syn_registry_SUITE.erl | 113 +++++++++++++++++++++--------------- 2 files changed, 85 insertions(+), 68 deletions(-) diff --git a/src/syn_registry.erl b/src/syn_registry.erl index 8ab60bec..4c4a4f4e 100755 --- a/src/syn_registry.erl +++ b/src/syn_registry.erl @@ -615,29 +615,25 @@ reconcile_remote_registry_snapshot(Node, RegistryTuplesOfRemoteNode, #state{ table_by_name = TableByName, table_by_pid = TableByPid }) -> - SnapshotNames = snapshot_names(RegistryTuplesOfRemoteNode), - ExistingTuples = get_registry_tuples_for_node(Node, TableByName), - lists:foreach(fun({Name, Pid, Meta, _Time}) -> - case is_map_key(Name, SnapshotNames) of - true -> - ok; + case get_registry_tuples_for_node(Node, TableByName) of + [] -> + ok; - false -> - remove_from_local_table(Name, Pid, TableByName, TableByPid), - syn_event_handler:call_event_handler(on_process_unregistered, - [Scope, Name, Pid, Meta, {syn_remote_scope_node_up, Scope, Node}] - ) - end - end, ExistingTuples). - --spec snapshot_names([syn_registry_tuple()]) -> #{term() => true}. -- if (?OTP_RELEASE >= 24). -snapshot_names(RegistryTuples) -> - maps:from_keys([Name || {Name, _Pid, _Meta, _Time} <- RegistryTuples], true). --else. -snapshot_names(RegistryTuples) -> - maps:from_list([{Name, true} || {Name, _Pid, _Meta, _Time} <- RegistryTuples]). --endif. + ExistingTuples -> + SnapshotNames = maps:from_list([{Name, true} || {Name, _Pid, _Meta, _Time} <- RegistryTuplesOfRemoteNode]), + lists:foreach(fun({Name, Pid, Meta, _Time}) -> + case is_map_key(Name, SnapshotNames) of + true -> + ok; + + false -> + remove_from_local_table(Name, Pid, TableByName, TableByPid), + syn_event_handler:call_event_handler(on_process_unregistered, + [Scope, Name, Pid, Meta, {syn_remote_scope_node_up, Scope, Node}] + ) + end + end, ExistingTuples) + end. -spec handle_registry_sync( Name :: term(), diff --git a/test/syn_registry_SUITE.erl b/test/syn_registry_SUITE.erl index 12e06793..5d230aa1 100644 --- a/test/syn_registry_SUITE.erl +++ b/test/syn_registry_SUITE.erl @@ -1784,8 +1784,9 @@ three_nodes_ack_sync_reconciles_registry_snapshot(Config) -> three_nodes_ack_sync_reconciles_large_snapshot(Config) -> %% Reconciliation checks every entry already held for the remote node against the %% names in the incoming snapshot. A list based membership test makes that check - %% quadratic, so a peer holding tens of thousands of names blocks the scope process - %% on every discovery, and every registration on this node waits behind it. + %% quadratic, which blocks the scope process (and every registration on this node) when + %% an ack_sync comes in while we still hold that peer's entries, e.g. when a node that + %% connects during scope startup gets discovered twice. %% The cost is asserted in reductions so that the bound does not depend on the machine. SlaveNode1 = proplists:get_value(syn_slave_1, Config), @@ -1796,55 +1797,75 @@ three_nodes_ack_sync_reconciles_large_snapshot(Config) -> ok = rpc:call(SlaveNode1, syn, add_node_to_scopes, [[scope_reconcile_large]]), syn_test_suite_helper:assert_scope_subcluster(registry, node(), scope_reconcile_large, [SlaveNode1]), + syn_test_suite_helper:assert_scope_subcluster(registry, SlaveNode1, scope_reconcile_large, [node()]), EntryCount = 10000, %% every name is held by its own process, the way a registry of this size looks in practice. %% Putting all the names on one pid would instead measure the per pid ets bag, which is a %% different cost and not the one under test here. - Entries = [ - {{name, I}, spawn(SlaveNode1, syn_test_suite_helper, process_main, [])} - || I <- lists:seq(1, EntryCount) - ], - lists:foreach(fun({Name, Pid}) -> - add_to_local_table(scope_reconcile_large, Name, Pid, meta, 1, undefined) - end, Entries), - %% an entry the snapshot does not carry, to keep checking that stale entries are removed - StalePid = syn_test_suite_helper:start_process(SlaveNode1), - add_to_local_table(scope_reconcile_large, "stale-remote", StalePid, stale_meta, 1, undefined), - - %% the last entry carries a new meta, so that waiting for it means the whole snapshot was applied - {LastName, LastPid} = lists:last(Entries), - Snapshot = [{Name, Pid, meta, 2} || {Name, Pid} <- lists:droplast(Entries)] - ++ [{LastName, LastPid, applied_meta, 2}], - - RemoteScopePid = rpc:call(SlaveNode1, erlang, whereis, [syn_registry_scope_reconcile_large]), - true = is_pid(RemoteScopePid), - ScopePid = whereis(syn_registry_scope_reconcile_large), - {reductions, ReductionsBefore} = erlang:process_info(ScopePid, reductions), - - ScopePid ! {'3.0', ack_sync, RemoteScopePid, Snapshot}, - - syn_test_suite_helper:assert_wait( - {LastPid, applied_meta}, - fun() -> syn:lookup(scope_reconcile_large, LastName) end - ), - {reductions, ReductionsAfter} = erlang:process_info(ScopePid, reductions), - Reductions = ReductionsAfter - ReductionsBefore, - ct:pal("reconciled and applied a snapshot of ~p entries in ~p reductions", [EntryCount, Reductions]), - - %% the stale entry is gone and nothing else was dropped - undefined = syn:lookup(scope_reconcile_large, "stale-remote"), - EntryCount = syn:registry_count(scope_reconcile_large, SlaveNode1), - - %% linear reconciliation stays far below one thousand reductions per entry, - %% a quadratic one needs thousands of times more - case Reductions < EntryCount * 1000 of - true -> ok; - false -> ct:fail("reconciliation took ~p reductions for ~p entries", [Reductions, EntryCount]) - end, - - %% the slave node outlives this case and the group is shuffled, so the entry processes go now - lists:foreach(fun(Pid) -> exit(Pid, kill) end, [StalePid | [Pid || {_Name, Pid} <- Entries]]). + %% They are all linked to a single owner on the slave, so killing the owner kills them all. + TestPid = self(), + OwnerPid = syn_test_suite_helper:start_process(SlaveNode1, fun() -> + Pids = [spawn_link(fun syn_test_suite_helper:process_main/0) || _ <- lists:seq(0, EntryCount)], + TestPid ! {entry_pids, self(), Pids}, + syn_test_suite_helper:process_main() + end), + try + [StalePid | EntryPids] = receive + {entry_pids, OwnerPid, Pids} -> Pids + after 5000 -> + ct:fail("entry processes were not started on ~p", [SlaveNode1]) + end, + + %% the slave replies to discovery with its own (empty) ack_sync, which would wipe the entries + %% added below if it got here after them. A register on the slave goes out right behind it + %% through the same sender, so once we see it here we're good. + %% The snapshot below doesn't include this name, so reconciliation has to remove it. + ok = rpc:call(SlaveNode1, syn, register, [scope_reconcile_large, "stale-remote", StalePid, stale_meta]), + syn_test_suite_helper:assert_wait( + {StalePid, stale_meta}, + fun() -> syn:lookup(scope_reconcile_large, "stale-remote") end + ), + + Entries = lists:zip([{name, I} || I <- lists:seq(1, EntryCount)], EntryPids), + lists:foreach(fun({Name, Pid}) -> + add_to_local_table(scope_reconcile_large, Name, Pid, meta, 1, undefined) + end, Entries), + + %% the last entry carries a new meta, so that waiting for it means the whole snapshot was applied + {LastName, LastPid} = lists:last(Entries), + Snapshot = [{Name, Pid, meta, 2} || {Name, Pid} <- lists:droplast(Entries)] + ++ [{LastName, LastPid, applied_meta, 2}], + + RemoteScopePid = rpc:call(SlaveNode1, erlang, whereis, [syn_registry_scope_reconcile_large]), + true = is_pid(RemoteScopePid), + ScopePid = whereis(syn_registry_scope_reconcile_large), + {reductions, ReductionsBefore} = erlang:process_info(ScopePid, reductions), + + ScopePid ! {'3.0', ack_sync, RemoteScopePid, Snapshot}, + + syn_test_suite_helper:assert_wait( + {LastPid, applied_meta}, + fun() -> syn:lookup(scope_reconcile_large, LastName) end + ), + {reductions, ReductionsAfter} = erlang:process_info(ScopePid, reductions), + Reductions = ReductionsAfter - ReductionsBefore, + ct:pal("reconciled and applied a snapshot of ~p entries in ~p reductions", [EntryCount, Reductions]), + + %% the stale entry is gone and nothing else was dropped + undefined = syn:lookup(scope_reconcile_large, "stale-remote"), + EntryCount = syn:registry_count(scope_reconcile_large, SlaveNode1), + + %% applying the snapshot is a few dozen reductions per entry, a quadratic reconciliation + %% adds about EntryCount / 2 per entry: keep EntryCount well above 2x the bound or this + %% stops catching it + case Reductions < EntryCount * 200 of + true -> ok; + false -> ct:fail("reconciliation took ~p reductions for ~p entries", [Reductions, EntryCount]) + end + after + syn_test_suite_helper:kill_process(OwnerPid) + end. four_nodes_concurrency(Config) -> %% get slaves