Optimize TSQL outbox cleanup by reducing batch size and adding rowlock hint

src/SqlPersistence.Tests/ApprovalFiles/OutboxCommandTests.Cleanup.MsSql.approved.txt

@@ -1,4 +1,4 @@
 
-delete top (@BatchSize) from [TheSchema].[TheTablePrefixOutboxData]
+delete top (@BatchSize) from [TheSchema].[TheTablePrefixOutboxData] with (rowlock)
 where Dispatched = 'true' and
       DispatchedAt < @DispatchedBefore

src/SqlPersistence/Outbox/OutboxPersister.cs

@@ -14,19 +14,16 @@ class OutboxPersister : IOutboxStorage
 {
     readonly IConnectionManager connectionManager;
     readonly SqlDialect sqlDialect;
-    readonly int cleanupBatchSize;
     readonly OutboxCommands outboxCommands;
     readonly Func<ISqlOutboxTransaction> outboxTransactionFactory;
 
     public OutboxPersister(IConnectionManager connectionManager, SqlDialect sqlDialect, OutboxCommands outboxCommands,
-        Func<ISqlOutboxTransaction> outboxTransactionFactory,
-        int cleanupBatchSize = 10000)
+        Func<ISqlOutboxTransaction> outboxTransactionFactory) 
     {
         this.connectionManager = connectionManager;
         this.sqlDialect = sqlDialect;
         this.outboxCommands = outboxCommands;
         this.outboxTransactionFactory = outboxTransactionFactory;
-        this.cleanupBatchSize = cleanupBatchSize;
     }
 
     public Task<IOutboxTransaction> BeginTransaction(ContextBag context, CancellationToken cancellationToken = default)
@@ -126,6 +123,8 @@ public Task Store(OutboxMessage message, IOutboxTransaction outboxTransaction, C
 
     public async Task RemoveEntriesOlderThan(DateTime dateTime, CancellationToken cancellationToken = default)
     {
+        const int CleanupBatchSize = 4000; // Keep below 4000 to prevent lock escalation
+
         using (var connection = await connectionManager.OpenNonContextualConnection(cancellationToken).ConfigureAwait(false))
         {
             var continuePurging = true;
@@ -137,7 +136,7 @@ public async Task RemoveEntriesOlderThan(DateTime dateTime, CancellationToken ca
                 {
                     command.CommandText = outboxCommands.Cleanup;
                     command.AddParameter("DispatchedBefore", dateTime);
-                    command.AddParameter("BatchSize", cleanupBatchSize);
+                    command.AddParameter("BatchSize", CleanupBatchSize);
                     var rowCount = await command.ExecuteNonQueryEx(cancellationToken).ConfigureAwait(false);
                     continuePurging = rowCount != 0;
                 }

src/SqlPersistence/Outbox/SqlDialect_MsSqlServer.cs

@@ -75,8 +75,9 @@ internal override string GetOutboxPessimisticCompleteCommand(string tableName)
 
             internal override string GetOutboxCleanupCommand(string tableName)
             {
+                // Rowlock hint to prevent lock escalation which can result in INSERT's and competing cleanup to dead-lock
                 return $@"
-delete top (@BatchSize) from {tableName}
+delete top (@BatchSize) from {tableName} with (rowlock)
 where Dispatched = 'true' and
       DispatchedAt < @DispatchedBefore";
             }