From d20482e772fab5abd492bbd1d4b1faf5b0d236e3 Mon Sep 17 00:00:00 2001 From: John Simons Date: Wed, 5 Aug 2026 12:48:36 +1000 Subject: [PATCH 1/2] Refactor message body full-text search configuration and tests The `EnableFullTextSearchOnBodies` setting is only honored by the RavenDB persister, which uses it to control whether message bodies are indexed for full-text search. EF Core persisters, by design, always index message bodies. This change removes the redundant `EnableFullTextSearchOnBodies` setting from the EF Core persistence configuration. Corresponding acceptance tests are updated: a new test is added specifically for RavenDB to verify that the setting correctly disables body search, while the previously shared test for disabled body search is removed as it's not applicable to EF Core. --- .../When_body_search_is_disabled.cs | 94 +++++++++++++++++++ ...failed_message_searched_by_body_content.cs | 42 +-------- .../EFPersistenceConfigurationBase.cs | 2 - 3 files changed, 95 insertions(+), 43 deletions(-) create mode 100644 src/ServiceControl.AcceptanceTests.RavenDB/Recoverability/MessageFailures/When_body_search_is_disabled.cs diff --git a/src/ServiceControl.AcceptanceTests.RavenDB/Recoverability/MessageFailures/When_body_search_is_disabled.cs b/src/ServiceControl.AcceptanceTests.RavenDB/Recoverability/MessageFailures/When_body_search_is_disabled.cs new file mode 100644 index 0000000000..42e1d71e37 --- /dev/null +++ b/src/ServiceControl.AcceptanceTests.RavenDB/Recoverability/MessageFailures/When_body_search_is_disabled.cs @@ -0,0 +1,94 @@ +namespace ServiceControl.AcceptanceTests.RavenDB.Recoverability.MessageFailures +{ + using System; + using System.Threading.Tasks; + using AcceptanceTesting; + using AcceptanceTesting.EndpointTemplates; + using CompositeViews.Messages; + using NServiceBus; + using NServiceBus.AcceptanceTesting; + using NServiceBus.AcceptanceTesting.Customization; + using NUnit.Framework; + + // EnableFullTextSearchOnBodies is honoured by the RavenDB persister only: it decides at ingestion + // whether the body is indexed. The EF persisters always index it, so this test cannot be shared. + class When_body_search_is_disabled : AcceptanceTest + { + [Test] + public async Task Should_not_be_found() + { + SetSettings = settings => settings.PersisterSpecificSettings.EnableFullTextSearchOnBodies = false; + + var searchString = "forty-two"; + + var context = await Define() + .WithEndpoint(b => b.When((bus, c) => bus.Send(new MyMessage + { + Something = "Somewhere in the body is the answer to all of the questions. forty-two" + }))) + .WithEndpoint(b => b.DoNotFailOnErrorMessages()) + .Done(async c => + { + if (c.MessageId != null && await this.TryGetMany($"/api/messages/search/{c.MessageId}")) + { + c.MessageIngested = true; + } + + if (!c.MessageIngested) + { + return false; + } + + c.MessageFound = await this.TryGetMany($"/api/messages/search/{searchString}"); + return true; + }) + .Run(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(context.MessageIngested, Is.True); + Assert.That(context.MessageFound, Is.False); + } + } + + public class Sender : EndpointConfigurationBuilder + { + public Sender() => + EndpointSetup(c => + { + var routing = c.ConfigureRouting(); + routing.RouteToEndpoint(typeof(MyMessage), typeof(Receiver)); + }); + } + + public class Receiver : EndpointConfigurationBuilder + { + public Receiver() => + EndpointSetup(c => c.NoRetries()); + + [Handler] + public class MyMessageHandler(MyContext scenarioContext) : IHandleMessages + { + public Task Handle(MyMessage message, IMessageHandlerContext context) + { + scenarioContext.MessageId = context.MessageId; + throw new Exception("Simulated exception"); + } + } + } + + public class MyMessage : ICommand + { + public string Something { get; set; } + } + + public class MyContext : ScenarioContext + { + public string MessageId { get; set; } + + public bool MessageIngested { get; set; } + + public bool MessageFound { get; set; } + } + } +} diff --git a/src/ServiceControl.AcceptanceTests/Recoverability/MessageFailures/When_failed_message_searched_by_body_content.cs b/src/ServiceControl.AcceptanceTests/Recoverability/MessageFailures/When_failed_message_searched_by_body_content.cs index 6ed34157e2..fd8a3dde81 100644 --- a/src/ServiceControl.AcceptanceTests/Recoverability/MessageFailures/When_failed_message_searched_by_body_content.cs +++ b/src/ServiceControl.AcceptanceTests/Recoverability/MessageFailures/When_failed_message_searched_by_body_content.cs @@ -13,11 +13,8 @@ class When_failed_message_searched_by_body_content : AcceptanceTest { [Test] - public async Task Should_be_found_when_fulltext_search_enabled() + public async Task Should_be_found() { - // setting it even if it is the default - SetSettings = settings => settings.PersisterSpecificSettings.EnableFullTextSearchOnBodies = true; - var searchString = "forty-two"; var context = await Define() @@ -45,43 +42,6 @@ public async Task Should_be_found_when_fulltext_search_enabled() Assert.That(context.MessageFound, Is.True); } - [Test] - public async Task Should_not_be_found_when_fulltext_search_disabled() - { - SetSettings = settings => settings.PersisterSpecificSettings.EnableFullTextSearchOnBodies = false; - - var searchString = "forty-two"; - - var context = await Define() - .WithEndpoint(b => b.When((bus, c) => bus.Send(new MyMessage - { - Something = "Somewhere in the body is the answer to all of the questions. forty-two" - }))) - .WithEndpoint(b => b.DoNotFailOnErrorMessages()) - .Done(async c => - { - if (c.MessageId != null && await this.TryGetMany($"/api/messages/search/{c.MessageId}")) - { - c.MessageIngested = true; - } - - if (!c.MessageIngested) - { - return false; - } - - c.MessageFound = await this.TryGetMany($"/api/messages/search/{searchString}"); - return true; - }) - .Run(); - - using (Assert.EnterMultipleScope()) - { - Assert.That(context.MessageIngested, Is.True); - Assert.That(context.MessageFound, Is.False); - } - } - public class Sender : EndpointConfigurationBuilder { public Sender() => diff --git a/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersistenceConfigurationBase.cs b/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersistenceConfigurationBase.cs index 9dda44ea66..16444defb9 100644 --- a/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersistenceConfigurationBase.cs +++ b/src/ServiceControl.Persistence.EFCore/Abstractions/EFPersistenceConfigurationBase.cs @@ -26,7 +26,6 @@ public abstract class EFPersistenceConfigurationBase : PersistenceConfiguration, const string MaxBodySizeToStoreKey = "MaxBodySizeToStore"; const string ErrorRetentionPeriodKey = "ErrorRetentionPeriod"; const string EventsRetentionPeriodKey = "EventsRetentionPeriod"; - const string EnableFullTextSearchOnBodiesKey = "EnableFullTextSearchOnBodies"; const string SubscriptionCacheDurationKey = "SubscriptionCacheDuration"; public PersistenceSettings CreateSettings(SettingsRootNamespace settingsRootNamespace) @@ -38,7 +37,6 @@ public PersistenceSettings CreateSettings(SettingsRootNamespace settingsRootName settings.CommandTimeout = SettingsReader.Read(settingsRootNamespace, CommandTimeoutKey, EFPersisterSettings.DefaultCommandTimeout); settings.ErrorRetentionPeriod = GetRequiredSetting(settingsRootNamespace, ErrorRetentionPeriodKey); settings.EventsRetentionPeriod = SettingsReader.Read(settingsRootNamespace, EventsRetentionPeriodKey, EFPersisterSettings.DefaultEventsRetentionPeriod); - settings.EnableFullTextSearchOnBodies = SettingsReader.Read(settingsRootNamespace, EnableFullTextSearchOnBodiesKey, true); settings.SubscriptionCacheDuration = SettingsReader.Read(settingsRootNamespace, SubscriptionCacheDurationKey, EFPersisterSettings.DefaultSubscriptionCacheDuration); return settings; From 6535642a5e382c4869494507e85d2237f698e141 Mon Sep 17 00:00:00 2001 From: John Simons Date: Fri, 7 Aug 2026 11:04:52 +1000 Subject: [PATCH 2/2] Implement IMessagesViewDataStore on the EF Core error persister Search introduces IFullTextSearchDialect beside IIngestionSqlDialect. The full text indexes already existed but nothing queried them. SQL Server ORs two FREETEXT predicates, PostgreSQL matches the indexed tsvector against websearch_to_tsquery with the terms rejoined by OR, so both keep the OR semantics RavenDB's Search defaults to. PostgreSQL only uses an expression index when the query expression parses to the same tree, and a mismatch is silent: search keeps working, on a sequential scan. The indexed expression is now a constant written the way EF renders it, and FullTextSearchIndexTests fails if the two drift apart. SQL Server setup now fails with a named error when the Full-Text Search feature is missing, rather than migrating into an instance whose search endpoint throws. --- .../FullTextSearchSql.cs | 27 +- .../PostgreSqlFullTextSearchDialect.cs | 27 ++ .../PostgreSqlPersistence.cs | 1 + .../FullTextSearchSql.cs | 21 +- .../20260722061323_AddFullTextSearch.cs | 1 + .../SqlServerFullTextSearchDialect.cs | 16 + .../SqlServerPersistence.cs | 1 + .../Implementation/FailedMessageViewMapper.cs | 6 +- .../Implementation/MessagesViewDataStore.cs | 39 +- .../Implementation/MessagesViewMapper.cs | 59 +++ .../MessagesViewQueryResults.cs | 24 ++ .../FailedMessageQueryFilters.cs | 49 +++ .../Infrastructure/IFullTextSearchDialect.cs | 18 + .../FullTextSearchIndexTests.cs | 51 +++ .../EFCore/MessagesViewDataStoreTests.cs | 364 ++++++++++++++++++ .../IngestedFailure.cs | 5 +- 16 files changed, 679 insertions(+), 30 deletions(-) create mode 100644 src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlFullTextSearchDialect.cs create mode 100644 src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerFullTextSearchDialect.cs create mode 100644 src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewMapper.cs create mode 100644 src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewQueryResults.cs create mode 100644 src/ServiceControl.Persistence.EFCore/Infrastructure/IFullTextSearchDialect.cs create mode 100644 src/ServiceControl.Persistence.Tests.PostgreSql/FullTextSearchIndexTests.cs create mode 100644 src/ServiceControl.Persistence.Tests/EFCore/MessagesViewDataStoreTests.cs diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/FullTextSearchSql.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/FullTextSearchSql.cs index 2266b6e087..f8d25c91d4 100644 --- a/src/ServiceControl.Persistence.EFCore.PostgreSql/FullTextSearchSql.cs +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/FullTextSearchSql.cs @@ -12,18 +12,21 @@ static class FullTextSearchSql // 'simple' rather than 'english': message and header content is technical, stemming and // stopword removal do more harm than good. - // The default parser reads a dotted name as a single host token, so - // "ServiceControl.MessageFailures.MyMessage" would not match a search for "MyMessage". The - // message type is therefore also indexed with its separators replaced by spaces, mirroring - // the SearchableMessageType that MessageTypeEnricher already produces for RavenDB. - public const string Up = $""" - CREATE INDEX {IndexName} - ON failed_messages - USING GIN (to_tsvector('simple', - coalesce(headers_json, '') || ' ' || - coalesce(body_text, '') || ' ' || - replace(replace(coalesce(message_type, ''), '.', ' '), '+', ' '))) - """; + public const string Configuration = "simple"; + + // Written the way PostgreSqlFullTextSearchDialect makes EF Core render it, down to the casing + // and the redundant looking parentheses: PostgreSQL only uses an expression index when the + // query expression parses to the same tree, and a mismatch downgrades search to a sequential + // scan silently. FullTextSearchIndexTests fails if the two drift apart. + // The message type is indexed a second time with its separators replaced by spaces because the + // default parser reads a dotted name as a single host token, so + // "ServiceControl.MessageFailures.MyMessage" would not otherwise match a search for + // "MyMessage". It mirrors the SearchableMessageType that MessageTypeEnricher produces for + // RavenDB, and is not the duplicate of the headers it looks like. + public const string IndexedExpression = + $"""to_tsvector('{Configuration}', headers_json || ' ' || COALESCE(body_text, '') || ' ' || replace(replace(COALESCE(message_type, ''), '.', ' '), '+', ' '))"""; + + public const string Up = $"CREATE INDEX {IndexName} ON failed_messages USING GIN ({IndexedExpression})"; public const string Down = $"DROP INDEX IF EXISTS {IndexName}"; } diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlFullTextSearchDialect.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlFullTextSearchDialect.cs new file mode 100644 index 0000000000..d2c01fb908 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlFullTextSearchDialect.cs @@ -0,0 +1,27 @@ +namespace ServiceControl.Persistence.EFCore.PostgreSql; + +using Microsoft.EntityFrameworkCore; +using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Persistence.EFCore.Infrastructure; + +class PostgreSqlFullTextSearchDialect : IFullTextSearchDialect +{ + // The tsvector expression has to be the one FullTextSearchSql indexes, character for character, + // or the planner cannot use the GIN index and the search degrades to a sequential scan instead + // of failing. FullTextSearchIndexTests pins the two together. + // The document has to be built by concatenation: an interpolated string compiles to + // string.Format, which EF Core cannot translate, and the query then throws. + public IQueryable Search(IQueryable source, string searchTerms) => + source.Where(message => + EF.Functions.ToTsVector(FullTextSearchSql.Configuration, + message.HeadersJson + " " + + (message.BodyText ?? "") + " " + + (message.MessageType ?? "").Replace(".", " ").Replace("+", " ")) + .Matches(EF.Functions.WebSearchToTsQuery(FullTextSearchSql.Configuration, ToOrQuery(searchTerms)))); + + // websearch_to_tsquery ANDs bare terms; the RavenDB persister ORs them, so the terms are + // rejoined with the operator that syntax understands. It also never throws on odd input, which + // a hand built tsquery would. + static string ToOrQuery(string searchTerms) => + string.Join(" OR ", searchTerms.Split(' ', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries)); +} diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlPersistence.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlPersistence.cs index 9e807fc358..c0901491d5 100644 --- a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlPersistence.cs +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlPersistence.cs @@ -15,6 +15,7 @@ public void AddPersistence(IServiceCollection services) RegisterDataStores(services, settings); services.AddSingleton(); + services.AddSingleton(); } public void AddInstaller(IServiceCollection services) diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/FullTextSearchSql.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/FullTextSearchSql.cs index 8f699723c0..a7140f672f 100644 --- a/src/ServiceControl.Persistence.EFCore.SqlServer/FullTextSearchSql.cs +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/FullTextSearchSql.cs @@ -10,12 +10,20 @@ static class FullTextSearchSql { const string CatalogName = "ServiceControlFullTextCatalog"; - // Both statements are guarded: an instance without Full-Text Search installed still migrates, - // it just has no full text index. They also cannot run inside a transaction, so the migration - // passes suppressTransaction. + // Message search is not optional, so an instance without Full-Text Search installed is not a + // degraded instance, it is a broken one: every /messages/search request would fail on a missing + // index. Failing the migration says so once, at setup, instead of at the first search. + public const string RequireFullTextSearch = """ + IF SERVERPROPERTY('IsFullTextInstalled') <> 1 + BEGIN + THROW 50000, 'ServiceControl requires the SQL Server Full-Text Search feature, which is not installed on this instance. Install it and run setup again.', 1; + END + """; + + // The statements are idempotent so that a re-run is harmless. They also cannot run inside a + // transaction, so the migration passes suppressTransaction. public const string CreateCatalog = $""" - IF SERVERPROPERTY('IsFullTextInstalled') = 1 - AND NOT EXISTS (SELECT 1 FROM sys.fulltext_catalogs WHERE name = '{CatalogName}') + IF NOT EXISTS (SELECT 1 FROM sys.fulltext_catalogs WHERE name = '{CatalogName}') BEGIN EXEC('CREATE FULLTEXT CATALOG {CatalogName}'); END @@ -26,8 +34,7 @@ AND NOT EXISTS (SELECT 1 FROM sys.fulltext_catalogs WHERE name = '{CatalogName}' // The message type needs no dedicated column here: the word breaker splits dotted names, and // the headers already carry the type. public const string CreateIndex = $""" - IF SERVERPROPERTY('IsFullTextInstalled') = 1 - AND NOT EXISTS (SELECT 1 FROM sys.fulltext_indexes WHERE object_id = OBJECT_ID('FailedMessages')) + IF NOT EXISTS (SELECT 1 FROM sys.fulltext_indexes WHERE object_id = OBJECT_ID('FailedMessages')) BEGIN EXEC('CREATE FULLTEXT INDEX ON FailedMessages(HeadersJson LANGUAGE 0, BodyText LANGUAGE 0) KEY INDEX PK_FailedMessages diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260722061323_AddFullTextSearch.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260722061323_AddFullTextSearch.cs index 9393cce434..128ecb5e66 100644 --- a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260722061323_AddFullTextSearch.cs +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260722061323_AddFullTextSearch.cs @@ -10,6 +10,7 @@ public partial class AddFullTextSearch : Migration /// protected override void Up(MigrationBuilder migrationBuilder) { + migrationBuilder.Sql(FullTextSearchSql.RequireFullTextSearch, suppressTransaction: true); migrationBuilder.Sql(FullTextSearchSql.CreateCatalog, suppressTransaction: true); migrationBuilder.Sql(FullTextSearchSql.CreateIndex, suppressTransaction: true); } diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerFullTextSearchDialect.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerFullTextSearchDialect.cs new file mode 100644 index 0000000000..cc323d9f2e --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerFullTextSearchDialect.cs @@ -0,0 +1,16 @@ +namespace ServiceControl.Persistence.EFCore.SqlServer; + +using Microsoft.EntityFrameworkCore; +using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Persistence.EFCore.Infrastructure; + +class SqlServerFullTextSearchDialect : IFullTextSearchDialect +{ + // FREETEXT ORs the terms itself, so the search string needs no parsing here. Both columns are + // covered by the index the AddFullTextSearch migration creates, and a NULL body simply does not + // match. + public IQueryable Search(IQueryable source, string searchTerms) => + source.Where(message => + EF.Functions.FreeText(message.HeadersJson, searchTerms) || + EF.Functions.FreeText(message.BodyText!, searchTerms)); +} diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerPersistence.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerPersistence.cs index 3c6d1384b3..db57f688bf 100644 --- a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerPersistence.cs +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerPersistence.cs @@ -15,6 +15,7 @@ public void AddPersistence(IServiceCollection services) RegisterDataStores(services, settings); services.AddSingleton(); + services.AddSingleton(); } public void AddInstaller(IServiceCollection services) diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageViewMapper.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageViewMapper.cs index 211151c0ee..418f17eba0 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageViewMapper.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageViewMapper.cs @@ -1,13 +1,11 @@ namespace ServiceControl.Persistence.EFCore.Implementation; -using System.Text.Json; using ServiceControl.Contracts.Operations; using ServiceControl.MessageFailures; using ServiceControl.MessageFailures.Api; using ServiceControl.Operations; using ServiceControl.Persistence.EFCore.Entities; using ServiceControl.Persistence.EFCore.Infrastructure; -using ServiceControl.Persistence.EFCore.Implementation.UnitOfWork; static class FailedMessageViewMapper { @@ -128,7 +126,7 @@ static ExceptionDetails ToExceptionDetails(this FailedMessageEntity entity, Dict StackTrace = headers.GetValueOrDefault(ExceptionStackTraceHeader) }; - static EndpointDetails? ToSendingEndpoint(this FailedMessageEntity entity) => + public static EndpointDetails? ToSendingEndpoint(this FailedMessageEntity entity) => entity.SendingEndpointName == null ? null : new EndpointDetails @@ -138,7 +136,7 @@ static ExceptionDetails ToExceptionDetails(this FailedMessageEntity entity, Dict HostId = entity.SendingEndpointHostId ?? Guid.Empty }; - static EndpointDetails? ToReceivingEndpoint(this FailedMessageEntity entity) => + public static EndpointDetails? ToReceivingEndpoint(this FailedMessageEntity entity) => entity.ReceivingEndpointName == null ? null : new EndpointDetails diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewDataStore.cs index c8439b1eca..5cea7bc390 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewDataStore.cs @@ -1,22 +1,49 @@ namespace ServiceControl.Persistence.EFCore.Implementation; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; using ServiceControl.CompositeViews.Messages; +using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Persistence.EFCore.Infrastructure; using ServiceControl.Persistence.Infrastructure; -public class MessagesViewDataStore : IMessagesViewDataStore +public class MessagesViewDataStore(IServiceScopeFactory scopeFactory, IFullTextSearchDialect fullTextSearch) : DataStoreBase(scopeFactory), IMessagesViewDataStore { public Task>> GetAllMessages(PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange? timeSentRange = null) => - throw new NotImplementedException(); + ExecuteWithDbContext(dbContext => dbContext.FailedMessages + .AsNoTracking() + .IncludeSystemMessagesWhere(includeSystemMessages) + .FilterBySentTimeRange(timeSentRange) + .ToPagedMessagesResult(pagingInfo, sortInfo)); public Task>> GetAllMessagesForEndpoint(string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange? timeSentRange = null) => - throw new NotImplementedException(); + ExecuteWithDbContext(dbContext => dbContext.FailedMessages + .AsNoTracking() + .Where(message => message.ReceivingEndpointName == endpointName) + .IncludeSystemMessagesWhere(includeSystemMessages) + .FilterBySentTimeRange(timeSentRange) + .ToPagedMessagesResult(pagingInfo, sortInfo)); + // includeSystemMessages is unused here: a conversation is incomplete without the system messages that took part in it. public Task>> GetAllMessagesByConversation(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages) => - throw new NotImplementedException(); + ExecuteWithDbContext(dbContext => dbContext.FailedMessages + .AsNoTracking() + .Where(message => message.ConversationId == conversationId) + .ToPagedMessagesResult(pagingInfo, sortInfo)); public Task>> GetAllMessagesForSearch(string searchTerms, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null) => - throw new NotImplementedException(); + ExecuteWithDbContext(dbContext => Search(dbContext.FailedMessages.AsNoTracking(), searchTerms) + .FilterBySentTimeRange(timeSentRange) + .ToPagedMessagesResult(pagingInfo, sortInfo)); public Task>> SearchEndpointMessages(string endpointName, string searchKeyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null) => - throw new NotImplementedException(); + ExecuteWithDbContext(dbContext => Search(dbContext.FailedMessages.AsNoTracking(), searchKeyword) + .Where(message => message.ReceivingEndpointName == endpointName) + .FilterBySentTimeRange(timeSentRange) + .ToPagedMessagesResult(pagingInfo, sortInfo)); + + // Neither search hides system messages: a caller who searched + // for something specific is not helped by hiding the message that matched it. + IQueryable Search(IQueryable source, string searchTerms) => + string.IsNullOrWhiteSpace(searchTerms) ? source : fullTextSearch.Search(source, searchTerms); } diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewMapper.cs b/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewMapper.cs new file mode 100644 index 0000000000..0b099929f0 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewMapper.cs @@ -0,0 +1,59 @@ +namespace ServiceControl.Persistence.EFCore.Implementation; + +using NServiceBus; +using ServiceControl.CompositeViews.Messages; +using ServiceControl.MessageFailures; +using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Persistence.EFCore.Infrastructure; + +static class MessagesViewMapper +{ + public static MessagesView ToMessagesView(this FailedMessageEntity entity) + { + var headers = MessageHeaders.Read(entity.HeadersJson); + + return new MessagesView + { + Id = entity.UniqueMessageId.ToString(), + MessageId = entity.MessageId, + MessageType = entity.MessageType, + SendingEndpoint = entity.ToSendingEndpoint(), + ReceivingEndpoint = entity.ToReceivingEndpoint(), + TimeSent = entity.TimeSent, + ProcessedAt = entity.LastAttemptedAt, + // The error instance never enriches the processing statistics: ProcessingStatisticsEnricher + // contributes TimeSent and nothing else. + CriticalTime = TimeSpan.Zero, + ProcessingTime = TimeSpan.Zero, + DeliveryTime = TimeSpan.Zero, + IsSystemMessage = entity.IsSystemMessage, + ConversationId = entity.ConversationId, + Headers = [.. headers.Select(header => new KeyValuePair(header.Key, header.Value))], + Status = entity.ToMessageStatus(), + MessageIntent = ReadMessageIntent(headers), + BodyUrl = $"/messages/{entity.UniqueMessageId}/body", + BodySize = entity.BodySize + }; + } + + public static MessageStatus ToMessageStatus(this FailedMessageEntity entity) => + entity.Status switch + { + FailedMessageStatus.Resolved => MessageStatus.ResolvedSuccessfully, + FailedMessageStatus.RetryIssued => MessageStatus.RetryIssued, + FailedMessageStatus.Archived => MessageStatus.ArchivedFailure, + FailedMessageStatus.Unresolved or _ => entity.NumberOfProcessingAttempts == 1 ? MessageStatus.Failed : MessageStatus.RepeatedFailure + }; + + static MessageIntent ReadMessageIntent(Dictionary headers) + { + var intent = default(MessageIntent); + + if (headers.TryGetValue(Headers.MessageIntent, out var value)) + { + Enum.TryParse(value, true, out intent); + } + + return intent; + } +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewQueryResults.cs b/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewQueryResults.cs new file mode 100644 index 0000000000..0dc362cd97 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewQueryResults.cs @@ -0,0 +1,24 @@ +namespace ServiceControl.Persistence.EFCore.Implementation; + +using Microsoft.EntityFrameworkCore; +using ServiceControl.CompositeViews.Messages; +using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Persistence.EFCore.Infrastructure; +using ServiceControl.Persistence.Infrastructure; + +static class MessagesViewQueryResults +{ + public static async Task>> ToPagedMessagesResult(this IQueryable source, PagingInfo pagingInfo, SortInfo sortInfo) + { + var stats = await source.ToQueryStatsInfo(); + + var entities = await source + .SortMessages(sortInfo) + .Page(pagingInfo) + .ToListAsync(); + + IList results = [.. entities.Select(entity => entity.ToMessagesView())]; + + return new QueryResult>(results, stats); + } +} diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/FailedMessageQueryFilters.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/FailedMessageQueryFilters.cs index 30a53e5470..ebf5c3c294 100644 --- a/src/ServiceControl.Persistence.EFCore/Infrastructure/FailedMessageQueryFilters.cs +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/FailedMessageQueryFilters.cs @@ -113,6 +113,55 @@ public static IQueryable Sort(this IQueryable FilterBySentTimeRange(this IQueryable source, DateTimeRange? timeSentRange) + { + if (timeSentRange?.From is { } from) + { + source = source.Where(message => message.TimeSent >= from); + } + + if (timeSentRange?.To is { } to) + { + source = source.Where(message => message.TimeSent <= to); + } + + return source; + } + + public static IQueryable IncludeSystemMessagesWhere(this IQueryable source, bool includeSystemMessages) => + includeSystemMessages ? source : source.Where(message => !message.IsSystemMessage); + + /// + /// The sort options of the message endpoints, which differ from the failed message endpoints. + /// + public static IQueryable SortMessages(this IQueryable source, SortInfo? sortInfo) + { + var descending = sortInfo?.Direction != "asc"; + + // critical_time, delivery_time and processing_time are accepted but fall through to + // time_sent: the error instance never enriches those statistics, so every message reports + // zero and sorting by them is meaningless here. RavenDB behaves the same way, its index + // fields for them are always null. + return sortInfo?.Sort switch + { + "id" or "message_id" => source.OrderBy(message => message.MessageId, descending), + "message_type" => source.OrderBy(message => message.MessageType, descending), + "processed_at" => source.OrderBy(message => message.LastAttemptedAt, descending), + // Ordering follows the status the view reports, not the one the column stores. + "status" => source.OrderBy(message => + message.Status == FailedMessageStatus.Resolved + ? MessageStatus.ResolvedSuccessfully + : message.Status == FailedMessageStatus.RetryIssued + ? MessageStatus.RetryIssued + : message.Status == FailedMessageStatus.Archived + ? MessageStatus.ArchivedFailure + : message.NumberOfProcessingAttempts == 1 + ? MessageStatus.Failed + : MessageStatus.RepeatedFailure, descending), + _ => source.OrderBy(message => message.TimeSent, descending) + }; + } + public static IQueryable Page(this IQueryable source, PagingInfo pagingInfo) => source.Skip(pagingInfo.Offset).Take(pagingInfo.Next); diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/IFullTextSearchDialect.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/IFullTextSearchDialect.cs new file mode 100644 index 0000000000..45b3642e93 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/IFullTextSearchDialect.cs @@ -0,0 +1,18 @@ +namespace ServiceControl.Persistence.EFCore.Infrastructure; + +using ServiceControl.Persistence.EFCore.Entities; + +/// +/// The provider specific full text predicate over the failed messages table. Implementations must +/// produce SQL the index created by the AddFullTextSearch migration can serve: on PostgreSQL that +/// means reproducing the indexed expression exactly, because an expression the planner cannot match +/// turns every search into a sequential scan rather than failing. +/// +public interface IFullTextSearchDialect +{ + /// + /// Terms are ORed, matching the RavenDB persister, whose Search defaults to SearchOperator.Or. + /// Callers guarantee the terms are not blank. + /// + IQueryable Search(IQueryable source, string searchTerms); +} diff --git a/src/ServiceControl.Persistence.Tests.PostgreSql/FullTextSearchIndexTests.cs b/src/ServiceControl.Persistence.Tests.PostgreSql/FullTextSearchIndexTests.cs new file mode 100644 index 0000000000..cbfeafd87b --- /dev/null +++ b/src/ServiceControl.Persistence.Tests.PostgreSql/FullTextSearchIndexTests.cs @@ -0,0 +1,51 @@ +namespace ServiceControl.Persistence.Tests; + +using System.Text.RegularExpressions; +using EFCore.PostgreSql; +using Microsoft.EntityFrameworkCore; +using NUnit.Framework; + +/// +/// PostgreSQL only uses the GIN index of the AddFullTextSearch migration when the query expression +/// parses to the same tree as the indexed one. A mismatch is silent: search keeps working, on a +/// sequential scan of every failed message. These tests need no database. +/// +class FullTextSearchIndexTests +{ + [Test] + public void Search_uses_the_indexed_expression() + { + var sql = WithoutTableAlias(SearchQuery("forty-two")); + + Assert.That(sql, Does.Contain(FullTextSearchSql.IndexedExpression)); + } + + [Test] + public void Terms_are_ored() + { + var sql = SearchQuery("forty two"); + + Assert.That(sql, Does.Contain("='forty OR two'")); + } + + static string SearchQuery(string searchTerms) + { + var options = new DbContextOptionsBuilder() + .UseNpgsql("Host=localhost;Database=servicecontrol") + .Options; + + using var dbContext = new PostgreSqlServiceControlDbContext(options); + + return new PostgreSqlFullTextSearchDialect() + .Search(dbContext.FailedMessages, searchTerms) + .ToQueryString(); + } + + // The DDL names the columns bare, the query qualifies them with whatever alias EF picked. + static string WithoutTableAlias(string sql) + { + var alias = Regex.Match(sql, @"FROM failed_messages AS (\w+)").Groups[1].Value; + + return sql.Replace($"{alias}.", string.Empty); + } +} diff --git a/src/ServiceControl.Persistence.Tests/EFCore/MessagesViewDataStoreTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/MessagesViewDataStoreTests.cs new file mode 100644 index 0000000000..cf680e2418 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/MessagesViewDataStoreTests.cs @@ -0,0 +1,364 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text; +using System.Threading.Tasks; +using NServiceBus; +using NUnit.Framework; +using ServiceControl.CompositeViews.Messages; +using ServiceControl.MessageFailures; +using ServiceControl.Operations; +using ServiceControl.Persistence.Infrastructure; + +class MessagesViewDataStoreTests : ErrorIngestionTestBase +{ + [Test] + public async Task Reports_a_stored_failure() + { + var failure = new IngestedFailure(); + + await Ingest(failure); + + var view = await SingleMessage(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(view.Id, Is.EqualTo(failure.UniqueMessageIdString)); + Assert.That(view.MessageId, Is.EqualTo(failure.MessageId)); + Assert.That(view.MessageType, Is.EqualTo(failure.MessageType)); + Assert.That(view.ConversationId, Is.EqualTo(failure.ConversationId)); + Assert.That(view.TimeSent, Is.EqualTo(failure.TimeSent)); + Assert.That(view.ProcessedAt, Is.EqualTo(failure.AttemptedAt)); + Assert.That(view.SendingEndpoint.Name, Is.EqualTo(failure.SendingEndpoint.Name)); + Assert.That(view.ReceivingEndpoint.Name, Is.EqualTo(failure.ReceivingEndpoint.Name)); + Assert.That(view.BodySize, Is.EqualTo(failure.Body.Length)); + Assert.That(view.BodyUrl, Is.EqualTo($"/messages/{failure.UniqueMessageIdString}/body")); + Assert.That(view.IsSystemMessage, Is.False); + Assert.That(view.Headers.ToDictionary(header => header.Key, header => header.Value), + Does.ContainKey(NServiceBus.Headers.EnclosedMessageTypes)); + } + } + + [Test] + public async Task Reports_the_message_intent() + { + await Ingest(new IngestedFailure { MessageIntent = MessageIntent.Reply }); + + var view = await SingleMessage(); + + Assert.That(view.MessageIntent, Is.EqualTo(MessageIntent.Reply)); + } + + /// + /// The error instance never enriches the processing statistics, so RavenDB reports zeroes here + /// too. The sort options for them fall through to time_sent for the same reason. + /// + [Test] + public async Task Reports_no_processing_statistics() + { + await Ingest(new IngestedFailure()); + + var view = await SingleMessage(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(view.CriticalTime, Is.EqualTo(TimeSpan.Zero)); + Assert.That(view.ProcessingTime, Is.EqualTo(TimeSpan.Zero)); + Assert.That(view.DeliveryTime, Is.EqualTo(TimeSpan.Zero)); + } + } + + [Test] + public async Task Reports_a_null_time_sent() + { + await Ingest(new IngestedFailure { TimeSent = null }); + + var view = await SingleMessage(); + + Assert.That(view.TimeSent, Is.Null); + } + + [TestCase(FailedMessageStatus.Unresolved, 1, MessageStatus.Failed)] + [TestCase(FailedMessageStatus.Unresolved, 2, MessageStatus.RepeatedFailure)] + [TestCase(FailedMessageStatus.Resolved, 1, MessageStatus.ResolvedSuccessfully)] + [TestCase(FailedMessageStatus.RetryIssued, 1, MessageStatus.RetryIssued)] + [TestCase(FailedMessageStatus.Archived, 1, MessageStatus.ArchivedFailure)] + [TestCase(FailedMessageStatus.Archived, 2, MessageStatus.ArchivedFailure)] + public async Task Reports_the_status(FailedMessageStatus status, int attempts, MessageStatus expected) + { + await Insert(new IngestedFailure(), status, attempts); + + var view = await SingleMessage(); + + Assert.That(view.Status, Is.EqualTo(expected)); + } + + [Test] + public async Task Hides_system_messages_unless_asked() + { + var system = new IngestedFailure { IsSystemMessage = true }; + + await Ingest(new IngestedFailure(), system); + + var hidden = await MessagesViewStore.GetAllMessages(new PagingInfo(), new SortInfo(), includeSystemMessages: false); + var included = await MessagesViewStore.GetAllMessages(new PagingInfo(), new SortInfo(), includeSystemMessages: true); + + using (Assert.EnterMultipleScope()) + { + Assert.That(hidden.Results.Select(view => view.Id), Does.Not.Contain(system.UniqueMessageIdString)); + Assert.That(included.Results.Select(view => view.Id), Does.Contain(system.UniqueMessageIdString)); + } + } + + [Test] + public async Task Sorts_by_time_sent_by_default() + { + var (oldest, middle, newest) = await IngestThreeSentMinutesApart(); + + var descending = await MessagesViewStore.GetAllMessages(new PagingInfo(), new SortInfo(), true); + var ascending = await MessagesViewStore.GetAllMessages(new PagingInfo(), new SortInfo(direction: "asc"), true); + + using (Assert.EnterMultipleScope()) + { + Assert.That(descending.Results.Select(view => view.Id), Is.EqualTo(new[] { newest, middle, oldest })); + Assert.That(ascending.Results.Select(view => view.Id), Is.EqualTo(new[] { oldest, middle, newest })); + } + } + + /// + /// The error instance reports zero for all three, so they sort by time sent like everything the + /// API does not sort by. + /// + [TestCase("critical_time")] + [TestCase("delivery_time")] + [TestCase("processing_time")] + public async Task Sorts_by_time_sent_for_the_statistics_options(string sort) + { + var (oldest, middle, newest) = await IngestThreeSentMinutesApart(); + + var result = await MessagesViewStore.GetAllMessages(new PagingInfo(), new SortInfo(sort), true); + + Assert.That(result.Results.Select(view => view.Id), Is.EqualTo(new[] { newest, middle, oldest })); + } + + [Test] + public async Task Sorts_by_status_as_reported() + { + var failed = new IngestedFailure(); + var archived = new IngestedFailure(); + + await Insert(failed, FailedMessageStatus.Unresolved, 1); + await Insert(archived, FailedMessageStatus.Archived, 1); + + var result = await MessagesViewStore.GetAllMessages(new PagingInfo(), new SortInfo("status", "asc"), true); + + // Failed is 1 and ArchivedFailure is 5 as MessageStatus, the reverse of the order the + // FailedMessageStatus column stores them in. + Assert.That(result.Results.Select(view => view.Id), Is.EqualTo(new[] { failed.UniqueMessageIdString, archived.UniqueMessageIdString })); + } + + [Test] + public async Task Pages_and_counts() + { + var (_, middle, _) = await IngestThreeSentMinutesApart(); + + var result = await MessagesViewStore.GetAllMessages(new PagingInfo(page: 2, pageSize: 1), new SortInfo(), true); + + using (Assert.EnterMultipleScope()) + { + Assert.That(result.Results.Select(view => view.Id), Is.EqualTo(new[] { middle })); + Assert.That(result.QueryStats.TotalCount, Is.EqualTo(3)); + } + } + + [Test] + public async Task Filters_by_time_sent_range() + { + var (_, middle, newest) = await IngestThreeSentMinutesApart(); + + var result = await MessagesViewStore.GetAllMessages(new PagingInfo(), new SortInfo(), true, + new DateTimeRange(BaseTimeSent, BaseTimeSent.AddMinutes(3))); + + using (Assert.EnterMultipleScope()) + { + Assert.That(result.Results.Select(view => view.Id), Is.EqualTo(new[] { newest, middle })); + Assert.That(result.QueryStats.TotalCount, Is.EqualTo(2)); + } + } + + [Test] + public async Task Filters_by_endpoint() + { + var sales = new IngestedFailure(); + var billing = new IngestedFailure { ReceivingEndpoint = new EndpointDetails { Name = "Billing", Host = "BillingHost", HostId = Guid.NewGuid() } }; + + await Ingest(sales, billing); + + var result = await MessagesViewStore.GetAllMessagesForEndpoint("Billing", new PagingInfo(), new SortInfo(), true); + + Assert.That(result.Results.Select(view => view.Id), Is.EqualTo(new[] { billing.UniqueMessageIdString })); + } + + [Test] + public async Task Filters_by_conversation() + { + var conversationId = Guid.NewGuid().ToString(); + var inConversation = new IngestedFailure { ConversationId = conversationId }; + + await Ingest(inConversation, new IngestedFailure()); + + var result = await MessagesViewStore.GetAllMessagesByConversation(conversationId, new PagingInfo(), new SortInfo(), false); + + Assert.That(result.Results.Select(view => view.Id), Is.EqualTo(new[] { inConversation.UniqueMessageIdString })); + } + + /// + /// The RavenDB persister ignores includeSystemMessages here, and so does this one: a + /// conversation is incomplete without the system messages that took part in it. + /// + [Test] + public async Task Keeps_system_messages_in_a_conversation() + { + var conversationId = Guid.NewGuid().ToString(); + var system = new IngestedFailure { ConversationId = conversationId, IsSystemMessage = true }; + + await Ingest(system); + + var result = await MessagesViewStore.GetAllMessagesByConversation(conversationId, new PagingInfo(), new SortInfo(), false); + + Assert.That(result.Results.Select(view => view.Id), Is.EqualTo(new[] { system.UniqueMessageIdString })); + } + + [Test] + public async Task Searches_the_headers() + { + var matching = new IngestedFailure { ExceptionMessage = "the zarquon overheated" }; + + await Ingest(matching, new IngestedFailure()); + + await AssertSearchFinds("zarquon", matching); + } + + [Test] + public async Task Searches_the_body() + { + var matching = new IngestedFailure { Body = Encoding.UTF8.GetBytes("zarquon") }; + + await Ingest(matching, new IngestedFailure()); + + await AssertSearchFinds("zarquon", matching); + } + + /// + /// The full name is a single token to the PostgreSQL parser, which is why the index carries a + /// separator stripped copy of the message type. + /// + [Test] + public async Task Searches_the_short_message_type() + { + var matching = new IngestedFailure { MessageType = "MyCompany.Sales.ZarquonOverheated" }; + + await Ingest(matching, new IngestedFailure()); + + await AssertSearchFinds("ZarquonOverheated", matching); + } + + /// + /// ServicePulse links to /messages/search/{messageId}, so the id has to be findable even though + /// the two parsers tokenise it differently: the SQL Server word breaker splits it on the + /// hyphens, PostgreSQL keeps it as a single uuid token. + /// + [Test] + public async Task Searches_the_message_id() + { + var matching = new IngestedFailure(); + + await Ingest(matching); + + await AssertSearchFinds(matching.MessageId, matching); + } + + [Test] + public async Task Ors_the_search_terms() + { + var first = new IngestedFailure { ExceptionMessage = "the zarquon overheated" }; + var second = new IngestedFailure { ExceptionMessage = "the flux capacitor melted" }; + + await Ingest(first, second); + + await AssertSearchFinds("zarquon capacitor", first, second); + } + + [Test] + public async Task Searches_within_an_endpoint() + { + var billing = new IngestedFailure + { + ExceptionMessage = "the zarquon overheated", + ReceivingEndpoint = new EndpointDetails { Name = "Billing", Host = "BillingHost", HostId = Guid.NewGuid() } + }; + var sales = new IngestedFailure { ExceptionMessage = "the zarquon overheated" }; + + await Ingest(billing, sales); + + await WaitForSearchResults( + () => MessagesViewStore.SearchEndpointMessages("Billing", "zarquon", new PagingInfo(), new SortInfo()), + billing); + } + + static readonly DateTime BaseTimeSent = new(2026, 7, 22, 10, 0, 0, DateTimeKind.Utc); + + async Task<(string Oldest, string Middle, string Newest)> IngestThreeSentMinutesApart() + { + var oldest = new IngestedFailure { TimeSent = BaseTimeSent.AddMinutes(-2) }; + var middle = new IngestedFailure { TimeSent = BaseTimeSent }; + var newest = new IngestedFailure { TimeSent = BaseTimeSent.AddMinutes(2) }; + + await Ingest(oldest, middle, newest); + + return (oldest.UniqueMessageIdString, middle.UniqueMessageIdString, newest.UniqueMessageIdString); + } + + async Task SingleMessage() + { + var result = await MessagesViewStore.GetAllMessages(new PagingInfo(), new SortInfo(), includeSystemMessages: true); + + return result.Results.Single(); + } + + Task AssertSearchFinds(string searchTerms, params IngestedFailure[] expected) => + WaitForSearchResults( + () => MessagesViewStore.GetAllMessagesForSearch(searchTerms, new PagingInfo(), new SortInfo()), + expected); + + /// + /// SQL Server populates its full text index asynchronously, so a search right after the write + /// legitimately returns nothing for a moment. PostgreSQL is current as soon as the transaction + /// commits and passes on the first attempt. + /// + static async Task WaitForSearchResults(Func>>> search, params IngestedFailure[] expected) + { + var expectedIds = expected.Select(failure => failure.UniqueMessageIdString).OrderBy(id => id).ToArray(); + IList results = []; + + await WaitUntil(async () => + { + results = (await search()).Results; + + return results.Count == expectedIds.Length; + }, $"Search returned {expectedIds.Length} message(s)", TimeSpan.FromSeconds(30)); + + Assert.That(results.Select(view => view.Id).OrderBy(id => id), Is.EqualTo(expectedIds)); + } + + async Task Insert(IngestedFailure failure, FailedMessageStatus status, int attempts) + { + var message = failure.ToFailedMessage(status, attempts); + message.Id = PersistenceTestsContext.GenerateFailedMessageRecordId(message.UniqueMessageId); + + await PersistenceTestsContext.InsertFailedMessages(message); + await CompleteDatabaseOperation(); + } +} diff --git a/src/ServiceControl.Persistence.Tests/IngestedFailure.cs b/src/ServiceControl.Persistence.Tests/IngestedFailure.cs index a8395c5527..12d1d48696 100644 --- a/src/ServiceControl.Persistence.Tests/IngestedFailure.cs +++ b/src/ServiceControl.Persistence.Tests/IngestedFailure.cs @@ -3,6 +3,7 @@ namespace ServiceControl.Persistence.Tests; using System; using System.Collections.Generic; using System.Text; +using NServiceBus; using NServiceBus.Extensibility; using NServiceBus.Transport; using ServiceControl.Contracts.Operations; @@ -18,7 +19,8 @@ class IngestedFailure public byte[] Body { get; init; } = Encoding.UTF8.GetBytes("1"); public DateTime AttemptedAt { get; init; } = new(2026, 7, 22, 10, 0, 0, DateTimeKind.Utc); public DateTime TimeOfFailure { get; init; } = new(2026, 7, 22, 10, 0, 0, DateTimeKind.Utc); - public DateTime TimeSent { get; init; } = new(2026, 7, 22, 9, 59, 0, DateTimeKind.Utc); + public DateTime? TimeSent { get; init; } = new(2026, 7, 22, 9, 59, 0, DateTimeKind.Utc); + public MessageIntent MessageIntent { get; init; } = MessageIntent.Send; public string MessageType { get; init; } = "MyCompany.Sales.OrderPlaced"; public string ConversationId { get; init; } = Guid.NewGuid().ToString(); public string QueueAddress { get; init; } = "error"; @@ -46,6 +48,7 @@ Dictionary BuildHeaders() [NServiceBus.Headers.ProcessingEndpoint] = EndpointName, [NServiceBus.Headers.ContentType] = ContentType, [NServiceBus.Headers.EnclosedMessageTypes] = MessageType, + [NServiceBus.Headers.MessageIntent] = MessageIntent.ToString(), ["NServiceBus.FailedQ"] = QueueAddress, ["NServiceBus.ExceptionInfo.ExceptionType"] = ExceptionType, ["NServiceBus.ExceptionInfo.Message"] = ExceptionMessage