Skip to content

Refactor and re-enable processing attempt acceptance tests #5771

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
johnsimons merged 1 commit into from
Aug 14, 2026
Merged

Refactor and re-enable processing attempt acceptance tests #5771

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 0 additions & 4 deletions ...erviceControl.AcceptanceTests.PostgreSql/ServiceControl.AcceptanceTests.PostgreSql.csproj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,10 +52,6 @@
<Compile Remove="..\ServiceControl.AcceptanceTests\Recoverability\ExternalIntegration\When_encountered_an_error.cs" />
<Compile Remove="..\ServiceControl.AcceptanceTests\Recoverability\When_edited_message_fails_to_process.cs" />

<!-- EF persistence retains only the latest processing attempt. -->
<Compile Remove="..\ServiceControl.AcceptanceTests\Recoverability\MessageFailures\When_errors_with_same_uniqueid_are_imported.cs" />
<Compile Remove="..\ServiceControl.AcceptanceTests\Recoverability\MessageFailures\When_a_messages_fails_multiple_times.cs" />

<!-- The EF custom-check query does not provide an ETag. Addressed by a separate PR. -->
<Compile Remove="..\ServiceControl.AcceptanceTests\WebApi\When_a_request_is_repeated_with_its_etag.cs" />

Expand Down
4 changes: 0 additions & 4 deletions src/ServiceControl.AcceptanceTests.SqlServer/ServiceControl.AcceptanceTests.SqlServer.csproj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,10 +52,6 @@
<Compile Remove="..\ServiceControl.AcceptanceTests\Recoverability\ExternalIntegration\When_encountered_an_error.cs" />
<Compile Remove="..\ServiceControl.AcceptanceTests\Recoverability\When_edited_message_fails_to_process.cs" />

<!-- EF persistence retains only the latest processing attempt. -->
<Compile Remove="..\ServiceControl.AcceptanceTests\Recoverability\MessageFailures\When_errors_with_same_uniqueid_are_imported.cs" />
<Compile Remove="..\ServiceControl.AcceptanceTests\Recoverability\MessageFailures\When_a_messages_fails_multiple_times.cs" />

<!-- The EF custom-check query does not provide an ETag. Addressed by a separate PR. -->
<Compile Remove="..\ServiceControl.AcceptanceTests\WebApi\When_a_request_is_repeated_with_its_etag.cs" />
</ItemGroup>
Expand Down
39 changes: 16 additions & 23 deletions ...ol.AcceptanceTests/Recoverability/MessageFailures/When_a_messages_fails_multiple_times.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
namespace ServiceControl.AcceptanceTests.Recoverability
namespace ServiceControl.AcceptanceTests.Recoverability
{
using NServiceBus.AcceptanceTesting;
using NServiceBus.Routing;
Expand All @@ -17,15 +17,14 @@
class When_a_messages_fails_multiple_times : AcceptanceTest
{
const int NumberOfFailedAttempts = 20;
const int MaximalNumberOfStoredFailedAttempts = 10;
const string AttemptIdHeaderKey = "testing.failed_attempt_no";
const string AttemptNumberHeaderKey = "testing.failed_attempt_no";

[Test]
public async Task Should_store_only_the_latest_processing_attempts()
public async Task Should_report_the_most_recent_attempt_last()
{
FailedMessage result = null;

var context = await Define<TestContext>()
await Define<TestContext>()
.WithEndpoint<AnEndpoint>()
.Done(async c =>
{
Expand All @@ -34,19 +33,23 @@ public async Task Should_store_only_the_latest_processing_attempts()
return false;
}

result = await this.TryGet<FailedMessage>($"/api/errors/{c.UniqueMessageId}");
result = await this.TryGet<FailedMessage>(
$"/api/errors/{c.UniqueMessageId}",
m => LatestAttemptNumber(m) == NumberOfFailedAttempts.ToString());

var failureTimes = result?.ProcessingAttempts.Select(pa => pa.Headers["NServiceBus.TimeOfFailure"]).ToArray() ?? [];

return failureTimes.SequenceEqual([.. c.LatestFailureTimes]);
return result != null;
})
.Run();

Assert.That(LatestAttemptNumber(result), Is.EqualTo(NumberOfFailedAttempts.ToString()));
}

static string LatestAttemptNumber(FailedMessage message) =>
message.ProcessingAttempts[^1].Headers.GetValueOrDefault(AttemptNumberHeaderKey);

class TestContext : ScenarioContext
{
public string UniqueMessageId { get; set; }
public List<string> LatestFailureTimes { get; set; } = [];
}

class AnEndpoint : EndpointConfigurationBuilder
Expand All @@ -66,35 +69,25 @@ protected override TransportOperations CreateMessage(TestContext context)
var transportOperations = Enumerable.Range(0, NumberOfFailedAttempts)
.Select(i =>
{
var timeOfFailure = DateTimeOffsetHelper.ToWireFormattedString(earliestTimeOfFailure.Add(TimeSpan.FromMinutes(i)));

var headers = new Dictionary<string, string>
{
[Headers.MessageId] = messageId,
[Headers.EnclosedMessageTypes] = typeof(MyMessage).FullName,
["NServiceBus.FailedQ"] = endpointName,
["$.diagnostics.hostid"] = Guid.NewGuid().ToString(),
["NServiceBus.TimeOfFailure"] = timeOfFailure,

[AttemptIdHeaderKey] = (i + 1).ToString()
["NServiceBus.TimeOfFailure"] = DateTimeOffsetHelper.ToWireFormattedString(earliestTimeOfFailure.Add(TimeSpan.FromMinutes(i))),
[AttemptNumberHeaderKey] = (i + 1).ToString()
};

context.LatestFailureTimes.Add(timeOfFailure);

return new TransportOperation(new OutgoingMessage(messageId, headers, Array.Empty<byte>()), new UnicastAddressTag("error"));
})
.ToArray();

context.LatestFailureTimes = context.LatestFailureTimes
.Skip(context.LatestFailureTimes.Count - MaximalNumberOfStoredFailedAttempts)
.Take(MaximalNumberOfStoredFailedAttempts)
.ToList();

return new TransportOperations(transportOperations);
}
}

class MyMessage : ICommand;
}
}
}
}
83 changes: 36 additions & 47 deletions ...ptanceTests/Recoverability/MessageFailures/When_errors_with_same_uniqueid_are_imported.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,8 @@
namespace ServiceControl.AcceptanceTests.Recoverability.MessageFailures
namespace ServiceControl.AcceptanceTests.Recoverability.MessageFailures
{
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using AcceptanceTesting;
using AcceptanceTesting.EndpointTemplates;
Expand All @@ -19,37 +18,32 @@

class When_errors_with_same_uniqueid_are_imported : AcceptanceTest
{
const int NumberOfDuplicates = 10;

[Test]
public async Task The_import_should_deduplicate_on_TimeOfFailure()
{
var criticalErrorExecuted = false;

SetSettings = settings => settings.MaximumConcurrencyLevel = 10;
CustomizeHostBuilder = builder => builder.Services.AddSingleton<CounterEnricher>();
CustomConfiguration = config =>
{
config.DefineCriticalErrorAction((_, _) =>
SetSettings = settings => settings.MaximumConcurrencyLevel = NumberOfDuplicates;
CustomizeHostBuilder = builder => builder.Services.AddSingleton<IEnrichImportedErrorMessages, CounterEnricher>();
CustomConfiguration = config => config.DefineCriticalErrorAction((_, _) =>
{
criticalErrorExecuted = true;
return Task.CompletedTask;
});
};

FailedMessage failure = null;
var context = await Define<MyContext>()
.WithEndpoint<SourceEndpoint>()
.Done(async c =>
{
if (c.UniqueId == null)
if (c.UniqueId == null || c.IngestedCount < NumberOfDuplicates)
{
return false;
}

var result = await this.TryGet<FailedMessage>($"/api/errors/{c.UniqueId}", m =>
{
Console.WriteLine("Processing attempts: " + m.ProcessingAttempts.Count);
return m.ProcessingAttempts.Count == 2;
});
var result = await this.TryGet<FailedMessage>($"/api/errors/{c.UniqueId}");
failure = result;
return criticalErrorExecuted || result;
})
Expand All @@ -64,8 +58,8 @@ public async Task The_import_should_deduplicate_on_TimeOfFailure()
var attempts = failure.ProcessingAttempts;
using (Assert.EnterMultipleScope())
{
Assert.That(attempts, Has.Count.EqualTo(2));
Assert.That(attempts.Select(a => a.AttemptedAt), Is.EquivalentTo(context.FailureTimes));
Assert.That(attempts, Has.Count.EqualTo(1));
Assert.That(attempts[^1].AttemptedAt, Is.EqualTo(context.FailureTime));
}
}

Expand Down Expand Up @@ -94,41 +88,34 @@ protected override TransportOperations CreateMessage(MyContext context)
{
var messageId = Guid.NewGuid().ToString();
context.UniqueId = DeterministicGuid.MakeId(messageId, "Error.SourceEndpoint").ToString();
context.FailureTimes = new[]
{
new DateTime(2020, 09, 05, 13, 20, 00, 0, DateTimeKind.Utc),
new DateTime(2020, 09, 05, 12, 20, 00, 0, DateTimeKind.Utc),
};
context.FailureTime = new DateTime(2020, 09, 05, 13, 20, 00, 0, DateTimeKind.Utc);

return new TransportOperations(GetMessages(context.UniqueId, context.FailureTimes).ToArray());
return new TransportOperations([.. GetMessages(context.UniqueId, context.FailureTime)]);
}

IEnumerable<TransportOperation> GetMessages(string uniqueId, DateTime[] failureTimes)
IEnumerable<TransportOperation> GetMessages(string uniqueId, DateTime failureTime)
{
for (var failureNo = 0; failureNo < failureTimes.Length; failureNo++)
for (var i = 0; i < NumberOfDuplicates; i++)
{
for (var i = 0; i < 5; i++)
var messageId = Guid.NewGuid().ToString();
var headers = new Dictionary<string, string>
{
var messageId = Guid.NewGuid().ToString();
var headers = new Dictionary<string, string>
{
[Headers.MessageId] = messageId,
["ServiceControl.Retry.UniqueMessageId"] = uniqueId,
[Headers.ProcessingEndpoint] = "Error.SourceEndpoint",
["NServiceBus.ExceptionInfo.ExceptionType"] = typeof(Exception).FullName,
["NServiceBus.ExceptionInfo.Message"] = "Bad thing happened",
["NServiceBus.ExceptionInfo.InnerExceptionType"] = "System.Exception",
["NServiceBus.ExceptionInfo.Source"] = "NServiceBus.Core",
["NServiceBus.ExceptionInfo.StackTrace"] = string.Empty,
["NServiceBus.FailedQ"] = "Error.SourceEndpoint",
["NServiceBus.TimeOfFailure"] = DateTimeOffsetHelper.ToWireFormattedString(failureTimes[failureNo]),
["Counter"] = i.ToString()
};

var outgoingMessage = new OutgoingMessage(messageId, headers, new byte[0]);

yield return new TransportOperation(outgoingMessage, new UnicastAddressTag("error"));
}
[Headers.MessageId] = messageId,
["ServiceControl.Retry.UniqueMessageId"] = uniqueId,
[Headers.ProcessingEndpoint] = "Error.SourceEndpoint",
["NServiceBus.ExceptionInfo.ExceptionType"] = typeof(Exception).FullName,
["NServiceBus.ExceptionInfo.Message"] = "Bad thing happened",
["NServiceBus.ExceptionInfo.InnerExceptionType"] = "System.Exception",
["NServiceBus.ExceptionInfo.Source"] = "NServiceBus.Core",
["NServiceBus.ExceptionInfo.StackTrace"] = string.Empty,
["NServiceBus.FailedQ"] = "Error.SourceEndpoint",
["NServiceBus.TimeOfFailure"] = DateTimeOffsetHelper.ToWireFormattedString(failureTime),
["Counter"] = i.ToString()
};

var outgoingMessage = new OutgoingMessage(messageId, headers, Array.Empty<byte>());

yield return new TransportOperation(outgoingMessage, new UnicastAddressTag("error"));
}
}
}
Expand All @@ -138,11 +125,13 @@ class MyContext : ScenarioContext
{
public string UniqueId { get; set; }

public DateTime[] FailureTimes { get; set; }
public DateTime FailureTime { get; set; }

public int IngestedCount => receivedMessages.Count;

public void OnMessage(string counter) => receivedMessages.AddOrUpdate(counter, true, (id, old) => true);

readonly ConcurrentDictionary<string, bool> receivedMessages = new();
}
}
}
}
46 changes: 46 additions & 0 deletions ...ServiceControl.Persistence.Tests.RavenDB/Recoverability/ProcessingAttemptTrimmingTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,46 @@
namespace ServiceControl.Persistence.Tests.RavenDB.Recoverability;

using System.Linq;
using System.Threading.Tasks;
using NUnit.Framework;

[TestFixture]
class ProcessingAttemptTrimmingTests : RavenPersistenceTestBase
{
const int MaxStoredAttempts = 10;
const int IngestedAttempts = 15;

[Test]
public async Task Only_the_latest_attempts_are_kept()
{
var failure = new IngestedFailure();

for (var i = 0; i < IngestedAttempts; i++)
{
await Ingest(failure.NextAttempt(failure.AttemptedAt.AddMinutes(i)));
}

var message = await FailedMessageQueryStore.GetFailedMessage(failure.UniqueMessageIdString);

using (Assert.EnterMultipleScope())
{
Assert.That(message.ProcessingAttempts, Has.Count.EqualTo(MaxStoredAttempts));
Assert.That(
message.ProcessingAttempts.Select(attempt => attempt.AttemptedAt),
Is.EqualTo(Enumerable
.Range(IngestedAttempts - MaxStoredAttempts, MaxStoredAttempts)
.Select(i => failure.AttemptedAt.AddMinutes(i))));
}
}

async Task Ingest(IngestedFailure failure)
{
await using (var unitOfWork = await UnitOfWorkFactory.StartNew())
{
await unitOfWork.Recoverability.RecordFailedProcessingAttempt(failure.Context, failure.ProcessingAttempt, failure.Groups);
await unitOfWork.Complete(TestContext.CurrentContext.CancellationToken);
}

await CompleteDatabaseOperation();
}
}