Fix deadlock in GetOrCreateTopicArnAsync causing CI tests to hang

RCA: EnsureTopicSubscriptionAsync held _lock and called GetOrCreateTopicArnAsync
which tried to acquire the same _lock. AsyncLock is not reentrant, causing deadlock.

Fix: Remove lock from GetOrCreateTopicArnAsync since caller already holds it.
Use ConcurrentDictionary.TryAdd for thread-safe caching of topic ARNs.

Author: niemyjski

src/Foundatio.AWS/Messaging/SQSMessageBus.cs

@@ -171,8 +171,8 @@ private string GetTopicName(Type messageType)
     /// <param name="cancellationToken">A cancellation token.</param>
     /// <returns>The ARN of the SNS topic.</returns>
     /// <remarks>
-    /// This method uses double-check locking to ensure thread-safe topic creation.
     /// Topic ARNs are cached in a ConcurrentDictionary to avoid redundant AWS API calls.
+    /// Caller must hold the lock when topic creation may occur.
     /// </remarks>
     private async Task<string> GetOrCreateTopicArnAsync(string topicName, CancellationToken cancellationToken)
     {
@@ -181,16 +181,9 @@ private async Task<string> GetOrCreateTopicArnAsync(string topicName, Cancellati
         if (_topicArns.TryGetValue(topicName, out string arn))
             return arn;
 
-        using (await _lock.LockAsync(cancellationToken).AnyContext())
-        {
-            // Double-check after acquiring lock
-            if (_topicArns.TryGetValue(topicName, out arn))
-                return arn;
-
-            arn = await CreateTopicImplAsync(topicName, cancellationToken).AnyContext();
-            _topicArns[topicName] = arn;
-            return arn;
-        }
+        arn = await CreateTopicImplAsync(topicName, cancellationToken).AnyContext();
+        _topicArns.TryAdd(topicName, arn);
+        return arn;
     }
 
     /// <summary>

tests/Foundatio.AWS.Tests/Messaging/SQSMessageBusTests.cs

@@ -213,7 +213,7 @@ public override Task CanHandlePoisonedMessageAsync()
         return base.CanHandlePoisonedMessageAsync();
     }
 
-    [Fact]
+    [Fact] // 2 minute timeout for durable queue test
     public async Task SubscribeAsync_WithDurableQueue_PersistsAcrossRestarts()
     {
         // Arrange
@@ -328,7 +328,7 @@ await messageBus.SubscribeAsync<SimpleMessageA>(msg =>
         Assert.Equal("DefaultTopicTest", receivedData);
     }
 
-    [Fact]
+    [Fact] // 2 minute timeout for multi-topic test
     public async Task PublishAsync_WithTopicResolver_RoutesToCorrectTopic()
     {
         // Arrange
@@ -512,7 +512,7 @@ public async Task GetOrCreateTopicArnAsync_WhenCalledConcurrently_OnlyCreatesOnc
         Assert.True(tasks.All(t => t.IsCompletedSuccessfully));
     }
 
-    [Fact]
+    [Fact] // 2 minute timeout for multi-subscriber test
     public async Task PublishAsync_WithDifferentTopicsPerType_SubscribersReceiveCorrectMessages()
     {
         // Arrange