Merge pull request #90 from tgiachi/develop

release: SQS subscription wiring fix

Author: tgiachi

src/SquidStd.Messaging.Sqs/Services/SqsTopicProvider.cs

@@ -149,8 +149,19 @@ Func<ReadOnlyMemory<byte>, CancellationToken, Task> handler
             _handler = handler;
         }
 
+        // SNS only fans a publish out to subscriptions that are already confirmed at publish time
+        // (unlike a plain SQS queue, it does not buffer for endpoints that subscribe later), so the
+        // subscribe/queue/policy/SNS-subscribe wiring below must complete before Start() returns.
+        // This mirrors RabbitMqTopicProvider.Subscription.Start(), which blocks the same way.
         public void Start()
-            => _loop = Task.Run(() => RunAsync(_cts.Token));
+        {
+            var queueUrl = SetupAsync(_cts.Token).GetAwaiter().GetResult();
+
+            if (queueUrl is not null)
+            {
+                _loop = Task.Run(() => ReceiveLoopAsync(queueUrl, _cts.Token));
+            }
+        }
 
         private static string BuildPolicy(string queueArn, string topicArn)
             => JsonSerializer.Serialize(
@@ -171,18 +182,21 @@ private static string BuildPolicy(string queueArn, string topicArn)
                 }
             );
 
-        private async Task RunAsync(CancellationToken cancellationToken)
+        /// <summary>
+        /// Creates the subscriber's dedicated queue, grants SNS permission to send to it and
+        /// confirms the SNS subscription. Returns the queue URL once the subscription is fully
+        /// wired and ready to receive fanned-out messages, or null on failure/cancellation.
+        /// </summary>
+        private async Task<string?> SetupAsync(CancellationToken cancellationToken)
         {
-            string queueUrl;
-
             try
             {
                 var topicArn = await _provider.EnsureTopicAsync(_topic, cancellationToken);
                 var queueName = SqsNames.Sanitize(_topic) + "-sub-" + _index;
-                queueUrl = (await _provider._sqs!.CreateQueueAsync(
-                                new CreateQueueRequest { QueueName = queueName },
-                                cancellationToken
-                            )).QueueUrl;
+                var queueUrl = (await _provider._sqs!.CreateQueueAsync(
+                                    new CreateQueueRequest { QueueName = queueName },
+                                    cancellationToken
+                                )).QueueUrl;
                 _queueUrl = queueUrl;
 
                 var attributes = await _provider._sqs.GetQueueAttributesAsync(
@@ -211,18 +225,23 @@ await _provider._sqs.SetQueueAttributesAsync(
                                         },
                                         cancellationToken
                                     )).SubscriptionArn;
+
+                return queueUrl;
             }
             catch (OperationCanceledException)
             {
-                return;
+                return null;
             }
             catch (Exception ex)
             {
                 _provider._logger.Warning(ex, "SQS topic '{Topic}' subscribe setup failed", _topic);
 
-                return;
+                return null;
             }
+        }
 
+        private async Task ReceiveLoopAsync(string queueUrl, CancellationToken cancellationToken)
+        {
             while (!cancellationToken.IsCancellationRequested)
             {
                 ReceiveMessageResponse response;

tests/SquidStd.Tests/Messaging/InMemoryQueueProviderTests.cs

@@ -163,7 +163,10 @@ public async Task TwoSubscribers_ReceiveRoundRobin()
             await provider.PublishAsync("q", Bytes("m"));
         }
 
-        Assert.True(done.Wait(Timeout));
+        // The single consumer loop is dispatched via Task.Run and dispatches round-robin
+        // strictly in registration order, so the split is deterministic once all four
+        // deliveries land; a generous bound just absorbs thread-pool scheduling delay under load.
+        Assert.True(done.Wait(TimeSpan.FromSeconds(30)));
         Assert.Equal(2, aCount);
         Assert.Equal(2, bCount);
     }

tests/SquidStd.Tests/Messaging/Sqs/SqsTopicProviderTests.cs

@@ -42,8 +42,8 @@ public async Task Publish_FansOutToAllSubscribers()
             }
         );
 
-        // Allow the subscriptions to be wired before publishing.
-        await Task.Delay(2000);
+        // Subscribe() now blocks until the SNS subscription is confirmed (see SqsTopicProvider),
+        // so both subscriptions are already fanned-in by the time it returns; no delay needed.
         await provider.PublishAsync(topic, Bytes("hello"));
 
         Assert.Equal("hello", await a.Task.WaitAsync(Timeout));

tests/SquidStd.Tests/Services/Core/MetricsCollectionServiceTests.cs

@@ -83,7 +83,14 @@ [new CountingMetricProvider("events", "published.total", 5)],
         );
 
         await service.StartAsync(CancellationToken.None);
-        await WaitUntilAsync(() => firstListener.LastEvent is not null && secondListener.LastEvent is not null);
+
+        // The collection loop's first tick still has to be scheduled via Task.Run, so give it
+        // more headroom than the default: on a loaded CI runner that scheduling can lag well
+        // past a couple of seconds even though the collector itself collects immediately.
+        await WaitUntilAsync(
+            () => firstListener.LastEvent is not null && secondListener.LastEvent is not null,
+            TimeSpan.FromSeconds(30)
+        );
         await service.StopAsync(CancellationToken.None);
 
         Assert.NotNull(firstListener.LastEvent);
@@ -161,9 +168,9 @@ public async Task StopAsync_StopsCollectionLoop()
         Assert.Equal(countAfterStop, provider.CollectionCount);
     }
 
-    private static async Task WaitUntilAsync(Func<bool> predicate)
+    private static async Task WaitUntilAsync(Func<bool> predicate, TimeSpan? timeout = null)
     {
-        var deadline = DateTime.UtcNow.AddSeconds(2);
+        var deadline = DateTime.UtcNow + (timeout ?? TimeSpan.FromSeconds(2));
 
         while (DateTime.UtcNow < deadline)
         {

tests/SquidStd.Tests/Services/Core/Scheduling/TimerWheelPumpServiceTests.cs

@@ -25,7 +25,9 @@ public async Task Pump_AdvancesTheWheel()
 
         await pump.StartAsync();
 
-        Assert.True(timer.Pumped.Wait(TimeSpan.FromSeconds(2)));
+        // The pump loop is dispatched via Task.Run; a generous bound absorbs thread-pool
+        // scheduling delay on loaded CI runners while still failing fast if it never pumps.
+        Assert.True(timer.Pumped.Wait(TimeSpan.FromSeconds(30)));
 
         await pump.StopAsync();
         pump.Dispose();