From d5977c6ed0601e09062afcee2166a9834fd5bc8a Mon Sep 17 00:00:00 2001 From: John Simons Date: Fri, 17 Jul 2026 14:55:09 +1000 Subject: [PATCH] Integrate EF Core into ingestion unit of work The `IIngestionUnitOfWork` interface is updated to `IAsyncDisposable` to properly manage asynchronous resources like `ServiceControlDbContext` and `AsyncServiceScope`. `EFIngestionUnitOfWork` now manages a dedicated `ServiceControlDbContext` instance within its own `AsyncServiceScope`, persisting changes via `SaveChangesAsync`. A new `IBodyStoragePersistence` interface is introduced for abstracting message body storage. --- .../Abstractions/BasePersistence.cs | 1 + .../UnitOfWork/EFIngestionUnitOfWork.cs | 22 +++++++++++++++---- .../EFIngestionUnitOfWorkFactory.cs | 22 ++++++++++++++----- .../Infrastructure/IBodyStoragePersistence.cs | 12 ++++++++++ .../Infrastructure/MessageBodyFileResult.cs | 11 ++++++++++ .../Expiration/MessageExpiryTests.cs | 12 +++++----- .../AttachmentsBodyStorageTests.cs | 2 +- .../MonitoringDataStoreTests.cs | 2 +- .../UnitOfWork/FallbackIngestionUnitOfWork.cs | 12 ++++++---- .../UnitOfWork/IIngestionUnitOfWork.cs | 2 +- .../UnitOfWork/IngestionUnitOfWorkBase.cs | 14 +++++------- .../Operations/ErrorIngestor.cs | 2 +- 12 files changed, 82 insertions(+), 32 deletions(-) create mode 100644 src/ServiceControl.Persistence.EFCore/Infrastructure/IBodyStoragePersistence.cs create mode 100644 src/ServiceControl.Persistence.EFCore/Infrastructure/MessageBodyFileResult.cs diff --git a/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs b/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs index 0076c9ae81..dccd2abcbb 100644 --- a/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs +++ b/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs @@ -15,6 +15,7 @@ public abstract class BasePersistence protected static void RegisterDataStores(IServiceCollection services) { services.AddSingleton(TimeProvider.System); + services.AddSingleton(); services.AddSingleton(); services.AddSingleton(p => p.GetRequiredService()); diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWork.cs b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWork.cs index 2e0b2ab1ae..33d7ae477f 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWork.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWork.cs @@ -1,21 +1,35 @@ namespace ServiceControl.Persistence.EFCore.Implementation.UnitOfWork; +using Microsoft.Extensions.DependencyInjection; +using ServiceControl.Persistence.EFCore.Abstractions; +using ServiceControl.Persistence.EFCore.DbContexts; +using ServiceControl.Persistence.EFCore.Infrastructure; using ServiceControl.Persistence.UnitOfWork; public class EFIngestionUnitOfWork : IIngestionUnitOfWork { + readonly ServiceControlDbContext dbContext; + readonly AsyncServiceScope scope; + + public EFIngestionUnitOfWork(AsyncServiceScope scope, ServiceControlDbContext dbContext, IBodyStoragePersistence storagePersistence, EFPersisterSettings settings) + { + this.scope = scope; + this.dbContext = dbContext; + } + public IMonitoringIngestionUnitOfWork Monitoring => throw new NotImplementedException(); public IRecoverabilityIngestionUnitOfWork Recoverability => throw new NotImplementedException(); - public Task Complete(CancellationToken cancellationToken) => - throw new NotImplementedException(); + public Task Complete(CancellationToken cancellationToken) => dbContext.SaveChangesAsync(cancellationToken); - public void Dispose() + public async ValueTask DisposeAsync() { - // Nothing to dispose yet + await dbContext.DisposeAsync(); + await scope.DisposeAsync(); + GC.SuppressFinalize(this); } } diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWorkFactory.cs b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWorkFactory.cs index 2aa8efe4c6..a425acd23e 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWorkFactory.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWorkFactory.cs @@ -1,12 +1,24 @@ namespace ServiceControl.Persistence.EFCore.Implementation.UnitOfWork; +using Microsoft.Extensions.DependencyInjection; +using ServiceControl.Persistence.EFCore.Abstractions; +using ServiceControl.Persistence.EFCore.DbContexts; +using ServiceControl.Persistence.EFCore.Infrastructure; using ServiceControl.Persistence.UnitOfWork; -public class EFIngestionUnitOfWorkFactory : IIngestionUnitOfWorkFactory +public class EFIngestionUnitOfWorkFactory( + IServiceProvider serviceProvider, + MinimumRequiredStorageState storageState, + IBodyStoragePersistence storagePersistence) : IIngestionUnitOfWorkFactory { - public ValueTask StartNew() => - throw new NotImplementedException(); + public ValueTask StartNew() + { + var scope = serviceProvider.CreateAsyncScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + var settings = scope.ServiceProvider.GetRequiredService(); + var unitOfWork = new EFIngestionUnitOfWork(scope, dbContext, storagePersistence, settings); + return ValueTask.FromResult(unitOfWork); + } - public bool CanIngestMore() => - throw new NotImplementedException(); + public bool CanIngestMore() => storageState.CanIngestMore; } diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/IBodyStoragePersistence.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/IBodyStoragePersistence.cs new file mode 100644 index 0000000000..b4ef500909 --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/IBodyStoragePersistence.cs @@ -0,0 +1,12 @@ +namespace ServiceControl.Persistence.EFCore.Infrastructure; + +using System; +using System.Threading; +using System.Threading.Tasks; + +public interface IBodyStoragePersistence +{ + Task WriteBody(string bodyId, DateTime createdOn, ReadOnlyMemory body, string contentType, CancellationToken cancellationToken = default); + Task ReadBody(string bodyId, DateTime createdOn, CancellationToken cancellationToken = default); + Task DeleteBody(string bodyId, CancellationToken cancellationToken = default); +} diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/MessageBodyFileResult.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/MessageBodyFileResult.cs new file mode 100644 index 0000000000..2d58946d9b --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/MessageBodyFileResult.cs @@ -0,0 +1,11 @@ +namespace ServiceControl.Persistence.EFCore.Infrastructure; + +using System.IO; + +public class MessageBodyFileResult +{ + public Stream Stream { get; set; } = null!; + public string ContentType { get; set; } = null!; + public int BodySize { get; set; } + public string Etag { get; set; } = null!; +} \ No newline at end of file diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/Expiration/MessageExpiryTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/Expiration/MessageExpiryTests.cs index 7e8ca92206..67b12d36b5 100644 --- a/src/ServiceControl.Persistence.Tests.RavenDB/Expiration/MessageExpiryTests.cs +++ b/src/ServiceControl.Persistence.Tests.RavenDB/Expiration/MessageExpiryTests.cs @@ -34,7 +34,7 @@ public async Task SingleMessageMarkedAsArchiveShouldExpire() { var (context, attempt) = CreateMessageContext(); - using (var uow = await IngestionUnitOfWorkFactory.StartNew()) + await using (var uow = await IngestionUnitOfWorkFactory.StartNew()) { await uow.Recoverability.RecordFailedProcessingAttempt(context, attempt, []); @@ -63,7 +63,7 @@ public async Task AllMessagesInUnArchivedGroupShouldNotExpire() var (contextA, attemptA) = CreateMessageContext(); var (contextB, attemptB) = CreateMessageContext(); - using (var uow = await IngestionUnitOfWorkFactory.StartNew()) + await using (var uow = await IngestionUnitOfWorkFactory.StartNew()) { await uow.Recoverability.RecordFailedProcessingAttempt(contextA, attemptA, [new FailedMessage.FailureGroup { Id = groupIdA }]); await uow.Recoverability.RecordFailedProcessingAttempt(contextB, attemptB, [new FailedMessage.FailureGroup { Id = groupIdB }]); @@ -100,7 +100,7 @@ public async Task AllMessagesInArchivedGroupShouldExpire() var groupId = Guid.NewGuid().ToString(); var (context, attempt) = CreateMessageContext(); - using (var uow = await IngestionUnitOfWorkFactory.StartNew()) + await using (var uow = await IngestionUnitOfWorkFactory.StartNew()) { await uow.Recoverability.RecordFailedProcessingAttempt(context, attempt, [new FailedMessage.FailureGroup { Id = groupId }]); @@ -123,7 +123,7 @@ public async Task SingleMessageMarkedAsResolvedShouldExpire() { var (context, attempt) = CreateMessageContext(); - using (var uow = await IngestionUnitOfWorkFactory.StartNew()) + await using (var uow = await IngestionUnitOfWorkFactory.StartNew()) { await uow.Recoverability.RecordFailedProcessingAttempt(context, attempt, []); @@ -146,7 +146,7 @@ public async Task RetryConfirmationProcessingShouldTriggerExpiration() { var (context, attempt) = CreateMessageContext(); - using (var uow = await IngestionUnitOfWorkFactory.StartNew()) + await using (var uow = await IngestionUnitOfWorkFactory.StartNew()) { await uow.Recoverability.RecordFailedProcessingAttempt(context, attempt, []); @@ -159,7 +159,7 @@ public async Task RetryConfirmationProcessingShouldTriggerExpiration() Assert.That(errors.Results, Has.Count.EqualTo(1), "Failed message should be available to query after ingestion"); - using (var uow = await IngestionUnitOfWorkFactory.StartNew()) + await using (var uow = await IngestionUnitOfWorkFactory.StartNew()) { await uow.Recoverability.RecordSuccessfulRetry(errors.Results.First().Id); diff --git a/src/ServiceControl.Persistence.Tests/BodyStorage/AttachmentsBodyStorageTests.cs b/src/ServiceControl.Persistence.Tests/BodyStorage/AttachmentsBodyStorageTests.cs index ac952f81d4..de9d11235e 100644 --- a/src/ServiceControl.Persistence.Tests/BodyStorage/AttachmentsBodyStorageTests.cs +++ b/src/ServiceControl.Persistence.Tests/BodyStorage/AttachmentsBodyStorageTests.cs @@ -44,7 +44,7 @@ async Task RunTest(Func, string> getIdToQuery) }; using (var cancellationSource = new CancellationTokenSource()) - using (var uow = await ingestionFactory.StartNew()) + await using (var uow = await ingestionFactory.StartNew()) { var context = new MessageContext(messageId, headers, body, new TransportTransaction(), "receiveAddress", new NServiceBus.Extensibility.ContextBag()); var processingAttempt = new FailedMessage.ProcessingAttempt diff --git a/src/ServiceControl.Persistence.Tests/MonitoringDataStoreTests.cs b/src/ServiceControl.Persistence.Tests/MonitoringDataStoreTests.cs index e2fec6a83c..7e8a7d3fce 100644 --- a/src/ServiceControl.Persistence.Tests/MonitoringDataStoreTests.cs +++ b/src/ServiceControl.Persistence.Tests/MonitoringDataStoreTests.cs @@ -131,7 +131,7 @@ public async Task Unit_of_work_detects_endpoint() EndpointDetails = new EndpointDetails { Host = "Host1", HostId = Guid.NewGuid(), Name = "Endpoint" } }; - using (var unitOfWork = await UnitOfWorkFactory.StartNew()) + await using (var unitOfWork = await UnitOfWorkFactory.StartNew()) { await unitOfWork.Monitoring.RecordKnownEndpoint(knownEndpoint); diff --git a/src/ServiceControl.Persistence/UnitOfWork/FallbackIngestionUnitOfWork.cs b/src/ServiceControl.Persistence/UnitOfWork/FallbackIngestionUnitOfWork.cs index 4fbb1e2885..98921ae799 100644 --- a/src/ServiceControl.Persistence/UnitOfWork/FallbackIngestionUnitOfWork.cs +++ b/src/ServiceControl.Persistence/UnitOfWork/FallbackIngestionUnitOfWork.cs @@ -25,12 +25,16 @@ public override Task Complete(CancellationToken cancellationToken) fallback.Complete(cancellationToken) ); - protected override void Dispose(bool disposing) + protected override async ValueTask DisposeAsyncCore() { - if (disposing) + if (primary != null) { - primary?.Dispose(); - fallback?.Dispose(); + await primary.DisposeAsync(); + } + + if (fallback != null) + { + await fallback.DisposeAsync(); } } } diff --git a/src/ServiceControl.Persistence/UnitOfWork/IIngestionUnitOfWork.cs b/src/ServiceControl.Persistence/UnitOfWork/IIngestionUnitOfWork.cs index a14bda0369..ce5f245ac9 100644 --- a/src/ServiceControl.Persistence/UnitOfWork/IIngestionUnitOfWork.cs +++ b/src/ServiceControl.Persistence/UnitOfWork/IIngestionUnitOfWork.cs @@ -4,7 +4,7 @@ using System.Threading; using System.Threading.Tasks; - public interface IIngestionUnitOfWork : IDisposable + public interface IIngestionUnitOfWork : IAsyncDisposable { IMonitoringIngestionUnitOfWork Monitoring { get; } IRecoverabilityIngestionUnitOfWork Recoverability { get; } diff --git a/src/ServiceControl.Persistence/UnitOfWork/IngestionUnitOfWorkBase.cs b/src/ServiceControl.Persistence/UnitOfWork/IngestionUnitOfWorkBase.cs index e277b0491b..c348964646 100644 --- a/src/ServiceControl.Persistence/UnitOfWork/IngestionUnitOfWorkBase.cs +++ b/src/ServiceControl.Persistence/UnitOfWork/IngestionUnitOfWorkBase.cs @@ -1,4 +1,4 @@ -namespace ServiceControl.Persistence.UnitOfWork +namespace ServiceControl.Persistence.UnitOfWork { using System; using System.Threading; @@ -6,20 +6,16 @@ public abstract class IngestionUnitOfWorkBase : IIngestionUnitOfWork { - protected virtual void Dispose(bool disposing) - { - } + protected virtual ValueTask DisposeAsyncCore() => ValueTask.CompletedTask; - public void Dispose() + public async ValueTask DisposeAsync() { - Dispose(true); + await DisposeAsyncCore(); GC.SuppressFinalize(this); } - ~IngestionUnitOfWorkBase() => Dispose(false); - public IMonitoringIngestionUnitOfWork Monitoring { get; protected set; } public IRecoverabilityIngestionUnitOfWork Recoverability { get; protected set; } public virtual Task Complete(CancellationToken cancellationToken) => Task.CompletedTask; } -} \ No newline at end of file +} diff --git a/src/ServiceControl/Operations/ErrorIngestor.cs b/src/ServiceControl/Operations/ErrorIngestor.cs index 39f4f264f5..2f90579005 100644 --- a/src/ServiceControl/Operations/ErrorIngestor.cs +++ b/src/ServiceControl/Operations/ErrorIngestor.cs @@ -116,7 +116,7 @@ async Task> PersistFailedMessages(List