Skip to content

Commit b19d9fd

Browse files
VulthilVulthil
authored
Merge pull request #292 from Vulthil/fix/outbox-relay-retry-and-cancellation
Harden outbox relay retry, cancellation, and retention-store diagnostics
2 parents bc22461 + 8ff7d5b commit b19d9fd

10 files changed

Lines changed: 327 additions & 3 deletions

File tree

‎src/Vulthil.Messaging.Inbox/InboxRetentionBackgroundService.cs‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ internal sealed class InboxRetentionBackgroundService(
1818
ILogger<InboxRetentionBackgroundService> logger) : BackgroundService
1919
{
2020
private readonly InboxRetentionOptions _options = options.Value.Retention;
21+
private bool _loggedMissingRetentionStore;
2122

2223
/// <inheritdoc />
2324
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
@@ -47,6 +48,12 @@ private async Task SweepAsync(CancellationToken cancellationToken)
4748
await using var scope = scopeFactory.CreateAsyncScope();
4849
if (scope.ServiceProvider.GetRequiredService<IIdempotencyStore>() is not IInboxRetentionStore store)
4950
{
51+
if (!_loggedMissingRetentionStore)
52+
{
53+
_loggedMissingRetentionStore = true;
54+
logger.LogWarning("Inbox retention is enabled, but the registered IIdempotencyStore does not implement IInboxRetentionStore; the retention sweep will not run.");
55+
}
56+
5057
return;
5158
}
5259

‎src/Vulthil.SharedKernel.Outbox.EntityFrameworkCore/EntityFrameworkOutboxStore.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -111,7 +111,7 @@ private async Task<int> ProcessBatchCoreAsync(Func<OutboxMessageData, Cancellati
111111
await transaction.CommitAsync(cancellationToken);
112112
}
113113

114-
return messages.Count;
114+
return successIds.Count;
115115
}
116116
finally
117117
{

‎src/Vulthil.SharedKernel.Outbox/IOutboxStore.cs‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,11 @@ public interface IOutboxStore
2323
/// against the row as a failed attempt.
2424
/// </param>
2525
/// <param name="cancellationToken">A token to observe for cancellation.</param>
26-
/// <returns>The number of messages fetched and processed in the batch.</returns>
26+
/// <returns>
27+
/// The number of messages <paramref name="dispatch"/> reported as successful. A caller uses this to decide
28+
/// whether more work is likely waiting (the count reaches the configured batch size) — a batch that included
29+
/// failures must not be mistaken for one that is exhausted.
30+
/// </returns>
2731
Task<int> ProcessBatchAsync(Func<OutboxMessageData, CancellationToken, Task<string?>> dispatch, CancellationToken cancellationToken);
2832

2933
/// <summary>

‎src/Vulthil.SharedKernel.Outbox/OutboxProcessor.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,7 @@ internal Task<int> ExecuteAsync(CancellationToken cancellationToken) =>
4040
Telemetry.Relayed.Add(1);
4141
return null;
4242
}
43-
catch (Exception exception)
43+
catch (Exception exception) when (exception is not OperationCanceledException || !cancellationToken.IsCancellationRequested)
4444
{
4545
logger.LogError(exception, "Failed to publish outbox message {MessageId}", outboxMessage.Id);
4646
Telemetry.Failed.Add(1);

‎src/Vulthil.SharedKernel.Outbox/OutboxRetentionBackgroundService.cs‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ internal sealed class OutboxRetentionBackgroundService(
1818
ILogger<OutboxRetentionBackgroundService> logger) : BackgroundService
1919
{
2020
private readonly OutboxRetentionOptions _options = options.Value.Retention;
21+
private bool _loggedMissingRetentionStore;
2122

2223
/// <inheritdoc />
2324
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
@@ -47,6 +48,12 @@ private async Task SweepAsync(CancellationToken cancellationToken)
4748
await using var scope = scopeFactory.CreateAsyncScope();
4849
if (scope.ServiceProvider.GetRequiredService<IOutboxStore>() is not IOutboxRetentionStore store)
4950
{
51+
if (!_loggedMissingRetentionStore)
52+
{
53+
_loggedMissingRetentionStore = true;
54+
logger.LogWarning("Outbox retention is enabled, but the registered IOutboxStore does not implement IOutboxRetentionStore; the retention sweep will not run.");
55+
}
56+
5057
return;
5158
}
5259

‎tests/Vulthil.Messaging.Inbox.Tests/InboxRetentionBackgroundServiceTests.cs‎

Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,79 @@
1+
using Microsoft.Extensions.DependencyInjection;
2+
using Microsoft.Extensions.Logging;
3+
using Microsoft.Extensions.Options;
4+
using Vulthil.xUnit;
5+
6+
namespace Vulthil.Messaging.Inbox.Tests;
7+
8+
public sealed class InboxRetentionBackgroundServiceTests : BaseUnitTestCase
9+
{
10+
private readonly Lazy<InboxRetentionBackgroundService> _lazyTarget;
11+
12+
private InboxRetentionBackgroundService Target => _lazyTarget.Value;
13+
14+
public InboxRetentionBackgroundServiceTests()
15+
{
16+
_lazyTarget = new(CreateInstance<InboxRetentionBackgroundService>);
17+
Use(TimeProvider.System);
18+
}
19+
20+
[Fact]
21+
public async Task StoreWithoutRetentionSupportLogsExactlyOneWarningAcrossMultipleSweeps()
22+
{
23+
// Arrange
24+
Use<IOptions<InboxOptions>>(Options.Create(new InboxOptions
25+
{
26+
Retention = { SweepInterval = TimeSpan.FromMilliseconds(5) }
27+
}));
28+
Use<IServiceScopeFactory>(new AutoMockerServiceScopeFactory(AutoMocker));
29+
var logger = new WarningCountingLogger();
30+
Use<ILogger<InboxRetentionBackgroundService>>(logger);
31+
32+
// Act
33+
await Target.StartAsync(CancellationToken);
34+
await Task.WhenAny(logger.FirstWarning, Task.Delay(TimeSpan.FromSeconds(5), CancellationToken));
35+
await Task.Delay(TimeSpan.FromMilliseconds(100), CancellationToken);
36+
await Target.StopAsync(CancellationToken);
37+
38+
// Assert
39+
logger.WarningCount.ShouldBe(1);
40+
}
41+
42+
private sealed class WarningCountingLogger : ILogger<InboxRetentionBackgroundService>
43+
{
44+
private readonly TaskCompletionSource _firstWarning = new(TaskCreationOptions.RunContinuationsAsynchronously);
45+
46+
public int WarningCount { get; private set; }
47+
48+
public Task FirstWarning => _firstWarning.Task;
49+
50+
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
51+
52+
public bool IsEnabled(LogLevel logLevel) => true;
53+
54+
public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func<TState, Exception?, string> formatter)
55+
{
56+
if (logLevel != LogLevel.Warning)
57+
{
58+
return;
59+
}
60+
61+
WarningCount++;
62+
_firstWarning.TrySetResult();
63+
}
64+
}
65+
66+
private sealed class AutoMockerServiceScopeFactory(IServiceProvider serviceProvider) : IServiceScopeFactory
67+
{
68+
public IServiceScope CreateScope() => new PassthroughServiceScope(serviceProvider);
69+
70+
private sealed class PassthroughServiceScope(IServiceProvider serviceProvider) : IServiceScope
71+
{
72+
public IServiceProvider ServiceProvider { get; } = serviceProvider;
73+
74+
public void Dispose()
75+
{
76+
}
77+
}
78+
}
79+
}

‎tests/Vulthil.SharedKernel.Outbox.EntityFrameworkCore.Tests/EntityFrameworkOutboxStoreTests.cs‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,29 @@ public async Task DoesNotFetchADeadLetteredMessage()
6464
dispatched.ShouldBeEmpty();
6565
}
6666

67+
[Fact]
68+
public async Task ReturnsOnlyTheSuccessfullyDispatchedCountWhenTheBatchHasFailures()
69+
{
70+
// Arrange
71+
await using var seed = NewContext();
72+
seed.OutboxMessages.Add(NewMessage(Guid.CreateVersion7(), DateTimeOffset.UtcNow));
73+
seed.OutboxMessages.Add(NewMessage(Guid.CreateVersion7(), DateTimeOffset.UtcNow));
74+
await seed.SaveChangesAsync(CancellationToken);
75+
await using var context = NewContext();
76+
var store = NewStore(context, maxRetries: 3);
77+
var dispatchCount = 0;
78+
79+
// Act
80+
var processed = await store.ProcessBatchAsync((_, _) =>
81+
{
82+
dispatchCount++;
83+
return Task.FromResult<string?>(dispatchCount == 1 ? "boom" : null);
84+
}, CancellationToken);
85+
86+
// Assert
87+
processed.ShouldBe(1);
88+
}
89+
6790
[Fact]
6891
public async Task FetchesMessagesOrderedByOccurredOnThenId()
6992
{

‎tests/Vulthil.SharedKernel.Outbox.Tests/OutboxBackgroundServiceTests.cs‎

Lines changed: 94 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
using Microsoft.Extensions.DependencyInjection;
12
using Microsoft.Extensions.Options;
23
using Vulthil.xUnit;
34

@@ -31,6 +32,59 @@ public async Task StoppingWhileWaitingForRelayGatesCompletesGracefully()
3132
Target.ExecuteTask!.Status.ShouldBe(TaskStatus.RanToCompletion);
3233
}
3334

35+
[Fact]
36+
public async Task ABatchWithFewerSuccessesThanTheBatchSizeDelaysTheNextFetch()
37+
{
38+
// Arrange
39+
const int batchSize = 10;
40+
const int baseDelaySeconds = 2;
41+
Use<IOptions<OutboxProcessingOptions>>(Options.Create(new OutboxProcessingOptions
42+
{
43+
BatchSize = batchSize,
44+
OutboxProcessingDelaySeconds = baseDelaySeconds
45+
}));
46+
Use<IEnumerable<IOutboxRelayGate>>([]);
47+
Use<IServiceScopeFactory>(new AutoMockerServiceScopeFactory(AutoMocker));
48+
var store = new CountingOutboxStore(result: batchSize - 3);
49+
Use<IOutboxStore>(store);
50+
Use(CreateInstance<OutboxProcessor>());
51+
TimeSpan? capturedDelay = null;
52+
GetMock<IOutboxSignal>()
53+
.Setup(signal => signal.WaitAsync(It.IsAny<TimeSpan>(), It.IsAny<CancellationToken>()))
54+
.Callback<TimeSpan, CancellationToken>((delay, _) => capturedDelay ??= delay)
55+
.Returns(Task.CompletedTask);
56+
57+
// Act
58+
await Target.StartAsync(CancellationToken);
59+
await store.SecondCallStarted;
60+
await Target.StopAsync(CancellationToken);
61+
62+
// Assert
63+
capturedDelay.ShouldNotBeNull();
64+
capturedDelay!.Value.ShouldBeGreaterThanOrEqualTo(TimeSpan.FromSeconds(baseDelaySeconds));
65+
}
66+
67+
[Fact]
68+
public async Task AFullySuccessfulFullBatchRefetchesImmediately()
69+
{
70+
// Arrange
71+
const int batchSize = 10;
72+
Use<IOptions<OutboxProcessingOptions>>(Options.Create(new OutboxProcessingOptions { BatchSize = batchSize }));
73+
Use<IEnumerable<IOutboxRelayGate>>([]);
74+
Use<IServiceScopeFactory>(new AutoMockerServiceScopeFactory(AutoMocker));
75+
var store = new CountingOutboxStore(result: batchSize);
76+
Use<IOutboxStore>(store);
77+
Use(CreateInstance<OutboxProcessor>());
78+
79+
// Act
80+
await Target.StartAsync(CancellationToken);
81+
await store.SecondCallStarted;
82+
await Target.StopAsync(CancellationToken);
83+
84+
// Assert
85+
GetMock<IOutboxSignal>().Verify(signal => signal.WaitAsync(It.IsAny<TimeSpan>(), It.IsAny<CancellationToken>()), Times.Never);
86+
}
87+
3488
private sealed class BlockingRelayGate : IOutboxRelayGate
3589
{
3690
private readonly TaskCompletionSource _entered = new(TaskCreationOptions.RunContinuationsAsynchronously);
@@ -43,4 +97,44 @@ public Task WaitUntilReadyAsync(CancellationToken cancellationToken)
4397
return Task.Delay(Timeout.Infinite, cancellationToken);
4498
}
4599
}
100+
101+
private sealed class CountingOutboxStore(int result) : IOutboxStore
102+
{
103+
private readonly TaskCompletionSource _secondCallStarted = new(TaskCreationOptions.RunContinuationsAsynchronously);
104+
private int _callCount;
105+
106+
public Task SecondCallStarted => _secondCallStarted.Task;
107+
108+
public bool IsInTransaction => false;
109+
110+
public void AddOutboxMessage(OutboxMessage message)
111+
{
112+
}
113+
114+
public Task<int> SaveChangesAsync(CancellationToken cancellationToken = default) => Task.FromResult(0);
115+
116+
public Task<int> ProcessBatchAsync(Func<OutboxMessageData, CancellationToken, Task<string?>> dispatch, CancellationToken cancellationToken)
117+
{
118+
if (Interlocked.Increment(ref _callCount) == 2)
119+
{
120+
_secondCallStarted.TrySetResult();
121+
}
122+
123+
return Task.FromResult(result);
124+
}
125+
}
126+
127+
private sealed class AutoMockerServiceScopeFactory(IServiceProvider serviceProvider) : IServiceScopeFactory
128+
{
129+
public IServiceScope CreateScope() => new PassthroughServiceScope(serviceProvider);
130+
131+
private sealed class PassthroughServiceScope(IServiceProvider serviceProvider) : IServiceScope
132+
{
133+
public IServiceProvider ServiceProvider { get; } = serviceProvider;
134+
135+
public void Dispose()
136+
{
137+
}
138+
}
139+
}
46140
}

‎tests/Vulthil.SharedKernel.Outbox.Tests/OutboxProcessorMetricsTests.cs‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
using System.Diagnostics.Metrics;
2+
using Microsoft.Extensions.Logging;
23
using Microsoft.Extensions.Options;
34
using Vulthil.xUnit;
45

@@ -47,6 +48,36 @@ public async Task AFailedDispatchIncrementsTheFailedCounter()
4748
count.ShouldBe(1);
4849
}
4950

51+
[Fact]
52+
public async Task CancellingMidBatchDoesNotIncrementTheFailedCounterOrLogAnError()
53+
{
54+
// Arrange
55+
using var cts = new CancellationTokenSource();
56+
await cts.CancelAsync();
57+
SetupDispatcherToThrowOperationCanceled();
58+
SetupStoreToDispatch();
59+
60+
// Act
61+
var count = await MeasureCounterAsync("vulthil.outbox.failed", () =>
62+
Should.ThrowAsync<OperationCanceledException>(() => Target.ExecuteAsync(cts.Token)));
63+
64+
// Assert
65+
count.ShouldBe(0);
66+
GetMock<ILogger<OutboxProcessor>>().Invocations.ShouldBeEmpty();
67+
}
68+
69+
private void SetupDispatcherToThrowOperationCanceled()
70+
{
71+
var dispatcher = GetMock<IOutboxDispatcher>();
72+
dispatcher.Setup(d => d.Handles(It.IsAny<OutboxDestination>())).Returns(true);
73+
dispatcher.Setup(d => d.DispatchAsync(It.IsAny<OutboxMessageData>(), It.IsAny<CancellationToken>()))
74+
.ThrowsAsync(new OperationCanceledException());
75+
76+
GetMock<IServiceProvider>()
77+
.Setup(sp => sp.GetService(typeof(IEnumerable<IOutboxDispatcher>)))
78+
.Returns(new[] { dispatcher.Object });
79+
}
80+
5081
private void SetupDispatcher(bool succeeds)
5182
{
5283
var dispatcher = GetMock<IOutboxDispatcher>();

0 commit comments

Comments
 (0)