diff --git a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs index 98136439..ce3eb904 100644 --- a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs +++ b/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 /// private Task? _sweepTask; + /// + /// Ordered, single-reader queue of cached-call audit-observer notifications. + /// The retry sweep posts to this channel instead of awaiting the + /// observer inline, so a slow observer (a SQLite audit write) cannot stretch + /// the sweep; the single reader () preserves + /// per-operation event order. Re-created in because + /// completes it (a restarted instance needs a fresh + /// channel). Before starts the pump, posts fall back + /// to inline processing (see ). + /// + private Channel> _observerQueue = + Channel.CreateUnbounded>(new UnboundedChannelOptions { SingleReader = true }); + + /// + /// The single-reader pump draining , or + /// null when the service is not started. Started in + /// , completed + awaited in . + /// + private Task? _observerPump; + /// /// How long 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>( + 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 } } + /// + /// Hands a cached-call observer notification to the ordered single-reader pump + /// (see ) 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 directly), + /// the notification is processed inline so it is still delivered synchronously. + /// The buffered 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. + /// + private ValueTask PostObserverNotification( + StoreAndForwardMessage message, + CachedCallAttemptOutcome outcome, + string? lastError, + int? httpStatus, + DateTime occurredAtUtc, + int? durationMs) + { + if (_cachedCallObserver == null) + { + return ValueTask.CompletedTask; + } + + Func 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; + } + /// /// Notify the registered /// 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 /// ( and /// ); the /// category has its diff --git a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/CachedCallAttemptEmissionTests.cs b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/CachedCallAttemptEmissionTests.cs index aa67d856..d04e8bb9 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/CachedCallAttemptEmissionTests.cs +++ b/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 _onNotify; + public DelegatingObserver(Func 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(); + 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.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.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(); } + } }