Integrate EF Core into ingestion unit of work

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<MinimumRequiredStorageState>();
 
         services.AddSingleton<IServiceControlSubscriptionStorage, SubscriptionStorage>();
         services.AddSingleton<ISubscriptionStorage>(p => p.GetRequiredService<IServiceControlSubscriptionStorage>());

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

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<IIngestionUnitOfWork> StartNew() =>
-        throw new NotImplementedException();
+    public ValueTask<IIngestionUnitOfWork> StartNew()
+    {
+        var scope = serviceProvider.CreateAsyncScope();
+        var dbContext = scope.ServiceProvider.GetRequiredService<ServiceControlDbContext>();
+        var settings = scope.ServiceProvider.GetRequiredService<EFPersisterSettings>();
+        var unitOfWork = new EFIngestionUnitOfWork(scope, dbContext, storagePersistence, settings);
+        return ValueTask.FromResult<IIngestionUnitOfWork>(unitOfWork);
+    }
 
-    public bool CanIngestMore() =>
-        throw new NotImplementedException();
+    public bool CanIngestMore() => storageState.CanIngestMore;
 }

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<byte> body, string contentType, CancellationToken cancellationToken = default);
+    Task<MessageBodyFileResult?> ReadBody(string bodyId, DateTime createdOn, CancellationToken cancellationToken = default);
+    Task DeleteBody(string bodyId, CancellationToken cancellationToken = default);
+}

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!;
+}

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);
 

src/ServiceControl.Persistence.Tests/BodyStorage/AttachmentsBodyStorageTests.cs

@@ -44,7 +44,7 @@ async Task RunTest(Func<Dictionary<string, string>, 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

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);
 

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

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; }

src/ServiceControl.Persistence/UnitOfWork/IngestionUnitOfWorkBase.cs

@@ -1,25 +1,21 @@
-namespace ServiceControl.Persistence.UnitOfWork
+namespace ServiceControl.Persistence.UnitOfWork
 {
     using System;
     using System.Threading;
     using System.Threading.Tasks;
 
     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;
     }
-}
+}

src/ServiceControl/Operations/ErrorIngestor.cs

@@ -116,7 +116,7 @@ async Task<IReadOnlyList<MessageContext>> PersistFailedMessages(List<MessageCont
 
             try
             {
-                using var unitOfWork = await unitOfWorkFactory.StartNew();
+                await using var unitOfWork = await unitOfWorkFactory.StartNew();
                 var storedFailedMessageContexts = await errorProcessor.Process(failedMessageContexts, unitOfWork);
                 await retryConfirmationProcessor.Process(retriedMessageContexts, unitOfWork);