Tolerate refused CLUSTER SLOTS/NODES during topology updates

src/StackExchange.Redis/ConnectionMultiplexer.cs

@@ -2157,13 +2157,41 @@ public EndPoint[] GetEndPoints(bool configuredOnly = false) =>
                 // node including those serving nothing. Asked as a pair for symmetry - trusting the topology
                 // cached from autoconfigure here would mean acting on possibly-stale data while deliberately
                 // re-reading the other half
+                // ObserveErrors: if either await below throws, the *other* task is abandoned mid-flight,
+                // and both ride the same connection - a faulted orphan must never surface as an
+                // UnobservedTaskException in the host
                 var slotsTask = ExecuteAsyncImpl(
-                    RedisServer.GetClusterSlotsMessage(CommandFlags.None), ResultProcessor.ClusterSlots, null, server);
+                    RedisServer.GetClusterSlotsMessage(CommandFlags.None), ResultProcessor.ClusterSlots, null, server).ObserveErrors();
                 var nodesTask = ExecuteAsyncImpl(
-                    RedisServer.GetClusterNodesMessage(CommandFlags.None), ResultProcessor.ClusterNodes, null, server);
+                    RedisServer.GetClusterNodesMessage(CommandFlags.None), ResultProcessor.ClusterNodes, null, server).ObserveErrors();
 
-                var slots = await slotsTask.ForAwait();
-                var clusterConfig = await nodesTask.ForAwait();
+                ClusterSlotsResult? slots;
+                try
+                {
+                    slots = await slotsTask.ForAwait();
+                }
+                catch (RedisServerException ex)
+                {
+                    // the server *refused* the command - a posture, not a fault: proxied deployments
+                    // commonly answer NODES while blocking SLOTS (Redis Enterprise exposes SLOTS only
+                    // under the OSS cluster API, for example; see #3274). Fall back to the NODES-only
+                    // path, exactly as for a pre-4.0 server that has no SLOTS view to give
+                    log?.LogInformationClusterSlotsRefused(new(server), ex.Message);
+                    slots = null;
+                }
+
+                ClusterConfiguration? clusterConfig;
+                try
+                {
+                    clusterConfig = await nodesTask.ForAwait();
+                }
+                catch (RedisServerException ex)
+                {
+                    // likewise; a server refusing NODES as well exposes no topology at all, however it
+                    // announced itself (a proxy's HELLO can say "cluster" while CLUSTER stays blocked)
+                    log?.LogInformationClusterNodesRefused(new(server), ex.Message);
+                    return null;
+                }
                 if (clusterConfig is null)
                 {
                     return null;

src/StackExchange.Redis/LoggerExtensions.cs

@@ -328,6 +328,18 @@ internal static void LogWithThreadPoolStats(this ILogger? log, string message)
         Message = "Encountered error while updating cluster config: {ErrorMessage}")]
     internal static partial void LogErrorEncounteredErrorWhileUpdatingClusterConfig(this ILogger logger, Exception exception, string errorMessage);
 
+    [LoggerMessage(
+        Level = LogLevel.Information,
+        EventId = 128,
+        Message = "{Server}: CLUSTER SLOTS was refused ({ErrorMessage}); continuing with CLUSTER NODES alone")]
+    internal static partial void LogInformationClusterSlotsRefused(this ILogger logger, ServerEndPointLogValue server, string errorMessage);
+
+    [LoggerMessage(
+        Level = LogLevel.Information,
+        EventId = 129,
+        Message = "{Server}: CLUSTER NODES was refused ({ErrorMessage}); no cluster topology is available")]
+    internal static partial void LogInformationClusterNodesRefused(this ILogger logger, ServerEndPointLogValue server, string errorMessage);
+
     [LoggerMessage(
         Level = LogLevel.Information,
         EventId = 111,

tests/StackExchange.Redis.Tests/ClusterCommandsRefusedUnitTests.cs

@@ -0,0 +1,75 @@
+using System;
+using System.Linq;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using StackExchange.Redis.Server;
+using Xunit;
+
+namespace StackExchange.Redis.Tests;
+
+/// <summary>
+/// Proxied deployments can refuse some or all <c>CLUSTER</c> subcommands while still identifying as a cluster
+/// (Redis Enterprise exposes <c>CLUSTER SLOTS</c> only under the OSS cluster API, and its <c>HELLO</c>/<c>INFO</c>
+/// can say "cluster" while <c>CLUSTER</c> stays blocked entirely). A refusal is a posture, not a fault: the
+/// topology pass must fall back to whatever view it can get, quietly - not log a server error on every
+/// reconfigure. See #3274.
+/// </summary>
+[RunPerProtocol]
+public class ClusterCommandsRefusedUnitTests(ITestOutputHelper log)
+{
+    private sealed class RefusingServer(ITestOutputHelper log, bool refuseNodes) : InProcessTestServer(log)
+    {
+        protected override TypedRedisValue ClusterSlots(RedisClient client, in RedisRequest request)
+            => TypedRedisValue.Error("ERR command is not allowed");
+
+        protected override TypedRedisValue ClusterNodes(RedisClient client, in RedisRequest request)
+            => refuseNodes ? TypedRedisValue.Error("ERR command is not allowed") : base.ClusterNodes(client, request);
+    }
+
+    [Fact]
+    public async Task SlotsRefusedFallsBackToNodes()
+    {
+        using var server = new RefusingServer(log, refuseNodes: false) { ServerType = ServerType.Cluster };
+        var a = server.DefaultEndPoint;
+        var b = server.AddEmptyNode();
+
+        var config = server.GetClientConfig(defaultOnly: true); // only connect to a initially
+        var logger = new CapturingLogger();
+        config.LoggerFactory = logger;
+        await using var conn = await ConnectionMultiplexer.ConnectAsync(config);
+
+        // discovery still happened, from the NODES view alone
+        var endpoints = conn.GetEndPoints();
+        Assert.Contains(a, endpoints);
+        Assert.Contains(b, endpoints);
+
+        log.WriteLine(logger.All);
+        Assert.Empty(logger.Matching("Encountered error while updating cluster config"));
+        Assert.NotEmpty(logger.Matching("CLUSTER SLOTS was refused"));
+
+        await conn.GetDatabase().PingAsync();
+    }
+
+    [Fact]
+    public async Task AllClusterCommandsRefusedIsQuiet()
+    {
+        // the server still *identifies* as a cluster (HELLO "mode", INFO "redis_mode"), so the client
+        // interrogates it; both commands being refused must leave a working connection, not a logged error
+        using var server = new RefusingServer(log, refuseNodes: true) { ServerType = ServerType.Cluster };
+
+        var config = server.GetClientConfig(defaultOnly: true);
+        var logger = new CapturingLogger();
+        config.LoggerFactory = logger;
+        await using var conn = await ConnectionMultiplexer.ConnectAsync(config);
+
+        log.WriteLine(logger.All);
+        Assert.Empty(logger.Matching("Encountered error while updating cluster config"));
+        Assert.NotEmpty(logger.Matching("CLUSTER NODES was refused"));
+
+        await conn.GetDatabase().PingAsync();
+
+        // ...and a later explicit reconfigure stays quiet too
+        Assert.True(await conn.ReconfigureAsync("test"));
+        Assert.Empty(logger.Matching("Encountered error while updating cluster config"));
+    }
+}

tests/StackExchange.Redis.Tests/ConnectFailureRefreshTests.cs

@@ -145,34 +145,4 @@ public async Task TheReadIsRateLimitedRatherThanOncePerFailure()
             Assert.True(refreshes <= permitted, $"expected at most {permitted} rate-limited reads, but saw {refreshes}");
         }
     }
-
-    private sealed class CapturingLogger : ILoggerFactory, ILogger
-    {
-        private readonly List<string> _messages = [];
-
-        public ILogger CreateLogger(string categoryName) => this;
-
-        public void AddProvider(ILoggerProvider provider) { }
-
-        public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
-
-        public bool IsEnabled(LogLevel logLevel) => true;
-
-        public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func<TState, Exception?, string> formatter)
-        {
-            lock (_messages) _messages.Add(formatter(state, exception));
-        }
-
-        public List<string> Matching(string fragment)
-        {
-            lock (_messages) return _messages.FindAll(x => x.Contains(fragment, StringComparison.Ordinal));
-        }
-
-        public string All
-        {
-            get { lock (_messages) return string.Join("\n", _messages); }
-        }
-
-        public void Dispose() { }
-    }
 }

tests/StackExchange.Redis.Tests/Helpers/CapturingLogger.cs

@@ -0,0 +1,39 @@
+using System;
+using System.Collections.Generic;
+using Microsoft.Extensions.Logging;
+
+namespace StackExchange.Redis.Tests;
+
+/// <summary>
+/// An <see cref="ILoggerFactory"/> that records every formatted message, so a test can assert on what
+/// was (or was not) logged.
+/// </summary>
+internal sealed class CapturingLogger : ILoggerFactory, ILogger
+{
+    private readonly List<string> _messages = [];
+
+    public ILogger CreateLogger(string categoryName) => this;
+
+    public void AddProvider(ILoggerProvider provider) { }
+
+    public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
+
+    public bool IsEnabled(LogLevel logLevel) => true;
+
+    public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func<TState, Exception?, string> formatter)
+    {
+        lock (_messages) _messages.Add(formatter(state, exception));
+    }
+
+    public List<string> Matching(string fragment)
+    {
+        lock (_messages) return _messages.FindAll(x => x.Contains(fragment, StringComparison.Ordinal));
+    }
+
+    public string All
+    {
+        get { lock (_messages) return string.Join("\n", _messages); }
+    }
+
+    public void Dispose() { }
+}