Upgrade clients

Signed-off-by: Tomasz Maruszak <maruszaktomasz@gmail.com>

Author: zarusz

src/Tests/SlimMessageBus.Host.Kafka.Test/KafkaMessageBusIt.cs

@@ -120,9 +120,11 @@ public async Task BasicPubSub(int numberOfMessages, int delayProducerAt, int del
                 // Remove self to cause only delay once (in case the same message gets repeated)
                 pauseAtOffsets.Remove(message.Message.Counter);
 
-                consumerControl.Stop().ContinueWith(async (_) =>
+                // Use fire-and-forget pattern to avoid blocking the event handler
+                _ = Task.Run(async () =>
                 {
                     Logger.LogInformation("Consumer stopped at message {MessageCounter}, waiting {DelayTime} before restart", message.Message.Counter, DelayTimeSpan);
+                    await consumerControl.Stop();
                     await Task.Delay(DelayTimeSpan);
                     await consumerControl.Start();
 
@@ -133,7 +135,7 @@ public async Task BasicPubSub(int numberOfMessages, int delayProducerAt, int del
                         await Task.Delay(100);
                     }
                     Logger.LogInformation("Consumer restarted after {ElapsedTime}, IsStarted: {IsStarted}", timeout.Elapsed, consumerControl.IsStarted);
-                }).Wait();
+                });
             }
         };
 
@@ -177,7 +179,7 @@ public async Task BasicPubSub(int numberOfMessages, int delayProducerAt, int del
         await consumedMessages.WaitUntilArriving(newMessagesTimeout: consumeTimeout, expectedCount: messages.Count);
 
         stopwatch.Stop();
-        Logger.LogInformation("Consumed {MessageCount} messages in {ConsumeTime} including simlulated delay {DelayTime}", consumedMessages.Count, stopwatch.Elapsed, DelayTimeSpan);
+        Logger.LogInformation("Consumed {ActualCount}/{ExpectedCount} messages in {ConsumeTime} including simlulated delay {DelayTime}", consumedMessages.Count, messages.Count, stopwatch.Elapsed, DelayTimeSpan);
 
         // assert
 

src/Tests/SlimMessageBus.Host.Outbox.Sql.DbContext.Test/BaseOutboxIntegrationTest.cs

@@ -28,6 +28,8 @@ protected async Task WaitForOutboxToBeDrained(int timeoutSeconds = 10)
     {
         var stopwatch = Stopwatch.StartNew();
         var lastPendingCount = int.MaxValue;
+        var stuckCount = 0;
+        const int maxStuckIterations = 5; // If count doesn't change after 5 checks, exit early
 
         while (stopwatch.Elapsed.TotalSeconds < timeoutSeconds)
         {
@@ -58,8 +60,20 @@ protected async Task WaitForOutboxToBeDrained(int timeoutSeconds = 10)
                         return;
                     }
 
-                    if (pendingCount != lastPendingCount)
+                    // Track if count is stuck (not changing)
+                    if (pendingCount == lastPendingCount)
                     {
+                        stuckCount++;
+                        if (stuckCount >= maxStuckIterations)
+                        {
+                            Logger.LogWarning("Outbox pending count stuck at {PendingCount} for {StuckIterations} iterations after {Elapsed}, exiting early", 
+                                pendingCount, stuckCount, stopwatch.Elapsed);
+                            return;
+                        }
+                    }
+                    else
+                    {
+                        stuckCount = 0;
                         Logger.LogInformation("Outbox has {PendingCount} pending messages after {Elapsed}", pendingCount, stopwatch.Elapsed);
                         lastPendingCount = pendingCount;
                     }

src/Tests/SlimMessageBus.Host.Outbox.Sql.DbContext.Test/OutboxTests.cs

@@ -177,16 +177,10 @@ public async Task Given_CommandHandlerInTransaction_When_ExceptionThrownDuringHa
 
         await Task.WhenAll(sendTasks);
 
-        // Wait for outbox to be fully processed and drained
-        await WaitForOutboxToBeDrained(timeoutSeconds: 20);
-
-        // Wait for messages to be processed through outbox:
-        // - Max processing delay per message: 250ms
-        // - Total max delay: 100 messages / 3 valid per batch × 250ms ≈ 8.3s
-        // - Outbox poll interval: 250ms (reduced from 1s)
-        // - Add buffer: 5s (reduced since outbox is drained)
-        // Total: ~15s timeout
-        await store.WaitUntilArriving(newMessagesTimeout: 15, expectedCount: validCommands.Count);
+        // Wait for messages to arrive first (they should start arriving quickly)
+        await store.WaitUntilArriving(newMessagesTimeout: 20);
+        
+        Logger.LogInformation("Received {ActualCount}/{ExpectedCount} messages after initial wait", store.Count, validCommands.Count);
 
         // assert
         using var scope = ServiceProvider!.CreateScope();
@@ -281,11 +275,10 @@ public async Task Given_PublishExternalEventInTransaction_When_ExceptionThrownDu
 
         await Task.WhenAll(publishTasks);
 
-        // Wait for outbox to be fully processed and drained
-        await WaitForOutboxToBeDrained(timeoutSeconds: 20);
-
-        // Wait for messages to be processed through outbox (same reasoning as other test)
-        await store.WaitUntilArriving(newMessagesTimeout: 15, expectedCount: validEvents.Count);
+        // Wait for messages to arrive (they should start arriving quickly)
+        await store.WaitUntilArriving(newMessagesTimeout: 20);
+        
+        Logger.LogInformation("Received {ActualCount}/{ExpectedCount} messages after initial wait", store.Count, validEvents.Count);
 
         // assert