fix(store-and-forward): inline ordered replication dispatch; Warning + counter on replication failures

src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs

@@ -37,6 +37,11 @@ public static class ScadaBridgeTelemetry
         Meter.CreateCounter<long>("scadabridge.inbound_api.requests", unit: "1",
             description: "Inbound API requests, tagged by method.");
 
+    /// <summary>Incremented each time an S&amp;F buffer replication op fails to dispatch/deliver to the peer.</summary>
+    private static readonly Counter<long> _replicationFailures =
+        Meter.CreateCounter<long>("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 ----------------
 
     /// <summary>Current count of open site connections, mutated via <see cref="Interlocked"/>.</summary>
@@ -70,6 +75,9 @@ public static class ScadaBridgeTelemetry
     public static void RecordInboundApiRequest(string method) =>
         _inboundApiRequests.Add(1, new KeyValuePair<string, object?>("method", method));
 
+    /// <summary>Records one failed S&amp;F replication dispatch.</summary>
+    public static void RecordReplicationFailure() => _replicationFailures.Add(1);
+
     /// <summary>Records that a site connection opened (increments the up-count gauge).</summary>
     public static void SiteConnectionOpened() => Interlocked.Increment(ref _siteConnectionsUp);
 

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;
         }
 

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);
+        }
     }
 }
 

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<ReplicationService>();
+        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
         };
     }
+
+    /// <summary>Minimal in-memory logger that records level + rendered message.</summary>
+    private sealed class CapturingLogger<T> : ILogger<T>
+    {
+        public List<(LogLevel Level, string Message)> Entries { get; } = new();
+        public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
+        public bool IsEnabled(LogLevel logLevel) => true;
+        public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception,
+            Func<TState, Exception?, string> formatter)
+            => Entries.Add((logLevel, formatter(state, exception)));
+    }
 }