Recover a replica lane whose peer's hello was lost

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,

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

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)
 

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,

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])