feat(store-and-forward): deferToSweep enqueue mode with immediate sweep kick for latency-sensitive callers

This commit is contained in:
Joseph Doherty
2026-07-08 20:46:08 -04:00
parent 283ce0647c
commit 05a71a0260
2 changed files with 109 additions and 1 deletions
@@ -13,6 +13,7 @@ public class StoreAndForwardServiceTests : IAsyncLifetime, IDisposable
private readonly StoreAndForwardStorage _storage;
private readonly StoreAndForwardService _service;
private readonly StoreAndForwardOptions _options;
private readonly List<SqliteConnection> _extraKeepAlives = new();
public StoreAndForwardServiceTests()
{
@@ -38,7 +39,35 @@ public class StoreAndForwardServiceTests : IAsyncLifetime, IDisposable
public Task DisposeAsync() => Task.CompletedTask;
public void Dispose() => _keepAlive.Dispose();
public void Dispose()
{
_keepAlive.Dispose();
foreach (var c in _extraKeepAlives) c.Dispose();
}
/// <summary>
/// Builds a fresh service over its own in-memory (shared-cache) SQLite so a test
/// can call StartAsync without racing the shared <c>_service</c>'s timer. The
/// keep-alive connection is tracked for disposal.
/// </summary>
private StoreAndForwardService CreateService(TimeSpan? retryTimerInterval = null)
{
var connStr = $"Data Source=DeferTests_{Guid.NewGuid():N};Mode=Memory;Cache=Shared";
var keepAlive = new SqliteConnection(connStr);
keepAlive.Open();
_extraKeepAlives.Add(keepAlive);
var storage = new StoreAndForwardStorage(connStr, NullLogger<StoreAndForwardStorage>.Instance);
storage.InitializeAsync().GetAwaiter().GetResult();
var options = new StoreAndForwardOptions
{
DefaultRetryInterval = TimeSpan.Zero,
DefaultMaxRetries = 3,
RetryTimerInterval = retryTimerInterval ?? TimeSpan.FromMinutes(10),
};
return new StoreAndForwardService(storage, options, NullLogger<StoreAndForwardService>.Instance);
}
// ── WP-10: Immediate delivery ──
@@ -746,4 +775,50 @@ public class StoreAndForwardServiceTests : IAsyncLifetime, IDisposable
do { current = Volatile.Read(ref target); if (value <= current) return; }
while (Interlocked.CompareExchange(ref target, value, current) != current);
}
// ── Task 12: deferToSweep enqueue mode ──
/// <summary>
/// <c>deferToSweep: true</c> must NOT invoke the delivery handler inline (the
/// caller is on a latency-sensitive thread), and the buffered row must be due on
/// the very next sweep — <c>LastAttemptAt</c> stays null so it is not held back by
/// RetryInterval.
/// </summary>
[Fact]
public async Task Enqueue_DeferToSweep_DoesNotInvokeHandlerInline_AndRowIsDueImmediately()
{
var service = CreateService();
var inlineInvoked = false;
service.RegisterDeliveryHandler(StoreAndForwardCategory.Notification,
_ => { inlineInvoked = true; return Task.FromResult(true); });
var result = await service.EnqueueAsync(StoreAndForwardCategory.Notification, "ops", "{}",
deferToSweep: true, messageId: "n1");
Assert.False(inlineInvoked);
Assert.True(result.WasBuffered);
var row = await service.GetMessageByIdAsync("n1");
Assert.Null(row!.LastAttemptAt); // due on the very next sweep, not after RetryInterval
}
/// <summary>
/// After StartAsync, <c>deferToSweep</c> must kick an immediate background sweep so
/// healthy-path latency is milliseconds, not one timer interval — proven with a
/// timer set to an hour so only the kicked sweep can deliver.
/// </summary>
[Fact]
public async Task Enqueue_DeferToSweep_AfterStart_KicksAnImmediateSweep()
{
var service = CreateService(retryTimerInterval: TimeSpan.FromHours(1)); // timer will never fire in-test
var delivered = new TaskCompletionSource();
service.RegisterDeliveryHandler(StoreAndForwardCategory.Notification,
_ => { delivered.TrySetResult(); return Task.FromResult(true); });
await service.StartAsync();
try
{
await service.EnqueueAsync(StoreAndForwardCategory.Notification, "ops", "{}", deferToSweep: true);
await delivered.Task.WaitAsync(TimeSpan.FromSeconds(5)); // pre-fix: times out
}
finally { await service.StopAsync(); }
}
}