NCBC-3807: ConfigPushHandler unit tests / Handle server version regressed edge case.

Motivation:
Test coverage and edge case discovery for ConfigPushHandler

Modifications:
* ConfigHandler doesn't blow up when attempting to log configs with Trace enabled.
* ConfigPushHandler mitigates looping if server returns version older than what was pushed.
* Basic unit tests for ConfigPushHandler (more to come)

Change-Id: If82713f43f62353e9f99816bfe8f240eeb928081
Reviewed-on: https://review.couchbase.org/c/couchbase-net-client/+/211964
Tested-by: Build Bot <[email protected]>
Reviewed-by: Jeffry Morris <[email protected]>

Author: RiPont

src/Couchbase/Core/Configuration/Server/ConfigHandler.cs

@@ -232,8 +232,15 @@ public void Publish(BucketConfig config)
             if (_logger.IsEnabled(LogLevel.Trace))
             {
                 // Only log if debug logging is enabled to avoid serialization cost
-                _logger.LogDebug(LoggingEvents.ConfigEvent,
-                    JsonSerializer.Serialize(config, InternalSerializationContext.Default.BucketConfig));
+                try
+                {
+                    _logger.LogDebug(LoggingEvents.ConfigEvent,
+                        JsonSerializer.Serialize(config, InternalSerializationContext.Default.BucketConfig));
+                }
+                catch (Exception ex)
+                {
+                    _logger.LogWarning(ex, "Failed to deserialize config");
+                }
             }
 
             if (_configQueue?.Completion.IsCompleted ?? true)

src/Couchbase/Core/Configuration/Server/ConfigPushHandler.cs

@@ -56,6 +56,7 @@ public ConfigPushHandler(BucketBase bucket, ClusterContext context, ILogger logg
     private async Task ProcessConfigPushesAsync(CancellationToken cancellationToken)
     {
         ConfigVersion lastSkippedVersion = new(0, 0);
+        long skipsWithNoPublish = 0;
         while (!cancellationToken.IsCancellationRequested)
         {
             // Don't release this semaphore, it is released whenever a new config is received.
@@ -143,6 +144,7 @@ private async Task ProcessConfigPushesAsync(CancellationToken cancellationToken)
                         // therefore, not-null means this is a new config that must be published to all subscribers.
                         LogConfigPublished(_redactor.SystemData(_bucket.Name), fetchedBucketConfig.ConfigVersion,
                             effectiveVersion);
+                        Interlocked.Exchange(ref skipsWithNoPublish, 0);
                         _context.PublishConfig(fetchedBucketConfig);
                     }
                     else if (fetchedVersion < pushedVersion)
@@ -163,6 +165,12 @@ private async Task ProcessConfigPushesAsync(CancellationToken cancellationToken)
                     {
                         // a new version came in while we were publishing.
                         LogAttemptedButSkipped(nextVersion, fetchedVersion);
+                        var skips = Interlocked.Increment(ref skipsWithNoPublish);
+                        if (skips > 100)
+                        {
+                            await Task.Delay(10).ConfigureAwait(false);
+                        }
+
                         TryReleaseNewVersionSemaphore();
                     }
                 }

tests/Couchbase.UnitTests/Core/Configuration/Server/ConfigPushHandlerTests.cs

@@ -0,0 +1,160 @@
+#nullable enable
+using System;
+using System.Collections.Generic;
+using System.Runtime.CompilerServices;
+using System.Threading;
+using System.Threading.Tasks;
+using Couchbase.Core;
+using Couchbase.Core.Bootstrapping;
+using Couchbase.Core.Configuration.Server;
+using Couchbase.Core.DI;
+using Couchbase.Core.Diagnostics.Tracing;
+using Couchbase.Core.IO.Operations;
+using Couchbase.Core.Logging;
+using Couchbase.Core.Retry;
+using Couchbase.Test.Common.Utils;
+using Couchbase.UnitTests.Utils;
+using Microsoft.Extensions.Logging;
+using Moq;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace Couchbase.UnitTests.Core.Configuration.Server;
+
+public class ConfigPushHandlerTests
+{
+    private readonly ITestOutputHelper _outputHelper;
+
+    public ConfigPushHandlerTests(ITestOutputHelper outputHelper)
+    {
+        _outputHelper = outputHelper;
+    }
+    [Fact]
+    public async Task ConfigPushHandler_ServerVersionRegressed()
+    {
+        // server pushes (1,3), but returns (1,1)
+        bool publishedAnything = false;
+        var initialBucketConfig = new BucketConfig() { RevEpoch = 1, Rev = 2 };
+        var versionPublished = new ConfigVersion(0, 0);
+        var mockBucket = CreateBucketMock(initialConfig: initialBucketConfig, onPublish: bc =>
+        {
+            publishedAnything = true;
+            Assert.NotNull(bc);
+            versionPublished = bc!.ConfigVersion;
+        });
+        ClusterContext mockContext = mockBucket.Context;
+        var mockNode = new Mock<IClusterNode>();
+        BucketConfig getClusterMapResult = new BucketConfig() { RevEpoch = 1, Rev = 1 };
+        getClusterMapResult.OnDeserialized();
+        IReadOnlyCollection<HostEndpointWithPort> endpoints = new List<HostEndpointWithPort>();
+        mockNode.Setup(x => x.GetClusterMap(It.IsAny<ConfigVersion?>(), It.IsAny<CancellationToken>()))
+            .Returns(Task.FromResult(getClusterMapResult));
+        mockNode.SetupGet(x => x.KeyEndPoints).Returns(endpoints);
+        mockContext.Nodes.Add(mockNode.Object);
+        mockBucket.Nodes.Add(mockNode.Object);
+        mockContext.RegisterBucket(mockBucket);
+        mockContext.Start();
+        var logger = new TestOutputLogger(_outputHelper, nameof(ConfigPushHandler_ServerVersionRegressed));
+        var redactor = new TypedRedactor(RedactionLevel.None);
+        using var configPushHandler = new ConfigPushHandler(mockBucket, mockContext, logger, redactor);
+        var pushedVersion = new ConfigVersion(1, 3);
+        configPushHandler.ProcessConfigPush(pushedVersion);
+        // while server is returning older version, do not publish
+        await Task.Delay(100);
+        Assert.NotEqual(versionPublished, pushedVersion);
+
+        // update the version of the config that is returned.  This should result in a publish.
+        getClusterMapResult.Rev = 3;
+        getClusterMapResult.OnDeserialized();
+        await Task.Delay(100);
+        Assert.Equal(versionPublished, pushedVersion);
+    }
+
+    [Fact]
+    public async Task ConfigPushHandler_BasicAdvance()
+    {
+        // server pushes (1,2), and returns (1,2)
+        bool publishedAnything = false;
+        var initialBucketConfig = new BucketConfig() { RevEpoch = 1, Rev = 1 };
+        var versionPublished = new ConfigVersion(0, 0);
+        var mockBucket = CreateBucketMock(initialConfig: initialBucketConfig, onPublish: bc =>
+        {
+            publishedAnything = true;
+            Assert.NotNull(bc);
+            versionPublished = bc!.ConfigVersion;
+        });
+
+        ClusterContext mockContext = mockBucket.Context;
+        var mockNode = new Mock<IClusterNode>();
+        BucketConfig getClusterMapResult = new BucketConfig() { RevEpoch = 1, Rev = 2 };
+        getClusterMapResult.OnDeserialized();
+        IReadOnlyCollection<HostEndpointWithPort> endpoints = new List<HostEndpointWithPort>();
+        mockNode.Setup(x => x.GetClusterMap(It.IsAny<ConfigVersion?>(), It.IsAny<CancellationToken>()))
+            .Returns(Task.FromResult(getClusterMapResult));
+        mockNode.SetupGet(x => x.KeyEndPoints).Returns(endpoints);
+        mockContext.Nodes.Add(mockNode.Object);
+        mockBucket.Nodes.Add(mockNode.Object);
+        mockContext.RegisterBucket(mockBucket);
+        mockContext.Start();
+        var logger = new TestOutputLogger(_outputHelper, nameof(ConfigPushHandler_ServerVersionRegressed));
+        var redactor = new TypedRedactor(RedactionLevel.None);
+        using var configPushHandler = new ConfigPushHandler(mockBucket, mockContext, logger, redactor);
+        configPushHandler.ProcessConfigPush(new ConfigVersion(1, 2));
+        await Task.Delay(100);
+        Assert.True(publishedAnything);
+        Assert.Equal(expected: getClusterMapResult.ConfigVersion, actual: versionPublished);
+    }
+
+    private BucketBase CreateBucketMock(
+        string bucketName = "default",
+        BucketConfig? initialConfig = null,
+        Action<BucketConfig?>? onPublish = null,
+        [CallerMemberName] string caller = "CreateBucketMock")
+    {
+        onPublish ??= _ => { };
+        initialConfig ??= new();
+        Action doNothing = () => { };
+        var mockCluster = new Mock<ICluster>(MockBehavior.Strict);
+        var mockConfigHandler = new Mock<IConfigHandler>(MockBehavior.Strict);
+        mockConfigHandler.Setup(ch => ch.Publish(It.IsAny<BucketConfig>())).Callback(onPublish);
+        mockConfigHandler.Setup(ch => ch.Subscribe(It.IsAny<IConfigUpdateEventSink>())).Callback(doNothing);
+        mockConfigHandler.Setup(ch => ch.Start(It.IsAny<bool>())).Callback(doNothing);
+        var clusterOptions = new ClusterOptions().WithLogging(new TestOutputLoggerFactory(_outputHelper));
+        clusterOptions.AddClusterService(mockConfigHandler.Object);
+        var mock = new Mock<BucketBase>(
+            bucketName,
+            new ClusterContext(mockCluster.Object, new CancellationTokenSource(), clusterOptions),
+            new Mock<IScopeFactory>().Object,
+            new Mock<IRetryOrchestrator>().Object,
+            new TestOutputLogger(_outputHelper, nameof(ConfigPushHandlerTests)),
+            new TypedRedactor(RedactionLevel.None),
+            new Mock<IBootstrapperFactory>().Object,
+            NoopRequestTracer.Instance,
+            new Mock<IOperationConfigurator>().Object,
+            new BestEffortRetryStrategy(),
+            initialConfig);
+
+        mock.SetupGet(it => it.Name).Returns(bucketName);
+        mock.Setup(it => it.ConfigUpdatedAsync(It.IsAny<BucketConfig>()))
+            .Returns((BucketConfig bc) =>
+            {
+                _outputHelper.WriteLine("Config Published: bucket={}, version={}", bucketName, bc.ConfigVersion);
+                return Task.CompletedTask;
+            });
+        return mock.Object;
+    }
+
+    internal class MockConfigUpdatedSink : IConfigUpdateEventSink
+    {
+        public Action<BucketConfig> ConfigUpdatedAction { get; set; } = _ => { };
+
+        public Task ConfigUpdatedAsync(BucketConfig newConfig)
+        {
+            ConfigUpdatedAction(newConfig);
+            return Task.CompletedTask;
+        }
+
+        public string Name => nameof(MockConfigUpdatedSink);
+        public IEnumerable<IClusterNode> ClusterNodes { get; set; }
+    }
+}