From 70e7dfd7b95674dd0db695bfc3c0df37f86e1ea1 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 8 Jul 2026 17:35:29 -0400 Subject: [PATCH] fix(store-and-forward): inline ordered replication dispatch; Warning + counter on replication failures --- .../Observability/ScadaBridgeTelemetry.cs | 8 +++ .../Actors/SiteReplicationActor.cs | 8 ++- .../ReplicationService.cs | 34 ++++++++---- .../ReplicationServiceTests.cs | 53 +++++++++++++++++++ 4 files changed, 92 insertions(+), 11 deletions(-) diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs index a1aafc55..a28c16c8 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs @@ -37,6 +37,11 @@ public static class ScadaBridgeTelemetry Meter.CreateCounter("scadabridge.inbound_api.requests", unit: "1", description: "Inbound API requests, tagged by method."); + /// Incremented each time an S&F buffer replication op fails to dispatch/deliver to the peer. + private static readonly Counter _replicationFailures = + Meter.CreateCounter("scadabridge.store_and_forward.replication.failures", unit: "1", + description: "S&F buffer replication operations that failed to dispatch/deliver to the peer node"); + // ---------------- Observable gauges ---------------- /// Current count of open site connections, mutated via . @@ -70,6 +75,9 @@ public static class ScadaBridgeTelemetry public static void RecordInboundApiRequest(string method) => _inboundApiRequests.Add(1, new KeyValuePair("method", method)); + /// Records one failed S&F replication dispatch. + public static void RecordReplicationFailure() => _replicationFailures.Add(1); + /// Records that a site connection opened (increments the up-count gauge). public static void SiteConnectionOpened() => Interlocked.Increment(ref _siteConnectionsUp); diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs index ed7d321f..bc321e0a 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs @@ -2,6 +2,7 @@ using Akka.Actor; using Akka.Cluster; using Akka.Event; using Microsoft.Extensions.Logging; +using ZB.MOM.WW.ScadaBridge.Commons.Observability; using ZB.MOM.WW.ScadaBridge.SiteRuntime.Deployment; using ZB.MOM.WW.ScadaBridge.SiteRuntime.Messages; using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence; @@ -140,7 +141,12 @@ public class SiteReplicationActor : ReceiveActor { if (_peerAddress == null) { - _logger.LogDebug("No peer available, dropping replication message {Type}", message.GetType().Name); + // A dropped op is a lost delta — surface it at Warning + a metric so a + // never-tracked peer (a dead standby) is visible, not silent (arch + // review 02). In single-node dev the peer is legitimately absent; the + // per-op warning is rate-tolerable (accepted per review). + ScadaBridgeTelemetry.RecordReplicationFailure(); + _logger.LogWarning("No peer available, dropping replication message {Type}", message.GetType().Name); return; } diff --git a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/ReplicationService.cs b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/ReplicationService.cs index 536ea2f9..abdb2393 100644 --- a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/ReplicationService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/ReplicationService.cs @@ -1,4 +1,5 @@ using Microsoft.Extensions.Logging; +using ZB.MOM.WW.ScadaBridge.Commons.Observability; using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; namespace ZB.MOM.WW.ScadaBridge.StoreAndForward; @@ -133,19 +134,32 @@ public class ReplicationService private void FireAndForget(ReplicationOperation operation) { - Task.Run(async () => + // Invoked inline, NOT via Task.Run: the handler is a non-blocking Akka + // Tell, and thread-pool hand-off destroyed Add/Remove ordering for the + // same message id (arch review 02). Inline invocation preserves issue + // order; Akka's per-sender/receiver guarantee preserves it on the wire. + try { - try + var task = _replicationHandler!.Invoke(operation); + if (!task.IsCompletedSuccessfully) { - await _replicationHandler!.Invoke(operation); + task.ContinueWith(t => + { + ScadaBridgeTelemetry.RecordReplicationFailure(); + _logger.LogWarning(t.Exception, + "Replication of {OpType} for message {MessageId} failed (best-effort); standby buffer may be diverging", + operation.OperationType, operation.MessageId); + }, + TaskContinuationOptions.OnlyOnFaulted); } - catch (Exception ex) - { - _logger.LogDebug(ex, - "Replication of {OpType} for message {MessageId} failed (best-effort)", - operation.OperationType, operation.MessageId); - } - }); + } + catch (Exception ex) + { + ScadaBridgeTelemetry.RecordReplicationFailure(); + _logger.LogWarning(ex, + "Replication of {OpType} for message {MessageId} failed (best-effort); standby buffer may be diverging", + operation.OperationType, operation.MessageId); + } } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/ReplicationServiceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/ReplicationServiceTests.cs index 6b8ce4ba..7df0d8e8 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/ReplicationServiceTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/ReplicationServiceTests.cs @@ -1,4 +1,5 @@ using Microsoft.Data.Sqlite; +using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; @@ -163,6 +164,47 @@ public class ReplicationServiceTests : IAsyncLifetime, IDisposable // No exception -- fire-and-forget, best-effort } + // ── Task 10 (arch review 02): ordered, observable replication dispatch ── + + [Fact] + public void ReplicationOperations_AreDispatchedInIssueOrder() + { + var seen = new List<(ReplicationOperationType, string)>(); + _replicationService.SetReplicationHandler(op => + { + seen.Add((op.OperationType, op.MessageId)); + return Task.CompletedTask; + }); + + for (var i = 0; i < 200; i++) + { + _replicationService.ReplicateEnqueue(CreateMessage($"m{i}")); + _replicationService.ReplicateRemove($"m{i}"); + } + + // Inline dispatch: by the time the calls return, every op was handed to the + // handler, Add strictly before Remove per id. Pre-fix (Task.Run) this was + // racy in both count and order. + Assert.Equal(400, seen.Count); + for (var i = 0; i < 200; i++) + { + Assert.Equal((ReplicationOperationType.Add, $"m{i}"), seen[2 * i]); + Assert.Equal((ReplicationOperationType.Remove, $"m{i}"), seen[2 * i + 1]); + } + } + + [Fact] + public void ReplicationHandlerThrow_IsSwallowed_AndLoggedAtWarning() + { + var logger = new CapturingLogger(); + var service = new ReplicationService(new StoreAndForwardOptions(), logger); + service.SetReplicationHandler(_ => throw new InvalidOperationException("peer gone")); + + service.ReplicateRemove("m1"); // must not throw + + Assert.Contains(logger.Entries, e => e.Level == LogLevel.Warning); + } + private static StoreAndForwardMessage CreateMessage(string id) { return new StoreAndForwardMessage @@ -178,4 +220,15 @@ public class ReplicationServiceTests : IAsyncLifetime, IDisposable Status = StoreAndForwardMessageStatus.Pending }; } + + /// Minimal in-memory logger that records level + rendered message. + private sealed class CapturingLogger : ILogger + { + public List<(LogLevel Level, string Message)> Entries { get; } = new(); + public IDisposable? BeginScope(TState state) where TState : notnull => null; + public bool IsEnabled(LogLevel logLevel) => true; + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, + Func formatter) + => Entries.Add((logLevel, formatter(state, exception))); + } }