From f80dc2654bec45153b62ff0bc2bb6eab3fcb0bcd Mon Sep 17 00:00:00 2001 From: John Simons Date: Mon, 3 Aug 2026 14:25:19 +1000 Subject: [PATCH] Implement retry batch creation Adds RetryBatches and RetryBatchNowForwarding, and swaps the string RetryId on FailedMessageRetries for a Guid RetryBatchId plus StageAttempts. Membership is not stored twice: a message belongs to at most one batch, so RetryBatchId is the membership and FailureRetries is a query over it. Claiming is one insert if absent through the provider dialect, so there is no conflict to recover from in code. Renames the contract to IRetryBatchStore, away from document vocabulary. --- ...20260803021015_AddRetryBatches.Designer.cs | 568 ++++++++++++++++++ .../20260803021015_AddRetryBatches.cs | 109 ++++ ...SqlServiceControlDbContextModelSnapshot.cs | 108 +++- .../PostgreSqlIngestionSqlDialect.cs | 17 + ...20260803021009_AddRetryBatches.Designer.cs | 459 ++++++++++++++ .../20260803021009_AddRetryBatches.cs | 109 ++++ ...verServiceControlDbContextModelSnapshot.cs | 84 ++- .../SqlServerIngestionSqlDialect.cs | 20 + .../DbContexts/ServiceControlDbContext.cs | 4 + .../Entities/FailedMessageRetryEntity.cs | 4 +- .../Entities/RetryBatchEntity.cs | 36 ++ .../Entities/RetryBatchNowForwardingEntity.cs | 10 + .../FailedMessageRetryConfiguration.cs | 4 +- .../RetryBatchConfiguration.cs | 27 + .../RetryBatchNowForwardingConfiguration.cs | 15 + .../Implementation/RetryBatchMapper.cs | 27 + .../Implementation/RetryBatchStore.cs | 209 ++++++- .../Infrastructure/IIngestionSqlDialect.cs | 6 + .../EFCore/ErrorIngestionTests.cs | 2 +- .../EFCore/RetryBatchStoreTests.cs | 248 ++++++++ 20 files changed, 2042 insertions(+), 24 deletions(-) create mode 100644 src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260803021015_AddRetryBatches.Designer.cs create mode 100644 src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260803021015_AddRetryBatches.cs create mode 100644 src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260803021009_AddRetryBatches.Designer.cs create mode 100644 src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260803021009_AddRetryBatches.cs create mode 100644 src/ServiceControl.Persistence.EFCore/Entities/RetryBatchEntity.cs create mode 100644 src/ServiceControl.Persistence.EFCore/Entities/RetryBatchNowForwardingEntity.cs create mode 100644 src/ServiceControl.Persistence.EFCore/EntityConfigurations/RetryBatchConfiguration.cs create mode 100644 src/ServiceControl.Persistence.EFCore/EntityConfigurations/RetryBatchNowForwardingConfiguration.cs create mode 100644 src/ServiceControl.Persistence.EFCore/Implementation/RetryBatchMapper.cs create mode 100644 src/ServiceControl.Persistence.Tests/EFCore/RetryBatchStoreTests.cs diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260803021015_AddRetryBatches.Designer.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260803021015_AddRetryBatches.Designer.cs new file mode 100644 index 0000000000..964f060ceb --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260803021015_AddRetryBatches.Designer.cs @@ -0,0 +1,568 @@ +// +using System; +using System.Collections.Generic; +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("20260803021015_AddRetryBatches")] + partial class AddRetryBatches + { + /// + 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.EventLogItemEntity", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("bigint") + .HasColumnName("id"); + + NpgsqlPropertyBuilderExtensions.UseIdentityByDefaultColumn(b.Property("Id")); + + b.Property("Category") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("category"); + + b.Property("Description") + .IsRequired() + .HasColumnType("text") + .HasColumnName("description"); + + b.Property("EventType") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("event_type"); + + b.Property("RaisedAt") + .HasColumnType("timestamp with time zone") + .HasColumnName("raised_at"); + + b.PrimitiveCollection>("RelatedTo") + .IsRequired() + .HasColumnType("text[]") + .HasColumnName("related_to"); + + b.Property("Severity") + .HasColumnType("integer") + .HasColumnName("severity"); + + b.HasKey("Id") + .HasName("pk_event_log_items"); + + b.HasIndex("RaisedAt", "Id") + .IsDescending() + .HasDatabaseName("ix_event_log_items_raised_at_id"); + + b.ToTable("EventLogItems", (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") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("failing_endpoint_address"); + + b.Property("FirstTimeOfFailure") + .HasColumnType("timestamp with time zone") + .HasColumnName("first_time_of_failure"); + + b.Property("HeadersJson") + .IsRequired() + .HasColumnType("text") + .HasColumnName("headers_json"); + + b.Property("IsSystemMessage") + .HasColumnType("boolean") + .HasColumnName("is_system_message"); + + b.Property("LastAttemptedAt") + .HasColumnType("timestamp with time zone") + .HasColumnName("last_attempted_at"); + + b.Property("LastModified") + .HasColumnType("timestamp with time zone") + .HasColumnName("last_modified"); + + b.Property("LastTimeOfFailure") + .HasColumnType("timestamp with time zone") + .HasColumnName("last_time_of_failure"); + + b.Property("MessageId") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("message_id"); + + b.Property("MessageType") + .HasColumnType("text") + .HasColumnName("message_type"); + + b.Property("NumberOfProcessingAttempts") + .HasColumnType("integer") + .HasColumnName("number_of_processing_attempts"); + + b.Property("ReceivingEndpointHost") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("receiving_endpoint_host"); + + b.Property("ReceivingEndpointHostId") + .HasColumnType("uuid") + .HasColumnName("receiving_endpoint_host_id"); + + b.Property("ReceivingEndpointName") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("receiving_endpoint_name"); + + b.Property("SendingEndpointHost") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("sending_endpoint_host"); + + b.Property("SendingEndpointHostId") + .HasColumnType("uuid") + .HasColumnName("sending_endpoint_host_id"); + + b.Property("SendingEndpointName") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("sending_endpoint_name"); + + b.Property("Status") + .HasColumnType("integer") + .HasColumnName("status"); + + b.Property("StatusChangedAt") + .HasColumnType("timestamp with time zone") + .HasColumnName("status_changed_at"); + + b.Property("TimeSent") + .HasColumnType("timestamp with time zone") + .HasColumnName("time_sent"); + + b.HasKey("UniqueMessageId") + .HasName("pk_failed_messages"); + + b.HasIndex("ConversationId") + .HasDatabaseName("ix_failed_messages_conversation_id"); + + b.HasIndex("FailingEndpointAddress") + .HasDatabaseName("ix_failed_messages_failing_endpoint_address"); + + b.HasIndex("ReceivingEndpointName") + .HasDatabaseName("ix_failed_messages_receiving_endpoint_name"); + + b.HasIndex("StatusChangedAt") + .HasDatabaseName("ix_failed_messages_status_changed_at") + .HasFilter("status IN (2, 4)"); + + b.HasIndex("TimeSent") + .HasDatabaseName("ix_failed_messages_time_sent"); + + b.HasIndex("Status", "LastModified") + .HasDatabaseName("ix_failed_messages_status_last_modified"); + + b.ToTable("failed_messages", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedMessageGroupEntity", b => + { + b.Property("FailedMessageUniqueId") + .HasColumnType("uuid") + .HasColumnName("failed_message_unique_id"); + + b.Property("GroupId") + .HasMaxLength(64) + .HasColumnType("character varying(64)") + .HasColumnName("group_id"); + + b.Property("Title") + .IsRequired() + .HasColumnType("text") + .HasColumnName("title"); + + b.Property("Type") + .IsRequired() + .HasMaxLength(255) + .HasColumnType("character varying(255)") + .HasColumnName("type"); + + b.HasKey("FailedMessageUniqueId", "GroupId") + .HasName("pk_failed_message_groups"); + + b.HasIndex("GroupId") + .HasDatabaseName("ix_failed_message_groups_group_id"); + + b.HasIndex("Type", "GroupId") + .HasDatabaseName("ix_failed_message_groups_type_group_id"); + + b.ToTable("failed_message_groups", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedMessageRetryEntity", b => + { + b.Property("UniqueMessageId") + .HasColumnType("uuid") + .HasColumnName("unique_message_id"); + + b.Property("RetryBatchId") + .HasColumnType("uuid") + .HasColumnName("retry_batch_id"); + + b.Property("StageAttempts") + .HasColumnType("integer") + .HasColumnName("stage_attempts"); + + b.HasKey("UniqueMessageId") + .HasName("pk_failed_message_retries"); + + b.HasIndex("RetryBatchId") + .HasDatabaseName("ix_failed_message_retries_retry_batch_id"); + + b.ToTable("failed_message_retries", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.GroupCommentEntity", b => + { + b.Property("GroupId") + .HasMaxLength(64) + .HasColumnType("character varying(64)") + .HasColumnName("group_id"); + + b.Property("Comment") + .IsRequired() + .HasColumnType("text") + .HasColumnName("comment"); + + b.HasKey("GroupId") + .HasName("pk_group_comments"); + + b.ToTable("group_comments", (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.MessageRedirectEntity", b => + { + b.Property("FromPhysicalAddress") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("from_physical_address"); + + b.Property("LastModified") + .HasColumnType("timestamp with time zone") + .HasColumnName("last_modified"); + + b.Property("ToPhysicalAddress") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("to_physical_address"); + + b.HasKey("FromPhysicalAddress") + .HasName("pk_message_redirects"); + + b.ToTable("message_redirects", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.RetryBatchEntity", b => + { + b.Property("Id") + .HasColumnType("uuid") + .HasColumnName("id"); + + b.Property("Classifier") + .HasColumnType("text") + .HasColumnName("classifier"); + + b.Property("Context") + .HasColumnType("text") + .HasColumnName("context"); + + b.Property("InitialBatchSize") + .HasColumnType("integer") + .HasColumnName("initial_batch_size"); + + b.Property("InitiatedById") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("initiated_by_id"); + + b.Property("InitiatedByName") + .HasColumnType("text") + .HasColumnName("initiated_by_name"); + + b.Property("Last") + .HasColumnType("timestamp with time zone") + .HasColumnName("last"); + + b.Property("OperationId") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("operation_id"); + + b.Property("Originator") + .HasColumnType("text") + .HasColumnName("originator"); + + b.Property("RequestId") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("request_id"); + + b.Property("RetrySessionId") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("retry_session_id"); + + b.Property("RetryType") + .HasColumnType("integer") + .HasColumnName("retry_type"); + + b.Property("StagingId") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("staging_id"); + + b.Property("StartTime") + .HasColumnType("timestamp with time zone") + .HasColumnName("start_time"); + + b.Property("Status") + .HasColumnType("integer") + .HasColumnName("status"); + + b.HasKey("Id") + .HasName("pk_retry_batches"); + + b.HasIndex("Status", "RetrySessionId") + .HasDatabaseName("ix_retry_batches_status_retry_session_id"); + + b.ToTable("retry_batches", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.RetryBatchNowForwardingEntity", b => + { + b.Property("Id") + .HasColumnType("integer") + .HasColumnName("id"); + + b.Property("RetryBatchId") + .HasColumnType("uuid") + .HasColumnName("retry_batch_id"); + + b.HasKey("Id") + .HasName("pk_retry_batch_now_forwarding"); + + b.ToTable("retry_batch_now_forwarding", (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/20260803021015_AddRetryBatches.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260803021015_AddRetryBatches.cs new file mode 100644 index 0000000000..13fcc9c8ae --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260803021015_AddRetryBatches.cs @@ -0,0 +1,109 @@ +using System; +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace ServiceControl.Persistence.EFCore.PostgreSql.Migrations +{ + /// + public partial class AddRetryBatches : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropColumn( + name: "retry_id", + table: "failed_message_retries"); + + migrationBuilder.AddColumn( + name: "retry_batch_id", + table: "failed_message_retries", + type: "uuid", + nullable: false, + defaultValue: new Guid("00000000-0000-0000-0000-000000000000")); + + migrationBuilder.AddColumn( + name: "stage_attempts", + table: "failed_message_retries", + type: "integer", + nullable: false, + defaultValue: 0); + + migrationBuilder.CreateTable( + name: "retry_batch_now_forwarding", + columns: table => new + { + id = table.Column(type: "integer", nullable: false), + retry_batch_id = table.Column(type: "uuid", nullable: false) + }, + constraints: table => + { + table.PrimaryKey("pk_retry_batch_now_forwarding", x => x.id); + }); + + migrationBuilder.CreateTable( + name: "retry_batches", + columns: table => new + { + id = table.Column(type: "uuid", nullable: false), + status = table.Column(type: "integer", nullable: false), + retry_session_id = table.Column(type: "character varying(450)", maxLength: 450, nullable: false), + request_id = table.Column(type: "character varying(450)", maxLength: 450, nullable: false), + retry_type = table.Column(type: "integer", nullable: false), + initial_batch_size = table.Column(type: "integer", nullable: false), + start_time = table.Column(type: "timestamp with time zone", nullable: false), + last = table.Column(type: "timestamp with time zone", nullable: true), + staging_id = table.Column(type: "character varying(450)", maxLength: 450, nullable: true), + context = table.Column(type: "text", nullable: true), + originator = table.Column(type: "text", nullable: true), + classifier = table.Column(type: "text", nullable: true), + initiated_by_id = table.Column(type: "character varying(450)", maxLength: 450, nullable: true), + initiated_by_name = table.Column(type: "text", nullable: true), + operation_id = table.Column(type: "character varying(450)", maxLength: 450, nullable: true) + }, + constraints: table => + { + table.PrimaryKey("pk_retry_batches", x => x.id); + }); + + migrationBuilder.CreateIndex( + name: "ix_failed_message_retries_retry_batch_id", + table: "failed_message_retries", + column: "retry_batch_id"); + + migrationBuilder.CreateIndex( + name: "ix_retry_batches_status_retry_session_id", + table: "retry_batches", + columns: new[] { "status", "retry_session_id" }); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropTable( + name: "retry_batch_now_forwarding"); + + migrationBuilder.DropTable( + name: "retry_batches"); + + migrationBuilder.DropIndex( + name: "ix_failed_message_retries_retry_batch_id", + table: "failed_message_retries"); + + migrationBuilder.DropColumn( + name: "retry_batch_id", + table: "failed_message_retries"); + + migrationBuilder.DropColumn( + name: "stage_attempts", + table: "failed_message_retries"); + + migrationBuilder.AddColumn( + name: "retry_id", + table: "failed_message_retries", + type: "character varying(450)", + maxLength: 450, + nullable: true); + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs index 6feea9b8a8..8184cac7ba 100644 --- a/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs @@ -371,14 +371,20 @@ protected override void BuildModel(ModelBuilder modelBuilder) .HasColumnType("uuid") .HasColumnName("unique_message_id"); - b.Property("RetryId") - .HasMaxLength(450) - .HasColumnType("character varying(450)") - .HasColumnName("retry_id"); + b.Property("RetryBatchId") + .HasColumnType("uuid") + .HasColumnName("retry_batch_id"); + + b.Property("StageAttempts") + .HasColumnType("integer") + .HasColumnName("stage_attempts"); b.HasKey("UniqueMessageId") .HasName("pk_failed_message_retries"); + b.HasIndex("RetryBatchId") + .HasDatabaseName("ix_failed_message_retries_retry_batch_id"); + b.ToTable("failed_message_retries", (string)null); }); @@ -455,6 +461,100 @@ protected override void BuildModel(ModelBuilder modelBuilder) b.ToTable("message_redirects", (string)null); }); + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.RetryBatchEntity", b => + { + b.Property("Id") + .HasColumnType("uuid") + .HasColumnName("id"); + + b.Property("Classifier") + .HasColumnType("text") + .HasColumnName("classifier"); + + b.Property("Context") + .HasColumnType("text") + .HasColumnName("context"); + + b.Property("InitialBatchSize") + .HasColumnType("integer") + .HasColumnName("initial_batch_size"); + + b.Property("InitiatedById") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("initiated_by_id"); + + b.Property("InitiatedByName") + .HasColumnType("text") + .HasColumnName("initiated_by_name"); + + b.Property("Last") + .HasColumnType("timestamp with time zone") + .HasColumnName("last"); + + b.Property("OperationId") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("operation_id"); + + b.Property("Originator") + .HasColumnType("text") + .HasColumnName("originator"); + + b.Property("RequestId") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("request_id"); + + b.Property("RetrySessionId") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("retry_session_id"); + + b.Property("RetryType") + .HasColumnType("integer") + .HasColumnName("retry_type"); + + b.Property("StagingId") + .HasMaxLength(450) + .HasColumnType("character varying(450)") + .HasColumnName("staging_id"); + + b.Property("StartTime") + .HasColumnType("timestamp with time zone") + .HasColumnName("start_time"); + + b.Property("Status") + .HasColumnType("integer") + .HasColumnName("status"); + + b.HasKey("Id") + .HasName("pk_retry_batches"); + + b.HasIndex("Status", "RetrySessionId") + .HasDatabaseName("ix_retry_batches_status_retry_session_id"); + + b.ToTable("retry_batches", (string)null); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.RetryBatchNowForwardingEntity", b => + { + b.Property("Id") + .HasColumnType("integer") + .HasColumnName("id"); + + b.Property("RetryBatchId") + .HasColumnType("uuid") + .HasColumnName("retry_batch_id"); + + b.HasKey("Id") + .HasName("pk_retry_batch_now_forwarding"); + + b.ToTable("retry_batch_now_forwarding", (string)null); + }); + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.SubscriptionEntity", b => { b.Property("MessageType") diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlIngestionSqlDialect.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlIngestionSqlDialect.cs index 6d57b747f9..3a0bdedafe 100644 --- a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlIngestionSqlDialect.cs +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlIngestionSqlDialect.cs @@ -65,6 +65,23 @@ ON CONFLICT (id) DO NOTHING } } + public async Task InsertMissingRetryClaims(ServiceControlDbContext dbContext, IReadOnlyList rows, CancellationToken cancellationToken) + { + foreach (var chunk in rows.Chunk(MaxRowsPerStatement)) + { + await Execute( + dbContext, + $""" + INSERT INTO failed_message_retries (unique_message_id, retry_batch_id, stage_attempts) + VALUES + {ParameterRows(chunk.Length, 3)} + ON CONFLICT (unique_message_id) DO NOTHING + """, + chunk.Select(retry => new object?[] { retry.UniqueMessageId, retry.RetryBatchId, retry.StageAttempts }), + cancellationToken); + } + } + static async Task Execute(ServiceControlDbContext dbContext, string sql, IEnumerable rows, CancellationToken cancellationToken) { await using var command = dbContext.Database.GetDbConnection().CreateCommand(); diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260803021009_AddRetryBatches.Designer.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260803021009_AddRetryBatches.Designer.cs new file mode 100644 index 0000000000..67a66855a7 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260803021009_AddRetryBatches.Designer.cs @@ -0,0 +1,459 @@ +// +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("20260803021009_AddRetryBatches")] + partial class AddRetryBatches + { + /// + 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.EventLogItemEntity", b => + { + b.Property("Id") + .ValueGeneratedOnAdd() + .HasColumnType("bigint"); + + SqlServerPropertyBuilderExtensions.UseIdentityColumn(b.Property("Id")); + + b.Property("Category") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("Description") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("EventType") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("RaisedAt") + .HasColumnType("datetime2"); + + b.PrimitiveCollection("RelatedTo") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("Severity") + .HasColumnType("int"); + + b.HasKey("Id"); + + b.HasIndex("RaisedAt", "Id") + .IsDescending(); + + b.ToTable("EventLogItems", (string)null); + }); + + 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") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("FirstTimeOfFailure") + .HasColumnType("datetime2"); + + b.Property("HeadersJson") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("IsSystemMessage") + .HasColumnType("bit"); + + b.Property("LastAttemptedAt") + .HasColumnType("datetime2"); + + b.Property("LastModified") + .HasColumnType("datetime2"); + + b.Property("LastTimeOfFailure") + .HasColumnType("datetime2"); + + b.Property("MessageId") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("MessageType") + .HasColumnType("nvarchar(max)"); + + b.Property("NumberOfProcessingAttempts") + .HasColumnType("int"); + + b.Property("ReceivingEndpointHost") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("ReceivingEndpointHostId") + .HasColumnType("uniqueidentifier"); + + b.Property("ReceivingEndpointName") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("SendingEndpointHost") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("SendingEndpointHostId") + .HasColumnType("uniqueidentifier"); + + b.Property("SendingEndpointName") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("Status") + .HasColumnType("int"); + + b.Property("StatusChangedAt") + .HasColumnType("datetime2"); + + b.Property("TimeSent") + .HasColumnType("datetime2"); + + b.HasKey("UniqueMessageId"); + + b.HasIndex("ConversationId"); + + b.HasIndex("FailingEndpointAddress"); + + b.HasIndex("ReceivingEndpointName"); + + b.HasIndex("StatusChangedAt") + .HasFilter("[Status] IN (2, 4)"); + + b.HasIndex("TimeSent"); + + b.HasIndex("Status", "LastModified"); + + b.ToTable("FailedMessages"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedMessageGroupEntity", b => + { + b.Property("FailedMessageUniqueId") + .HasColumnType("uniqueidentifier"); + + b.Property("GroupId") + .HasMaxLength(64) + .HasColumnType("nvarchar(64)"); + + b.Property("Title") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.Property("Type") + .IsRequired() + .HasMaxLength(255) + .HasColumnType("nvarchar(255)"); + + b.HasKey("FailedMessageUniqueId", "GroupId"); + + b.HasIndex("GroupId"); + + b.HasIndex("Type", "GroupId"); + + b.ToTable("FailedMessageGroups"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.FailedMessageRetryEntity", b => + { + b.Property("UniqueMessageId") + .HasColumnType("uniqueidentifier"); + + b.Property("RetryBatchId") + .HasColumnType("uniqueidentifier"); + + b.Property("StageAttempts") + .HasColumnType("int"); + + b.HasKey("UniqueMessageId"); + + b.HasIndex("RetryBatchId"); + + b.ToTable("FailedMessageRetries"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.GroupCommentEntity", b => + { + b.Property("GroupId") + .HasMaxLength(64) + .HasColumnType("nvarchar(64)"); + + b.Property("Comment") + .IsRequired() + .HasColumnType("nvarchar(max)"); + + b.HasKey("GroupId"); + + b.ToTable("GroupComments"); + }); + + 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.MessageRedirectEntity", b => + { + b.Property("FromPhysicalAddress") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("LastModified") + .HasColumnType("datetime2"); + + b.Property("ToPhysicalAddress") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.HasKey("FromPhysicalAddress"); + + b.ToTable("MessageRedirects"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.RetryBatchEntity", b => + { + b.Property("Id") + .HasColumnType("uniqueidentifier"); + + b.Property("Classifier") + .HasColumnType("nvarchar(max)"); + + b.Property("Context") + .HasColumnType("nvarchar(max)"); + + b.Property("InitialBatchSize") + .HasColumnType("int"); + + b.Property("InitiatedById") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("InitiatedByName") + .HasColumnType("nvarchar(max)"); + + b.Property("Last") + .HasColumnType("datetime2"); + + b.Property("OperationId") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("Originator") + .HasColumnType("nvarchar(max)"); + + b.Property("RequestId") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("RetrySessionId") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("RetryType") + .HasColumnType("int"); + + b.Property("StagingId") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("StartTime") + .HasColumnType("datetime2"); + + b.Property("Status") + .HasColumnType("int"); + + b.HasKey("Id"); + + b.HasIndex("Status", "RetrySessionId"); + + b.ToTable("RetryBatches"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.RetryBatchNowForwardingEntity", b => + { + b.Property("Id") + .HasColumnType("int"); + + b.Property("RetryBatchId") + .HasColumnType("uniqueidentifier"); + + b.HasKey("Id"); + + b.ToTable("RetryBatchNowForwarding"); + }); + + 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/20260803021009_AddRetryBatches.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260803021009_AddRetryBatches.cs new file mode 100644 index 0000000000..5952ccc0e3 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/20260803021009_AddRetryBatches.cs @@ -0,0 +1,109 @@ +using System; +using Microsoft.EntityFrameworkCore.Migrations; + +#nullable disable + +namespace ServiceControl.Persistence.EFCore.SqlServer.Migrations +{ + /// + public partial class AddRetryBatches : Migration + { + /// + protected override void Up(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropColumn( + name: "RetryId", + table: "FailedMessageRetries"); + + migrationBuilder.AddColumn( + name: "RetryBatchId", + table: "FailedMessageRetries", + type: "uniqueidentifier", + nullable: false, + defaultValue: new Guid("00000000-0000-0000-0000-000000000000")); + + migrationBuilder.AddColumn( + name: "StageAttempts", + table: "FailedMessageRetries", + type: "int", + nullable: false, + defaultValue: 0); + + migrationBuilder.CreateTable( + name: "RetryBatches", + columns: table => new + { + Id = table.Column(type: "uniqueidentifier", nullable: false), + Status = table.Column(type: "int", nullable: false), + RetrySessionId = table.Column(type: "nvarchar(450)", maxLength: 450, nullable: false), + RequestId = table.Column(type: "nvarchar(450)", maxLength: 450, nullable: false), + RetryType = table.Column(type: "int", nullable: false), + InitialBatchSize = table.Column(type: "int", nullable: false), + StartTime = table.Column(type: "datetime2", nullable: false), + Last = table.Column(type: "datetime2", nullable: true), + StagingId = table.Column(type: "nvarchar(450)", maxLength: 450, nullable: true), + Context = table.Column(type: "nvarchar(max)", nullable: true), + Originator = table.Column(type: "nvarchar(max)", nullable: true), + Classifier = table.Column(type: "nvarchar(max)", nullable: true), + InitiatedById = table.Column(type: "nvarchar(450)", maxLength: 450, nullable: true), + InitiatedByName = table.Column(type: "nvarchar(max)", nullable: true), + OperationId = table.Column(type: "nvarchar(450)", maxLength: 450, nullable: true) + }, + constraints: table => + { + table.PrimaryKey("PK_RetryBatches", x => x.Id); + }); + + migrationBuilder.CreateTable( + name: "RetryBatchNowForwarding", + columns: table => new + { + Id = table.Column(type: "int", nullable: false), + RetryBatchId = table.Column(type: "uniqueidentifier", nullable: false) + }, + constraints: table => + { + table.PrimaryKey("PK_RetryBatchNowForwarding", x => x.Id); + }); + + migrationBuilder.CreateIndex( + name: "IX_FailedMessageRetries_RetryBatchId", + table: "FailedMessageRetries", + column: "RetryBatchId"); + + migrationBuilder.CreateIndex( + name: "IX_RetryBatches_Status_RetrySessionId", + table: "RetryBatches", + columns: new[] { "Status", "RetrySessionId" }); + } + + /// + protected override void Down(MigrationBuilder migrationBuilder) + { + migrationBuilder.DropTable( + name: "RetryBatches"); + + migrationBuilder.DropTable( + name: "RetryBatchNowForwarding"); + + migrationBuilder.DropIndex( + name: "IX_FailedMessageRetries_RetryBatchId", + table: "FailedMessageRetries"); + + migrationBuilder.DropColumn( + name: "RetryBatchId", + table: "FailedMessageRetries"); + + migrationBuilder.DropColumn( + name: "StageAttempts", + table: "FailedMessageRetries"); + + migrationBuilder.AddColumn( + name: "RetryId", + table: "FailedMessageRetries", + type: "nvarchar(450)", + maxLength: 450, + nullable: true); + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs index 08098897fd..e2c68d1c5b 100644 --- a/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/Migrations/SqlServerServiceControlDbContextModelSnapshot.cs @@ -295,12 +295,16 @@ protected override void BuildModel(ModelBuilder modelBuilder) b.Property("UniqueMessageId") .HasColumnType("uniqueidentifier"); - b.Property("RetryId") - .HasMaxLength(450) - .HasColumnType("nvarchar(450)"); + b.Property("RetryBatchId") + .HasColumnType("uniqueidentifier"); + + b.Property("StageAttempts") + .HasColumnType("int"); b.HasKey("UniqueMessageId"); + b.HasIndex("RetryBatchId"); + b.ToTable("FailedMessageRetries"); }); @@ -364,6 +368,80 @@ protected override void BuildModel(ModelBuilder modelBuilder) b.ToTable("MessageRedirects"); }); + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.RetryBatchEntity", b => + { + b.Property("Id") + .HasColumnType("uniqueidentifier"); + + b.Property("Classifier") + .HasColumnType("nvarchar(max)"); + + b.Property("Context") + .HasColumnType("nvarchar(max)"); + + b.Property("InitialBatchSize") + .HasColumnType("int"); + + b.Property("InitiatedById") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("InitiatedByName") + .HasColumnType("nvarchar(max)"); + + b.Property("Last") + .HasColumnType("datetime2"); + + b.Property("OperationId") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("Originator") + .HasColumnType("nvarchar(max)"); + + b.Property("RequestId") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("RetrySessionId") + .IsRequired() + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("RetryType") + .HasColumnType("int"); + + b.Property("StagingId") + .HasMaxLength(450) + .HasColumnType("nvarchar(450)"); + + b.Property("StartTime") + .HasColumnType("datetime2"); + + b.Property("Status") + .HasColumnType("int"); + + b.HasKey("Id"); + + b.HasIndex("Status", "RetrySessionId"); + + b.ToTable("RetryBatches"); + }); + + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.RetryBatchNowForwardingEntity", b => + { + b.Property("Id") + .HasColumnType("int"); + + b.Property("RetryBatchId") + .HasColumnType("uniqueidentifier"); + + b.HasKey("Id"); + + b.ToTable("RetryBatchNowForwarding"); + }); + modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.SubscriptionEntity", b => { b.Property("MessageType") diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerIngestionSqlDialect.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerIngestionSqlDialect.cs index 85f78a253a..de266a73af 100644 --- a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerIngestionSqlDialect.cs +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerIngestionSqlDialect.cs @@ -76,6 +76,26 @@ WHEN NOT MATCHED THEN INSERT ([Id], [Name], [HostId], [Host], [Monitored]) } } + public async Task InsertMissingRetryClaims(ServiceControlDbContext dbContext, IReadOnlyList rows, CancellationToken cancellationToken) + { + foreach (var chunk in rows.Chunk(MaxRowsPerStatement)) + { + await Execute( + dbContext, + $""" + MERGE [FailedMessageRetries] WITH (HOLDLOCK) AS t + USING (VALUES + {ParameterRows(chunk.Length, 3)} + ) AS s ([UniqueMessageId], [RetryBatchId], [StageAttempts]) + ON t.[UniqueMessageId] = s.[UniqueMessageId] + WHEN NOT MATCHED THEN INSERT ([UniqueMessageId], [RetryBatchId], [StageAttempts]) + VALUES (s.[UniqueMessageId], s.[RetryBatchId], s.[StageAttempts]); + """, + chunk.Select(retry => new object?[] { retry.UniqueMessageId, retry.RetryBatchId, retry.StageAttempts }), + cancellationToken); + } + } + static async Task Execute(ServiceControlDbContext dbContext, string sql, IEnumerable rows, CancellationToken cancellationToken) { await using var command = dbContext.Database.GetDbConnection().CreateCommand(); diff --git a/src/ServiceControl.Persistence.EFCore/DbContexts/ServiceControlDbContext.cs b/src/ServiceControl.Persistence.EFCore/DbContexts/ServiceControlDbContext.cs index fc71c45e37..8973e49541 100644 --- a/src/ServiceControl.Persistence.EFCore/DbContexts/ServiceControlDbContext.cs +++ b/src/ServiceControl.Persistence.EFCore/DbContexts/ServiceControlDbContext.cs @@ -13,6 +13,8 @@ public abstract class ServiceControlDbContext(DbContextOptions options) : DbCont public DbSet FailedMessageGroups { get; set; } public DbSet GroupComments { get; set; } public DbSet MessageRedirects { get; set; } + public DbSet RetryBatches { get; set; } + public DbSet RetryBatchNowForwarding { get; set; } public DbSet FailedMessageRetries { get; set; } public DbSet FailedErrorImports { get; set; } public DbSet TrialMetadata { get; set; } @@ -34,6 +36,8 @@ protected override void OnModelCreating(ModelBuilder modelBuilder) modelBuilder.ApplyConfiguration(new FailedMessageRetryConfiguration()); modelBuilder.ApplyConfiguration(new GroupCommentConfiguration()); modelBuilder.ApplyConfiguration(new MessageRedirectConfiguration()); + modelBuilder.ApplyConfiguration(new RetryBatchConfiguration()); + modelBuilder.ApplyConfiguration(new RetryBatchNowForwardingConfiguration()); modelBuilder.ApplyConfiguration(new KnownEndpointConfiguration()); modelBuilder.ApplyConfiguration(new SubscriptionConfiguration()); modelBuilder.ApplyConfiguration(new TrialMetadataConfiguration()); diff --git a/src/ServiceControl.Persistence.EFCore/Entities/FailedMessageRetryEntity.cs b/src/ServiceControl.Persistence.EFCore/Entities/FailedMessageRetryEntity.cs index 5fe0816f2d..ac06233137 100644 --- a/src/ServiceControl.Persistence.EFCore/Entities/FailedMessageRetryEntity.cs +++ b/src/ServiceControl.Persistence.EFCore/Entities/FailedMessageRetryEntity.cs @@ -4,5 +4,7 @@ public class FailedMessageRetryEntity { public Guid UniqueMessageId { get; set; } - public string? RetryId { get; set; } + public Guid RetryBatchId { get; set; } + + public int StageAttempts { get; set; } } diff --git a/src/ServiceControl.Persistence.EFCore/Entities/RetryBatchEntity.cs b/src/ServiceControl.Persistence.EFCore/Entities/RetryBatchEntity.cs new file mode 100644 index 0000000000..582c916d36 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Entities/RetryBatchEntity.cs @@ -0,0 +1,36 @@ +namespace ServiceControl.Persistence.EFCore.Entities; + +using ServiceControl.Persistence; + +public class RetryBatchEntity +{ + public Guid Id { get; set; } + + public RetryBatchStatus Status { get; set; } + + public required string RetrySessionId { get; set; } + + public required string RequestId { get; set; } + + public RetryType RetryType { get; set; } + + public int InitialBatchSize { get; set; } + + public DateTime StartTime { get; set; } + + public DateTime? Last { get; set; } + + public string? StagingId { get; set; } + + public string? Context { get; set; } + + public string? Originator { get; set; } + + public string? Classifier { get; set; } + + public string? InitiatedById { get; set; } + + public string? InitiatedByName { get; set; } + + public string? OperationId { get; set; } +} diff --git a/src/ServiceControl.Persistence.EFCore/Entities/RetryBatchNowForwardingEntity.cs b/src/ServiceControl.Persistence.EFCore/Entities/RetryBatchNowForwardingEntity.cs new file mode 100644 index 0000000000..dc80b2b930 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Entities/RetryBatchNowForwardingEntity.cs @@ -0,0 +1,10 @@ +namespace ServiceControl.Persistence.EFCore.Entities; + +public class RetryBatchNowForwardingEntity +{ + public const int SingleRowId = 1; + + public int Id { get; set; } = SingleRowId; + + public Guid RetryBatchId { get; set; } +} diff --git a/src/ServiceControl.Persistence.EFCore/EntityConfigurations/FailedMessageRetryConfiguration.cs b/src/ServiceControl.Persistence.EFCore/EntityConfigurations/FailedMessageRetryConfiguration.cs index 3cc7007040..e9bd7597f8 100644 --- a/src/ServiceControl.Persistence.EFCore/EntityConfigurations/FailedMessageRetryConfiguration.cs +++ b/src/ServiceControl.Persistence.EFCore/EntityConfigurations/FailedMessageRetryConfiguration.cs @@ -10,6 +10,8 @@ public void Configure(EntityTypeBuilder builder) { builder.HasKey(e => e.UniqueMessageId); builder.Property(e => e.UniqueMessageId).ValueGeneratedNever(); - builder.Property(e => e.RetryId).HasMaxLength(ColumnLengths.ShortTextLength); + builder.Property(e => e.StageAttempts).IsRequired(); + + builder.HasIndex(e => e.RetryBatchId); } } diff --git a/src/ServiceControl.Persistence.EFCore/EntityConfigurations/RetryBatchConfiguration.cs b/src/ServiceControl.Persistence.EFCore/EntityConfigurations/RetryBatchConfiguration.cs new file mode 100644 index 0000000000..3a51b139ff --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/EntityConfigurations/RetryBatchConfiguration.cs @@ -0,0 +1,27 @@ +namespace ServiceControl.Persistence.EFCore.EntityConfigurations; + +using Entities; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Metadata.Builders; + +class RetryBatchConfiguration : IEntityTypeConfiguration +{ + public void Configure(EntityTypeBuilder builder) + { + builder.HasKey(e => e.Id); + builder.Property(e => e.Id).ValueGeneratedNever(); + + builder.Property(e => e.Status).IsRequired(); + builder.Property(e => e.RetryType).IsRequired(); + builder.Property(e => e.InitialBatchSize).IsRequired(); + builder.Property(e => e.StartTime).IsRequired(); + + builder.Property(e => e.RetrySessionId).IsRequired().HasMaxLength(ColumnLengths.ShortTextLength); + builder.Property(e => e.RequestId).IsRequired().HasMaxLength(ColumnLengths.ShortTextLength); + builder.Property(e => e.StagingId).HasMaxLength(ColumnLengths.ShortTextLength); + builder.Property(e => e.OperationId).HasMaxLength(ColumnLengths.ShortTextLength); + builder.Property(e => e.InitiatedById).HasMaxLength(ColumnLengths.ShortTextLength); + + builder.HasIndex(e => new { e.Status, e.RetrySessionId }); + } +} diff --git a/src/ServiceControl.Persistence.EFCore/EntityConfigurations/RetryBatchNowForwardingConfiguration.cs b/src/ServiceControl.Persistence.EFCore/EntityConfigurations/RetryBatchNowForwardingConfiguration.cs new file mode 100644 index 0000000000..eac8ab3791 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/EntityConfigurations/RetryBatchNowForwardingConfiguration.cs @@ -0,0 +1,15 @@ +namespace ServiceControl.Persistence.EFCore.EntityConfigurations; + +using Entities; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Metadata.Builders; + +class RetryBatchNowForwardingConfiguration : IEntityTypeConfiguration +{ + public void Configure(EntityTypeBuilder builder) + { + builder.HasKey(e => e.Id); + builder.Property(e => e.Id).ValueGeneratedNever(); + builder.Property(e => e.RetryBatchId).IsRequired(); + } +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/RetryBatchMapper.cs b/src/ServiceControl.Persistence.EFCore/Implementation/RetryBatchMapper.cs new file mode 100644 index 0000000000..3552d50044 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Implementation/RetryBatchMapper.cs @@ -0,0 +1,27 @@ +namespace ServiceControl.Persistence.EFCore.Implementation; + +using ServiceControl.Persistence.EFCore.Entities; + +static class RetryBatchMapper +{ + public static RetryBatch ToRetryBatch(this RetryBatchEntity entity, IList failureRetries) => + new() + { + Id = entity.Id.ToString(), + Status = entity.Status, + RetrySessionId = entity.RetrySessionId, + RequestId = entity.RequestId, + RetryType = entity.RetryType, + InitialBatchSize = entity.InitialBatchSize, + StartTime = entity.StartTime, + Last = entity.Last, + StagingId = entity.StagingId, + Context = entity.Context, + Originator = entity.Originator, + Classifier = entity.Classifier, + InitiatedById = entity.InitiatedById, + InitiatedByName = entity.InitiatedByName, + OperationId = entity.OperationId, + FailureRetries = failureRetries + }; +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/RetryBatchStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/RetryBatchStore.cs index f4134a4ea7..06411a2945 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/RetryBatchStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/RetryBatchStore.cs @@ -1,41 +1,222 @@ namespace ServiceControl.Persistence.EFCore.Implementation; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; using ServiceControl.MessageFailures; +using ServiceControl.Persistence.EFCore.DbContexts; +using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Persistence.EFCore.Infrastructure; using ServiceControl.Persistence.Infrastructure; -using ServiceControl.Recoverability; -public class RetryBatchStore : IRetryBatchStore +public class RetryBatchStore(IServiceScopeFactory scopeFactory, IIngestionSqlDialect dialect) : DataStoreBase(scopeFactory), IRetryBatchStore { public Task CreateBatch(string retrySessionId, string requestId, RetryType retryType, string[] failedMessageRetryIds, string originator, DateTime startTime, DateTime? last = null, string? batchName = null, string? classifier = null, string? initiatedById = null, string? initiatedByName = null, string? operationId = null) => - throw new NotImplementedException(); + ExecuteWithDbContext(async dbContext => + { + var batch = new RetryBatchEntity + { + Id = Guid.NewGuid(), + Status = RetryBatchStatus.MarkingDocuments, + RetrySessionId = retrySessionId, + RequestId = requestId, + RetryType = retryType, + InitialBatchSize = failedMessageRetryIds.Length, + StartTime = startTime, + Last = last, + Context = batchName, + Originator = originator, + Classifier = classifier, + InitiatedById = initiatedById, + InitiatedByName = initiatedByName, + OperationId = operationId + }; + dbContext.RetryBatches.Add(batch); + + await dbContext.SaveChangesAsync(); + + return batch.Id.ToString(); + }); + + /// + /// Claims the messages for the batch. A message already claimed by another batch keeps that claim, so the batch it is staged with is whichever one got there first. + /// public Task AssignMessagesToBatch(string batchId, string[] messageIds) => - throw new NotImplementedException(); + ExecuteWithDbContext(async dbContext => + { + var batch = ParseBatchId(batchId); + + // Message ids reach this from the API, so an id that is not a message id is a caller's + // typo rather than a fault. It cannot match a stored message, so it is left unclaimed. + var uniqueMessageIds = new HashSet(); + + foreach (var messageId in messageIds) + { + if (Guid.TryParse(messageId, out var uniqueMessageId)) + { + uniqueMessageIds.Add(uniqueMessageId); + } + } + + var claims = uniqueMessageIds + .Select(uniqueMessageId => new FailedMessageRetryEntity { UniqueMessageId = uniqueMessageId, RetryBatchId = batch }) + .ToArray(); + + if (claims.Length == 0) + { + return; + } + + // The dialect writes on the connection directly, so it needs a transaction of its own. + var strategy = dbContext.Database.CreateExecutionStrategy(); + + await strategy.ExecuteAsync(async () => + { + await using var transaction = await dbContext.Database.BeginTransactionAsync(); + + await dialect.InsertMissingRetryClaims(dbContext, claims, CancellationToken.None); + + await transaction.CommitAsync(); + }); + }); + + public Task MoveBatchToStaging(string batchId) + { + var batch = ParseBatchId(batchId); + + return ExecuteWithDbContext(dbContext => dbContext.RetryBatches + .Where(row => row.Id == batch) + .ExecuteUpdateAsync(setters => setters.SetProperty(row => row.Status, RetryBatchStatus.Staging))); + } - public Task MoveBatchToStaging(string batchId) => - throw new NotImplementedException(); + // Batch ids only ever come from CreateBatch, so anything else is a programming error. + static Guid ParseBatchId(string batchId) => + Guid.TryParse(batchId, out var parsed) + ? parsed + : throw new ArgumentException($"'{batchId}' is not a retry batch id issued by this store.", nameof(batchId)); public Task>> GetOrphanedBatches(string retrySessionId) => - throw new NotImplementedException(); + ExecuteWithDbContext(async dbContext => + { + var orphaned = await dbContext.RetryBatches + .AsNoTracking() + .Where(batch => batch.Status == RetryBatchStatus.MarkingDocuments && batch.RetrySessionId != retrySessionId) + .ToListAsync(); + + var membership = await ReadMembership(dbContext, [.. orphaned.Select(batch => batch.Id)]); + + IList batches = [.. orphaned.Select(batch => batch.ToRetryBatch(membership.GetValueOrDefault(batch.Id, [])))]; + + return new QueryResult>(batches, new QueryStatsInfo(string.Empty, batches.Count, false)); + }); public Task> GetAvailableBatchGroups() => - throw new NotImplementedException(); + ExecuteWithDbContext>(async dbContext => + { + var groups = await dbContext.RetryBatches + .AsNoTracking() + .Where(batch => batch.Status == RetryBatchStatus.Staging || batch.Status == RetryBatchStatus.Forwarding) + .GroupBy(batch => new { batch.RequestId, batch.RetryType }) + .Select(group => new + { + group.Key.RequestId, + group.Key.RetryType, + HasStagingBatches = group.Any(batch => batch.Status == RetryBatchStatus.Staging), + HasForwardingBatches = group.Any(batch => batch.Status == RetryBatchStatus.Forwarding), + InitialBatchSize = group.Sum(batch => batch.InitialBatchSize), + StartTime = group.Min(batch => batch.StartTime), + Last = group.Max(batch => batch.Last), + Originator = group.Max(batch => batch.Originator), + Classifier = group.Max(batch => batch.Classifier) + }) + .ToListAsync(); + + return [.. groups.Select(group => new RetryBatchGroup + { + RequestId = group.RequestId, + RetryType = group.RetryType, + HasStagingBatches = group.HasStagingBatches, + HasForwardingBatches = group.HasForwardingBatches, + InitialBatchSize = group.InitialBatchSize, + StartTime = group.StartTime, + Last = group.Last ?? default, + Originator = group.Originator, + Classifier = group.Classifier + })]; + }); - public Task GetCurrentForwardingBatch() => - throw new NotImplementedException(); + public Task GetCurrentForwardingBatch() => + ExecuteWithDbContext(async dbContext => + { + var nowForwarding = await dbContext.RetryBatchNowForwarding + .AsNoTracking() + .SingleOrDefaultAsync(); + + if (nowForwarding == null) + { + return null; + } + + return await dbContext.RetryBatches + .AsNoTracking() + .Where(batch => batch.Id == nowForwarding.RetryBatchId) + .Select(batch => new ForwardingRetryBatch(batch.RequestId, batch.RetryType, batch.Originator!, batch.Classifier!)) + .SingleOrDefaultAsync(); + }); public Task ForEachUnresolvedMessage(Func callback) => - throw new NotImplementedException(); + ForEach(Unresolved, callback); public Task ForEachUnresolvedMessageForEndpoint(string endpoint, Func callback) => - throw new NotImplementedException(); + ForEach(dbContext => Unresolved(dbContext) + .Where(message => message.ReceivingEndpointName == endpoint), callback); public Task ForEachMessageForQueueAddress(string failedQueueAddress, FailedMessageStatus status, Func callback) => - throw new NotImplementedException(); + ForEach(dbContext => Unresolved(dbContext) + .Where(message => message.FailingEndpointAddress == failedQueueAddress && message.Status == status), callback); public Task ForEachUnresolvedMessageInGroup(string groupId, Func callback) => - throw new NotImplementedException(); + ForEach(dbContext => Unresolved(dbContext) + .Where(message => dbContext.FailedMessageGroups.Any(group => group.GroupId == groupId && group.FailedMessageUniqueId == message.UniqueMessageId)), callback); + + Task ForEach(Func> query, Func callback) => + ExecuteWithDbContext(dbContext => Stream(query(dbContext), callback)); + + static IQueryable Unresolved(ServiceControlDbContext dbContext) => + dbContext.FailedMessages + .AsNoTracking() + .Where(message => message.Status == FailedMessageStatus.Unresolved); + + static async Task Stream(IQueryable messages, Func callback) + { + var rows = messages + .Select(message => new { message.UniqueMessageId, message.LastTimeOfFailure }) + .AsAsyncEnumerable(); + + await foreach (var row in rows) + { + await callback(row.UniqueMessageId.ToString(), row.LastTimeOfFailure); + } + } + + static async Task>> ReadMembership(ServiceControlDbContext dbContext, Guid[] batchIds) + { + if (batchIds.Length == 0) + { + return []; + } + + var rows = await dbContext.FailedMessageRetries + .AsNoTracking() + .Where(retry => batchIds.Contains(retry.RetryBatchId)) + .Select(retry => new { retry.RetryBatchId, retry.UniqueMessageId }) + .ToListAsync(); + + return rows + .GroupBy(row => row.RetryBatchId) + .ToDictionary(group => group.Key, group => group.Select(row => row.UniqueMessageId.ToString()).ToList()); + } } diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/IIngestionSqlDialect.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/IIngestionSqlDialect.cs index 3e24d92faa..c94e75fa7f 100644 --- a/src/ServiceControl.Persistence.EFCore/Infrastructure/IIngestionSqlDialect.cs +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/IIngestionSqlDialect.cs @@ -27,4 +27,10 @@ public interface IIngestionSqlDialect /// Insert if absent, never update: existing endpoints keep their Monitored flag. /// Task InsertMissingKnownEndpoints(ServiceControlDbContext dbContext, IReadOnlyList rows, CancellationToken cancellationToken); + + /// + /// Insert if absent, never update: a message already claimed stays with the batch that claimed + /// it first, so two retry requests covering the same message cannot both stage it. + /// + Task InsertMissingRetryClaims(ServiceControlDbContext dbContext, IReadOnlyList rows, CancellationToken cancellationToken); } diff --git a/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionTests.cs index 0952009c4e..bb5c89598f 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionTests.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/ErrorIngestionTests.cs @@ -218,7 +218,7 @@ public async Task A_confirmed_retry_resolves_the_message_and_drops_its_retry_row { var failure = new IngestedFailure(); await Ingest(failure); - await Store(new FailedMessageRetryEntity { UniqueMessageId = failure.UniqueMessageId, RetryId = "RetryBatches/1" }); + await Store(new FailedMessageRetryEntity { UniqueMessageId = failure.UniqueMessageId, RetryBatchId = Guid.NewGuid() }); await ConfirmRetry(failure.UniqueMessageIdString); diff --git a/src/ServiceControl.Persistence.Tests/EFCore/RetryBatchStoreTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/RetryBatchStoreTests.cs new file mode 100644 index 0000000000..9beb782757 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/RetryBatchStoreTests.cs @@ -0,0 +1,248 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using NUnit.Framework; +using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Recoverability; + +class RetryBatchStoreTests : ErrorIngestionTestBase +{ + const string OtherSession = "another-session"; + + static readonly DateTime Noon = new(2026, 8, 1, 12, 0, 0, DateTimeKind.Utc); + + [Test] + public async Task Creates_a_batch() + { + var batchId = await CreateBatch("request-1", RetryType.FailureGroup, ["OrderPlaced failures"], messageCount: 3); + + var orphaned = await Orphaned(); + + var batch = orphaned.Single(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(batch.Id, Is.EqualTo(batchId)); + Assert.That(batch.Status, Is.EqualTo(RetryBatchStatus.MarkingDocuments)); + Assert.That(batch.RequestId, Is.EqualTo("request-1")); + Assert.That(batch.RetryType, Is.EqualTo(RetryType.FailureGroup)); + Assert.That(batch.InitialBatchSize, Is.EqualTo(3)); + Assert.That(batch.Originator, Is.EqualTo("OrderPlaced failures")); + Assert.That(batch.StartTime, Is.EqualTo(Noon)); + } + } + + [Test] + public async Task Does_not_report_batches_of_the_current_session_as_orphaned() + { + await CreateBatch("request-1", retrySessionId: OtherSession); + + var orphaned = await RetryBatchStore.GetOrphanedBatches(OtherSession); + + Assert.That(orphaned.Results, Is.Empty); + } + + [Test] + public async Task Does_not_report_staged_batches_as_orphaned() + { + var batchId = await CreateBatch("request-1"); + + await RetryBatchStore.MoveBatchToStaging(batchId); + + Assert.That(await Orphaned(), Is.Empty); + } + + [Test] + public async Task Claims_the_messages_of_a_batch() + { + var first = Guid.NewGuid(); + var second = Guid.NewGuid(); + var batchId = await CreateBatch("request-1"); + + await RetryBatchStore.AssignMessagesToBatch(batchId, [first.ToString(), second.ToString()]); + + var batch = (await Orphaned()).Single(); + + Assert.That(batch.FailureRetries, Is.EquivalentTo(new[] { first.ToString(), second.ToString() })); + } + + [Test] + public async Task Leaves_a_message_claimed_by_an_earlier_batch_alone() + { + var shared = Guid.NewGuid().ToString(); + var firstBatch = await CreateBatch("request-1"); + var secondBatch = await CreateBatch("request-2"); + + await RetryBatchStore.AssignMessagesToBatch(firstBatch, [shared]); + await RetryBatchStore.AssignMessagesToBatch(secondBatch, [shared]); + + var batches = (await Orphaned()).ToDictionary(batch => batch.Id); + + using (Assert.EnterMultipleScope()) + { + Assert.That(batches[firstBatch].FailureRetries, Is.EquivalentTo(new[] { shared })); + Assert.That(batches[secondBatch].FailureRetries, Is.Empty); + } + } + + [Test] + public async Task Claims_each_message_once_when_two_batches_race_for_them() + { + var shared = Enumerable.Range(0, 25).Select(_ => Guid.NewGuid().ToString()).ToArray(); + var firstBatch = await CreateBatch("request-1"); + var secondBatch = await CreateBatch("request-2"); + + await Task.WhenAll( + RetryBatchStore.AssignMessagesToBatch(firstBatch, shared), + RetryBatchStore.AssignMessagesToBatch(secondBatch, shared)); + + var batches = (await Orphaned()).ToDictionary(batch => batch.Id); + var claimed = batches[firstBatch].FailureRetries.Concat(batches[secondBatch].FailureRetries); + + Assert.That(claimed, Is.EquivalentTo(shared)); + } + + [Test] + public async Task Reports_staged_batches_as_available() + { + var batchId = await CreateBatch("request-1", RetryType.FailureGroup, ["OrderPlaced failures"], messageCount: 2); + + await RetryBatchStore.MoveBatchToStaging(batchId); + + var group = (await RetryBatchStore.GetAvailableBatchGroups()).Single(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(group.RequestId, Is.EqualTo("request-1")); + Assert.That(group.RetryType, Is.EqualTo(RetryType.FailureGroup)); + Assert.That(group.HasStagingBatches, Is.True); + Assert.That(group.HasForwardingBatches, Is.False); + Assert.That(group.InitialBatchSize, Is.EqualTo(2)); + Assert.That(group.Originator, Is.EqualTo("OrderPlaced failures")); + } + } + + [Test] + public async Task Adds_up_the_batches_of_one_request() + { + var first = await CreateBatch("request-1", messageCount: 2); + var second = await CreateBatch("request-1", messageCount: 5); + + await RetryBatchStore.MoveBatchToStaging(first); + await RetryBatchStore.MoveBatchToStaging(second); + + var group = (await RetryBatchStore.GetAvailableBatchGroups()).Single(); + + Assert.That(group.InitialBatchSize, Is.EqualTo(7)); + } + + [Test] + public async Task Does_not_report_batches_still_marking_documents_as_available() + { + await CreateBatch("request-1"); + + Assert.That(await RetryBatchStore.GetAvailableBatchGroups(), Is.Empty); + } + + [Test] + public async Task Returns_no_forwarding_batch_when_none_is_in_flight() + { + await CreateBatch("request-1"); + + Assert.That(await RetryBatchStore.GetCurrentForwardingBatch(), Is.Null); + } + + [Test] + public async Task Returns_the_batch_being_forwarded() + { + var batchId = await CreateBatch("request-1", RetryType.FailureGroup, ["OrderPlaced failures"]); + + await Store(new RetryBatchNowForwardingEntity { RetryBatchId = Guid.Parse(batchId) }); + + var batch = await RetryBatchStore.GetCurrentForwardingBatch(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(batch.RequestId, Is.EqualTo("request-1")); + Assert.That(batch.RetryType, Is.EqualTo(RetryType.FailureGroup)); + Assert.That(batch.Originator, Is.EqualTo("OrderPlaced failures")); + Assert.That(batch.Classifier, Is.EqualTo("Message Type")); + } + } + + [Test] + public async Task Streams_every_unresolved_message() + { + var unresolved = new IngestedFailure(); + await Ingest(unresolved); + await Insert(new IngestedFailure(), MessageFailures.FailedMessageStatus.Archived); + + var streamed = await Collect(callback => RetryBatchStore.ForEachUnresolvedMessage(callback)); + + Assert.That(streamed, Is.EquivalentTo(new[] { unresolved.UniqueMessageIdString })); + } + + [Test] + public async Task Streams_the_unresolved_messages_of_an_endpoint() + { + var sales = new IngestedFailure { ReceivingEndpoint = new() { Name = "Sales", Host = "H", HostId = Guid.NewGuid() } }; + var shipping = new IngestedFailure { ReceivingEndpoint = new() { Name = "Shipping", Host = "H", HostId = Guid.NewGuid() } }; + + await Ingest(sales, shipping); + + var streamed = await Collect(callback => RetryBatchStore.ForEachUnresolvedMessageForEndpoint("Sales", callback)); + + Assert.That(streamed, Is.EquivalentTo(new[] { sales.UniqueMessageIdString })); + } + + [Test] + public async Task Streams_the_unresolved_messages_of_a_group() + { + var group = new MessageFailures.FailedMessage.FailureGroup { Id = Guid.NewGuid().ToString(), Title = "OrderPlaced", Type = "Message Type" }; + var inGroup = new IngestedFailure { Groups = [group] }; + var outsideGroup = new IngestedFailure(); + + await Ingest(inGroup, outsideGroup); + + var streamed = await Collect(callback => RetryBatchStore.ForEachUnresolvedMessageInGroup(group.Id, callback)); + + Assert.That(streamed, Is.EquivalentTo(new[] { inGroup.UniqueMessageIdString })); + } + + Task CreateBatch(string requestId, RetryType retryType = RetryType.MultipleMessages, string[] originator = null, int messageCount = 1, string retrySessionId = "this-session") => + RetryBatchStore.CreateBatch( + retrySessionId, + requestId, + retryType, + [.. Enumerable.Range(0, messageCount).Select(_ => Guid.NewGuid().ToString())], + originator?.Single(), + Noon, + classifier: "Message Type"); + + async Task> Orphaned() => (await RetryBatchStore.GetOrphanedBatches(OtherSession)).Results; + + static async Task> Collect(Func, Task> stream) + { + var streamed = new List(); + + await stream((uniqueMessageId, _) => + { + streamed.Add(uniqueMessageId); + return Task.CompletedTask; + }); + + return streamed; + } + + async Task Insert(IngestedFailure failure, MessageFailures.FailedMessageStatus status) + { + var message = failure.ToFailedMessage(status); + message.Id = PersistenceTestsContext.GenerateFailedMessageRecordId(message.UniqueMessageId); + + await PersistenceTestsContext.InsertFailedMessages(message); + await CompleteDatabaseOperation(); + } +}