diff --git a/CLAUDE.md b/CLAUDE.md index 3ca2c6c3..9b2eb9ad 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -89,7 +89,7 @@ spec for each is `docs/requirements/Component-.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. diff --git a/docker/site-a-node-a/appsettings.Site.json b/docker/site-a-node-a/appsettings.Site.json index 19b6fb0b..2562f6e7 100644 --- a/docker/site-a-node-a/appsettings.Site.json +++ b/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, diff --git a/docker/site-a-node-b/appsettings.Site.json b/docker/site-a-node-b/appsettings.Site.json index 96d84abe..a06f9ff1 100644 --- a/docker/site-a-node-b/appsettings.Site.json +++ b/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, diff --git a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs index e0143d4c..1cd2e3d1 100644 --- a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs +++ b/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 /// was at capacity (WP2.6c). Not reset across /// / cycles — a diagnostic total for - /// the lifetime of this service instance. + /// the lifetime of this service instance. Incremented unconditionally on every drop + /// by , independent of that method's log throttling. /// private long _observerQueueDroppedCount; /// Diagnostic counter — see . public long ObserverQueueDroppedCount => Interlocked.Read(ref _observerQueueDroppedCount); + /// + /// 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 floods the + /// log at sweep rate — one Warning per dropped item. Only the LOGGING is throttled; + /// still counts every drop. + /// + private static readonly TimeSpan ObserverQueueDropLogRollupInterval = TimeSpan.FromMinutes(1); + + /// + /// of the last observer-queue-drop Warning, or + /// -1 (its initial value — TickCount64 is never negative) if none has been + /// logged yet this service-instance lifetime. Not reset across + /// / cycles, matching + /// . + /// + private long _observerQueueLastDropLogTicks = -1; + + /// + /// Drops accumulated since was last + /// logged — the count a rollup Warning reports before resetting to 0. + /// + private long _observerQueueDroppedSinceLastLog; + + /// + /// 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 + /// 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 + /// , so overlapping droppers accumulate + /// into without double-logging. + /// + 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); + } + } + /// /// Builds a bounded, single-reader observer queue with DropOldest overflow, invoking /// (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()) diff --git a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs index 82f6c724..f24032d6 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs +++ b/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); } } + + /// + /// Captures every message logged through it, keyed by . Minimal + /// test double — no scopes, no filtering — just enough to assert on log VOLUME. + /// + private sealed class CapturingLogger : ILogger + { + 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 state) where TState : notnull => null; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log( + LogLevel logLevel, EventId eventId, TState state, Exception? exception, + Func formatter) + { + var message = formatter(state, exception); + lock (_gate) _entries.Add((logLevel, message)); + } + } + + /// + /// arch-review adversarial finding F3: the observer-queue onDropped callback used to + /// log a Warning PER dropped item — a stuck observer with a large + /// would flood the log at + /// sweep rate. Extends + /// (same bounded-queue setup) to pin the fix: many drops in one episode still + /// increment 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. + /// + [Fact] + public async Task ObserverQueue_ManyDropsInOneEpisode_LogsExactlyOneWarning() + { + var gate = new TaskCompletionSource(); + var observer = new BlockingObserver(gate); + var logger = new CapturingLogger(); + + var localDb = TestLocalDb.CreateTemp("ObsQueueDropLogRollup"); + var storage = new StoreAndForwardStorage(localDb.Db, NullLogger.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); + } + } }