Check receiver batch boundaries with generated histories

test/README.md

@@ -6,6 +6,7 @@
 mix test                           # all tests
 mix test test/group_test.exs       # local only
 mix test test/group_property_test.exs # local model properties
+mix test test/replication_property_test.exs # receiver protocol properties
 mix test test/distributed_test.exs # distributed only
 ```
 
@@ -15,6 +16,7 @@ mix test test/distributed_test.exs # distributed only
 |------|---------------|
 | `group_test.exs` | Single-node: register/unregister, join/leave, members, monitor/demonitor, named clusters, concurrent operations |
 | `group_property_test.exs` | StreamData local command histories checked against an independent map/set model |
+| `replication_property_test.exs` | Generated receiver batch boundaries checked against a map-based oracle |
 | `distributed_test.exs` | Multi-node: replication, peer discovery, node disconnect cleanup, partition healing, conflict resolution, event ordering, rolling restarts |
 
 ## Local model properties
@@ -57,13 +59,34 @@ mix test test/group_property_test.exs --seed 12345
 
 StreamData reports the shrunk command sequence. Promote discovered failures to
 focused regression tests. A seed reproduces generation, not BEAM scheduling.
-This suite deliberately settles between commands: it does **not** test remote
-batch boundaries, equal-timestamp conflicts, stale replication after disconnect,
-TTL, or fairness under remote load. Those need separate targeted properties on
-top of the existing distributed tests, not more machinery in this local model.
+The local model deliberately settles between commands. Receiver buffering is
+covered separately below, without growing the command model into a scheduler
+or transport simulator.
 If the model grows into a substantial state-machine framework, evaluate
 PropCheck/PropEr rather than implementing that framework here.
 
+## Targeted protocol properties
+
+`replication_property_test.exs` exercises both registry and PG receiver lanes:
+
+- **Batch boundaries:** generate ordered writes, updates, and removals over hot
+  keys, then partition each history into singleton, buffer-minus-one, buffer,
+  buffer-plus-one, and whole-history messages. Each partition gets a fresh Group
+  instance and is checked against a map-based oracle for contents and per-key
+  event order, including metadata and removal reasons. Runs 100 examples per lane.
+
+These are controlled receiver-protocol scenarios using real local owner PIDs;
+they do not traverse Erlang distribution or exercise sender batching. Registry
+keys retain one owner, leaving owner-conflict resolution to separate scenarios.
+The distributed suite remains necessary for real peer lifecycle and transport
+behavior. These properties do not claim reproducible network races or fairness
+under an infinite stream of control messages.
+
+`Group.PropertyFixture` owns the Group supervisor, owners, and a mailbox-preserving
+observer per example/shrink attempt. It tears them down in `after`, stops
+subscribers before Registry, and uses observer-addressed mailbox barriers for
+event delivery. Fixed names avoid atom growth during shrinking.
+
 ## How distribution works
 
 The test node starts as a named Erlang node in `test_helper.exs`:

test/replication_property_test.exs

@@ -0,0 +1,169 @@
+defmodule Group.ReplicationPropertyTest do
+  use ExUnit.Case, async: false
+  use ExUnitProperties
+
+  import Group.PropertyFixture
+  alias Group.Replica
+  alias Group.TestCluster
+
+  @name :replication_property
+  @keys ["hot/a", "hot/b", "hot/nested/a", "other/a"]
+
+  for kind <- [:registry, :pg] do
+    property "#{kind} batch boundaries preserve settled contents and per-key events" do
+      check all(
+              shards <- member_of([1, 2, 4]),
+              buffer <- member_of([2, 4, 8]),
+              cluster <- member_of([nil, "shared"]),
+              count <- member_of([buffer - 1, buffer, buffer + 1, 2 * buffer + 1]),
+              commands <-
+                list_of(
+                  tuple(
+                    {member_of([:put, :put, :remove]), integer(0..3), integer(0..1),
+                     integer(0..2)}
+                  ),
+                  length: count
+                ),
+              max_runs: 100
+            ) do
+        # Compare each partition against a map-based oracle, not merely another
+        # Group run that could make the same mistake. Include one entire batch.
+        for chunk_size <- Enum.uniq([1, buffer - 1, buffer, buffer + 1, count]) do
+          with_group(@name, options(shards, buffer), fn actors ->
+            if cluster, do: Group.connect(@name, cluster)
+            subscribe(actors.observer, cluster)
+
+            {ops, entries, expected_events} =
+              compile_history(unquote(kind), cluster, commands, actors)
+
+            deliver(unquote(kind), ops, chunk_size, shards)
+            actual_events = events_after_barrier(@name, actors.observer)
+            assert_per_key_events(actual_events, expected_events)
+            assert_entries(entries, unquote(kind), cluster)
+            assert :ok = TestCluster.assert_ets_consistent(@name)
+          end)
+        end
+      end
+    end
+  end
+
+  defp options(shards, buffer) do
+    [
+      shards: shards,
+      log: false,
+      replicated_pg_receiver_buffer_size: buffer,
+      replicated_registry_receiver_buffer_size: buffer,
+      replicated_pg_receiver_flush_interval: 60_000,
+      replicated_registry_receiver_flush_interval: 60_000
+    ]
+  end
+
+  defp subscribe(observer, cluster) do
+    assert :ok = in_process(observer, fn -> Group.monitor(@name, :all, cluster: cluster) end)
+  end
+
+  # Build well-ordered wire histories with monotonic timestamps. Registry keys
+  # keep one owner (conflicts are a separate property); PG keys can have two.
+  defp compile_history(kind, cluster, commands, actors) do
+    {ops, entries, events} =
+      commands
+      |> Enum.with_index(1)
+      |> Enum.reduce({[], %{}, []}, fn {{action, key_id, owner, value}, time},
+                                       {ops, entries, events} ->
+        key = Enum.at(@keys, key_id)
+        pid = actors[if(kind == :registry, do: rem(key_id, 2), else: owner)]
+        meta = %{v: value}
+        previous = Map.get(entries, {key, pid})
+
+        case action do
+          :put ->
+            op =
+              case kind do
+                :registry ->
+                  {:register, cluster, key, pid, meta, time, node(pid)}
+
+                :pg ->
+                  reason = if previous, do: :update, else: :join
+                  {:join, cluster, key, pid, meta, time, reason, node(pid)}
+              end
+
+            e = event(kind, cluster, key, pid, meta, previous, nil)
+            {[op | ops], Map.put(entries, {key, pid}, meta), [e | events]}
+
+          :remove ->
+            reason = if kind == :registry, do: :unregister, else: :leave
+            op = {reason, cluster, key, pid, previous || meta, reason}
+
+            events =
+              if previous,
+                do: [event(kind, cluster, key, pid, previous, nil, reason) | events],
+                else: events
+
+            {[op | ops], Map.delete(entries, {key, pid}), events}
+        end
+      end)
+
+    {Enum.reverse(ops), entries, Enum.reverse(events)}
+  end
+
+  defp event(kind, cluster, key, pid, meta, previous, reason) do
+    type =
+      if reason,
+        do: removal_type(kind),
+        else: if(kind == :registry, do: :registered, else: :joined)
+
+    %Group.Event{
+      supervisor: @name,
+      type: type,
+      cluster: cluster,
+      key: key,
+      pid: pid,
+      meta: meta,
+      previous_meta: previous,
+      reason: reason
+    }
+  end
+
+  defp removal_type(:registry), do: :unregistered
+  defp removal_type(:pg), do: :left
+
+  defp deliver(kind, ops, chunk_size, shards) do
+    tag = if kind == :registry, do: :replicate_registry_batch, else: :replicate_pg_batch
+
+    ops
+    |> Enum.group_by(fn op -> Replica.shard_index_for(elem(op, 1), elem(op, 2), shards) end)
+    |> Enum.each(fn {shard, shard_ops} ->
+      for chunk <- Enum.chunk_every(shard_ops, chunk_size) do
+        send(Replica.shard_name(@name, shard), {tag, chunk})
+      end
+
+      # Order our sends before the observer's barrier (a different sender).
+      # This system message does not itself flush a partial receiver buffer.
+      :sys.get_state(Replica.shard_name(@name, shard))
+    end)
+  end
+
+  defp assert_per_key_events(actual, expected) do
+    group = fn events -> Enum.group_by(events, &{&1.cluster, &1.key}) end
+    assert group.(actual) == group.(expected)
+  end
+
+  defp assert_entries(entries, kind, cluster) do
+    expected =
+      Enum.map(entries, fn {{key, pid}, meta} -> {kind, cluster, key, pid, meta} end)
+
+    assert Enum.sort(Group.local_entries(@name)) == Enum.sort(expected)
+
+    for key <- @keys do
+      members = for {{^key, pid}, meta} <- entries, do: {pid, meta}
+
+      case kind do
+        :registry ->
+          assert Group.lookup(@name, key, cluster: cluster) == List.first(members)
+
+        :pg ->
+          assert Enum.sort(Group.members(@name, key, cluster: cluster)) == Enum.sort(members)
+      end
+    end
+  end
+end

test/support/property_fixture.ex

@@ -0,0 +1,92 @@
+defmodule Group.PropertyFixture do
+  @moduledoc false
+
+  import ExUnit.Callbacks
+
+  # Invoke inside check all, not setup: failed examples and shrink candidates
+  # must not share owners, subscriptions, tables, buffers, or lease timers.
+  def with_group(name, opts, fun) do
+    children =
+      [{Group, Keyword.put(opts, :name, name)}] ++
+        for id <- [:observer, 0, 1] do
+          Supervisor.child_spec({Task, &actor_loop/0}, id: id)
+        end
+
+    id = {__MODULE__, name}
+
+    supervisor =
+      start_supervised!(%{
+        id: id,
+        start: {Supervisor, :start_link, [children, [strategy: :one_for_one]]},
+        type: :supervisor
+      })
+
+    try do
+      actors = Map.new(Supervisor.which_children(supervisor), fn {id, pid, _, _} -> {id, pid} end)
+      fun.(actors)
+    after
+      # Reverse child shutdown stops the subscriber before Registry. The ExUnit
+      # process never subscribes, so Registry shutdown cannot abort shrinking.
+      stop_supervised!(id)
+      :persistent_term.erase({Group, name})
+    end
+  end
+
+  def in_process(pid, fun) do
+    ref = Process.monitor(pid)
+    send(pid, {:run, self(), ref, fun})
+
+    try do
+      receive do
+        {^ref, {:ok, result}} -> result
+        {^ref, {:error, error, stacktrace}} -> reraise error, stacktrace
+        {:DOWN, ^ref, :process, ^pid, reason} -> raise "property actor exited: #{inspect(reason)}"
+      after
+        5_000 -> raise "property actor timed out"
+      end
+    after
+      Process.demonitor(ref, [:flush])
+    end
+  end
+
+  defp actor_loop do
+    # Selective receive preserves event messages until the observer drains them.
+    receive do
+      {:run, caller, ref, fun} ->
+        result =
+          try do
+            {:ok, fun.()}
+          rescue
+            error -> {:error, error, __STACKTRACE__}
+          end
+
+        send(caller, {ref, result})
+        actor_loop()
+    end
+  end
+
+  def events_after_barrier(name, observer) do
+    in_process(observer, fn ->
+      for shard <- 0..(Group.get_config(name).num_shards - 1) do
+        ref = make_ref()
+        send(Group.Replica.shard_name(name, shard), {:group_dispatch, [self()], {:settled, ref}})
+
+        receive do
+          {:settled, ^ref} -> :ok
+        after
+          1_000 -> raise "property barrier timed out"
+        end
+      end
+
+      drain_events(name)
+    end)
+  end
+
+  defp drain_events(name) do
+    receive do
+      {:group, events, %{name: ^name}} -> events ++ drain_events(name)
+    after
+      0 -> []
+    end
+  end
+end