NCBC-3393: OperationTaskException may continue after rebalance completes

Motivation
----------
Fixes a bug where the SDK incorrectly assumes it has the latest config
if there is a failure processing the cluster map. Note that if another
config was received and successfully processed, things would progress
successfully.

Modifications
-------------
 - Ensure OperationTaskException's are turned into UnambiguousTimeoutExceptions
if they never make it onto the wire. This can happen if a circuit breaker is
tripped.
 - Ensure that exceptions that happen in the ClusterContext are thrown and
handled by the calling code within the CouchbaseBucket
 - Fix unit tests

Change-Id: I245411d924e46c94f4e17c2c79a662354e891bd5
Reviewed-on: https://review.couchbase.org/c/couchbase-net-client/+/191786
Reviewed-by: Richard Ponton <richard.ponton@couchbase.com>
Tested-by: Build Bot <build@couchbase.com>

src/Couchbase/Core/ClusterContext.cs

@@ -813,6 +813,7 @@ public async Task ProcessClusterMapAsync(BucketBase bucket, BucketConfig config)
             catch (Exception e)
             {
                 _logger.LogError(LoggingEvents.ConfigEvent, "ERROR:{e}", e);
+                throw;
             }
             finally
             {

src/Couchbase/Core/IO/Operations/OperationBase.cs

@@ -39,7 +39,6 @@ internal abstract class OperationBase : IOperation, IValueTaskSource<ResponseSta
         private bool _isOrphaned;
         private volatile string? _lastDispatchedFrom;
         private volatile string? _lastDispatchedTo;
-        private long _totalExpiredTime;
 
         protected OperationBase()
         {
@@ -203,7 +202,11 @@ public CancellationToken Token
         /// </summary>
         public ObjectPool<OperationBuilder> OperationBuilderPool { get; set; } = null!;  // Assumes we always initialize with OperationConfigurator
 
-        public TimeSpan Elapsed => TimeSpan.FromMilliseconds(_totalExpiredTime);
+        public TimeSpan Elapsed
+        {
+            get;
+            private set;
+        } = TimeSpan.Zero;
 
         #endregion
 
@@ -688,9 +691,6 @@ public void HandleOperationCompleted(in SlicedMemoryOwner<byte> data)
             {
                 //for measuring latency using an LoggingMeter or similar.
                 StopRecording();
-
-                //Since an operation may be retried, we want to add to the total elapsed time.
-                _totalExpiredTime = Interlocked.Add(ref _totalExpiredTime, _stopwatch.ElapsedMilliseconds);
             }
         }
 
@@ -707,6 +707,7 @@ public bool WasNmvb()
         public string? LastDispatchedTo => _lastDispatchedTo;
 
         public string? LastErrorMessage { get; set; }
+
         public virtual bool CanStream => false;
 
         public bool IsCompleted => _isCompleted == 1;
@@ -719,6 +720,9 @@ public bool WasNmvb()
         public void StopRecording()
         {
             _stopwatch.Stop();
+
+            //Since an operation may be retried, we want to add to the total elapsed time.
+            Elapsed =  Elapsed.Add(_stopwatch.Elapsed);
             MetricTracker.KeyValue.TrackOperation(OpCode, _stopwatch.Elapsed);
         }
         #endregion

src/Couchbase/Core/Retry/IRequest.cs

@@ -14,6 +14,11 @@ public interface IRequest
         List<RetryReason> RetryReasons { get; set; }
         IRetryStrategy RetryStrategy { get; set; }
         TimeSpan Timeout { get; set; }
+
+        /// <summary>
+        /// The total time expired at the time the operation is called. If another retry happens,
+        /// it will be updated once the response is received.
+        /// </summary>
         TimeSpan Elapsed { get; }
         CancellationToken Token
         {

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

@@ -1,4 +1,5 @@
 using System.Collections.Generic;
+using System.Threading;
 using System.Threading.Tasks;
 using Couchbase.Core;
 using Couchbase.Core.Bootstrapping;
@@ -512,11 +513,24 @@ public async Task Test_HasConfigChanges()
 
     CouchbaseBucket CreateBucket(BucketConfig bootstrapConfig)
     {
+        var clusterNodeFactory = new Mock<IClusterNodeFactory>();
+        var node = new Mock<IClusterNode>();
+        node.Setup(x => x.KeyEndPoints).Returns(new List<HostEndpointWithPort>
+            { new("127.0.0.1", 11210) });
+
+        clusterNodeFactory.Setup(x => x.CreateAndConnectAsync(
+                It.IsAny<HostEndpointWithPort>(), It.IsAny<NodeAdapter>(),
+                It.IsAny<CancellationToken>())).
+            Returns(Task.FromResult(node.Object));
+
+        var options = new ClusterOptions().AddClusterService(clusterNodeFactory.Object);
+        var clusterCtx = new ClusterContext(new CancellationTokenSource(), options)
+        {
+            SupportsCollections = true
+        };
+
         var bucket = new CouchbaseBucket("default",
-            new ClusterContext
-            {
-                SupportsCollections = true
-            },
+            clusterCtx,
             new Mock<IScopeFactory>().Object,
             new Mock<IRetryOrchestrator>().Object,
             new Mock<IVBucketKeyMapperFactory>().Object,
@@ -528,6 +542,9 @@ CouchbaseBucket CreateBucket(BucketConfig bootstrapConfig)
             new BestEffortRetryStrategy(),
             bootstrapConfig);
 
+        node.Setup(x => x.Owner).Returns(bucket);
+
+
         return bucket;
     }
 }

tests/Couchbase.UnitTests/Core/Diagnostics/Tracing/Fakes/FakeOperation.cs

@@ -40,6 +40,7 @@ public CancellationToken Token
         public CancellationTokenPair TokenPair { get; set; }
         public string? ClientContextId { get; set; }
         public string? Statement { get; set; }
+
         public bool PreserveTtl { get; }
         public OpCode OpCode { get; }
         public string? BucketName { get; }

tests/Couchbase.UnitTests/Utils/FakeOperation.cs

@@ -64,6 +64,7 @@ public CancellationToken Token
         public CancellationTokenPair TokenPair { get; set; }
         public string ClientContextId { get; set; }
         public string Statement { get; set; }
+
         public bool PreserveTtl { get; }
         public OpCode OpCode { get; }
         public string Key { get; }