fix: properly handle acknowledge in OutboxWorker

BEagle1984 · May 19, 2023 · eb5dd2a · eb5dd2a
1 parent 37f06ae
commit eb5dd2a
Show file tree

Hide file tree

Showing 4 changed files with 26 additions and 23 deletions.
diff --git a/Directory.Build.props b/Directory.Build.props
@@ -1,7 +1,7 @@
 <Project>
   <PropertyGroup Label="Package information">
     <BaseVersionSuffix></BaseVersionSuffix>
-    <BaseVersion>4.3.1$(BaseVersionSuffix)</BaseVersion>
+    <BaseVersion>4.3.2$(BaseVersionSuffix)</BaseVersion>
     <DatabasePackagesRevision>1</DatabasePackagesRevision>
     <DatabasePackagesVersionSuffix>$(BaseVersionSuffix)</DatabasePackagesVersionSuffix>
   </PropertyGroup>

diff --git a/docs/releases.md b/docs/releases.md
@@ -4,6 +4,12 @@ uid: releases
 
 # Releases
 
+## [4.3.2](https://github.com/BEagle1984/silverback/releases/tag/v4.3.2)
+
+### Fixes
+
+* Fix potential message loss in `OutboxWorker`
+
 ## [4.3.1](https://github.com/BEagle1984/silverback/releases/tag/v4.3.1)
 
 ### Fixes

diff --git a/src/Silverback.Integration/Messaging/Outbound/TransactionalOutbox/OutboxWorker.cs b/src/Silverback.Integration/Messaging/Outbound/TransactionalOutbox/OutboxWorker.cs
@@ -130,7 +130,7 @@ protected virtual Task ProduceMessageAsync(
 
         private static async Task AcknowledgeAllAsync(
             IOutboxReader outboxReader,
-            List<OutboxStoredMessage> messages,
+            IEnumerable<OutboxStoredMessage> messages,
             ConcurrentBag<OutboxStoredMessage> failedMessages)
         {
             await outboxReader.RetryAsync(failedMessages).ConfigureAwait(false);
@@ -157,6 +157,8 @@ private async Task ProcessQueueAsync(
                 return;
             }
 
+            var lastProduced = -1;
+
             try
             {
                 Interlocked.Add(ref _pendingProduceOperations, outboxMessages.Count);
@@ -165,34 +167,28 @@ private async Task ProcessQueueAsync(
                 {
                     _logger.LogProcessingOutboxStoredMessage(i + 1, outboxMessages.Count);
 
-                    try
-                    {
-                        await ProcessMessageAsync(
-                                outboxMessages[i],
-                                failedMessages,
-                                outboxReader,
-                                serviceProvider)
-                            .ConfigureAwait(false);
-                    }
-                    catch (Exception)
-                    {
-                        // Subtract the produce operations that will never be initiated
-                        Interlocked.Add(ref _pendingProduceOperations, -(outboxMessages.Count - i - 1));
-                        throw;
-                    }
+                    lastProduced = i;
+                    await ProcessMessageAsync(
+                            outboxMessages[i],
+                            failedMessages,
+                            outboxReader,
+                            serviceProvider)
+                        .ConfigureAwait(false);
 
                     if (stoppingToken.IsCancellationRequested)
-                    {
-                        // Subtract the produce operations that will never be initiated
-                        Interlocked.Add(ref _pendingProduceOperations, -(outboxMessages.Count - i - 1));
                         break;
-                    }
                 }
             }
             finally
             {
+                if (lastProduced < outboxMessages.Count - 1)
+                {
+                    // Subtract the produce operations that will never be initiated
+                    Interlocked.Add(ref _pendingProduceOperations, -(outboxMessages.Count - lastProduced - 1));
+                }
+
                 await WaitAllAsync().ConfigureAwait(false);
-                await AcknowledgeAllAsync(outboxReader, outboxMessages, failedMessages).ConfigureAwait(false);
+                await AcknowledgeAllAsync(outboxReader, outboxMessages.Take(lastProduced + 1), failedMessages).ConfigureAwait(false);
             }
         }
 

diff --git a/.../Silverback.Integration.Tests/Messaging/Outbound/TransactionalOutbox/OutboxWorkerTests.cs b/.../Silverback.Integration.Tests/Messaging/Outbound/TransactionalOutbox/OutboxWorkerTests.cs
@@ -213,10 +213,11 @@ await _outboxWriter.WriteAsync(
 
             await _worker.ProcessQueueAsync(CancellationToken.None);
 
-            _broker.ProducedMessages.Should().HaveCount(3);
+            _broker.ProducedMessages.Should().HaveCount(4);
             _broker.ProducedMessages[0].Endpoint.Name.Should().Be("topic1");
             _broker.ProducedMessages[1].Endpoint.Name.Should().Be("topic2");
             _broker.ProducedMessages[2].Endpoint.Name.Should().Be("topic1");
+            _broker.ProducedMessages[3].Endpoint.Name.Should().Be("topic1");
         }
     }
 }