From c1ef28e60c8c92c43cd19be9a3dc1d0b5f18b693 Mon Sep 17 00:00:00 2001 From: Vulthil Date: Tue, 21 Jul 2026 16:54:03 +0200 Subject: [PATCH] test: cover the broker-outbox seam's filters, parallel dispatch, and end-to-end relay --- .../TransactionalOutboxIntegrationTests.cs | 71 +++++ .../Vulthil.IntegrationTests.csproj | 1 + .../BrokerOutboxDispatcherTests.cs | 122 ++++++++ .../TransactionalConsumeFilterTests.cs | 84 ++++++ .../TransactionalPublishFilterTests.cs | 68 +++++ ...xEngineServiceCollectionExtensionsTests.cs | 96 +++++++ .../OutboxProcessorMetricsTests.cs | 10 + .../OutboxProcessorTests.cs | 272 ++++++++++++++++++ .../OutboxRetentionBackgroundServiceTests.cs | 94 ++++++ 9 files changed, 818 insertions(+) create mode 100644 tests/Vulthil.Messaging.Outbox.Tests/TransactionalConsumeFilterTests.cs create mode 100644 tests/Vulthil.SharedKernel.Outbox.Tests/OutboxProcessorTests.cs diff --git a/tests/Vulthil.IntegrationTests/TransactionalOutboxIntegrationTests.cs b/tests/Vulthil.IntegrationTests/TransactionalOutboxIntegrationTests.cs index 127d5139..f0d13f34 100644 --- a/tests/Vulthil.IntegrationTests/TransactionalOutboxIntegrationTests.cs +++ b/tests/Vulthil.IntegrationTests/TransactionalOutboxIntegrationTests.cs @@ -1,8 +1,10 @@ using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; +using Vulthil.Extensions.Testing; using Vulthil.IntegrationTests.Fixtures; using Vulthil.Messaging.Abstractions.Publishers; using Vulthil.Messaging.TestHarness; +using Vulthil.Results; using Vulthil.SharedKernel.Outbox; using Vulthil.SharedKernel.Outbox.EntityFrameworkCore; using Vulthil.TestHost.Data; @@ -15,6 +17,7 @@ public sealed class TransactionalOutboxIntegrationTests(TestHarnessWebApplicatio : BaseIntegrationTestCase(factory, testOutputHelper), IClassFixture { private ITestHarness Harness => Factory.Services.GetRequiredService(); + private IRequester Requester => Factory.Services.GetRequiredService(); [Fact] public async Task PublishInsideATransactionIsCapturedIntoTheOutboxAndNotSentDirectly() @@ -44,4 +47,72 @@ await dbContext.ExecuteInTransactionAsync(async cancellationToken => capturedDuringTransaction.ShouldBeGreaterThan(0); sentDuringTransaction.ShouldBeFalse(); } + + [Fact] + public async Task CommittingATransactionRelaysTheMessageAndTheConsumerReceivesItWithTheStableMessageId() + { + // Arrange + var id = Guid.NewGuid(); + var messageId = Guid.NewGuid().ToString(); + var dbContext = ScopedServices.GetRequiredService(); + var publisher = ScopedServices.GetRequiredService(); + + // Act — publish inside a committed transaction, then wait for the real relay to dispatch it. + await dbContext.ExecuteInTransactionAsync(async cancellationToken => + { + await publisher.PublishAsync( + new ProbeCreatedIntegrationEvent(id), + context => + { + context.SetMessageId(messageId); + return ValueTask.CompletedTask; + }, + cancellationToken); + + return 0; + }, CancellationToken); + + var polled = await Polling.WaitAsync(TimeSpan.FromSeconds(15), async () => + { + var consumed = Harness.Consumed().FirstOrDefault(message => message.Message.Id == id); + return consumed is not null + ? Result.Success(consumed) + : Result.Failure>( + Error.NotFound("Probe.NotYetRelayed", "The captured message has not been relayed and consumed yet.")); + }, CancellationToken); + + // Assert + polled.IsSuccess.ShouldBeTrue(); + polled.Value.Envelope.MessageId.ShouldBe(messageId); + + var sideEffects = await Requester.RequestAsync>(new GetProbeSideEffects(id), CancellationToken); + sideEffects.IsSuccess.ShouldBeTrue(); + sideEffects.Value.ShouldContain(sideEffect => sideEffect.ProbeId == id); + } + + [Fact] + public async Task RollingBackATransactionNeverRelaysTheCapturedMessage() + { + // Arrange + var id = Guid.NewGuid(); + var dbContext = ScopedServices.GetRequiredService(); + var publisher = ScopedServices.GetRequiredService(); + + // Act — the operation reports failure, so ExecuteInTransactionAsync rolls back instead of committing. + await dbContext.ExecuteInTransactionAsync( + async cancellationToken => + { + await publisher.PublishAsync(new ProbeCreatedIntegrationEvent(id), cancellationToken); + return false; + }, + static operationSucceeded => operationSucceeded, + CancellationToken); + await Task.Delay(TimeSpan.FromSeconds(1), CancellationToken); + + // Assert — fetch every row and filter in memory: Content is mapped to jsonb, which Npgsql cannot LIKE-match server-side. + var outboxMessages = await dbContext.OutboxMessages.AsNoTracking().ToListAsync(CancellationToken); + outboxMessages.ShouldNotContain(message => message.Content.Contains(id.ToString())); + Harness.Published().ShouldNotContain(message => message.Message.Id == id); + Harness.Consumed().ShouldNotContain(message => message.Message.Id == id); + } } diff --git a/tests/Vulthil.IntegrationTests/Vulthil.IntegrationTests.csproj b/tests/Vulthil.IntegrationTests/Vulthil.IntegrationTests.csproj index bddca033..f32886b2 100644 --- a/tests/Vulthil.IntegrationTests/Vulthil.IntegrationTests.csproj +++ b/tests/Vulthil.IntegrationTests/Vulthil.IntegrationTests.csproj @@ -10,6 +10,7 @@ + diff --git a/tests/Vulthil.Messaging.Outbox.Tests/BrokerOutboxDispatcherTests.cs b/tests/Vulthil.Messaging.Outbox.Tests/BrokerOutboxDispatcherTests.cs index f98f24dd..4de706a9 100644 --- a/tests/Vulthil.Messaging.Outbox.Tests/BrokerOutboxDispatcherTests.cs +++ b/tests/Vulthil.Messaging.Outbox.Tests/BrokerOutboxDispatcherTests.cs @@ -1,3 +1,6 @@ +using System.Text.Json; +using Vulthil.Messaging.Abstractions.Publishers; +using Vulthil.Messaging.Transport; using Vulthil.SharedKernel.Outbox; using Vulthil.xUnit; @@ -26,4 +29,123 @@ public async Task DispatchAsyncThrowsDescriptiveErrorWhenMessageTypeCannotBeReso // Assert exception.Message.ShouldContain("Vulthil.Nonexistent.PhantomMessage"); } + + [Fact] + public async Task DispatchAsyncPublishesThePayloadWithTheCapturedMessageIdCorrelationIdRoutingKeyAndHeaders() + { + // Arrange + var jsonOptions = new JsonSerializerOptions(); + GetMock().Setup(provider => provider.JsonSerializerOptions).Returns(jsonOptions); + Func? capturedConfigure = null; + GetMock() + .Setup(publisher => publisher.PublishAsync(It.IsAny(), It.IsAny>(), It.IsAny())) + .Callback, CancellationToken>((_, configure, _) => capturedConfigure = configure) + .Returns(Task.CompletedTask); + var metadata = new BrokerOutboxMetadata + { + MessageId = "stable-message-id", + CorrelationId = "correlation-1", + RoutingKey = "routing-key-1", + Headers = new Dictionary { ["tenant"] = "acme" }, + }; + var dispatcher = CreateInstance(); + var message = new OutboxMessageData( + Id: Guid.NewGuid(), + Type: typeof(TestMessage).FullName!, + Content: JsonSerializer.Serialize(new TestMessage("hello"), jsonOptions), + TraceParent: null, + TraceState: null, + Destination: OutboxDestination.Publish, + Metadata: JsonSerializer.Serialize(metadata, jsonOptions)); + + // Act + await dispatcher.DispatchAsync(message, CancellationToken); + + // Assert + GetMock().Verify( + publisher => publisher.PublishAsync(It.Is(m => m.Value == "hello"), It.IsAny>(), CancellationToken), + Times.Once); + capturedConfigure.ShouldNotBeNull(); + var appliedContext = new PublishContext(); + await capturedConfigure!(appliedContext); + appliedContext.MessageId.ShouldBe("stable-message-id"); + appliedContext.CorrelationId.ShouldBe("correlation-1"); + appliedContext.RoutingKey.ShouldBe("routing-key-1"); + appliedContext.Headers["tenant"].ShouldBe("acme"); + } + + [Fact] + public async Task DispatchAsyncSendsThePayloadToTheCapturedDestinationAddressWithFidelity() + { + // Arrange + var jsonOptions = new JsonSerializerOptions(); + GetMock().Setup(provider => provider.JsonSerializerOptions).Returns(jsonOptions); + var sendEndpoint = new Mock(); + Func? capturedConfigure = null; + sendEndpoint + .Setup(endpoint => endpoint.SendAsync(It.IsAny(), It.IsAny>(), It.IsAny())) + .Callback, CancellationToken>((_, configure, _) => capturedConfigure = configure) + .Returns(Task.CompletedTask); + var destinationAddress = new Uri("queue:order-commands"); + GetMock() + .Setup(provider => provider.GetSendEndpointAsync(destinationAddress, It.IsAny())) + .ReturnsAsync(sendEndpoint.Object); + var metadata = new BrokerOutboxMetadata + { + MessageId = "stable-send-id", + DestinationAddress = destinationAddress.ToString(), + Headers = new Dictionary { ["tenant"] = "acme" }, + }; + var dispatcher = CreateInstance(); + var message = new OutboxMessageData( + Id: Guid.NewGuid(), + Type: typeof(TestMessage).FullName!, + Content: JsonSerializer.Serialize(new TestMessage("hello"), jsonOptions), + TraceParent: null, + TraceState: null, + Destination: OutboxDestination.Send, + Metadata: JsonSerializer.Serialize(metadata, jsonOptions)); + + // Act + await dispatcher.DispatchAsync(message, CancellationToken); + + // Assert + sendEndpoint.Verify( + endpoint => endpoint.SendAsync(It.Is(m => m.Value == "hello"), It.IsAny>(), CancellationToken), + Times.Once); + capturedConfigure.ShouldNotBeNull(); + var appliedContext = new PublishContext(); + await capturedConfigure!(appliedContext); + appliedContext.MessageId.ShouldBe("stable-send-id"); + appliedContext.Headers["tenant"].ShouldBe("acme"); + } + + [Fact] + public async Task DispatchAsyncThrowsDescriptiveErrorWhenASendMessageHasNoDestinationAddress() + { + // Arrange + var jsonOptions = new JsonSerializerOptions(); + GetMock().Setup(provider => provider.JsonSerializerOptions).Returns(jsonOptions); + var dispatcher = CreateInstance(); + var message = new OutboxMessageData( + Id: Guid.NewGuid(), + Type: typeof(TestMessage).FullName!, + Content: JsonSerializer.Serialize(new TestMessage("hello"), jsonOptions), + TraceParent: null, + TraceState: null, + Destination: OutboxDestination.Send, + Metadata: null); + + // Act + var exception = await Should.ThrowAsync( + () => dispatcher.DispatchAsync(message, CancellationToken)); + + // Assert + exception.Message.ShouldContain("missing its destination address"); + GetMock().Verify( + provider => provider.GetSendEndpointAsync(It.IsAny(), It.IsAny()), + Times.Never); + } + + private sealed record TestMessage(string Value); } diff --git a/tests/Vulthil.Messaging.Outbox.Tests/TransactionalConsumeFilterTests.cs b/tests/Vulthil.Messaging.Outbox.Tests/TransactionalConsumeFilterTests.cs new file mode 100644 index 00000000..fd9c35a3 --- /dev/null +++ b/tests/Vulthil.Messaging.Outbox.Tests/TransactionalConsumeFilterTests.cs @@ -0,0 +1,84 @@ +using Vulthil.Messaging.Abstractions.Consumers; +using Vulthil.Messaging.Transport; +using Vulthil.SharedKernel.Application.Data; +using Vulthil.xUnit; + +namespace Vulthil.Messaging.Outbox.Tests; + +public sealed class TransactionalConsumeFilterTests : BaseUnitTestCase +{ + [Fact] + public async Task ConsumeAsyncRunsTheNextDelegateInsideUnitOfWorkExecuteInTransactionAsync() + { + // Arrange + SetUpUnitOfWorkToInvokeTheOperation(); + var filter = CreateInstance>(); + var context = NewContext(); + var nextInvoked = false; + ConsumeDelegate next = receivedContext => + { + nextInvoked = ReferenceEquals(receivedContext, context); + return Task.CompletedTask; + }; + + // Act + await filter.ConsumeAsync(context, next); + + // Assert + nextInvoked.ShouldBeTrue(); + GetMock().Verify( + unitOfWork => unitOfWork.ExecuteInTransactionAsync(It.IsAny>>(), context.CancellationToken), + Times.Once); + } + + [Fact] + public async Task ConsumeAsyncNeverRunsTheNextDelegateWhenTheUnitOfWorkDoesNotInvokeTheOperation() + { + // Arrange + GetMock() + .Setup(unitOfWork => unitOfWork.ExecuteInTransactionAsync(It.IsAny>>(), It.IsAny())) + .ReturnsAsync(false); + var filter = CreateInstance>(); + var context = NewContext(); + var nextInvoked = false; + ConsumeDelegate next = _ => + { + nextInvoked = true; + return Task.CompletedTask; + }; + + // Act + await filter.ConsumeAsync(context, next); + + // Assert + nextInvoked.ShouldBeFalse(); + } + + [Fact] + public async Task ConsumeAsyncPropagatesAnExceptionThrownByTheNextDelegate() + { + // Arrange + SetUpUnitOfWorkToInvokeTheOperation(); + var filter = CreateInstance>(); + var context = NewContext(); + ConsumeDelegate next = _ => throw new InvalidOperationException("handler failed"); + + // Act & Assert + await Should.ThrowAsync(() => filter.ConsumeAsync(context, next)); + } + + private void SetUpUnitOfWorkToInvokeTheOperation() => + GetMock() + .Setup(unitOfWork => unitOfWork.ExecuteInTransactionAsync(It.IsAny>>(), It.IsAny())) + .Returns((Func> operation, CancellationToken token) => operation(token)); + + private static MessageContext NewContext() => new() + { + Message = new TestMessage("hello"), + RoutingKey = string.Empty, + Headers = new Dictionary(), + CancellationToken = CancellationToken, + }; + + private sealed record TestMessage(string Value); +} diff --git a/tests/Vulthil.Messaging.Outbox.Tests/TransactionalPublishFilterTests.cs b/tests/Vulthil.Messaging.Outbox.Tests/TransactionalPublishFilterTests.cs index dbfa67ce..481acf21 100644 --- a/tests/Vulthil.Messaging.Outbox.Tests/TransactionalPublishFilterTests.cs +++ b/tests/Vulthil.Messaging.Outbox.Tests/TransactionalPublishFilterTests.cs @@ -1,3 +1,4 @@ +using System.Diagnostics; using System.Text.Json; using System.Transactions; using Microsoft.Extensions.Options; @@ -88,6 +89,73 @@ public async Task PublishingInsideAnEntityFrameworkTransactionCapturesTheMessage GetMock().Verify(store => store.SaveChangesAsync(It.IsAny()), Times.Once); } + [Fact] + public async Task EnableTracingCapturesTheCurrentActivityOntoTheCapturedRow() + { + // Arrange + GetMock().Setup(store => store.IsInTransaction).Returns(true); + Use>(Options.Create(new OutboxProcessingOptions { EnableTracing = true })); + Use(TimeProvider.System); + GetMock().Setup(provider => provider.JsonSerializerOptions).Returns(new JsonSerializerOptions()); + var filter = CreateInstance(); + var context = NewContext(); + OutboxMessage? captured = null; + GetMock() + .Setup(store => store.AddOutboxMessage(It.IsAny())) + .Callback(message => captured = message); + using var activity = new Activity("test-publish"); + activity.TraceStateString = "congo=t61rcWkgMzE"; + + // Act + activity.Start(); + try + { + await filter.PublishAsync(context, _ => Task.CompletedTask); + } + finally + { + activity.Stop(); + } + + // Assert + captured.ShouldNotBeNull(); + captured.TraceParent.ShouldBe(activity.Id); + captured.TraceState.ShouldBe(activity.TraceStateString); + } + + [Fact] + public async Task EnableTracingDisabledLeavesTraceFieldsNullEvenWithAnActiveActivity() + { + // Arrange + GetMock().Setup(store => store.IsInTransaction).Returns(true); + Use>(Options.Create(new OutboxProcessingOptions { EnableTracing = false })); + Use(TimeProvider.System); + GetMock().Setup(provider => provider.JsonSerializerOptions).Returns(new JsonSerializerOptions()); + var filter = CreateInstance(); + var context = NewContext(); + OutboxMessage? captured = null; + GetMock() + .Setup(store => store.AddOutboxMessage(It.IsAny())) + .Callback(message => captured = message); + using var activity = new Activity("test-publish"); + + // Act + activity.Start(); + try + { + await filter.PublishAsync(context, _ => Task.CompletedTask); + } + finally + { + activity.Stop(); + } + + // Assert + captured.ShouldNotBeNull(); + captured.TraceParent.ShouldBeNull(); + captured.TraceState.ShouldBeNull(); + } + private static PublishFilterContext NewContext() => new() { Message = new TestMessage("hello"), diff --git a/tests/Vulthil.SharedKernel.Outbox.Tests/OutboxEngineServiceCollectionExtensionsTests.cs b/tests/Vulthil.SharedKernel.Outbox.Tests/OutboxEngineServiceCollectionExtensionsTests.cs index 34d2cace..49fbdab2 100644 --- a/tests/Vulthil.SharedKernel.Outbox.Tests/OutboxEngineServiceCollectionExtensionsTests.cs +++ b/tests/Vulthil.SharedKernel.Outbox.Tests/OutboxEngineServiceCollectionExtensionsTests.cs @@ -1,6 +1,8 @@ using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Options; +using OpenTelemetry.Metrics; +using OpenTelemetry.Trace; using Vulthil.xUnit; namespace Vulthil.SharedKernel.Outbox.Tests; @@ -89,6 +91,100 @@ public void CallingAddOutboxEngineTwiceDoesNotDuplicateHostedServices() services.Count(descriptor => descriptor.ServiceType == typeof(OutboxProcessor)).ShouldBe(1); } + [Fact] + public void EnableMetricsRegistersTheMeterProviderService() + { + // Arrange + var services = new ServiceCollection(); + + // Act + services.AddOutboxEngine(o => o.EnableMetrics = true); + + // Assert + services.ShouldContain(descriptor => descriptor.ServiceType == typeof(MeterProvider)); + } + + [Fact] + public void EnableMetricsDisabledSkipsMeterProviderRegistration() + { + // Arrange + var services = new ServiceCollection(); + + // Act + services.AddOutboxEngine(o => o.EnableMetrics = false); + + // Assert + services.ShouldNotContain(descriptor => descriptor.ServiceType == typeof(MeterProvider)); + } + + [Fact] + public void EnableTracingRegistersTheTracerProviderService() + { + // Arrange + var services = new ServiceCollection(); + + // Act + services.AddOutboxEngine(o => o.EnableTracing = true); + + // Assert + services.ShouldContain(descriptor => descriptor.ServiceType == typeof(TracerProvider)); + } + + [Fact] + public void EnableTracingDisabledSkipsTracerProviderRegistration() + { + // Arrange + var services = new ServiceCollection(); + + // Act + services.AddOutboxEngine(o => o.EnableTracing = false); + + // Assert + services.ShouldNotContain(descriptor => descriptor.ServiceType == typeof(TracerProvider)); + } + + [Fact] + public void RetentionEnabledRegistersTheRetentionBackgroundService() + { + // Arrange + var services = new ServiceCollection(); + + // Act + services.AddOutboxEngine(o => o.Retention.Enabled = true); + + // Assert + services.ShouldContain(descriptor => descriptor.ServiceType == typeof(IHostedService) && descriptor.ImplementationType == typeof(OutboxRetentionBackgroundService)); + } + + [Fact] + public void RetentionDisabledDoesNotRegisterTheRetentionBackgroundService() + { + // Arrange + var services = new ServiceCollection(); + + // Act + services.AddOutboxEngine(o => o.Retention.Enabled = false); + + // Assert + services.ShouldNotContain(descriptor => descriptor.ServiceType == typeof(IHostedService) && descriptor.ImplementationType == typeof(OutboxRetentionBackgroundService)); + } + + [Fact] + public void RetentionEnabledWithAZeroRetentionPeriodFailsValidationAtStartup() + { + // Arrange + var services = new ServiceCollection(); + services.AddOutboxEngine(o => + { + o.Retention.Enabled = true; + o.Retention.RetentionPeriod = TimeSpan.Zero; + }); + using var provider = services.BuildServiceProvider(); + + // Act & Assert + Should.Throw(() => provider.GetRequiredService>().Value); + } + private sealed class FakeOutboxStore : IOutboxStore { public Type ContextType { get; } = typeof(TContext); diff --git a/tests/Vulthil.SharedKernel.Outbox.Tests/OutboxProcessorMetricsTests.cs b/tests/Vulthil.SharedKernel.Outbox.Tests/OutboxProcessorMetricsTests.cs index 9f1793d2..84c94bca 100644 --- a/tests/Vulthil.SharedKernel.Outbox.Tests/OutboxProcessorMetricsTests.cs +++ b/tests/Vulthil.SharedKernel.Outbox.Tests/OutboxProcessorMetricsTests.cs @@ -5,6 +5,16 @@ namespace Vulthil.SharedKernel.Outbox.Tests; +/// +/// Groups every test that measures the process-wide counters within a narrow +/// listening window, so xUnit never runs them concurrently with each other — a real +/// dispatch anywhere else in the process (e.g. ) increments the same static +/// counters and would otherwise be misattributed to a concurrently-running measurement. +/// +[CollectionDefinition(nameof(OutboxTelemetryCollection))] +public sealed class OutboxTelemetryCollection; + +[Collection(nameof(OutboxTelemetryCollection))] public sealed class OutboxProcessorMetricsTests : BaseUnitTestCase { private static readonly OutboxMessageData Message = new( diff --git a/tests/Vulthil.SharedKernel.Outbox.Tests/OutboxProcessorTests.cs b/tests/Vulthil.SharedKernel.Outbox.Tests/OutboxProcessorTests.cs new file mode 100644 index 00000000..d9fddcf1 --- /dev/null +++ b/tests/Vulthil.SharedKernel.Outbox.Tests/OutboxProcessorTests.cs @@ -0,0 +1,272 @@ +using System.Collections.Concurrent; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; +using Vulthil.xUnit; + +namespace Vulthil.SharedKernel.Outbox.Tests; + +/// +/// Shares with : every test here +/// drives a real to a successful or failed dispatch, incrementing the same +/// process-wide counters that class measures within a narrow window. +/// +[Collection(nameof(OutboxTelemetryCollection))] +public sealed class OutboxProcessorTests : BaseUnitTestCase +{ + private readonly Lazy _lazyTarget; + + private OutboxProcessor Target => _lazyTarget.Value; + + public OutboxProcessorTests() => _lazyTarget = new(CreateInstance); + + [Fact] + public async Task ParallelPublishingDisabledDispatchesOnTheRootServiceProviderWithoutCreatingAScope() + { + // Arrange + Use>(Options.Create(new OutboxProcessingOptions { EnableParallelPublishing = false })); + var dispatcher = new RecordingDispatcher(); + GetMock().Setup(sp => sp.GetService(typeof(IEnumerable))).Returns(new IOutboxDispatcher[] { dispatcher }); + Use(new FakeParallelOutboxStore(Messages(1), maxDegreeOfParallelism: null)); + + // Act + var processed = await Target.ExecuteAsync(CancellationToken); + + // Assert + processed.ShouldBe(1); + dispatcher.CallCount.ShouldBe(1); + GetMock().Verify(factory => factory.CreateScope(), Times.Never); + } + + [Fact] + public async Task ParallelPublishingEnabledDispatchesEachMessageOnAnIsolatedDependencyInjectionScope() + { + // Arrange + const int messageCount = 3; + Use>(Options.Create(new OutboxProcessingOptions { EnableParallelPublishing = true })); + var scopedDispatchers = new ConcurrentBag(); + var scopeFactory = new RecordingServiceScopeFactory(() => + { + var dispatcher = new RecordingDispatcher(); + scopedDispatchers.Add(dispatcher); + return new SingleDispatcherServiceProvider(dispatcher); + }); + Use(scopeFactory); + Use(new FakeParallelOutboxStore(Messages(messageCount), messageCount)); + + // Act + var processed = await Target.ExecuteAsync(CancellationToken); + + // Assert + processed.ShouldBe(messageCount); + scopeFactory.ScopesCreated.ShouldBe(messageCount); + scopedDispatchers.Count.ShouldBe(messageCount); + scopedDispatchers.ShouldAllBe(dispatcher => dispatcher.CallCount == 1); + } + + [Fact] + public async Task MaxDegreeOfParallelismBoundsTheNumberOfConcurrentDispatches() + { + // Arrange + const int maxDegreeOfParallelism = 2; + const int messageCount = 6; + Use>(Options.Create(new OutboxProcessingOptions + { + EnableParallelPublishing = true, + MaxDegreeOfParallelism = maxDegreeOfParallelism, + })); + var dispatcher = new ConcurrencyTrackingDispatcher(); + Use(new RecordingServiceScopeFactory(() => new SingleDispatcherServiceProvider(dispatcher))); + Use(new FakeParallelOutboxStore(Messages(messageCount), maxDegreeOfParallelism)); + + // Act + var processed = await Target.ExecuteAsync(CancellationToken); + + // Assert + processed.ShouldBe(messageCount); + dispatcher.MaxObservedConcurrency.ShouldBeLessThanOrEqualTo(maxDegreeOfParallelism); + dispatcher.MaxObservedConcurrency.ShouldBe(maxDegreeOfParallelism); + } + + [Fact] + public async Task AFailingParallelDispatchDoesNotCorruptTheOtherDispatchesOutcomes() + { + // Arrange + var messages = Messages(3); + var failingMessageId = messages[1].Id; + Use>(Options.Create(new OutboxProcessingOptions + { + EnableParallelPublishing = true, + MaxDegreeOfParallelism = 3, + })); + Use(new RecordingServiceScopeFactory(() => new SingleDispatcherServiceProvider(new FailingForOneMessageDispatcher(failingMessageId)))); + var store = new FakeParallelOutboxStore(messages, 3); + Use(store); + + // Act + var processed = await Target.ExecuteAsync(CancellationToken); + + // Assert + processed.ShouldBe(2); + store.Outcomes[failingMessageId].ShouldNotBeNull(); + store.Outcomes.Where(outcome => outcome.Key != failingMessageId).ShouldAllBe(outcome => outcome.Value == null); + } + + private static List Messages(int count) => + Enumerable.Range(0, count) + .Select(_ => new OutboxMessageData(Guid.NewGuid(), "Some.Event", "{}", null, null, OutboxDestination.DomainEvent, null)) + .ToList(); + + private sealed class RecordingServiceScopeFactory(Func scopedProviderFactory) : IServiceScopeFactory + { + private int _scopesCreated; + + public int ScopesCreated => _scopesCreated; + + public IServiceScope CreateScope() + { + Interlocked.Increment(ref _scopesCreated); + return new FakeScope(scopedProviderFactory()); + } + + private sealed class FakeScope(IServiceProvider serviceProvider) : IServiceScope + { + public IServiceProvider ServiceProvider { get; } = serviceProvider; + + public void Dispose() + { + } + } + } + + /// Resolves a single, fixed — stands in for one isolated DI scope's container. + private sealed class SingleDispatcherServiceProvider(IOutboxDispatcher dispatcher) : IServiceProvider + { + public object? GetService(Type serviceType) => + serviceType == typeof(IEnumerable) ? new[] { dispatcher } : null; + } + + private sealed class RecordingDispatcher : IOutboxDispatcher + { + private int _callCount; + + public int CallCount => _callCount; + + public bool Handles(OutboxDestination destination) => true; + + public Task DispatchAsync(OutboxMessageData message, CancellationToken cancellationToken) + { + Interlocked.Increment(ref _callCount); + return Task.CompletedTask; + } + } + + /// Tracks concurrent in-flight calls to observe the peak concurrency a caller allowed. + private sealed class ConcurrencyTrackingDispatcher : IOutboxDispatcher + { + private int _active; + private int _maxObservedConcurrency; + + public int MaxObservedConcurrency => _maxObservedConcurrency; + + public bool Handles(OutboxDestination destination) => true; + + public async Task DispatchAsync(OutboxMessageData message, CancellationToken cancellationToken) + { + var current = Interlocked.Increment(ref _active); + TrackMax(current); + try + { + await Task.Delay(TimeSpan.FromMilliseconds(100), cancellationToken); + } + finally + { + Interlocked.Decrement(ref _active); + } + } + + private void TrackMax(int observed) + { + int current; + do + { + current = Volatile.Read(ref _maxObservedConcurrency); + if (observed <= current) + { + return; + } + } + while (Interlocked.CompareExchange(ref _maxObservedConcurrency, observed, current) != current); + } + } + + private sealed class FailingForOneMessageDispatcher(Guid failingMessageId) : IOutboxDispatcher + { + public bool Handles(OutboxDestination destination) => true; + + public Task DispatchAsync(OutboxMessageData message, CancellationToken cancellationToken) => + message.Id == failingMessageId + ? throw new InvalidOperationException("Simulated dispatch failure.") + : Task.CompletedTask; + } + + /// + /// Fake that mirrors EntityFrameworkOutboxStore's + /// EnableParallelPublishing/MaxDegreeOfParallelism semaphore-throttled dispatch, so + /// can be exercised under the same concurrency contract the real store applies + /// without depending on the EF Core store project. + /// + private sealed class FakeParallelOutboxStore(IReadOnlyList messages, int? maxDegreeOfParallelism) : IOutboxStore + { + public ConcurrentDictionary Outcomes { get; } = new(); + + public bool IsInTransaction => false; + + public void AddOutboxMessage(OutboxMessage message) + { + } + + public Task SaveChangesAsync(CancellationToken cancellationToken = default) => Task.FromResult(0); + + public async Task ProcessBatchAsync(Func> dispatch, CancellationToken cancellationToken) + { + if (maxDegreeOfParallelism is not { } max) + { + var successCount = 0; + foreach (var message in messages) + { + var error = await dispatch(message, cancellationToken); + Outcomes[message.Id] = error; + successCount += error is null ? 1 : 0; + } + + return successCount; + } + + using var throttle = new SemaphoreSlim(max); + var results = await Task.WhenAll(messages.Select(message => DispatchThrottledAsync(message, dispatch, throttle, cancellationToken))); + foreach (var (id, error) in results) + { + Outcomes[id] = error; + } + + return results.Count(result => result.Error is null); + } + + private static async Task<(Guid Id, string? Error)> DispatchThrottledAsync( + OutboxMessageData message, + Func> dispatch, + SemaphoreSlim throttle, + CancellationToken cancellationToken) + { + await throttle.WaitAsync(cancellationToken); + try + { + return (message.Id, await dispatch(message, cancellationToken)); + } + finally + { + throttle.Release(); + } + } + } +} diff --git a/tests/Vulthil.SharedKernel.Outbox.Tests/OutboxRetentionBackgroundServiceTests.cs b/tests/Vulthil.SharedKernel.Outbox.Tests/OutboxRetentionBackgroundServiceTests.cs index 419a7cdd..bd554949 100644 --- a/tests/Vulthil.SharedKernel.Outbox.Tests/OutboxRetentionBackgroundServiceTests.cs +++ b/tests/Vulthil.SharedKernel.Outbox.Tests/OutboxRetentionBackgroundServiceTests.cs @@ -57,6 +57,100 @@ public async Task AForeignCancellationDuringASweepDoesNotEndTheExecuteTask() Target.ExecuteTask!.Status.ShouldBe(TaskStatus.RanToCompletion); } + [Fact] + public async Task TheSweepPassesACutoffComputedFromTheConfiguredRetentionPeriod() + { + // Arrange + var now = new DateTimeOffset(2026, 1, 15, 12, 0, 0, TimeSpan.Zero); + var retentionPeriod = TimeSpan.FromDays(3); + Use(new FixedTimeProvider(now)); + Use>(Options.Create(new OutboxProcessingOptions + { + Retention = { SweepInterval = TimeSpan.FromSeconds(60), RetentionPeriod = retentionPeriod } + })); + Use(new AutoMockerServiceScopeFactory(AutoMocker)); + var store = new RecordingRetentionStore([0]); + Use(store); + + // Act + await Target.StartAsync(CancellationToken); + await store.LastCallCompleted; + await Target.StopAsync(CancellationToken); + + // Assert + store.ObservedCutoffs.ShouldHaveSingleItem().ShouldBe(now - retentionPeriod); + } + + [Fact] + public async Task TheSweepKeepsDeletingBatchesUntilFewerThanBatchSizeRowsRemain() + { + // Arrange + const int batchSize = 5; + Use>(Options.Create(new OutboxProcessingOptions + { + Retention = { SweepInterval = TimeSpan.FromSeconds(60), BatchSize = batchSize } + })); + Use(new AutoMockerServiceScopeFactory(AutoMocker)); + var store = new RecordingRetentionStore([batchSize, batchSize, 2]); + Use(store); + + // Act + await Target.StartAsync(CancellationToken); + await store.LastCallCompleted; + await Target.StopAsync(CancellationToken); + + // Assert + store.CallCount.ShouldBe(3); + store.ObservedBatchSizes.ShouldAllBe(size => size == batchSize); + } + + private sealed class FixedTimeProvider(DateTimeOffset now) : TimeProvider + { + public override DateTimeOffset GetUtcNow() => now; + } + + /// Records every call's cutoff/batch-size and replays a fixed result sequence. + private sealed class RecordingRetentionStore(IReadOnlyList resultsInOrder) : IOutboxStore, IOutboxRetentionStore + { + private readonly TaskCompletionSource _lastCallCompleted = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly List _observedCutoffs = []; + private readonly List _observedBatchSizes = []; + private int _callCount; + + public Task LastCallCompleted => _lastCallCompleted.Task; + + public IReadOnlyList ObservedCutoffs => _observedCutoffs; + + public IReadOnlyList ObservedBatchSizes => _observedBatchSizes; + + public int CallCount => _callCount; + + public bool IsInTransaction => false; + + public void AddOutboxMessage(OutboxMessage message) + { + } + + public Task SaveChangesAsync(CancellationToken cancellationToken = default) => Task.FromResult(0); + + public Task ProcessBatchAsync(Func> dispatch, CancellationToken cancellationToken) => Task.FromResult(0); + + public Task DeleteProcessedAsync(DateTimeOffset olderThanUtc, int batchSize, CancellationToken cancellationToken) + { + var index = Interlocked.Increment(ref _callCount) - 1; + _observedCutoffs.Add(olderThanUtc); + _observedBatchSizes.Add(batchSize); + + var result = index < resultsInOrder.Count ? resultsInOrder[index] : resultsInOrder[^1]; + if (index == resultsInOrder.Count - 1) + { + _lastCallCompleted.TrySetResult(); + } + + return Task.FromResult(result); + } + } + private sealed class TimeoutThrowingRetentionStore : IOutboxStore, IOutboxRetentionStore { private readonly TaskCompletionSource _firstSweepAttempted = new(TaskCreationOptions.RunContinuationsAsynchronously);