Fix flakey Tests (#129)

* Make sure we are unsubscribed from the channel.

Use async unsubscribe with a bounded timeout to prevent hanging

* Improves Redis unsubscribe reliability

Uses fire and forget unsubscribe to prevent potential hanging issues.

Adds unique topic names to Redis tests to prevent interference
between tests.

* Handles exceptions during Redis unsubscribe

Adds try-catch blocks when unsubscribing from Redis topics to prevent potential unhandled exceptions from causing issues, particularly in test environments where flaky behavior might occur. Logs any exceptions encountered during the unsubscribe process for better debugging.

* Added trace logging to failing test

* Updated timeout

Author: niemyjski

src/Foundatio.Redis/Foundatio.Redis.csproj

@@ -1,7 +1,7 @@
 <Project Sdk="Microsoft.NET.Sdk">
   <Import Project="..\..\build\common.props" />
   <ItemGroup>
-    <PackageReference Include="Foundatio" Version="12.0.1-preview.0.42" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
+    <PackageReference Include="Foundatio" Version="12.0.1-preview.0.44" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
     <ProjectReference Include="..\..\..\Foundatio\src\Foundatio\Foundatio.csproj" Condition="'$(ReferenceFoundatioSource)' == 'true'" />
 
     <PackageReference Include="StackExchange.Redis" Version="2.10.1" />

src/Foundatio.Redis/Queues/RedisQueue.cs

@@ -778,7 +778,14 @@ public override void Dispose()
                 if (_isSubscribed)
                 {
                     _logger.LogTrace("Unsubscribing from topic {Topic}", GetTopicName());
-                    _subscriber.Unsubscribe(RedisChannel.Literal(GetTopicName()), OnTopicMessage, CommandFlags.FireAndForget);
+                    try
+                    {
+                        _subscriber.Unsubscribe(RedisChannel.Literal(GetTopicName()), OnTopicMessage, CommandFlags.FireAndForget);
+                    }
+                    catch (Exception ex)
+                    {
+                        _logger.LogError(ex, "Error unsubscribing from topic {Topic}: {Message}", GetTopicName(), ex.Message);
+                    }
                     _isSubscribed = false;
                     _logger.LogTrace("Unsubscribed from topic {Topic}", GetTopicName());
                 }

tests/Foundatio.Redis.Tests/Foundatio.Redis.Tests.csproj

@@ -8,7 +8,7 @@
     <PackageReference Include="xunit.runner.visualstudio" Version="3.1.5" />
     <PackageReference Include="GitHubActionsTestLogger" Version="3.0.1" PrivateAssets="All" />
 
-    <PackageReference Include="Foundatio.TestHarness" Version="12.0.1-preview.0.42" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
+    <PackageReference Include="Foundatio.TestHarness" Version="12.0.1-preview.0.44" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
     <ProjectReference Include="..\..\..\Foundatio\src\Foundatio.TestHarness\Foundatio.TestHarness.csproj" Condition="'$(ReferenceFoundatioSource)' == 'true'" />
   </ItemGroup>
   <ItemGroup>

tests/Foundatio.Redis.Tests/Locks/RedisLockTests.cs

@@ -12,14 +12,15 @@ namespace Foundatio.Redis.Tests.Locks;
 
 public class RedisLockTests : LockTestBase, IDisposable, IAsyncLifetime
 {
+    private readonly string _topic = $"test-lock-{Guid.NewGuid().ToString("N")[..10]}";
     private readonly ICacheClient _cache;
     private readonly IMessageBus _messageBus;
 
     public RedisLockTests(ITestOutputHelper output) : base(output)
     {
         var muxer = SharedConnection.GetMuxer(Log);
         _cache = new RedisCacheClient(o => o.ConnectionMultiplexer(muxer).LoggerFactory(Log));
-        _messageBus = new RedisMessageBus(o => o.Subscriber(muxer.GetSubscriber()).Topic("test-lock").LoggerFactory(Log));
+        _messageBus = new RedisMessageBus(o => o.Subscriber(muxer.GetSubscriber()).Topic(_topic).LoggerFactory(Log));
     }
 
     protected override ILockProvider GetThrottlingLockProvider(int maxHits, TimeSpan period)

tests/Foundatio.Redis.Tests/Messaging/RedisMessageBusTests.cs

@@ -16,6 +16,8 @@ namespace Foundatio.Redis.Tests.Messaging;
 
 public class RedisMessageBusTests : MessageBusTestBase, IAsyncLifetime
 {
+    private readonly string _topic = $"test-messages-{Guid.NewGuid().ToString("N")[..10]}";
+
     public RedisMessageBusTests(ITestOutputHelper output) : base(output)
     {
     }
@@ -25,7 +27,7 @@ protected override IMessageBus GetMessageBus(Func<SharedMessageBusOptions, Share
         return new RedisMessageBus(o =>
         {
             o.Subscriber(SharedConnection.GetMuxer(Log).GetSubscriber());
-            o.Topic("test-messages");
+            o.Topic(_topic);
             o.LoggerFactory(Log);
             if (config != null)
                 config(o.Target);
@@ -152,7 +154,7 @@ public override Task CanHandlePoisonedMessageAsync()
     public async Task CanDisposeCacheAndQueueAndReceiveSubscribedMessages()
     {
         var muxer = SharedConnection.GetMuxer(Log);
-        var messageBus1 = new RedisMessageBus(new RedisMessageBusOptions { Subscriber = muxer.GetSubscriber(), Topic = "test-messages", LoggerFactory = Log });
+        var messageBus1 = new RedisMessageBus(new RedisMessageBusOptions { Subscriber = muxer.GetSubscriber(), Topic = _topic, LoggerFactory = Log });
 
         var cache = new RedisCacheClient(new RedisCacheClientOptions { ConnectionMultiplexer = muxer });
         Assert.NotNull(cache);

tests/Foundatio.Redis.Tests/Queues/RedisQueueTests.cs

@@ -25,6 +25,8 @@ namespace Foundatio.Redis.Tests.Queues;
 
 public class RedisQueueTests : QueueTestBase, IAsyncLifetime
 {
+    private readonly string _topic = $"test-queue-{Guid.NewGuid().ToString("N")[..10]}";
+
     public RedisQueueTests(ITestOutputHelper output) : base(output)
     {
     }
@@ -185,7 +187,7 @@ public override async Task CanDequeueWithLockingAsync()
     {
         var muxer = SharedConnection.GetMuxer(Log);
         using var cache = new RedisCacheClient(new RedisCacheClientOptions { ConnectionMultiplexer = muxer, LoggerFactory = Log });
-        using var messageBus = new RedisMessageBus(new RedisMessageBusOptions { Subscriber = muxer.GetSubscriber(), Topic = "test-queue", LoggerFactory = Log });
+        using var messageBus = new RedisMessageBus(new RedisMessageBusOptions { Subscriber = muxer.GetSubscriber(), Topic = _topic, LoggerFactory = Log });
         var distributedLock = new CacheLockProvider(cache, messageBus, null, Log);
         await CanDequeueWithLockingImpAsync(distributedLock);
     }
@@ -195,7 +197,7 @@ public override async Task CanHaveMultipleQueueInstancesWithLockingAsync()
     {
         var muxer = SharedConnection.GetMuxer(Log);
         using var cache = new RedisCacheClient(new RedisCacheClientOptions { ConnectionMultiplexer = muxer, LoggerFactory = Log });
-        using var messageBus = new RedisMessageBus(new RedisMessageBusOptions { Subscriber = muxer.GetSubscriber(), Topic = "test-queue", LoggerFactory = Log });
+        using var messageBus = new RedisMessageBus(new RedisMessageBusOptions { Subscriber = muxer.GetSubscriber(), Topic = _topic, LoggerFactory = Log });
         var distributedLock = new CacheLockProvider(cache, messageBus, null, Log);
         await CanHaveMultipleQueueInstancesWithLockingImplAsync(distributedLock);
     }