Skip to content

Commit f80dc26

Browse files
johnsimonsjohnsimons
committed
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.
1 parent d26ec99 commit f80dc26

20 files changed

Lines changed: 2042 additions & 24 deletions

‎src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260803021015_AddRetryBatches.Designer.cs‎

Lines changed: 568 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/20260803021015_AddRetryBatches.cs‎

Lines changed: 109 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,109 @@
1+
using System;
2+
using Microsoft.EntityFrameworkCore.Migrations;
3+
4+
#nullable disable
5+
6+
namespace ServiceControl.Persistence.EFCore.PostgreSql.Migrations
7+
{
8+
/// <inheritdoc />
9+
public partial class AddRetryBatches : Migration
10+
{
11+
/// <inheritdoc />
12+
protected override void Up(MigrationBuilder migrationBuilder)
13+
{
14+
migrationBuilder.DropColumn(
15+
name: "retry_id",
16+
table: "failed_message_retries");
17+
18+
migrationBuilder.AddColumn<Guid>(
19+
name: "retry_batch_id",
20+
table: "failed_message_retries",
21+
type: "uuid",
22+
nullable: false,
23+
defaultValue: new Guid("00000000-0000-0000-0000-000000000000"));
24+
25+
migrationBuilder.AddColumn<int>(
26+
name: "stage_attempts",
27+
table: "failed_message_retries",
28+
type: "integer",
29+
nullable: false,
30+
defaultValue: 0);
31+
32+
migrationBuilder.CreateTable(
33+
name: "retry_batch_now_forwarding",
34+
columns: table => new
35+
{
36+
id = table.Column<int>(type: "integer", nullable: false),
37+
retry_batch_id = table.Column<Guid>(type: "uuid", nullable: false)
38+
},
39+
constraints: table =>
40+
{
41+
table.PrimaryKey("pk_retry_batch_now_forwarding", x => x.id);
42+
});
43+
44+
migrationBuilder.CreateTable(
45+
name: "retry_batches",
46+
columns: table => new
47+
{
48+
id = table.Column<Guid>(type: "uuid", nullable: false),
49+
status = table.Column<int>(type: "integer", nullable: false),
50+
retry_session_id = table.Column<string>(type: "character varying(450)", maxLength: 450, nullable: false),
51+
request_id = table.Column<string>(type: "character varying(450)", maxLength: 450, nullable: false),
52+
retry_type = table.Column<int>(type: "integer", nullable: false),
53+
initial_batch_size = table.Column<int>(type: "integer", nullable: false),
54+
start_time = table.Column<DateTime>(type: "timestamp with time zone", nullable: false),
55+
last = table.Column<DateTime>(type: "timestamp with time zone", nullable: true),
56+
staging_id = table.Column<string>(type: "character varying(450)", maxLength: 450, nullable: true),
57+
context = table.Column<string>(type: "text", nullable: true),
58+
originator = table.Column<string>(type: "text", nullable: true),
59+
classifier = table.Column<string>(type: "text", nullable: true),
60+
initiated_by_id = table.Column<string>(type: "character varying(450)", maxLength: 450, nullable: true),
61+
initiated_by_name = table.Column<string>(type: "text", nullable: true),
62+
operation_id = table.Column<string>(type: "character varying(450)", maxLength: 450, nullable: true)
63+
},
64+
constraints: table =>
65+
{
66+
table.PrimaryKey("pk_retry_batches", x => x.id);
67+
});
68+
69+
migrationBuilder.CreateIndex(
70+
name: "ix_failed_message_retries_retry_batch_id",
71+
table: "failed_message_retries",
72+
column: "retry_batch_id");
73+
74+
migrationBuilder.CreateIndex(
75+
name: "ix_retry_batches_status_retry_session_id",
76+
table: "retry_batches",
77+
columns: new[] { "status", "retry_session_id" });
78+
}
79+
80+
/// <inheritdoc />
81+
protected override void Down(MigrationBuilder migrationBuilder)
82+
{
83+
migrationBuilder.DropTable(
84+
name: "retry_batch_now_forwarding");
85+
86+
migrationBuilder.DropTable(
87+
name: "retry_batches");
88+
89+
migrationBuilder.DropIndex(
90+
name: "ix_failed_message_retries_retry_batch_id",
91+
table: "failed_message_retries");
92+
93+
migrationBuilder.DropColumn(
94+
name: "retry_batch_id",
95+
table: "failed_message_retries");
96+
97+
migrationBuilder.DropColumn(
98+
name: "stage_attempts",
99+
table: "failed_message_retries");
100+
101+
migrationBuilder.AddColumn<string>(
102+
name: "retry_id",
103+
table: "failed_message_retries",
104+
type: "character varying(450)",
105+
maxLength: 450,
106+
nullable: true);
107+
}
108+
}
109+
}

‎src/ServiceControl.Persistence.EFCore.PostgreSql/Migrations/PostgreSqlServiceControlDbContextModelSnapshot.cs‎

Lines changed: 104 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -371,14 +371,20 @@ protected override void BuildModel(ModelBuilder modelBuilder)
371371
.HasColumnType("uuid")
372372
.HasColumnName("unique_message_id");
373373

374-
b.Property<string>("RetryId")
375-
.HasMaxLength(450)
376-
.HasColumnType("character varying(450)")
377-
.HasColumnName("retry_id");
374+
b.Property<Guid>("RetryBatchId")
375+
.HasColumnType("uuid")
376+
.HasColumnName("retry_batch_id");
377+
378+
b.Property<int>("StageAttempts")
379+
.HasColumnType("integer")
380+
.HasColumnName("stage_attempts");
378381

379382
b.HasKey("UniqueMessageId")
380383
.HasName("pk_failed_message_retries");
381384

385+
b.HasIndex("RetryBatchId")
386+
.HasDatabaseName("ix_failed_message_retries_retry_batch_id");
387+
382388
b.ToTable("failed_message_retries", (string)null);
383389
});
384390

@@ -455,6 +461,100 @@ protected override void BuildModel(ModelBuilder modelBuilder)
455461
b.ToTable("message_redirects", (string)null);
456462
});
457463

464+
modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.RetryBatchEntity", b =>
465+
{
466+
b.Property<Guid>("Id")
467+
.HasColumnType("uuid")
468+
.HasColumnName("id");
469+
470+
b.Property<string>("Classifier")
471+
.HasColumnType("text")
472+
.HasColumnName("classifier");
473+
474+
b.Property<string>("Context")
475+
.HasColumnType("text")
476+
.HasColumnName("context");
477+
478+
b.Property<int>("InitialBatchSize")
479+
.HasColumnType("integer")
480+
.HasColumnName("initial_batch_size");
481+
482+
b.Property<string>("InitiatedById")
483+
.HasMaxLength(450)
484+
.HasColumnType("character varying(450)")
485+
.HasColumnName("initiated_by_id");
486+
487+
b.Property<string>("InitiatedByName")
488+
.HasColumnType("text")
489+
.HasColumnName("initiated_by_name");
490+
491+
b.Property<DateTime?>("Last")
492+
.HasColumnType("timestamp with time zone")
493+
.HasColumnName("last");
494+
495+
b.Property<string>("OperationId")
496+
.HasMaxLength(450)
497+
.HasColumnType("character varying(450)")
498+
.HasColumnName("operation_id");
499+
500+
b.Property<string>("Originator")
501+
.HasColumnType("text")
502+
.HasColumnName("originator");
503+
504+
b.Property<string>("RequestId")
505+
.IsRequired()
506+
.HasMaxLength(450)
507+
.HasColumnType("character varying(450)")
508+
.HasColumnName("request_id");
509+
510+
b.Property<string>("RetrySessionId")
511+
.IsRequired()
512+
.HasMaxLength(450)
513+
.HasColumnType("character varying(450)")
514+
.HasColumnName("retry_session_id");
515+
516+
b.Property<int>("RetryType")
517+
.HasColumnType("integer")
518+
.HasColumnName("retry_type");
519+
520+
b.Property<string>("StagingId")
521+
.HasMaxLength(450)
522+
.HasColumnType("character varying(450)")
523+
.HasColumnName("staging_id");
524+
525+
b.Property<DateTime>("StartTime")
526+
.HasColumnType("timestamp with time zone")
527+
.HasColumnName("start_time");
528+
529+
b.Property<int>("Status")
530+
.HasColumnType("integer")
531+
.HasColumnName("status");
532+
533+
b.HasKey("Id")
534+
.HasName("pk_retry_batches");
535+
536+
b.HasIndex("Status", "RetrySessionId")
537+
.HasDatabaseName("ix_retry_batches_status_retry_session_id");
538+
539+
b.ToTable("retry_batches", (string)null);
540+
});
541+
542+
modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.RetryBatchNowForwardingEntity", b =>
543+
{
544+
b.Property<int>("Id")
545+
.HasColumnType("integer")
546+
.HasColumnName("id");
547+
548+
b.Property<Guid>("RetryBatchId")
549+
.HasColumnType("uuid")
550+
.HasColumnName("retry_batch_id");
551+
552+
b.HasKey("Id")
553+
.HasName("pk_retry_batch_now_forwarding");
554+
555+
b.ToTable("retry_batch_now_forwarding", (string)null);
556+
});
557+
458558
modelBuilder.Entity("ServiceControl.Persistence.EFCore.Entities.SubscriptionEntity", b =>
459559
{
460560
b.Property<string>("MessageType")

‎src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlIngestionSqlDialect.cs‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,23 @@ ON CONFLICT (id) DO NOTHING
6565
}
6666
}
6767

68+
public async Task InsertMissingRetryClaims(ServiceControlDbContext dbContext, IReadOnlyList<FailedMessageRetryEntity> rows, CancellationToken cancellationToken)
69+
{
70+
foreach (var chunk in rows.Chunk(MaxRowsPerStatement))
71+
{
72+
await Execute(
73+
dbContext,
74+
$"""
75+
INSERT INTO failed_message_retries (unique_message_id, retry_batch_id, stage_attempts)
76+
VALUES
77+
{ParameterRows(chunk.Length, 3)}
78+
ON CONFLICT (unique_message_id) DO NOTHING
79+
""",
80+
chunk.Select(retry => new object?[] { retry.UniqueMessageId, retry.RetryBatchId, retry.StageAttempts }),
81+
cancellationToken);
82+
}
83+
}
84+
6885
static async Task Execute(ServiceControlDbContext dbContext, string sql, IEnumerable<object?[]> rows, CancellationToken cancellationToken)
6986
{
7087
await using var command = dbContext.Database.GetDbConnection().CreateCommand();

0 commit comments

Comments
 (0)