From 05a71a0260c532d30058673260a612e12f70d210 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 8 Jul 2026 20:46:08 -0400 Subject: [PATCH] feat(store-and-forward): deferToSweep enqueue mode with immediate sweep kick for latency-sensitive callers --- .../StoreAndForwardService.cs | 33 ++++++++ .../StoreAndForwardServiceTests.cs | 77 ++++++++++++++++++- 2 files changed, 109 insertions(+), 1 deletion(-) diff --git a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs index 2f1ef7b6..98136439 100644 --- a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs @@ -451,6 +451,14 @@ public class StoreAndForwardService /// When false, the caller has already made its own delivery attempt and the /// message is buffered directly for the retry sweep (the handler is not invoked here). /// + /// + /// When true, the handler is never invoked inline — the message is buffered + /// due-immediately ( stays null) + /// and a background sweep is kicked. Use for callers on latency-sensitive threads + /// (e.g. script dispatchers) whose delivery involves a long Ask timeout. Unlike + /// : false, no delivery attempt is + /// presumed to have already been made, so the row is due on the very next sweep. + /// /// /// An explicit, caller-supplied message id. null (the default) makes the /// service mint a fresh GUID. The Notification Outbox enqueue path supplies its own @@ -487,6 +495,7 @@ public class StoreAndForwardService int? maxRetries = null, TimeSpan? retryInterval = null, bool attemptImmediateDelivery = true, + bool deferToSweep = false, string? messageId = null, Guid? executionId = null, string? sourceScript = null, @@ -509,6 +518,18 @@ public class StoreAndForwardService ParentExecutionId = parentExecutionId }; + // Latency-sensitive caller: never run the handler inline. Buffer + // due-immediately (LastAttemptAt intentionally left null so the row is due + // on the very next sweep, not held back by RetryInterval) and kick a + // background sweep so the healthy-path latency is milliseconds. + if (deferToSweep) + { + await BufferAsync(message); + RaiseActivity("Queued", category, $"Deferred to sweep: {target}"); + TriggerSweep(); + return new StoreAndForwardResult(true, message.Id, true); + } + // Attempt immediate delivery — unless the caller has already made a // delivery attempt of its own (attemptImmediateDelivery: false). In that // case re-invoking the handler here would dispatch the request twice. @@ -571,6 +592,18 @@ public class StoreAndForwardService Interlocked.Increment(ref _bufferedCount); } + /// + /// Kicks a background retry sweep now (fire-and-forget). No-op before + /// (storage may be uninitialized and the timer, whose + /// presence gates this, is not yet set). Overlap-safe: the sweep's + /// CAS makes a concurrent kick a cheap no-op. + /// + public void TriggerSweep() + { + if (_retryTimer == null) return; + Volatile.Write(ref _sweepTask, RetryPendingMessagesAsync()); + } + /// /// Background retry sweep. Processes all pending messages that are due for retry. /// diff --git a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs index 8fd5fda9..4facffa7 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs @@ -13,6 +13,7 @@ public class StoreAndForwardServiceTests : IAsyncLifetime, IDisposable private readonly StoreAndForwardStorage _storage; private readonly StoreAndForwardService _service; private readonly StoreAndForwardOptions _options; + private readonly List _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(); + } + + /// + /// Builds a fresh service over its own in-memory (shared-cache) SQLite so a test + /// can call StartAsync without racing the shared _service's timer. The + /// keep-alive connection is tracked for disposal. + /// + 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.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.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 ── + + /// + /// deferToSweep: true 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 — LastAttemptAt stays null so it is not held back by + /// RetryInterval. + /// + [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 + } + + /// + /// After StartAsync, deferToSweep 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. + /// + [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(); } + } }