Check local write fairness at receiver flush boundaries

test/README.md

@@ -16,7 +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 and buffered/snapshot cluster isolation |
+| `replication_property_test.exs` | Generated receiver batch boundaries, buffered/snapshot cluster isolation, and local-write fairness |
 | `distributed_test.exs` | Multi-node: replication, peer discovery, node disconnect cleanup, partition healing, conflict resolution, event ordering, rolling restarts |
 
 ## Local model properties
@@ -59,9 +59,9 @@ 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.
-The local model deliberately settles between commands. Receiver buffering is
-covered separately below, without growing the command model into a scheduler
-or transport simulator.
+The local model deliberately settles between commands. Receiver buffering and
+fairness are 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.
 
@@ -79,6 +79,11 @@ PropCheck/PropEr rather than implementing that framework here.
   may repopulate the disconnected cluster or emit transient apply-then-purge
   events. Reconnect cycles must accept fresh snapshots while preserving entries
   in the default and another named cluster. Runs 50 examples per lane.
+- **Fairness:** suspend one shard, enqueue a finite receiver backlog, and wait
+  until a public local register/join request is also queued. On resume, the local
+  event must appear at the first receiver flush, before the backlog finishes.
+  The assertion uses the shard's event timeline rather than a wall-clock
+  latency threshold or queue-length observation. Runs 50 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

test/replication_property_test.exs

@@ -130,6 +130,73 @@ defmodule Group.ReplicationPropertyTest do
         end)
       end
     end
+
+    property "#{kind} backlog yields to a queued local write at the first receiver flush" do
+      check all(
+              buffer <- member_of([1, 2, 4, 8]),
+              chunk_size <- member_of(Enum.uniq([1, max(buffer - 1, 1), buffer, buffer + 1])),
+              count <- member_of([2 * buffer + chunk_size, 4 * buffer + chunk_size]),
+              local_operation <- member_of([:register, :join]),
+              max_runs: 50
+            ) do
+        # One shard gives a single event timeline. Assert the position of local
+        # work in that timeline, not the scheduler-sensitive queue length when
+        # the caller happens to wake up.
+        with_group(@name, options(1, buffer), fn actors ->
+          subscribe(actors.observer, nil)
+          commands = for i <- 1..count, do: {:put, 0, 0, i}
+          {ops, entries, remote_events} = compile_history(unquote(kind), nil, commands, actors)
+
+          tag =
+            if unquote(kind) == :registry,
+              do: :replicate_registry_batch,
+              else: :replicate_pg_batch
+
+          shard = Process.whereis(Replica.shard_name(@name, 0))
+          :ok = :sys.suspend(shard)
+
+          try do
+            for chunk <- Enum.chunk_every(ops, chunk_size), do: send(shard, {tag, chunk})
+
+            caller =
+              Task.async(fn ->
+                in_process(actors[1], fn ->
+                  apply(Group, local_operation, [@name, "local/a", %{local: true}])
+                end)
+              end)
+
+            try do
+              TestCluster.assert_eventually(
+                fn ->
+                  {:messages, messages} = Process.info(shard, :messages)
+                  Enum.any?(messages, &(is_tuple(&1) and elem(&1, 0) == :group_local_request))
+                end,
+                interval: 1
+              )
+
+              :ok = :sys.resume(shard)
+              assert :ok = Task.await(caller, 5_000)
+              events = events_after_barrier(@name, actors.observer)
+              first_flush = div(buffer + chunk_size - 1, chunk_size) * chunk_size
+              assert Enum.find_index(events, &(&1.key == "local/a")) == first_flush
+              assert first_flush < count
+              assert Enum.reject(events, &(&1.key == "local/a")) == remote_events
+
+              local_kind = if local_operation == :register, do: :registry, else: :pg
+              local_event = event(local_kind, nil, "local/a", actors[1], %{local: true}, nil, nil)
+              assert Enum.filter(events, &(&1.key == "local/a")) == [local_event]
+              local_entry = {local_kind, nil, "local/a", actors[1], %{local: true}}
+              assert_entries(entries, unquote(kind), nil, [local_entry])
+              assert :ok = TestCluster.assert_ets_consistent(@name)
+            after
+              Task.shutdown(caller, :brutal_kill)
+            end
+          after
+            :sys.resume(shard)
+          end
+        end)
+      end
+    end
   end
 
   defp options(shards, buffer) do