fix: prevent deadlock in OutboxWorker

BEagle1984 · BEagle1984 · commit 37f06aea1c83 · 2023-05-18T21:29:14.000+02:00
diff --git a/Directory.Build.props b/Directory.Build.props
@@ -1,7 +1,7 @@
 <Project>
   <PropertyGroup Label="Package information">
     <BaseVersionSuffix></BaseVersionSuffix>
-    <BaseVersion>4.3.0$(BaseVersionSuffix)</BaseVersion>
+    <BaseVersion>4.3.1$(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.1](https://github.com/BEagle1984/silverback/releases/tag/v4.3.1)
+
+### Fixes
+
+* Fix deadlock in `OutboxWorker` when `enforceMessageOrder=true` (default)
+
 ## [4.3.0](https://github.com/BEagle1984/silverback/releases/tag/v4.3.0)
 
 ### What's new
diff --git a/src/Silverback.Integration/Messaging/Outbound/TransactionalOutbox/OutboxWorker.cs b/src/Silverback.Integration/Messaging/Outbound/TransactionalOutbox/OutboxWorker.cs
@@ -165,15 +165,28 @@ private async Task ProcessQueueAsync(
                 {
                     _logger.LogProcessingOutboxStoredMessage(i + 1, outboxMessages.Count);
 
-                    await ProcessMessageAsync(
-                            outboxMessages[i],
-                            failedMessages,
-                            outboxReader,
-                            serviceProvider)
-                        .ConfigureAwait(false);
+                    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;
+                    }
 
                     if (stoppingToken.IsCancellationRequested)
+                    {
+                        // Subtract the produce operations that will never be initiated
+                        Interlocked.Add(ref _pendingProduceOperations, -(outboxMessages.Count - i - 1));
                         break;
+                    }
                 }
             }
             finally
@@ -247,8 +260,7 @@ private IProducerEndpoint GetTargetEndpoint(
 
             var targetEndpoint = outboundRoutes
                 .SelectMany(route => route.GetOutboundRouter(serviceProvider).Endpoints)
-                .FirstOrDefault(
-                    endpoint => endpoint.Name == endpointName || endpoint.DisplayName == endpointName);
+                .FirstOrDefault(endpoint => endpoint.Name == endpointName || endpoint.DisplayName == endpointName);
 
             if (targetEndpoint == null)
             {
diff --git a/src/Silverback.Storage.Memory/Silverback.Storage.Memory.csproj b/src/Silverback.Storage.Memory/Silverback.Storage.Memory.csproj
@@ -0,0 +1,34 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFrameworks>$(DefaultTargetFrameworks)</TargetFrameworks>
+    <RootNamespace>$(RootNamespace)</RootNamespace>
+    <Version>$(BaseVersion)</Version>
+    <Authors>$(Authors)</Authors>
+    <Company>$(Company)</Company>
+    <PackageLicenseExpression>$(License)</PackageLicenseExpression>
+    <Copyright>$(Copyright)</Copyright>
+    <PackageProjectUrl>$(ProjectUrl)</PackageProjectUrl>
+    <RepositoryUrl>$(RepositoryUrl)</RepositoryUrl>
+    <RepositoryType>$(RepositoryType)</RepositoryType>
+    <GeneratePackageOnBuild>${GeneratePackageOnBuild}</GeneratePackageOnBuild>
+    <Description>$(Description) This package contains an implementation of Silverback.Storage that stores the data in memory.</Description>
+    <PackageIconUrl>$(IconUrl)</PackageIconUrl>
+    <PackageTags>$(Tags)</PackageTags>
+    <LangVersion>$(LangVersion)</LangVersion>
+    <EnforceCodeStyleInBuild>true</EnforceCodeStyleInBuild>
+    <PackageId>Silverback.Storage.Memory</PackageId>
+    <Product>Silverback.Storage.Memory</Product>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\Silverback.Core\Silverback.Core.csproj" />
+    <ProjectReference Include="..\Silverback.Integration\Silverback.Integration.csproj" />
+    <ProjectReference Include="..\Silverback.Integration.Kafka\Silverback.Integration.Kafka.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <PackageReference Update="Microsoft.SourceLink.GitHub" Version="1.1.1" />
+  </ItemGroup>
+
+</Project>
diff --git a/src/Silverback.Storage.Sqlite/Silverback.Storage.Sqlite.csproj b/src/Silverback.Storage.Sqlite/Silverback.Storage.Sqlite.csproj
@@ -0,0 +1,36 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFrameworks>$(DefaultTargetFrameworks)</TargetFrameworks>
+    <RootNamespace>$(RootNamespace)</RootNamespace>
+    <Version>$(BaseVersion)</Version>
+    <Authors>$(Authors)</Authors>
+    <Company>$(Company)</Company>
+    <PackageLicenseExpression>$(License)</PackageLicenseExpression>
+    <Copyright>$(Copyright)</Copyright>
+    <PackageProjectUrl>$(ProjectUrl)</PackageProjectUrl>
+    <RepositoryUrl>$(RepositoryUrl)</RepositoryUrl>
+    <RepositoryType>$(RepositoryType)</RepositoryType>
+    <GeneratePackageOnBuild>${GeneratePackageOnBuild}</GeneratePackageOnBuild>
+    <Description>$(Description) This package contains an implementation of Silverback.Storage that stores the data in Sqlite.</Description>
+    <PackageIconUrl>$(IconUrl)</PackageIconUrl>
+    <PackageTags>$(Tags)</PackageTags>
+    <LangVersion>$(LangVersion)</LangVersion>
+    <EnforceCodeStyleInBuild>true</EnforceCodeStyleInBuild>
+    <PackageId>Silverback.Storage.Sqlite</PackageId>
+    <Product>Silverback.Storage.Sqlite</Product>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\Silverback.Core\Silverback.Core.csproj" />
+    <ProjectReference Include="..\Silverback.Integration\Silverback.Integration.csproj" />
+    <ProjectReference Include="..\Silverback.Storage.Memory\Silverback.Storage.Memory.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Data.Sqlite" Version="7.0.1" />
+    <PackageReference Include="System.Linq.Async" Version="6.0.1" />
+    <PackageReference Update="Microsoft.SourceLink.GitHub" Version="1.1.1" />
+  </ItemGroup>
+
+</Project>
diff --git a/tests/Silverback.Integration.Tests/Messaging/Outbound/TransactionalOutbox/OutboxWorkerTests.cs b/tests/Silverback.Integration.Tests/Messaging/Outbound/TransactionalOutbox/OutboxWorkerTests.cs
@@ -167,5 +167,56 @@ await _outboxWriter.WriteAsync(
 
             _broker.ProducedMessages.Should().HaveCount(3);
         }
+
+        [Fact]
+        public async Task ProcessQueue_ProduceError_Retried()
+        {
+            await _outboxWriter.WriteAsync(
+                new TestEventOne { Content = "Test" },
+                null,
+                null,
+                "topic1",
+                "topic1");
+            await _outboxWriter.WriteAsync(
+                new TestEventTwo { Content = "Test" },
+                null,
+                null,
+                "topic2",
+                "topic2");
+            await _outboxWriter.WriteAsync(
+                new TestEventOne { Content = "Test" },
+                null,
+                null,
+                "topic1",
+                "topic1");
+            await _outboxWriter.WriteAsync(
+                new TestEventOne { Content = "Test" },
+                null,
+                null,
+                "topic1",
+                "topic1");
+            await _outboxWriter.CommitAsync();
+
+            _broker.FailProduceNumber = new[] { 2, 3 }; // Note: counter is per producer / topic
+
+            await _worker.ProcessQueueAsync(CancellationToken.None);
+
+            _broker.ProducedMessages.Should().HaveCount(2);
+            _broker.ProducedMessages[0].Endpoint.Name.Should().Be("topic1");
+            _broker.ProducedMessages[1].Endpoint.Name.Should().Be("topic2");
+
+            await _worker.ProcessQueueAsync(CancellationToken.None);
+
+            _broker.ProducedMessages.Should().HaveCount(2);
+            _broker.ProducedMessages[0].Endpoint.Name.Should().Be("topic1");
+            _broker.ProducedMessages[1].Endpoint.Name.Should().Be("topic2");
+
+            await _worker.ProcessQueueAsync(CancellationToken.None);
+
+            _broker.ProducedMessages.Should().HaveCount(3);
+            _broker.ProducedMessages[0].Endpoint.Name.Should().Be("topic1");
+            _broker.ProducedMessages[1].Endpoint.Name.Should().Be("topic2");
+            _broker.ProducedMessages[2].Endpoint.Name.Should().Be("topic1");
+        }
     }
 }
diff --git a/tests/Silverback.Integration.Tests/TestTypes/TestBroker.cs b/tests/Silverback.Integration.Tests/TestTypes/TestBroker.cs
@@ -22,6 +22,8 @@ public TestBroker(IServiceProvider serviceProvider)
 
         public bool SimulateConnectIssues { get; set; }
 
+        public IReadOnlyCollection<int>? FailProduceNumber { get; set; }
+
         protected override Task ConnectAsync(
             IReadOnlyCollection<IProducer> producers,
             IReadOnlyCollection<IConsumer> consumers)
diff --git a/tests/Silverback.Integration.Tests/TestTypes/TestProducer.cs b/tests/Silverback.Integration.Tests/TestTypes/TestProducer.cs
@@ -4,6 +4,7 @@
 using System;
 using System.Collections.Generic;
 using System.IO;
+using System.Linq;
 using System.Threading.Tasks;
 using NSubstitute;
 using Silverback.Diagnostics;
@@ -17,6 +18,8 @@ namespace Silverback.Tests.Integration.TestTypes
 {
     public class TestProducer : Producer<TestBroker, TestProducerEndpoint>
     {
+        private int _produceCount;
+
         public TestProducer(
             TestBroker broker,
             TestProducerEndpoint endpoint,
@@ -51,7 +54,7 @@ public TestProducer(
             IReadOnlyCollection<MessageHeader>? headers,
             string actualEndpointName)
         {
-            ProducedMessages.Add(new ProducedMessage(messageBytes, headers, Endpoint));
+            PerformMockProduce(messageBytes, headers);
             return null;
         }
 
@@ -80,7 +83,7 @@ protected override void ProduceCore(
             Action<IBrokerMessageIdentifier?> onSuccess,
             Action<Exception> onError)
         {
-            ProducedMessages.Add(new ProducedMessage(messageBytes, headers, Endpoint));
+            PerformMockProduce(messageBytes, headers);
             onSuccess.Invoke(null);
         }
 
@@ -101,7 +104,7 @@ await messageStream.ReadAllAsync().ConfigureAwait(false),
             IReadOnlyCollection<MessageHeader>? headers,
             string actualEndpointName)
         {
-            ProducedMessages.Add(new ProducedMessage(messageBytes, headers, Endpoint));
+            PerformMockProduce(messageBytes, headers);
             return Task.FromResult<IBrokerMessageIdentifier?>(null);
         }
 
@@ -128,9 +131,17 @@ protected override Task ProduceCoreAsync(
             Action<IBrokerMessageIdentifier?> onSuccess,
             Action<Exception> onError)
         {
-            ProducedMessages.Add(new ProducedMessage(messageBytes, headers, Endpoint));
+            PerformMockProduce(messageBytes, headers);
             onSuccess.Invoke(null);
             return Task.CompletedTask;
         }
+
+        private void PerformMockProduce(byte[]? messageBytes, IReadOnlyCollection<MessageHeader>? headers)
+        {
+            if (Broker.FailProduceNumber != null && Broker.FailProduceNumber.Contains(++_produceCount))
+                throw new InvalidOperationException("Produce failed (mock).");
+
+            ProducedMessages.Add(new ProducedMessage(messageBytes, headers, Endpoint));
+        }
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -4,6 +4,7 @@`
`4`	`4`	`using System;`
`5`	`5`	`using System.Collections.Generic;`
`6`	`6`	`using System.IO;`
	`7`	`+using System.Linq;`
`7`	`8`	`using System.Threading.Tasks;`
`8`	`9`	`using NSubstitute;`
`9`	`10`	`using Silverback.Diagnostics;`
`@@ -17,6 +18,8 @@ namespace Silverback.Tests.Integration.TestTypes`
`17`	`18`	`{`
`18`	`19`	`public class TestProducer : Producer<TestBroker, TestProducerEndpoint>`
`19`	`20`	`{`
	`21`	`+ private int _produceCount;`
	`22`	`+`
`20`	`23`	`public TestProducer(`
`21`	`24`	`TestBroker broker,`
`22`	`25`	`TestProducerEndpoint endpoint,`
`@@ -51,7 +54,7 @@ public TestProducer(`
`51`	`54`	`IReadOnlyCollection<MessageHeader>? headers,`
`52`	`55`	`string actualEndpointName)`
`53`	`56`	`{`
`54`		`- ProducedMessages.Add(new ProducedMessage(messageBytes, headers, Endpoint));`
	`57`	`+ PerformMockProduce(messageBytes, headers);`
`55`	`58`	`return null;`
`56`	`59`	`}`
`57`	`60`
`@@ -80,7 +83,7 @@ protected override void ProduceCore(`
`80`	`83`	`Action<IBrokerMessageIdentifier?> onSuccess,`
`81`	`84`	`Action<Exception> onError)`
`82`	`85`	`{`
`83`		`- ProducedMessages.Add(new ProducedMessage(messageBytes, headers, Endpoint));`
	`86`	`+ PerformMockProduce(messageBytes, headers);`
`84`	`87`	`onSuccess.Invoke(null);`
`85`	`88`	`}`
`86`	`89`
`@@ -101,7 +104,7 @@ await messageStream.ReadAllAsync().ConfigureAwait(false),`
`101`	`104`	`IReadOnlyCollection<MessageHeader>? headers,`
`102`	`105`	`string actualEndpointName)`
`103`	`106`	`{`
`104`		`- ProducedMessages.Add(new ProducedMessage(messageBytes, headers, Endpoint));`
	`107`	`+ PerformMockProduce(messageBytes, headers);`
`105`	`108`	`return Task.FromResult<IBrokerMessageIdentifier?>(null);`
`106`	`109`	`}`
`107`	`110`
`@@ -128,9 +131,17 @@ protected override Task ProduceCoreAsync(`
`128`	`131`	`Action<IBrokerMessageIdentifier?> onSuccess,`
`129`	`132`	`Action<Exception> onError)`
`130`	`133`	`{`
`131`		`- ProducedMessages.Add(new ProducedMessage(messageBytes, headers, Endpoint));`
	`134`	`+ PerformMockProduce(messageBytes, headers);`
`132`	`135`	`onSuccess.Invoke(null);`
`133`	`136`	`return Task.CompletedTask;`
`134`	`137`	`}`
	`138`	`+`
	`139`	`+ private void PerformMockProduce(byte[]? messageBytes, IReadOnlyCollection<MessageHeader>? headers)`
	`140`	`+ {`
	`141`	`+ if (Broker.FailProduceNumber != null && Broker.FailProduceNumber.Contains(++_produceCount))`
	`142`	`+ throw new InvalidOperationException("Produce failed (mock).");`
	`143`	`+`
	`144`	`+ ProducedMessages.Add(new ProducedMessage(messageBytes, headers, Endpoint));`
	`145`	`+ }`
`135`	`146`	`}`
`136`	`147`	`}`