Moved the latch signal up the stack, to ensure it signals when the connection setup fails.

xwipeoutx · mauroservienti · commit 4b94d4870949 · 2025-12-09T11:09:04.000+01:00
diff --git a/src/NServiceBus.Transport.Sql.Shared/Receiving/ProcessWithNativeTransaction.cs b/src/NServiceBus.Transport.Sql.Shared/Receiving/ProcessWithNativeTransaction.cs
@@ -26,6 +26,7 @@ public override async Task ProcessMessage(CancellationTokenSource stopBatchCance
         {
             Message message = null;
             var context = new ContextBag();
+            var hasLatchBeenSignalled = false;
 
             try
             {
@@ -41,6 +42,7 @@ public override async Task ProcessMessage(CancellationTokenSource stopBatchCance
                     finally
                     {
                         receiveLatch.Signal();
+                        hasLatchBeenSignalled = true;
                     }
 
                     if (receiveResult == MessageReadResult.NoMessage)
@@ -87,6 +89,13 @@ public override async Task ProcessMessage(CancellationTokenSource stopBatchCance
                 }
                 failureInfoStorage.RecordFailureInfoForMessage(message.TransportId, ex, context);
             }
+            finally
+            {
+                if (!hasLatchBeenSignalled)
+                {
+                    receiveLatch.Signal();
+                }
+            }
         }
 
         async Task<bool> TryProcess(Message message, TransportTransaction transportTransaction, ContextBag context, CancellationToken cancellationToken)
diff --git a/src/NServiceBus.Transport.Sql.Shared/Receiving/ProcessWithNoTransaction.cs b/src/NServiceBus.Transport.Sql.Shared/Receiving/ProcessWithNoTransaction.cs
@@ -21,6 +21,7 @@ public override async Task ProcessMessage(CancellationTokenSource stopBatchCance
         {
             Message message = null;
             var context = new ContextBag();
+            var hasLatchBeenSignalled = false;
 
             using (var connection = await connectionFactory.OpenNewConnection(cancellationToken).ConfigureAwait(false))
             {
@@ -37,6 +38,7 @@ public override async Task ProcessMessage(CancellationTokenSource stopBatchCance
                         finally
                         {
                             receiveLatch.Signal();
+                            hasLatchBeenSignalled = true;
                         }
 
                         if (receiveResult == MessageReadResult.NoMessage)
@@ -87,6 +89,13 @@ await ErrorQueue
                     // Since this is TransactionMode.None, we don't care whether error handling says handled or retry. Message is gone either way.
                     _ = await HandleError(ex, message, transportTransaction, 1, context, cancellationToken).ConfigureAwait(false);
                 }
+                finally
+                {
+                    if (!hasLatchBeenSignalled)
+                    {
+                        receiveLatch.Signal();
+                    }
+                }
             }
         }
 
diff --git a/src/NServiceBus.Transport.Sql.Shared/Receiving/ProcessWithTransactionScope.cs b/src/NServiceBus.Transport.Sql.Shared/Receiving/ProcessWithTransactionScope.cs
@@ -22,6 +22,7 @@ public override async Task ProcessMessage(CancellationTokenSource stopBatchCance
         {
             Message message = null;
             var context = new ContextBag();
+            var hasLatchBeenSignalled = false;
 
             try
             {
@@ -36,6 +37,7 @@ public override async Task ProcessMessage(CancellationTokenSource stopBatchCance
                     finally
                     {
                         receiveLatch.Signal();
+                        hasLatchBeenSignalled = true;
                     }
 
                     if (receiveResult == MessageReadResult.NoMessage)
@@ -79,6 +81,13 @@ public override async Task ProcessMessage(CancellationTokenSource stopBatchCance
                 }
                 failureInfoStorage.RecordFailureInfoForMessage(message.TransportId, ex, context);
             }
+            finally
+            {
+                if (!hasLatchBeenSignalled)
+                {
+                    receiveLatch.Signal();
+                }
+            }
         }
 
         async Task<bool> TryProcess(Message message, TransportTransaction transportTransaction, ContextBag context, CancellationToken cancellationToken)

Original file line number	Diff line number	Diff line change
`@@ -26,6 +26,7 @@ public override async Task ProcessMessage(CancellationTokenSource stopBatchCance`
`26`	`26`	`{`
`27`	`27`	`Message message = null;`
`28`	`28`	`var context = new ContextBag();`
	`29`	`+ var hasLatchBeenSignalled = false;`
`29`	`30`
`30`	`31`	`try`
`31`	`32`	`{`
`@@ -41,6 +42,7 @@ public override async Task ProcessMessage(CancellationTokenSource stopBatchCance`
`41`	`42`	`finally`
`42`	`43`	`{`
`43`	`44`	`receiveLatch.Signal();`
	`45`	`+ hasLatchBeenSignalled = true;`
`44`	`46`	`}`
`45`	`47`
`46`	`48`	`if (receiveResult == MessageReadResult.NoMessage)`
`@@ -87,6 +89,13 @@ public override async Task ProcessMessage(CancellationTokenSource stopBatchCance`
`87`	`89`	`}`
`88`	`90`	`failureInfoStorage.RecordFailureInfoForMessage(message.TransportId, ex, context);`
`89`	`91`	`}`
	`92`	`+ finally`
	`93`	`+ {`
	`94`	`+ if (!hasLatchBeenSignalled)`
	`95`	`+ {`
	`96`	`+ receiveLatch.Signal();`
	`97`	`+ }`
	`98`	`+ }`
`90`	`99`	`}`
`91`	`100`
`92`	`101`	`async Task<bool> TryProcess(Message message, TransportTransaction transportTransaction, ContextBag context, CancellationToken cancellationToken)`
Original file line number	Diff line number	Diff line change
`@@ -21,6 +21,7 @@ public override async Task ProcessMessage(CancellationTokenSource stopBatchCance`
`21`	`21`	`{`
`22`	`22`	`Message message = null;`
`23`	`23`	`var context = new ContextBag();`
	`24`	`+ var hasLatchBeenSignalled = false;`
`24`	`25`
`25`	`26`	`using (var connection = await connectionFactory.OpenNewConnection(cancellationToken).ConfigureAwait(false))`
`26`	`27`	`{`
`@@ -37,6 +38,7 @@ public override async Task ProcessMessage(CancellationTokenSource stopBatchCance`
`37`	`38`	`finally`
`38`	`39`	`{`
`39`	`40`	`receiveLatch.Signal();`
	`41`	`+ hasLatchBeenSignalled = true;`
`40`	`42`	`}`
`41`	`43`
`42`	`44`	`if (receiveResult == MessageReadResult.NoMessage)`
`@@ -87,6 +89,13 @@ await ErrorQueue`
`87`	`89`	`// Since this is TransactionMode.None, we don't care whether error handling says handled or retry. Message is gone either way.`
`88`	`90`	`_ = await HandleError(ex, message, transportTransaction, 1, context, cancellationToken).ConfigureAwait(false);`
`89`	`91`	`}`
	`92`	`+ finally`
	`93`	`+ {`
	`94`	`+ if (!hasLatchBeenSignalled)`
	`95`	`+ {`
	`96`	`+ receiveLatch.Signal();`
	`97`	`+ }`
	`98`	`+ }`
`90`	`99`	`}`
`91`	`100`	`}`
`92`	`101`
Original file line number	Diff line number	Diff line change
`@@ -22,6 +22,7 @@ public override async Task ProcessMessage(CancellationTokenSource stopBatchCance`
`22`	`22`	`{`
`23`	`23`	`Message message = null;`
`24`	`24`	`var context = new ContextBag();`
	`25`	`+ var hasLatchBeenSignalled = false;`
`25`	`26`
`26`	`27`	`try`
`27`	`28`	`{`
`@@ -36,6 +37,7 @@ public override async Task ProcessMessage(CancellationTokenSource stopBatchCance`
`36`	`37`	`finally`
`37`	`38`	`{`
`38`	`39`	`receiveLatch.Signal();`
	`40`	`+ hasLatchBeenSignalled = true;`
`39`	`41`	`}`
`40`	`42`
`41`	`43`	`if (receiveResult == MessageReadResult.NoMessage)`
`@@ -79,6 +81,13 @@ public override async Task ProcessMessage(CancellationTokenSource stopBatchCance`
`79`	`81`	`}`
`80`	`82`	`failureInfoStorage.RecordFailureInfoForMessage(message.TransportId, ex, context);`
`81`	`83`	`}`
	`84`	`+ finally`
	`85`	`+ {`
	`86`	`+ if (!hasLatchBeenSignalled)`
	`87`	`+ {`
	`88`	`+ receiveLatch.Signal();`
	`89`	`+ }`
	`90`	`+ }`
`82`	`91`	`}`
`83`	`92`
`84`	`93`	`async Task<bool> TryProcess(Message message, TransportTransaction transportTransaction, ContextBag context, CancellationToken cancellationToken)`