diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260731003309_DropDuplicateQueueAddress.Designer.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260731003309_DropDuplicateQueueAddress.Designer.cs new file mode 100644 index 0000000000..7570ecd8f8 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260731003309_DropDuplicateQueueAddress.Designer.cs @@ -0,0 +1,375 @@ +// +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("20260731003309_DropDuplicateQueueAddress")] + partial class DropDuplicateQueueAddress + { + /// + 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.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/20260731003309_DropDuplicateQueueAddress.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260731003309_DropDuplicateQueueAddress.cs new file mode 100644 index 0000000000..ee9c0704bc --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260731003309_DropDuplicateQueueAddress.cs @@ -0,0 +1,38 @@ +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace ServiceControl.Persistence.EFCore.PostgreSql.Migrations +{ + /// + public partial class DropDuplicateQueueAddress : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropIndex( + name: "ix_failed_messages_queue_address", + table: "failed_messages"); + + migrationBuilder.DropColumn( + name: "queue_address", + table: "failed_messages"); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.AddColumn( + name: "queue_address", + table: "failed_messages", + type: "character varying(450)", + maxLength: 450, + nullable: true); + + migrationBuilder.CreateIndex( + name: "ix_failed_messages_queue_address", + table: "failed_messages", + column: "queue_address"); + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs index c62889ba93..cca72b3640 100644 --- a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs @@ -212,11 +212,6 @@ protected override void BuildModel(ModelBuilder modelBuilder) .HasColumnType("integer") .HasColumnName("number_of_processing_attempts"); - b.Property("QueueAddress") - .HasMaxLength(450) - .HasColumnType("character varying(450)") - .HasColumnName("queue_address"); - b.Property("ReceivingEndpointHost") .HasMaxLength(450) .HasColumnType("character varying(450)") @@ -266,9 +261,6 @@ protected override void BuildModel(ModelBuilder modelBuilder) b.HasIndex("FailingEndpointAddress") .HasDatabaseName("ix_failed_messages_failing_endpoint_address"); - b.HasIndex("QueueAddress") - .HasDatabaseName("ix_failed_messages_queue_address"); - b.HasIndex("ReceivingEndpointName") .HasDatabaseName("ix_failed_messages_receiving_endpoint_name"); diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlIngestionSqlDialect.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlIngestionSqlDialect.cs index 96421bd7bd..6d57b747f9 100644 --- a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlIngestionSqlDialect.cs +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlIngestionSqlDialect.cs @@ -90,7 +90,7 @@ static async Task Execute(ServiceControlDbContext dbContext, string sql, IEnumer // The columns the newer attempt wins wholesale static readonly string[] PayloadColumns = [ - "message_id", "message_type", "time_sent", "conversation_id", "queue_address", + "message_id", "message_type", "time_sent", "conversation_id", "failing_endpoint_address", "sending_endpoint_name", "sending_endpoint_host_id", "sending_endpoint_host", "receiving_endpoint_name", "receiving_endpoint_host_id", "receiving_endpoint_host", "exception_type", "exception_message", "is_system_message", @@ -109,7 +109,7 @@ .. PayloadColumns [ row.UniqueMessageId, (int)row.Status, row.StatusChangedAt, row.LastModified, row.NumberOfProcessingAttempts, row.FirstTimeOfFailure, row.LastTimeOfFailure, row.LastAttemptedAt, - row.MessageId, row.MessageType, row.TimeSent, row.ConversationId, row.QueueAddress, + row.MessageId, row.MessageType, row.TimeSent, row.ConversationId, row.FailingEndpointAddress, row.SendingEndpointName, row.SendingEndpointHostId, row.SendingEndpointHost, row.ReceivingEndpointName, row.ReceivingEndpointHostId, row.ReceivingEndpointHost, row.ExceptionType, row.ExceptionMessage, row.IsSystemMessage, diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260731003301_DropDuplicateQueueAddress.Designer.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260731003301_DropDuplicateQueueAddress.Designer.cs new file mode 100644 index 0000000000..2d959177d0 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260731003301_DropDuplicateQueueAddress.Designer.cs @@ -0,0 +1,306 @@ +// +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("20260731003301_DropDuplicateQueueAddress")] + partial class DropDuplicateQueueAddress + { + /// + 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.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/20260731003301_DropDuplicateQueueAddress.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260731003301_DropDuplicateQueueAddress.cs new file mode 100644 index 0000000000..2694196c62 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260731003301_DropDuplicateQueueAddress.cs @@ -0,0 +1,38 @@ +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace ServiceControl.Persistence.EFCore.SqlServer.Migrations +{ + /// + public partial class DropDuplicateQueueAddress : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropIndex( + name: "IX_FailedMessages_QueueAddress", + table: "FailedMessages"); + + migrationBuilder.DropColumn( + name: "QueueAddress", + table: "FailedMessages"); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.AddColumn( + name: "QueueAddress", + table: "FailedMessages", + type: "nvarchar(450)", + maxLength: 450, + nullable: true); + + migrationBuilder.CreateIndex( + name: "IX_FailedMessages_QueueAddress", + table: "FailedMessages", + column: "QueueAddress"); + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs index b8140a169f..829753e136 100644 --- a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs @@ -172,10 +172,6 @@ protected override void BuildModel(ModelBuilder modelBuilder) b.Property("NumberOfProcessingAttempts") .HasColumnType("int"); - b.Property("QueueAddress") - .HasMaxLength(450) - .HasColumnType("nvarchar(450)"); - b.Property("ReceivingEndpointHost") .HasMaxLength(450) .HasColumnType("nvarchar(450)"); @@ -213,8 +209,6 @@ protected override void BuildModel(ModelBuilder modelBuilder) b.HasIndex("FailingEndpointAddress"); - b.HasIndex("QueueAddress"); - b.HasIndex("ReceivingEndpointName"); b.HasIndex("StatusChangedAt") diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerIngestionSqlDialect.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerIngestionSqlDialect.cs index 86f31948b4..85f78a253a 100644 --- a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerIngestionSqlDialect.cs +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerIngestionSqlDialect.cs @@ -109,7 +109,7 @@ static async Task Execute(ServiceControlDbContext dbContext, string sql, IEnumer // The columns the newer attempt wins wholesale static readonly string[] PayloadColumns = [ - "MessageId", "MessageType", "TimeSent", "ConversationId", "QueueAddress", + "MessageId", "MessageType", "TimeSent", "ConversationId", "FailingEndpointAddress", "SendingEndpointName", "SendingEndpointHostId", "SendingEndpointHost", "ReceivingEndpointName", "ReceivingEndpointHostId", "ReceivingEndpointHost", "ExceptionType", "ExceptionMessage", "IsSystemMessage", @@ -128,7 +128,7 @@ .. PayloadColumns [ row.UniqueMessageId, (int)row.Status, row.StatusChangedAt, row.LastModified, row.NumberOfProcessingAttempts, row.FirstTimeOfFailure, row.LastTimeOfFailure, row.LastAttemptedAt, - row.MessageId, row.MessageType, row.TimeSent, row.ConversationId, row.QueueAddress, + row.MessageId, row.MessageType, row.TimeSent, row.ConversationId, row.FailingEndpointAddress, row.SendingEndpointName, row.SendingEndpointHostId, row.SendingEndpointHost, row.ReceivingEndpointName, row.ReceivingEndpointHostId, row.ReceivingEndpointHost, row.ExceptionType, row.ExceptionMessage, row.IsSystemMessage, diff --git a/src/ServiceControl.Persistence.EFCore/Entities/FailedMessageEntity.cs b/src/ServiceControl.Persistence.EFCore/Entities/FailedMessageEntity.cs index 85e360594f..81a83456f0 100644 --- a/src/ServiceControl.Persistence.EFCore/Entities/FailedMessageEntity.cs +++ b/src/ServiceControl.Persistence.EFCore/Entities/FailedMessageEntity.cs @@ -28,8 +28,6 @@ public class FailedMessageEntity public string? ConversationId { get; set; } - public string? QueueAddress { get; set; } - public string? SendingEndpointName { get; set; } public Guid? SendingEndpointHostId { get; set; } diff --git a/src/ServiceControl.Persistence.EFCore/EntityConfigurations/FailedMessageConfiguration.cs b/src/ServiceControl.Persistence.EFCore/EntityConfigurations/FailedMessageConfiguration.cs index a378c6593b..b746d1f7b1 100644 --- a/src/ServiceControl.Persistence.EFCore/EntityConfigurations/FailedMessageConfiguration.cs +++ b/src/ServiceControl.Persistence.EFCore/EntityConfigurations/FailedMessageConfiguration.cs @@ -21,7 +21,6 @@ public void Configure(EntityTypeBuilder builder) builder.Property(e => e.MessageId).HasMaxLength(ColumnLengths.ShortTextLength); builder.Property(e => e.ConversationId).HasMaxLength(ColumnLengths.ShortTextLength); - builder.Property(e => e.QueueAddress).HasMaxLength(ColumnLengths.ShortTextLength); builder.Property(e => e.SendingEndpointName).HasMaxLength(ColumnLengths.ShortTextLength); builder.Property(e => e.SendingEndpointHost).HasMaxLength(ColumnLengths.ShortTextLength); builder.Property(e => e.ReceivingEndpointName).HasMaxLength(ColumnLengths.ShortTextLength); @@ -39,7 +38,6 @@ public void Configure(EntityTypeBuilder builder) builder.HasIndex(e => e.FailingEndpointAddress); builder.HasIndex(e => e.ConversationId); builder.HasIndex(e => e.TimeSent); - builder.HasIndex(e => e.QueueAddress); // Drives the retention sweep. The index is restricted to the statuses the sweep deletes // (Resolved and Archived) by a provider specific filter, applied in the provider DbContext. diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryDataStore.cs index 182882aa97..c0cb7ee2fa 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryDataStore.cs @@ -1,30 +1,148 @@ namespace ServiceControl.Persistence.EFCore.Implementation; +using System.Linq.Expressions; +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; public class FailedMessageQueryDataStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), IFailedMessageQueryDataStore { - public Task>> GetFailedMessages(string status, string modified, string queueAddress, PagingInfo pagingInfo, SortInfo sortInfo) => - throw new NotImplementedException(); + 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)); - public Task GetFailedMessagesStats(string status, string modified, string queueAddress) => - throw new NotImplementedException(); + public Task GetFailedMessagesStats(string? status, string? modified, string? queueAddress) => + ExecuteWithDbContext(dbContext => dbContext.FailedMessages + .AsNoTracking() + .FilterByStatus(status) + .FilterByLastModifiedRange(modified) + .FilterByQueueAddress(queueAddress) + .ToQueryStatsInfo()); - public Task>> GetFailedMessagesByEndpoint(string status, string endpointName, string modified, PagingInfo pagingInfo, SortInfo sortInfo) => - throw new NotImplementedException(); + 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)); public Task> GetFailedMessagesSummary() => - throw new NotImplementedException(); + ExecuteWithDbContext(async dbContext => + { + var endpoints = await CountBy(dbContext, message => message.ReceivingEndpointName); + var hosts = await CountBy(dbContext, message => message.ReceivingEndpointHost); + var messageTypes = await CountBy(dbContext, message => message.MessageType); - public Task GetLatestFailedMessageView(string failedMessageId) => - throw new NotImplementedException(); + return (IDictionary)new Dictionary + { + ["Endpoints"] = endpoints, + ["Hosts"] = hosts, + ["Message types"] = messageTypes + }; + }); - public Task GetFailedMessage(string failedMessageId) => - throw new NotImplementedException(); + public Task GetLatestFailedMessageView(string failedMessageId) => + ExecuteWithDbContext(async dbContext => + { + if (!Guid.TryParse(failedMessageId, out var uniqueMessageId)) + { + return null; + } + + var entity = await dbContext.FailedMessages + .AsNoTracking() + .SingleOrDefaultAsync(message => message.UniqueMessageId == uniqueMessageId); + + return entity?.ToFailedMessageView(); + }); + + public Task GetFailedMessage(string failedMessageId) => + ExecuteWithDbContext(async dbContext => + { + if (!Guid.TryParse(failedMessageId, out var uniqueMessageId)) + { + return null; + } + + var entity = await dbContext.FailedMessages + .AsNoTracking() + .SingleOrDefaultAsync(message => message.UniqueMessageId == uniqueMessageId); + + if (entity == null) + { + return null; + } + + var groups = await ReadGroups(dbContext, [uniqueMessageId]); + + return entity.ToFailedMessage(groups.GetValueOrDefault(uniqueMessageId, [])); + }); public Task GetFailedMessagesByIds(Guid[] ids) => - throw new NotImplementedException(); + ExecuteWithDbContext(async dbContext => + { + var entities = await dbContext.FailedMessages + .AsNoTracking() + .Where(message => ids.Contains(message.UniqueMessageId)) + .ToListAsync(); + + if (entities.Count == 0) + { + return []; + } + + var groups = await ReadGroups(dbContext, [.. entities.Select(entity => entity.UniqueMessageId)]); + + 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() + .Where(message => message.Status == FailedMessageStatus.Unresolved) + .Select(selector) + .Where(value => value != null && value != string.Empty) + .GroupBy(value => value) + .Select(group => new { Value = group.Key, Count = group.Count() }) + .ToDictionaryAsync(row => row.Value!, row => (object)row.Count); + + static async Task>> ReadGroups(ServiceControlDbContext dbContext, Guid[] uniqueMessageIds) + { + var groups = await dbContext.FailedMessageGroups + .AsNoTracking() + .Where(group => uniqueMessageIds.Contains(group.FailedMessageUniqueId)) + .ToListAsync(); + + return groups + .GroupBy(group => group.FailedMessageUniqueId) + .ToDictionary(group => group.Key, group => group.ToList()); + } } diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageViewMapper.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageViewMapper.cs new file mode 100644 index 0000000000..d9115160ca --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageViewMapper.cs @@ -0,0 +1,152 @@ +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.Implementation.UnitOfWork; + +static class FailedMessageViewMapper +{ + const string EditOfHeader = "ServiceControl.EditOf"; + const string ExceptionSourceHeader = "NServiceBus.ExceptionInfo.Source"; + const string ExceptionStackTraceHeader = "NServiceBus.ExceptionInfo.StackTrace"; + + public static FailedMessageView ToFailedMessageView(this FailedMessageEntity entity) + { + var headers = entity.ReadHeaders(); + var editOf = headers.GetValueOrDefault(EditOfHeader); + + return new FailedMessageView + { + Id = entity.UniqueMessageId.ToString(), + MessageType = entity.MessageType, + IsSystemMessage = entity.IsSystemMessage, + TimeSent = entity.TimeSent, + MessageId = entity.MessageId, + Exception = entity.ToExceptionDetails(headers), + QueueAddress = entity.FailingEndpointAddress, + NumberOfProcessingAttempts = entity.NumberOfProcessingAttempts, + Status = entity.Status, + SendingEndpoint = entity.ToSendingEndpoint(), + ReceivingEndpoint = entity.ToReceivingEndpoint(), + TimeOfFailure = entity.LastTimeOfFailure, + LastModified = entity.LastModified, + Edited = editOf != null, + EditOf = editOf ?? string.Empty + }; + } + + public static FailedMessage ToFailedMessage(this FailedMessageEntity entity, IReadOnlyCollection groups) + { + var headers = entity.ReadHeaders(); + + var attempt = new FailedMessage.ProcessingAttempt + { + AttemptedAt = entity.LastAttemptedAt, + MessageId = entity.MessageId, + Headers = headers, + Body = entity.BodyText, + MessageMetadata = entity.ToMetadata(), + FailureDetails = new FailureDetails + { + TimeOfFailure = entity.LastTimeOfFailure, + AddressOfFailingEndpoint = entity.FailingEndpointAddress, + Exception = entity.ToExceptionDetails(headers) + } + }; + + return new FailedMessage + { + Id = entity.UniqueMessageId.ToString(), + UniqueMessageId = entity.UniqueMessageId.ToString(), + Status = entity.Status, + ProcessingAttempts = BuildAttempts(attempt, entity.NumberOfProcessingAttempts), + FailureGroups = [.. groups.Select(group => new FailedMessage.FailureGroup + { + Id = group.GroupId, + Title = group.Title, + Type = group.Type + })] + }; + } + + static List BuildAttempts(FailedMessage.ProcessingAttempt last, int numberOfAttempts) + { + var attempts = new List(Math.Max(numberOfAttempts, 1)); + + for (var i = 1; i < numberOfAttempts; i++) + { + attempts.Add(new FailedMessage.ProcessingAttempt()); + } + + attempts.Add(last); + + return attempts; + } + + static Dictionary ToMetadata(this FailedMessageEntity entity) + { + var metadata = new Dictionary + { + ["IsSystemMessage"] = entity.IsSystemMessage, + ["ContentLength"] = entity.BodySize + }; + + AddIfPresent(metadata, "MessageId", entity.MessageId); + AddIfPresent(metadata, "MessageType", entity.MessageType); + AddIfPresent(metadata, "ConversationId", entity.ConversationId); + AddIfPresent(metadata, "ContentType", entity.BodyContentType); + AddIfPresent(metadata, "SendingEndpoint", entity.ToSendingEndpoint()); + AddIfPresent(metadata, "ReceivingEndpoint", entity.ToReceivingEndpoint()); + + if (entity.TimeSent.HasValue) + { + metadata["TimeSent"] = entity.TimeSent.Value; + } + + return metadata; + } + + static void AddIfPresent(Dictionary metadata, string key, object? value) + { + if (value != null) + { + metadata[key] = value; + } + } + + static ExceptionDetails ToExceptionDetails(this FailedMessageEntity entity, Dictionary headers) => + new() + { + ExceptionType = entity.ExceptionType, + Message = entity.ExceptionMessage, + Source = headers.GetValueOrDefault(ExceptionSourceHeader), + StackTrace = headers.GetValueOrDefault(ExceptionStackTraceHeader) + }; + + static EndpointDetails? ToSendingEndpoint(this FailedMessageEntity entity) => + entity.SendingEndpointName == null + ? null + : new EndpointDetails + { + Name = entity.SendingEndpointName, + Host = entity.SendingEndpointHost, + HostId = entity.SendingEndpointHostId ?? Guid.Empty + }; + + static EndpointDetails? ToReceivingEndpoint(this FailedMessageEntity entity) => + entity.ReceivingEndpointName == null + ? null + : new EndpointDetails + { + Name = entity.ReceivingEndpointName, + Host = entity.ReceivingEndpointHost, + HostId = entity.ReceivingEndpointHostId ?? Guid.Empty + }; + + static Dictionary ReadHeaders(this FailedMessageEntity entity) => + JsonSerializer.Deserialize(entity.HeadersJson, HeadersJsonContext.Default.DictionaryStringString) ?? []; +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFRecoverabilityIngestionUnitOfWork.cs b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFRecoverabilityIngestionUnitOfWork.cs index 823b5e3495..92d3a37ab5 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFRecoverabilityIngestionUnitOfWork.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFRecoverabilityIngestionUnitOfWork.cs @@ -41,7 +41,6 @@ public Task RecordFailedProcessingAttempt(MessageContext context, MessageType = GetMetadata(processingAttempt, "MessageType"), TimeSent = GetMetadata(processingAttempt, "TimeSent"), ConversationId = GetMetadata(processingAttempt, "ConversationId"), - QueueAddress = context.Headers.GetValueOrDefault(NServiceBus.Faults.FaultsHeaderKeys.FailedQ), SendingEndpointName = sendingEndpoint?.Name, SendingEndpointHostId = sendingEndpoint?.HostId, SendingEndpointHost = sendingEndpoint?.Host, diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/FailedMessageBatchWriter.cs b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/FailedMessageBatchWriter.cs index 7ebc091c79..247b19157c 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/FailedMessageBatchWriter.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/FailedMessageBatchWriter.cs @@ -78,7 +78,6 @@ await strategy.ExecuteAsync(async ct => MessageType = last.MessageType, TimeSent = last.TimeSent, ConversationId = last.ConversationId, - QueueAddress = last.QueueAddress, SendingEndpointName = last.SendingEndpointName, SendingEndpointHostId = last.SendingEndpointHostId, SendingEndpointHost = last.SendingEndpointHost, diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/RecordedFailedProcessingAttempt.cs b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/RecordedFailedProcessingAttempt.cs index 6120ef0dc3..87182927c5 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/RecordedFailedProcessingAttempt.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/RecordedFailedProcessingAttempt.cs @@ -14,7 +14,6 @@ sealed class RecordedFailedProcessingAttempt public string? MessageType { get; init; } public DateTime? TimeSent { get; init; } public string? ConversationId { get; init; } - public string? QueueAddress { get; init; } public string? SendingEndpointName { get; init; } public Guid? SendingEndpointHostId { get; init; } public string? SendingEndpointHost { get; init; } diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/FailedMessageQueryFilters.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/FailedMessageQueryFilters.cs new file mode 100644 index 0000000000..d04487ed76 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/FailedMessageQueryFilters.cs @@ -0,0 +1,146 @@ +namespace ServiceControl.Persistence.EFCore.Infrastructure; + +using System.Globalization; +using Microsoft.EntityFrameworkCore; +using ServiceControl.MessageFailures; +using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Persistence.Infrastructure; + +static class FailedMessageQueryFilters +{ + static readonly string[] ModifiedRangeSeparator = ["..."]; + + const string InvalidModifiedRange = + "Invalid modified date range, dates need to be in ISO8601 format and it needs to be a range eg. 2016-03-11T00:27:15.474Z...2016-03-16T03:27:15.474Z"; + + public static IQueryable FilterByStatus(this IQueryable source, string? status) + { + if (status == null) + { + return source; + } + + var includes = new List(); + var excludes = new List(); + + foreach (var filter in status.Replace(" ", string.Empty).Split(',')) + { + if (filter.StartsWith('-')) + { + if (Enum.TryParse(filter[1..], true, out var excluded)) + { + excludes.Add(excluded); + } + + continue; + } + + if (Enum.TryParse(filter, true, out var included)) + { + includes.Add(included); + } + } + + if (includes.Count > 0) + { + source = source.Where(message => includes.Contains(message.Status)); + } + + foreach (var exclude in excludes) + { + // Captured per iteration so each exclusion closes over its own value. + var excluded = exclude; + source = source.Where(message => message.Status != excluded); + } + + return source; + } + + /// + /// Accepts an ISO8601 range in the form "from...to". + /// + public static IQueryable FilterByLastModifiedRange(this IQueryable source, string? modified) + { + if (modified == null) + { + return source; + } + + var filters = modified.Split(ModifiedRangeSeparator, StringSplitOptions.None); + + if (filters.Length != 2) + { + throw new Exception(InvalidModifiedRange); + } + + DateTime from; + DateTime to; + + try + { + from = DateTime.Parse(filters[0], CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind); + to = DateTime.Parse(filters[1], CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind); + } + catch (Exception) + { + throw new Exception(InvalidModifiedRange); + } + + return source.Where(message => message.LastModified >= from && message.LastModified <= to); + } + + public static IQueryable FilterByQueueAddress(this IQueryable source, string? queueAddress) + { + if (string.IsNullOrWhiteSpace(queueAddress)) + { + return source; + } + + var address = queueAddress.ToLowerInvariant(); + + // The ToLower() here causes a full table scan! + return source.Where(message => message.FailingEndpointAddress != null && message.FailingEndpointAddress.ToLower() == address); + } + + public static IQueryable Sort(this IQueryable source, SortInfo? sortInfo) + { + var descending = sortInfo?.Direction != "asc"; + var sort = sortInfo?.Sort; + + if (sort == null || !SortInfo.AllowedSortOptions.Contains(sort)) + { + sort = "time_sent"; + } + + return sort switch + { + "id" or "message_id" => source.OrderBy(message => message.MessageId, descending), + "message_type" => source.OrderBy(message => message.MessageType, descending), + "status" => source.OrderBy(message => message.Status, descending), + "modified" => source.OrderBy(message => message.LastModified, descending), + "time_of_failure" => source.OrderBy(message => message.LastTimeOfFailure, descending), + _ => source.OrderBy(message => message.TimeSent, descending) + }; + } + + public static IQueryable Page(this IQueryable source, PagingInfo pagingInfo) => + source.Skip(pagingInfo.Offset).Take(pagingInfo.Next); + + public static async Task ToQueryStatsInfo(this IQueryable source, CancellationToken cancellationToken = default) + { + var stats = await source + .GroupBy(_ => 1) + .Select(group => new { Count = group.Count(), Latest = group.Max(message => (DateTime?)message.LastModified) }) + .SingleOrDefaultAsync(cancellationToken); + + var count = stats?.Count ?? 0; + var latest = stats?.Latest ?? DateTime.MinValue; + + return new QueryStatsInfo($"{count}-{latest.Ticks}", count, false); + } + + static IOrderedQueryable OrderBy(this IQueryable source, System.Linq.Expressions.Expression> keySelector, bool descending) => + descending + ? source.OrderByDescending(keySelector).ThenByDescending(message => message.UniqueMessageId) + : source.OrderBy(keySelector).ThenBy(message => message.UniqueMessageId); +} diff --git a/src/ServiceControl.Persistence.RavenDB/ErrorMessagesDataStore.cs b/src/ServiceControl.Persistence.RavenDB/ErrorMessagesDataStore.cs index 91cd3f481d..89a01c6bb8 100644 --- a/src/ServiceControl.Persistence.RavenDB/ErrorMessagesDataStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/ErrorMessagesDataStore.cs @@ -245,35 +245,29 @@ SortInfo sortInfo public async Task> GetFailedMessagesSummary() { using var session = await sessionProvider.OpenSession(); + + // The facets are named after the index fields and renamed afterwards. Setting + // DisplayFieldName instead makes the server resolve the field by that name, which then + // fails with "Field Endpoints not found in Index". var facetResults = await session.Query() .AggregateBy(new List { - new Facet - { - FieldName = "Name", - DisplayFieldName = "Endpoints" - }, - new Facet - { - FieldName = "Host", - DisplayFieldName = "Hosts" - }, - new Facet - { - FieldName = "MessageType", - DisplayFieldName = "Message types" - } + new Facet { FieldName = "Name" }, + new Facet { FieldName = "Host" }, + new Facet { FieldName = "MessageType" } }).ExecuteAsync(); - var results = facetResults - .ToDictionary( - x => x.Key, - x => (object)x.Value - ); - - return results; + return new Dictionary + { + ["Endpoints"] = Counts(facetResults, "Name"), + ["Hosts"] = Counts(facetResults, "Host"), + ["Message types"] = Counts(facetResults, "MessageType") + }; } + static Dictionary Counts(Dictionary facetResults, string fieldName) => + facetResults[fieldName].Values.ToDictionary(value => value.Range, value => (object)value.Count); + public async Task GetFailedMessage(string failedMessageId) { using var session = await sessionProvider.OpenSession(); diff --git a/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionTests.cs index 7a4529c1f6..8f7d7dc677 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionTests.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionTests.cs @@ -31,7 +31,7 @@ public async Task First_failure_stores_the_message() Assert.That(row.MessageType, Is.EqualTo(failure.MessageType)); Assert.That(row.TimeSent, Is.EqualTo(failure.TimeSent)); Assert.That(row.ConversationId, Is.EqualTo(failure.ConversationId)); - Assert.That(row.QueueAddress, Is.EqualTo(failure.QueueAddress)); + Assert.That(row.FailingEndpointAddress, Is.EqualTo(failure.QueueAddress)); Assert.That(row.SendingEndpointName, Is.EqualTo(failure.SendingEndpoint.Name)); Assert.That(row.SendingEndpointHostId, Is.EqualTo(failure.SendingEndpoint.HostId)); Assert.That(row.SendingEndpointHost, Is.EqualTo(failure.SendingEndpoint.Host)); diff --git a/src/ServiceControl.Persistence.Tests/EFCore/PersistenceTestsContext.cs b/src/ServiceControl.Persistence.Tests/EFCore/PersistenceTestsContext.cs index d63e6a85e6..7476d806bd 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/PersistenceTestsContext.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/PersistenceTestsContext.cs @@ -46,7 +46,6 @@ static async Task InsertFailedMessagesDirect(IServiceProvider serviceProvider, F MessageType = GetMetadata(attempt, "MessageType"), TimeSent = GetMetadata(attempt, "TimeSent"), ConversationId = GetMetadata(attempt, "ConversationId"), - QueueAddress = attempt.Headers.GetValueOrDefault(NServiceBus.Faults.FaultsHeaderKeys.FailedQ), SendingEndpointName = sendingEndpoint?.Name, SendingEndpointHostId = sendingEndpoint?.HostId, SendingEndpointHost = sendingEndpoint?.Host, @@ -66,6 +65,19 @@ static async Task InsertFailedMessagesDirect(IServiceProvider serviceProvider, F LastModified = now, NumberOfProcessingAttempts = ordered.Select(pa => pa.AttemptedAt).Distinct().Count(), }); + + // RavenDB carries the failure groups inside the document, so seeding them has to be + // explicit here for the two persisters to start from the same state. + db.FailedMessageGroups.AddRange(failedMessage.FailureGroups + .Where(group => group.Id != null) + .DistinctBy(group => group.Id) + .Select(group => new FailedMessageGroupEntity + { + FailedMessageUniqueId = Guid.Parse(failedMessage.UniqueMessageId), + GroupId = group.Id, + Title = group.Title ?? string.Empty, + Type = group.Type ?? string.Empty + })); } await db.SaveChangesAsync(); diff --git a/src/ServiceControl.Persistence.Tests/EFCore/IngestedFailure.cs b/src/ServiceControl.Persistence.Tests/IngestedFailure.cs similarity index 55% rename from src/ServiceControl.Persistence.Tests/EFCore/IngestedFailure.cs rename to src/ServiceControl.Persistence.Tests/IngestedFailure.cs index fa95d4d694..472787f55a 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/IngestedFailure.cs +++ b/src/ServiceControl.Persistence.Tests/IngestedFailure.cs @@ -22,8 +22,12 @@ class IngestedFailure public string MessageType { get; init; } = "MyCompany.Sales.OrderPlaced"; public string ConversationId { get; init; } = Guid.NewGuid().ToString(); public string QueueAddress { get; init; } = "error"; + public string FailingEndpointAddress { get; init; } public string ExceptionType { get; init; } = "System.InvalidOperationException"; public string ExceptionMessage { get; init; } = "Something went wrong"; + public string ExceptionSource { get; init; } + public string ExceptionStackTrace { get; init; } + public string EditOf { get; init; } public bool IsSystemMessage { get; init; } public EndpointDetails SendingEndpoint { get; init; } = new() { Name = "Ordering", Host = "SenderHost", HostId = Guid.NewGuid() }; public EndpointDetails ReceivingEndpoint { get; init; } = new() { Name = "Sales", Host = "ReceiverHost", HostId = Guid.NewGuid() }; @@ -32,16 +36,38 @@ class IngestedFailure new() { Id = Guid.NewGuid().ToString(), Title = "OrderPlaced", Type = "Message Type" } ]; - public Dictionary Headers => field ??= new Dictionary + public Dictionary Headers => field ??= BuildHeaders(); + + Dictionary BuildHeaders() { - [NServiceBus.Headers.MessageId] = MessageId, - [NServiceBus.Headers.ProcessingEndpoint] = EndpointName, - [NServiceBus.Headers.ContentType] = ContentType, - [NServiceBus.Headers.EnclosedMessageTypes] = MessageType, - ["NServiceBus.FailedQ"] = QueueAddress, - ["NServiceBus.ExceptionInfo.ExceptionType"] = ExceptionType, - ["NServiceBus.ExceptionInfo.Message"] = ExceptionMessage - }; + var headers = new Dictionary + { + [NServiceBus.Headers.MessageId] = MessageId, + [NServiceBus.Headers.ProcessingEndpoint] = EndpointName, + [NServiceBus.Headers.ContentType] = ContentType, + [NServiceBus.Headers.EnclosedMessageTypes] = MessageType, + ["NServiceBus.FailedQ"] = QueueAddress, + ["NServiceBus.ExceptionInfo.ExceptionType"] = ExceptionType, + ["NServiceBus.ExceptionInfo.Message"] = ExceptionMessage + }; + + if (ExceptionSource != null) + { + headers["NServiceBus.ExceptionInfo.Source"] = ExceptionSource; + } + + if (ExceptionStackTrace != null) + { + headers["NServiceBus.ExceptionInfo.StackTrace"] = ExceptionStackTrace; + } + + if (EditOf != null) + { + headers["ServiceControl.EditOf"] = EditOf; + } + + return headers; + } public string UniqueMessageIdString => Headers.UniqueId(); @@ -68,11 +94,13 @@ class IngestedFailure FailureDetails = new FailureDetails { TimeOfFailure = TimeOfFailure, - AddressOfFailingEndpoint = QueueAddress, + AddressOfFailingEndpoint = FailingEndpointAddress ?? QueueAddress, Exception = new ExceptionDetails { ExceptionType = ExceptionType, - Message = ExceptionMessage + Message = ExceptionMessage, + Source = ExceptionSource, + StackTrace = ExceptionStackTrace } } }; @@ -88,12 +116,44 @@ class IngestedFailure MessageType = MessageType, ConversationId = ConversationId, QueueAddress = QueueAddress, + FailingEndpointAddress = FailingEndpointAddress, ExceptionType = ExceptionType, ExceptionMessage = ExceptionMessage, + ExceptionSource = ExceptionSource, + ExceptionStackTrace = ExceptionStackTrace, + EditOf = EditOf, IsSystemMessage = IsSystemMessage, SendingEndpoint = SendingEndpoint, ReceivingEndpoint = ReceivingEndpoint, TimeSent = TimeSent, Groups = Groups }; + + /// + /// The same failure as a stored document, for seeding through + /// rather than through ingestion. + /// + public FailedMessage ToFailedMessage(FailedMessageStatus status = FailedMessageStatus.Unresolved, int numberOfAttempts = 1) + { + var attempts = new List(); + + // Earlier attempts only have to be distinct in AttemptedAt, which is what the attempt count + // is derived from. Their content is never read. + for (var i = numberOfAttempts - 1; i > 0; i--) + { + attempts.Add(NextAttempt(AttemptedAt.AddMinutes(-i)).ProcessingAttempt); + } + + attempts.Add(ProcessingAttempt); + + // Id is left for the caller to stamp through IPersistenceTestsContext, because RavenDB + // stores the document under it while the relational persisters ignore it. + return new FailedMessage + { + UniqueMessageId = UniqueMessageIdString, + Status = status, + ProcessingAttempts = attempts, + FailureGroups = Groups + }; + } } diff --git a/src/ServiceControl.Persistence.Tests/MessageFailures/FailedMessageQueryAfterIngestionTests.cs b/src/ServiceControl.Persistence.Tests/MessageFailures/FailedMessageQueryAfterIngestionTests.cs new file mode 100644 index 0000000000..44779e7deb --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/MessageFailures/FailedMessageQueryAfterIngestionTests.cs @@ -0,0 +1,125 @@ +namespace ServiceControl.Persistence.Tests; + +using System.Linq; +using System.Threading.Tasks; +using NUnit.Framework; +using ServiceControl.MessageFailures; +using ServiceControl.Persistence.Infrastructure; + +class FailedMessageQueryAfterIngestionTests : PersistenceTestBase +{ + [Test] + public async Task Ingested_failure_is_returned_by_the_query_store() + { + var failure = new IngestedFailure + { + ExceptionSource = "MyCompany.Sales.Handlers", + ExceptionStackTrace = " at MyCompany.Sales.Handlers.OrderPlacedHandler.Handle()" + }; + + await Ingest(failure); + + var result = await FailedMessageQueryStore.GetFailedMessages(null, null, null, new PagingInfo(), new SortInfo()); + + var view = result.Results.Single(); + + 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.Status, Is.EqualTo(FailedMessageStatus.Unresolved)); + Assert.That(view.TimeSent, Is.EqualTo(failure.TimeSent)); + Assert.That(view.TimeOfFailure, Is.EqualTo(failure.TimeOfFailure)); + Assert.That(view.NumberOfProcessingAttempts, Is.EqualTo(1)); + Assert.That(view.SendingEndpoint.Name, Is.EqualTo(failure.SendingEndpoint.Name)); + Assert.That(view.ReceivingEndpoint.Name, Is.EqualTo(failure.ReceivingEndpoint.Name)); + Assert.That(view.Exception.ExceptionType, Is.EqualTo(failure.ExceptionType)); + Assert.That(view.Exception.Message, Is.EqualTo(failure.ExceptionMessage)); + Assert.That(view.Exception.Source, Is.EqualTo(failure.ExceptionSource)); + Assert.That(view.Exception.StackTrace, Is.EqualTo(failure.ExceptionStackTrace)); + } + } + + [Test] + public async Task Ingestion_stores_the_failing_endpoint_address() + { + var failure = new IngestedFailure { FailingEndpointAddress = "Sales@MACHINE" }; + + await Ingest(failure); + + var view = await FailedMessageQueryStore.GetLatestFailedMessageView(failure.UniqueMessageIdString); + + Assert.That(view.QueueAddress, Is.EqualTo("Sales@MACHINE")); + } + + [Test] + public async Task Ingested_failure_can_be_filtered_by_its_failing_endpoint_address() + { + var matching = new IngestedFailure { FailingEndpointAddress = "Sales@MACHINE" }; + var other = new IngestedFailure { FailingEndpointAddress = "Billing@MACHINE" }; + + await Ingest(matching, other); + + var result = await FailedMessageQueryStore.GetFailedMessages(null, null, "Sales@MACHINE", new PagingInfo(), new SortInfo()); + + Assert.That( + result.Results.Select(view => view.Id), + Is.EqualTo(new[] { matching.UniqueMessageIdString })); + } + + [Test] + public async Task Ingested_failure_is_returned_by_id() + { + var failure = new IngestedFailure(); + + await Ingest(failure); + + var message = await FailedMessageQueryStore.GetFailedMessage(failure.UniqueMessageIdString); + + using (Assert.EnterMultipleScope()) + { + Assert.That(message.UniqueMessageId, Is.EqualTo(failure.UniqueMessageIdString)); + Assert.That(message.Status, Is.EqualTo(FailedMessageStatus.Unresolved)); + Assert.That(message.ProcessingAttempts.Last().MessageId, Is.EqualTo(failure.MessageId)); + Assert.That( + message.ProcessingAttempts.Last().FailureDetails.AddressOfFailingEndpoint, + Is.EqualTo(failure.QueueAddress)); + Assert.That(message.FailureGroups.Select(group => group.Id), Is.EquivalentTo(failure.Groups.Select(group => group.Id))); + } + } + + [Test] + public async Task Repeated_failures_are_counted_as_attempts() + { + var failure = new IngestedFailure(); + var secondAttempt = failure.NextAttempt(failure.AttemptedAt.AddMinutes(5)); + + await Ingest(failure); + await Ingest(secondAttempt); + + var view = await FailedMessageQueryStore.GetLatestFailedMessageView(failure.UniqueMessageIdString); + var message = await FailedMessageQueryStore.GetFailedMessage(failure.UniqueMessageIdString); + + using (Assert.EnterMultipleScope()) + { + Assert.That(view.NumberOfProcessingAttempts, Is.EqualTo(2)); + Assert.That(message.ProcessingAttempts, Has.Count.EqualTo(2)); + } + } + + async Task Ingest(params IngestedFailure[] failures) + { + await using (var unitOfWork = await UnitOfWorkFactory.StartNew()) + { + foreach (var failure in failures) + { + await unitOfWork.Recoverability.RecordFailedProcessingAttempt(failure.Context, failure.ProcessingAttempt, failure.Groups); + } + + await unitOfWork.Complete(TestContext.CurrentContext.CancellationToken); + } + + await CompleteDatabaseOperation(); + } +} diff --git a/src/ServiceControl.Persistence.Tests/MessageFailures/FailedMessageQueryDataStoreTests.cs b/src/ServiceControl.Persistence.Tests/MessageFailures/FailedMessageQueryDataStoreTests.cs new file mode 100644 index 0000000000..057b693ff7 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/MessageFailures/FailedMessageQueryDataStoreTests.cs @@ -0,0 +1,438 @@ +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.MessageFailures.Api; +using ServiceControl.Operations; +using ServiceControl.Persistence.Infrastructure; + +class FailedMessageQueryDataStoreTests : PersistenceTestBase +{ + [Test] + public async Task Returns_a_stored_failure() + { + var failure = new IngestedFailure + { + ExceptionSource = "MyCompany.Sales.Handlers", + ExceptionStackTrace = " at MyCompany.Sales.Handlers.OrderPlacedHandler.Handle()" + }; + + await Insert(failure); + + var result = await FailedMessageQueryStore.GetFailedMessages(null, null, null, new PagingInfo(), new SortInfo()); + + var view = result.Results.Single(); + + 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.Status, Is.EqualTo(FailedMessageStatus.Unresolved)); + Assert.That(view.TimeSent, Is.EqualTo(failure.TimeSent)); + Assert.That(view.TimeOfFailure, Is.EqualTo(failure.TimeOfFailure)); + Assert.That(view.NumberOfProcessingAttempts, Is.EqualTo(1)); + Assert.That(view.QueueAddress, Is.EqualTo(failure.QueueAddress)); + Assert.That(view.SendingEndpoint.Name, Is.EqualTo(failure.SendingEndpoint.Name)); + Assert.That(view.ReceivingEndpoint.Name, Is.EqualTo(failure.ReceivingEndpoint.Name)); + Assert.That(view.Exception.ExceptionType, Is.EqualTo(failure.ExceptionType)); + Assert.That(view.Exception.Message, Is.EqualTo(failure.ExceptionMessage)); + } + } + + [Test] + public async Task Exposes_the_full_exception_details() + { + var failure = new IngestedFailure + { + ExceptionSource = "MyCompany.Sales.Handlers", + ExceptionStackTrace = " at MyCompany.Sales.Handlers.OrderPlacedHandler.Handle()" + }; + + await Insert(failure); + + var view = await FailedMessageQueryStore.GetLatestFailedMessageView(failure.UniqueMessageIdString); + + using (Assert.EnterMultipleScope()) + { + Assert.That(view.Exception.Source, Is.EqualTo(failure.ExceptionSource)); + Assert.That(view.Exception.StackTrace, Is.EqualTo(failure.ExceptionStackTrace)); + } + } + + [Test] + public async Task Reports_an_edited_message() + { + var original = new IngestedFailure(); + var edited = new IngestedFailure { EditOf = original.UniqueMessageIdString }; + + await Insert(original, edited); + + var originalView = await FailedMessageQueryStore.GetLatestFailedMessageView(original.UniqueMessageIdString); + var editedView = await FailedMessageQueryStore.GetLatestFailedMessageView(edited.UniqueMessageIdString); + + using (Assert.EnterMultipleScope()) + { + Assert.That(originalView.Edited, Is.False); + Assert.That(originalView.EditOf, Is.Empty); + Assert.That(editedView.Edited, Is.True); + Assert.That(editedView.EditOf, Is.EqualTo(original.UniqueMessageIdString)); + } + } + + [Test] + public async Task Filters_by_a_single_status() + { + var unresolved = new IngestedFailure(); + var archived = new IngestedFailure(); + + await Insert(unresolved.ToFailedMessage(), archived.ToFailedMessage(FailedMessageStatus.Archived)); + + var result = await FailedMessageQueryStore.GetFailedMessages("archived", null, null, new PagingInfo(), new SortInfo()); + + Assert.That(Ids(result), Is.EqualTo(new[] { archived.UniqueMessageIdString })); + } + + [Test] + public async Task Filters_by_several_statuses() + { + var unresolved = new IngestedFailure(); + var archived = new IngestedFailure(); + var resolved = new IngestedFailure(); + + await Insert( + unresolved.ToFailedMessage(), + archived.ToFailedMessage(FailedMessageStatus.Archived), + resolved.ToFailedMessage(FailedMessageStatus.Resolved)); + + var result = await FailedMessageQueryStore.GetFailedMessages("unresolved,archived", null, null, new PagingInfo(), new SortInfo()); + + Assert.That(Ids(result), Is.EquivalentTo(new[] { unresolved.UniqueMessageIdString, archived.UniqueMessageIdString })); + } + + // A leading '-' excludes rather than includes, which RavenDB has supported for as long as the + // status parameter has existed. No caller in this repository uses the form, but it is part of + // the API contract, so the persisters have to agree on it. + [Test] + public async Task Excludes_a_status_prefixed_with_a_minus() + { + var unresolved = new IngestedFailure(); + var archived = new IngestedFailure(); + + await Insert(unresolved.ToFailedMessage(), archived.ToFailedMessage(FailedMessageStatus.Archived)); + + var result = await FailedMessageQueryStore.GetFailedMessages("-archived", null, null, new PagingInfo(), new SortInfo()); + + Assert.That(Ids(result), Is.EqualTo(new[] { unresolved.UniqueMessageIdString })); + } + + [Test] + public async Task Ignores_an_unknown_status() + { + var failure = new IngestedFailure(); + + await Insert(failure); + + var result = await FailedMessageQueryStore.GetFailedMessages("not_a_status", null, null, new PagingInfo(), new SortInfo()); + + Assert.That(Ids(result), Is.EqualTo(new[] { failure.UniqueMessageIdString })); + } + + [Test] + public async Task Filters_by_a_modified_range() + { + var failure = new IngestedFailure(); + + await Insert(failure); + + var enclosing = Range(DateTime.UtcNow.AddDays(-1), DateTime.UtcNow.AddDays(1)); + var past = Range(DateTime.UtcNow.AddDays(-2), DateTime.UtcNow.AddDays(-1)); + + var included = await FailedMessageQueryStore.GetFailedMessages(null, enclosing, null, new PagingInfo(), new SortInfo()); + var excluded = await FailedMessageQueryStore.GetFailedMessages(null, past, null, new PagingInfo(), new SortInfo()); + + using (Assert.EnterMultipleScope()) + { + Assert.That(Ids(included), Is.EqualTo(new[] { failure.UniqueMessageIdString })); + Assert.That(Ids(excluded), Is.Empty); + } + } + + [Test] + public void Rejects_a_malformed_modified_range() + { + Assert.That( + async () => await FailedMessageQueryStore.GetFailedMessages(null, "2016-03-11T00:27:15.474Z", null, new PagingInfo(), new SortInfo()), + Throws.Exception.With.Message.Contains("Invalid modified date range")); + } + + [Test] + public async Task Filters_by_the_failing_endpoint_address() + { + var matching = new IngestedFailure { QueueAddress = "error", FailingEndpointAddress = "Sales@MACHINE" }; + var other = new IngestedFailure { QueueAddress = "error", FailingEndpointAddress = "Billing@MACHINE" }; + + await Insert(matching, other); + + var result = await FailedMessageQueryStore.GetFailedMessages(null, null, "Sales@MACHINE", new PagingInfo(), new SortInfo()); + + using (Assert.EnterMultipleScope()) + { + Assert.That(Ids(result), Is.EqualTo(new[] { matching.UniqueMessageIdString })); + Assert.That(result.Results.Single().QueueAddress, Is.EqualTo("Sales@MACHINE")); + } + } + + [Test] + public async Task Matches_the_failing_endpoint_address_regardless_of_case() + { + var failure = new IngestedFailure { FailingEndpointAddress = "Sales@MACHINE" }; + + await Insert(failure); + + var result = await FailedMessageQueryStore.GetFailedMessages(null, null, "sales@machine", new PagingInfo(), new SortInfo()); + + Assert.That(Ids(result), Is.EqualTo(new[] { failure.UniqueMessageIdString })); + } + + [Test] + public async Task Filters_by_endpoint() + { + var sales = new IngestedFailure(); + var billing = new IngestedFailure + { + ReceivingEndpoint = new EndpointDetails { Name = "Billing", Host = "Host", HostId = Guid.NewGuid() } + }; + + await Insert(sales, billing); + + var result = await FailedMessageQueryStore.GetFailedMessagesByEndpoint(null, "Billing", null, new PagingInfo(), new SortInfo()); + + Assert.That(Ids(result), Is.EqualTo(new[] { billing.UniqueMessageIdString })); + } + + [TestCase("time_sent", "asc")] + [TestCase("time_sent", "desc")] + [TestCase("message_type", "asc")] + [TestCase("message_type", "desc")] + [TestCase("status", "asc")] + [TestCase("status", "desc")] + [TestCase("time_of_failure", "asc")] + [TestCase("time_of_failure", "desc")] + public async Task Sorts_by_the_requested_field(string sort, string direction) + { + var baseTime = new DateTime(2026, 7, 20, 9, 0, 0, DateTimeKind.Utc); + + var first = new IngestedFailure + { + MessageType = "AAA.First", + TimeSent = baseTime, + AttemptedAt = baseTime.AddMinutes(1), + TimeOfFailure = baseTime.AddMinutes(1) + }; + + var second = new IngestedFailure + { + MessageType = "ZZZ.Second", + TimeSent = baseTime.AddHours(1), + AttemptedAt = baseTime.AddHours(1).AddMinutes(1), + TimeOfFailure = baseTime.AddHours(1).AddMinutes(1) + }; + + await Insert( + first.ToFailedMessage(), + second.ToFailedMessage(FailedMessageStatus.Archived)); + + var result = await FailedMessageQueryStore.GetFailedMessages(null, null, null, new PagingInfo(), new SortInfo(sort, direction)); + + var expected = direction == "asc" + ? new[] { first.UniqueMessageIdString, second.UniqueMessageIdString } + : new[] { second.UniqueMessageIdString, first.UniqueMessageIdString }; + + Assert.That(Ids(result), Is.EqualTo(expected)); + } + + [Test] + public async Task Falls_back_to_time_sent_for_an_unknown_sort() + { + var baseTime = new DateTime(2026, 7, 20, 9, 0, 0, DateTimeKind.Utc); + var older = new IngestedFailure { TimeSent = baseTime }; + var newer = new IngestedFailure { TimeSent = baseTime.AddHours(1) }; + + await Insert(older, newer); + + var result = await FailedMessageQueryStore.GetFailedMessages(null, null, null, new PagingInfo(), new SortInfo("not_a_field", "desc")); + + Assert.That(Ids(result), Is.EqualTo(new[] { newer.UniqueMessageIdString, older.UniqueMessageIdString })); + } + + [Test] + public async Task Pages_through_the_results() + { + var baseTime = new DateTime(2026, 7, 20, 9, 0, 0, DateTimeKind.Utc); + + var failures = Enumerable.Range(0, 5) + .Select(i => new IngestedFailure { TimeSent = baseTime.AddMinutes(i) }) + .ToArray(); + + await Insert(failures); + + var firstPage = await FailedMessageQueryStore.GetFailedMessages(null, null, null, new PagingInfo(1, 2), new SortInfo("time_sent", "asc")); + var secondPage = await FailedMessageQueryStore.GetFailedMessages(null, null, null, new PagingInfo(2, 2), new SortInfo("time_sent", "asc")); + var lastPage = await FailedMessageQueryStore.GetFailedMessages(null, null, null, new PagingInfo(3, 2), new SortInfo("time_sent", "asc")); + + using (Assert.EnterMultipleScope()) + { + Assert.That(Ids(firstPage), Is.EqualTo(failures.Take(2).Select(f => f.UniqueMessageIdString))); + Assert.That(Ids(secondPage), Is.EqualTo(failures.Skip(2).Take(2).Select(f => f.UniqueMessageIdString))); + Assert.That(Ids(lastPage), Is.EqualTo(failures.Skip(4).Select(f => f.UniqueMessageIdString))); + + // The total is the size of the whole filtered set, not of the page. + Assert.That(firstPage.QueryStats.TotalCount, Is.EqualTo(5)); + } + } + + [Test] + public async Task Reports_stats_matching_the_query() + { + var unresolved = new IngestedFailure(); + var archived = new IngestedFailure(); + + await Insert(unresolved.ToFailedMessage(), archived.ToFailedMessage(FailedMessageStatus.Archived)); + + var query = await FailedMessageQueryStore.GetFailedMessages("unresolved", null, null, new PagingInfo(), new SortInfo()); + var stats = await FailedMessageQueryStore.GetFailedMessagesStats("unresolved", null, null); + + using (Assert.EnterMultipleScope()) + { + Assert.That(stats.TotalCount, Is.EqualTo(1)); + Assert.That(stats.TotalCount, Is.EqualTo(query.QueryStats.TotalCount)); + Assert.That(stats.ETag, Is.EqualTo(query.QueryStats.ETag)); + } + } + + [Test] + public async Task Repeats_the_etag_while_nothing_changes() + { + await Insert(new IngestedFailure()); + + var first = await FailedMessageQueryStore.GetFailedMessagesStats(null, null, null); + var second = await FailedMessageQueryStore.GetFailedMessagesStats(null, null, null); + + Assert.That(second.ETag, Is.EqualTo(first.ETag)); + } + + [Test] + public async Task Changes_the_etag_when_the_set_changes() + { + await Insert(new IngestedFailure()); + + var before = await FailedMessageQueryStore.GetFailedMessagesStats(null, null, null); + + await Insert(new IngestedFailure()); + + var after = await FailedMessageQueryStore.GetFailedMessagesStats(null, null, null); + + Assert.That(after.ETag, Is.Not.EqualTo(before.ETag)); + } + + + [Test] + public async Task Returns_the_stored_message_by_id() + { + var failure = new IngestedFailure(); + + await Insert(failure); + + var message = await FailedMessageQueryStore.GetFailedMessage(failure.UniqueMessageIdString); + + using (Assert.EnterMultipleScope()) + { + Assert.That(message.UniqueMessageId, Is.EqualTo(failure.UniqueMessageIdString)); + Assert.That(message.Status, Is.EqualTo(FailedMessageStatus.Unresolved)); + Assert.That(message.ProcessingAttempts.Last().MessageId, Is.EqualTo(failure.MessageId)); + Assert.That(message.ProcessingAttempts.Last().FailureDetails.AddressOfFailingEndpoint, Is.EqualTo(failure.QueueAddress)); + Assert.That(message.FailureGroups.Select(g => g.Id), Is.EquivalentTo(failure.Groups.Select(g => g.Id))); + } + } + + [Test] + public async Task Reports_the_number_of_attempts_on_the_stored_message() + { + var failure = new IngestedFailure(); + + await Insert(failure.ToFailedMessage(numberOfAttempts: 3)); + + var message = await FailedMessageQueryStore.GetFailedMessage(failure.UniqueMessageIdString); + + using (Assert.EnterMultipleScope()) + { + Assert.That(message.ProcessingAttempts, Has.Count.EqualTo(3)); + Assert.That(message.ProcessingAttempts.Last().MessageId, Is.EqualTo(failure.MessageId)); + } + } + + [Test] + public async Task Returns_nothing_for_an_unknown_id() + { + var view = await FailedMessageQueryStore.GetLatestFailedMessageView(Guid.NewGuid().ToString()); + var message = await FailedMessageQueryStore.GetFailedMessage(Guid.NewGuid().ToString()); + + using (Assert.EnterMultipleScope()) + { + Assert.That(view, Is.Null); + Assert.That(message, Is.Null); + } + } + + [Test] + public async Task Returns_the_stored_messages_by_ids() + { + var wanted = new IngestedFailure(); + var alsoWanted = new IngestedFailure(); + var other = new IngestedFailure(); + + await Insert(wanted, alsoWanted, other); + + var messages = await FailedMessageQueryStore.GetFailedMessagesByIds([wanted.UniqueMessageId, alsoWanted.UniqueMessageId]); + + Assert.That( + messages.Select(m => m.UniqueMessageId), + Is.EquivalentTo(new[] { wanted.UniqueMessageIdString, alsoWanted.UniqueMessageIdString })); + } + + [Test] + public async Task Skips_ids_that_are_not_stored() + { + var stored = new IngestedFailure(); + + await Insert(stored); + + var messages = await FailedMessageQueryStore.GetFailedMessagesByIds([stored.UniqueMessageId, Guid.NewGuid()]); + + Assert.That(messages.Select(m => m.UniqueMessageId), Is.EqualTo(new[] { stored.UniqueMessageIdString })); + } + + 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(); + } + + static IEnumerable Ids(QueryResult> result) => + result.Results.Select(view => view.Id); + + static string Range(DateTime from, DateTime to) => $"{from:O}...{to:O}"; +} diff --git a/src/ServiceControl.Persistence.Tests/MessageFailures/FailedMessageSummaryTests.cs b/src/ServiceControl.Persistence.Tests/MessageFailures/FailedMessageSummaryTests.cs new file mode 100644 index 0000000000..1b7b7b6e9e --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/MessageFailures/FailedMessageSummaryTests.cs @@ -0,0 +1,74 @@ +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.Operations; + +class FailedMessageSummaryTests : PersistenceTestBase +{ + [Test] + public async Task Summarises_by_endpoint_host_and_message_type() + { + var sales = new IngestedFailure(); + var alsoSales = new IngestedFailure(); + var billing = new IngestedFailure + { + MessageType = "MyCompany.Billing.InvoiceRaised", + ReceivingEndpoint = new EndpointDetails { Name = "Billing", Host = "BillingHost", HostId = Guid.NewGuid() } + }; + + await Insert(sales, alsoSales, billing); + + var summary = await FailedMessageQueryStore.GetFailedMessagesSummary(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(Counts(summary, "Endpoints"), Is.EqualTo(new Dictionary { ["Sales"] = 2, ["Billing"] = 1 })); + Assert.That(Counts(summary, "Hosts"), Is.EqualTo(new Dictionary { ["ReceiverHost"] = 2, ["BillingHost"] = 1 })); + Assert.That(Counts(summary, "Message types"), Is.EqualTo(new Dictionary + { + ["MyCompany.Sales.OrderPlaced"] = 2, + ["MyCompany.Billing.InvoiceRaised"] = 1 + })); + } + } + + [Test] + public async Task Counts_only_unresolved_messages() + { + await Insert( + new IngestedFailure().ToFailedMessage(), + new IngestedFailure().ToFailedMessage(FailedMessageStatus.Archived), + new IngestedFailure().ToFailedMessage(FailedMessageStatus.Resolved)); + + var summary = await FailedMessageQueryStore.GetFailedMessagesSummary(); + + Assert.That(Counts(summary, "Endpoints"), Is.EqualTo(new Dictionary { ["Sales"] = 1 })); + } + + static Dictionary Counts(IDictionary summary, string key) + { + Assert.That(summary, Contains.Key(key)); + + return ((IDictionary)summary[key]) + .ToDictionary(entry => entry.Key, entry => Convert.ToInt32(entry.Value)); + } + + 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(); + } +} diff --git a/src/ServiceControl.Persistence/IFailedMessageQueryDataStore.cs b/src/ServiceControl.Persistence/IFailedMessageQueryDataStore.cs index 66b7d8ed4a..c299dcf10b 100644 --- a/src/ServiceControl.Persistence/IFailedMessageQueryDataStore.cs +++ b/src/ServiceControl.Persistence/IFailedMessageQueryDataStore.cs @@ -1,3 +1,4 @@ +#nullable enable namespace ServiceControl.Persistence { using System; @@ -9,12 +10,13 @@ namespace ServiceControl.Persistence public interface IFailedMessageQueryDataStore { - Task>> GetFailedMessages(string status, string modified, string queueAddress, PagingInfo pagingInfo, SortInfo sortInfo); - Task GetFailedMessagesStats(string status, string modified, string queueAddress); - Task>> GetFailedMessagesByEndpoint(string status, string endpointName, string modified, PagingInfo pagingInfo, SortInfo sortInfo); + Task>> GetFailedMessages(string? status, string? modified, string? queueAddress, PagingInfo pagingInfo, SortInfo sortInfo); + Task GetFailedMessagesStats(string? status, string? modified, string? queueAddress); + Task>> GetFailedMessagesByEndpoint(string? status, string endpointName, string? modified, PagingInfo pagingInfo, SortInfo sortInfo); Task> GetFailedMessagesSummary(); - Task GetLatestFailedMessageView(string failedMessageId); - Task GetFailedMessage(string failedMessageId); + Task GetLatestFailedMessageView(string failedMessageId); + Task GetFailedMessage(string failedMessageId); + /// Ids with no stored failed message are skipped, so the result can be shorter than . Task GetFailedMessagesByIds(Guid[] ids); } } diff --git a/src/ServiceControl.UnitTests/MessageFailures/AsyncRangeAndQueueAuditTests.cs b/src/ServiceControl.UnitTests/MessageFailures/AsyncRangeAndQueueAuditTests.cs index e1bbbc56db..6e2784070a 100644 --- a/src/ServiceControl.UnitTests/MessageFailures/AsyncRangeAndQueueAuditTests.cs +++ b/src/ServiceControl.UnitTests/MessageFailures/AsyncRangeAndQueueAuditTests.cs @@ -163,15 +163,15 @@ internal sealed class StubErrorMessageDataStore : IFailedMessageQueryDataStore, public Task RemoveFailedMessageRetry(string uniqueMessageId) => Task.CompletedTask; public Task UnArchiveMessagesByRange(DateTime from, DateTime to) => Task.FromResult(UnArchiveByRangeResult); public Task UnArchiveMessages(IEnumerable failedMessageIds) => Task.FromResult(UnArchiveMessagesResult); - public Task GetFailedMessage(string failedMessageId) => Task.FromResult(ErrorByResult); + public Task GetFailedMessage(string failedMessageId) => Task.FromResult(ErrorByResult); public Task MarkAsArchived(string failedMessageId) => Task.CompletedTask; public Task GetFailedMessagesByIds(Guid[] ids) => throw new NotImplementedException(); - public Task>> GetFailedMessages(string status, string modified, string queueAddress, PagingInfo pagingInfo, SortInfo sortInfo) => throw new NotImplementedException(); - public Task GetFailedMessagesStats(string status, string modified, string queueAddress) => throw new NotImplementedException(); - public Task>> GetFailedMessagesByEndpoint(string status, string endpointName, string modified, PagingInfo pagingInfo, SortInfo sortInfo) => throw new NotImplementedException(); + public Task>> GetFailedMessages(string? status, string? modified, string? queueAddress, PagingInfo pagingInfo, SortInfo sortInfo) => throw new NotImplementedException(); + public Task GetFailedMessagesStats(string? status, string? modified, string? queueAddress) => throw new NotImplementedException(); + public Task>> GetFailedMessagesByEndpoint(string? status, string endpointName, string? modified, PagingInfo pagingInfo, SortInfo sortInfo) => throw new NotImplementedException(); public Task> GetFailedMessagesSummary() => throw new NotImplementedException(); - public Task GetLatestFailedMessageView(string failedMessageId) => throw new NotImplementedException(); + public Task GetLatestFailedMessageView(string failedMessageId) => throw new NotImplementedException(); public Task MarkAsResolved(string failedMessageId) => throw new NotImplementedException(); public Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func processCallback) => throw new NotImplementedException(); public Task RevertRetry(string messageUniqueId) => throw new NotImplementedException(); diff --git a/src/ServiceControl.UnitTests/MessageFailures/EditFailedMessagesControllerAuditTests.cs b/src/ServiceControl.UnitTests/MessageFailures/EditFailedMessagesControllerAuditTests.cs index e03faea222..76bd5f3056 100644 --- a/src/ServiceControl.UnitTests/MessageFailures/EditFailedMessagesControllerAuditTests.cs +++ b/src/ServiceControl.UnitTests/MessageFailures/EditFailedMessagesControllerAuditTests.cs @@ -53,7 +53,7 @@ public void Dispose() } public Task SaveChanges() => Task.CompletedTask; - public Task GetFailedMessage(string failedMessageId) => Task.FromResult(null!); + public Task GetFailedMessage(string failedMessageId) => Task.FromResult(null); public Task GetCurrentEditingRequestId(string failedMessageId) => Task.FromResult(CurrentEditingRequestId); public Task SetCurrentEditingRequestId(string editingMessageId) => Task.CompletedTask; public Task SetFailedMessageAsResolved() => Task.CompletedTask; @@ -65,13 +65,13 @@ sealed class StubErrorMessageDataStore : IFailedMessageQueryDataStore, IEditFail public FakeEditFailedMessagesManager EditManager { get; } = new(); public Task CreateEditFailedMessageManager() => Task.FromResult(EditManager); - public Task GetFailedMessage(string failedMessageId) => Task.FromResult(ErrorByResult!); + public Task GetFailedMessage(string failedMessageId) => Task.FromResult(ErrorByResult); public Task GetFailedMessagesByIds(Guid[] ids) => throw new NotImplementedException(); - public Task>> GetFailedMessages(string status, string modified, string queueAddress, PagingInfo pagingInfo, SortInfo sortInfo) => throw new NotImplementedException(); - public Task GetFailedMessagesStats(string status, string modified, string queueAddress) => throw new NotImplementedException(); - public Task>> GetFailedMessagesByEndpoint(string status, string endpointName, string modified, PagingInfo pagingInfo, SortInfo sortInfo) => throw new NotImplementedException(); + public Task>> GetFailedMessages(string? status, string? modified, string? queueAddress, PagingInfo pagingInfo, SortInfo sortInfo) => throw new NotImplementedException(); + public Task GetFailedMessagesStats(string? status, string? modified, string? queueAddress) => throw new NotImplementedException(); + public Task>> GetFailedMessagesByEndpoint(string? status, string endpointName, string? modified, PagingInfo pagingInfo, SortInfo sortInfo) => throw new NotImplementedException(); public Task> GetFailedMessagesSummary() => throw new NotImplementedException(); - public Task GetLatestFailedMessageView(string failedMessageId) => throw new NotImplementedException(); + public Task GetLatestFailedMessageView(string failedMessageId) => throw new NotImplementedException(); } }