From 099c3648353102cf5146d9f108c2770c7be28621 Mon Sep 17 00:00:00 2001 From: John Simons Date: Fri, 31 Jul 2026 15:58:57 +1000 Subject: [PATCH] Add EF Core-backed failure group queries Implements the read side of IGroupsDataStore for the SQL Server and PostgreSQL persisters. Groups are aggregated on the fly by joining FailedMessageGroups to FailedMessages and grouping by (GroupId, Title, Type) --- ...AddFailureGroupClassifierIndex.Designer.cs | 378 ++++++++++++++++++ ...31043221_AddFailureGroupClassifierIndex.cs | 27 ++ ...SqlServiceControlDbContextModelSnapshot.cs | 3 + ...AddFailureGroupClassifierIndex.Designer.cs | 308 ++++++++++++++ ...31043129_AddFailureGroupClassifierIndex.cs | 27 ++ ...verServiceControlDbContextModelSnapshot.cs | 2 + .../FailedMessageGroupConfiguration.cs | 3 + .../FailedMessageQueryDataStore.cs | 42 +- .../FailedMessageQueryResults.cs | 24 ++ .../Implementation/GroupsDataStore.cs | 79 +++- .../Infrastructure/FailureGroupQueries.cs | 32 ++ .../Recoverability/GroupsDataStoreTests.cs | 254 ++++++++++++ 12 files changed, 1142 insertions(+), 37 deletions(-) create mode 100644 src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260731043221_AddFailureGroupClassifierIndex.Designer.cs create mode 100644 src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260731043221_AddFailureGroupClassifierIndex.cs create mode 100644 src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260731043129_AddFailureGroupClassifierIndex.Designer.cs create mode 100644 src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260731043129_AddFailureGroupClassifierIndex.cs create mode 100644 src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryResults.cs create mode 100644 src/ServiceControl.Persistence.EFCore/Infrastructure/FailureGroupQueries.cs create mode 100644 src/ServiceControl.Persistence.Tests/Recoverability/GroupsDataStoreTests.cs diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260731043221_AddFailureGroupClassifierIndex.Designer.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260731043221_AddFailureGroupClassifierIndex.Designer.cs new file mode 100644 index 0000000000..5ed94cb171 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260731043221_AddFailureGroupClassifierIndex.Designer.cs @@ -0,0 +1,378 @@ +// +using System; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Migrations; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; +using Npgsql.EntityFrameworkCore.PostgreSQL.Metadata; +using ServiceControl.Persistence.EFCore.PostgreSql; + +#nullable disable + +namespace ServiceControl.Persistence.EFCore.PostgreSql.Migrations +{ + [DbContext(typeof(PostgreSqlServiceControlDbContext))] + [Migration("20260731043221_AddFailureGroupClassifierIndex")] + partial class AddFailureGroupClassifierIndex + { + /// + protected override void BuildTargetModel(ModelBuilder modelBuilder) + { +#pragma warning disable 612, 618 + modelBuilder + .HasAnnotation("ProductVersion", "10.0.10") + .HasAnnotation("Relational:MaxIdentifierLength", 63); + + NpgsqlModelBuilderExtensions.UseIdentityByDefaultColumns(modelBuilder); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.EndpointSettingsEntity", b => + { + b.Property("Name") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("name"); + + b.Property("TrackInstances") + .HasColumnType("boolean") + .HasColumnName("track_instances"); + + b.HasKey("Name") + .HasName("pk_endpoint_settings"); + + b.ToTable("endpoint_settings", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedErrorImportEntity", b => + { + b.Property("UniqueMessageId") + .HasColumnType("uuid") + .HasColumnName("unique_message_id"); + + b.Property("Body") + .IsRequired() + .HasColumnType("bytea") + .HasColumnName("body"); + + b.Property("BodyStoredExternally") + .HasColumnType("boolean") + .HasColumnName("body_stored_externally"); + + b.Property("ExceptionInfo") + .IsRequired() + .HasColumnType("text") + .HasColumnName("exception_info"); + + b.Property("FailedAt") + .HasColumnType("timestamp with time zone") + .HasColumnName("failed_at"); + + b.Property("HeadersJson") + .IsRequired() + .HasColumnType("text") + .HasColumnName("headers_json"); + + b.Property("MessageId") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("message_id"); + + b.HasKey("UniqueMessageId") + .HasName("pk_failed_error_imports"); + + b.HasIndex("FailedAt") + .HasDatabaseName("ix_failed_error_imports_failed_at"); + + b.ToTable("failed_error_imports", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedMessageEntity", b => + { + b.Property("UniqueMessageId") + .HasColumnType("uuid") + .HasColumnName("unique_message_id"); + + b.Property("BodyContentType") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("body_content_type"); + + b.Property("BodySize") + .HasColumnType("integer") + .HasColumnName("body_size"); + + b.Property("BodyStoredExternally") + .HasColumnType("boolean") + .HasColumnName("body_stored_externally"); + + b.Property("BodyText") + .HasColumnType("text") + .HasColumnName("body_text"); + + b.Property("ConversationId") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("conversation_id"); + + b.Property("ExceptionMessage") + .HasColumnType("text") + .HasColumnName("exception_message"); + + b.Property("ExceptionType") + .HasColumnType("text") + .HasColumnName("exception_type"); + + b.Property("FailingEndpointAddress") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("failing_endpoint_address"); + + b.Property("FirstTimeOfFailure") + .HasColumnType("timestamp with time zone") + .HasColumnName("first_time_of_failure"); + + b.Property("HeadersJson") + .IsRequired() + .HasColumnType("text") + .HasColumnName("headers_json"); + + b.Property("IsSystemMessage") + .HasColumnType("boolean") + .HasColumnName("is_system_message"); + + b.Property("LastAttemptedAt") + .HasColumnType("timestamp with time zone") + .HasColumnName("last_attempted_at"); + + b.Property("LastModified") + .HasColumnType("timestamp with time zone") + .HasColumnName("last_modified"); + + b.Property("LastTimeOfFailure") + .HasColumnType("timestamp with time zone") + .HasColumnName("last_time_of_failure"); + + b.Property("MessageId") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("message_id"); + + b.Property("MessageType") + .HasColumnType("text") + .HasColumnName("message_type"); + + b.Property("NumberOfProcessingAttempts") + .HasColumnType("integer") + .HasColumnName("number_of_processing_attempts"); + + b.Property("ReceivingEndpointHost") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("receiving_endpoint_host"); + + b.Property("ReceivingEndpointHostId") + .HasColumnType("uuid") + .HasColumnName("receiving_endpoint_host_id"); + + b.Property("ReceivingEndpointName") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("receiving_endpoint_name"); + + b.Property("SendingEndpointHost") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("sending_endpoint_host"); + + b.Property("SendingEndpointHostId") + .HasColumnType("uuid") + .HasColumnName("sending_endpoint_host_id"); + + b.Property("SendingEndpointName") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("sending_endpoint_name"); + + b.Property("Status") + .HasColumnType("integer") + .HasColumnName("status"); + + b.Property("StatusChangedAt") + .HasColumnType("timestamp with time zone") + .HasColumnName("status_changed_at"); + + b.Property("TimeSent") + .HasColumnType("timestamp with time zone") + .HasColumnName("time_sent"); + + b.HasKey("UniqueMessageId") + .HasName("pk_failed_messages"); + + b.HasIndex("ConversationId") + .HasDatabaseName("ix_failed_messages_conversation_id"); + + b.HasIndex("FailingEndpointAddress") + .HasDatabaseName("ix_failed_messages_failing_endpoint_address"); + + b.HasIndex("ReceivingEndpointName") + .HasDatabaseName("ix_failed_messages_receiving_endpoint_name"); + + b.HasIndex("StatusChangedAt") + .HasDatabaseName("ix_failed_messages_status_changed_at") + .HasFilter("status IN (2, 4)"); + + b.HasIndex("TimeSent") + .HasDatabaseName("ix_failed_messages_time_sent"); + + b.HasIndex("Status", "LastModified") + .HasDatabaseName("ix_failed_messages_status_last_modified"); + + b.ToTable("failed_messages", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedMessageGroupEntity", b => + { + b.Property("FailedMessageUniqueId") + .HasColumnType("uuid") + .HasColumnName("failed_message_unique_id"); + + b.Property("GroupId") + .HasMaxLength(64) + .HasColumnType("character varying(64)") + .HasColumnName("group_id"); + + b.Property("Title") + .IsRequired() + .HasColumnType("text") + .HasColumnName("title"); + + b.Property("Type") + .IsRequired() + .HasMaxLength(255) + .HasColumnType("character varying(255)") + .HasColumnName("type"); + + b.HasKey("FailedMessageUniqueId", "GroupId") + .HasName("pk_failed_message_groups"); + + b.HasIndex("GroupId") + .HasDatabaseName("ix_failed_message_groups_group_id"); + + b.HasIndex("Type", "GroupId") + .HasDatabaseName("ix_failed_message_groups_type_group_id"); + + b.ToTable("failed_message_groups", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedMessageRetryEntity", b => + { + b.Property("UniqueMessageId") + .HasColumnType("uuid") + .HasColumnName("unique_message_id"); + + b.Property("RetryId") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("retry_id"); + + b.HasKey("UniqueMessageId") + .HasName("pk_failed_message_retries"); + + b.ToTable("failed_message_retries", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.KnownEndpointEntity", b => + { + b.Property("Id") + .HasColumnType("uuid") + .HasColumnName("id"); + + b.Property("Host") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("host"); + + b.Property("HostId") + .HasColumnType("uuid") + .HasColumnName("host_id"); + + b.Property("Monitored") + .HasColumnType("boolean") + .HasColumnName("monitored"); + + b.Property("Name") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("name"); + + b.HasKey("Id") + .HasName("pk_known_endpoints"); + + b.ToTable("known_endpoints", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.SubscriptionEntity", b => + { + b.Property("MessageType") + .HasMaxLength(200) + .HasColumnType("character varying(200)") + .HasColumnName("message_type"); + + b.Property("TransportAddress") + .HasMaxLength(200) + .HasColumnType("character varying(200)") + .HasColumnName("transport_address"); + + b.Property("Endpoint") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("endpoint"); + + b.HasKey("MessageType", "TransportAddress") + .HasName("pk_subscriptions"); + + b.ToTable("subscriptions", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.TrialMetadataEntity", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("integer") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("TrialEndDate") + .HasColumnType("date") + .HasColumnName("trial_end_date"); + + b.HasKey("Id") + .HasName("pk_trial_metadata"); + + b.ToTable("trial_metadata", (string)null); + + b.HasData( + new + { + Id = 1 + }); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedMessageGroupEntity", b => + { + b.HasOne("ServiceControl.Persistence.EFCore.Entities.FailedMessageEntity", null) + .WithMany() + .HasForeignKey("FailedMessageUniqueId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired() + .HasConstraintName("fk_failed_message_groups_failed_messages_failed_message_unique"); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260731043221_AddFailureGroupClassifierIndex.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260731043221_AddFailureGroupClassifierIndex.cs new file mode 100644 index 0000000000..f61b3ea87e --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260731043221_AddFailureGroupClassifierIndex.cs @@ -0,0 +1,27 @@ +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace ServiceControl.Persistence.EFCore.PostgreSql.Migrations +{ + /// + public partial class AddFailureGroupClassifierIndex : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.CreateIndex( + name: "ix_failed_message_groups_type_group_id", + table: "failed_message_groups", + columns: new[] { "type", "group_id" }); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropIndex( + name: "ix_failed_message_groups_type_group_id", + table: "failed_message_groups"); + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs index cca72b3640..1623ec1da1 100644 --- a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs @@ -305,6 +305,9 @@ protected override void BuildModel(ModelBuilder modelBuilder) b.HasIndex("GroupId") .HasDatabaseName("ix_failed_message_groups_group_id"); + b.HasIndex("Type", "GroupId") + .HasDatabaseName("ix_failed_message_groups_type_group_id"); + b.ToTable("failed_message_groups", (string)null); }); diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260731043129_AddFailureGroupClassifierIndex.Designer.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260731043129_AddFailureGroupClassifierIndex.Designer.cs new file mode 100644 index 0000000000..7418ae7263 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260731043129_AddFailureGroupClassifierIndex.Designer.cs @@ -0,0 +1,308 @@ +// +using System; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Infrastructure; +using Microsoft.EntityFrameworkCore.Metadata; +using Microsoft.EntityFrameworkCore.Migrations; +using Microsoft.EntityFrameworkCore.Storage.ValueConversion; +using ServiceControl.Persistence.EFCore.SqlServer; + +#nullable disable + +namespace ServiceControl.Persistence.EFCore.SqlServer.Migrations +{ + [DbContext(typeof(SqlServerServiceControlDbContext))] + [Migration("20260731043129_AddFailureGroupClassifierIndex")] + partial class AddFailureGroupClassifierIndex + { + /// + protected override void BuildTargetModel(ModelBuilder modelBuilder) + { +#pragma warning disable 612, 618 + modelBuilder + .HasAnnotation("ProductVersion", "10.0.10") + .HasAnnotation("Relational:MaxIdentifierLength", 128); + + SqlServerModelBuilderExtensions.UseIdentityColumns(modelBuilder); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.EndpointSettingsEntity", b => + { + b.Property("Name") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("TrackInstances") + .HasColumnType("bit"); + + b.HasKey("Name"); + + b.ToTable("EndpointSettings"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedErrorImportEntity", b => + { + b.Property("UniqueMessageId") + .HasColumnType("uniqueidentifier"); + + b.Property("Body") + .IsRequired() + .HasColumnType("varbinary(max)"); + + b.Property("BodyStoredExternally") + .HasColumnType("bit"); + + b.Property("ExceptionInfo") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("FailedAt") + .HasColumnType("datetime2"); + + b.Property("HeadersJson") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("MessageId") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.HasKey("UniqueMessageId"); + + b.HasIndex("FailedAt"); + + b.ToTable("FailedErrorImports"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedMessageEntity", b => + { + b.Property("UniqueMessageId") + .HasColumnType("uniqueidentifier"); + + b.Property("BodyContentType") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("BodySize") + .HasColumnType("int"); + + b.Property("BodyStoredExternally") + .HasColumnType("bit"); + + b.Property("BodyText") + .HasColumnType("nvarchar(max)"); + + b.Property("ConversationId") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("ExceptionMessage") + .HasColumnType("nvarchar(max)"); + + b.Property("ExceptionType") + .HasColumnType("nvarchar(max)"); + + b.Property("FailingEndpointAddress") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("FirstTimeOfFailure") + .HasColumnType("datetime2"); + + b.Property("HeadersJson") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("IsSystemMessage") + .HasColumnType("bit"); + + b.Property("LastAttemptedAt") + .HasColumnType("datetime2"); + + b.Property("LastModified") + .HasColumnType("datetime2"); + + b.Property("LastTimeOfFailure") + .HasColumnType("datetime2"); + + b.Property("MessageId") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("MessageType") + .HasColumnType("nvarchar(max)"); + + b.Property("NumberOfProcessingAttempts") + .HasColumnType("int"); + + b.Property("ReceivingEndpointHost") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("ReceivingEndpointHostId") + .HasColumnType("uniqueidentifier"); + + b.Property("ReceivingEndpointName") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("SendingEndpointHost") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("SendingEndpointHostId") + .HasColumnType("uniqueidentifier"); + + b.Property("SendingEndpointName") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("Status") + .HasColumnType("int"); + + b.Property("StatusChangedAt") + .HasColumnType("datetime2"); + + b.Property("TimeSent") + .HasColumnType("datetime2"); + + b.HasKey("UniqueMessageId"); + + b.HasIndex("ConversationId"); + + b.HasIndex("FailingEndpointAddress"); + + b.HasIndex("ReceivingEndpointName"); + + b.HasIndex("StatusChangedAt") + .HasFilter("[Status] IN (2, 4)"); + + b.HasIndex("TimeSent"); + + b.HasIndex("Status", "LastModified"); + + b.ToTable("FailedMessages"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedMessageGroupEntity", b => + { + b.Property("FailedMessageUniqueId") + .HasColumnType("uniqueidentifier"); + + b.Property("GroupId") + .HasMaxLength(64) + .HasColumnType("nvarchar(64)"); + + b.Property("Title") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("Type") + .IsRequired() + .HasMaxLength(255) + .HasColumnType("nvarchar(255)"); + + b.HasKey("FailedMessageUniqueId", "GroupId"); + + b.HasIndex("GroupId"); + + b.HasIndex("Type", "GroupId"); + + b.ToTable("FailedMessageGroups"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedMessageRetryEntity", b => + { + b.Property("UniqueMessageId") + .HasColumnType("uniqueidentifier"); + + b.Property("RetryId") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.HasKey("UniqueMessageId"); + + b.ToTable("FailedMessageRetries"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.KnownEndpointEntity", b => + { + b.Property("Id") + .HasColumnType("uniqueidentifier"); + + b.Property("Host") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("HostId") + .HasColumnType("uniqueidentifier"); + + b.Property("Monitored") + .HasColumnType("bit"); + + b.Property("Name") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.HasKey("Id"); + + b.ToTable("KnownEndpoints"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.SubscriptionEntity", b => + { + b.Property("MessageType") + .HasMaxLength(200) + .HasColumnType("nvarchar(200)"); + + b.Property("TransportAddress") + .HasMaxLength(200) + .HasColumnType("nvarchar(200)"); + + b.Property("Endpoint") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.HasKey("MessageType", "TransportAddress"); + + b.ToTable("Subscriptions"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.TrialMetadataEntity", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("int"); + + SqlServerPropertyBuilderExtensions.UseIdentityColumn(b.Property("Id")); + + b.Property("TrialEndDate") + .HasColumnType("date"); + + b.HasKey("Id"); + + b.ToTable("TrialMetadata"); + + b.HasData( + new + { + Id = 1 + }); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedMessageGroupEntity", b => + { + b.HasOne("ServiceControl.Persistence.EFCore.Entities.FailedMessageEntity", null) + .WithMany() + .HasForeignKey("FailedMessageUniqueId") + .OnDelete(DeleteBehavior.Cascade) + .IsRequired(); + }); +#pragma warning restore 612, 618 + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260731043129_AddFailureGroupClassifierIndex.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260731043129_AddFailureGroupClassifierIndex.cs new file mode 100644 index 0000000000..eac954e32f --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260731043129_AddFailureGroupClassifierIndex.cs @@ -0,0 +1,27 @@ +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace ServiceControl.Persistence.EFCore.SqlServer.Migrations +{ + /// + public partial class AddFailureGroupClassifierIndex : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.CreateIndex( + name: "IX_FailedMessageGroups_Type_GroupId", + table: "FailedMessageGroups", + columns: new[] { "Type", "GroupId" }); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropIndex( + name: "IX_FailedMessageGroups_Type_GroupId", + table: "FailedMessageGroups"); + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs index 829753e136..42cb2b31bd 100644 --- a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs @@ -243,6 +243,8 @@ protected override void BuildModel(ModelBuilder modelBuilder) b.HasIndex("GroupId"); + b.HasIndex("Type", "GroupId"); + b.ToTable("FailedMessageGroups"); }); diff --git a/src/ServiceControl.Persistence.EFCore/EntityConfigurations/FailedMessageGroupConfiguration.cs b/src/ServiceControl.Persistence.EFCore/EntityConfigurations/FailedMessageGroupConfiguration.cs index 591081cb80..535913c3ef 100644 --- a/src/ServiceControl.Persistence.EFCore/EntityConfigurations/FailedMessageGroupConfiguration.cs +++ b/src/ServiceControl.Persistence.EFCore/EntityConfigurations/FailedMessageGroupConfiguration.cs @@ -17,6 +17,9 @@ public void Configure(EntityTypeBuilder builder) builder.HasIndex(e => e.GroupId); + // Drives the per-classifier group aggregate, which filters on Type and groups by GroupId. + builder.HasIndex(e => new { e.Type, e.GroupId }); + builder.HasOne() .WithMany() .HasForeignKey(e => e.FailedMessageUniqueId) diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryDataStore.cs index c0cb7ee2fa..9060da9cf9 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryDataStore.cs @@ -13,14 +13,12 @@ namespace ServiceControl.Persistence.EFCore.Implementation; public class FailedMessageQueryDataStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), IFailedMessageQueryDataStore { public Task>> GetFailedMessages(string? status, string? modified, string? queueAddress, PagingInfo pagingInfo, SortInfo sortInfo) => - ExecuteWithDbContext(dbContext => Page( - dbContext.FailedMessages - .AsNoTracking() - .FilterByStatus(status) - .FilterByLastModifiedRange(modified) - .FilterByQueueAddress(queueAddress), - pagingInfo, - sortInfo)); + ExecuteWithDbContext(dbContext => dbContext.FailedMessages + .AsNoTracking() + .FilterByStatus(status) + .FilterByLastModifiedRange(modified) + .FilterByQueueAddress(queueAddress) + .ToPagedResult(pagingInfo, sortInfo)); public Task GetFailedMessagesStats(string? status, string? modified, string? queueAddress) => ExecuteWithDbContext(dbContext => dbContext.FailedMessages @@ -31,14 +29,12 @@ public Task GetFailedMessagesStats(string? status, string? modif .ToQueryStatsInfo()); public Task>> GetFailedMessagesByEndpoint(string? status, string endpointName, string? modified, PagingInfo pagingInfo, SortInfo sortInfo) => - ExecuteWithDbContext(dbContext => Page( - dbContext.FailedMessages - .AsNoTracking() - .Where(message => message.ReceivingEndpointName == endpointName) - .FilterByStatus(status) - .FilterByLastModifiedRange(modified), - pagingInfo, - sortInfo)); + ExecuteWithDbContext(dbContext => dbContext.FailedMessages + .AsNoTracking() + .Where(message => message.ReceivingEndpointName == endpointName) + .FilterByStatus(status) + .FilterByLastModifiedRange(modified) + .ToPagedResult(pagingInfo, sortInfo)); public Task> GetFailedMessagesSummary() => ExecuteWithDbContext(async dbContext => @@ -110,20 +106,6 @@ public Task GetFailedMessagesByIds(Guid[] ids) => return [.. entities.Select(entity => entity.ToFailedMessage(groups.GetValueOrDefault(entity.UniqueMessageId, [])))]; }); - static async Task>> Page(IQueryable query, PagingInfo pagingInfo, SortInfo sortInfo) - { - var stats = await query.ToQueryStatsInfo(); - - var entities = await query - .Sort(sortInfo) - .Page(pagingInfo) - .ToListAsync(); - - IList results = [.. entities.Select(entity => entity.ToFailedMessageView())]; - - return new QueryResult>(results, stats); - } - static async Task> CountBy(ServiceControlDbContext dbContext, Expression> selector) => await dbContext.FailedMessages .AsNoTracking() diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryResults.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryResults.cs new file mode 100644 index 0000000000..9c6fe90250 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryResults.cs @@ -0,0 +1,24 @@ +namespace ServiceControl.Persistence.EFCore.Implementation; + +using Microsoft.EntityFrameworkCore; +using ServiceControl.MessageFailures.Api; +using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Persistence.EFCore.Infrastructure; +using ServiceControl.Persistence.Infrastructure; + +static class FailedMessageQueryResults +{ + public static async Task>> ToPagedResult(this IQueryable source, PagingInfo pagingInfo, SortInfo sortInfo) + { + var stats = await source.ToQueryStatsInfo(); + + var entities = await source + .Sort(sortInfo) + .Page(pagingInfo) + .ToListAsync(); + + IList results = [.. entities.Select(entity => entity.ToFailedMessageView())]; + + return new QueryResult>(results, stats); + } +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/GroupsDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/GroupsDataStore.cs index 0ef4ea6a5c..56760ece42 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/GroupsDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/GroupsDataStore.cs @@ -1,35 +1,100 @@ namespace ServiceControl.Persistence.EFCore.Implementation; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using ServiceControl.MessageFailures; using ServiceControl.MessageFailures.Api; +using ServiceControl.Persistence.EFCore.DbContexts; +using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Persistence.EFCore.Infrastructure; using ServiceControl.Persistence.Infrastructure; using ServiceControl.Recoverability; -public class GroupsDataStore : IGroupsDataStore +public class GroupsDataStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), IGroupsDataStore { public Task> GetFailureGroupsByClassifier(string classifier, string classifierFilter) => - throw new NotImplementedException(); + ExecuteWithDbContext(dbContext => + { + var groups = ByClassifier(dbContext, classifier); + + if (!string.IsNullOrWhiteSpace(classifierFilter)) + { + groups = groups.Where(group => group.Title == classifierFilter); + } + + return MostRecent(groups.AggregateGroups(WithStatus(dbContext, FailedMessageStatus.Unresolved))); + }); public Task> GetArchivedFailureGroupsByClassifier(string classifier) => - throw new NotImplementedException(); + ExecuteWithDbContext(dbContext => MostRecent( + ByClassifier(dbContext, classifier).AggregateGroups(WithStatus(dbContext, FailedMessageStatus.Archived)))); + // Implemented once retry batches are persisted, together with IRetryDocumentDataStore. public Task GetCurrentForwardingBatch() => throw new NotImplementedException(); public Task>> GetGroup(string groupId, string status, string modified) => - throw new NotImplementedException(); + ExecuteWithDbContext(async dbContext => + { + var groups = await ById(dbContext, groupId, FailedMessageStatus.Unresolved, status, modified).ToListAsync(); + + return new QueryResult>(groups, groups.ToQueryStatsInfo()); + }); public Task> GetFailureGroupView(string groupId, string status, string modified) => - throw new NotImplementedException(); + ExecuteWithDbContext(async dbContext => + { + var groups = await ById(dbContext, groupId, FailedMessageStatus.Archived, status, modified).ToListAsync(); + + // A missing group is reported as a null result, the same as the RavenDB persister does. + return new QueryResult(groups.FirstOrDefault()!, groups.ToQueryStatsInfo()); + }); public Task>> GetGroupErrors(string groupId, string status, string modified, SortInfo sortInfo, PagingInfo pagingInfo) => - throw new NotImplementedException(); + ExecuteWithDbContext(dbContext => InGroup(dbContext, groupId, status, modified).ToPagedResult(pagingInfo, sortInfo)); public Task GetGroupErrorsCount(string groupId, string status, string modified) => - throw new NotImplementedException(); + ExecuteWithDbContext(dbContext => InGroup(dbContext, groupId, status, modified).ToQueryStatsInfo()); public Task EditComment(string groupId, string comment) => throw new NotImplementedException(); public Task DeleteComment(string groupId) => throw new NotImplementedException(); + + static IQueryable ByClassifier(ServiceControlDbContext dbContext, string classifier) => + dbContext.FailedMessageGroups + .AsNoTracking() + .Where(group => group.Type == classifier); + + /// + /// The status a group is read at, before the caller's own status and modified filters narrow it + /// further. RavenDB reads open groups out of an unresolved-only index and archived groups out of + /// an archived-only one, which is what stands in for here. + /// + static IQueryable ById(ServiceControlDbContext dbContext, string groupId, FailedMessageStatus baseline, string status, string modified) => + dbContext.FailedMessageGroups + .AsNoTracking() + .Where(group => group.GroupId == groupId) + .AggregateGroups(WithStatus(dbContext, baseline) + .FilterByStatus(status) + .FilterByLastModifiedRange(modified)); + + static IQueryable WithStatus(ServiceControlDbContext dbContext, FailedMessageStatus status) => + dbContext.FailedMessages + .AsNoTracking() + .Where(message => message.Status == status); + + static IQueryable InGroup(ServiceControlDbContext dbContext, string groupId, string status, string modified) => + dbContext.FailedMessages + .AsNoTracking() + .Where(message => dbContext.FailedMessageGroups.Any(group => group.GroupId == groupId && group.FailedMessageUniqueId == message.UniqueMessageId)) + .FilterByStatus(status) + .FilterByLastModifiedRange(modified); + + static async Task> MostRecent(IQueryable groups) => + await groups + .OrderByDescending(group => group.Last) + .Take(FailureGroupQueries.MaxGroups) + .ToListAsync(); } diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/FailureGroupQueries.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/FailureGroupQueries.cs new file mode 100644 index 0000000000..0767a7d4b2 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/FailureGroupQueries.cs @@ -0,0 +1,32 @@ +namespace ServiceControl.Persistence.EFCore.Infrastructure; + +using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Persistence.Infrastructure; +using ServiceControl.Recoverability; + +static class FailureGroupQueries +{ + public const int MaxGroups = 200; + + public static IQueryable AggregateGroups(this IQueryable groups, IQueryable messages) => + from failureGroup in groups + join message in messages on failureGroup.FailedMessageUniqueId equals message.UniqueMessageId + group message by new { failureGroup.GroupId, failureGroup.Title, failureGroup.Type } + into aggregate + select new FailureGroupView + { + Id = aggregate.Key.GroupId, + Title = aggregate.Key.Title, + Type = aggregate.Key.Type, + Count = aggregate.Count(), + First = aggregate.Min(message => message.FirstTimeOfFailure), + Last = aggregate.Max(message => message.LastTimeOfFailure) + }; + + public static QueryStatsInfo ToQueryStatsInfo(this IReadOnlyCollection groups) + { + var latest = groups.Count == 0 ? DateTime.MinValue : groups.Max(group => group.Last); + + return new QueryStatsInfo($"{groups.Count}-{latest.Ticks}", groups.Count, false); + } +} diff --git a/src/ServiceControl.Persistence.Tests/Recoverability/GroupsDataStoreTests.cs b/src/ServiceControl.Persistence.Tests/Recoverability/GroupsDataStoreTests.cs new file mode 100644 index 0000000000..4f8654f193 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/Recoverability/GroupsDataStoreTests.cs @@ -0,0 +1,254 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using NUnit.Framework; +using ServiceControl.MessageFailures; +using ServiceControl.Persistence.Infrastructure; +using ServiceControl.Recoverability; + +class GroupsDataStoreTests : PersistenceTestBase +{ + const string Classifier = "Exception Type and Stack Trace"; + const string OtherClassifier = "Message Type"; + + static readonly DateTime Noon = new(2026, 7, 22, 12, 0, 0, DateTimeKind.Utc); + + [Test] + public async Task Aggregates_the_messages_of_a_group() + { + var group = NewGroup("OrderPlaced"); + + await Insert( + InGroup(group, failedAt: Noon), + InGroup(group, failedAt: Noon.AddHours(2))); + + var view = (await GroupsStore.GetFailureGroupsByClassifier(Classifier, null)).Single(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(view.Id, Is.EqualTo(group.Id)); + Assert.That(view.Title, Is.EqualTo("OrderPlaced")); + Assert.That(view.Type, Is.EqualTo(Classifier)); + Assert.That(view.Count, Is.EqualTo(2)); + Assert.That(view.First, Is.EqualTo(Noon)); + Assert.That(view.Last, Is.EqualTo(Noon.AddHours(2))); + } + } + + [Test] + public async Task Returns_only_the_groups_of_the_requested_classifier() + { + var requested = NewGroup("OrderPlaced"); + var other = NewGroup("OrderPlaced", OtherClassifier); + + await Insert(InGroup(requested), InGroup(other)); + + var groups = await GroupsStore.GetFailureGroupsByClassifier(Classifier, null); + + Assert.That(groups.Select(group => group.Id), Is.EqualTo(new[] { requested.Id })); + } + + [Test] + public async Task Narrows_the_groups_to_the_classifier_filter() + { + var matching = NewGroup("OrderPlaced"); + var other = NewGroup("OrderCancelled"); + + await Insert(InGroup(matching), InGroup(other)); + + var groups = await GroupsStore.GetFailureGroupsByClassifier(Classifier, "OrderPlaced"); + + Assert.That(groups.Select(group => group.Id), Is.EqualTo(new[] { matching.Id })); + } + + [Test] + public async Task Counts_only_unresolved_messages() + { + var group = NewGroup("OrderPlaced"); + + await Insert( + InGroup(group).ToFailedMessage(), + InGroup(group).ToFailedMessage(FailedMessageStatus.Archived), + InGroup(group).ToFailedMessage(FailedMessageStatus.Resolved)); + + var view = (await GroupsStore.GetFailureGroupsByClassifier(Classifier, null)).Single(); + + Assert.That(view.Count, Is.EqualTo(1)); + } + + [Test] + public async Task Returns_archived_groups_separately() + { + var group = NewGroup("OrderPlaced"); + + await Insert( + InGroup(group).ToFailedMessage(), + InGroup(group).ToFailedMessage(FailedMessageStatus.Archived)); + + var view = (await GroupsStore.GetArchivedFailureGroupsByClassifier(Classifier)).Single(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(view.Id, Is.EqualTo(group.Id)); + Assert.That(view.Count, Is.EqualTo(1)); + } + } + + [Test] + public async Task Orders_groups_by_the_most_recent_failure() + { + var oldest = NewGroup("Oldest"); + var newest = NewGroup("Newest"); + var middle = NewGroup("Middle"); + + await Insert( + InGroup(oldest, failedAt: Noon), + InGroup(newest, failedAt: Noon.AddHours(4)), + InGroup(middle, failedAt: Noon.AddHours(2))); + + var groups = await GroupsStore.GetFailureGroupsByClassifier(Classifier, null); + + Assert.That(groups.Select(group => group.Title), Is.EqualTo(new[] { "Newest", "Middle", "Oldest" })); + } + + [Test] + public async Task Returns_a_single_group_by_id() + { + var requested = NewGroup("OrderPlaced"); + + await Insert(InGroup(requested), InGroup(NewGroup("OrderCancelled"))); + + var result = await GroupsStore.GetGroup(requested.Id, null, null); + + var view = result.Results.Single(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(view.Id, Is.EqualTo(requested.Id)); + Assert.That(view.Count, Is.EqualTo(1)); + } + } + + [Test] + public async Task Returns_no_group_for_an_unknown_id() + { + await Insert(InGroup(NewGroup("OrderPlaced"))); + + var result = await GroupsStore.GetGroup(Guid.NewGuid().ToString(), null, null); + + Assert.That(result.Results, Is.Empty); + } + + [Test] + public async Task Returns_an_archived_group_view_by_id() + { + var group = NewGroup("OrderPlaced"); + + await Insert(InGroup(group).ToFailedMessage(FailedMessageStatus.Archived)); + + var result = await GroupsStore.GetFailureGroupView(group.Id, null, null); + + using (Assert.EnterMultipleScope()) + { + Assert.That(result.Results.Id, Is.EqualTo(group.Id)); + Assert.That(result.Results.Count, Is.EqualTo(1)); + } + } + + [Test] + public async Task Returns_no_archived_group_view_for_an_unknown_id() + { + await Insert(InGroup(NewGroup("OrderPlaced")).ToFailedMessage(FailedMessageStatus.Archived)); + + var result = await GroupsStore.GetFailureGroupView(Guid.NewGuid().ToString(), null, null); + + Assert.That(result.Results, Is.Null); + } + + [Test] + public async Task Returns_the_errors_of_a_group() + { + var group = NewGroup("OrderPlaced"); + var inGroup = InGroup(group); + var outsideGroup = InGroup(NewGroup("OrderCancelled")); + + await Insert(inGroup, outsideGroup); + + var result = await GroupsStore.GetGroupErrors(group.Id, null, null, new SortInfo(), new PagingInfo()); + + Assert.That(result.Results.Select(view => view.Id), Is.EqualTo(new[] { inGroup.UniqueMessageIdString })); + } + + [Test] + public async Task Pages_the_errors_of_a_group() + { + var group = NewGroup("OrderPlaced"); + var newest = InGroup(group, sentAt: Noon.AddHours(2)); + var oldest = InGroup(group, sentAt: Noon); + + await Insert(newest, oldest); + + var result = await GroupsStore.GetGroupErrors(group.Id, null, null, new SortInfo("time_sent", "desc"), new PagingInfo(page: 1, pageSize: 1)); + + using (Assert.EnterMultipleScope()) + { + Assert.That(result.Results.Select(view => view.Id), Is.EqualTo(new[] { newest.UniqueMessageIdString })); + Assert.That(result.QueryStats.TotalCount, Is.EqualTo(2)); + } + } + + [Test] + public async Task Returns_the_errors_of_a_group_by_status() + { + var group = NewGroup("OrderPlaced"); + var unresolved = InGroup(group); + var archived = InGroup(group); + + await Insert(unresolved.ToFailedMessage(), archived.ToFailedMessage(FailedMessageStatus.Archived)); + + var result = await GroupsStore.GetGroupErrors(group.Id, "archived", null, new SortInfo(), new PagingInfo()); + + Assert.That(result.Results.Select(view => view.Id), Is.EqualTo(new[] { archived.UniqueMessageIdString })); + } + + [Test] + public async Task Counts_the_errors_of_a_group() + { + var group = NewGroup("OrderPlaced"); + + await Insert(InGroup(group), InGroup(group), InGroup(NewGroup("OrderCancelled"))); + + var stats = await GroupsStore.GetGroupErrorsCount(group.Id, null, null); + + Assert.That(stats.TotalCount, Is.EqualTo(2)); + } + + static FailedMessage.FailureGroup NewGroup(string title, string type = Classifier) => + new() { Id = Guid.NewGuid().ToString(), Title = title, Type = type }; + + static IngestedFailure InGroup(FailedMessage.FailureGroup group, DateTime? failedAt = null, DateTime? sentAt = null) => + new() + { + Groups = [group], + AttemptedAt = failedAt ?? Noon, + TimeOfFailure = failedAt ?? Noon, + TimeSent = sentAt ?? Noon.AddMinutes(-1) + }; + + Task Insert(params IngestedFailure[] failures) => + Insert([.. failures.Select(failure => failure.ToFailedMessage())]); + + async Task Insert(params FailedMessage[] messages) + { + foreach (var message in messages) + { + message.Id = PersistenceTestsContext.GenerateFailedMessageRecordId(message.UniqueMessageId); + } + + await PersistenceTestsContext.InsertFailedMessages(messages); + await CompleteDatabaseOperation(); + } +}