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.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/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; 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