Merge remote-tracking branch 'origin/marc/aa2' into marc/aa2

Author: mgravell

src/StackExchange.Redis/ConnectionMultiplexer.Debug.cs

@@ -4,9 +4,18 @@ namespace StackExchange.Redis;
 
 public partial class ConnectionMultiplexer
 {
-    private static int _collectedWithoutDispose;
+    private static int _collectedWithoutDispose, s_DisposedCount, s_MuxerCreateCount;
     internal static int CollectedWithoutDispose => Volatile.Read(ref _collectedWithoutDispose);
 
+    internal static int GetLiveObjectCount(out int created, out int disposed, out int finalized)
+    {
+        // read destroy first, to prevent negative numbers in race conditions
+        disposed = Volatile.Read(ref s_DisposedCount);
+        created = Volatile.Read(ref s_MuxerCreateCount);
+        finalized = Volatile.Read(ref _collectedWithoutDispose);
+        return created - (disposed + finalized);
+    }
+
     /// <summary>
     /// Invoked by the garbage collector.
     /// </summary>

src/StackExchange.Redis/ConnectionMultiplexer.cs

@@ -128,6 +128,8 @@ static ConnectionMultiplexer()
 
         private ConnectionMultiplexer(ConfigurationOptions configuration, ServerType? serverType = null, EndPointCollection? endpoints = null)
         {
+            Interlocked.Increment(ref s_MuxerCreateCount);
+
             RawConfig = configuration ?? throw new ArgumentNullException(nameof(configuration));
             EndPoints = endpoints ?? RawConfig.EndPoints.Clone();
             EndPoints.SetDefaultPorts(serverType, ssl: RawConfig.Ssl);
@@ -2260,7 +2262,8 @@ ConfigurationChangedChannel is byte[] channel
         public void Dispose()
         {
             GC.SuppressFinalize(this);
-            Close(!_isDisposed);
+            if (!_isDisposed) Interlocked.Increment(ref s_DisposedCount);
+            Close(!_isDisposed); // marks disposed
             sentinelConnection?.Dispose();
             var oldTimer = Interlocked.Exchange(ref sentinelPrimaryReconnectTimer, null);
             oldTimer?.Dispose();
@@ -2272,7 +2275,8 @@ public void Dispose()
         public async ValueTask DisposeAsync()
         {
             GC.SuppressFinalize(this);
-            await CloseAsync(!_isDisposed).ForAwait();
+            if (!_isDisposed) Interlocked.Increment(ref s_DisposedCount);
+            await CloseAsync(!_isDisposed).ForAwait(); // marks disposed
             if (sentinelConnection is ConnectionMultiplexer sentinel)
             {
                 await sentinel.DisposeAsync().ForAwait();

src/StackExchange.Redis/ExceptionFactory.cs

@@ -347,6 +347,12 @@ private static void AddCommonDetail(
                 Add(data, sb, "Last-Result-Bytes", "last-in", bs.Connection.BytesLastResult.ToString());
                 Add(data, sb, "Inbound-Buffer-Bytes", "cur-in", bs.Connection.BytesInBuffer.ToString());
 
+                var liveMuxers = ConnectionMultiplexer.GetLiveObjectCount(out var created, out var disposed, out var finalized);
+                if (created > 1)
+                {
+                    Add(data, sb, "Live-Multiplexers", "lm", $"{liveMuxers}/{created}/{disposed}/{finalized}");
+                }
+
                 Add(data, sb, "Sync-Ops", "sync-ops", multiplexer.syncOps.ToString());
                 Add(data, sb, "Async-Ops", "async-ops", multiplexer.asyncOps.ToString());
 

src/StackExchange.Redis/PhysicalBridge.cs

@@ -1622,8 +1622,13 @@ private WriteResult WriteMessageToServerInsideWriteLock(PhysicalConnection conne
 
                 if (_nextHighIntegrityToken is not 0
                     && !connection.TransactionActive // validated in the UNWATCH/EXEC/DISCARD
-                    && message.Command is not (RedisCommand.AUTH or RedisCommand.HELLO)) // if auth fails, ECHO may also fail; avoid confusion
+                    && message.Command is not RedisCommand.AUTH // if auth fails, later commands may also fail; avoid confusion
+                    && message.Command is not RedisCommand.HELLO)
                 {
+                    // note on the Command match above: curiously, .NET 10 and .NET 11 SDKs emit *opposite* errors here
+                    // re "CS9336: The pattern is redundant." ("fixing" one "breaks" the other); possibly a fixed bool inversion
+                    // in the analyzer? to avoid pain, we'll just use the most obviously correct form
+
                     // make sure this value exists early to avoid a race condition
                     // if the response comes back super quickly
                     message.WithHighIntegrity(NextHighIntegrityTokenInsideLock());

tests/StackExchange.Redis.Tests/BasicOpTests.cs

@@ -12,11 +12,13 @@ public class BasicOpsTests(ITestOutputHelper output, SharedConnectionFixture fix
 {
 }
 
+/*
 [RunPerProtocol]
 public class InProcBasicOpsTests(ITestOutputHelper output, InProcServerFixture fixture)
     : BasicOpsTestsBase(output, null, fixture)
 {
 }
+*/
 
 [RunPerProtocol]
 public abstract class BasicOpsTestsBase(ITestOutputHelper output, SharedConnectionFixture? connection, InProcServerFixture? server)

tests/StackExchange.Redis.Tests/HighIntegrityBasicOpsTests.cs

@@ -7,7 +7,9 @@ public class HighIntegrityBasicOpsTests(ITestOutputHelper output, SharedConnecti
     internal override bool HighIntegrity => true;
 }
 
+/*
 public class InProcHighIntegrityBasicOpsTests(ITestOutputHelper output, InProcServerFixture fixture) : InProcBasicOpsTests(output, fixture)
 {
     internal override bool HighIntegrity => true;
 }
+*/

tests/StackExchange.Redis.Tests/InProcessTestServer.cs

@@ -25,24 +25,14 @@ public InProcessTestServer(ITestOutputHelper? log = null, EndPoint? endpoint = n
         Tunnel = new InProcTunnel(this);
     }
 
-    public Task<ConnectionMultiplexer> ConnectAsync(bool withPubSub = false, TextWriter? log = null)
-        => ConnectionMultiplexer.ConnectAsync(GetClientConfig(withPubSub), log);
+    public Task<ConnectionMultiplexer> ConnectAsync(TextWriter? log = null)
+        => ConnectionMultiplexer.ConnectAsync(GetClientConfig(), log);
 
-    public ConfigurationOptions GetClientConfig(bool withPubSub = false)
+    public ConfigurationOptions GetClientConfig()
     {
         var commands = GetCommands();
-        if (!withPubSub)
-        {
-            commands.Remove(nameof(RedisCommand.SUBSCRIBE));
-            commands.Remove(nameof(RedisCommand.PSUBSCRIBE));
-            commands.Remove(nameof(RedisCommand.SSUBSCRIBE));
-            commands.Remove(nameof(RedisCommand.UNSUBSCRIBE));
-            commands.Remove(nameof(RedisCommand.PUNSUBSCRIBE));
-            commands.Remove(nameof(RedisCommand.SUNSUBSCRIBE));
-            commands.Remove(nameof(RedisCommand.PUBLISH));
-            commands.Remove(nameof(RedisCommand.SPUBLISH));
-        }
-        // transactions don't work yet
+
+        // transactions don't work yet (needs v3 buffer features)
         commands.Remove(nameof(RedisCommand.MULTI));
         commands.Remove(nameof(RedisCommand.EXEC));
         commands.Remove(nameof(RedisCommand.DISCARD));
@@ -82,6 +72,22 @@ protected override void OnMoved(RedisClient client, int hashSlot, Node node)
         base.OnMoved(client, hashSlot, node);
     }
 
+    protected override void OnOutOfBand(RedisClient client, TypedRedisValue message)
+    {
+        if (message.IsAggregate
+            && message.Span is { IsEmpty: false } span
+            && !span[0].IsAggregate)
+        {
+            _log?.WriteLine($"Client {client.Id}: {span[0].AsRedisValue()} {message} ");
+        }
+        else
+        {
+            _log?.WriteLine($"Client {client.Id}: {message}");
+        }
+
+        base.OnOutOfBand(client, message);
+    }
+
     public override TypedRedisValue OnUnknownCommand(in RedisClient client, in RedisRequest request, ReadOnlySpan<byte> command)
     {
         _log?.WriteLine($"[{client.Id}] unknown command: {Encoding.ASCII.GetString(command)}");

tests/StackExchange.Redis.Tests/PubSubTests.cs

@@ -11,12 +11,31 @@
 namespace StackExchange.Redis.Tests;
 
 [RunPerProtocol]
-public class PubSubTests(ITestOutputHelper output, SharedConnectionFixture fixture) : TestBase(output, fixture)
+public class PubSubTests(ITestOutputHelper output, SharedConnectionFixture fixture)
+    : PubSubTestBase(output, fixture, null)
+{
+}
+
+/*
+[RunPerProtocol]
+public class InProcPubSubTests(ITestOutputHelper output, InProcServerFixture fixture)
+    : PubSubTestBase(output, null, fixture)
+{
+    protected override bool UseDedicatedInProcessServer => false;
+}
+*/
+
+[RunPerProtocol]
+public abstract class PubSubTestBase(
+    ITestOutputHelper output,
+    SharedConnectionFixture? connection,
+    InProcServerFixture? server)
+    : TestBase(output, connection, server)
 {
     [Fact]
     public async Task ExplicitPublishMode()
     {
-        await using var conn = Create(channelPrefix: "foo:", log: Writer);
+        await using var conn = ConnectFactory(channelPrefix: "foo:");
 
         var pub = conn.GetSubscriber();
         int a = 0, b = 0, c = 0, d = 0;
@@ -54,9 +73,9 @@ await UntilConditionAsync(
     [InlineData("Foo:", true, "f")]
     public async Task TestBasicPubSub(string? channelPrefix, bool wildCard, string breaker)
     {
-        await using var conn = Create(channelPrefix: channelPrefix, shared: false, log: Writer);
+        await using var conn = ConnectFactory(channelPrefix: channelPrefix, shared: false);
 
-        var pub = GetAnyPrimary(conn);
+        var pub = GetAnyPrimary(conn.DefaultClient);
         var sub = conn.GetSubscriber();
         await PingAsync(pub, sub).ForAwait();
         HashSet<string?> received = [];
@@ -139,10 +158,10 @@ public async Task TestBasicPubSub(string? channelPrefix, bool wildCard, string b
     [Fact]
     public async Task TestBasicPubSubFireAndForget()
     {
-        await using var conn = Create(shared: false, log: Writer);
+        await using var conn = ConnectFactory(shared: false);
 
-        var profiler = conn.AddProfiler();
-        var pub = GetAnyPrimary(conn);
+        var profiler = conn.DefaultClient.AddProfiler();
+        var pub = GetAnyPrimary(conn.DefaultClient);
         var sub = conn.GetSubscriber();
 
         RedisChannel key = RedisChannel.Literal(Me() + Guid.NewGuid());
@@ -214,9 +233,9 @@ private async Task PingAsync(IServer pub, ISubscriber sub, int times = 1)
     [Fact]
     public async Task TestPatternPubSub()
     {
-        await using var conn = Create(shared: false, log: Writer);
+        await using var conn = ConnectFactory(shared: false);
 
-        var pub = GetAnyPrimary(conn);
+        var pub = GetAnyPrimary(conn.DefaultClient);
         var sub = conn.GetSubscriber();
 
         HashSet<string?> received = [];
@@ -273,7 +292,7 @@ public async Task TestPatternPubSub()
     [Fact]
     public async Task TestPublishWithNoSubscribers()
     {
-        await using var conn = Create();
+        await using var conn = ConnectFactory();
 
         var sub = conn.GetSubscriber();
 #pragma warning disable CS0618
@@ -285,7 +304,7 @@ public async Task TestPublishWithNoSubscribers()
     public async Task TestMassivePublishWithWithoutFlush_Local()
     {
         Skip.UnlessLongRunning();
-        await using var conn = Create();
+        await using var conn = ConnectFactory();
 
         var sub = conn.GetSubscriber();
         TestMassivePublish(sub, Me(), "local");
@@ -335,7 +354,7 @@ private void TestMassivePublish(ISubscriber sub, string channel, string caption)
     [Fact]
     public async Task SubscribeAsyncEnumerable()
     {
-        await using var conn = Create(syncTimeout: 20000, shared: false, log: Writer);
+        await using var conn = ConnectFactory(shared: false);
 
         var sub = conn.GetSubscriber();
         RedisChannel channel = RedisChannel.Literal(Me());
@@ -370,7 +389,7 @@ public async Task SubscribeAsyncEnumerable()
     [Fact]
     public async Task PubSubGetAllAnyOrder()
     {
-        await using var conn = Create(syncTimeout: 20000, shared: false, log: Writer);
+        await using var conn = ConnectFactory(shared: false);
 
         var sub = conn.GetSubscriber();
         RedisChannel channel = RedisChannel.Literal(Me());
@@ -625,9 +644,10 @@ public async Task PubSubGetAllCorrectOrder_OnMessage_Async()
     [Fact]
     public async Task TestPublishWithSubscribers()
     {
-        await using var connA = Create(shared: false, log: Writer);
-        await using var connB = Create(shared: false, log: Writer);
-        await using var connPub = Create();
+        await using var pair = ConnectFactory(shared: false);
+        await using var connA = pair.DefaultClient;
+        await using var connB = pair.CreateClient();
+        await using var connPub = pair.CreateClient();
 
         var channel = Me();
         var listenA = connA.GetSubscriber();
@@ -652,9 +672,10 @@ public async Task TestPublishWithSubscribers()
     [Fact]
     public async Task TestMultipleSubscribersGetMessage()
     {
-        await using var connA = Create(shared: false, log: Writer);
-        await using var connB = Create(shared: false, log: Writer);
-        await using var connPub = Create();
+        await using var pair = ConnectFactory(shared: false);
+        await using var connA = pair.DefaultClient;
+        await using var connB = pair.CreateClient();
+        await using var connPub = pair.CreateClient();
 
         var channel = RedisChannel.Literal(Me());
         var listenA = connA.GetSubscriber();
@@ -682,7 +703,7 @@ public async Task TestMultipleSubscribersGetMessage()
     [Fact]
     public async Task Issue38()
     {
-        await using var conn = Create(log: Writer);
+        await using var conn = ConnectFactory();
 
         var sub = conn.GetSubscriber();
         int count = 0;
@@ -717,9 +738,10 @@ public async Task Issue38()
     [Fact]
     public async Task TestPartialSubscriberGetMessage()
     {
-        await using var connA = Create();
-        await using var connB = Create();
-        await using var connPub = Create();
+        await using var pair = ConnectFactory();
+        await using var connA = pair.DefaultClient;
+        await using var connB = pair.CreateClient();
+        await using var connPub = pair.CreateClient();
 
         int gotA = 0, gotB = 0;
         var listenA = connA.GetSubscriber();
@@ -750,8 +772,9 @@ public async Task TestPartialSubscriberGetMessage()
     [Fact]
     public async Task TestSubscribeUnsubscribeAndSubscribeAgain()
     {
-        await using var connPub = Create();
-        await using var connSub = Create();
+        await using var pair = ConnectFactory();
+        await using var connPub = pair.DefaultClient;
+        await using var connSub = pair.CreateClient();
 
         var prefix = Me();
         var pub = connPub.GetSubscriber();