Track activation requests in partition identity tests (#2267)

Author: rogeralsing

tests/Proto.Cluster.PartitionIdentity.Tests/PartitionIdentityTests.cs

@@ -12,6 +12,7 @@
 using System.Threading.Tasks;
 using ClusterTest.Messages;
 using FluentAssertions;
+using Proto.Cluster;
 using Proto.Cluster.Identity;
 using Proto.Cluster.Partition;
 using Proto.Cluster.Tests;
@@ -91,12 +92,20 @@ PartitionIdentityLookup.Send send
         var actorStates = fixture.Repository.Contents.ToList();
 
         var totalCalls = actorStates.Select(it => it.TotalCount).Sum();
-        var restarts = actorStates.Select(it => it.Events.Count(it => it is ActorStopped) - 1).Sum();
+        var restarts = actorStates.Select(it => it.Events.Count(e => e is ActorStopped) - 1).Sum();
+        var totalActivationRequests =
+            actorStates.Select(it => it.Events.Count(e => e is ActivationRequested)).Sum();
 
-        _output.WriteLine($"{totalCalls} requests, {restarts} restarts against " + actorStates.Count + " identities");
+        _output.WriteLine(
+            $"{totalCalls} requests, {restarts} restarts, {totalActivationRequests} activation requests against " +
+            actorStates.Count + " identities");
 
         foreach (var actorState in actorStates)
         {
+            var activationReqs = actorState.Events.Count(e => e is ActivationRequested);
+            var starts = actorState.Events.Count(e => e is ActorStarted);
+            activationReqs.Should().Be(starts);
+
             if (actorState.Inconsistent)
             {
                 Assert.False(actorState.Inconsistent, actorState.ToString());
@@ -298,12 +307,22 @@ protected override ClusterKind[] ClusterKinds
         };
 
     protected override IIdentityLookup GetIdentityLookup(string clusterName) =>
-        new PartitionIdentityLookup(new PartitionConfig
-        {
-            GetPidTimeout = TimeSpan.FromSeconds(5),
-            HandoverChunkSize = _chunkSize,
-            RebalanceRequestTimeout = TimeSpan.FromSeconds(3),
-            Mode = _mode,
-            Send = _send
-        });
+        new PartitionIdentityLookup(
+            new PartitionConfig
+            {
+                GetPidTimeout = TimeSpan.FromSeconds(5),
+                HandoverChunkSize = _chunkSize,
+                RebalanceRequestTimeout = TimeSpan.FromSeconds(3),
+                Mode = _mode,
+                Send = _send
+            },
+            props => props.WithReceiverMiddleware(next => async (ctx, env) =>
+            {
+                if (env.Message is ActivationRequest req)
+                {
+                    Repository.Get(req.Identity, this).RecordActivationRequest(ctx.System.Id);
+                }
+
+                await next(ctx, env);
+            }));
 }

tests/Proto.Cluster.Tests/ConcurrencyVerificationActor.cs

@@ -105,6 +105,9 @@ public record ActorStarted(string Member, PID Activation, DateTimeOffset When, i
 public record ActorStopped(string Member, PID Activation, DateTimeOffset When, int StoredCount, long GlobalCount)
     : VerificationEvent(Activation, When);
 
+public record ActivationRequested(string Member, DateTimeOffset When)
+    : VerificationEvent(null!, When);
+
 public record ConsistencyError(
         PID Activation,
         DateTimeOffset When,
@@ -139,6 +142,9 @@ public ActorState(string id, IClusterFixture clusterFixture)
     public long TotalCount => Interlocked.Read(ref _totalCount);
     public ConcurrentBag<VerificationEvent> Events { get; } = new();
 
+    public void RecordActivationRequest(string member) =>
+        Events.Add(new ActivationRequested(member, DateTimeOffset.Now));
+
     public void RecordStarted(IContext context)
     {
         lock (_padlock)
@@ -200,6 +206,7 @@ private string EventsToString() =>
                 {
                     ActorStarted started     => $"[{started.When:O}] Actor started on member {started.Member}\n",
                     ActorStopped stopped     => $"[{stopped.When:O}] Actor stopped on member {stopped.Member}\n",
+                    ActivationRequested req => $"[{req.When:O}] Activation requested on member {req.Member}\n",
                     ClusterSnapshot snapshot => $"[{snapshot.When:O}] Cluster snapshot:\n{snapshot.Snapshot}\n",
                     ConsistencyError err =>
                         $"[{err.When:O}] Consistency error, actual: {err.ActualCount}, expected: {err.ExpectedCount}, stored: {err.StoredCount}, global: {err.GlobalCount}",