diff --git a/docs/error-ingestion-design.md b/docs/error-ingestion-design.md index 523fe70515..38cf4a74bd 100644 --- a/docs/error-ingestion-design.md +++ b/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 diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlDialect.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlDialect.cs new file mode 100644 index 0000000000..d177ef3f59 --- /dev/null +++ b/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 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; +} diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlIngestionSqlDialect.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlFailedMessageIngestionSqlDialect.cs similarity index 68% rename from src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlIngestionSqlDialect.cs rename to src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlFailedMessageIngestionSqlDialect.cs index 3a0bdedafe..17184e1aeb 100644 --- a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlIngestionSqlDialect.cs +++ b/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 rows, CancellationToken cancellationToken) { @@ -65,45 +63,6 @@ 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(); - 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; } diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlPersistence.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlPersistence.cs index 9e807fc358..aa7c69f8c8 100644 --- a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlPersistence.cs +++ b/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlPersistence.cs @@ -14,7 +14,8 @@ public void AddPersistence(IServiceCollection services) ConfigureDbContext(services); RegisterDataStores(services, settings); - services.AddSingleton(); + services.AddSingleton(); + services.AddSingleton(); } public void AddInstaller(IServiceCollection services) diff --git a/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlRetryBatchSqlDialect.cs b/src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlRetryBatchSqlDialect.cs new file mode 100644 index 0000000000..a15cb17c58 --- /dev/null +++ b/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 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); + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerDialect.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerDialect.cs new file mode 100644 index 0000000000..6a2ff7de55 --- /dev/null +++ b/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 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; +} diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerIngestionSqlDialect.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerFailedMessageIngestionSqlDialect.cs similarity index 65% rename from src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerIngestionSqlDialect.cs rename to src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerFailedMessageIngestionSqlDialect.cs index de266a73af..6206d3bfa4 100644 --- a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerIngestionSqlDialect.cs +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerFailedMessageIngestionSqlDialect.cs @@ -1,23 +1,21 @@ namespace ServiceControl.Persistence.EFCore.SqlServer; -using System.Data; 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; // MERGE WITH (HOLDLOCK) over an inline VALUES source. HOLDLOCK closes the race where two writers // both miss a key and collide on the insert, and the single statement keeps every guard reading // the same row state. Rows are chunked to stay clear of the 2100 parameter limit while keeping // statement texts down to a few reusable shapes. -class SqlServerIngestionSqlDialect : IIngestionSqlDialect +class SqlServerFailedMessageIngestionSqlDialect : SqlServerDialect, IFailedMessageIngestionSqlDialect { public async Task UpsertFailedMessages(ServiceControlDbContext dbContext, IReadOnlyList rows, CancellationToken cancellationToken) { - foreach (var chunk in rows.Chunk(MaxRowsPerStatement)) + var maxRowsPerStatement = MaxRowsPerStatement(FailedMessageColumns.Length); + foreach (var chunk in rows.Chunk(maxRowsPerStatement)) { await Execute( dbContext, @@ -38,7 +36,8 @@ WHEN NOT MATCHED THEN INSERT ({FailedMessageColumnList}) public async Task InsertGroups(ServiceControlDbContext dbContext, IReadOnlyList rows, CancellationToken cancellationToken) { - foreach (var chunk in rows.Chunk(MaxRowsPerStatement)) + var maxRowsPerStatement = MaxRowsPerStatement(4); + foreach (var chunk in rows.Chunk(maxRowsPerStatement)) { await Execute( dbContext, @@ -58,7 +57,8 @@ WHEN NOT MATCHED THEN INSERT ([FailedMessageUniqueId], [GroupId], [Title], [Type public async Task InsertMissingKnownEndpoints(ServiceControlDbContext dbContext, IReadOnlyList rows, CancellationToken cancellationToken) { - foreach (var chunk in rows.Chunk(MaxRowsPerStatement)) + var maxRowsPerStatement = MaxRowsPerStatement(5); + foreach (var chunk in rows.Chunk(maxRowsPerStatement)) { await Execute( dbContext, @@ -76,56 +76,6 @@ 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(); - 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; - - // 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); - } - // The columns the newer attempt wins wholesale static readonly string[] PayloadColumns = [ @@ -187,31 +137,4 @@ WHEN MATCHED THEN 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(); - } - - // 27 columns * 50 rows stays well below the 2100 parameter limit - const int MaxRowsPerStatement = 50; } diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerPersistence.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerPersistence.cs index 3c6d1384b3..a2b8be30e1 100644 --- a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerPersistence.cs +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerPersistence.cs @@ -14,7 +14,8 @@ public void AddPersistence(IServiceCollection services) ConfigureDbContext(services); RegisterDataStores(services, settings); - services.AddSingleton(); + services.AddSingleton(); + services.AddSingleton(); } public void AddInstaller(IServiceCollection services) diff --git a/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerRetryBatchSqlDialect.cs b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerRetryBatchSqlDialect.cs new file mode 100644 index 0000000000..0774e5c64b --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerRetryBatchSqlDialect.cs @@ -0,0 +1,29 @@ +namespace ServiceControl.Persistence.EFCore.SqlServer; + +using DbContexts; +using Entities; +using Infrastructure; + +class SqlServerRetryBatchSqlDialect : SqlServerDialect, IRetryBatchSqlDialect +{ + public async Task InsertMissingRetryClaims(ServiceControlDbContext dbContext, IReadOnlyList rows, CancellationToken cancellationToken) + { + var maxRowsPerStatement = MaxRowsPerStatement(3); + 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); + } + } +} diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/RetryBatchStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/RetryBatchStore.cs index 0adf2ae6f4..11fa9b8681 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/RetryBatchStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/RetryBatchStore.cs @@ -8,7 +8,7 @@ namespace ServiceControl.Persistence.EFCore.Implementation; using ServiceControl.Persistence.EFCore.Infrastructure; using ServiceControl.Persistence.Infrastructure; -public class RetryBatchStore(IServiceScopeFactory scopeFactory, IIngestionSqlDialect dialect) : DataStoreBase(scopeFactory), IRetryBatchStore +public class RetryBatchStore(IServiceScopeFactory scopeFactory, IRetryBatchSqlDialect dialect) : DataStoreBase(scopeFactory), IRetryBatchStore { public Task CreateBatch(string retrySessionId, string requestId, RetryType retryType, string[] failedMessageRetryIds, string originator, DateTime startTime, DateTime? last = null, diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWork.cs b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWork.cs index 07f434269a..0f67fcc1af 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWork.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWork.cs @@ -12,14 +12,14 @@ public class EFIngestionUnitOfWork : IIngestionUnitOfWork { readonly ServiceControlDbContext dbContext; readonly IAsyncDisposable scope; - readonly IIngestionSqlDialect dialect; + readonly IFailedMessageIngestionSqlDialect dialect; readonly TimeProvider timeProvider; readonly ConcurrentQueue failedProcessingAttempts = new(); readonly ConcurrentQueue bodyWrites = new(); readonly ConcurrentQueue knownEndpoints = new(); readonly ConcurrentQueue confirmedRetries = new(); - public EFIngestionUnitOfWork(IAsyncDisposable scope, ServiceControlDbContext dbContext, IBodyStoragePersistence storagePersistence, EFPersisterSettings settings, IIngestionSqlDialect dialect, TimeProvider timeProvider) + public EFIngestionUnitOfWork(IAsyncDisposable scope, ServiceControlDbContext dbContext, IBodyStoragePersistence storagePersistence, EFPersisterSettings settings, IFailedMessageIngestionSqlDialect dialect, TimeProvider timeProvider) { this.scope = scope; this.dbContext = dbContext; diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWorkFactory.cs b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWorkFactory.cs index 9c2422a249..c62ec80634 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWorkFactory.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWorkFactory.cs @@ -10,7 +10,7 @@ public class EFIngestionUnitOfWorkFactory( IServiceProvider serviceProvider, MinimumRequiredStorageState storageState, IBodyStoragePersistence storagePersistence, - IIngestionSqlDialect dialect, + IFailedMessageIngestionSqlDialect dialect, TimeProvider timeProvider) : IIngestionUnitOfWorkFactory { public ValueTask StartNew() diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/FailedMessageBatchWriter.cs b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/FailedMessageBatchWriter.cs index 646001b249..84d347494f 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/FailedMessageBatchWriter.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/FailedMessageBatchWriter.cs @@ -10,7 +10,7 @@ namespace ServiceControl.Persistence.EFCore.Implementation.UnitOfWork; // differ on (the upserts) come from the injected dialect; everything portable stays here as // set-based EF operations. Statement order matters: a message that fails and is retry-confirmed // in the same batch must end Resolved. -class FailedMessageBatchWriter(ServiceControlDbContext dbContext, IIngestionSqlDialect dialect) +class FailedMessageBatchWriter(ServiceControlDbContext dbContext, IFailedMessageIngestionSqlDialect dialect) { public async Task Write( IReadOnlyCollection attempts, diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/IIngestionSqlDialect.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/IFailedMessageIngestionSqlDialect.cs similarity index 64% rename from src/ServiceControl.Persistence.EFCore/Infrastructure/IIngestionSqlDialect.cs rename to src/ServiceControl.Persistence.EFCore/Infrastructure/IFailedMessageIngestionSqlDialect.cs index 7dd9fcd21a..c43273546a 100644 --- a/src/ServiceControl.Persistence.EFCore/Infrastructure/IIngestionSqlDialect.cs +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/IFailedMessageIngestionSqlDialect.cs @@ -4,11 +4,12 @@ namespace ServiceControl.Persistence.EFCore.Infrastructure; using ServiceControl.Persistence.EFCore.Entities; /// -/// The provider specific SQL of the error ingestion batch. Implementations run on the DbContext -/// connection inside the transaction the caller has already opened, and every statement must stay -/// correct under concurrent writers: a same-key race between two instances may not fail the batch. +/// The provider-specific SQL of the failed-message ingestion batch. Implementations run on the +/// DbContext connection inside the transaction the caller has already opened, and every statement +/// must stay correct under concurrent writers: a same-key race between two instances may not fail +/// the batch. /// -public interface IIngestionSqlDialect +public interface IFailedMessageIngestionSqlDialect { /// /// One row per message, distinct by UniqueMessageId. Inserts new rows; for existing rows the @@ -27,10 +28,4 @@ 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.EFCore/Infrastructure/IRetryBatchSqlDialect.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/IRetryBatchSqlDialect.cs new file mode 100644 index 0000000000..f1c0bbbb87 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/IRetryBatchSqlDialect.cs @@ -0,0 +1,16 @@ +namespace ServiceControl.Persistence.EFCore.Infrastructure; + +using ServiceControl.Persistence.EFCore.DbContexts; +using ServiceControl.Persistence.EFCore.Entities; + +/// +/// The provider-specific SQL used to claim messages for retry batches. +/// +public interface IRetryBatchSqlDialect +{ + /// + /// 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/IngestionSqlDialectTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/FailedMessageIngestionSqlDialectTests.cs similarity index 97% rename from src/ServiceControl.Persistence.Tests/EFCore/IngestionSqlDialectTests.cs rename to src/ServiceControl.Persistence.Tests/EFCore/FailedMessageIngestionSqlDialectTests.cs index d20a9a350a..62f798c979 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/IngestionSqlDialectTests.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/FailedMessageIngestionSqlDialectTests.cs @@ -11,7 +11,7 @@ namespace ServiceControl.Persistence.Tests; using ServiceControl.Persistence.EFCore.Entities; using ServiceControl.Persistence.EFCore.Infrastructure; -class IngestionSqlDialectTests : ErrorIngestionTestBase +class FailedMessageIngestionSqlDialectTests : ErrorIngestionTestBase { [Test] public async Task Writes_every_mapped_column_of_a_failed_message() @@ -84,7 +84,7 @@ async Task Upsert(FailedMessageEntity row) { using var scope = ServiceProvider.CreateScope(); var dbContext = scope.ServiceProvider.GetRequiredService(); - var dialect = scope.ServiceProvider.GetRequiredService(); + var dialect = scope.ServiceProvider.GetRequiredService(); var strategy = dbContext.Database.CreateExecutionStrategy();