fix(ops): wonder site config gains required audit DB path; explicit LocalDb read-page cap; rate-limited observer drop logging

F1: deploy/wonder-app-vd03/appsettings.Site.json (outside git, WP1.2's
StartupValidator gate applies live on next install/upgrade) was missing the
now-required AuditLog:SiteWriter:DatabasePath, added pointing at
E:\ApiInstall\ScadaBridge\site\data\auditlog.db alongside the file's
existing SiteEventLog/LocalDb paths; scanned deploy/ for other Site-role
appsettings with the same gap (none) and confirmed wonder does not pin
LocalDb:Replication:MaxBatchSize (F2 doesn't apply there).

F2: re-pin an explicit LocalDb:Replication:MaxBatchSize=64 on docker/site-a
node-a and node-b. MaxBatchBytes (2 MB default) only bounds the wire
message via the per-message split in SyncSession.PumpLoopAsync;
MaxBatchSize separately bounds the DB read page in
OplogStore.ReadBatchAboveAsync/SnapshotStreamer, which materializes the
whole page into memory before that split runs. Left at the 500 default, a
reconnect drain of worst-case config_json rows could transiently allocate
~35 MB per read even though every wire message stayed within budget.
Updated the CLAUDE.md LocalDb bullet to stop implying the row cap is fully
redundant with the byte budget (topology-guide.md has no matching claim).

F3: StoreAndForwardService's observer-queue onDropped callback logged a
Warning per dropped item, flooding logs at sweep rate for a stuck observer
with a large queue. LogObserverQueueDrop now logs once immediately on the
first drop of an episode, then throttles to at most one rollup Warning per
minute while drops continue, reporting the count dropped since the last
log; the cumulative ObserverQueueDroppedCount counter is unaffected.
Extended StoreAndForwardServiceTests with
ObserverQueue_ManyDropsInOneEpisode_LogsExactlyOneWarning, which floods the
bounded queue and pins exactly one drop-related Warning log for the
episode via a small CapturingLogger test double.

dotnet build ZB.MOM.WW.ScadaBridge.slnx: 0 warnings, 0 errors.
dotnet test StoreAndForward.Tests: 134/134 passed.
dotnet test Host.Tests: 490/490 passed.

CLAUDE.md

@@ -89,7 +89,7 @@ spec for each is `docs/requirements/Component-<Name>.md`, and `README.md` carrie
     - CDC replication does all three jobs now: config deploys reach the standby as ordinary row changes — **the standby makes no fetch at all** during a deploy (`SiteReconciliationActor`'s node-STARTUP fetch when central reports gaps is a different, surviving path) — and buffer mutations replicate via triggers on `sf_messages`. `ReplaceAllAsync` was a destructive delete-all-then-insert-all resync and is **unsafe to reintroduce**: a mass DELETE on a replicated table would be captured and shipped to the peer. LocalDb's snapshot resync merges per row under LWW and never deletes, which is also why the old N1 directional-authority guard is gone — there is no wipe left to gate.
     - **`notification_lists` and `smtp_configurations` are created but deliberately NOT registered.** They are permanently empty on a site (no writer since 2026-07-10, the migrator skips them, the active-node purge keeps them empty), and registering them would open a standing replication channel whose only historical payload was plaintext SMTP passwords. Pinned by a security-named test, and verified live: those two tables have **no CDC triggers** on either rig node.
     - **Operational constraints (read before upgrading a site pair):** stop and start both nodes TOGETHER — rolling one at a time is no longer supported, since the legacy `SfBufferSnapshot` compatibility handler went with the replicator. And a node offline longer than `LocalDb:Replication:TombstoneRetention` (default 7 days) can resurrect deleted rows on rejoin. See `docs/deployment/topology-guide.md`.
-    - **Batching is by BYTE BUDGET as of LocalDb 0.2.0** — `LocalDb:Replication:MaxBatchBytes` (default **2 MB**, sized under the 4 MB gRPC cap) bounds a delta/snapshot message by summed serialized size, with `MaxBatchSize` demoted to a secondary row cap; a single row over budget is sent alone rather than stalling the stream. **The rig's `MaxBatchSize = 16` pin is retired** (it was a hand-computed proxy: ~70 KB worst-case `config_json` x the 500 default is ~35 MB); both keys are now left unset on `docker/`.
+    - **Batching is by BYTE BUDGET as of LocalDb 0.2.0** — `LocalDb:Replication:MaxBatchBytes` (default **2 MB**, sized under the 4 MB gRPC cap) bounds a delta/snapshot message by summed serialized size via a per-message split in `SyncSession.PumpLoopAsync`, with `MaxBatchSize` demoted to a secondary row cap; a single row over budget is sent alone rather than stalling the stream. **The rig's old `MaxBatchSize = 16` pin (a hand-computed byte-budget proxy: ~70 KB worst-case `config_json` x the 500 default is ~35 MB) is retired, but `MaxBatchSize` is NOT fully redundant with `MaxBatchBytes`** — it also bounds the separate DB READ page in `OplogStore.ReadBatchAboveAsync`/`SnapshotStreamer`, which materializes the whole page into memory *before* the byte-budget split runs, so an unset (500-default) `MaxBatchSize` still lets a reconnect drain transiently allocate ~35 MB per read even though every wire message stays under `MaxBatchBytes` (arch-review adversarial finding F2). Both site-a nodes on `docker/` therefore pin an explicit `"MaxBatchSize": 64` to bound that transient allocation, while `MaxBatchBytes` stays unset (its 2 MB default) to bound the wire message; site-b/site-c stay unreplicated so the key doesn't apply there.
     - **CDC registration is conditional, and both directions self-heal at boot** (arch-review WP1.3 + WP3.3, `Host/SiteLocalDbSetup.cs`). Capture triggers are installed only when this node has replication configured — `PeerAddress` **or** `ApiKey`, an OR because only the dialling half sets `PeerAddress` while the passive half carries the key alone. An unreplicated node calls **`DeregisterReplicated`** on all ten tables at boot, dropping triggers an earlier build left behind and pruning their oplog/row-version rows (idempotent; logs once at Information when something was actually cleaned). A replicating node registers with **`baselineExistingRows: true`**, which seeds `__localdb_row_version` for pre-existing rows at the LWW floor (HLC `0`, this node's id) and flags a snapshot resync — so turning replication ON for a site that has been running without it now converges on the rows already in the file instead of only on writes made after the restart. Deregistration must be symmetric (the handshake compares registered-table digests fail-closed), so replication is a both-nodes-together change in either direction. LocalDb 0.2.0 also makes backlog depth O(1) and drops the unused `__localdb_oplog_hlc` index; the on-disk bookkeeping schema goes to v2, upgraded in place on open, with no wire change (0.1.x peers still sync).
 - All timestamps are UTC throughout the system.
 - Inter-cluster communication uses **three** transports, not two — **all cross-cluster command/control and data now rides gRPC** after the ClusterClient→gRPC migration's Phase 4 (`docs/plans/2026-07-22-clusterclient-to-grpc-plan.md`) deleted Akka `ClusterClient`/`ClusterClientReceptionist`: (1) **gRPC command/control** — site→central over the central-hosted `CentralControlService` (`GrpcCentralTransport`, sticky central-a→central-b channel pair; deployments/notifications/health/heartbeat/audit-ingest/reconcile), and central→site over the site-hosted `SiteCommandService` (`GrpcSiteTransport`, per-site NodeA→NodeB channel pair; the 28 lifecycle/OPC-UA/query/parked/route/failover commands); (2) **gRPC** server-streaming for real-time data (attribute values, alarm states, `SiteStreamService`); and (3) **plain token-gated HTTP** for the deployment config itself — notify-and-fetch, the site pulls the config from `DeploymentConfigEndpoints` (`ManagementService/DeploymentConfigEndpoints.cs`) with an `X-Deployment-Token` header, `AllowAnonymous` with the per-deployment token as the entire security boundary. The gRPC boundary is per-site PSK-authenticated (`ControlPlaneAuthInterceptor`, unchanged). There is **no receptionist registration** — discovery is by dialling configured endpoints; central builds one `SitePairChannelProvider` per site (addresses from `Site.GrpcNodeAAddress`/`GrpcNodeBAddress`, refreshed from the DB every 60s and on admin changes), sites dial `ScadaBridge:Communication:CentralGrpcEndpoints` (both central nodes, h2c on `CentralGrpcPort` 8083, **NOT** via Traefik). **Discovery is asymmetric by design:** central discovers site gRPC addresses from the *database* (refreshable at runtime), sites discover central from *appsettings* (`CentralGrpcEndpoints`, static — restart required; `StartupValidator` requires a Site node to list at least one). `Akka.Cluster.Tools` stays for ClusterSingleton; only the ClusterClient part is gone. **Central never buffers for an unreachable site** — the send fails with the caller's Ask/deadline timing out; a `ConnectionStateChanged` mechanism built for this was deleted as dead code.

docker/site-a-node-a/appsettings.Site.json

@@ -104,14 +104,18 @@
       "ApiKey": "dev-site-a-localdb-sync-key",
       // ---- Phase 2 sizing, from the Task 1 rig soak (not from the defaults) ----
       //
-      // The MaxBatchSize = 16 pin is RETIRED as of LocalDb 0.2.0 (arch-review WP3.3).
-      // It existed only as a hand-computed proxy for a byte budget: batching was
-      // row-count-only, and 70 KB of production config_json x the 500 default is
-      // ~35 MB against gRPC's 4 MB receive limit. The library now bounds a batch by
-      // MaxBatchBytes (default 2 MB of summed serialized size), with the row count
-      // demoted to a secondary cap, so both are left at their defaults here - a
-      // deliberately unset MaxBatchBytes is the 2 MB default, and the widest row no
-      // longer has to be guessed at deploy time.
+      // The old MaxBatchSize = 16 pin (a hand-computed byte-budget proxy) was retired
+      // when LocalDb 0.2.0 added MaxBatchBytes (arch-review WP3.3) - but MaxBatchSize
+      // is NOT purely redundant with it. MaxBatchBytes bounds the WIRE message via a
+      // per-message split in SyncSession.PumpLoopAsync; MaxBatchSize separately bounds
+      // the DB READ page in OplogStore.ReadBatchAboveAsync / SnapshotStreamer, which
+      // materializes the whole page into memory BEFORE the byte-budget split runs. At
+      // the unset 500 default, a reconnect drain of 70 KB worst-case config_json rows
+      // can transiently allocate ~35 MB per read even though every wire message stays
+      // under the 2 MB MaxBatchBytes default (arch-review adversarial finding F2). Pin
+      // it explicitly here to bound that transient allocation; MaxBatchBytes is left
+      // unset (its 2 MB default) to bound the wire message.
+      "MaxBatchSize": 64,
       //
       // Backlog caps bound the oplog while the peer is offline. Exceeding them is
       // NOT data loss: the oplog is pruned to the ceiling and needs_snapshot is set,

docker/site-a-node-b/appsettings.Site.json

@@ -97,14 +97,18 @@
       "ApiKey": "dev-site-a-localdb-sync-key",
       // ---- Phase 2 sizing, from the Task 1 rig soak (not from the defaults) ----
       //
-      // The MaxBatchSize = 16 pin is RETIRED as of LocalDb 0.2.0 (arch-review WP3.3).
-      // It existed only as a hand-computed proxy for a byte budget: batching was
-      // row-count-only, and 70 KB of production config_json x the 500 default is
-      // ~35 MB against gRPC's 4 MB receive limit. The library now bounds a batch by
-      // MaxBatchBytes (default 2 MB of summed serialized size), with the row count
-      // demoted to a secondary cap, so both are left at their defaults here - a
-      // deliberately unset MaxBatchBytes is the 2 MB default, and the widest row no
-      // longer has to be guessed at deploy time.
+      // The old MaxBatchSize = 16 pin (a hand-computed byte-budget proxy) was retired
+      // when LocalDb 0.2.0 added MaxBatchBytes (arch-review WP3.3) - but MaxBatchSize
+      // is NOT purely redundant with it. MaxBatchBytes bounds the WIRE message via a
+      // per-message split in SyncSession.PumpLoopAsync; MaxBatchSize separately bounds
+      // the DB READ page in OplogStore.ReadBatchAboveAsync / SnapshotStreamer, which
+      // materializes the whole page into memory BEFORE the byte-budget split runs. At
+      // the unset 500 default, a reconnect drain of 70 KB worst-case config_json rows
+      // can transiently allocate ~35 MB per read even though every wire message stays
+      // under the 2 MB MaxBatchBytes default (arch-review adversarial finding F2). Pin
+      // it explicitly here to bound that transient allocation; MaxBatchBytes is left
+      // unset (its 2 MB default) to bound the wire message.
+      "MaxBatchSize": 64,
       //
       // Backlog caps bound the oplog while the peer is offline. Exceeding them is
       // NOT data loss: the oplog is pruned to the ceiling and needs_snapshot is set,

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

@@ -148,13 +148,85 @@ public class StoreAndForwardService
     /// Cumulative count of cached-call audit-observer notifications dropped because
     /// <see cref="_observerQueue"/> was at capacity (WP2.6c). Not reset across
     /// <see cref="StartAsync"/>/<see cref="StopAsync"/> cycles — a diagnostic total for
-    /// the lifetime of this service instance.
+    /// the lifetime of this service instance. Incremented unconditionally on every drop
+    /// by <see cref="LogObserverQueueDrop"/>, independent of that method's log throttling.
     /// </summary>
     private long _observerQueueDroppedCount;
 
     /// <summary>Diagnostic counter — see <see cref="_observerQueueDroppedCount"/>.</summary>
     public long ObserverQueueDroppedCount => Interlocked.Read(ref _observerQueueDroppedCount);
 
+    /// <summary>
+    /// How often a sustained run of observer-queue drops re-logs after the first Warning
+    /// of an episode (arch-review adversarial finding F3). Without this, a stuck observer
+    /// with a large <see cref="StoreAndForwardOptions.ObserverQueueCapacity"/> floods the
+    /// log at sweep rate — one Warning per dropped item. Only the LOGGING is throttled;
+    /// <see cref="_observerQueueDroppedCount"/> still counts every drop.
+    /// </summary>
+    private static readonly TimeSpan ObserverQueueDropLogRollupInterval = TimeSpan.FromMinutes(1);
+
+    /// <summary>
+    /// <see cref="Environment.TickCount64"/> of the last observer-queue-drop Warning, or
+    /// -1 (its initial value — <c>TickCount64</c> is never negative) if none has been
+    /// logged yet this service-instance lifetime. Not reset across
+    /// <see cref="StartAsync"/>/<see cref="StopAsync"/> cycles, matching
+    /// <see cref="_observerQueueDroppedCount"/>.
+    /// </summary>
+    private long _observerQueueLastDropLogTicks = -1;
+
+    /// <summary>
+    /// Drops accumulated since <see cref="_observerQueueLastDropLogTicks"/> was last
+    /// logged — the count a rollup Warning reports before resetting to 0.
+    /// </summary>
+    private long _observerQueueDroppedSinceLastLog;
+
+    /// <summary>
+    /// Records one observer-queue drop and logs about it: a Warning for the FIRST drop of
+    /// an episode, then at most one rollup Warning per
+    /// <see cref="ObserverQueueDropLogRollupInterval"/> while drops keep happening — never
+    /// one Warning per dropped item (arch-review adversarial finding F3). Safe to call
+    /// concurrently: the log slot for an episode is claimed via a CAS on
+    /// <see cref="_observerQueueLastDropLogTicks"/>, so overlapping droppers accumulate
+    /// into <see cref="_observerQueueDroppedSinceLastLog"/> without double-logging.
+    /// </summary>
+    private void LogObserverQueueDrop()
+    {
+        var total = Interlocked.Increment(ref _observerQueueDroppedCount);
+        Interlocked.Increment(ref _observerQueueDroppedSinceLastLog);
+
+        var now = Environment.TickCount64;
+        var lastLog = Interlocked.Read(ref _observerQueueLastDropLogTicks);
+        var isFirstEverDrop = lastLog < 0;
+        var dueForRollup = !isFirstEverDrop
+            && now - lastLog >= (long)ObserverQueueDropLogRollupInterval.TotalMilliseconds;
+
+        if (!isFirstEverDrop && !dueForRollup)
+            return;
+
+        // Claims the log slot for this episode; a concurrent caller that loses the CAS
+        // simply leaves its increment above in the rollup's next count instead of logging.
+        if (Interlocked.CompareExchange(ref _observerQueueLastDropLogTicks, now, lastLog) != lastLog)
+            return;
+
+        var countSinceLog = Interlocked.Exchange(ref _observerQueueDroppedSinceLastLog, 0);
+        if (isFirstEverDrop)
+        {
+            _logger.LogWarning(
+                "Cached-call audit-observer queue exceeded its bounded capacity ({Capacity}); " +
+                "oldest pending notification dropped (total dropped: {Dropped})",
+                _options.ObserverQueueCapacity, total);
+        }
+        else
+        {
+            _logger.LogWarning(
+                "Cached-call audit-observer queue still exceeding its bounded capacity " +
+                "({Capacity}); {DroppedSinceLastLog} oldest pending notifications dropped in " +
+                "the last {IntervalMinutes} minute(s) (total dropped: {Dropped})",
+                _options.ObserverQueueCapacity, countSinceLog,
+                ObserverQueueDropLogRollupInterval.TotalMinutes, total);
+        }
+    }
+
     /// <summary>
     /// Builds a bounded, single-reader observer queue with DropOldest overflow, invoking
     /// <paramref name="onDropped"/> (if supplied) on every eviction.
@@ -417,14 +489,12 @@ public class StoreAndForwardService
         // WP2.6c: bounded + DropOldest, sized from options; a drop increments
         // _observerQueueDroppedCount (surfaced via ObserverQueueDroppedCount) and is
         // logged at Warning so a stuck observer is visible, not just silently lossy.
-        _observerQueue = CreateObserverQueue(_options.ObserverQueueCapacity, onDropped: () =>
-        {
-            Interlocked.Increment(ref _observerQueueDroppedCount);
-            _logger.LogWarning(
-                "Cached-call audit-observer queue exceeded its bounded capacity ({Capacity}); " +
-                "oldest pending notification dropped (total dropped: {Dropped})",
-                _options.ObserverQueueCapacity, Interlocked.Read(ref _observerQueueDroppedCount));
-        });
+        // F3: logging itself is rate-limited by LogObserverQueueDrop (first-drop Warning
+        // + a rollup at most once per ObserverQueueDropLogRollupInterval) — otherwise a
+        // stuck observer with a large queue floods the log at sweep rate, one Warning per
+        // dropped item. The counter is unaffected by that throttling.
+        _observerQueue = CreateObserverQueue(
+            _options.ObserverQueueCapacity, onDropped: LogObserverQueueDrop);
         _observerPump = Task.Run(async () =>
         {
             await foreach (var work in _observerQueue.Reader.ReadAllAsync())

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

@@ -1,3 +1,4 @@
+using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Logging.Abstractions;
 using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
 using ZB.MOM.WW.ScadaBridge.Commons.Types;
@@ -965,4 +966,109 @@ public class StoreAndForwardServiceTests : IAsyncLifetime, IDisposable
             TestLocalDb.DeleteFiles(path);
         }
     }
+
+    /// <summary>
+    /// Captures every message logged through it, keyed by <see cref="LogLevel"/>. Minimal
+    /// test double — no scopes, no filtering — just enough to assert on log VOLUME.
+    /// </summary>
+    private sealed class CapturingLogger<T> : ILogger<T>
+    {
+        private readonly List<(LogLevel Level, string Message)> _entries = new();
+        private readonly object _gate = new();
+
+        public IReadOnlyList<(LogLevel Level, string Message)> Entries
+        {
+            get { lock (_gate) return _entries.ToList(); }
+        }
+
+        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)
+        {
+            var message = formatter(state, exception);
+            lock (_gate) _entries.Add((logLevel, message));
+        }
+    }
+
+    /// <summary>
+    /// arch-review adversarial finding F3: the observer-queue onDropped callback used to
+    /// log a Warning PER dropped item — a stuck observer with a large
+    /// <see cref="StoreAndForwardOptions.ObserverQueueCapacity"/> would flood the log at
+    /// sweep rate. Extends <see cref="ObserverQueue_BoundedCapacity_DropsOldestAndCountsDrops"/>
+    /// (same bounded-queue setup) to pin the fix: many drops in one episode still
+    /// increment <see cref="StoreAndForwardService.ObserverQueueDroppedCount"/> per drop, but
+    /// produce exactly ONE observer-queue-drop Warning log — the first-drop Warning — because
+    /// the episode never runs long enough to cross the periodic rollup interval.
+    /// </summary>
+    [Fact]
+    public async Task ObserverQueue_ManyDropsInOneEpisode_LogsExactlyOneWarning()
+    {
+        var gate = new TaskCompletionSource();
+        var observer = new BlockingObserver(gate);
+        var logger = new CapturingLogger<StoreAndForwardService>();
+
+        var localDb = TestLocalDb.CreateTemp("ObsQueueDropLogRollup");
+        var storage = new StoreAndForwardStorage(localDb.Db, NullLogger<StoreAndForwardStorage>.Instance);
+        await storage.InitializeAsync();
+        var service = new StoreAndForwardService(
+            storage,
+            new StoreAndForwardOptions
+            {
+                DefaultRetryInterval = TimeSpan.Zero,
+                DefaultMaxRetries = 5,
+                RetryTimerInterval = TimeSpan.FromHours(1), // timer never fires in-test
+                ObserverQueueCapacity = 2,
+            },
+            logger,
+            cachedCallObserver: observer,
+            siteId: "site-78");
+
+        await service.StartAsync();
+        try
+        {
+            service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem,
+                _ => throw new HttpRequestException("transient"));
+
+            // Enqueue far more than the bounded capacity (2) so the queue overflows many
+            // times over in one sweep — same mechanism as the sibling test above, just a
+            // larger flood to make a per-item log flood obvious if the fix regresses.
+            for (var i = 0; i < 50; i++)
+            {
+                await service.EnqueueAsync(
+                    StoreAndForwardCategory.ExternalSystem, $"t{i}", "{}",
+                    attemptImmediateDelivery: false, retryInterval: TimeSpan.Zero,
+                    messageId: TrackedOperationId.New().ToString());
+            }
+
+            await service.RetryPendingMessagesAsync();
+
+            var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(5);
+            while (DateTime.UtcNow < deadline && service.ObserverQueueDroppedCount < 10)
+                await Task.Delay(10);
+
+            Assert.True(service.ObserverQueueDroppedCount >= 10,
+                "expected many notifications to be dropped once the bounded queue filled");
+
+            var dropWarnings = logger.Entries
+                .Where(e => e.Level == LogLevel.Warning
+                    && e.Message.Contains("audit-observer queue", StringComparison.Ordinal))
+                .ToList();
+
+            Assert.True(dropWarnings.Count == 1,
+                $"expected exactly one observer-queue-drop Warning per episode (rate-limited), " +
+                $"but got {dropWarnings.Count} for {service.ObserverQueueDroppedCount} drops");
+        }
+        finally
+        {
+            gate.TrySetResult();
+            await service.StopAsync();
+            var path = localDb.Path;
+            localDb.Dispose();
+            TestLocalDb.DeleteFiles(path);
+        }
+    }
 }