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(); + } +}