perf(store-and-forward): post audit-observer notifications to an ordered single-reader channel, latency-isolating the sweep

src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs

@@ -1,3 +1,4 @@
+using System.Threading.Channels;
 using Microsoft.Extensions.Logging;
 using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
 using ZB.MOM.WW.ScadaBridge.Commons.Observability;
@@ -112,6 +113,26 @@ public class StoreAndForwardService
     /// </summary>
     private Task? _sweepTask;
 
+    /// <summary>
+    /// Ordered, single-reader queue of cached-call audit-observer notifications.
+    /// The retry sweep <b>posts</b> to this channel instead of awaiting the
+    /// observer inline, so a slow observer (a SQLite audit write) cannot stretch
+    /// the sweep; the single reader (<see cref="_observerPump"/>) preserves
+    /// per-operation event order. Re-created in <see cref="StartAsync"/> because
+    /// <see cref="StopAsync"/> completes it (a restarted instance needs a fresh
+    /// channel). Before <see cref="StartAsync"/> starts the pump, posts fall back
+    /// to inline processing (see <see cref="PostObserverNotification"/>).
+    /// </summary>
+    private Channel<Func<Task>> _observerQueue =
+        Channel.CreateUnbounded<Func<Task>>(new UnboundedChannelOptions { SingleReader = true });
+
+    /// <summary>
+    /// The single-reader pump draining <see cref="_observerQueue"/>, or
+    /// <c>null</c> when the service is not started. Started in
+    /// <see cref="StartAsync"/>, completed + awaited in <see cref="StopAsync"/>.
+    /// </summary>
+    private Task? _observerPump;
+
     /// <summary>
     /// How long <see cref="StopAsync"/> waits for an
     /// in-flight retry sweep to finish before returning. The default — 10 s —
@@ -349,6 +370,28 @@ public class StoreAndForwardService
             ScadaBridgeTelemetry.SetQueueDepthProvider(provider);
         }
 
+        // (Re)create the observer queue + start its single-reader pump. A prior
+        // StopAsync completes the channel, so a restarted instance needs a fresh
+        // one. The pump is best-effort: an observer that throws is logged and
+        // swallowed so a failing audit pipeline never corrupts retry bookkeeping.
+        _observerQueue = Channel.CreateUnbounded<Func<Task>>(
+            new UnboundedChannelOptions { SingleReader = true });
+        _observerPump = Task.Run(async () =>
+        {
+            await foreach (var work in _observerQueue.Reader.ReadAllAsync())
+            {
+                try
+                {
+                    await work();
+                }
+                catch (Exception ex)
+                {
+                    _logger.LogWarning(ex,
+                        "Cached-call audit observer threw; ignored (best-effort per alog.md §7)");
+                }
+            }
+        });
+
         _retryTimer = new Timer(
             _ => Volatile.Write(ref _sweepTask, RetryPendingMessagesAsync()),
             null,
@@ -404,6 +447,25 @@ public class StoreAndForwardService
             }
         }
 
+        // Complete the observer queue and wait (bounded) for the pump to drain any
+        // posted-but-not-yet-processed notifications, so shutdown does not silently
+        // drop audit events that the sweep already handed off.
+        _observerQueue.Writer.TryComplete();
+        var pump = _observerPump;
+        if (pump is not null)
+        {
+            try
+            {
+                await pump.WaitAsync(SweepShutdownWaitTimeout).ConfigureAwait(false);
+            }
+            catch (TimeoutException)
+            {
+                _logger.LogWarning(
+                    "Audit-observer pump did not drain within {Timeout}", SweepShutdownWaitTimeout);
+            }
+            _observerPump = null;
+        }
+
         var provider = _queueDepthProvider;
         if (provider is not null)
         {
@@ -717,7 +779,7 @@ public class StoreAndForwardService
 
                 // Terminal Delivered observer notification — the audit
                 // bridge maps this to Attempted + CachedResolve(Delivered).
-                await NotifyCachedCallObserverAsync(
+                await PostObserverNotification(
                     message,
                     CachedCallAttemptOutcome.Delivered,
                     lastError: null,
@@ -747,7 +809,7 @@ public class StoreAndForwardService
 
             // Terminal PermanentFailure observer notification — the
             // audit bridge maps this to Attempted(Failed) + CachedResolve(Parked).
-            await NotifyCachedCallObserverAsync(
+            await PostObserverNotification(
                 message,
                 CachedCallAttemptOutcome.PermanentFailure,
                 lastError: message.LastError,
@@ -786,7 +848,7 @@ public class StoreAndForwardService
 
                 // Terminal ParkedMaxRetries observer notification — the
                 // audit bridge maps this to Attempted(Failed) + CachedResolve(Parked).
-                await NotifyCachedCallObserverAsync(
+                await PostObserverNotification(
                     message,
                     CachedCallAttemptOutcome.ParkedMaxRetries,
                     lastError: ex.Message,
@@ -810,7 +872,7 @@ public class StoreAndForwardService
 
                 // Per-attempt TransientFailure observer notification —
                 // the audit bridge maps this to Attempted(Failed).
-                await NotifyCachedCallObserverAsync(
+                await PostObserverNotification(
                     message,
                     CachedCallAttemptOutcome.TransientFailure,
                     lastError: ex.Message,
@@ -822,10 +884,48 @@ public class StoreAndForwardService
         }
     }
 
+    /// <summary>
+    /// Hands a cached-call observer notification to the ordered single-reader pump
+    /// (see <see cref="_observerQueue"/>) and returns immediately, so a slow
+    /// observer cannot stretch the sweep. When the service is not started (no pump —
+    /// e.g. a unit test driving <see cref="RetryPendingMessagesAsync"/> directly),
+    /// the notification is processed inline so it is still delivered synchronously.
+    /// The buffered <paramref name="message"/> is not mutated again after this call
+    /// within a sweep (the next tick loads fresh row instances), so the deferred
+    /// read of its fields by the pump is safe.
+    /// </summary>
+    private ValueTask PostObserverNotification(
+        StoreAndForwardMessage message,
+        CachedCallAttemptOutcome outcome,
+        string? lastError,
+        int? httpStatus,
+        DateTime occurredAtUtc,
+        int? durationMs)
+    {
+        if (_cachedCallObserver == null)
+        {
+            return ValueTask.CompletedTask;
+        }
+
+        Func<Task> work = () => NotifyCachedCallObserverAsync(
+            message, outcome, lastError, httpStatus, occurredAtUtc, durationMs);
+
+        if (_observerPump is null)
+        {
+            // Not started: process inline (preserves pre-Task-19 test behavior).
+            return new ValueTask(work());
+        }
+
+        _observerQueue.Writer.TryWrite(work);
+        return ValueTask.CompletedTask;
+    }
+
     /// <summary>
     /// Notify the registered
     /// <see cref="ICachedCallLifecycleObserver"/> of the just-completed
-    /// attempt. Only fires for cached-call categories
+    /// attempt. Invoked from the single-reader observer pump (latency-isolated
+    /// from the sweep; order preserved per posting sequence) — or inline when the
+    /// service is not started. Only fires for cached-call categories
     /// (<see cref="StoreAndForwardCategory.ExternalSystem"/> and
     /// <see cref="StoreAndForwardCategory.CachedDbWrite"/>); the
     /// <see cref="StoreAndForwardCategory.Notification"/> category has its

tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/CachedCallAttemptEmissionTests.cs

@@ -452,4 +452,69 @@ public class CachedCallAttemptEmissionTests : IAsyncLifetime, IDisposable
         var msg = await _storage.GetMessageByIdAsync(trackedId.ToString());
         Assert.Null(msg);
     }
+
+    // ── Task 19: audit-observer decoupled from the sweep via an ordered channel ──
+
+    private sealed class DelegatingObserver : ICachedCallLifecycleObserver
+    {
+        private readonly Func<CachedCallAttemptContext, Task> _onNotify;
+        public DelegatingObserver(Func<CachedCallAttemptContext, Task> onNotify) => _onNotify = onNotify;
+        public Task OnAttemptCompletedAsync(CachedCallAttemptContext context, CancellationToken ct = default)
+            => _onNotify(context);
+    }
+
+    [Fact]
+    public async Task SlowObserver_DoesNotBlockTheSweep_AndEventsArriveInOrder()
+    {
+        var gate = new TaskCompletionSource();
+        var observed = new List<CachedCallAttemptOutcome>();
+        var observer = new DelegatingObserver(async ctx =>
+        {
+            await gate.Task; // first call blocks until released
+            lock (observed) observed.Add(ctx.Outcome);
+        });
+
+        // Fresh service over its own storage so StartAsync's pump/timer is isolated
+        // from the shared _service.
+        var connStr = $"Data Source=SlowObs_{Guid.NewGuid():N};Mode=Memory;Cache=Shared";
+        using var keepAlive = new SqliteConnection(connStr);
+        keepAlive.Open();
+        var storage = new StoreAndForwardStorage(connStr, NullLogger<StoreAndForwardStorage>.Instance);
+        await storage.InitializeAsync();
+        var service = new StoreAndForwardService(
+            storage,
+            new StoreAndForwardOptions
+            {
+                DefaultRetryInterval = TimeSpan.Zero,
+                DefaultMaxRetries = 3,
+                RetryTimerInterval = TimeSpan.FromHours(1), // timer never fires in-test
+            },
+            NullLogger<StoreAndForwardService>.Instance,
+            replication: null,
+            cachedCallObserver: observer,
+            siteId: "site-77");
+
+        await service.StartAsync();
+        try
+        {
+            service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem,
+                _ => Task.FromResult(true));
+            await service.EnqueueAsync(StoreAndForwardCategory.ExternalSystem, "t", "{}",
+                attemptImmediateDelivery: false, retryInterval: TimeSpan.Zero,
+                messageId: TrackedOperationId.New().ToString());
+
+            var sweep = service.RetryPendingMessagesAsync();
+            await sweep.WaitAsync(TimeSpan.FromSeconds(5)); // pre-fix: blocks on gate → times out
+
+            gate.SetResult();
+            var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(5);
+            while (DateTime.UtcNow < deadline)
+            {
+                lock (observed) if (observed.Count == 1) break;
+                await Task.Delay(10);
+            }
+            lock (observed) Assert.Single(observed);
+        }
+        finally { await service.StopAsync(); }
+    }
 }