Record a start time for every retry operation, and keep SQL Server statements under the parameter limit

src/ServiceControl.AcceptanceTests.RavenDB/Recoverability/MessageFailures/When_a_failed_message_is_retried.cs

@@ -16,6 +16,7 @@
     using NServiceBus.Transport;
     using NUnit.Framework;
     using ServiceControl.MessageFailures;
+    using ServiceControl.Persistence;
     using ServiceControl.Recoverability;
 
     class When_a_failed_message_is_retried : AcceptanceTest
@@ -108,6 +109,7 @@ await Define<Context>()
         public async Task Should_remove_UnacknowledgedOperation_when_retrying_individual_messages()
         {
             RetryHistory retryHistory = null;
+            var beforeTheRetryWasAskedFor = DateTime.UtcNow;
 
             await Define<Context>()
                 .WithEndpoint<FailingEndpoint>(b => b.When(async ctx =>
@@ -142,7 +144,13 @@ await Define<Context>()
                 })
                 .Run();
 
-            Assert.That(retryHistory.UnacknowledgedOperations, Is.Empty, "Unucknowledged retry operation not removed");
+            var historic = retryHistory.HistoricOperations.Single(operation => operation.RetryType == RetryType.MultipleMessages);
+            using (Assert.EnterMultipleScope())
+            {
+                Assert.That(retryHistory.UnacknowledgedOperations, Is.Empty, "Unucknowledged retry operation not removed");
+                Assert.That(historic.StartTime, Is.GreaterThanOrEqualTo(beforeTheRetryWasAskedFor), "a retry that never waited used to record 01 Jan 0001 as its start time");
+                Assert.That(historic.StartTime, Is.LessThanOrEqualTo(historic.CompletionTime), "a retry cannot have finished before it started");
+            }
         }
 
         [Test]

src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlDialect.cs

@@ -53,5 +53,6 @@ protected static string ParameterRows(int rowCount, int columnCount)
         return sql.ToString();
     }
 
+    // Not a ceiling like SQL Server's, which PostgreSQL is nowhere near: a fixed chunk keeps the text down to a full shape and a remainder, so the planner can cache both.
     protected const int MaxRowsPerStatement = 50;
 }

src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerDialect.cs

@@ -62,6 +62,12 @@ protected static string ParameterRows(int rowCount, int columnCount)
         return sql.ToString();
     }
 
-    protected static int MaxRowsPerStatement(int columns) => MaxSqlParameters / columns;
+    // sharedParameters is for a statement that also carries values of its own, outside the per-row ones.
+    protected static int MaxRowsPerStatement(int columns, int sharedParameters = 0) =>
+        (MaxSqlParameters - ExecuteSqlOverhead - sharedParameters) / columns;
+
     const int MaxSqlParameters = 2100;
+
+    // The client sends every parameterised command through sp_executesql, which spends two of the 2100 on the statement and the parameter list.
+    const int ExecuteSqlOverhead = 2;
 }

src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerFailedMessageIngestionSqlDialect.cs

@@ -80,7 +80,8 @@ public async Task ResolveRetriedMessages(ServiceControlDbContext dbContext, IRea
     {
         const int resolved = (int)FailedMessageStatus.Resolved;
 
-        var maxRowsPerStatement = MaxRowsPerStatement(2);
+        // @p0 carries "now" for every row in the statement
+        var maxRowsPerStatement = MaxRowsPerStatement(2, sharedParameters: 1);
         foreach (var chunk in rows.Chunk(maxRowsPerStatement))
         {
             await Execute(

src/ServiceControl.Persistence.Tests.PostgreSql/ServiceControl.Persistence.Tests.PostgreSql.csproj

@@ -31,9 +31,6 @@
   <ItemGroup>
     <Compile Include="..\ServiceControl.Persistence.Tests\**\*.cs" LinkBase="Shared" />
 
-    <!-- as features get implemented remove these exclusions -->
-    <Compile Remove="..\ServiceControl.Persistence.Tests\Recoverability\RetryConfirmationProcessorTests.cs" />
-    <Compile Remove="..\ServiceControl.Persistence.Tests\RetryStateTests.cs" />
   </ItemGroup>
 
 </Project>

src/ServiceControl.Persistence.Tests.PostgreSql/app.config

@@ -0,0 +1,8 @@
+<?xml version="1.0" encoding="utf-8" ?>
+<configuration>
+  <appSettings>
+    <!-- Settings has no default for these two and throws without them -->
+    <add key="ServiceControl/ForwardErrorMessages" value="false" />
+    <add key="ServiceControl/ErrorRetentionPeriod" value="10.00:00:00" />
+  </appSettings>
+</configuration>

src/ServiceControl.Persistence.Tests.SqlServer/ServiceControl.Persistence.Tests.SqlServer.csproj

@@ -31,9 +31,6 @@
   <ItemGroup>
     <Compile Include="..\ServiceControl.Persistence.Tests\**\*.cs" LinkBase="Shared" />
 
-    <!-- as features get implemented remove these exclusions -->
-    <Compile Remove="..\ServiceControl.Persistence.Tests\Recoverability\RetryConfirmationProcessorTests.cs" />
-    <Compile Remove="..\ServiceControl.Persistence.Tests\RetryStateTests.cs" />
   </ItemGroup>
 
 </Project>

src/ServiceControl.Persistence.Tests.SqlServer/app.config

@@ -0,0 +1,8 @@
+<?xml version="1.0" encoding="utf-8" ?>
+<configuration>
+  <appSettings>
+    <!-- Settings has no default for these two and throws without them -->
+    <add key="ServiceControl/ForwardErrorMessages" value="false" />
+    <add key="ServiceControl/ErrorRetentionPeriod" value="10.00:00:00" />
+  </appSettings>
+</configuration>

src/ServiceControl.Persistence.Tests/EFCore/FailedMessageIngestionSqlDialectTests.cs

@@ -1,7 +1,9 @@
-namespace ServiceControl.Persistence.Tests;
+namespace ServiceControl.Persistence.Tests;
 
 using System;
+using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.DependencyInjection;
@@ -80,7 +82,129 @@ System.Reflection.PropertyInfo[] MappedProperties()
             .Select(property => property!)];
     }
 
-    async Task Upsert(FailedMessageEntity row)
+    // Every one of these row counts fills at least one whole statement, which is where a chunk
+    // size that lands on the database's parameter ceiling instead of under it gets rejected.
+
+    [Test]
+    public async Task Upserts_more_messages_than_fit_in_one_statement()
+    {
+        var rows = Enumerable.Range(0, 200).Select(_ => MinimalFailedMessage()).ToArray();
+
+        await InDialectTransaction((dialect, dbContext, ct) => dialect.UpsertFailedMessages(dbContext, rows, ct));
+
+        Assert.That(await CountStored(rows.Select(row => row.UniqueMessageId)), Is.EqualTo(rows.Length));
+    }
+
+    [Test]
+    public async Task Inserts_more_groups_than_fit_in_one_statement()
+    {
+        var messages = Enumerable.Range(0, 600).Select(_ => MinimalFailedMessage()).ToArray();
+        await InDialectTransaction((dialect, dbContext, ct) => dialect.UpsertFailedMessages(dbContext, messages, ct));
+
+        var rows = messages.Select(message => new FailedMessageGroupEntity
+        {
+            FailedMessageUniqueId = message.UniqueMessageId,
+            GroupId = Guid.NewGuid().ToString(),
+            Title = "a group",
+            Type = "ExceptionType"
+        }).ToArray();
+
+        await InDialectTransaction((dialect, dbContext, ct) => dialect.InsertGroups(dbContext, rows, ct));
+
+        Assert.That(await CountGroups(rows.Select(row => row.FailedMessageUniqueId)), Is.EqualTo(rows.Length));
+    }
+
+    [Test]
+    public async Task Inserts_more_known_endpoints_than_fit_in_one_statement()
+    {
+        var rows = Enumerable.Range(0, 500).Select(_ => new KnownEndpointEntity
+        {
+            Id = Guid.NewGuid(),
+            Name = "Sales",
+            HostId = Guid.NewGuid(),
+            Host = "SalesHost",
+            Monitored = false
+        }).ToArray();
+
+        await InDialectTransaction((dialect, dbContext, ct) => dialect.InsertMissingKnownEndpoints(dbContext, rows, ct));
+
+        Assert.That(await CountEndpoints(rows.Select(row => row.Id)), Is.EqualTo(rows.Length));
+    }
+
+    [Test]
+    public async Task Resolves_more_retried_messages_than_fit_in_one_statement()
+    {
+        var rows = Enumerable.Range(0, 1100).Select(_ => MinimalFailedMessage()).ToArray();
+        await InDialectTransaction((dialect, dbContext, ct) => dialect.UpsertFailedMessages(dbContext, rows, ct));
+
+        var succeededAt = rows[0].LastAttemptedAt.AddMinutes(1);
+        var retries = rows.Select(row => new ConfirmedRetry(row.UniqueMessageId, succeededAt)).ToArray();
+
+        await InDialectTransaction((dialect, dbContext, ct) => dialect.ResolveRetriedMessages(dbContext, retries, succeededAt, ct));
+
+        Assert.That(await CountResolved(rows.Select(row => row.UniqueMessageId)), Is.EqualTo(rows.Length));
+    }
+
+    static FailedMessageEntity MinimalFailedMessage()
+    {
+        var now = new DateTime(2026, 8, 3, 9, 30, 0, DateTimeKind.Utc);
+
+        return new FailedMessageEntity
+        {
+            UniqueMessageId = Guid.NewGuid(),
+            Status = FailedMessageStatus.Unresolved,
+            StatusChangedAt = now,
+            LastModified = now,
+            NumberOfProcessingAttempts = 1,
+            FirstTimeOfFailure = now,
+            LastTimeOfFailure = now,
+            LastAttemptedAt = now,
+            MessageId = "message-id",
+            HeadersJson = "{}",
+            FailingEndpointAddress = "Sales@MACHINE"
+        };
+    }
+
+    async Task<int> CountStored(IEnumerable<Guid> ids)
+    {
+        var wanted = ids.ToHashSet();
+        using var scope = ServiceProvider.CreateScope();
+        var dbContext = scope.ServiceProvider.GetRequiredService<ServiceControlDbContext>();
+
+        return await dbContext.FailedMessages.CountAsync(row => wanted.Contains(row.UniqueMessageId));
+    }
+
+    async Task<int> CountResolved(IEnumerable<Guid> ids)
+    {
+        var wanted = ids.ToHashSet();
+        using var scope = ServiceProvider.CreateScope();
+        var dbContext = scope.ServiceProvider.GetRequiredService<ServiceControlDbContext>();
+
+        return await dbContext.FailedMessages.CountAsync(row => wanted.Contains(row.UniqueMessageId) && row.Status == FailedMessageStatus.Resolved);
+    }
+
+    async Task<int> CountGroups(IEnumerable<Guid> ids)
+    {
+        var wanted = ids.ToHashSet();
+        using var scope = ServiceProvider.CreateScope();
+        var dbContext = scope.ServiceProvider.GetRequiredService<ServiceControlDbContext>();
+
+        return await dbContext.FailedMessageGroups.CountAsync(row => wanted.Contains(row.FailedMessageUniqueId));
+    }
+
+    async Task<int> CountEndpoints(IEnumerable<Guid> ids)
+    {
+        var wanted = ids.ToHashSet();
+        using var scope = ServiceProvider.CreateScope();
+        var dbContext = scope.ServiceProvider.GetRequiredService<ServiceControlDbContext>();
+
+        return await dbContext.KnownEndpoints.CountAsync(row => wanted.Contains(row.Id));
+    }
+
+    Task Upsert(FailedMessageEntity row) =>
+        InDialectTransaction((dialect, dbContext, ct) => dialect.UpsertFailedMessages(dbContext, [row], ct));
+
+    async Task InDialectTransaction(Func<IFailedMessageIngestionSqlDialect, ServiceControlDbContext, CancellationToken, Task> work)
     {
         using var scope = ServiceProvider.CreateScope();
         var dbContext = scope.ServiceProvider.GetRequiredService<ServiceControlDbContext>();
@@ -92,7 +216,7 @@ await strategy.ExecuteAsync(async () =>
         {
             await using var transaction = await dbContext.Database.BeginTransactionAsync();
 
-            await dialect.UpsertFailedMessages(dbContext, [row], TestContext.CurrentContext.CancellationToken);
+            await work(dialect, dbContext, TestContext.CurrentContext.CancellationToken);
 
             await transaction.CommitAsync();
         });

src/ServiceControl.Persistence.Tests/Recoverability/RetryConfirmationProcessorTests.cs

@@ -3,6 +3,7 @@
     using System;
     using System.Collections.Generic;
     using System.Threading.Tasks;
+    using Contracts.Operations;
     using MessageFailures;
     using NServiceBus.Extensibility;
     using NServiceBus.Transport;
@@ -19,14 +20,21 @@ public async Task Setup()
             var domainEvents = new FakeDomainEvents();
             Processor = new RetryConfirmationProcessor(domainEvents);
 
-            await PersistenceTestsContext.InsertFailedMessages(
-                new FailedMessage
-                {
-                    Id = MessageId,
-                    UniqueMessageId = Guid.NewGuid().ToString(),
-                    Status = FailedMessageStatus.Unresolved
-                }
-            );
+            await SeedFailedMessage(new FailedMessage
+            {
+                UniqueMessageId = MessageId,
+                Status = FailedMessageStatus.Unresolved,
+                ProcessingAttempts =
+                [
+                    new FailedMessage.ProcessingAttempt
+                    {
+                        AttemptedAt = DateTime.UtcNow,
+                        MessageMetadata = [],
+                        FailureDetails = new FailureDetails { AddressOfFailingEndpoint = "TestEndpoint" },
+                        Headers = []
+                    }
+                ]
+            });
 
             var batchId = Guid.NewGuid().ToString();
             await RetryBatchStore.AssignMessagesToBatch(batchId, new[] { MessageId });