From 6b8879ddbc58dd7240b311ee011551ba3dce2f14 Mon Sep 17 00:00:00 2001 From: Warwick Schroeder Date: Wed, 2 Sep 2026 12:54:51 +0800 Subject: [PATCH 1/4] fix: stamp 'started' when the op did not wait. --- .../FailureGroupsRetryControllerAuditTests.cs | 2 +- .../OperationProgressClockTests.cs | 2 +- .../Recoverability/RetryOperationTests.cs | 20 +++++---- .../Recoverability/RetryStartTimeTests.cs | 44 ++++++++++++++++++- .../Recoverability/Retrying/InMemoryRetry.cs | 13 +++++- .../Recoverability/Retrying/RetriesGateway.cs | 14 +++--- .../Retrying/RetryingManager.cs | 6 +-- 7 files changed, 79 insertions(+), 22 deletions(-) diff --git a/src/ServiceControl.UnitTests/Recoverability/FailureGroupsRetryControllerAuditTests.cs b/src/ServiceControl.UnitTests/Recoverability/FailureGroupsRetryControllerAuditTests.cs index 9d55ffba33..1b8914c96f 100644 --- a/src/ServiceControl.UnitTests/Recoverability/FailureGroupsRetryControllerAuditTests.cs +++ b/src/ServiceControl.UnitTests/Recoverability/FailureGroupsRetryControllerAuditTests.cs @@ -42,7 +42,7 @@ public async Task Group_retry_skipped_as_already_in_progress_is_not_audited() var session = new TestableMessageSession(); var audit = new RecordingMessageActionAuditLog(); var retryingManager = new RetryingManager(new FakeDomainEvents(), TestRetryMetrics.Create(), NullLogger.Instance, TimeProvider.System); - await retryingManager.Preparing("group-42", RetryType.FailureGroup, totalNumberOfMessages: 10); + await retryingManager.Preparing("group-42", RetryType.FailureGroup, totalNumberOfMessages: 10, startTime: DateTime.UtcNow); var controller = new FailureGroupsRetryController(session, retryingManager, new StubCurrentUserAccessor(new AuditUser("alice-sub", "Alice")), audit, TimeProvider.System); await controller.ArchiveGroupErrors("group-42"); diff --git a/src/ServiceControl.UnitTests/Recoverability/OperationProgressClockTests.cs b/src/ServiceControl.UnitTests/Recoverability/OperationProgressClockTests.cs index 533a5aa283..3d48aba4ec 100644 --- a/src/ServiceControl.UnitTests/Recoverability/OperationProgressClockTests.cs +++ b/src/ServiceControl.UnitTests/Recoverability/OperationProgressClockTests.cs @@ -84,7 +84,7 @@ public async Task Retry_completion_comes_from_the_injected_clock() { var retry = new InMemoryRetry("abc123", RetryType.FailureGroup, new FakeDomainEvents(), TestRetryMetrics.Create(), NullLogger.Instance, new FixedClock(FixedNow)); - await retry.Prepare(1000); + await retry.Prepare(1000, FixedNow, null); await retry.PrepareBatch(1000); await retry.Forwarding(); await retry.BatchForwarded(1000); diff --git a/src/ServiceControl.UnitTests/Recoverability/RetryOperationTests.cs b/src/ServiceControl.UnitTests/Recoverability/RetryOperationTests.cs index 2e7141842f..2997fbd4fc 100644 --- a/src/ServiceControl.UnitTests/Recoverability/RetryOperationTests.cs +++ b/src/ServiceControl.UnitTests/Recoverability/RetryOperationTests.cs @@ -39,7 +39,7 @@ public void Fail_should_set_failed() public async Task Prepare_should_set_prepare_state() { var summary = new InMemoryRetry("abc123", RetryType.FailureGroup, new FakeDomainEvents(), TestRetryMetrics.Create(), NullLogger.Instance, TimeProvider.System); - await summary.Prepare(1000); + await summary.Prepare(1000, StartedAt, null); using (Assert.EnterMultipleScope()) { Assert.That(summary.RetryState, Is.EqualTo(RetryState.Preparing)); @@ -52,7 +52,7 @@ public async Task Prepare_should_set_prepare_state() public async Task Prepared_batch_should_set_prepare_state() { var summary = new InMemoryRetry("abc123", RetryType.FailureGroup, new FakeDomainEvents(), TestRetryMetrics.Create(), NullLogger.Instance, TimeProvider.System); - await summary.Prepare(1000); + await summary.Prepare(1000, StartedAt, null); await summary.PrepareBatch(1000); using (Assert.EnterMultipleScope()) { @@ -66,7 +66,7 @@ public async Task Prepared_batch_should_set_prepare_state() public async Task Forwarding_should_set_forwarding_state() { var summary = new InMemoryRetry("abc123", RetryType.FailureGroup, new FakeDomainEvents(), TestRetryMetrics.Create(), NullLogger.Instance, TimeProvider.System); - await summary.Prepare(1000); + await summary.Prepare(1000, StartedAt, null); await summary.PrepareBatch(1000); await summary.Forwarding(); @@ -82,7 +82,7 @@ public async Task Forwarding_should_set_forwarding_state() public async Task Batch_forwarded_should_set_forwarding_state() { var summary = new InMemoryRetry("abc123", RetryType.FailureGroup, new FakeDomainEvents(), TestRetryMetrics.Create(), NullLogger.Instance, TimeProvider.System); - await summary.Prepare(1000); + await summary.Prepare(1000, StartedAt, null); await summary.PrepareBatch(1000); await summary.Forwarding(); await summary.BatchForwarded(500); @@ -100,7 +100,7 @@ public async Task Should_raise_domain_events() { var domainEvents = new FakeDomainEvents(); var summary = new InMemoryRetry("abc123", RetryType.FailureGroup, domainEvents, TestRetryMetrics.Create(), NullLogger.Instance, TimeProvider.System); - await summary.Prepare(1000); + await summary.Prepare(1000, StartedAt, null); await summary.PrepareBatch(1000); await summary.Forwarding(); await summary.BatchForwarded(1000); @@ -119,7 +119,7 @@ public async Task Should_raise_domain_events() public async Task Batch_forwarded_all_forwarded_should_set_completed_state() { var summary = new InMemoryRetry("abc123", RetryType.FailureGroup, new FakeDomainEvents(), TestRetryMetrics.Create(), NullLogger.Instance, TimeProvider.System); - await summary.Prepare(1000); + await summary.Prepare(1000, StartedAt, null); await summary.PrepareBatch(1000); await summary.Forwarding(); await summary.BatchForwarded(1000); @@ -137,7 +137,7 @@ public async Task Skip_should_set_update_skipped_messages() { var summary = new InMemoryRetry("abc123", RetryType.FailureGroup, new FakeDomainEvents(), TestRetryMetrics.Create(), NullLogger.Instance, TimeProvider.System); await summary.Wait(DateTime.UtcNow); - await summary.Prepare(2000); + await summary.Prepare(2000, StartedAt, null); await summary.PrepareBatch(1000); await summary.Skip(1000); @@ -153,7 +153,7 @@ public async Task Skip_should_complete_when_all_skipped() { var summary = new InMemoryRetry("abc123", RetryType.FailureGroup, new FakeDomainEvents(), TestRetryMetrics.Create(), NullLogger.Instance, TimeProvider.System); await summary.Wait(DateTime.UtcNow); - await summary.Prepare(1000); + await summary.Prepare(1000, StartedAt, null); await summary.PrepareBatch(1000); await summary.Skip(1000); @@ -169,7 +169,7 @@ public async Task Skip_and_forward_combination_should_complete_when_done() { var summary = new InMemoryRetry("abc123", RetryType.FailureGroup, new FakeDomainEvents(), TestRetryMetrics.Create(), NullLogger.Instance, TimeProvider.System); await summary.Wait(DateTime.UtcNow); - await summary.Prepare(2000); + await summary.Prepare(2000, StartedAt, null); await summary.PrepareBatch(1000); await summary.Skip(1000); await summary.Forwarding(); @@ -182,5 +182,7 @@ public async Task Skip_and_forward_combination_should_complete_when_done() Assert.That(summary.NumberOfMessagesSkipped, Is.EqualTo(1000)); } } + + static readonly DateTime StartedAt = new(2026, 9, 2, 11, 0, 0, DateTimeKind.Utc); } } \ No newline at end of file diff --git a/src/ServiceControl.UnitTests/Recoverability/RetryStartTimeTests.cs b/src/ServiceControl.UnitTests/Recoverability/RetryStartTimeTests.cs index 4d2a3a3179..9fcbe4135f 100644 --- a/src/ServiceControl.UnitTests/Recoverability/RetryStartTimeTests.cs +++ b/src/ServiceControl.UnitTests/Recoverability/RetryStartTimeTests.cs @@ -45,7 +45,7 @@ public async Task A_completed_group_retry_never_finishes_before_it_started() await NewController(new TestableMessageSession(), retryingManager, clock).ArchiveGroupErrors("group-42"); clock.Advance(TimeSpan.FromMinutes(5)); - await retryingManager.Preparing("group-42", RetryType.FailureGroup, totalNumberOfMessages: 1); + await retryingManager.Preparing("group-42", RetryType.FailureGroup, totalNumberOfMessages: 1, clock.GetUtcNow().UtcDateTime); await retryingManager.PreparedBatch("group-42", RetryType.FailureGroup, numberOfMessagesPrepared: 1); await retryingManager.Forwarding("group-42", RetryType.FailureGroup); await retryingManager.ForwardedBatch("group-42", RetryType.FailureGroup, numberOfMessagesForwarded: 1); @@ -58,6 +58,48 @@ public async Task A_completed_group_retry_never_finishes_before_it_started() } } + [Test] + public async Task A_retry_that_never_waited_records_when_it_was_asked_for() + { + var clock = new FakeTimeProvider(ClockStart); + var retryingManager = NewManager(clock); + + await retryingManager.Preparing("selection-1", RetryType.MultipleMessages, totalNumberOfMessages: 1, + clock.GetUtcNow().UtcDateTime, "all messages for endpoint Endpoint1"); + + var operation = retryingManager.GetStatusForRetryOperation("selection-1", RetryType.MultipleMessages); + using (Assert.EnterMultipleScope()) + { + Assert.That(operation.Started, Is.EqualTo(ClockStart.UtcDateTime), "only a group retry goes through Wait, so every other type has to be stamped here"); + Assert.That(operation.Originator, Is.EqualTo("all messages for endpoint Endpoint1"), "without this the screen labels a bulk retry as a selection of individual messages"); + } + } + + [Test] + public async Task A_completed_retry_that_runs_again_records_the_later_start() + { + var clock = new FakeTimeProvider(ClockStart); + var retryingManager = NewManager(clock); + await RunToCompletion(retryingManager, clock.GetUtcNow().UtcDateTime); + + clock.Advance(TimeSpan.FromHours(1)); + await RunToCompletion(retryingManager, clock.GetUtcNow().UtcDateTime); + + Assert.That(retryingManager.GetStatusForRetryOperation("selection-1", RetryType.MultipleMessages).Started, + Is.EqualTo(ClockStart.UtcDateTime.AddHours(1))); + } + + static async Task RunToCompletion(RetryingManager retryingManager, DateTime startTime) + { + await retryingManager.Preparing("selection-1", RetryType.MultipleMessages, totalNumberOfMessages: 1, startTime); + await retryingManager.PreparedBatch("selection-1", RetryType.MultipleMessages, numberOfMessagesPrepared: 1); + await retryingManager.Forwarding("selection-1", RetryType.MultipleMessages); + await retryingManager.ForwardedBatch("selection-1", RetryType.MultipleMessages, numberOfMessagesForwarded: 1); + } + + static RetryingManager NewManager(TimeProvider clock) => + new(new FakeDomainEvents(), TestRetryMetrics.Create(), NullLogger.Instance, clock); + static FailureGroupsRetryController NewController(TestableMessageSession session, RetryingManager retryingManager, TimeProvider clock) => new(session, retryingManager, new StubCurrentUserAccessor(new AuditUser("alice-sub", "Alice")), new RecordingMessageActionAuditLog(), clock); } diff --git a/src/ServiceControl/Recoverability/Retrying/InMemoryRetry.cs b/src/ServiceControl/Recoverability/Retrying/InMemoryRetry.cs index 9aef908c44..e66a9b4a97 100644 --- a/src/ServiceControl/Recoverability/Retrying/InMemoryRetry.cs +++ b/src/ServiceControl/Recoverability/Retrying/InMemoryRetry.cs @@ -70,14 +70,23 @@ public void Fail() Failed = true; } - public Task Prepare(int totalNumberOfMessages, CancellationToken cancellationToken = default) + public Task Prepare(int totalNumberOfMessages, DateTime startTime, string originator, CancellationToken cancellationToken = default) { // A completed operation being prepared again is a new run that never went through Wait. - if (RetryState == RetryState.Completed) + var isNewRun = RetryState == RetryState.Completed; + + if (isNewRun) { operationStartTimestamp = metrics.GetTimestamp(); } + // Only a group retry goes through Wait, which stamps these; every other type gets them here. + if (isNewRun || Started == default) + { + Started = startTime; + Originator = originator; + } + RetryState = RetryState.Preparing; TotalNumberOfMessages = totalNumberOfMessages; NumberOfMessagesForwarded = 0; diff --git a/src/ServiceControl/Recoverability/Retrying/RetriesGateway.cs b/src/ServiceControl/Recoverability/Retrying/RetriesGateway.cs index f749cdd8be..17dcc6b348 100644 --- a/src/ServiceControl/Recoverability/Retrying/RetriesGateway.cs +++ b/src/ServiceControl/Recoverability/Retrying/RetriesGateway.cs @@ -36,8 +36,10 @@ public async Task StartRetryForSingleMessage(string uniqueMessageId, AuditUser? using var preparation = metrics.BeginPreparation(retryType, cancellationToken); - await operationManager.Preparing(requestId, retryType, numberOfMessages, cancellationToken); - await AssignMessagesToBatch(requestId, retryType, new[] { uniqueMessageId }, timeProvider.GetUtcNow().UtcDateTime, cancellationToken, initiatedBy: initiatedBy, operationId: operationId); + var startedAt = timeProvider.GetUtcNow().UtcDateTime; + + await operationManager.Preparing(requestId, retryType, numberOfMessages, startedAt, cancellationToken: cancellationToken); + await AssignMessagesToBatch(requestId, retryType, new[] { uniqueMessageId }, startedAt, cancellationToken, initiatedBy: initiatedBy, operationId: operationId); await operationManager.PreparedBatch(requestId, retryType, numberOfMessages, cancellationToken); preparation.Complete(); @@ -53,8 +55,10 @@ public async Task StartRetryForMessageSelection(string[] uniqueMessageIds, Audit using var preparation = metrics.BeginPreparation(retryType, cancellationToken); - await operationManager.Preparing(requestId, retryType, numberOfMessages, cancellationToken); - await AssignMessagesToBatch(requestId, retryType, uniqueMessageIds, timeProvider.GetUtcNow().UtcDateTime, cancellationToken, initiatedBy: initiatedBy, operationId: operationId); + var startedAt = timeProvider.GetUtcNow().UtcDateTime; + + await operationManager.Preparing(requestId, retryType, numberOfMessages, startedAt, cancellationToken: cancellationToken); + await AssignMessagesToBatch(requestId, retryType, uniqueMessageIds, startedAt, cancellationToken, initiatedBy: initiatedBy, operationId: operationId); await operationManager.PreparedBatch(requestId, retryType, numberOfMessages, cancellationToken); preparation.Complete(); @@ -107,7 +111,7 @@ async Task ProcessRequest(BulkRetryRequest request, CancellationToken cancellati { var numberOfMessagesAdded = 0; - await operationManager.Preparing(request.RequestId, request.RetryType, totalMessages, cancellationToken); + await operationManager.Preparing(request.RequestId, request.RetryType, totalMessages, request.StartTime, request.Originator, cancellationToken); for (var i = 0; i < batches.Count; i++) { diff --git a/src/ServiceControl/Recoverability/Retrying/RetryingManager.cs b/src/ServiceControl/Recoverability/Retrying/RetryingManager.cs index f7697ed2ff..27db806330 100644 --- a/src/ServiceControl/Recoverability/Retrying/RetryingManager.cs +++ b/src/ServiceControl/Recoverability/Retrying/RetryingManager.cs @@ -44,18 +44,18 @@ public bool IsRetryInProgressFor(string requestId) return retryOperations.Values.Any(o => o.RequestId == requestId && o.IsInProgress()); } - public async Task Preparing(string requestId, RetryType retryType, int totalNumberOfMessages, CancellationToken cancellationToken = default) + public async Task Preparing(string requestId, RetryType retryType, int totalNumberOfMessages, DateTime startTime, string originator = null, CancellationToken cancellationToken = default) { var summary = GetOrCreate(retryType, requestId); - await summary.Prepare(totalNumberOfMessages, cancellationToken); + await summary.Prepare(totalNumberOfMessages, startTime, originator, cancellationToken); } public async Task PreparedAdoptedBatch(string requestId, RetryType retryType, int numberOfMessagesPrepared, int totalNumberOfMessages, string originator, string classifier, DateTime startTime, DateTime last, CancellationToken cancellationToken = default) { var summary = GetOrCreate(retryType, requestId); - await summary.Prepare(totalNumberOfMessages, cancellationToken); + await summary.Prepare(totalNumberOfMessages, startTime, originator, cancellationToken); await summary.PrepareAdoptedBatch(numberOfMessagesPrepared, originator, classifier, startTime, last, cancellationToken); } From b6e4ff4706bb131e0129b033185b6b67915e8371 Mon Sep 17 00:00:00 2001 From: Warwick Schroeder Date: Wed, 2 Sep 2026 15:02:22 +0800 Subject: [PATCH 2/4] - feat: enhance retry operations to maintain accurate start times and titles. - add remaining database agnostic tests. --- .../When_a_failed_message_is_retried.cs | 10 +- .../SqlServerDialect.cs | 6 +- ...ontrol.Persistence.Tests.PostgreSql.csproj | 3 - .../app.config | 8 ++ ...Control.Persistence.Tests.SqlServer.csproj | 3 - .../app.config | 8 ++ .../AppSettingsFixture.cs | 0 .../RetryConfirmationProcessorTests.cs | 24 +++-- .../RetryStateTests.cs | 96 ++++++++++++++----- .../Recoverability/RetryOperationTests.cs | 44 ++++++++- .../Recoverability/RetryStartTimeTests.cs | 19 ++++ .../Recoverability/Retrying/InMemoryRetry.cs | 3 + 12 files changed, 184 insertions(+), 40 deletions(-) create mode 100644 src/ServiceControl.Persistence.Tests.PostgreSql/app.config create mode 100644 src/ServiceControl.Persistence.Tests.SqlServer/app.config rename src/{ServiceControl.Persistence.Tests.RavenDB => ServiceControl.Persistence.Tests}/AppSettingsFixture.cs (100%) diff --git a/src/ServiceControl.AcceptanceTests.RavenDB/Recoverability/MessageFailures/When_a_failed_message_is_retried.cs b/src/ServiceControl.AcceptanceTests.RavenDB/Recoverability/MessageFailures/When_a_failed_message_is_retried.cs index 95a133bd5b..94c8aba9e3 100644 --- a/src/ServiceControl.AcceptanceTests.RavenDB/Recoverability/MessageFailures/When_a_failed_message_is_retried.cs +++ b/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() public async Task Should_remove_UnacknowledgedOperation_when_retrying_individual_messages() { RetryHistory retryHistory = null; + var beforeTheRetryWasAskedFor = DateTime.UtcNow; await Define() .WithEndpoint(b => b.When(async ctx => @@ -142,7 +144,13 @@ await Define() }) .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] diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerDialect.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerDialect.cs index a053469147..2101d84db8 100644 --- a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerDialect.cs +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerDialect.cs @@ -62,6 +62,10 @@ protected static string ParameterRows(int rowCount, int columnCount) return sql.ToString(); } - protected static int MaxRowsPerStatement(int columns) => MaxSqlParameters / columns; + protected static int MaxRowsPerStatement(int columns) => (MaxSqlParameters - ExecuteSqlOverhead) / 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; } diff --git a/src/ServiceControl.Persistence.Tests.PostgreSql/ServiceControl.Persistence.Tests.PostgreSql.csproj b/src/ServiceControl.Persistence.Tests.PostgreSql/ServiceControl.Persistence.Tests.PostgreSql.csproj index 5401253c40..184b878a82 100644 --- a/src/ServiceControl.Persistence.Tests.PostgreSql/ServiceControl.Persistence.Tests.PostgreSql.csproj +++ b/src/ServiceControl.Persistence.Tests.PostgreSql/ServiceControl.Persistence.Tests.PostgreSql.csproj @@ -31,9 +31,6 @@ - - - diff --git a/src/ServiceControl.Persistence.Tests.PostgreSql/app.config b/src/ServiceControl.Persistence.Tests.PostgreSql/app.config new file mode 100644 index 0000000000..4c1ee5c4a5 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests.PostgreSql/app.config @@ -0,0 +1,8 @@ + + + + + + + + diff --git a/src/ServiceControl.Persistence.Tests.SqlServer/ServiceControl.Persistence.Tests.SqlServer.csproj b/src/ServiceControl.Persistence.Tests.SqlServer/ServiceControl.Persistence.Tests.SqlServer.csproj index 7fd542c6f8..d9c54a88e9 100644 --- a/src/ServiceControl.Persistence.Tests.SqlServer/ServiceControl.Persistence.Tests.SqlServer.csproj +++ b/src/ServiceControl.Persistence.Tests.SqlServer/ServiceControl.Persistence.Tests.SqlServer.csproj @@ -31,9 +31,6 @@ - - - diff --git a/src/ServiceControl.Persistence.Tests.SqlServer/app.config b/src/ServiceControl.Persistence.Tests.SqlServer/app.config new file mode 100644 index 0000000000..4c1ee5c4a5 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests.SqlServer/app.config @@ -0,0 +1,8 @@ + + + + + + + + diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/AppSettingsFixture.cs b/src/ServiceControl.Persistence.Tests/AppSettingsFixture.cs similarity index 100% rename from src/ServiceControl.Persistence.Tests.RavenDB/AppSettingsFixture.cs rename to src/ServiceControl.Persistence.Tests/AppSettingsFixture.cs diff --git a/src/ServiceControl.Persistence.Tests/Recoverability/RetryConfirmationProcessorTests.cs b/src/ServiceControl.Persistence.Tests/Recoverability/RetryConfirmationProcessorTests.cs index 62bf5eda6f..f72f6852a8 100644 --- a/src/ServiceControl.Persistence.Tests/Recoverability/RetryConfirmationProcessorTests.cs +++ b/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 }); diff --git a/src/ServiceControl.Persistence.Tests/RetryStateTests.cs b/src/ServiceControl.Persistence.Tests/RetryStateTests.cs index 68df8d8d83..bf1a051fde 100644 --- a/src/ServiceControl.Persistence.Tests/RetryStateTests.cs +++ b/src/ServiceControl.Persistence.Tests/RetryStateTests.cs @@ -8,6 +8,7 @@ using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging.Abstractions; + using Microsoft.Extensions.Time.Testing; using NServiceBus.Transport; using NUnit.Framework; using ServiceBus.Management.Infrastructure.Settings; @@ -38,6 +39,45 @@ public async Task When_a_group_is_processed_it_is_set_to_the_Preparing_state() Assert.That(status.RetryState, Is.EqualTo(RetryState.Preparing)); } + [Test] + public async Task When_a_bulk_retry_is_processed_the_operation_records_when_it_was_asked_for() + { + var askedAt = new DateTime(2020, 1, 1, 0, 0, 0, DateTimeKind.Utc); + var retryManager = new RetryingManager(new FakeDomainEvents(), TestRetryMetrics.Create(), NullLogger.Instance, TimeProvider.System); + + await CreateAFailedMessageAndMarkAsPartOfRetryBatch(retryManager, "Test-group", true, null, null, askedAt, Guid.NewGuid().ToString()); + + var operation = retryManager.GetStatusForRetryOperation("Test-group", RetryType.FailureGroup); + using (Assert.EnterMultipleScope()) + { + Assert.That(operation.Started, Is.EqualTo(askedAt), "the bulk route carries the time the operator asked on the request, and used to drop it on the way to the operation"); + Assert.That(operation.Originator, Is.EqualTo("Test-Context"), "without this the history row has nothing to describe the retry with"); + } + } + + [Test] + public async Task When_a_single_message_is_retried_the_operation_and_the_batch_agree_on_when_it_started() + { + var clock = new FakeTimeProvider(new DateTimeOffset(2020, 1, 1, 0, 0, 0, TimeSpan.Zero)); + var retryManager = new RetryingManager(new FakeDomainEvents(), TestRetryMetrics.Create(), NullLogger.Instance, clock); + var messageId = Guid.NewGuid().ToString(); + + await InsertUnresolvedFailedMessages("Test-group", messageId); + + var gateway = new CustomRetriesGateway(true, RetryBatchStore, retryManager, clock); + await gateway.StartRetryForSingleMessage(messageId); + await CompleteDatabaseOperation(); + + var operation = retryManager.GetStatusForRetryOperation(messageId, RetryType.SingleMessage); + var batchGroup = (await RetryBatchStore.GetAvailableBatchGroups()).Single(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(operation.Started, Is.EqualTo(clock.GetUtcNow().UtcDateTime), "a single-message retry used to record 01 Jan 0001 as its start time"); + Assert.That(batchGroup.StartTime, Is.EqualTo(operation.Started), "the batch is what the start time is rebuilt from after a restart, so the two must not drift apart"); + } + } + [Test] public async Task When_a_group_is_prepared_and_SC_is_started_the_group_is_marked_as_failed() { @@ -199,14 +239,17 @@ public async Task When_there_is_one_poison_message_it_is_removed_from_batch_and_ var domainEvents = new FakeDomainEvents(); var retryManager = new RetryingManager(domainEvents, TestRetryMetrics.Create(), NullLogger.Instance, TimeProvider.System); - await CreateAFailedMessageAndMarkAsPartOfRetryBatch(retryManager, "Test-group", true, "A", "B", "C"); + var ids = new[] { Guid.NewGuid().ToString(), Guid.NewGuid().ToString(), Guid.NewGuid().ToString() }; + var poisonRecordId = PersistenceTestsContext.GenerateFailedMessageRecordId(ids[1]); + + await CreateAFailedMessageAndMarkAsPartOfRetryBatch(retryManager, "Test-group", true, ids); var sender = new TestSender { Callback = operation => { - //Always fails staging message B - if (operation.Message.MessageId == "FailedMessages/B") + //Always fails staging the second message + if (operation.Message.MessageId == poisonRecordId) { throw new Exception("Simulated"); } @@ -272,7 +315,7 @@ public async Task When_a_selection_is_staged_each_message_is_audited_as_a_batch( var retryManager = new RetryingManager(domainEvents, TestRetryMetrics.Create(), NullLogger.Instance, TimeProvider.System); var user = new AuditUser("alice-sub", "Alice"); const string operationId = "op-sel"; - var ids = new[] { "A", "B" }; + var ids = new[] { Guid.NewGuid().ToString(), Guid.NewGuid().ToString() }; var messages = ids.Select(id => new FailedMessage { @@ -294,7 +337,7 @@ public async Task When_a_selection_is_staged_each_message_is_audited_as_a_batch( await PersistenceTestsContext.InsertFailedMessages(messages); await CompleteDatabaseOperation(); - var gateway = new CustomRetriesGateway(true, RetryBatchStore, retryManager); + var gateway = new CustomRetriesGateway(true, RetryBatchStore, retryManager, TimeProvider.System); await gateway.StartRetryForMessageSelection(ids, user, operationId); await CompleteDatabaseOperation(); @@ -323,7 +366,9 @@ public async Task When_a_group_is_staged_each_message_is_audited_with_the_initia var user = new AuditUser("alice-sub", "Alice"); const string operationId = "op-abc"; - await CreateAFailedMessageAndMarkAsPartOfRetryBatch(retryManager, "Test-group", true, user, operationId, "A", "B"); + var ids = new[] { Guid.NewGuid().ToString(), Guid.NewGuid().ToString() }; + + await CreateAFailedMessageAndMarkAsPartOfRetryBatch(retryManager, "Test-group", true, user, operationId, ids); var audit = new RecordingMessageActionAuditLog(); var sender = new TestSender(); @@ -333,7 +378,7 @@ public async Task When_a_group_is_staged_each_message_is_audited_with_the_initia await processor.ProcessBatches(); // stage (emits per-message audit) await processor.ProcessBatches(); // forward - Assert.That(audit.Messages.Select(m => m.MessageId), Is.EquivalentTo(new[] { "A", "B" })); + Assert.That(audit.Messages.Select(m => m.MessageId), Is.EquivalentTo(ids)); using (Assert.EnterMultipleScope()) { Assert.That(audit.Messages, Has.All.Matches(m => m.User.Equals(user))); @@ -376,7 +421,26 @@ Task CreateAFailedMessageAndMarkAsPartOfRetryBatch(RetryingManager retryManager, Task CreateAFailedMessageAndMarkAsPartOfRetryBatch(RetryingManager retryManager, string groupId, bool progressToStaged, params string[] messageIds) => CreateAFailedMessageAndMarkAsPartOfRetryBatch(retryManager, groupId, progressToStaged, null, null, messageIds); - async Task CreateAFailedMessageAndMarkAsPartOfRetryBatch(RetryingManager retryManager, string groupId, bool progressToStaged, AuditUser? initiatedBy, string operationId, params string[] messageIds) + Task CreateAFailedMessageAndMarkAsPartOfRetryBatch(RetryingManager retryManager, string groupId, bool progressToStaged, AuditUser? initiatedBy, string operationId, params string[] messageIds) => + CreateAFailedMessageAndMarkAsPartOfRetryBatch(retryManager, groupId, progressToStaged, initiatedBy, operationId, DateTime.UtcNow, messageIds); + + async Task CreateAFailedMessageAndMarkAsPartOfRetryBatch(RetryingManager retryManager, string groupId, bool progressToStaged, AuditUser? initiatedBy, string operationId, DateTime startTime, params string[] messageIds) + { + await InsertUnresolvedFailedMessages(groupId, messageIds); + + var gateway = new CustomRetriesGateway(progressToStaged, RetryBatchStore, retryManager, TimeProvider.System); + + gateway.EnqueueRetryForFailureGroup(new RetriesGateway.RetryForFailureGroup(groupId, "Test-Context", groupType: null, startTime, initiatedBy, operationId)); + + await CompleteDatabaseOperation(); + + await gateway.ProcessNextBulkRetry(); + + // Wait for indexes to catch up + await CompleteDatabaseOperation(); + } + + async Task InsertUnresolvedFailedMessages(string groupId, params string[] messageIds) { var messages = messageIds.Select(id => new FailedMessage { @@ -409,24 +473,12 @@ async Task CreateAFailedMessageAndMarkAsPartOfRetryBatch(RetryingManager retryMa // Needs index FailedMessages_ByGroup // Needs index FailedMessages_UniqueMessageIdAndTimeOfFailures await CompleteDatabaseOperation(); - - var documentManager = new CustomRetryDocumentManager(progressToStaged, RetryBatchStore, retryManager); - var gateway = new CustomRetriesGateway(progressToStaged, RetryBatchStore, retryManager); - - gateway.EnqueueRetryForFailureGroup(new RetriesGateway.RetryForFailureGroup(groupId, "Test-Context", groupType: null, DateTime.UtcNow, initiatedBy, operationId)); - - await CompleteDatabaseOperation(); - - await gateway.ProcessNextBulkRetry(); - - // Wait for indexes to catch up - await CompleteDatabaseOperation(); } class CustomRetriesGateway : RetriesGateway { - public CustomRetriesGateway(bool progressToStaged, IRetryBatchStore store, RetryingManager retryManager) - : base(store, retryManager, TestRetryMetrics.Create(), NullLogger.Instance, TimeProvider.System) + public CustomRetriesGateway(bool progressToStaged, IRetryBatchStore store, RetryingManager retryManager, TimeProvider timeProvider) + : base(store, retryManager, TestRetryMetrics.Create(), NullLogger.Instance, timeProvider) { this.progressToStaged = progressToStaged; } diff --git a/src/ServiceControl.UnitTests/Recoverability/RetryOperationTests.cs b/src/ServiceControl.UnitTests/Recoverability/RetryOperationTests.cs index 2997fbd4fc..d6b4d0c755 100644 --- a/src/ServiceControl.UnitTests/Recoverability/RetryOperationTests.cs +++ b/src/ServiceControl.UnitTests/Recoverability/RetryOperationTests.cs @@ -151,8 +151,9 @@ public async Task Skip_should_set_update_skipped_messages() [Test] public async Task Skip_should_complete_when_all_skipped() { + var waitedAt = DateTime.UtcNow; var summary = new InMemoryRetry("abc123", RetryType.FailureGroup, new FakeDomainEvents(), TestRetryMetrics.Create(), NullLogger.Instance, TimeProvider.System); - await summary.Wait(DateTime.UtcNow); + await summary.Wait(waitedAt); await summary.Prepare(1000, StartedAt, null); await summary.PrepareBatch(1000); await summary.Skip(1000); @@ -161,14 +162,16 @@ public async Task Skip_should_complete_when_all_skipped() { Assert.That(summary.RetryState, Is.EqualTo(RetryState.Completed)); Assert.That(summary.NumberOfMessagesSkipped, Is.EqualTo(1000)); + Assert.That(summary.Started, Is.EqualTo(waitedAt), "Prepare overwriting this would complete the operation before it started"); } } [Test] public async Task Skip_and_forward_combination_should_complete_when_done() { + var waitedAt = DateTime.UtcNow; var summary = new InMemoryRetry("abc123", RetryType.FailureGroup, new FakeDomainEvents(), TestRetryMetrics.Create(), NullLogger.Instance, TimeProvider.System); - await summary.Wait(DateTime.UtcNow); + await summary.Wait(waitedAt); await summary.Prepare(2000, StartedAt, null); await summary.PrepareBatch(1000); await summary.Skip(1000); @@ -180,6 +183,43 @@ public async Task Skip_and_forward_combination_should_complete_when_done() Assert.That(summary.RetryState, Is.EqualTo(RetryState.Completed)); Assert.That(summary.NumberOfMessagesForwarded, Is.EqualTo(1000)); Assert.That(summary.NumberOfMessagesSkipped, Is.EqualTo(1000)); + Assert.That(summary.Started, Is.EqualTo(waitedAt), "Prepare overwriting this would complete the operation before it started"); + } + } + + [Test] + public async Task A_second_run_does_not_inherit_the_first_runs_skipped_messages() + { + var summary = new InMemoryRetry("abc123", RetryType.All, new FakeDomainEvents(), TestRetryMetrics.Create(), NullLogger.Instance, TimeProvider.System); + await summary.Prepare(2, StartedAt, "all messages"); + await summary.PrepareBatch(2); + await summary.Forwarding(); + await summary.Skip(1); + await summary.BatchForwarded(1); + Assert.That(summary.RetryState, Is.EqualTo(RetryState.Completed), "the first run should have completed"); + + await summary.Prepare(2, StartedAt.AddHours(1), "all messages"); + await summary.PrepareBatch(2); + await summary.Forwarding(); + await summary.BatchForwarded(2); + + Assert.That(summary.RetryState, Is.EqualTo(RetryState.Completed), "a skip count left over from the previous run makes the operation miss its own finish line and sit on Forwarding for ever"); + } + + [Test] + public async Task A_second_run_does_not_inherit_the_first_runs_failure() + { + var summary = new InMemoryRetry("abc123", RetryType.All, new FakeDomainEvents(), TestRetryMetrics.Create(), NullLogger.Instance, TimeProvider.System); + await summary.Prepare(1, StartedAt, "all messages"); + summary.Fail(); + await summary.BatchForwarded(1); + + await summary.Prepare(1, StartedAt.AddHours(1), "all messages"); + + using (Assert.EnterMultipleScope()) + { + Assert.That(summary.Failed, Is.False, "a run that has not failed yet would be written to retry history as a failure"); + Assert.That(summary.CompletionTime, Is.Null, "the previous run's completion time makes a running operation look finished"); } } diff --git a/src/ServiceControl.UnitTests/Recoverability/RetryStartTimeTests.cs b/src/ServiceControl.UnitTests/Recoverability/RetryStartTimeTests.cs index 9fcbe4135f..6944f7dea8 100644 --- a/src/ServiceControl.UnitTests/Recoverability/RetryStartTimeTests.cs +++ b/src/ServiceControl.UnitTests/Recoverability/RetryStartTimeTests.cs @@ -75,6 +75,25 @@ await retryingManager.Preparing("selection-1", RetryType.MultipleMessages, total } } + [Test] + public async Task A_group_retry_keeps_the_title_it_waited_with() + { + var clock = new FakeTimeProvider(ClockStart); + var retryingManager = NewManager(clock); + + await retryingManager.Wait("group-42", RetryType.FailureGroup, clock.GetUtcNow().UtcDateTime, "OrderPlaced failures"); + + clock.Advance(TimeSpan.FromMinutes(5)); + await retryingManager.Preparing("group-42", RetryType.FailureGroup, totalNumberOfMessages: 1, clock.GetUtcNow().UtcDateTime); + + var operation = retryingManager.GetStatusForRetryOperation("group-42", RetryType.FailureGroup); + using (Assert.EnterMultipleScope()) + { + Assert.That(operation.Originator, Is.EqualTo("OrderPlaced failures"), "Prepare must leave alone what Wait already stamped, or the group loses its title on the history row"); + Assert.That(operation.Started, Is.EqualTo(ClockStart.UtcDateTime), "Prepare must leave alone what Wait already stamped, or the group reports the wrong start time"); + } + } + [Test] public async Task A_completed_retry_that_runs_again_records_the_later_start() { diff --git a/src/ServiceControl/Recoverability/Retrying/InMemoryRetry.cs b/src/ServiceControl/Recoverability/Retrying/InMemoryRetry.cs index e66a9b4a97..448162ab10 100644 --- a/src/ServiceControl/Recoverability/Retrying/InMemoryRetry.cs +++ b/src/ServiceControl/Recoverability/Retrying/InMemoryRetry.cs @@ -78,6 +78,9 @@ public Task Prepare(int totalNumberOfMessages, DateTime startTime, string origin if (isNewRun) { operationStartTimestamp = metrics.GetTimestamp(); + NumberOfMessagesSkipped = 0; + CompletionTime = null; + Failed = false; } // Only a group retry goes through Wait, which stamps these; every other type gets them here. From b9cd7d9da617adb13b2103556832d2bd6d876b83 Mon Sep 17 00:00:00 2001 From: Warwick Schroeder Date: Wed, 2 Sep 2026 15:12:21 +0800 Subject: [PATCH 3/4] feat: enhance MaxRowsPerStatement to account for shared parameters in SQL execution --- .../SqlServerDialect.cs | 4 +- ...lServerFailedMessageIngestionSqlDialect.cs | 3 +- .../FailedMessageIngestionSqlDialectTests.cs | 130 +++++++++++++++++- 3 files changed, 132 insertions(+), 5 deletions(-) diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerDialect.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerDialect.cs index 2101d84db8..14c88b87bd 100644 --- a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerDialect.cs +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerDialect.cs @@ -62,7 +62,9 @@ protected static string ParameterRows(int rowCount, int columnCount) return sql.ToString(); } - protected static int MaxRowsPerStatement(int columns) => (MaxSqlParameters - ExecuteSqlOverhead) / 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; diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerFailedMessageIngestionSqlDialect.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerFailedMessageIngestionSqlDialect.cs index 89c3c5609a..bf7bd76f4a 100644 --- a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerFailedMessageIngestionSqlDialect.cs +++ b/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( diff --git a/src/ServiceControl.Persistence.Tests/EFCore/FailedMessageIngestionSqlDialectTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/FailedMessageIngestionSqlDialectTests.cs index 62f798c979..5aadca967c 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/FailedMessageIngestionSqlDialectTests.cs +++ b/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 CountStored(IEnumerable ids) + { + var wanted = ids.ToHashSet(); + using var scope = ServiceProvider.CreateScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + + return await dbContext.FailedMessages.CountAsync(row => wanted.Contains(row.UniqueMessageId)); + } + + async Task CountResolved(IEnumerable ids) + { + var wanted = ids.ToHashSet(); + using var scope = ServiceProvider.CreateScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + + return await dbContext.FailedMessages.CountAsync(row => wanted.Contains(row.UniqueMessageId) && row.Status == FailedMessageStatus.Resolved); + } + + async Task CountGroups(IEnumerable ids) + { + var wanted = ids.ToHashSet(); + using var scope = ServiceProvider.CreateScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + + return await dbContext.FailedMessageGroups.CountAsync(row => wanted.Contains(row.FailedMessageUniqueId)); + } + + async Task CountEndpoints(IEnumerable ids) + { + var wanted = ids.ToHashSet(); + using var scope = ServiceProvider.CreateScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + + 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 work) { using var scope = ServiceProvider.CreateScope(); var dbContext = scope.ServiceProvider.GetRequiredService(); @@ -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(); }); From 43359b973c0a97770d723c55e00ed0543695157e Mon Sep 17 00:00:00 2001 From: Warwick Schroeder Date: Wed, 2 Sep 2026 15:28:28 +0800 Subject: [PATCH 4/4] add clarification comment regarding MaxRowsPerStatement behavior for postgresql --- .../PostgreSqlDialect.cs | 1 + 1 file changed, 1 insertion(+) diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlDialect.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlDialect.cs index 385b61ddc5..5e387119d2 100644 --- a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlDialect.cs +++ b/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; }