feat(pubsub): add thread safety to PubSub handler access in BaseClient

- Add volatile modifier to _pubSubHandler field for memory barrier guarantees
- Add _pubSubLock object for coordinating thread-safe access
- Implement lock-based handler access in HandlePubSubMessage() to prevent race conditions
- Update InitializePubSubHandler() with thread-safe initialization
- Enhance CleanupPubSubResources() with proper synchronization and timeout-based cleanup
- Add thread-safe access to PubSubQueue property

Add comprehensive thread safety tests:
- Test concurrent message processing from 100+ threads
- Test disposal during active message processing
- Add stress test with 100 iterations of concurrent operations
- Test concurrent access to PubSubQueue and HasPubSubSubscriptions properties
- Test rapid create/dispose cycles for memory leak detection
- Test disposal timeout handling

All 1,771 tests pass (250 unit + 1,521 integration tests).

Addresses requirements 2.1-2.6 from pubsub-critical-fixes spec.

Signed-off-by: Joe Brinkman <joe.brinkman@improving.com>

Author: jbrinkman

sources/Valkey.Glide/BaseClient.cs

@@ -45,11 +45,22 @@ public void Dispose()
     /// <summary>
     /// Get the PubSub message queue for manual message retrieval.
     /// Returns null if no PubSub subscriptions are configured.
+    /// Uses thread-safe access to prevent race conditions.
     /// </summary>
-    public PubSubMessageQueue? PubSubQueue => _pubSubHandler?.GetQueue();
+    public PubSubMessageQueue? PubSubQueue
+    {
+        get
+        {
+            lock (_pubSubLock)
+            {
+                return _pubSubHandler?.GetQueue();
+            }
+        }
+    }
 
     /// <summary>
     /// Indicates whether this client has PubSub subscriptions configured.
+    /// Uses volatile read for thread-safe access without locking.
     /// </summary>
     public bool HasPubSubSubscriptions => _pubSubHandler != null;
 
@@ -259,6 +270,7 @@ private static PubSubMessage MarshalPubSubMessage(
 
     /// <summary>
     /// Initializes PubSub message handling if PubSub subscriptions are configured.
+    /// Uses thread-safe initialization to ensure proper visibility across threads.
     /// </summary>
     /// <param name="config">The PubSub subscription configuration.</param>
     private void InitializePubSubHandler(BasePubSubSubscriptionConfig? config)
@@ -268,45 +280,82 @@ private void InitializePubSubHandler(BasePubSubSubscriptionConfig? config)
             return;
         }
 
-        // Create the PubSub message handler
-        _pubSubHandler = new PubSubMessageHandler(config.Callback, config.Context);
+        // Create the PubSub message handler with thread-safe initialization
+        lock (_pubSubLock)
+        {
+            _pubSubHandler = new PubSubMessageHandler(config.Callback, config.Context);
+        }
     }
 
     /// <summary>
     /// Handles incoming PubSub messages from the FFI layer.
-    /// This method is called directly by the FFI callback.
+    /// This method is called directly by the FFI callback and uses thread-safe access to the handler.
     /// </summary>
     /// <param name="message">The PubSub message to handle.</param>
     internal virtual void HandlePubSubMessage(PubSubMessage message)
     {
-        try
+        // Thread-safe access to handler - use local copy to avoid race conditions
+        PubSubMessageHandler? handler;
+        lock (_pubSubLock)
         {
-            _pubSubHandler?.HandleMessage(message);
+            handler = _pubSubHandler;
         }
-        catch (Exception ex)
+
+        if (handler != null)
         {
-            // Log the error but don't let exceptions escape
-            Logger.Log(Level.Error, "BaseClient", $"Error handling PubSub message: {ex.Message}", ex);
+            try
+            {
+                handler.HandleMessage(message);
+            }
+            catch (Exception ex)
+            {
+                // Log the error but don't let exceptions escape
+                Logger.Log(Level.Error, "BaseClient", $"Error handling PubSub message: {ex.Message}", ex);
+            }
         }
     }
 
     /// <summary>
-    /// Cleans up PubSub resources during client disposal.
+    /// Cleans up PubSub resources during client disposal with proper synchronization.
+    /// Uses locking to coordinate safe disposal and prevent conflicts with concurrent message processing.
     /// </summary>
     private void CleanupPubSubResources()
     {
-        if (_pubSubHandler != null)
+        PubSubMessageHandler? handler = null;
+
+        // Acquire lock and capture handler reference, then set to null
+        lock (_pubSubLock)
+        {
+            handler = _pubSubHandler;
+            _pubSubHandler = null;
+        }
+
+        // Dispose outside of lock to prevent deadlocks
+        if (handler != null)
         {
             try
             {
-                // Dispose the message handler
-                _pubSubHandler.Dispose();
-                _pubSubHandler = null;
+                // Create a task to dispose the handler with timeout
+                var disposeTask = Task.Run(() => handler.Dispose());
+
+                // Wait for disposal with timeout (5 seconds)
+                if (!disposeTask.Wait(TimeSpan.FromSeconds(5)))
+                {
+                    Logger.Log(Level.Warn, "BaseClient",
+                        "PubSub handler disposal did not complete within timeout (5 seconds)");
+                }
+            }
+            catch (AggregateException ex)
+            {
+                // Log the error but continue with disposal
+                Logger.Log(Level.Warn, "BaseClient",
+                    $"Error cleaning up PubSub resources: {ex.InnerException?.Message ?? ex.Message}", ex);
             }
             catch (Exception ex)
             {
                 // Log the error but continue with disposal
-                Logger.Log(Level.Warn, "BaseClient", $"Error cleaning up PubSub resources: {ex.Message}", ex);
+                Logger.Log(Level.Warn, "BaseClient",
+                    $"Error cleaning up PubSub resources: {ex.Message}", ex);
             }
         }
     }
@@ -348,7 +397,11 @@ private delegate void PubSubAction(
     protected Version? _serverVersion; // cached server version
 
     /// PubSub message handler for routing messages to callbacks or queues.
-    private PubSubMessageHandler? _pubSubHandler;
+    /// Uses volatile to ensure visibility across threads without locking on every read.
+    private volatile PubSubMessageHandler? _pubSubHandler;
+
+    /// Lock object for coordinating PubSub handler access and disposal.
+    private readonly object _pubSubLock = new();
 
     #endregion private fields
 }

tests/Valkey.Glide.UnitTests/PubSubFFIMemoryLeakTests.cs

@@ -1,14 +1,7 @@
 // Copyright Valkey GLIDE Project Contributors - SPDX Identifier: Apache-2.0
 
-using System;
 using System.Diagnostics;
 using System.Runtime.InteropServices;
-using System.Threading;
-using System.Threading.Tasks;
-
-using Valkey.Glide.Internals;
-
-using Xunit;
 
 namespace Valkey.Glide.UnitTests;
 
@@ -88,7 +81,7 @@ public void ProcessVariousMessageSizes_NoMemoryLeak_ConsistentBehavior()
     {
         // Arrange
         const int iterationsPerSize = 1_000;
-        var messageSizes = new[] { 10, 100, 1_000, 10_000, 100_000 }; // Various sizes
+        int[] messageSizes = [10, 100, 1_000, 10_000, 100_000]; // Various sizes
 
         GC.Collect();
         GC.WaitForPendingFinalizers();
@@ -102,7 +95,7 @@ public void ProcessVariousMessageSizes_NoMemoryLeak_ConsistentBehavior()
         {
             Console.WriteLine($"Testing message size: {messageSize} bytes");
 
-            string largeMessage = new string('X', messageSize);
+            string largeMessage = new('X', messageSize);
             string channel = "test-channel";
 
             long beforeSizeTest = GC.GetTotalMemory(true);
@@ -164,7 +157,7 @@ public void ProcessMessagesUnderGCPressure_NoMemoryLeak_StableUnderPressure()
             if (i % gcInterval == 0)
             {
                 // Create some temporary objects to increase GC pressure
-                var tempObjects = new object[1000];
+                object[] tempObjects = new object[1000];
                 for (int j = 0; j < tempObjects.Length; j++)
                 {
                     tempObjects[j] = new byte[1024]; // 1KB objects
@@ -209,8 +202,8 @@ public void ProcessConcurrentMessages_NoMemoryLeak_ThreadSafeMemoryManagement()
         Console.WriteLine($"Initial memory for concurrent test: {initialMemory:N0} bytes");
 
         // Act: Process messages concurrently from multiple threads
-        var tasks = new Task[threadsCount];
-        var exceptions = new Exception?[threadsCount];
+        Task[] tasks = new Task[threadsCount];
+        Exception?[] exceptions = new Exception?[threadsCount];
 
         for (int threadIndex = 0; threadIndex < threadsCount; threadIndex++)
         {
@@ -278,9 +271,9 @@ public void ProcessExtendedDuration_NoMemoryLeak_StableOverTime()
         Console.WriteLine($"Starting extended duration test for {durationSeconds} seconds");
         Console.WriteLine($"Initial memory: {initialMemory:N0} bytes");
 
-        var stopwatch = Stopwatch.StartNew();
+        Stopwatch stopwatch = Stopwatch.StartNew();
         int messageCount = 0;
-        var memorySnapshots = new List<(TimeSpan Time, long Memory)>();
+        List<(TimeSpan Time, long Memory)> memorySnapshots = [];
 
         // Act: Process messages for extended duration
         while (stopwatch.Elapsed.TotalSeconds < durationSeconds)