Split up sql dialects by consuming store

docs/error-ingestion-design.md

@@ -11,7 +11,9 @@ ordinary change-tracked entity saves.
 The unit of ingestion is a **batch**: the transport hands the ingester up to `MaximumConcurrency`
 messages at a time, and the whole batch is written in a single database transaction. The relevant
 types are `EFIngestionUnitOfWork` (accumulation), `FailedMessageBatchWriter` (the write), and the
-per-provider `IIngestionSqlDialect` implementations (the statements that differ by provider).
+per-provider `IFailedMessageIngestionSqlDialect` implementations (the statements that differ by
+provider). Retry claim insertion is a separate persistence capability behind
+`IRetryBatchSqlDialect`; it is used by `RetryBatchStore`, not by the ingestion unit of work.
 
 ## Data model
 
@@ -108,8 +110,8 @@ The order matters: a message that both fails and is retry-confirmed in the same
 Most of ServiceControl prefers standard abstractions, and the portable parts of this write path do
 use them: the group delete and the retry resolution are ordinary set-based EF operations
 (`ExecuteDelete`/`ExecuteUpdate`). The **upserts** are hand-written SQL, per provider, behind the
-`IIngestionSqlDialect` seam. Three requirements together force that, and no ORM-level API satisfies
-all three at once.
+`IFailedMessageIngestionSqlDialect` seam. Three requirements together force that, and no ORM-level
+API satisfies all three at once.
 
 ### 1. The upsert is a conditional merge, not a save
 
@@ -169,10 +171,14 @@ that the database can cache a plan for, with no per-row round trips and no tempo
 ### What stays portable
 
 Only the genuinely divergent statements are raw. The retry resolution and the group delete are set
-based and identical across providers, so they remain EF operations in the shared writer. The raw
-SQL is confined to the two dialect classes, one per provider, each responsible only for the
-upserts. The guard semantics are kept identical between the two dialects; the shared test suite
-runs every ingestion test against both providers to keep them from drifting.
+based and identical across providers, so they remain EF operations in the shared writer. Failed
+message upserts and insert-if-absent group and endpoint writes are owned by each provider's
+`IFailedMessageIngestionSqlDialect` implementation. Insert-if-absent retry claims belong to the
+separate retry-batch persistence seam, `IRetryBatchSqlDialect`. Provider-local base classes share
+only parameter and transaction plumbing between those capabilities; each capability class still
+owns its SQL and domain semantics. The guard semantics are kept identical between providers, and
+the shared test suite runs every ingestion and retry-batch test against both providers to keep them
+from drifting.
 
 ## Transactions and retries
 

src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlDialect.cs

@@ -0,0 +1,57 @@
+namespace ServiceControl.Persistence.EFCore.PostgreSql;
+
+using System.Text;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Storage;
+using ServiceControl.Persistence.EFCore.DbContexts;
+
+abstract class PostgreSqlDialect
+{
+    protected static async Task Execute(ServiceControlDbContext dbContext, string sql, IEnumerable<object?[]> rows, CancellationToken cancellationToken)
+    {
+        await using var command = dbContext.Database.GetDbConnection().CreateCommand();
+        command.Transaction = (dbContext.Database.CurrentTransaction
+            ?? throw new InvalidOperationException("Dialect statements must run inside a transaction")).GetDbTransaction();
+        command.CommandText = sql;
+
+        var index = 0;
+        foreach (var row in rows)
+        {
+            foreach (var value in row)
+            {
+                var parameter = command.CreateParameter();
+                parameter.ParameterName = $"@p{index++}";
+                parameter.Value = value ?? DBNull.Value;
+                command.Parameters.Add(parameter);
+            }
+        }
+
+        await command.ExecuteNonQueryAsync(cancellationToken);
+    }
+
+    protected static string ParameterRows(int rowCount, int columnCount)
+    {
+        var sql = new StringBuilder();
+
+        for (var row = 0; row < rowCount; row++)
+        {
+            sql.Append(row == 0 ? "(" : ",\n(");
+
+            for (var column = 0; column < columnCount; column++)
+            {
+                if (column > 0)
+                {
+                    sql.Append(", ");
+                }
+
+                sql.Append("@p").Append((row * columnCount) + column);
+            }
+
+            sql.Append(')');
+        }
+
+        return sql.ToString();
+    }
+
+    protected const int MaxRowsPerStatement = 50;
+}

src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlIngestionSqlDialect.cs → src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlFailedMessageIngestionSqlDialect.cs

@@ -1,18 +1,16 @@
 namespace ServiceControl.Persistence.EFCore.PostgreSql;
 
 using System.Text;
-using Microsoft.EntityFrameworkCore;
-using Microsoft.EntityFrameworkCore.Storage;
-using ServiceControl.MessageFailures;
-using ServiceControl.Persistence.EFCore.DbContexts;
-using ServiceControl.Persistence.EFCore.Entities;
-using ServiceControl.Persistence.EFCore.Infrastructure;
+using MessageFailures;
+using DbContexts;
+using Entities;
+using Infrastructure;
 
 // INSERT ... ON CONFLICT rather than MERGE: PostgreSQL's MERGE can fail with unique_violation
 // when two writers insert the same key concurrently, ON CONFLICT cannot. All references to the
 // target table inside DO UPDATE read the pre-update row, so the guards are consistent within one
 // atomic statement. Rows are chunked to keep statement texts down to a few reusable shapes.
-class PostgreSqlIngestionSqlDialect : IIngestionSqlDialect
+class PostgreSqlFailedMessageIngestionSqlDialect : PostgreSqlDialect, IFailedMessageIngestionSqlDialect
 {
     public async Task UpsertFailedMessages(ServiceControlDbContext dbContext, IReadOnlyList<FailedMessageEntity> rows, CancellationToken cancellationToken)
     {
@@ -65,45 +63,6 @@ ON CONFLICT (id) DO NOTHING
         }
     }
 
-    public async Task InsertMissingRetryClaims(ServiceControlDbContext dbContext, IReadOnlyList<FailedMessageRetryEntity> 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<object?[]> rows, CancellationToken cancellationToken)
-    {
-        await using var command = dbContext.Database.GetDbConnection().CreateCommand();
-        command.Transaction = (dbContext.Database.CurrentTransaction
-            ?? throw new InvalidOperationException("Ingestion statements must run inside the batch transaction")).GetDbTransaction();
-        command.CommandText = sql;
-
-        var index = 0;
-        foreach (var row in rows)
-        {
-            foreach (var value in row)
-            {
-                var parameter = command.CreateParameter();
-                parameter.ParameterName = $"@p{index++}";
-                parameter.Value = value ?? DBNull.Value;
-                command.Parameters.Add(parameter);
-            }
-        }
-
-        await command.ExecuteNonQueryAsync(cancellationToken);
-    }
-
     // The columns the newer attempt wins wholesale
     static readonly string[] PayloadColumns =
     [
@@ -163,30 +122,4 @@ ON CONFLICT (unique_message_id) DO UPDATE SET
 
         return sql.ToString();
     }
-
-    static string ParameterRows(int rowCount, int columnCount)
-    {
-        var sql = new StringBuilder();
-
-        for (var row = 0; row < rowCount; row++)
-        {
-            sql.Append(row == 0 ? "(" : ",\n(");
-
-            for (var column = 0; column < columnCount; column++)
-            {
-                if (column > 0)
-                {
-                    sql.Append(", ");
-                }
-
-                sql.Append("@p").Append((row * columnCount) + column);
-            }
-
-            sql.Append(')');
-        }
-
-        return sql.ToString();
-    }
-
-    const int MaxRowsPerStatement = 50;
 }

src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlPersistence.cs

@@ -14,7 +14,8 @@ public void AddPersistence(IServiceCollection services)
         ConfigureDbContext(services);
         RegisterDataStores(services, settings);
 
-        services.AddSingleton<IIngestionSqlDialect, PostgreSqlIngestionSqlDialect>();
+        services.AddSingleton<IFailedMessageIngestionSqlDialect, PostgreSqlFailedMessageIngestionSqlDialect>();
+        services.AddSingleton<IRetryBatchSqlDialect, PostgreSqlRetryBatchSqlDialect>();
     }
 
     public void AddInstaller(IServiceCollection services)

src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlRetryBatchSqlDialect.cs

@@ -0,0 +1,25 @@
+namespace ServiceControl.Persistence.EFCore.PostgreSql;
+
+using DbContexts;
+using Entities;
+using Infrastructure;
+
+class PostgreSqlRetryBatchSqlDialect : PostgreSqlDialect, IRetryBatchSqlDialect
+{
+    public async Task InsertMissingRetryClaims(ServiceControlDbContext dbContext, IReadOnlyList<FailedMessageRetryEntity> 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);
+        }
+    }
+}

src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerDialect.cs

@@ -0,0 +1,67 @@
+namespace ServiceControl.Persistence.EFCore.SqlServer;
+
+using System.Data;
+using System.Text;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Storage;
+using ServiceControl.Persistence.EFCore.DbContexts;
+
+abstract class SqlServerDialect
+{
+    protected static async Task Execute(ServiceControlDbContext dbContext, string sql, IEnumerable<object?[]> rows, CancellationToken cancellationToken)
+    {
+        await using var command = dbContext.Database.GetDbConnection().CreateCommand();
+        command.Transaction = (dbContext.Database.CurrentTransaction
+            ?? throw new InvalidOperationException("Dialect statements must run inside a transaction")).GetDbTransaction();
+        command.CommandText = sql;
+
+        var index = 0;
+        foreach (var row in rows)
+        {
+            foreach (var value in row)
+            {
+                var parameter = command.CreateParameter();
+                parameter.ParameterName = $"@p{index++}";
+                parameter.Value = value ?? DBNull.Value;
+
+                // Attempt de-duplication compares LastAttemptedAt for equality, so datetime
+                // parameters must keep datetime2 precision instead of the datetime default.
+                if (value is DateTime)
+                {
+                    parameter.DbType = DbType.DateTime2;
+                }
+
+                command.Parameters.Add(parameter);
+            }
+        }
+
+        await command.ExecuteNonQueryAsync(cancellationToken);
+    }
+
+    protected static string ParameterRows(int rowCount, int columnCount)
+    {
+        var sql = new StringBuilder();
+
+        for (var row = 0; row < rowCount; row++)
+        {
+            sql.Append(row == 0 ? "(" : ",\n(");
+
+            for (var column = 0; column < columnCount; column++)
+            {
+                if (column > 0)
+                {
+                    sql.Append(", ");
+                }
+
+                sql.Append("@p").Append((row * columnCount) + column);
+            }
+
+            sql.Append(')');
+        }
+
+        return sql.ToString();
+    }
+
+    protected static int MaxRowsPerStatement(int columns) => MaxSqlParameters / columns;
+    const int MaxSqlParameters = 2100;
+}