intermediate, investigating weird connection oddity

Author: mgravell

StackExchange.Redis.sln

@@ -117,8 +117,6 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "ConsoleTest", "tests\Consol
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "ConsoleTestBaseline", "tests\ConsoleTestBaseline\ConsoleTestBaseline.csproj", "{69A0ACF2-DF1F-4F49-B554-F732DCA938A3}"
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "SimpleCancellationDemo", "tests\SimpleCancellationDemo\SimpleCancellationDemo.csproj", "{368A06AA-9DEB-4C55-B9CF-A1070B85E502}"
-EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "docs", "docs\docs.csproj", "{64CF03B6-6B29-4C4C-88B8-7B9E317D631A}"
 EndProject
 Global
@@ -171,10 +169,6 @@ Global
 		{69A0ACF2-DF1F-4F49-B554-F732DCA938A3}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{69A0ACF2-DF1F-4F49-B554-F732DCA938A3}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{69A0ACF2-DF1F-4F49-B554-F732DCA938A3}.Release|Any CPU.Build.0 = Release|Any CPU
-		{368A06AA-9DEB-4C55-B9CF-A1070B85E502}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
-		{368A06AA-9DEB-4C55-B9CF-A1070B85E502}.Debug|Any CPU.Build.0 = Debug|Any CPU
-		{368A06AA-9DEB-4C55-B9CF-A1070B85E502}.Release|Any CPU.ActiveCfg = Release|Any CPU
-		{368A06AA-9DEB-4C55-B9CF-A1070B85E502}.Release|Any CPU.Build.0 = Release|Any CPU
 		{64CF03B6-6B29-4C4C-88B8-7B9E317D631A}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
 		{64CF03B6-6B29-4C4C-88B8-7B9E317D631A}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{64CF03B6-6B29-4C4C-88B8-7B9E317D631A}.Release|Any CPU.ActiveCfg = Release|Any CPU
@@ -200,7 +194,6 @@ Global
 		{A9F81DA3-DA82-423E-A5DD-B11C37548E06} = {96E891CD-2ED7-4293-A7AB-4C6F5D8D2B05}
 		{A0F89B8B-32A3-4C28-8F1B-ADE343F16137} = {73A5C363-CA1F-44C4-9A9B-EF791A76BA6A}
 		{69A0ACF2-DF1F-4F49-B554-F732DCA938A3} = {73A5C363-CA1F-44C4-9A9B-EF791A76BA6A}
-		{368A06AA-9DEB-4C55-B9CF-A1070B85E502} = {73A5C363-CA1F-44C4-9A9B-EF791A76BA6A}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {193AA352-6748-47C1-A5FC-C9AA6B5F000B}

src/StackExchange.Redis/ChannelMessageQueue.cs

@@ -184,7 +184,7 @@ private async Task OnMessageSyncImpl()
                 catch (ChannelClosedException) { break; } // expected
                 catch (Exception ex)
                 {
-                    _parent?.multiplexer?.OnInternalError(ex);
+                    _parent?.Multiplexer?.OnInternalError(ex);
                     break;
                 }
 
@@ -305,7 +305,7 @@ private async Task OnMessageAsyncImpl()
                 catch (ChannelClosedException) { break; } // expected
                 catch (Exception ex)
                 {
-                    _parent?.multiplexer?.OnInternalError(ex);
+                    _parent?.Multiplexer?.OnInternalError(ex);
                     break;
                 }
 

src/StackExchange.Redis/CursorEnumerable.cs

@@ -246,7 +246,7 @@ private void ThrowTimeout(Message message)
             {
                 try
                 {
-                    throw ExceptionFactory.Timeout(parent.redis.multiplexer, null, message, parent.server);
+                    throw ExceptionFactory.Timeout(parent.redis.Multiplexer, null, message, parent.server);
                 }
                 catch (Exception ex)
                 {

src/StackExchange.Redis/Exceptions.cs

@@ -1,5 +1,6 @@
 using System;
 using System.ComponentModel;
+using System.Diagnostics;
 using System.Runtime.Serialization;
 
 namespace StackExchange.Redis
@@ -106,6 +107,7 @@ public RedisConnectionException(ConnectionFailureType failureType, string messag
         /// <param name="commandStatus">The status of the command.</param>
         public RedisConnectionException(ConnectionFailureType failureType, string message, Exception? innerException, CommandStatus commandStatus) : base(message, innerException)
         {
+            Debug.WriteLine($"{failureType}: {message}");
             FailureType = failureType;
             CommandStatus = commandStatus;
         }

src/StackExchange.Redis/PublicAPI/PublicAPI.Unshipped.txt

@@ -1,6 +1,5 @@
 #nullable enable
 StackExchange.Redis.RedisCancellationExtensions
-static StackExchange.Redis.RedisCancellationExtensions.GetEffectiveCancellationToken(this StackExchange.Redis.IConnectionMultiplexer! redis) -> System.Threading.CancellationToken
 static StackExchange.Redis.RedisCancellationExtensions.WithCancellation(this StackExchange.Redis.IConnectionMultiplexer! redis, System.Threading.CancellationToken cancellationToken) -> System.IDisposable!
 static StackExchange.Redis.RedisCancellationExtensions.WithCancellationAndTimeout(this StackExchange.Redis.IConnectionMultiplexer! redis, System.Threading.CancellationToken cancellationToken, int milliseconds) -> System.IDisposable!
 static StackExchange.Redis.RedisCancellationExtensions.WithCancellationAndTimeout(this StackExchange.Redis.IConnectionMultiplexer! redis, System.Threading.CancellationToken cancellationToken, System.TimeSpan timeout) -> System.IDisposable!

src/StackExchange.Redis/RedisBase.cs

@@ -5,21 +5,21 @@
 
 namespace StackExchange.Redis
 {
-    internal abstract partial class RedisBase : IRedis
+    internal abstract class RedisBase : IRedis
     {
         internal static readonly DateTime UnixEpoch = new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc);
-        internal readonly ConnectionMultiplexer multiplexer;
-        protected readonly object? asyncState;
+        internal readonly ConnectionMultiplexer Multiplexer;
+        internal readonly object? AsyncState;
 
         internal RedisBase(ConnectionMultiplexer multiplexer, object? asyncState)
         {
-            this.multiplexer = multiplexer;
-            this.asyncState = asyncState;
+            Multiplexer = multiplexer;
+            AsyncState = asyncState;
         }
 
-        IConnectionMultiplexer IRedisAsync.Multiplexer => multiplexer;
+        IConnectionMultiplexer IRedisAsync.Multiplexer => Multiplexer;
 
-        internal CancellationToken GetEffectiveCancellationToken() => multiplexer.GetEffectiveCancellationToken();
+        internal CancellationToken GetEffectiveCancellationToken() => Multiplexer.GetEffectiveCancellationToken();
 
         public virtual TimeSpan Ping(CommandFlags flags = CommandFlags.None)
         {
@@ -33,48 +33,48 @@ public virtual Task<TimeSpan> PingAsync(CommandFlags flags = CommandFlags.None)
             return ExecuteAsync(msg, ResultProcessor.ResponseTimer);
         }
 
-        public override string ToString() => multiplexer.ToString();
+        public override string ToString() => Multiplexer.ToString();
 
-        public bool TryWait(Task task) => task.Wait(multiplexer.TimeoutMilliseconds);
+        public bool TryWait(Task task) => task.Wait(Multiplexer.TimeoutMilliseconds);
 
-        public void Wait(Task task) => multiplexer.Wait(task);
+        public void Wait(Task task) => Multiplexer.Wait(task);
 
-        public T Wait<T>(Task<T> task) => multiplexer.Wait(task);
+        public T Wait<T>(Task<T> task) => Multiplexer.Wait(task);
 
-        public void WaitAll(params Task[] tasks) => multiplexer.WaitAll(tasks);
+        public void WaitAll(params Task[] tasks) => Multiplexer.WaitAll(tasks);
 
         internal virtual Task<T> ExecuteAsync<T>(Message? message, ResultProcessor<T>? processor, T defaultValue, ServerEndPoint? server = null)
         {
-            if (message is null) return CompletedTask<T>.FromDefault(defaultValue, asyncState);
-            multiplexer.CheckMessage(message);
+            if (message is null) return CompletedTask<T>.FromDefault(defaultValue, AsyncState);
+            Multiplexer.CheckMessage(message);
 
             // The message already captures the ambient cancellation token when it was created,
             // so we don't need to pass it again. This ensures resent messages preserve their original cancellation context.
-            return multiplexer.ExecuteAsyncImpl<T>(message, processor, asyncState, server, defaultValue);
+            return Multiplexer.ExecuteAsyncImpl<T>(message, processor, AsyncState, server, defaultValue);
         }
 
         internal virtual Task<T?> ExecuteAsync<T>(Message? message, ResultProcessor<T>? processor, ServerEndPoint? server = null)
         {
-            if (message is null) return CompletedTask<T>.Default(asyncState);
-            multiplexer.CheckMessage(message);
+            if (message is null) return CompletedTask<T>.Default(AsyncState);
+            Multiplexer.CheckMessage(message);
 
             // The message already captures the ambient cancellation token when it was created,
             // so we don't need to pass it again. This ensures resent messages preserve their original cancellation context.
-            return multiplexer.ExecuteAsyncImpl<T>(message, processor, asyncState, server);
+            return Multiplexer.ExecuteAsyncImpl<T>(message, processor, AsyncState, server);
         }
 
         [return: NotNullIfNotNull("defaultValue")]
         internal virtual T? ExecuteSync<T>(Message? message, ResultProcessor<T>? processor, ServerEndPoint? server = null, T? defaultValue = default)
         {
             if (message is null) return defaultValue; // no-op
-            multiplexer.CheckMessage(message);
-            return multiplexer.ExecuteSyncImpl<T>(message, processor, server, defaultValue);
+            Multiplexer.CheckMessage(message);
+            return Multiplexer.ExecuteSyncImpl<T>(message, processor, server, defaultValue);
         }
 
         internal virtual RedisFeatures GetFeatures(in RedisKey key, CommandFlags flags, RedisCommand command, out ServerEndPoint? server)
         {
-            server = multiplexer.SelectServer(command, flags, key);
-            var version = server == null ? multiplexer.RawConfig.DefaultVersion : server.Version;
+            server = Multiplexer.SelectServer(command, flags, key);
+            var version = server == null ? Multiplexer.RawConfig.DefaultVersion : server.Version;
             return new RedisFeatures(version);
         }
 
@@ -118,7 +118,7 @@ protected static void WhenAlwaysOrNotExists(When when)
         private ResultProcessor.TimingProcessor.TimerMessage GetTimerMessage(CommandFlags flags)
         {
             // do the best we can with available commands
-            var map = multiplexer.CommandMap;
+            var map = Multiplexer.CommandMap;
             var cancellationToken = GetEffectiveCancellationToken();
             if (map.IsAvailable(RedisCommand.PING))
                 return ResultProcessor.TimingProcessor.CreateMessage(-1, flags, RedisCommand.PING, default, cancellationToken);
@@ -128,7 +128,7 @@ private ResultProcessor.TimingProcessor.TimerMessage GetTimerMessage(CommandFlag
                 return ResultProcessor.TimingProcessor.CreateMessage(-1, flags, RedisCommand.ECHO, RedisLiterals.PING, cancellationToken);
             // as our fallback, we'll do something odd... we'll treat a key like a value, out of sheer desperation
             // note: this usually means: twemproxy/envoyproxy - in which case we're fine anyway, since the proxy does the routing
-            return ResultProcessor.TimingProcessor.CreateMessage(0, flags, RedisCommand.EXISTS, (RedisValue)multiplexer.UniqueId, cancellationToken);
+            return ResultProcessor.TimingProcessor.CreateMessage(0, flags, RedisCommand.EXISTS, (RedisValue)Multiplexer.UniqueId, cancellationToken);
         }
 
         internal static class CursorUtils

src/StackExchange.Redis/RedisBatch.cs

@@ -8,7 +8,7 @@ internal class RedisBatch : RedisDatabase, IBatch
     {
         private List<Message>? pending;
 
-        public RedisBatch(RedisDatabase wrapped, object? asyncState) : base(wrapped.multiplexer, wrapped.Database, asyncState ?? wrapped.AsyncState) { }
+        public RedisBatch(RedisDatabase wrapped, object? asyncState) : base(wrapped.Multiplexer, wrapped.Database, asyncState ?? wrapped.AsyncState) { }
 
         public void Execute()
         {
@@ -24,17 +24,17 @@ public void Execute()
             List<Message>? lastList = null;
             foreach (var message in snapshot)
             {
-                var server = multiplexer.SelectServer(message);
+                var server = Multiplexer.SelectServer(message);
                 if (server == null)
                 {
-                    FailNoServer(multiplexer, snapshot);
-                    throw ExceptionFactory.NoConnectionAvailable(multiplexer, message, server);
+                    FailNoServer(Multiplexer, snapshot);
+                    throw ExceptionFactory.NoConnectionAvailable(Multiplexer, message, server);
                 }
                 var bridge = server.GetBridge(message);
                 if (bridge == null)
                 {
-                    FailNoServer(multiplexer, snapshot);
-                    throw ExceptionFactory.NoConnectionAvailable(multiplexer, message, server);
+                    FailNoServer(Multiplexer, snapshot);
+                    throw ExceptionFactory.NoConnectionAvailable(Multiplexer, message, server);
                 }
 
                 // identity a list
@@ -58,15 +58,15 @@ public void Execute()
             {
                 if (!pair.Key.TryEnqueue(pair.Value, pair.Key.ServerEndPoint.IsReplica))
                 {
-                    FailNoServer(multiplexer, pair.Value);
+                    FailNoServer(Multiplexer, pair.Value);
                 }
             }
         }
 
         internal override Task<T> ExecuteAsync<T>(Message? message, ResultProcessor<T>? processor, T defaultValue, ServerEndPoint? server = null)
         {
-            if (message == null) return CompletedTask<T>.FromDefault(defaultValue, asyncState);
-            multiplexer.CheckMessage(message);
+            if (message == null) return CompletedTask<T>.FromDefault(defaultValue, ((RedisBase)this).AsyncState);
+            Multiplexer.CheckMessage(message);
 
             // prepare the inner command as a task
             Task<T> task;
@@ -76,7 +76,7 @@ internal override Task<T> ExecuteAsync<T>(Message? message, ResultProcessor<T>?
             }
             else
             {
-                var source = TaskResultBox<T>.Create(GetEffectiveCancellationToken(), out var tcs, asyncState);
+                var source = TaskResultBox<T>.Create(GetEffectiveCancellationToken(), out var tcs, ((RedisBase)this).AsyncState);
                 task = tcs.Task;
                 message.SetSource(source, processor);
             }
@@ -88,8 +88,8 @@ internal override Task<T> ExecuteAsync<T>(Message? message, ResultProcessor<T>?
 
         internal override Task<T?> ExecuteAsync<T>(Message? message, ResultProcessor<T>? processor, ServerEndPoint? server = null) where T : default
         {
-            if (message == null) return CompletedTask<T>.Default(asyncState);
-            multiplexer.CheckMessage(message);
+            if (message == null) return CompletedTask<T>.Default(((RedisBase)this).AsyncState);
+            Multiplexer.CheckMessage(message);
 
             // prepare the inner command as a task
             Task<T?> task;
@@ -99,7 +99,7 @@ internal override Task<T> ExecuteAsync<T>(Message? message, ResultProcessor<T>?
             }
             else
             {
-                var source = TaskResultBox<T?>.Create(message.CancellationToken, out var tcs, asyncState);
+                var source = TaskResultBox<T?>.Create(message.CancellationToken, out var tcs, ((RedisBase)this).AsyncState);
                 task = tcs.Task;
                 message.SetSource(source!, processor);
             }

src/StackExchange.Redis/RedisCancellationExtensions.cs

@@ -112,7 +112,7 @@ public static IDisposable WithCancellationAndTimeout(
         /// combining any ambient cancellation token and timeout.
         /// </summary>
         /// <returns>The effective cancellation token, or CancellationToken.None if no ambient context is set.</returns>
-        public static CancellationToken GetEffectiveCancellationToken(this IConnectionMultiplexer redis)
+        internal static CancellationToken GetEffectiveCancellationToken(this IConnectionMultiplexer redis, bool checkForCancellation = true)
         {
             var scope = _context.Value;
 
@@ -126,7 +126,11 @@ public static CancellationToken GetEffectiveCancellationToken(this IConnectionMu
                     if (fromScope is not null && fromScope.Equals(redis))
                     {
                         var token = scope.Token;
-                        token.ThrowIfCancellationRequested();
+                        if (checkForCancellation)
+                        {
+                            token.ThrowIfCancellationRequested();
+                        }
+
                         return token;
                     }
                     scope = scope.Previous;
@@ -137,10 +141,30 @@ public static CancellationToken GetEffectiveCancellationToken(this IConnectionMu
         }
 
         /// <summary>
-        /// Gets the current cancellation context for diagnostic purposes.
+        /// Gets the current cancellation scope for diagnostic purposes.
         /// </summary>
-        /// <returns>The current context, or null if no ambient context is set.</returns>
-        internal static object? GetCurrentScope() => _context.Value;
+        /// <returns>The current scope, or null if no ambient context is set.</returns>
+        internal static object? GetCurrentScope(this IConnectionMultiplexer redis)
+        {
+            var scope = _context.Value;
+
+            // ReSharper disable once ConditionIsAlwaysTrueOrFalseAccordingToNullableAPIContract - trust no-one
+            if (redis is not null)
+            {
+                // check for the top scope *that relates to this redis instance*
+                while (scope is not null)
+                {
+                    var scopeMuxer = scope.Target; // need to null-check because weak-ref / GC
+                    if (scopeMuxer is not null && scopeMuxer.Equals(redis))
+                    {
+                        return scope;
+                    }
+                    scope = scope.Previous;
+                }
+            }
+
+            return null;
+        }
 
         private static readonly AsyncLocal<CancellationScope?> _context = new();