diff --git a/CHANGELOG.md b/CHANGELOG.md index 4b97e61..ddf0aa4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,13 @@ ## Unreleased +- Fix a replica lane that never recovered after its lease expired and the + peer's `replica_hello` was dropped by a busy link. The peer answered every + repeated probe with a constant-size ACK because its own route was ready, + while this side kept the probe pending forever because the ACK could not + certify a route it never installed, so no hello was ever re-sent. A ready + ACK that finds no route now requests the peer's authority + (`replica_hello_request`), and a hello refused by a busy link is no longer + recorded as sent. - Materialize per-shard exact and slash-prefix process-group cardinalities so `Group.member_count/3` and `Group.local_member_count/3` no longer scan memberships. Counts follow resident-row diffs across local writes, diff --git a/lib/group/replica.ex b/lib/group/replica.ex index e3f6510..519ee33 100644 --- a/lib/group/replica.ex +++ b/lib/group/replica.ex @@ -75,6 +75,10 @@ defmodule Group.Replica do - constant-size periodic heartbeats provide a bounded peer lease without creating remote process monitors. A generation/epoch-revision mismatch requests a fresh authoritative hello. + - every message above is a nonblocking send that a busy link drops. A probe + answered as ready while this side still holds no route asks the peer for + its hello again (replica_hello_request), so one lost hello cannot leave a + lane unrouted for the life of the connection. Data installs an exact remote authority and its shared-cluster dual-index projection in one serialized operation. A concurrent local cluster connect @@ -1274,13 +1278,24 @@ defmodule Group.Replica do first_ack? = Map.get(state.peer_connect_ack_seen, remote_node) != remote_pid + # The receiver certifies its own route and answers a repeated probe with + # this ACK alone, so a lost hello leaves this side without a route while + # every later ACK says the receiver is ready. Ask for the hello + # explicitly: the request is constant-size and the receiver rate-limits + # the catalog it answers with. + route_missing? = + route_ready? and Map.get(state.remote_shards, remote_node) != remote_pid + state = %{ state | peer_connect_ack_seen: Map.put(state.peer_connect_ack_seen, remote_node, remote_pid), pending_peer_probes: pending_peer_probes } - {:noreply, maybe_send_discovery_hello(state, remote_node, first_ack?)} + state = maybe_send_discovery_hello(state, remote_node, first_ack?) + + {:noreply, + if(route_missing?, do: request_replica_authority(state, remote_node), else: state)} end end @@ -3307,8 +3322,6 @@ defmodule Group.Replica do Data.local_cluster_epoch_revision(state.name), state.replica_transport.id(), descriptor} ) end - - state end defp peer_connect_message(state, target_node) do @@ -3335,13 +3348,19 @@ defmodule Group.Replica do if force? or is_nil(last_sent) or elem(last_sent, 1) != authority or now - elem(last_sent, 0) >= retry_interval do - state = send_replica_hello(state, target_node) + # A hello refused by a busy link was never sent; recording it would + # silence the next request for a whole retry interval. + case send_replica_hello(state, target_node) do + :ok -> + %{ + state + | discovery_hello_last_sent: + Map.put(state.discovery_hello_last_sent, target_node, {now, authority}) + } - %{ - state - | discovery_hello_last_sent: - Map.put(state.discovery_hello_last_sent, target_node, {now, authority}) - } + :busy -> + state + end else state end diff --git a/test/distributed_test.exs b/test/distributed_test.exs index c85f0fe..2c8a09e 100644 --- a/test/distributed_test.exs +++ b/test/distributed_test.exs @@ -4410,10 +4410,16 @@ defmodule Group.DistributedTest do ["authority-new"] ]) + # A lane whose hello outran shard zero's exact authority has already + # re-probed once; an ACK is only honored for the lane's current epoch. Enum.each(b_lanes, fn {shard, b_pid} -> + a_pid = TestCluster.rpc!(node_a, Process, :whereis, [shard_name(name, shard)]) + a_state = TestCluster.rpc!(node_a, :sys, :get_state, [a_pid]) + probe_epoch = Map.get(a_state.peer_probe_epochs, node_b, 0) + TestCluster.rpc!(node_a, :erlang, :send, [ shard_name(name, shard), - {:peer_connect_ack, b_pid, shard, shards, 0, true} + {:peer_connect_ack, b_pid, shard, shards, probe_epoch, true} ]) end) diff --git a/test/formal/check_replica_ack_mutations.exs b/test/formal/check_replica_ack_mutations.exs index a1b3a8d..5d74fa5 100644 --- a/test/formal/check_replica_ack_mutations.exs +++ b/test/formal/check_replica_ack_mutations.exs @@ -2,21 +2,19 @@ defmodule Group.Formal.ReplicaAckQualification do @moduledoc false @mutations [ - {"stale ACK token", " /\\ m.tok = s.sendToken\n", "", "INVARIANT", - "QuietHasCurrentAck", 12, "Invariant QuietHasCurrentAck is violated"}, + {"stale ACK token", " /\\ m.tok = s.sendToken\n", "", "INVARIANT", "QuietHasCurrentAck", 12, + "Invariant QuietHasCurrentAck is violated"}, {"stale need full send", " s.pending /\\ m.aux = s.head /\\ m.n <= s.head\n", " m.aux = s.head /\\ m.n <= s.head\n", "INVARIANT", "NoUnjustifiedFullSend", 12, "Invariant NoUnjustifiedFullSend is violated"}, - {"gratuitous full hello", " /\\ s.helloDue\n", "", "INVARIANT", - "NoUnjustifiedFullSend", 12, "Invariant NoUnjustifiedFullSend is violated"}, + {"gratuitous full hello", " /\\ s.helloDue\n", "", "INVARIANT", "NoUnjustifiedFullSend", 12, + "Invariant NoUnjustifiedFullSend is violated"}, {"lost recovery probe after a stale hello", - "ProbeAllowed == ~s.connected \\/ s.probePending\n", - "ProbeAllowed == ~s.connected\n", "PROPERTY", "HealedConvergence", 13, - "Temporal property HealedConvergence was violated"}, + "ProbeAllowed == ~s.connected \\/ s.probePending\n", "ProbeAllowed == ~s.connected\n", + "PROPERTY", "HealedConvergence", 13, "Temporal property HealedConvergence was violated"}, {"premature ACK for an uninstalled receiver PID", - "m.probe = s.probe /\\ m.aux = 1 /\\ s.connected", - "m.probe = s.probe /\\ s.connected", "PROPERTY", "HealedConvergence", 13, - "Temporal property HealedConvergence was violated"}, + "m.probe = s.probe /\\ m.aux = 1 /\\ s.connected", "m.probe = s.probe /\\ s.connected", + "PROPERTY", "HealedConvergence", 13, "Temporal property HealedConvergence was violated"}, {"premature ACK for stale receiver authority", "m.epoch = s.knownEpoch\n THEN 1 ELSE 0", "TRUE\n THEN 1 ELSE 0", "PROPERTY", "HealedConvergence", 13, diff --git a/test/replica_ack_test.exs b/test/replica_ack_test.exs index 58fc91b..4f1f4d7 100644 --- a/test/replica_ack_test.exs +++ b/test/replica_ack_test.exs @@ -427,6 +427,225 @@ defmodule Group.ReplicaAckTest do assert 6 == queued_replica_hellos(context.receiver, receiver_shard) end + test "a ready ACK without a route asks the receiver for its hello", context do + # Two shards: expiring one lane then retires only that lane and keeps the + # node's authority, so the remaining repair paths (heartbeats, heads) all + # see "authority known, view missing" and stay quiet, as in production. + name = :"replica_ack_lanes_#{System.unique_integer([:positive])}" + + opts = [ + name: name, + shards: 2, + replica_transport: Group.TestReplicaTransport, + replicated_anti_entropy_interval: 25, + replicated_peer_lease_timeout: 5_000 + ] + + for node <- [context.source, context.receiver] do + {:ok, _pid} = TestCluster.start_group(node, opts) + :ok = TestCluster.rpc!(node, Group, :connect, [name, "org"]) + end + + on_exit(fn -> + for node <- [context.source, context.receiver] do + TestCluster.rpc!(node, Group.TestReplicaTransport, :clear, [name]) + end + end) + + source_shard = + TestCluster.rpc!(context.source, Process, :whereis, [Group.Replica.shard_name(name, 0)]) + + receiver_shard = + TestCluster.rpc!(context.receiver, Process, :whereis, [Group.Replica.shard_name(name, 0)]) + + shard_name = Group.Replica.shard_name(name, 0) + + # Both control lanes are routed and current before the fault. + TestCluster.assert_eventually(fn -> + source_state = TestCluster.rpc!(context.source, :sys, :get_state, [source_shard]) + receiver_state = TestCluster.rpc!(context.receiver, :sys, :get_state, [receiver_shard]) + + Map.get(source_state.remote_shards, context.receiver) == receiver_shard and + Map.get(receiver_state.remote_shards, context.source) == source_shard and + view_current?(context.receiver, name, context.source) and + view_current?(context.source, name, context.receiver) + end) + + # The source's lease on the receiver's control lane runs out, as it does + # when a busy link drops heartbeats. The helper withholds the probe the + # timer would send, so the loss below is deterministic. + _ = + TestCluster.rpc!(context.source, TestCluster, :expire_replica_lane_without_probe, [ + name, + 0, + context.receiver + ]) + + source_state = TestCluster.rpc!(context.source, :sys, :get_state, [source_shard]) + probe_epoch = Map.fetch!(source_state.pending_peer_probes, context.receiver) + refute Map.has_key?(source_state.remote_shards, context.receiver) + + # The lane is gone but the receiver's authority is still known here. + refute is_nil( + TestCluster.rpc!(context.source, Group.Replica.Data, :remote_generation, [ + name, + context.receiver + ]) + ) + + {generation, revision, _epochs} = + TestCluster.rpc!(context.source, Group.Replica.Data, :local_replica_authority, [name]) + + probe = {:peer_connect, source_shard, 0, 2, probe_epoch, generation, revision} + + # The receiver processes that probe epoch, but its hello and ACK are lost + # on the way back: nothing is registered under the source shard's name + # while they arrive, so the node drops them as a busy link would. + true = TestCluster.rpc!(context.source, Process, :unregister, [shard_name]) + + try do + send(receiver_shard, probe) + _state = TestCluster.rpc!(context.receiver, :sys, :get_state, [receiver_shard]) + Process.sleep(200) + after + true = TestCluster.rpc!(context.source, Process, :register, [source_shard, shard_name]) + end + + # The receiver now treats this epoch as seen and still has its own route, + # so it answers every repeat with an ACK alone; the source has no route and + # the probe pending. This is the wedge observed in production. + receiver_state = TestCluster.rpc!(context.receiver, :sys, :get_state, [receiver_shard]) + + assert {^source_shard, ^probe_epoch} = + Map.fetch!(receiver_state.remote_probe_epochs, context.source) + + assert Map.get(receiver_state.remote_shards, context.source) == source_shard + assert view_current?(context.receiver, name, context.source) + + source_state = TestCluster.rpc!(context.source, :sys, :get_state, [source_shard]) + refute Map.has_key?(source_state.remote_shards, context.receiver) + assert Map.fetch!(source_state.pending_peer_probes, context.receiver) == probe_epoch + + # The source re-sends the same probe epoch (by hand here: the expiry helper + # stopped its anti-entropy timer). The lane must come back: an ACK that + # finds no route requests the missed hello, and the next ACK settles the + # probe against the installed route. + TestCluster.assert_eventually( + fn -> + send(receiver_shard, probe) + _state = TestCluster.rpc!(context.receiver, :sys, :get_state, [receiver_shard]) + _state = TestCluster.rpc!(context.source, :sys, :get_state, [source_shard]) + state = TestCluster.rpc!(context.source, :sys, :get_state, [source_shard]) + + Map.get(state.remote_shards, context.receiver) == receiver_shard and + not Map.has_key?(state.pending_peer_probes, context.receiver) + end, + timeout: 4_000 + ) + + source_state = TestCluster.rpc!(context.source, :sys, :get_state, [source_shard]) + assert Map.has_key?(source_state.peer_last_seen, context.receiver) + assert source_state.authority_request_tokens == %{} + assert view_current?(context.source, name, context.receiver) + end + + test "a ready ACK without a route rebuilds a data lane from the shared authority", context do + # Same wedge on a data lane (shard 1). Its repair differs from shard zero's: + # the ACK makes the lane rebuild its view from the authority shard zero + # still holds, and the peer's next heartbeat reinstalls the route. + name = :"replica_ack_data_lane_#{System.unique_integer([:positive])}" + + opts = [ + name: name, + shards: 2, + replica_transport: Group.TestReplicaTransport, + replicated_anti_entropy_interval: 25, + replicated_peer_lease_timeout: 5_000 + ] + + for node <- [context.source, context.receiver] do + {:ok, _pid} = TestCluster.start_group(node, opts) + :ok = TestCluster.rpc!(node, Group, :connect, [name, "org"]) + end + + on_exit(fn -> + for node <- [context.source, context.receiver] do + TestCluster.rpc!(node, Group.TestReplicaTransport, :clear, [name]) + end + end) + + shard_name = Group.Replica.shard_name(name, 1) + source_lane = TestCluster.rpc!(context.source, Process, :whereis, [shard_name]) + receiver_lane = TestCluster.rpc!(context.receiver, Process, :whereis, [shard_name]) + + TestCluster.assert_eventually(fn -> + source_state = TestCluster.rpc!(context.source, :sys, :get_state, [source_lane]) + receiver_state = TestCluster.rpc!(context.receiver, :sys, :get_state, [receiver_lane]) + + Map.get(source_state.remote_shards, context.receiver) == receiver_lane and + Map.get(receiver_state.remote_shards, context.source) == source_lane and + view_current?(context.receiver, name, context.source, 1) and + view_current?(context.source, name, context.receiver, 1) + end) + + _ = + TestCluster.rpc!(context.source, TestCluster, :expire_replica_lane_without_probe, [ + name, + 1, + context.receiver + ]) + + source_state = TestCluster.rpc!(context.source, :sys, :get_state, [source_lane]) + probe_epoch = Map.fetch!(source_state.pending_peer_probes, context.receiver) + refute Map.has_key?(source_state.remote_shards, context.receiver) + + # Shard zero's lane survived, so the receiver's authority is still exact + # here; only this lane's view and route are gone. + assert view_current?(context.source, name, context.receiver, 0) + refute view_current?(context.source, name, context.receiver, 1) + + {generation, revision, _epochs} = + TestCluster.rpc!(context.source, Group.Replica.Data, :local_replica_authority, [name]) + + probe = {:peer_connect, source_lane, 1, 2, probe_epoch, generation, revision} + + true = TestCluster.rpc!(context.source, Process, :unregister, [shard_name]) + + try do + send(receiver_lane, probe) + _state = TestCluster.rpc!(context.receiver, :sys, :get_state, [receiver_lane]) + Process.sleep(200) + after + true = TestCluster.rpc!(context.source, Process, :register, [source_lane, shard_name]) + end + + receiver_state = TestCluster.rpc!(context.receiver, :sys, :get_state, [receiver_lane]) + + assert {^source_lane, ^probe_epoch} = + Map.fetch!(receiver_state.remote_probe_epochs, context.source) + + assert Map.get(receiver_state.remote_shards, context.source) == source_lane + + source_state = TestCluster.rpc!(context.source, :sys, :get_state, [source_lane]) + refute Map.has_key?(source_state.remote_shards, context.receiver) + assert Map.fetch!(source_state.pending_peer_probes, context.receiver) == probe_epoch + + TestCluster.assert_eventually( + fn -> + send(receiver_lane, probe) + _state = TestCluster.rpc!(context.receiver, :sys, :get_state, [receiver_lane]) + _state = TestCluster.rpc!(context.source, :sys, :get_state, [source_lane]) + state = TestCluster.rpc!(context.source, :sys, :get_state, [source_lane]) + + Map.get(state.remote_shards, context.receiver) == receiver_lane and + not Map.has_key?(state.pending_peer_probes, context.receiver) + end, + timeout: 4_000 + ) + + assert view_current?(context.source, name, context.receiver, 1) + end + test "a heartbeat that restores a lane advertises writes missed without a route", context do source_shard = Group.Replica.shard_name(context.name, 0) @@ -992,6 +1211,23 @@ defmodule Group.ReplicaAckTest do end) end + # A node's shard-0 view of a peer is current when its stored authority and + # the shard's view agree: the conditions behind route_ready? in the replica. + defp view_current?(node, name, peer, shard \\ 0) do + read = fn fun, args -> TestCluster.rpc!(node, Group.Replica.Data, fun, args) end + + generation = read.(:remote_generation, [name, peer]) + observed = read.(:remote_cluster_epoch_observed_revision, [name, peer]) + + not is_nil(generation) and + read.(:remote_replica_authority_hint, [name, peer]) == {generation, observed} and + read.(:remote_cluster_epoch_revision, [name, peer]) == observed and + read.(:remote_view_generation, [name, shard, peer]) == generation and + read.(:remote_view_cluster_epoch_revision, [name, shard, peer]) == + read.(:remote_cluster_epoch_exact_revision, [name, peer]) and + read.(:remote_view_observed_revision, [name, shard, peer]) == observed + end + defp queued_replica_hellos(node, receiver_shard) do {:messages, messages} = TestCluster.rpc!(node, Process, :info, [receiver_shard, :messages])