Fixed TopicActor problem with nullref when loading subscribers (#1716)

* fixed problem with nullref when loading subscribers

* test for default registration

* limit cluster size to 1 for the empty kv store

Author: marcinbudny

src/Proto.Actor/Utils/EmptyKeyValueStore.cs

@@ -14,7 +14,7 @@ namespace Proto.Utils;
 /// <typeparam name="T"></typeparam>
 public class EmptyKeyValueStore<T> : IKeyValueStore<T>
 {
-    public Task<T> GetAsync(string id, CancellationToken ct) => Task.FromResult(default(T));
+    public Task<T?> GetAsync(string id, CancellationToken ct) => Task.FromResult(default(T));
 
     public Task SetAsync(string id, T state, CancellationToken ct) => Task.CompletedTask;
 

src/Proto.Actor/Utils/IKeyValueStore.cs

@@ -22,7 +22,7 @@ public interface IKeyValueStore<T>
     /// <param name="id">Key</param>
     /// <param name="ct"></param>
     /// <returns></returns>
-    Task<T> GetAsync(string id, CancellationToken ct);
+    Task<T?> GetAsync(string id, CancellationToken ct);
 
     /// <summary>
     /// Set the value for the given key.

src/Proto.Cluster/PubSub/TopicActor.cs

@@ -221,7 +221,7 @@ private async Task<Subscribers> LoadSubscriptions(string topic)
             //TODO: cancellation token config?
             var state = await _subscriptionStore.GetAsync(topic, CancellationToken.None);
             Logger.LogDebug("Topic {Topic} loaded subscriptions {Subscriptions}", _topic, state);
-            return state;
+            return state ?? new Subscribers();
         }
         catch (Exception e)
         {

tests/Proto.Cluster.PubSub.Tests/PubSubClusterFixture.cs

@@ -18,6 +18,7 @@ public record Response;
 
 public class PubSubClusterFixture : BaseInMemoryClusterFixture
 {
+    private readonly bool _useDefaultTopicRegistration;
     public const string SubscriberKind = "Subscriber";
     public const string TimeoutSubscriberKind = "TimeoutSubscriber";
 
@@ -28,26 +29,34 @@ public class PubSubClusterFixture : BaseInMemoryClusterFixture
 
     public ITestOutputHelper? Output;
 
-    public PubSubClusterFixture() : base(3, config =>
+    public PubSubClusterFixture() : this(3, false) {} // xunit requires single, public, parameterless constructor
+    
+    internal PubSubClusterFixture(int clusterSize, bool useDefaultTopicRegistration) : base(clusterSize, config =>
         config
             .WithActorRequestTimeout(TimeSpan.FromSeconds(1))
             .WithPubSubConfig(PubSubConfig.Setup()
                 .WithSubscriberTimeout(TimeSpan.FromSeconds(2))
             )
-    )
-    {
-    }
+    ) => _useDefaultTopicRegistration = useDefaultTopicRegistration;
 
     private readonly InMemorySubscribersStore _subscribersStore = new();
 
     public Task<Subscribers> GetSubscribersForTopic(string topic) => _subscribersStore.GetAsync(topic, CancellationToken.None);
 
-    protected override ClusterKind[] ClusterKinds => new[]
-    {
-        new ClusterKind(TopicActor.Kind, Props.FromProducer(() => new TopicActor(_subscribersStore))),
-        new ClusterKind(SubscriberKind, SubscriberProps()),
-        new ClusterKind(TimeoutSubscriberKind, TimeoutSubscriberProps())
-    };
+    protected override ClusterKind[] ClusterKinds {
+        get {
+            var kinds = new List<ClusterKind>
+            {
+                new(SubscriberKind, SubscriberProps()),
+                new(TimeoutSubscriberKind, TimeoutSubscriberProps())
+            };
+            
+            if(!_useDefaultTopicRegistration)
+                kinds.Add(new ClusterKind(TopicActor.Kind, Props.FromProducer(() => new TopicActor(_subscribersStore))));
+
+            return kinds.ToArray();
+        }
+    }
 
     private Props SubscriberProps()
     {
@@ -136,29 +145,39 @@ public async Task UnsubscribeAllFrom(string topic, string[] subscriberIds)
 
     public async Task SubscribeTo(string topic, string identity, string kind = PubSubClusterFixture.SubscriberKind)
     {
-        var subRes = await RandomMember().Subscribe(topic, identity, kind);
+        var subRes = await RandomMember().Subscribe(topic, identity, kind, CancellationTokens.FromSeconds(5));
         if (subRes == null)
             Output?.WriteLine($"{kind}/{identity} failed to subscribe due to timeout");
+        subRes.Should().NotBeNull("subscribing should not time out");
     }
 
     public async Task UnsubscribeFrom(string topic, string identity, string kind = PubSubClusterFixture.SubscriberKind)
     {
-        var unsubRes = await RandomMember().Unsubscribe(topic, identity, kind);
+        var unsubRes = await RandomMember().Unsubscribe(topic, identity, kind, CancellationTokens.FromSeconds(5));
         if (unsubRes == null)
             Output?.WriteLine($"{kind}/{identity} failed to subscribe due to timeout");
+        unsubRes.Should().NotBeNull("unsubscribing should not time out");
     }
 
     public Task<PublishResponse?> PublishData(string topic, int data, CancellationToken cancel = default)
-        => RandomMember()
+    {
+        if(cancel == default) cancel = CancellationTokens.FromSeconds(5);
+        
+        return RandomMember()
             .Publisher()
             .Publish(topic, new DataPublished(data), cancel);
+    }
 
     public Task<PublishResponse?> PublishDataBatch(string topic, int[] data, CancellationToken cancel = default)
-        => RandomMember()
+    {
+        if (cancel == default) cancel = CancellationTokens.FromSeconds(5);
+
+        return RandomMember()
             .Publisher()
             .PublishBatch(
                 topic,
                 data.Select(d => new DataPublished(d)).ToArray(),
                 cancel
             );
+    }
 }

tests/Proto.Cluster.PubSub.Tests/PubSubDefaultTopicRegistrationTests.cs

@@ -0,0 +1,40 @@
+// -----------------------------------------------------------------------
+// <copyright file = "PubSubDefaultTopicRegistrationTests.cs" company = "Asynkron AB">
+//      Copyright (C) 2015-2022 Asynkron AB All rights reserved
+// </copyright>
+// -----------------------------------------------------------------------
+using FluentAssertions;
+using Xunit;
+
+namespace Proto.Cluster.PubSub.Tests;
+
+[Collection("PubSub")] // The CI is just to slow to run cluster fixture based tests in parallel
+public class PubSubDefaultTopicRegistrationTests : IAsyncLifetime
+{
+    private readonly PubSubClusterFixture _fixture;
+
+    public PubSubDefaultTopicRegistrationTests() => _fixture = new PubSubClusterFixture(clusterSize: 1, useDefaultTopicRegistration: true);
+
+    public Task InitializeAsync() => _fixture.InitializeAsync();
+    
+    [Fact]
+    public async Task Pub_sub_works_with_default_topic_registration()
+    {
+        var subscriberIds = _fixture.SubscriberIds("topic-default", 20);
+        const string topic = "topic-default-registration";
+        const int numMessages = 100;
+
+        await _fixture.SubscribeAllTo(topic, subscriberIds);
+
+        for (var i = 0; i < numMessages; i++)
+        {
+            var response = await _fixture.PublishData(topic, i);
+            response.Should().NotBeNull("publishing should not time out");
+            response!.Status.Should().Be(PublishStatus.Ok);
+        }
+
+        await _fixture.VerifyAllSubscribersGotAllTheData(subscriberIds, numMessages);
+    }
+
+    public Task DisposeAsync() => _fixture.DisposeAsync();
+}