diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260728233024_AddFailedErrorImports.Designer.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260728233024_AddFailedErrorImports.Designer.cs new file mode 100644 index 0000000000..df07d97e7a --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260728233024_AddFailedErrorImports.Designer.cs @@ -0,0 +1,359 @@ +// +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("20260728233024_AddFailedErrorImports")] + partial class AddFailedErrorImports + { + /// + 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("QueueAddress") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("queue_address"); + + 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("QueueAddress") + .HasDatabaseName("ix_failed_messages_queue_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.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/20260728233024_AddFailedErrorImports.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260728233024_AddFailedErrorImports.cs new file mode 100644 index 0000000000..b0cbbbdaa6 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260728233024_AddFailedErrorImports.cs @@ -0,0 +1,44 @@ +using System; +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace ServiceControl.Persistence.EFCore.PostgreSql.Migrations +{ + /// + public partial class AddFailedErrorImports : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.CreateTable( + name: "failed_error_imports", + columns: table => new + { + unique_message_id = table.Column(type: "uuid", nullable: false), + failed_at = table.Column(type: "timestamp with time zone", nullable: false), + message_id = table.Column(type: "character varying(450)", maxLength: 450, nullable: false), + headers_json = table.Column(type: "text", nullable: false), + body = table.Column(type: "bytea", nullable: false), + body_stored_externally = table.Column(type: "boolean", nullable: false), + exception_info = table.Column(type: "text", nullable: false) + }, + constraints: table => + { + table.PrimaryKey("pk_failed_error_imports", x => x.unique_message_id); + }); + + migrationBuilder.CreateIndex( + name: "ix_failed_error_imports_failed_at", + table: "failed_error_imports", + column: "failed_at"); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropTable( + name: "failed_error_imports"); + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs index 1bbf286a9f..5887803530 100644 --- a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs @@ -39,6 +39,50 @@ protected override void BuildModel(ModelBuilder modelBuilder) 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") diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260728233017_AddFailedErrorImports.Designer.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260728233017_AddFailedErrorImports.Designer.cs new file mode 100644 index 0000000000..518b0a56eb --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260728233017_AddFailedErrorImports.Designer.cs @@ -0,0 +1,292 @@ +// +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("20260728233017_AddFailedErrorImports")] + partial class AddFailedErrorImports + { + /// + 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("QueueAddress") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + 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("QueueAddress"); + + 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.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/20260728233017_AddFailedErrorImports.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260728233017_AddFailedErrorImports.cs new file mode 100644 index 0000000000..82ff157a93 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260728233017_AddFailedErrorImports.cs @@ -0,0 +1,44 @@ +using System; +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace ServiceControl.Persistence.EFCore.SqlServer.Migrations +{ + /// + public partial class AddFailedErrorImports : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.CreateTable( + name: "FailedErrorImports", + columns: table => new + { + UniqueMessageId = table.Column(type: "uniqueidentifier", nullable: false), + FailedAt = table.Column(type: "datetime2", nullable: false), + MessageId = table.Column(type: "nvarchar(450)", maxLength: 450, nullable: false), + HeadersJson = table.Column(type: "nvarchar(max)", nullable: false), + Body = table.Column(type: "varbinary(max)", nullable: false), + BodyStoredExternally = table.Column(type: "bit", nullable: false), + ExceptionInfo = table.Column(type: "nvarchar(max)", nullable: false) + }, + constraints: table => + { + table.PrimaryKey("PK_FailedErrorImports", x => x.UniqueMessageId); + }); + + migrationBuilder.CreateIndex( + name: "IX_FailedErrorImports_FailedAt", + table: "FailedErrorImports", + column: "FailedAt"); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropTable( + name: "FailedErrorImports"); + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs index 0a2cd26beb..e7082075b4 100644 --- a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs @@ -33,7 +33,42 @@ protected override void BuildModel(ModelBuilder modelBuilder) b.HasKey("Name"); - b.ToTable("EndpointSettings", (string)null); + 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 => diff --git a/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs b/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs index d0639f3c0a..2ead0819d0 100644 --- a/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs +++ b/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs @@ -60,6 +60,8 @@ protected static void RegisterDataStores(IServiceCollection services, EFPersiste // Settings are registered under their concrete type so each store resolves only what it can act on. static void RegisterBodyStorage(IServiceCollection services, EFPersisterSettings settings) { + services.AddSingleton(settings.BodyStorage); + switch (settings.BodyStorage) { case FileSystemBodyStorageSettings fileSystem: diff --git a/src/ServiceControl.Persistence.EFCore/DbContexts/ServiceControlDbContext.cs b/src/ServiceControl.Persistence.EFCore/DbContexts/ServiceControlDbContext.cs index 437abb4707..045b0f1414 100644 --- a/src/ServiceControl.Persistence.EFCore/DbContexts/ServiceControlDbContext.cs +++ b/src/ServiceControl.Persistence.EFCore/DbContexts/ServiceControlDbContext.cs @@ -11,6 +11,7 @@ public abstract class ServiceControlDbContext(DbContextOptions options) : DbCont public DbSet FailedMessages { get; set; } public DbSet FailedMessageGroups { get; set; } public DbSet FailedMessageRetries { get; set; } + public DbSet FailedErrorImports { get; set; } public DbSet TrialMetadata { get; set; } protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder) @@ -21,6 +22,7 @@ protected override void OnModelCreating(ModelBuilder modelBuilder) base.OnModelCreating(modelBuilder); modelBuilder.ApplyConfiguration(new EndpointSettingsConfiguration()); + modelBuilder.ApplyConfiguration(new FailedErrorImportConfiguration()); modelBuilder.ApplyConfiguration(new FailedMessageConfiguration()); modelBuilder.ApplyConfiguration(new FailedMessageGroupConfiguration()); modelBuilder.ApplyConfiguration(new FailedMessageRetryConfiguration()); diff --git a/src/ServiceControl.Persistence.EFCore/Entities/FailedErrorImportEntity.cs b/src/ServiceControl.Persistence.EFCore/Entities/FailedErrorImportEntity.cs new file mode 100644 index 0000000000..529b75e894 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Entities/FailedErrorImportEntity.cs @@ -0,0 +1,22 @@ +namespace ServiceControl.Persistence.EFCore.Entities; + +public class FailedErrorImportEntity +{ + public Guid UniqueMessageId { get; set; } + + public DateTime FailedAt { get; set; } + + public required string MessageId { get; set; } + + public required string HeadersJson { get; set; } + + // Holds the inline body, or an empty array when the body was spilled to external storage. + // BodyStoredExternally, not the contents here, decides where the body lives. + public required byte[] Body { get; set; } + + public bool BodyStoredExternally { get; set; } + + public required string ExceptionInfo { get; set; } + + public static string ExternalBodyId(Guid uniqueMessageId) => $"failedimport-{uniqueMessageId}"; +} diff --git a/src/ServiceControl.Persistence.EFCore/EntityConfigurations/FailedErrorImportConfiguration.cs b/src/ServiceControl.Persistence.EFCore/EntityConfigurations/FailedErrorImportConfiguration.cs new file mode 100644 index 0000000000..44c46981cc --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/EntityConfigurations/FailedErrorImportConfiguration.cs @@ -0,0 +1,23 @@ +namespace ServiceControl.Persistence.EFCore.EntityConfigurations; + +using Entities; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Metadata.Builders; + +class FailedErrorImportConfiguration : IEntityTypeConfiguration +{ + public void Configure(EntityTypeBuilder builder) + { + builder.HasKey(e => e.UniqueMessageId); + builder.Property(e => e.UniqueMessageId).ValueGeneratedNever(); + + builder.Property(e => e.FailedAt).IsRequired(); + builder.Property(e => e.MessageId).IsRequired().HasMaxLength(ColumnLengths.ShortTextLength); + builder.Property(e => e.HeadersJson).IsRequired(); + builder.Property(e => e.Body).IsRequired(); + builder.Property(e => e.BodyStoredExternally).IsRequired(); + builder.Property(e => e.ExceptionInfo).IsRequired(); + + builder.HasIndex(e => e.FailedAt); + } +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/ErrorMessagesDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/ErrorMessagesDataStore.cs index 692ce2b0c9..0b0930a5bf 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/ErrorMessagesDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/ErrorMessagesDataStore.cs @@ -1,17 +1,26 @@ namespace ServiceControl.Persistence.EFCore.Implementation; +using System.Text.Json; using Entities; using Microsoft.Extensions.DependencyInjection; +using NServiceBus; using Persistence.UnitOfWork; using ServiceControl.CompositeViews.Messages; using ServiceControl.EventLog; using ServiceControl.MessageFailures; using ServiceControl.MessageFailures.Api; using ServiceControl.Operations; +using ServiceControl.Persistence.EFCore.Abstractions; +using ServiceControl.Persistence.EFCore.Implementation.UnitOfWork; +using ServiceControl.Persistence.EFCore.Infrastructure; using ServiceControl.Persistence.Infrastructure; using ServiceControl.Recoverability; -public class ErrorMessagesDataStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), IErrorMessageDataStore +public class ErrorMessagesDataStore( + IServiceScopeFactory scopeFactory, + IBodyStoragePersistence bodyStorage, + BodyStorageSettings bodyStorageSettings, + TimeProvider timeProvider) : DataStoreBase(scopeFactory), IErrorMessageDataStore { public Task>> GetAllMessages(PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange? timeSentRange = null) => throw new NotImplementedException(); @@ -35,8 +44,46 @@ public Task FailedMessageMarkAsArchived(string failedMessageId) => public Task FailedMessagesFetch(Guid[] ids) => throw new NotImplementedException(); + // Update-first, then insert. The dedupe key is deterministic, so a repeat failure updates the + // existing row and concurrent writers that both miss it race only on the insert. The loser of + // that race confirms the row is now present (the winner stored the same logical failure) and + // otherwise rethrows, so the caller never treats a message as stored when it is not. public Task StoreFailedErrorImport(FailedErrorImport failure) => - throw new NotImplementedException(); + ExecuteWithDbContext(async dbContext => + { + var uniqueMessageId = FailedErrorImport.DeriveKey(failure.Message.Headers, failure.Message.Id); + var body = failure.Message.Body ?? []; + var storeExternally = body.Length > bodyStorageSettings.MaxBodySizeToStore; + + if (storeExternally) + { + var contentType = failure.Message.Headers.GetValueOrDefault(Headers.ContentType) ?? "application/octet-stream"; + await bodyStorage.WriteBody(FailedErrorImportEntity.ExternalBodyId(uniqueMessageId), body, contentType); + } + + var failedAt = timeProvider.GetUtcNow().UtcDateTime; + var headersJson = JsonSerializer.Serialize(failure.Message.Headers, HeadersJsonContext.Default.DictionaryStringString); + byte[] storedBody = storeExternally ? [] : body; + + await dbContext.UpsertAsync([uniqueMessageId], () => new FailedErrorImportEntity + { + UniqueMessageId = uniqueMessageId, + FailedAt = failedAt, + MessageId = failure.Message.Id, + HeadersJson = headersJson, + Body = storedBody, + BodyStoredExternally = storeExternally, + ExceptionInfo = failure.ExceptionInfo + }, (entity) => + { + entity.FailedAt = failedAt; + entity.MessageId = failure.Message.Id; + entity.HeadersJson = headersJson; + entity.Body = storedBody; + entity.BodyStoredExternally = storeExternally; + entity.ExceptionInfo = failure.ExceptionInfo; + }); + }); public Task CreateEditFailedMessageManager() => throw new NotImplementedException(); diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FailedErrorImportDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FailedErrorImportDataStore.cs index ed2dd126db..26cc3be1a6 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/FailedErrorImportDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FailedErrorImportDataStore.cs @@ -1,12 +1,156 @@ namespace ServiceControl.Persistence.EFCore.Implementation; +using System.IO; +using System.Text.Json; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; using ServiceControl.Operations; +using ServiceControl.Persistence.EFCore.DbContexts; +using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Persistence.EFCore.Implementation.UnitOfWork; +using ServiceControl.Persistence.EFCore.Infrastructure; -public class FailedErrorImportDataStore : IFailedErrorImportDataStore +public class FailedErrorImportDataStore( + IServiceScopeFactory scopeFactory, + IBodyStoragePersistence bodyStorage, + ILogger logger) : DataStoreBase(scopeFactory), IFailedErrorImportDataStore { - public Task ProcessFailedErrorImports(Func processMessage, CancellationToken cancellationToken) => - throw new NotImplementedException(); + const int BatchSize = 100; public Task QueryContainsFailedImports() => - throw new NotImplementedException(); + ExecuteWithDbContext(dbContext => dbContext.FailedErrorImports.AsNoTracking().AnyAsync()); + + // Replays oldest-first. Successful imports delete their row; failures are left in place, so the + // count of failures so far is exactly the offset to the next unseen row. This walks the whole + // set once without retrying a failure within the same run. + public async Task ProcessFailedErrorImports(Func processMessage, CancellationToken cancellationToken) + { + var succeeded = 0; + var failed = 0; + + while (!cancellationToken.IsCancellationRequested) + { + var batch = await ReadBatch(failed, cancellationToken); + + if (batch.Count == 0) + { + break; + } + + foreach (var import in batch) + { + if (cancellationToken.IsCancellationRequested) + { + break; + } + + try + { + var transportMessage = await ToTransportMessage(import, cancellationToken); + + await processMessage(transportMessage); + + await DeleteImport(import, cancellationToken); + + succeeded++; + + logger.LogDebug("Successfully re-imported failed error message {MessageId}", import.MessageId); + } + catch (OperationCanceledException e) when (cancellationToken.IsCancellationRequested) + { + logger.LogInformation(e, "Cancelled"); + } + catch (Exception e) + { + logger.LogError(e, "Error while attempting to re-import failed error message {MessageId}", import.MessageId); + failed++; + } + } + + if (batch.Count < BatchSize) + { + break; + } + } + + logger.LogInformation("Done re-importing failed errors. Successfully re-imported {SucceededCount} messages. Failed re-importing {FailedCount} messages", succeeded, failed); + + if (failed > 0) + { + logger.LogWarning("{FailedCount} messages could not be re-imported. This could indicate a problem with the data. Contact Particular support if you need help with recovering the messages", failed); + } + } + + async Task> ReadBatch(int offset, CancellationToken cancellationToken) + { + await using var scope = scopeFactory.CreateAsyncScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + + return await dbContext.FailedErrorImports + .AsNoTracking() + .OrderBy(import => import.FailedAt) + .ThenBy(import => import.UniqueMessageId) + .Skip(offset) + .Take(BatchSize) + .ToListAsync(cancellationToken); + } + + async Task ToTransportMessage(FailedErrorImportEntity import, CancellationToken cancellationToken) + { + var headers = JsonSerializer.Deserialize(import.HeadersJson, HeadersJsonContext.Default.DictionaryStringString) ?? []; + + var body = import.BodyStoredExternally + ? await ReadExternalBody(import.UniqueMessageId, cancellationToken) + : import.Body; + + return new FailedTransportMessage + { + Id = import.MessageId, + Headers = headers, + Body = body + }; + } + + async Task ReadExternalBody(Guid uniqueMessageId, CancellationToken cancellationToken) + { + var bodyId = FailedErrorImportEntity.ExternalBodyId(uniqueMessageId); + var stored = await bodyStorage.ReadBody(bodyId, cancellationToken) + ?? throw new InvalidOperationException($"The body for failed error import {uniqueMessageId} was not found in body storage under {bodyId}."); + + await using var stream = stored.Stream; + using var buffer = new MemoryStream(stored.BodySize); + await stream.CopyToAsync(buffer, cancellationToken); + return buffer.ToArray(); + } + + // The row is removed before its external body: a surviving row with a missing body would replay + // as an empty message, whereas an orphaned body is only a leak. + async Task DeleteImport(FailedErrorImportEntity import, CancellationToken cancellationToken) + { + await using var scope = scopeFactory.CreateAsyncScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + + await dbContext.FailedErrorImports + .Where(row => row.UniqueMessageId == import.UniqueMessageId) + .ExecuteDeleteAsync(cancellationToken); + + if (import.BodyStoredExternally) + { + await DeleteExternalBody(import.UniqueMessageId, cancellationToken); + } + } + + async Task DeleteExternalBody(Guid uniqueMessageId, CancellationToken cancellationToken) + { + try + { + await bodyStorage.DeleteBody(FailedErrorImportEntity.ExternalBodyId(uniqueMessageId), cancellationToken); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + // Re-import must not stall on a missing or unavailable body. + logger.LogWarning(ex, "Could not delete the external body for re-imported failed error {UniqueMessageId}", uniqueMessageId); + } + } } diff --git a/src/ServiceControl.Persistence.Tests/EFCore/FailedErrorImportTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/FailedErrorImportTests.cs new file mode 100644 index 0000000000..54eee84d0b --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/FailedErrorImportTests.cs @@ -0,0 +1,302 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using NServiceBus; +using NUnit.Framework; +using ServiceControl.Operations; +using ServiceControl.Persistence; +using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Persistence.Infrastructure; + +class FailedErrorImportTests : ErrorIngestionTestBase +{ + [Test] + public async Task Stores_and_replays_a_failed_import() + { + var headers = WellFormedHeaders(); + var body = Encoding.UTF8.GetBytes("1"); + + await StoreImport(headers, body, nativeId: "native-1"); + + Assert.That(await FailedImportStore.QueryContainsFailedImports(), Is.True); + + var replayed = await Replay(); + + Assert.That(replayed, Has.Count.EqualTo(1)); + using (Assert.EnterMultipleScope()) + { + Assert.That(replayed[0].Id, Is.EqualTo("native-1")); + Assert.That(replayed[0].Headers, Is.EqualTo(headers)); + Assert.That(replayed[0].Body, Is.EqualTo(body)); + } + + Assert.That(await FailedImportStore.QueryContainsFailedImports(), Is.False); + } + + [Test] + public async Task Round_trips_a_binary_body_with_nul_bytes() + { + var headers = WellFormedHeaders(); + var body = new byte[] { 0x00, 0x01, 0x02, 0x00, 0xFF, 0x00 }; + + await StoreImport(headers, body, nativeId: "native-1"); + + var replayed = await Replay(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(replayed, Has.Count.EqualTo(1)); + Assert.That(replayed[0].Body, Is.EqualTo(body)); + Assert.That(RecordedBodies.Written, Is.Empty); + } + } + + [Test] + public async Task Spills_a_large_body_to_external_storage_and_replays_it() + { + var headers = WellFormedHeaders(); + var body = LargeBody(); + var externalId = FailedErrorImportEntity.ExternalBodyId(FailedErrorImport.DeriveKey(headers, "native-1")); + + await StoreImport(headers, body, nativeId: "native-1"); + + Assert.That(RecordedBodies.Written.Select(written => written.BodyId), Does.Contain(externalId)); + + var replayed = await Replay(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(replayed, Has.Count.EqualTo(1)); + Assert.That(replayed[0].Body, Is.EqualTo(body)); + Assert.That(RecordedBodies.Deleted, Does.Contain(externalId)); + } + + Assert.That(await FailedImportStore.QueryContainsFailedImports(), Is.False); + } + + [Test] + public async Task Repeated_failure_of_the_same_message_keeps_one_row_with_the_latest_details() + { + var headers = WellFormedHeaders(); + + await StoreImport(headers, Encoding.UTF8.GetBytes("first"), "first failure", "native-1"); + await StoreImport(headers, Encoding.UTF8.GetBytes("second"), "second failure", "native-1"); + + var replayed = await Replay(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(replayed, Has.Count.EqualTo(1)); + Assert.That(replayed[0].Body, Is.EqualTo(Encoding.UTF8.GetBytes("second"))); + } + } + + [Test] + public async Task Stores_and_replays_a_message_with_no_derivable_endpoint() + { + var headers = new Dictionary(); + + await StoreImport(headers, Encoding.UTF8.GetBytes("body"), nativeId: "native-1"); + await StoreImport(headers, Encoding.UTF8.GetBytes("body-again"), nativeId: "native-1"); + + var replayed = await Replay(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(replayed, Has.Count.EqualTo(1)); + Assert.That(replayed[0].Id, Is.EqualTo("native-1")); + Assert.That(replayed[0].Body, Is.EqualTo(Encoding.UTF8.GetBytes("body-again"))); + } + } + + [Test] + public async Task A_failing_re_import_is_left_behind_while_the_rest_are_processed() + { + await Store( + Import("native-1", BaseTime), + Import("native-2", BaseTime.AddSeconds(1)), + Import("native-3", BaseTime.AddSeconds(2))); + + var replayed = new List(); + await FailedImportStore.ProcessFailedErrorImports( + message => + { + replayed.Add(message.Id); + return message.Id == "native-2" ? throw new InvalidOperationException("boom") : Task.CompletedTask; + }, + TestContext.CurrentContext.CancellationToken); + + using (Assert.EnterMultipleScope()) + { + Assert.That(replayed, Is.EqualTo(new[] { "native-1", "native-2", "native-3" })); + Assert.That(await FailedImportStore.QueryContainsFailedImports(), Is.True); + } + + var secondRun = new List(); + await FailedImportStore.ProcessFailedErrorImports( + message => { secondRun.Add(message.Id); return Task.CompletedTask; }, + TestContext.CurrentContext.CancellationToken); + + Assert.That(secondRun, Is.EqualTo(new[] { "native-2" })); + } + + [Test] + public async Task Replays_across_multiple_pages() + { + var imports = Enumerable.Range(0, 250) + .Select(i => Import($"native-{i:D4}", BaseTime.AddSeconds(i))) + .ToArray(); + + await Store(imports); + + var replayed = await Replay(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(replayed, Has.Count.EqualTo(250)); + Assert.That(await FailedImportStore.QueryContainsFailedImports(), Is.False); + } + } + + [Test] + public async Task Stops_replaying_when_cancelled() + { + await Store( + Import("native-1", BaseTime), + Import("native-2", BaseTime.AddSeconds(1)), + Import("native-3", BaseTime.AddSeconds(2))); + + using var cts = new CancellationTokenSource(); + var replayed = new List(); + + await FailedImportStore.ProcessFailedErrorImports( + message => + { + replayed.Add(message.Id); + cts.Cancel(); + return Task.CompletedTask; + }, + cts.Token); + + using (Assert.EnterMultipleScope()) + { + Assert.That(replayed, Has.Count.EqualTo(1)); + Assert.That(await FailedImportStore.QueryContainsFailedImports(), Is.True); + } + } + + [Test] + public async Task Retention_sweep_does_not_touch_failed_imports() + { + await Store(Import("native-1", new DateTime(2020, 1, 1, 0, 0, 0, DateTimeKind.Utc))); + + await RunRetentionSweep(); + + Assert.That(await FailedImportStore.QueryContainsFailedImports(), Is.True); + } + + [Test] + public async Task A_failing_external_body_delete_does_not_fail_the_re_import() + { + var headers = WellFormedHeaders(); + var body = LargeBody(); + var externalId = FailedErrorImportEntity.ExternalBodyId(FailedErrorImport.DeriveKey(headers, "native-1")); + + await StoreImport(headers, body, nativeId: "native-1"); + RecordedBodies.FailDeleteFor.Add(externalId); + + var replayed = await Replay(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(replayed, Has.Count.EqualTo(1)); + Assert.That(replayed[0].Body, Is.EqualTo(body)); + Assert.That(await FailedImportStore.QueryContainsFailedImports(), Is.False); + } + } + + [Test] + public async Task A_missing_external_body_fails_the_re_import_without_blocking_the_others() + { + var headers = WellFormedHeaders(); + var externalId = FailedErrorImportEntity.ExternalBodyId(FailedErrorImport.DeriveKey(headers, "native-1")); + + var otherHeaders = WellFormedHeaders(); + otherHeaders[Headers.MessageId] = "m2"; + + await StoreImport(headers, LargeBody(), nativeId: "native-1"); + await StoreImport(otherHeaders, Encoding.UTF8.GetBytes("intact"), nativeId: "native-2"); + + RecordedBodies.Evict(externalId); + + var replayed = await Replay(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(replayed.Select(message => message.Id), Is.EqualTo(new[] { "native-2" })); + Assert.That(await FailedImportStore.QueryContainsFailedImports(), Is.True); + } + + var secondRun = await Replay(); + + Assert.That(secondRun, Is.Empty, "the row with the missing body is retried and fails again"); + } + + IFailedErrorImportDataStore FailedImportStore => ServiceProvider.GetRequiredService(); + + Task StoreImport(Dictionary headers, byte[] body, string exceptionInfo = "boom", string nativeId = null) + { + nativeId ??= Guid.NewGuid().ToString(); + + return ErrorStore.StoreFailedErrorImport(new FailedErrorImport + { + Id = FailedErrorImport.DeriveKey(headers, nativeId).ToString(), + Message = new FailedTransportMessage { Id = nativeId, Headers = headers, Body = body }, + ExceptionInfo = exceptionInfo + }); + } + + async Task> Replay() + { + var replayed = new List(); + + await FailedImportStore.ProcessFailedErrorImports( + message => { replayed.Add(message); return Task.CompletedTask; }, + TestContext.CurrentContext.CancellationToken); + + return replayed; + } + + byte[] LargeBody() + { + var body = new byte[EFSettings.BodyStorage.MaxBodySizeToStore + 1]; + Random.Shared.NextBytes(body); + return body; + } + + static FailedErrorImportEntity Import(string nativeId, DateTime failedAt) => new() + { + UniqueMessageId = DeterministicGuid.MakeId(nativeId), + FailedAt = failedAt, + MessageId = nativeId, + HeadersJson = "{}", + Body = Encoding.UTF8.GetBytes(nativeId), + BodyStoredExternally = false, + ExceptionInfo = "boom" + }; + + static Dictionary WellFormedHeaders() => new() + { + [Headers.MessageId] = "m1", + [Headers.ProcessingEndpoint] = "Sales", + [Headers.ContentType] = "text/xml" + }; + + static readonly DateTime BaseTime = new(2026, 7, 22, 10, 0, 0, DateTimeKind.Utc); +} diff --git a/src/ServiceControl.Persistence.Tests/EFCore/InMemoryBodyStoragePersistence.cs b/src/ServiceControl.Persistence.Tests/EFCore/InMemoryBodyStoragePersistence.cs index f3b2c7c0cc..4190e37f58 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/InMemoryBodyStoragePersistence.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/InMemoryBodyStoragePersistence.cs @@ -71,6 +71,14 @@ public Task ReadBody(string bodyId, CancellationToken can return Task.FromResult(result); } + public void Evict(string bodyId) + { + lock (gate) + { + store.Remove(bodyId); + } + } + public Task DeleteBody(string bodyId, CancellationToken cancellationToken = default) { if (FailDeleteFor.Contains(bodyId))