diff --git a/docs/requirements/Component-Communication.md b/docs/requirements/Component-Communication.md index bcf92152..9bb0a2f2 100644 --- a/docs/requirements/Component-Communication.md +++ b/docs/requirements/Component-Communication.md @@ -75,8 +75,11 @@ Delivered 2026-07-10 (`docs/plans/2026-07-10-aggregated-live-alarm-stream-plan.m - **Central live cache** (`ISiteAlarmLiveCache`, singleton `SiteAlarmLiveCacheService`): a DI singleton on the active central node. For each site with ≥1 active viewer it runs ONE shared, **reference-counted** per-site aggregator (`SiteAlarmAggregatorActor`); the first `Subscribe(siteId, onChanged)` starts it, the last subscriber leaving stops it after a short **linger** to avoid re-seed thrash. `GetCurrentAlarms(siteId)` returns the current immutable snapshot; `IsLive(siteId)` reports whether the aggregator has seeded and published at least once. - **Seed-then-stream** (copied from `DebugStreamBridgeActor` ordering): open the `SubscribeSite` stream first (buffer live deltas), run the snapshot fan-out once via the existing `DebugViewSnapshot` path (bounded by `LiveAlarmCacheSeedConcurrency`), flush the buffer with **dedup by `(InstanceUniqueName, AlarmName, SourceReference)`**, then live pass-through. Placeholders are seeded from the snapshot and never expected on the live stream. - **Failover & drift**: NodeA↔NodeB reconnect (client via `SiteStreamGrpcClient.SubscribeSiteAsync`, factory endpoint failover) re-seeds rather than serving stale; a periodic **reconcile snapshot** (default 60s) corrects any missed enable/disable so the cache can't drift indefinitely. `[PERM]` (`docs/plans/2026-05-29-native-alarms-design.md`): the cache is **purely in-memory** — no EF entity/table/migration, no persisted central alarm store — so a new active node simply re-seeds from scratch. -- **Options** (`Communication` section, `CommunicationOptions`; eagerly validated by `CommunicationOptionsValidator` / `ValidateOnStart`): `LiveAlarmCacheLinger` (default 30s), `LiveAlarmCacheReconcileInterval` (default 60s), `LiveAlarmCacheSeedConcurrency` (default 8), `LiveAlarmCacheMaxSubscribersPerSite` (default 200). +- **Options** (`Communication` section, `CommunicationOptions`; eagerly validated by `CommunicationOptionsValidator` / `ValidateOnStart`): `LiveAlarmCacheLinger` (default 30s), `LiveAlarmCacheReconcileInterval` (default 60s), `LiveAlarmCacheSeedConcurrency` (default 8), `LiveAlarmCacheMaxSubscribersPerSite` (default 200), `LiveAlarmCachePublishCoalesce` (default 250ms; `0` = publish per delta — legacy — batches an alarm storm into one snapshot copy + one viewer fan-out per window; arch review 02 round 2, N6). - **Telemetry** (`ScadaBridgeTelemetry` meter): observable gauge `scadabridge.site.alarm_cache.aggregators.active` (running per-site aggregators) and counter `scadabridge.site.alarm_cache.reconnects` (site-wide stream reconnects — a NodeA↔NodeB flip or reconcile-driven reopen; a sustained climb signals a flapping site link). +- **Accepted limitations (arch review 02 round 2, N8):** + - **Standby-node aggregators**: the live cache is per-node and `SetActorSystem` is wired on every central node, so browsing the standby node directly (diagnostic ports, e.g. 9002) starts a second, fully-functional read-only aggregator + `SubscribeSite` stream there. Accepted — not gated: the aggregator is read-only, bounded (one stream/site, viewer-capped), and torn down by the viewer linger; gating `SetActorSystem` behind the active check would break the diagnostic-browse path and buy nothing. The `[PERM]` "lives only on the active central node" claim is corrected to "per-node; in routine operation only the active node hosts viewers". + - **Site deleted while an Alarm Summary viewer is open**: the viewer's aggregator reconciles to an empty snapshot (the deleted site's instances vanish from the fan-out) until its viewers leave, then the linger stop reaps it; the site's cached gRPC channels are disposed at deletion via `SiteStreamGrpcClientFactory.RemoveSiteAsync` (see `ManagementActor.HandleDeleteSite`, arch review 02 round 2, N8). #### Site-Side gRPC Streaming Components diff --git a/docs/requirements/Component-StoreAndForward.md b/docs/requirements/Component-StoreAndForward.md index a2d39f56..f10f255f 100644 --- a/docs/requirements/Component-StoreAndForward.md +++ b/docs/requirements/Component-StoreAndForward.md @@ -80,7 +80,7 @@ There is **no maximum buffer size**. Messages accumulate in the buffer until del - The standby node applies the same operations to its own local SQLite database but is **passive**: it never runs the delivery sweep. The retry sweep is **gated to the active node** (the oldest Up member / singleton host, re-evaluated every sweep tick), so only one node delivers at a time. The standby applies replicated operations purely to keep its copy warm for a future failover. - On failover, the new active node has a near-complete copy of the buffer. In rare cases, the most recent operations may not have been replicated (e.g., a message added or removed just before failover). This can result in a few **duplicate deliveries** (message delivered but its `Remove` not yet replicated) or a few **missed retries** (message added but not replicated). Duplicate deliveries are therefore confined to the **failover window** — an in-flight delivery whose `Remove` had not yet replicated — and never occur in steady-state operation (the standby's gate keeps it from delivering the same rows). Both are acceptable trade-offs for the latency benefit. - On failover, the new active node's gate flips to active within one sweep interval and it resumes delivery from its local copy. -- **Peer-join anti-entropy resync.** Asynchronous, no-ack replication keeps the standby warm in steady state, but a standby that was **down for an extended period** (a crash, a long maintenance window) misses every operation replicated while it was gone and would otherwise diverge from the active node's buffer forever. To close that gap, whenever a node **(re)tracks its peer**, a **standby** requests a full-buffer snapshot (`RequestSfBufferResync`); the **active** node answers with up to `MaxResyncRows` (10 000) of its oldest rows (`SfBufferSnapshot`), and the standby **replaces its entire local buffer** with that snapshot (`ReplaceAllAsync`, one transaction). Only the active node answers; only a standby applies (each side checks the repo-standard leader+Up active-node predicate, safe-by-default to standby). Because replicated applies are **upserts** (see the replication apply path), any Add/Remove/Park that lands after the resync merges cleanly onto the resynced state — no primary-key conflict, no lost delta. If the buffer exceeds the 10 000-row cap the snapshot is flagged `Truncated` and the standby logs a Warning; the residual divergence beyond the cap drains naturally as the active node delivers. +- **Peer-join anti-entropy resync (chunked, ack-confirmed).** Asynchronous, no-ack replication keeps the standby warm in steady state, but a standby that was **down for an extended period** (a crash, a long maintenance window) misses every operation replicated while it was gone and would otherwise diverge from the active node's buffer forever. To close that gap, whenever a node **(re)tracks its peer**, a **standby** requests a full-buffer snapshot (`RequestSfBufferResync`); the **active** node loads up to `MaxResyncRows` (10 000) of its oldest rows and answers with a **sequence of byte-budgeted chunks** (`SfBufferSnapshotChunk`), each carrying a shared `ResyncId`, a 1-based `Sequence`, and the `TotalChunks` count. Chunking is mandatory because the monolithic snapshot exceeds Akka remoting's **default 128 000-byte frame** for any realistic backlog and `BuildHocon` sets no override — a single oversized message is silently undeliverable (review 02 round 2, **N2 High**). Rows accumulate into a chunk until the estimated payload budget (`MaxResyncChunkBytes` = 64 000, ≈50% frame headroom) or the row cap (`MaxResyncChunkRows` = 200) is hit; a single row whose payload alone exceeds the budget ships solo with a Warning. The standby **assembles all chunks of one `ResyncId`** (a new `ResyncId` discards any stale partial assembly; a partial that never completes is dropped after `resyncAssemblyTimeout`, default 30 s, and counted a replication failure), then **replaces its entire local buffer** with the assembled snapshot (`ReplaceAllAsync`, one transaction) and returns a delivery confirmation (`SfBufferResyncAck`). The active node arms an ack window (`resyncAckTimeout`, default 60 s): an acknowledged resync increments `scadabridge.store_and_forward.resync.completed`; an unacknowledged one (lost chunks / dead peer) logs a Warning and increments `scadabridge.store_and_forward.resync.ack_missing` — closing N2's silent-loss mode (previously nothing counted a lost snapshot and nothing retried until the next peer-track). Only the active node answers; only a standby applies, and both sides re-check at apply time (a mid-flight active-flip aborts the wipe) — each side checks the repo-standard **oldest-Up member** active-node predicate (singleton-host semantics via the shared `ActiveNodeEvaluator`, the **same predicate as the S&F delivery gate**; review 02 round 2, **N1 Critical** — using cluster *leadership* here let a rolling restart of the lower-address node make the delivering node wipe its own live buffer). Because replicated applies are **upserts** (see the replication apply path), any Add/Remove/Park that lands after the resync merges cleanly onto the resynced state — no primary-key conflict, no lost delta; the one accepted exception is the rare **N5** orphan-row race (a replicated `Remove` ordered before the snapshot chunks can re-add the removed row, re-delivered once and self-corrected at the next resync — inherent to no-ack replication). If the buffer exceeds the 10 000-row cap the snapshot is flagged `Truncated` and the standby logs a Warning; the residual divergence beyond the cap drains naturally as the active node delivers. The legacy monolithic `SfBufferSnapshot` message + standby handler are **retained** for rolling-upgrade compatibility (an old active node's monolithic snapshot is still applied by a new standby). ### Operation Tracking Table (lives in Site Runtime, not here) diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs index 1f8cc2ad..0d1474c2 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs @@ -47,6 +47,14 @@ public static class ScadaBridgeTelemetry 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"); + private static readonly Counter _sfResyncCompleted = + Meter.CreateCounter("scadabridge.store_and_forward.resync.completed", unit: "1", + description: "S&F anti-entropy resyncs the standby acknowledged as applied"); + + private static readonly Counter _sfResyncAckMissing = + Meter.CreateCounter("scadabridge.store_and_forward.resync.ack_missing", unit: "1", + description: "S&F resyncs answered by the active node but never acknowledged by the standby within the ack window (lost chunks / dead peer)"); + /// /// Incremented each time a per-site live-alarm aggregator re-establishes its site-wide /// gRPC stream — a NodeA↔NodeB failover flip or a reconcile-driven reopen after the @@ -107,6 +115,12 @@ public static class ScadaBridgeTelemetry /// Records one failed S&F replication dispatch. public static void RecordReplicationFailure() => _replicationFailures.Add(1); + /// Records one acknowledged (applied) S&F buffer resync. + public static void RecordSfResyncCompleted() => _sfResyncCompleted.Add(1); + + /// Records one resync whose ack window expired. + public static void RecordSfResyncAckMissing() => _sfResyncAckMissing.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.Communication/Actors/SiteAlarmAggregatorActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs index 7d19e0cb..c1c53432 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs @@ -52,21 +52,26 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers private readonly string _grpcNodeAAddress; private readonly string _grpcNodeBAddress; private readonly TimeSpan _reconcileInterval; + private readonly TimeSpan _publishCoalesce; private const int MaxRetries = 3; private const string ReconnectTimerKey = "alarm-grpc-reconnect"; private const string StabilityTimerKey = "alarm-grpc-stability"; private const string ReconcileTimerKey = "alarm-reconcile"; + private const string PublishTimerKey = "alarm-publish-coalesce"; - /// Delay between gRPC reconnection attempts. Settable for tests. - internal static TimeSpan ReconnectDelay { get; set; } = TimeSpan.FromSeconds(5); + /// True while a coalesced publish is armed (dirty deltas awaiting one tick). Actor-thread only. + private bool _publishPending; + + /// Delay between gRPC reconnection attempts (ctor-injected; production default 5s). + private readonly TimeSpan _reconnectDelay; /// /// How long a freshly-opened gRPC stream must stay up before its retry budget is - /// considered recovered (mirrors ). - /// Settable for tests. + /// considered recovered (mirrors ); + /// ctor-injected (production default 60s). /// - internal static TimeSpan StabilityWindow { get; set; } = TimeSpan.FromSeconds(60); + private readonly TimeSpan _stabilityWindow; private int _retryCount; private bool _useNodeA = true; @@ -94,6 +99,21 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers /// Ordered buffer of live deltas that arrived while a fan-out was in flight. Actor-thread only. private readonly List _buffer = new(); + /// + /// A failover re-seed was requested while a fan-out was already in flight; it must run + /// right after the in-flight one completes rather than being silently dropped (N7.1) — + /// the in-flight snapshot's read-time predates the stream death. Actor-thread only. + /// + private bool _reseedQueued; + + /// + /// Monotonic stream generation stamped on each opened gRPC stream and echoed back on its + /// error callback: a late error raced out of a previous (cancelled) stream carries a stale + /// generation and is ignored so it never burns retry budget or double-flips (N7.2). + /// Actor-thread only. + /// + private int _streamGeneration; + private const int BufferWarnThreshold = 10_000; private bool _bufferWarned; @@ -118,6 +138,15 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers /// gRPC address of the site's node A. /// gRPC address of the site's node B. /// Periodic reconcile snapshot cadence. + /// + /// Publish-coalescing window for live deltas: a positive value batches a delta storm + /// into one publish per window (review 02 round 2, N6); + /// restores per-delta publishing (legacy). Seed/reconcile publishes stay immediate. + /// + /// Delay between gRPC reconnection attempts (production 5s). + /// + /// How long a fresh gRPC stream must stay up before its retry budget recovers (production 60s). + /// public SiteAlarmAggregatorActor( string siteIdentifier, string correlationId, @@ -126,7 +155,10 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers SiteStreamGrpcClientFactory grpcFactory, string grpcNodeAAddress, string grpcNodeBAddress, - TimeSpan reconcileInterval) + TimeSpan reconcileInterval, + TimeSpan publishCoalesce, + TimeSpan reconnectDelay, + TimeSpan stabilityWindow) { _siteIdentifier = siteIdentifier; _correlationId = correlationId; @@ -136,6 +168,9 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers _grpcNodeAAddress = grpcNodeAAddress; _grpcNodeBAddress = grpcNodeBAddress; _reconcileInterval = reconcileInterval; + _publishCoalesce = publishCoalesce; + _reconnectDelay = reconnectDelay; + _stabilityWindow = stabilityWindow; // Live delta from the site-wide alarm stream (marshalled in via Self.Tell). // A received delta must NOT reset the retry budget (a flapping stream that @@ -152,6 +187,13 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers // Periodic reconcile tick (and the re-seed kicked after a reconnect). Receive(_ => OnReconcileTick()); + // Coalesced-publish tick: one publish for a batch of dirtying deltas (N6). + Receive(_ => + { + _publishPending = false; + if (!_stopped) Publish(); + }); + // Stream stayed up for StabilityWindow — recover the retry budget. Receive(_ => { @@ -163,6 +205,15 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers // gRPC stream error — flip node + reconnect + re-seed. Receive(msg => { + // Ignore a late error raced out of a previous (cancelled) stream — the + // RpcException(Cancelled) filter at SiteStreamGrpcClient.cs covers the normal + // path, but a genuine socket fault can beat the cancel (N7.2). + if (msg.Generation != _streamGeneration) + { + _log.Debug("Ignoring stale gRPC error from stream generation {0} (current {1})", + msg.Generation, _streamGeneration); + return; + } _log.Warning("Site-alarm gRPC stream error for {0}: {1}", _siteIdentifier, msg.Exception.Message); HandleGrpcError(); }); @@ -247,7 +298,11 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers if (_stopped) return; if (_fanoutInFlight) { - // Already seeding/reconciling — don't stack a second fan-out. + // A failover re-seed requested mid-fan-out must run right after the in-flight + // one — its snapshot read-time predates the stream death, so skipping it would + // serve stale up to the next 60s reconcile (N7.1). An initial-seed collision + // never queues (there is only ever one). + if (!isInitial) _reseedQueued = true; return; } @@ -300,7 +355,18 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers _log.Debug("Site-alarm {0} {1} complete: {2} alarm row(s)", _siteIdentifier, msg.IsInitial ? "seed" : "reconcile", _cache.Count); + // The fresh snapshot already carries the buffered deltas; drop any armed coalesce + // tick so we publish once, immediately. + Timers.Cancel(PublishTimerKey); + _publishPending = false; Publish(); + + // A failover re-seed requested while this fan-out was in flight runs now (N7.1). + if (_reseedQueued) + { + _reseedQueued = false; + StartFanout(isInitial: false); + } } private void OnSeedFailed(SeedFailed msg) @@ -319,7 +385,18 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers // Only publish if we already had a seed (so IsLive doesn't flip true on a // failed initial seed — the page keeps its poll fallback until we truly seed). if (_seeded) + { + Timers.Cancel(PublishTimerKey); + _publishPending = false; Publish(); + } + + // A failover re-seed requested while this fan-out was in flight runs now (N7.1). + if (_reseedQueued) + { + _reseedQueued = false; + StartFanout(isInitial: false); + } } /// @@ -362,9 +439,23 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers return; } - // Pass-through: apply and publish only if the cache actually changed. + // Pass-through: apply and (coalesced) publish only if the cache actually changed. if (ApplyDelta(delta, requireStrictlyNewer: false)) - Publish(); + SchedulePublish(); + } + + /// + /// Coalesced publish: with a positive window, the first dirtying delta arms a + /// single-shot timer and further deltas ride the same tick — one snapshot copy and + /// one viewer fan-out per window instead of per transition (N6). Zero = legacy + /// immediate publish. Last write wins, so batching never changes final state. + /// + private void SchedulePublish() + { + if (_publishCoalesce <= TimeSpan.Zero) { Publish(); return; } + if (_publishPending) return; + _publishPending = true; + Timers.StartSingleTimer(PublishTimerKey, new PublishCoalesced(), _publishCoalesce); } /// @@ -416,8 +507,9 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers _grpcCts?.Dispose(); _grpcCts = new CancellationTokenSource(); - Timers.StartSingleTimer(StabilityTimerKey, new GrpcAlarmStreamStable(), StabilityWindow); + Timers.StartSingleTimer(StabilityTimerKey, new GrpcAlarmStreamStable(), _stabilityWindow); + var generation = ++_streamGeneration; var client = _grpcFactory.GetOrCreate(_siteIdentifier, endpoint); var self = Self; var ct = _grpcCts.Token; @@ -427,7 +519,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers await client.SubscribeSiteAsync( _correlationId, alarm => self.Tell(alarm), - ex => self.Tell(new GrpcAlarmStreamError(ex)), + ex => self.Tell(new GrpcAlarmStreamError(ex, generation)), ct); }, ct); } @@ -474,7 +566,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers if (_retryCount == 1) Self.Tell(new ReconnectAlarmStream()); else - Timers.StartSingleTimer(ReconnectTimerKey, new ReconnectAlarmStream(), ReconnectDelay); + Timers.StartSingleTimer(ReconnectTimerKey, new ReconnectAlarmStream(), _reconnectDelay); } private void CleanupGrpc() @@ -520,8 +612,12 @@ internal sealed record SeedFailed(Exception Exception, bool IsInitial); /// Internal: periodic reconcile tick (and the re-seed kicked after a reconnect). internal sealed record RunReconcile; -/// Internal: site-alarm gRPC stream error occurred. -internal sealed record GrpcAlarmStreamError(Exception Exception); +/// Internal: coalesced-publish tick — flush the dirty cache to viewers once (N6). +internal sealed record PublishCoalesced; + +/// Internal: site-alarm gRPC stream error occurred, stamped with the stream +/// generation it came from so a late error from a cancelled stream can be ignored (N7.2). +internal sealed record GrpcAlarmStreamError(Exception Exception, int Generation); /// Internal: reconnect the site-alarm gRPC stream (flip node). internal sealed record ReconnectAlarmStream; diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs index b3027ac6..4667b2ec 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs @@ -67,10 +67,10 @@ public class SiteCommunicationActor : ReceiveActor, IWithTimers /// Local reference to the Deployment Manager singleton proxy. /// /// Optional override returning true when this node - /// is the active member of the site cluster. null uses the real - /// Akka leader check (the default for production - /// wiring); tests pass a stub so they do not need to load Akka.Cluster - /// into the TestKit ActorSystem. + /// is the active member of the site cluster. null uses the shared oldest-Up + /// evaluator (production wiring passes the Host's singleton-host delegate); tests + /// pass a stub so they do not need to load Akka.Cluster into the TestKit + /// ActorSystem. /// public SiteCommunicationActor( string siteId, @@ -506,24 +506,16 @@ public class SiteCommunicationActor : ReceiveActor, IWithTimers } /// - /// Default active-node check used when no override is - /// supplied. Mirrors ActiveNodeGate in the Host (and - /// ActiveNodeHealthCheck): the node is the active member of the - /// site cluster when it is the current cluster leader AND its own - /// is . Any other - /// state (still joining, leaving, no leader yet) reports standby — - /// safe-by-default, matching the standby case. + /// Default active-node check used when no override is supplied: oldest-Up member + /// semantics via the shared — the + /// same predicate as the S&F delivery gate and the replication resync authority + /// (review 02 round 2, N1). Unscoped by role: a site cluster's members all carry + /// the site role, so role scoping is a no-op here; production wiring passes the + /// Host's role-scoped IClusterNodeProvider delegate anyway. Any other state + /// (still joining, leaving) reports standby — safe-by-default. /// - private bool DefaultIsActiveCheck() - { - var cluster = Cluster.Get(Context.System); - var self = cluster.SelfMember; - if (self.Status != MemberStatus.Up) - return false; - - var leader = cluster.State.Leader; - return leader != null && leader == self.Address; - } + private bool DefaultIsActiveCheck() => + ClusterState.ActiveNodeEvaluator.SelfIsOldestUp(Cluster.Get(Context.System)); // ── Internal messages ── diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/ClusterState/ActiveNodeEvaluator.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/ClusterState/ActiveNodeEvaluator.cs new file mode 100644 index 00000000..57d7b425 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/ClusterState/ActiveNodeEvaluator.cs @@ -0,0 +1,39 @@ +using Akka.Cluster; + +namespace ZB.MOM.WW.ScadaBridge.Communication.ClusterState; + +/// +/// THE single definition of "active node" (review 01 [High]; review 02 round 2 [Critical] N1): +/// a node is active when it is the OLDEST Up member (optionally within a role scope) — i.e. +/// the member the ClusterSingletonManager places singletons on. Cluster LEADERSHIP (lowest +/// address) is an Akka-internal concept that diverges from singleton placement permanently +/// once the original first node restarts and rejoins; every product-level active/standby +/// decision must use this evaluator, never cluster.State.Leader. +/// +/// Lives in Communication (not Host) so BOTH SiteCommunicationActor and +/// SiteReplicationActor can default to it — Host cannot be referenced from either. +/// The Host's ClusterActivityEvaluator.SelfIsOldest delegates here, so the S&F +/// delivery gate (IClusterNodeProvider.SelfIsPrimary), the resync authority checks, +/// and the heartbeat IsActive stamp all share one implementation. +/// +/// +public static class ActiveNodeEvaluator +{ + /// True when self is Up and no other Up member (in the role scope) is older. + /// The Akka cluster to evaluate. + /// Optional role scope; when set, only members with this role are considered. + /// true when self is Up and the oldest Up member in the role scope. + public static bool SelfIsOldestUp(Cluster cluster, string? role = null) + { + var self = cluster.SelfMember; + if (self.Status != MemberStatus.Up) + return false; + if (role != null && !self.HasRole(role)) + return false; + + return cluster.State.Members + .Where(m => m.Status == MemberStatus.Up) + .Where(m => role == null || m.HasRole(role)) + .All(m => m.UniqueAddress.Equals(self.UniqueAddress) || self.IsOlderThan(m)); + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs index c5957947..fc9e9ffe 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs @@ -122,4 +122,13 @@ public class CommunicationOptions /// count; this just bounds the subscriber list. Default 200. /// public int LiveAlarmCacheMaxSubscribersPerSite { get; set; } = 200; + + /// + /// Publish-coalescing window for live alarm deltas: an applied delta marks the cache + /// dirty and one publish (fresh snapshot + per-viewer onChanged fan-out) fires after + /// this window, batching an alarm storm into ~4 publishes/second instead of one per + /// transition (review 02 round 2, N6). Zero = publish per delta (legacy). Seed and + /// reconcile publishes are always immediate. Default 250 ms. + /// + public TimeSpan LiveAlarmCachePublishCoalesce { get; set; } = TimeSpan.FromMilliseconds(250); } diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptionsValidator.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptionsValidator.cs index 19236555..31151b67 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptionsValidator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptionsValidator.cs @@ -85,5 +85,10 @@ public sealed class CommunicationOptionsValidator : OptionsValidatorBase= 1, $"Communication:LiveAlarmCacheMaxSubscribersPerSite must be at least 1 (was {options.LiveAlarmCacheMaxSubscribersPerSite})."); + + // Publish-coalescing window drives a single-shot timer; TimeSpan.Zero is valid + // (publish per delta — legacy), only a negative value is invalid. + builder.RequireThat(options.LiveAlarmCachePublishCoalesce >= TimeSpan.Zero, + $"Communication:LiveAlarmCachePublishCoalesce must be zero or a positive duration (was {options.LiveAlarmCachePublishCoalesce})."); } } diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/ISiteAlarmLiveCache.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/ISiteAlarmLiveCache.cs index 44a13efe..8fe7eebc 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/ISiteAlarmLiveCache.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/ISiteAlarmLiveCache.cs @@ -11,9 +11,12 @@ namespace ZB.MOM.WW.ScadaBridge.Communication; /// reflects alarm transitions in near-real-time instead of re-polling every 15s. /// /// Hard constraint (locked [PERM]): there is NO persisted central alarm -/// store. This cache is purely in-memory and lives only on the active central node — -/// no EF entity/table/migration backs it. On a NodeA↔NodeB failover the new active -/// node re-seeds from scratch. +/// store. This cache is purely in-memory, per-node — no EF entity/table/migration backs +/// it. In routine operation only the active node (behind Traefik) hosts viewers, so only +/// it runs aggregators; browsing the standby node directly (diagnostic ports) starts an +/// independent read-only aggregator there — accepted (arch review 02 round 2, N8): +/// bounded (one stream/site, viewer-capped), read-only, and torn down by the viewer +/// linger. On a NodeA↔NodeB failover the new active node re-seeds from scratch. /// /// /// Reference-counted lifecycle: the first for a site diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs index ea0c5036..f27d50a7 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs @@ -233,7 +233,10 @@ public sealed class SiteAlarmLiveCacheService : ISiteAlarmLiveCache _grpcFactory, grpcA, grpcB, - _options.LiveAlarmCacheReconcileInterval)); + _options.LiveAlarmCacheReconcileInterval, + _options.LiveAlarmCachePublishCoalesce, + TimeSpan.FromSeconds(5), // reconnect delay — former ReconnectDelay static default + TimeSpan.FromSeconds(60))); // stability window — former StabilityWindow static default entry.Actor = system.ActorOf(props, $"site-alarm-aggregator-{entry.SiteId}-{Guid.NewGuid():N}"); entry.StartRetryTimer?.Dispose(); @@ -291,7 +294,9 @@ public sealed class SiteAlarmLiveCacheService : ISiteAlarmLiveCache return null; } - if (string.IsNullOrWhiteSpace(site.GrpcNodeAAddress) || string.IsNullOrWhiteSpace(site.GrpcNodeBAddress)) + var nodeA = site.GrpcNodeAAddress; + var nodeB = site.GrpcNodeBAddress; + if (string.IsNullOrWhiteSpace(nodeA) && string.IsNullOrWhiteSpace(nodeB)) { _logger.LogWarning( "Live alarm cache: site {SiteId} ({SiteIdentifier}) has no gRPC node addresses; " + @@ -299,7 +304,13 @@ public sealed class SiteAlarmLiveCacheService : ISiteAlarmLiveCache return null; } - return (site.SiteIdentifier, site.GrpcNodeAAddress!, site.GrpcNodeBAddress!); + // Single-endpoint site (review 02 round 2, N9): reuse the sole configured + // endpoint for both slots — the aggregator's NodeA↔NodeB failover flip + // degenerates to a reconnect against the same node, which is exactly the + // right behavior for a one-node site. + var grpcA = !string.IsNullOrWhiteSpace(nodeA) ? nodeA! : nodeB!; + var grpcB = !string.IsNullOrWhiteSpace(nodeB) ? nodeB! : nodeA!; + return (site.SiteIdentifier, grpcA, grpcB); } // ── Snapshot fan-out (the seed / reconcile) ───────────────────────────────── diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs index 7f060921..7a7d8517 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs @@ -772,10 +772,20 @@ akka {{ var replicationLogger = _serviceProvider.GetRequiredService() .CreateLogger(); + // ONE active-node predicate instance governs the S&F delivery gate, the resync + // authority checks (SiteReplicationActor), and the heartbeat IsActive stamp + // (SiteCommunicationActor, wired below) — review 02 round 2, N1. Null in + // non-clustered test hosts: the actors fall back to the shared oldest-Up + // evaluator, never to a leader check. + var clusterNodeProvider = _serviceProvider.GetService(); + Func? activeNodeCheck = clusterNodeProvider != null + ? () => clusterNodeProvider.SelfIsPrimary + : null; + var replicationActor = _actorSystem!.ActorOf( Props.Create(() => new SiteReplicationActor( storage, sfStorage, replicationService, siteRole, replicationLogger, - deploymentConfigFetcher, null, siteRuntimeOptionsValue, null)), + deploymentConfigFetcher, activeNodeCheck, siteRuntimeOptionsValue, null)), "site-replication"); // Wire S&F replication handler to forward operations via the replication actor @@ -823,7 +833,8 @@ akka {{ Props.Create(() => new SiteCommunicationActor( _nodeOptions.SiteId!, _communicationOptions, - dmProxy)), + dmProxy, + activeNodeCheck)), "site-communication"); // Register local handlers with SiteCommunicationActor @@ -872,10 +883,11 @@ akka {{ // tick so failover resumes delivery within one RetryTimerInterval. // IClusterNodeProvider.SelfIsPrimary is the canonical "this node is the // oldest Up member (singleton host)" check from the cluster-infrastructure - // fix plan — the shared helper this seam was designed to accept. In a - // non-clustered test host the provider is unregistered, so the gate stays - // unset and the sweep is ungated (legacy behaviour, preserved). - var clusterNodeProvider = _serviceProvider.GetService(); + // fix plan — the shared helper this seam was designed to accept. The provider + // is resolved once above (the same instance gating the resync authority and + // heartbeat IsActive stamp — N1). In a non-clustered test host the provider is + // unregistered, so the gate stays unset and the sweep is ungated (legacy + // behaviour, preserved). if (clusterNodeProvider != null) { storeAndForwardService.SetDeliveryGate(() => clusterNodeProvider.SelfIsPrimary); diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/Health/ClusterActivityEvaluator.cs b/src/ZB.MOM.WW.ScadaBridge.Host/Health/ClusterActivityEvaluator.cs index 382ade94..b4729cb5 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/Health/ClusterActivityEvaluator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/Health/ClusterActivityEvaluator.cs @@ -13,23 +13,15 @@ namespace ZB.MOM.WW.ScadaBridge.Host.Health; /// public static class ClusterActivityEvaluator { - /// True when self is Up and no other Up member (in the role scope) is older. + /// True when self is Up and no other Up member (in the role scope) is older. + /// Delegates to the shared — + /// one implementation for the delivery gate, the resync authority checks, and the + /// heartbeat IsActive stamp (review 02 round 2, N1). /// The Akka cluster to evaluate. /// Optional role scope; when set, only members with this role are considered. /// true when self is Up and the oldest Up member in the role scope. - public static bool SelfIsOldest(Cluster cluster, string? role = null) - { - var self = cluster.SelfMember; - if (self.Status != MemberStatus.Up) - return false; - if (role != null && !self.HasRole(role)) - return false; - - return cluster.State.Members - .Where(m => m.Status == MemberStatus.Up) - .Where(m => role == null || m.HasRole(role)) - .All(m => m.UniqueAddress.Equals(self.UniqueAddress) || self.IsOlderThan(m)); - } + public static bool SelfIsOldest(Cluster cluster, string? role = null) => + Communication.ClusterState.ActiveNodeEvaluator.SelfIsOldestUp(cluster, role); /// The oldest Up member in the role scope, or null while none is Up. Used for Primary/Standby labelling. /// The Akka cluster to evaluate. diff --git a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs index 9f057d10..71e692e4 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs @@ -1583,6 +1583,18 @@ public class ManagementActor : ReceiveActor await repo.SaveChangesAsync(); var commService = sp.GetService(); commService?.RefreshSiteAddresses(); + + // Dispose the deleted site's cached gRPC channels — RemoveSiteAsync is the + // factory's designed site-deletion disposal path and previously had no + // production caller (arch review 02 round 2, N8): a deleted site's channels + // (both node endpoints) otherwise persist until process restart. Null-safe: + // test/composition roots without the factory skip it. Any live-alarm + // aggregator for the site reconciles to an empty snapshot and is reaped by + // the viewer linger (documented acceptance — see Component-Communication.md). + var grpcFactory = sp.GetService(); + if (grpcFactory is not null && site is not null) + await grpcFactory.RemoveSiteAsync(site.SiteIdentifier); + await AuditAsync(sp, user, "Delete", "Site", cmd.SiteId.ToString(), site?.Name ?? cmd.SiteId.ToString(), null); return true; } diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs index 0123b049..63066d2d 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs @@ -18,7 +18,7 @@ namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors; /// Inbound: receives replicated operations from peer and applies to local SQLite. /// Uses fire-and-forget (Tell) — no ack wait per design. /// -public class SiteReplicationActor : ReceiveActor +public class SiteReplicationActor : ReceiveActor, IWithTimers { private readonly SiteStorageService _storage; private readonly StoreAndForwardStorage _sfStorage; @@ -32,6 +32,30 @@ public class SiteReplicationActor : ReceiveActor private readonly TimeSpan _configFetchRetryDelay; private Address? _peerAddress; + /// Akka timer scheduler injected by the framework via . + public ITimerScheduler Timers { get; set; } = null!; + + // ── Chunked-resync assembly (standby side; actor-thread only) ── + private string? _assemblingResyncId; + private int _assemblingTotalChunks; + private bool _assemblingTruncated; + private readonly Dictionary> _assemblingChunks = new(); + private const string ResyncAssemblyTimerKey = "sf-resync-assembly-timeout"; + + /// How long a partial chunk assembly may wait for its missing chunks before + /// being discarded (a lost chunk = lost resync; the next peer-track retries). Ctor + /// test seam; production default 30 s. + private readonly TimeSpan _resyncAssemblyTimeout; + + // ── Resync delivery confirmation (active side; actor-thread only) ── + private string? _pendingAckResyncId; + private const string ResyncAckTimerKey = "sf-resync-ack-timeout"; + + /// How long the active node waits for the standby's + /// before warning + counting the resync as unacknowledged (lost chunks / dead peer). Ctor + /// test seam; production default 60 s. + private readonly TimeSpan _resyncAckTimeout; + /// /// Maximum rows an active node returns in a single anti-entropy resync snapshot. /// A standby whose buffer exceeded this (Truncated snapshot) resyncs the oldest @@ -39,6 +63,48 @@ public class SiteReplicationActor : ReceiveActor /// private const int MaxResyncRows = 10_000; + /// + /// Estimated per-chunk payload budget for a resync snapshot. Akka remoting's default + /// maximum-frame-size is 128 000 bytes and BuildHocon sets no override, + /// so the monolithic is silently undeliverable for any + /// realistic backlog (review 02 round 2, N2). 64 000 bytes leaves ≈50% headroom for + /// the JSON envelope, CLR type manifests, and the non-payload columns. + /// + internal const int MaxResyncChunkBytes = 64_000; + + /// Row cap per resync chunk (bounds a chunk even when every row is tiny). + internal const int MaxResyncChunkRows = 200; + + /// + /// Splits a resync snapshot into chunks that fit Akka remoting's default + /// 128 000-byte frame (review 02 round 2, N2): rows accumulate until the estimated + /// payload budget or the row cap is hit. Estimation is payload-dominated + /// (payload_json length + 512 bytes fixed overhead per row); a single row whose + /// payload exceeds the budget ships alone (Warning at the call site). Order is + /// preserved (oldest-first, matching GetAllMessagesAsync). + /// + internal static List> ChunkForRemoting( + IReadOnlyList rows, int maxChunkBytes, int maxChunkRows) + { + var chunks = new List>(); + var current = new List(); + var currentBytes = 0; + foreach (var row in rows) + { + var estimate = (row.PayloadJson?.Length ?? 0) + 512; + if (current.Count > 0 && (currentBytes + estimate > maxChunkBytes || current.Count >= maxChunkRows)) + { + chunks.Add(current); + current = new List(); + currentBytes = 0; + } + current.Add(row); + currentBytes += estimate; + } + if (current.Count > 0) chunks.Add(current); + return chunks; + } + /// /// Initializes a new and registers Akka message handlers. /// @@ -54,10 +120,11 @@ public class SiteReplicationActor : ReceiveActor /// config never crosses the intra-site Akka hop. Null on nodes/tests without a fetcher. /// /// - /// Test seam for the active-node check that gates the buffer-resync roles (a - /// standby requests a resync, the active node answers). Null (production) uses - /// the repo-standard leader+Up check via — swap point for - /// plan 01's shared active-node helper. + /// Active-node check that gates the buffer-resync roles (a standby requests a + /// resync, the active node answers). Production wiring passes the Host's + /// IClusterNodeProvider.SelfIsPrimary delegate (the same instance gating the + /// S&F delivery sweep); null falls back to the shared oldest-Up evaluator + /// (). /// /// Site runtime options, including the config-fetch retry count; production defaults apply when null. /// Delay between config-fetch retry attempts; defaults to 2 seconds when null. @@ -70,7 +137,9 @@ public class SiteReplicationActor : ReceiveActor IDeploymentConfigFetcher? configFetcher = null, Func? isActiveOverride = null, SiteRuntimeOptions? options = null, - TimeSpan? configFetchRetryDelay = null) + TimeSpan? configFetchRetryDelay = null, + TimeSpan? resyncAssemblyTimeout = null, + TimeSpan? resyncAckTimeout = null) { _storage = storage; _sfStorage = sfStorage; @@ -80,6 +149,8 @@ public class SiteReplicationActor : ReceiveActor _logger = logger; _cluster = Cluster.Get(Context.System); _isActive = isActiveOverride ?? DefaultIsActive; + _resyncAssemblyTimeout = resyncAssemblyTimeout ?? TimeSpan.FromSeconds(30); + _resyncAckTimeout = resyncAckTimeout ?? TimeSpan.FromSeconds(60); // UA2: bound the standby's replicated-config fetch retries. At least one // attempt always runs; the fixed inter-attempt delay is a test seam // (production default 2 s). @@ -110,7 +181,30 @@ public class SiteReplicationActor : ReceiveActor // Anti-entropy — full S&F buffer resync on peer (re)join Receive(HandleRequestSfBufferResync); - Receive(HandleSfBufferSnapshot); + Receive(HandleSfResyncSnapshotLoaded); + Receive(HandleSfBufferSnapshotChunk); + Receive(HandleResyncAssemblyTimedOut); + Receive(msg => + { + if (msg.ResyncId == _pendingAckResyncId) + { + _pendingAckResyncId = null; + Timers.Cancel(ResyncAckTimerKey); + } + ScadaBridgeTelemetry.RecordSfResyncCompleted(); + _logger.LogInformation("S&F resync {ResyncId} acknowledged by standby: {Rows} row(s) applied", + msg.ResyncId, msg.RowCount); + }); + Receive(msg => + { + if (msg.ResyncId != _pendingAckResyncId) return; + _pendingAckResyncId = null; + ScadaBridgeTelemetry.RecordSfResyncAckMissing(); + _logger.LogWarning( + "S&F resync {ResyncId} was never acknowledged within {Window} — snapshot chunks may have been lost (frame drop / dead peer); the next peer-track retries", + msg.ResyncId, _resyncAckTimeout); + }); + Receive(HandleSfBufferSnapshot); // legacy monolithic handler — retained for rolling compat } /// @@ -181,21 +275,17 @@ public class SiteReplicationActor : ReceiveActor } /// - /// Repo-standard active-node check: this node is active when it is the current - /// cluster leader AND its own is - /// . Mirrors SiteCommunicationActor.DefaultIsActiveCheck - /// (swap point for plan 01's shared helper). Any other state reports standby — - /// safe-by-default. + /// Repo-standard active-node check: this node is active when it is the OLDEST Up + /// member carrying the site role — the same oldest-Up semantics as the S&F delivery + /// gate (IClusterNodeProvider.SelfIsPrimary → ClusterActivityEvaluator → shared + /// ActiveNodeEvaluator). NEVER the cluster leader: leadership is lowest-address and + /// diverges from singleton/delivery placement permanently after the lower-address + /// node restarts — the divergence that made the delivering node wipe its own live + /// buffer via a wrong-direction resync (review 02 round 2, N1 Critical). Any other + /// state reports standby — safe-by-default. /// - private bool DefaultIsActive() - { - var self = _cluster.SelfMember; - if (self.Status != MemberStatus.Up) - return false; - - var leader = _cluster.State.Leader; - return leader != null && leader == self.Address; - } + private bool DefaultIsActive() => + Communication.ClusterState.ActiveNodeEvaluator.SelfIsOldestUp(_cluster, _siteRole); /// /// Evaluates the active-node check, treating a throwing check as standby @@ -391,9 +481,10 @@ public class SiteReplicationActor : ReceiveActor /// /// Active-node side of the anti-entropy resync: answers a standby's - /// with a full-buffer - /// (up to oldest rows). A non-active node ignores the - /// request — only the authoritative node may answer. + /// with a sequence of byte-budgeted + /// s (up to oldest rows). + /// A non-active node ignores the request — only the authoritative node may answer. + /// The snapshot is piped back to Self so chunking + ack bookkeeping stays actor-safe. /// private void HandleRequestSfBufferResync(RequestSfBufferResync msg) { @@ -405,9 +496,37 @@ public class SiteReplicationActor : ReceiveActor var replyTo = Sender; _sfStorage.GetAllMessagesAsync(MaxResyncRows).PipeTo( - replyTo, Self, - success: result => new SfBufferSnapshot(result.Messages, result.Truncated)); + failure: ex => new Status.Failure(ex), + success: result => new SfResyncSnapshotLoaded(replyTo, result.Messages, result.Truncated)); + } + + /// + /// Active-node continuation: the resync snapshot finished loading; chunk it to fit the + /// remoting frame and send the sequenced chunks to the requester (all sharing one + /// resyncId). Task 7 arms the ack-timeout here. + /// + private void HandleSfResyncSnapshotLoaded(SfResyncSnapshotLoaded msg) + { + var chunks = ChunkForRemoting(msg.Messages, MaxResyncChunkBytes, MaxResyncChunkRows); + if (chunks.Count == 0) chunks.Add(new List()); // empty buffer still resyncs (clears the standby) + var resyncId = Guid.NewGuid().ToString("N"); + for (var i = 0; i < chunks.Count; i++) + { + if (chunks[i].Count == 1 && (chunks[i][0].PayloadJson?.Length ?? 0) + 512 > MaxResyncChunkBytes) + _logger.LogWarning( + "Resync row {Id} alone exceeds the chunk budget ({Bytes}B payload); sending solo — it may exceed the remoting frame", + chunks[i][0].Id, chunks[i][0].PayloadJson?.Length ?? 0); + msg.ReplyTo.Tell(new SfBufferSnapshotChunk(resyncId, i + 1, chunks.Count, chunks[i], msg.Truncated), Self); + } + _logger.LogInformation( + "Answered S&F resync request with {Rows} row(s) in {Chunks} chunk(s), resyncId={ResyncId}", + msg.Messages.Count, chunks.Count, resyncId); + // Arm the ack window: absence of an SfBufferResyncAck within it surfaces as a + // Warning + counter (the silent-loss mode N2 flagged). Single-outstanding-resync + // bookkeeping: a new request supersedes by overwriting the id and restarting the timer. + _pendingAckResyncId = resyncId; + Timers.StartSingleTimer(ResyncAckTimerKey, new ResyncAckTimedOut(resyncId), _resyncAckTimeout); } /// @@ -435,13 +554,120 @@ public class SiteReplicationActor : ReceiveActor _logger.LogInformation( "Applying S&F buffer resync snapshot ({Count} rows), replacing local buffer", msg.Messages.Count); - _sfStorage.ReplaceAllAsync(msg.Messages) + Task.Run(async () => + { + // Belt-and-braces (N1): re-check at apply time. ReplaceAllAsync discards + // every in-flight row (StoreAndForwardStorage.cs "Never call on an active + // node"); a flip between message receipt and this point must abort. + if (SafeIsActive()) + { + _logger.LogWarning( + "Discarding S&F buffer resync snapshot: this node became active before apply"); + return; + } + await _sfStorage.ReplaceAllAsync(msg.Messages); + }) .ContinueWith(t => { if (t.IsFaulted) _logger.LogError(t.Exception, "Failed to apply S&F buffer resync snapshot"); }); } + + /// + /// Standby-node side of the chunked anti-entropy resync: accumulates the chunks of one + /// ResyncId, and once all have arrived, assembles them in sequence order and + /// replaces the local buffer atomically, then acks. A new ResyncId discards any + /// stale partial assembly (review 02 round 2, N2). An active node ignores chunks. + /// + private void HandleSfBufferSnapshotChunk(SfBufferSnapshotChunk msg) + { + if (SafeIsActive()) + { + _logger.LogDebug("Ignoring S&F resync chunk — this node is active"); + return; + } + + if (_assemblingResyncId != msg.ResyncId) + { + if (_assemblingResyncId != null) + _logger.LogWarning( + "Discarding partial S&F resync assembly {Old} ({Have}/{Want} chunks): a new resync {New} superseded it", + _assemblingResyncId, _assemblingChunks.Count, _assemblingTotalChunks, msg.ResyncId); + _assemblingResyncId = msg.ResyncId; + _assemblingTotalChunks = msg.TotalChunks; + _assemblingTruncated = msg.Truncated; + _assemblingChunks.Clear(); + } + + _assemblingChunks[msg.Sequence] = msg.Messages; + Timers.StartSingleTimer(ResyncAssemblyTimerKey, new ResyncAssemblyTimedOut(msg.ResyncId), _resyncAssemblyTimeout); + + if (_assemblingChunks.Count < _assemblingTotalChunks) + return; + + // Complete: assemble in sequence order and apply atomically. + var assembled = Enumerable.Range(1, _assemblingTotalChunks) + .SelectMany(seq => _assemblingChunks[seq]) + .ToList(); + var resyncId = _assemblingResyncId!; + var truncated = _assemblingTruncated; + _assemblingResyncId = null; + _assemblingChunks.Clear(); + Timers.Cancel(ResyncAssemblyTimerKey); + + if (truncated) + _logger.LogWarning( + "S&F buffer resync snapshot truncated at {Cap} rows; divergence beyond the cap drains naturally", + MaxResyncRows); + _logger.LogInformation( + "Applying chunked S&F resync {ResyncId} ({Count} rows), replacing local buffer", resyncId, assembled.Count); + + // KNOWN, ACCEPTED race (review 02 round 2, N5 — do NOT "fix" this into something + // worse): a replicated Remove sent after the active node read its snapshot but + // before the snapshot's chunks is ordered BEFORE them on the wire (same + // sender/receiver pair), so this apply can re-add the removed row → an orphan + // Pending row that a later failover re-delivers ONCE. Bounded, self-correcting at + // the next resync, and inherent to no-ack replication; a delivered-side dedup or + // op-sequencing scheme would cost far more than one rare duplicate. + var replyTo = Sender; + Task.Run(async () => + { + if (SafeIsActive()) // belt-and-braces, mirrors the monolithic path (T3) + { + _logger.LogWarning("Discarding chunked S&F resync {ResyncId}: node became active before apply", resyncId); + return; + } + await _sfStorage.ReplaceAllAsync(assembled); + replyTo.Tell(new SfBufferResyncAck(resyncId, assembled.Count)); + }) + .ContinueWith(t => + { + if (t.IsFaulted) + _logger.LogError(t.Exception, "Failed to apply chunked S&F resync {ResyncId}", resyncId); + }); + } + + private void HandleResyncAssemblyTimedOut(ResyncAssemblyTimedOut msg) + { + if (_assemblingResyncId != msg.ResyncId) return; // superseded already + _logger.LogWarning( + "S&F resync assembly {ResyncId} timed out with {Have}/{Want} chunks — discarding partial (a lost chunk; the next peer-track retries)", + msg.ResyncId, _assemblingChunks.Count, _assemblingTotalChunks); + ScadaBridgeTelemetry.RecordReplicationFailure(); + _assemblingResyncId = null; + _assemblingChunks.Clear(); + } + + /// Internal: the resync snapshot finished loading; chunk and send to the requester. + internal sealed record SfResyncSnapshotLoaded( + IActorRef ReplyTo, List Messages, bool Truncated); + + /// Internal: a partial chunk assembly for exceeded its window. + internal sealed record ResyncAssemblyTimedOut(string ResyncId); + + /// Internal: the active node's ack window for expired. + internal sealed record ResyncAckTimedOut(string ResyncId); } /// @@ -458,3 +684,24 @@ public sealed record RequestSfBufferResync; /// Akka remoting; the message list rides the default serializer. /// public sealed record SfBufferSnapshot(List Messages, bool Truncated); + +/// +/// Active→standby: one sequenced chunk of a full-buffer anti-entropy snapshot +/// (review 02 round 2, N2 — the monolithic exceeds Akka +/// remoting's default 128 000-byte frame for any realistic backlog). All chunks of one +/// resync share ; is 1-based up to +/// . Additive message — the legacy monolithic snapshot +/// handler is retained for rolling upgrades. Crosses intra-site Akka remoting (NOT +/// ClusterClient — ClusterClientContractLockTests is intentionally not involved). +/// +public sealed record SfBufferSnapshotChunk( + string ResyncId, int Sequence, int TotalChunks, + List Messages, bool Truncated); + +/// +/// Standby→active: delivery confirmation — the standby assembled all chunks of +/// and applied them atomically ( +/// rows installed). Absence within the ack window is surfaced by the active node +/// (Warning + counter) — the silent-loss mode N2 flagged. +/// +public sealed record SfBufferResyncAck(string ResyncId, int RowCount); diff --git a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardOptionsValidator.cs b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardOptionsValidator.cs index a9ce0f2f..b1a16d33 100644 --- a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardOptionsValidator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardOptionsValidator.cs @@ -31,5 +31,13 @@ public sealed class StoreAndForwardOptionsValidator : OptionsValidatorBase= 0, $"ScadaBridge:StoreAndForward:DefaultMaxRetries must be >= 0 " + $"(was {options.DefaultMaxRetries})."); + + builder.RequireThat(options.SweepBatchLimit >= 0, + $"ScadaBridge:StoreAndForward:SweepBatchLimit must be >= 0 " + + $"(was {options.SweepBatchLimit}); it bounds due rows per retry sweep — 0 means unlimited (legacy)."); + + builder.RequireThat(options.SweepTargetParallelism >= 1, + $"ScadaBridge:StoreAndForward:SweepTargetParallelism must be >= 1 " + + $"(was {options.SweepTargetParallelism}); it caps concurrent (category,target) sweep lanes — 1 means serial."); } } diff --git a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs index 4553c82c..2b8dda17 100644 --- a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs @@ -394,7 +394,7 @@ public class StoreAndForwardService }); _retryTimer = new Timer( - _ => Volatile.Write(ref _sweepTask, RetryPendingMessagesAsync()), + _ => KickSweep(), null, _options.RetryTimerInterval, _options.RetryTimerInterval); @@ -658,17 +658,37 @@ public class StoreAndForwardService /// /// Kicks a background retry sweep now (fire-and-forget). No-op before /// (storage may be uninitialized and the timer, whose - /// presence gates this, is not yet set). Overlap-safe: the sweep's - /// CAS makes a concurrent kick a cheap no-op. + /// presence gates this, is not yet set). Overlap-safe: publishes into + /// only when the CAS is won — see . /// public void TriggerSweep() { if (_retryTimer == null) return; - Volatile.Write(ref _sweepTask, RetryPendingMessagesAsync()); + KickSweep(); + } + + /// The current drain handle — test seam for the N3 clobber regression. + internal Task? CurrentSweepTaskForTest => Volatile.Read(ref _sweepTask); + + /// + /// Starts a sweep IF none is in flight, publishing the new task into + /// only when this call wins the + /// CAS. A kick that loses the CAS returns without touching — + /// pre-fix it overwrote the drain handle with an instantly-completed no-op, so + /// proceeded with disposal under a still-running sweep + /// (review 02 round 2, N3) — defeating the drain exactly when a sweep outlives a tick. + /// + private void KickSweep() + { + if (Interlocked.CompareExchange(ref _retryInProgress, 1, 0) != 0) + return; + Volatile.Write(ref _sweepTask, RunSweepOwnedAsync()); } /// /// Background retry sweep. Processes all pending messages that are due for retry. + /// Self-CASes for direct callers (existing tests); the entry CAS is skipped when + /// ownership is already held by . /// /// A task representing the asynchronous retry sweep. internal async Task RetryPendingMessagesAsync() @@ -676,7 +696,16 @@ public class StoreAndForwardService // Prevent overlapping retry sweeps if (Interlocked.CompareExchange(ref _retryInProgress, 1, 0) != 0) return; + await RunSweepOwnedAsync(); + } + /// + /// The actual retry sweep body. The caller MUST already own the + /// flag (won the CAS); this method releases it in its + /// finally. Never call directly without holding ownership. + /// + private async Task RunSweepOwnedAsync() + { try { var gate = _deliveryGate; diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/ActiveNodeEvaluatorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/ActiveNodeEvaluatorTests.cs new file mode 100644 index 00000000..07589409 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/ActiveNodeEvaluatorTests.cs @@ -0,0 +1,57 @@ +using System.Net; +using System.Net.Sockets; +using Akka.Actor; +using Akka.Configuration; +using Xunit; +using ZB.MOM.WW.ScadaBridge.Communication.ClusterState; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Tests; + +public class ActiveNodeEvaluatorTests : IAsyncLifetime +{ + private ActorSystem? _system; + + public async Task InitializeAsync() + { + var port = FreePort(); + var config = ConfigurationFactory.ParseString($@" +akka {{ + actor.provider = cluster + remote.dot-netty.tcp {{ hostname = ""127.0.0.1"", port = {port} }} + cluster {{ + seed-nodes = [""akka.tcp://ane-test@127.0.0.1:{port}""] + roles = [""site-x""] + min-nr-of-members = 1 + }} +}}"); + _system = ActorSystem.Create("ane-test", config); + var cluster = Akka.Cluster.Cluster.Get(_system); + var deadline = DateTime.UtcNow.AddSeconds(20); + while (cluster.SelfMember.Status != Akka.Cluster.MemberStatus.Up && DateTime.UtcNow < deadline) + await Task.Delay(100); + } + + public async Task DisposeAsync() { if (_system != null) await _system.Terminate(); } + + [Fact] + public void SelfIsOldestUp_SoleUpMember_ReturnsTrue() + { + var cluster = Akka.Cluster.Cluster.Get(_system!); + Assert.True(ActiveNodeEvaluator.SelfIsOldestUp(cluster)); + Assert.True(ActiveNodeEvaluator.SelfIsOldestUp(cluster, "site-x")); + } + + [Fact] + public void SelfIsOldestUp_RoleNotHeld_ReturnsFalse() + { + var cluster = Akka.Cluster.Cluster.Get(_system!); + Assert.False(ActiveNodeEvaluator.SelfIsOldestUp(cluster, "role-nonexistent")); + } + + private static int FreePort() + { + using var l = new TcpListener(IPAddress.Loopback, 0); + l.Start(); + return ((IPEndPoint)l.LocalEndpoint).Port; + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsValidatorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsValidatorTests.cs index c29d677c..dafba565 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsValidatorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsValidatorTests.cs @@ -86,4 +86,22 @@ public class CommunicationOptionsValidatorTests Assert.True(result.Failed); Assert.Contains("LiveAlarmCacheMaxSubscribersPerSite", result.FailureMessage); } + + // ── R2 T10: live-delta publish-coalescing window (N6) ──────────────────────── + + [Fact] + public void ZeroLiveAlarmCachePublishCoalesce_IsValid() + { + // Zero = publish per delta (legacy behavior). + var result = Validate(new CommunicationOptions { LiveAlarmCachePublishCoalesce = TimeSpan.Zero }); + Assert.True(result.Succeeded, result.FailureMessage); + } + + [Fact] + public void NegativeLiveAlarmCachePublishCoalesce_IsRejected() + { + var result = Validate(new CommunicationOptions { LiveAlarmCachePublishCoalesce = TimeSpan.FromMilliseconds(-1) }); + Assert.True(result.Failed); + Assert.Contains("LiveAlarmCachePublishCoalesce", result.FailureMessage); + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs index 32846846..35c09662 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs @@ -24,10 +24,12 @@ public class SiteAlarmAggregatorActorTests : TestKit public SiteAlarmAggregatorActorTests() : base(@"akka.loglevel = WARNING") { - SiteAlarmAggregatorActor.ReconnectDelay = TimeSpan.FromMilliseconds(50); - SiteAlarmAggregatorActor.StabilityWindow = TimeSpan.FromSeconds(30); } + // Former process-global static test seams, now ctor-injected per actor (R2 T12). + private static readonly TimeSpan TestReconnectDelay = TimeSpan.FromMilliseconds(50); + private static readonly TimeSpan TestStabilityWindow = TimeSpan.FromSeconds(30); + // ── Test doubles ──────────────────────────────────────────────────────────── /// @@ -137,7 +139,7 @@ public class SiteAlarmAggregatorActorTests : TestKit } private (IActorRef Actor, SeedStub Seed, PublishSink Sink, MockSiteAlarmStreamClientFactory Factory) CreateActor( - TimeSpan? reconcileInterval = null) + TimeSpan? reconcileInterval = null, TimeSpan? publishCoalesce = null) { var seed = new SeedStub(); var sink = new PublishSink(); @@ -145,7 +147,10 @@ public class SiteAlarmAggregatorActorTests : TestKit var props = Props.Create(() => new SiteAlarmAggregatorActor( SiteId, "corr-1", seed.Seed, sink.Publish, factory, GrpcNodeA, GrpcNodeB, - reconcileInterval ?? TimeSpan.FromMinutes(10))); + reconcileInterval ?? TimeSpan.FromMinutes(10), + publishCoalesce ?? TimeSpan.Zero, + TestReconnectDelay, + TestStabilityWindow)); var actor = Sys.ActorOf(props); return (actor, seed, sink, factory); @@ -396,6 +401,89 @@ public class SiteAlarmAggregatorActorTests : TestKit AwaitCondition(() => TotalSubs() > before, TimeSpan.FromSeconds(3)); } + // ── R2 T11: queued re-seed after reconnect + stream-generation stamp (N7.1/N7.2) ── + + [Fact] + public void ReseedRequestedDuringInFlightFanout_IsQueued_NotDropped() + { + var (actor, seed, sink, factory) = CreateActor(publishCoalesce: TimeSpan.Zero); + AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); + seed.CompleteNext(); // initial seed done (CallCount 1) + AwaitAssert(() => Assert.Equal(1, sink.Count)); + AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3)); + + actor.Tell(new RunReconcile()); // reconcile fan-out now in flight (CallCount 2) + AwaitAssert(() => Assert.Equal(2, seed.CallCount)); + + // Stream error while the reconcile is in flight → the failover re-seed must not be + // silently skipped. Pre-fix: StartFanout no-ops and CallCount stays 2 until the next + // 60s reconcile tick. + factory.ClientFor(GrpcNodeA).Subs.Last().OnError(new Exception("stream fault")); + + seed.CompleteNext(); // finish the in-flight reconcile + AwaitAssert(() => Assert.Equal(3, seed.CallCount)); // queued re-seed ran immediately after + } + + [Fact] + public void LateErrorFromAPreviousStreamGeneration_IsIgnored() + { + var (_, seed, _, factory) = CreateActor(publishCoalesce: TimeSpan.Zero); + AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); + seed.CompleteNext(); + AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3)); + + int TotalSubs() => factory.ClientFor(GrpcNodeA).Subs.Count + factory.ClientFor(GrpcNodeB).Subs.Count; + + var firstSub = factory.ClientFor(GrpcNodeA).Subs.Single(); + firstSub.OnError(new Exception("real fault")); // gen 1 error → flip, reconnect + AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 1, TimeSpan.FromSeconds(5)); // gen 2 stream open + AwaitAssert(() => Assert.Equal(2, TotalSubs())); + + firstSub.OnError(new Exception("late zombie fault")); // stale gen-1 error races in + // Pre-fix this burns retry budget and opens a THIRD stream; post-fix it is ignored. + Thread.Sleep(400); + Assert.Equal(2, TotalSubs()); + } + + // ── R2 T10: live-delta publish coalescing (N6) ── + + [Fact] + public void DeltaBurst_IsCoalesced_IntoFewPublishes_ContainingEveryRow() + { + var (_, seed, sink, factory) = CreateActor(publishCoalesce: TimeSpan.FromMilliseconds(150)); + AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); + seed.CompleteNext(); // empty initial seed + AwaitAssert(() => Assert.Equal(1, sink.Count)); // seed publish, immediate + + AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3)); + var sub = factory.ClientFor(GrpcNodeA).Subs.Single(); + var now = DateTimeOffset.UtcNow; + for (var i = 0; i < 50; i++) + sub.OnAlarm(Alarm($"A{i}", "", 900, now)); // 50 distinct keys, one burst + + AwaitAssert(() => + { + Assert.Equal(50, sink.Latest!.Count); // nothing lost + Assert.InRange(sink.Count, 2, 4); // pre-fix: 51 publishes (1 seed + 1 per delta) + }, TimeSpan.FromSeconds(3)); + } + + [Fact] + public void ZeroCoalesce_PreservesLegacyPerDeltaPublish() + { + var (_, seed, sink, factory) = CreateActor(publishCoalesce: TimeSpan.Zero); + AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); + seed.CompleteNext(); + AwaitAssert(() => Assert.Equal(1, sink.Count)); + + AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3)); + var sub = factory.ClientFor(GrpcNodeA).Subs.Single(); + var now = DateTimeOffset.UtcNow; + sub.OnAlarm(Alarm("A1", "", 900, now)); + sub.OnAlarm(Alarm("A2", "", 900, now)); + AwaitAssert(() => Assert.Equal(3, sink.Count)); // 1 seed + 1 per delta + } + [Fact] public void Stop_Message_TearsDown_Grpc_And_Stops_Actor() { diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheServiceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheServiceTests.cs index 3d7e07ad..eec76ba1 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheServiceTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheServiceTests.cs @@ -258,4 +258,48 @@ public class SiteAlarmLiveCacheServiceTests : TestKit Assert.False(service.IsLive(999)); Assert.Empty(service.GetCurrentAlarms(999)); } + + // ── R2 T15: single-endpoint sites can go live (N9) ── + + [Fact] + public void SiteWithOnlyNodeAEndpoint_StartsAnAggregator_AgainstThatEndpoint() + { + var site = new Site("Three", "site-3") + { + Id = 3, + GrpcNodeAAddress = "http://only-node:8083", + GrpcNodeBAddress = null, // single-endpoint site — pre-fix: never goes live + }; + + var siteRepo = Substitute.For(); + siteRepo.GetSiteByIdAsync(3, Arg.Any()).Returns(site); + + var instanceRepo = Substitute.For(); + instanceRepo.GetInstancesBySiteIdAsync(3, Arg.Any()) + .Returns(new List()); + + var services = new ServiceCollection(); + services.AddScoped(_ => siteRepo); + services.AddScoped(_ => instanceRepo); + var provider = services.BuildServiceProvider(); + + var options = Options.Create(new CommunicationOptions + { + LiveAlarmCacheLinger = TimeSpan.FromSeconds(30), + LiveAlarmCacheReconcileInterval = TimeSpan.FromMinutes(10), + }); + var comm = new CommunicationService(Options.Create(new CommunicationOptions()), + NullLogger.Instance); + var factory = new CountingFactory(); + + var service = new SiteAlarmLiveCacheService( + provider, comm, factory, options, NullLogger.Instance); + service.SetActorSystem(Sys); + + using var sub = service.Subscribe(3, () => { }); + + // Pre-fix: ResolveSiteAsync required BOTH endpoints and bailed → never live, no client. + AwaitCondition(() => service.IsLive(3), TimeSpan.FromSeconds(5)); + Assert.True(factory.GetOrCreateCount >= 1); // a client was created for the single endpoint + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/SfBufferResyncPredicateTests.cs b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/SfBufferResyncPredicateTests.cs new file mode 100644 index 00000000..0dac6c10 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/SfBufferResyncPredicateTests.cs @@ -0,0 +1,94 @@ +using Akka.Actor; +using Microsoft.Extensions.Logging.Abstractions; +using ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors; +using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence; +using ZB.MOM.WW.ScadaBridge.StoreAndForward; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; + +namespace ZB.MOM.WW.ScadaBridge.IntegrationTests.Cluster; + +/// +/// N1 regression (review 02 round 2, Critical): the resync authority must use the same +/// oldest-Up predicate as the S&F delivery gate. Divergence scenario = the delivering node +/// is OLDEST but not LEADER (leader = lowest address), the exact state a rolling restart of +/// the lower-address node produces. Pre-fix the delivering node requests a resync from the +/// stale peer and ReplaceAllAsync wipes its live buffer. +/// +public class SfBufferResyncPredicateTests +{ + [Fact] + public async Task OldestButNotLeaderNode_KeepsItsBuffer_AndSeedsTheJoiner() + { + // Two explicit ports, deliberately assigned so the FIRST-started (oldest, + // delivering) node has the HIGHER address → the second node is cluster leader. + var p1 = TwoNodeClusterFixture.GetFreeTcpPort(); + var p2 = TwoNodeClusterFixture.GetFreeTcpPort(); + var (portHigh, portLow) = p1 > p2 ? (p1, p2) : (p2, p1); + + await using var fixture = await TwoNodeClusterFixture.StartAsync( + role: "site-int", portA: portHigh, portB: portLow); + + // Real S&F storage + replication actor per node, production default predicate + // (no isActiveOverride) — the exact wiring under test. + var (storageOldest, _) = await CreateReplicationActorAsync(fixture.NodeA, "oldest"); + var (storageJoiner, _) = await CreateReplicationActorAsync(fixture.NodeB, "joiner"); + + // The delivering (oldest) node has a live buffered row the standby never saw. + await storageOldest.EnqueueAsync(NewMessage("live-row")); + + // Trigger peer (re)tracking on both sides: each actor got InitialStateAsSnapshot + // in PreStart, but the enqueue raced it — re-deliver via a fresh MemberUp is not + // needed; OnPeerTracked already fired on join. The resync exchange is async: + // wait until the JOINER holds the row (proves the snapshot flowed oldest→joiner, + // the correct direction). Pre-fix this times out (the joiner, as leader, never + // requests) AND the oldest node's row is deleted by the stale wipe. + await AwaitAsync(async () => await storageJoiner.GetMessageByIdAsync("live-row") != null, + TimeSpan.FromSeconds(20), + "joiner never received the resync snapshot (resync ran in the wrong direction)"); + + // And the delivering node's buffer is untouched — the N1 wipe assertion. + Assert.NotNull(await storageOldest.GetMessageByIdAsync("live-row")); + } + + private static async Task<(StoreAndForwardStorage Storage, IActorRef Actor)> CreateReplicationActorAsync( + ActorSystem node, string tag) + { + var sfDb = Path.Combine(Path.GetTempPath(), $"sf-resync-{tag}-{Guid.NewGuid():N}.db"); + var siteDb = Path.Combine(Path.GetTempPath(), $"site-resync-{tag}-{Guid.NewGuid():N}.db"); + var sfStorage = new StoreAndForwardStorage($"Data Source={sfDb}", + NullLogger.Instance); + await sfStorage.InitializeAsync(); + var siteStorage = new SiteStorageService($"Data Source={siteDb}", + NullLogger.Instance); + var replicationService = new ReplicationService( + new StoreAndForwardOptions(), NullLogger.Instance); + // Name MUST be "site-replication" — SendToPeer targets /user/site-replication. + var actor = node.ActorOf(Props.Create(() => new SiteReplicationActor( + siteStorage, sfStorage, replicationService, "site-int", + NullLogger.Instance, null, null, null, null)), + "site-replication"); + return (sfStorage, actor); + } + + private static StoreAndForwardMessage NewMessage(string id) => new() + { + Id = id, + Category = StoreAndForwardCategory.Notification, + Target = "central", + PayloadJson = "{}", + CreatedAt = DateTimeOffset.UtcNow, + Status = StoreAndForwardMessageStatus.Pending, + MaxRetries = 0, + }; + + private static async Task AwaitAsync(Func> condition, TimeSpan timeout, string why) + { + var deadline = DateTime.UtcNow + timeout; + while (DateTime.UtcNow < deadline) + { + if (await condition()) return; + await Task.Delay(250); + } + throw new TimeoutException(why); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/TwoNodeClusterFixture.cs b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/TwoNodeClusterFixture.cs index ede98c32..3d1c94da 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/TwoNodeClusterFixture.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Cluster/TwoNodeClusterFixture.cs @@ -22,11 +22,12 @@ public sealed class TwoNodeClusterFixture : IAsyncDisposable public int PortB { get; private set; } public static async Task StartAsync( - string role = "Central", TimeSpan? stableAfter = null) + string role = "Central", TimeSpan? stableAfter = null, + int? portA = null, int? portB = null) { var f = new TwoNodeClusterFixture(); - f.PortA = GetFreeTcpPort(); - f.PortB = GetFreeTcpPort(); + f.PortA = portA ?? GetFreeTcpPort(); + f.PortB = portB ?? GetFreeTcpPort(); f.NodeA = f.StartNode(f.PortA, role, stableAfter); await WaitForMembersUp(f.NodeA, 1, TimeSpan.FromSeconds(20)); f.NodeB = f.StartNode(f.PortB, role, stableAfter); @@ -98,7 +99,7 @@ public sealed class TwoNodeClusterFixture : IAsyncDisposable throw new TimeoutException($"Member {removed} was never removed from {cluster.SelfAddress}'s view — SBR did not down it."); } - private static int GetFreeTcpPort() + public static int GetFreeTcpPort() { var listener = new System.Net.Sockets.TcpListener(System.Net.IPAddress.Loopback, 0); listener.Start(); diff --git a/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Grpc/SiteAlarmStreamEndToEndTests.cs b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Grpc/SiteAlarmStreamEndToEndTests.cs index 9694b975..4d51bba3 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Grpc/SiteAlarmStreamEndToEndTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Grpc/SiteAlarmStreamEndToEndTests.cs @@ -223,7 +223,8 @@ public class SiteAlarmStreamEndToEndTests : TestKit var sink = new PublishSink(); var aggregator = Sys.ActorOf(Props.Create(() => new SiteAlarmAggregatorActor( "site-alpha", "e2e-parity", _ => Task.FromResult>(new[] { seedRow }), - sink.Publish, factory, "http://a:5100", "http://b:5100", TimeSpan.FromMinutes(10)))); + sink.Publish, factory, "http://a:5100", "http://b:5100", TimeSpan.FromMinutes(10), + TimeSpan.Zero, TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(60)))); await WaitForConditionAsync(() => sink.Latest is { Count: 1 }); // seed applied aggregator.Tell(liveRow); diff --git a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/ManagementActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/ManagementActorTests.cs index afde74b3..b52d42e5 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/ManagementActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/ManagementActorTests.cs @@ -5,8 +5,10 @@ using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; using NSubstitute; using ZB.MOM.WW.ScadaBridge.Communication; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; using NSubstitute.ExceptionExtensions; using ZB.MOM.WW.ScadaBridge.Commons.Entities.Instances; +using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites; using ZB.MOM.WW.ScadaBridge.Commons.Entities.Schemas; using ZB.MOM.WW.ScadaBridge.Commons.Entities.SecuredWrites; using ZB.MOM.WW.ScadaBridge.Commons.Entities.Scripts; @@ -3599,6 +3601,51 @@ public class ManagementActorTests : TestKit, IDisposable ExpectMsg(TimeSpan.FromSeconds(5)); } + // ── R2 T13: site delete disposes the site's cached gRPC channels (N8) ── + + [Fact] + public void DeleteSite_DisposesTheSitesCachedGrpcChannels() + { + var siteRepo = Substitute.For(); + siteRepo.GetSiteByIdAsync(7, Arg.Any()) + .Returns(new Site("Seven", "site-7") { Id = 7 }); + siteRepo.GetInstancesBySiteIdAsync(7, Arg.Any()) + .Returns((IReadOnlyList)new List()); + _services.AddScoped(_ => siteRepo); + + var factory = new TrackingGrpcFactory(); + var cached = (TrackingClient)factory.GetOrCreate("site-7", "http://node-a:8083"); + _services.AddSingleton(factory); + + var actor = CreateActor(); + actor.Tell(Envelope(new DeleteSiteCommand(7), "Administrator")); + ExpectMsg(TimeSpan.FromSeconds(5)); + + AwaitAssert(() => + { + Assert.True(cached.Disposed); + Assert.Null(factory.TryGet("site-7", "http://node-a:8083")); + }, TimeSpan.FromSeconds(5)); + } + + private sealed class TrackingClient : SiteStreamGrpcClient + { + public TrackingClient(string endpoint) : base(endpoint) { } + public bool Disposed { get; private set; } + public override void Dispose() => Disposed = true; + public override ValueTask DisposeAsync() + { + Disposed = true; + return ValueTask.CompletedTask; + } + } + + private sealed class TrackingGrpcFactory : SiteStreamGrpcClientFactory + { + public TrackingGrpcFactory() : base(NullLoggerFactory.Instance) { } + protected override SiteStreamGrpcClient CreateClient(string grpcEndpoint) => new TrackingClient(grpcEndpoint); + } + /// /// Records remote-query relays for the actor-dispatch tests. The browse / /// search / verify / cert-trust methods on diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs index 7dccdddd..eb67ee4f 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs @@ -1,4 +1,5 @@ using System.Collections.Concurrent; +using System.Diagnostics.Metrics; using Akka.Actor; using Akka.TestKit.Xunit2; using Microsoft.Extensions.Logging; @@ -9,6 +10,7 @@ using ZB.MOM.WW.ScadaBridge.SiteRuntime.Deployment; using ZB.MOM.WW.ScadaBridge.SiteRuntime.Messages; using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence; using ZB.MOM.WW.ScadaBridge.StoreAndForward; +using ZB.MOM.WW.ScadaBridge.Commons.Observability; using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Actors; @@ -248,8 +250,10 @@ akka { } [Fact] - public async Task ActiveNode_AnswersResyncRequest_WithFullBufferSnapshot() + public async Task ActiveNode_AnswersResyncRequest_WithChunkedSnapshot() { + // Post-R2-T5 the active node answers with byte-budgeted SfBufferSnapshotChunk(s) + // (a single small row rides one chunk) rather than the monolithic SfBufferSnapshot. await _sfStorage.EnqueueAsync(NewSfMessage("m1")); var probe = CreateTestProbe(); var actor = ActorOf(Props.Create(() => new ResyncTestActor( @@ -258,9 +262,11 @@ akka { actor.Tell(new RequestSfBufferResync(), TestActor); - var snapshot = ExpectMsg(TimeSpan.FromSeconds(3)); - Assert.Single(snapshot.Messages); - Assert.False(snapshot.Truncated); + var chunk = ExpectMsg(TimeSpan.FromSeconds(3)); + Assert.Equal(1, chunk.TotalChunks); + Assert.Equal(1, chunk.Sequence); + Assert.Single(chunk.Messages); + Assert.False(chunk.Truncated); } [Fact] @@ -281,6 +287,170 @@ akka { }, TimeSpan.FromSeconds(5)); } + // ── R2 T5: chunked resync answer ── + + [Fact] + public void ChunkForRemoting_SplitsByByteBudget_PreservingOrderAndSequence() + { + var rows = Enumerable.Range(0, 10) + .Select(i => NewMessage($"m{i}", payloadJson: new string('x', 20_000))) + .ToList(); + + var chunks = SiteReplicationActor.ChunkForRemoting(rows, maxChunkBytes: 64_000, maxChunkRows: 200); + + Assert.True(chunks.Count > 1); // 10 × 20 KB cannot ride one 64 KB chunk + Assert.Equal(rows.Select(r => r.Id), chunks.SelectMany(c => c).Select(r => r.Id)); // order preserved + Assert.All(chunks, c => Assert.True( + c.Sum(r => r.PayloadJson.Length) <= 64_000 || c.Count == 1)); // budget honored (oversized row isolated) + } + + [Fact] + public void ChunkForRemoting_RowCapHonored_AndSingleOversizedRowIsolated() + { + var many = Enumerable.Range(0, 500).Select(i => NewMessage($"s{i}", payloadJson: "{}")).ToList(); + Assert.All(SiteReplicationActor.ChunkForRemoting(many, 64_000, 200), c => Assert.True(c.Count <= 200)); + + var oversized = new List + { NewMessage("big", payloadJson: new string('y', 100_000)), NewMessage("small", payloadJson: "{}") }; + var chunks = SiteReplicationActor.ChunkForRemoting(oversized, 64_000, 200); + Assert.Equal(2, chunks.Count); // the oversized row rides alone + } + + [Fact] + public async Task ActiveNode_AnswersResyncRequest_WithSequencedChunks_SharingOneResyncId() + { + for (var i = 0; i < 3; i++) + await _sfStorage.EnqueueAsync(NewMessage($"c{i}", payloadJson: new string('z', 30_000))); + var actor = CreateResyncActor(isActive: () => true); + + actor.Tell(new RequestSfBufferResync(), TestActor); + + var first = ExpectMsg(TimeSpan.FromSeconds(5)); + var rest = Enumerable.Range(1, first.TotalChunks - 1) + .Select(_ => ExpectMsg(TimeSpan.FromSeconds(5))) + .Prepend(first) + .ToList(); + + Assert.True(first.TotalChunks > 1); + Assert.All(rest, c => Assert.Equal(first.ResyncId, c.ResyncId)); + Assert.Equal(Enumerable.Range(1, first.TotalChunks), rest.Select(c => c.Sequence)); + Assert.Equal(3, rest.Sum(c => c.Messages.Count)); + } + + // ── R2 T6: standby chunk assembly + atomic apply + ack ── + + [Fact] + public async Task StandbyNode_AssemblesChunks_AppliesOnce_AndAcks() + { + await _sfStorage.EnqueueAsync(NewMessage("stale")); + var actor = CreateResyncActor(isActive: () => false); + var resyncId = "r1"; + + actor.Tell(new SfBufferSnapshotChunk(resyncId, 1, 2, + new List { NewMessage("f1") }, false), TestActor); + actor.Tell(new SfBufferSnapshotChunk(resyncId, 2, 2, + new List { NewMessage("f2") }, false), TestActor); + + var ack = ExpectMsg(TimeSpan.FromSeconds(5)); + Assert.Equal(resyncId, ack.ResyncId); + Assert.Equal(2, ack.RowCount); + await AwaitAssertAsync(async () => + { + Assert.Null(await _sfStorage.GetMessageByIdAsync("stale")); // replaced wholesale + Assert.NotNull(await _sfStorage.GetMessageByIdAsync("f1")); + Assert.NotNull(await _sfStorage.GetMessageByIdAsync("f2")); + }); + } + + [Fact] + public async Task StandbyNode_NewResyncId_DiscardsStalePartialAssembly() + { + var actor = CreateResyncActor(isActive: () => false); + actor.Tell(new SfBufferSnapshotChunk("old", 1, 2, + new List { NewMessage("orphan") }, false), TestActor); + actor.Tell(new SfBufferSnapshotChunk("new", 1, 1, + new List { NewMessage("fresh") }, false), TestActor); + + ExpectMsg(TimeSpan.FromSeconds(5)); // "new" completed + await AwaitAssertAsync(async () => + { + Assert.NotNull(await _sfStorage.GetMessageByIdAsync("fresh")); + Assert.Null(await _sfStorage.GetMessageByIdAsync("orphan")); // stale partial never applied + }); + } + + [Fact] + public void ActiveNode_IgnoresChunks_NeverAcks() + { + var actor = CreateResyncActor(isActive: () => true); + actor.Tell(new SfBufferSnapshotChunk("r", 1, 1, + new List { NewMessage("x") }, false), TestActor); + ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + } + + // ── R2 T7: active-side resync ack confirmation + telemetry ── + // + // NOTE (deviation from plan): the actor logs via Microsoft ILogger (NullLogger in + // tests), NOT Akka's EventStream, so the plan's EventFilter.Warning assertions can + // never observe these warnings. We observe the two OTel counters via a MeterListener + // instead — the equivalent, and stronger, observable signal. + + [Fact] + public async Task ActiveNode_ReceivingAck_CountsResyncCompleted() + { + long completed = 0; + using var listener = ListenCounter("scadabridge.store_and_forward.resync.completed", + m => Interlocked.Add(ref completed, m)); + + await _sfStorage.EnqueueAsync(NewMessage("m1")); + var actor = CreateResyncActor(isActive: () => true, ackTimeout: TimeSpan.FromSeconds(30)); + actor.Tell(new RequestSfBufferResync(), TestActor); + var chunk = ExpectMsg(TimeSpan.FromSeconds(5)); + + actor.Tell(new SfBufferResyncAck(chunk.ResyncId, 1), TestActor); + + await AwaitAssertAsync(() => + { + Assert.True(Interlocked.Read(ref completed) >= 1); // ack recorded the completion + return Task.CompletedTask; + }, TimeSpan.FromSeconds(5)); + } + + [Fact] + public async Task ActiveNode_MissingAck_WarnsAfterAckTimeout() + { + long ackMissing = 0; + using var listener = ListenCounter("scadabridge.store_and_forward.resync.ack_missing", + m => Interlocked.Add(ref ackMissing, m)); + + await _sfStorage.EnqueueAsync(NewMessage("m1")); + var actor = CreateResyncActor(isActive: () => true, ackTimeout: TimeSpan.FromMilliseconds(200)); + actor.Tell(new RequestSfBufferResync(), TestActor); + ExpectMsg(TimeSpan.FromSeconds(5)); + // No ack is sent → the ack window expires and the resync is counted unacknowledged. + + await AwaitAssertAsync(() => + { + Assert.True(Interlocked.Read(ref ackMissing) >= 1); + return Task.CompletedTask; + }, TimeSpan.FromSeconds(5)); + } + + /// Attaches a to a single ScadaBridge counter by name, + /// forwarding each recorded increment to . + private static MeterListener ListenCounter(string instrumentName, Action onMeasurement) + { + var listener = new MeterListener(); + listener.InstrumentPublished = (inst, l) => + { + if (inst.Meter.Name == ScadaBridgeTelemetry.MeterName && inst.Name == instrumentName) + l.EnableMeasurementEvents(inst); + }; + listener.SetMeasurementEventCallback((_, m, _, _) => onMeasurement(m)); + listener.Start(); + return listener; + } + private static StoreAndForwardMessage NewSfMessage(string id) => new() { Id = id, @@ -294,6 +464,31 @@ akka { Status = StoreAndForwardMessageStatus.Pending, }; + /// + /// Builds a resync-test message with a settable payload (additive to + /// — the chunker sizes on PayloadJson length). + /// + private static StoreAndForwardMessage NewMessage(string id, string payloadJson = "{}") => new() + { + Id = id, + Category = StoreAndForwardCategory.ExternalSystem, + Target = "t", + PayloadJson = payloadJson, + RetryCount = 0, + MaxRetries = 50, + RetryIntervalMs = 30000, + CreatedAt = DateTimeOffset.UtcNow, + Status = StoreAndForwardMessageStatus.Pending, + }; + + /// Constructs a with the given active-node check + /// (the resync chunk/ack tests Tell to and expect from ). + /// is the active-side ack window seam (T7). + private IActorRef CreateResyncActor(Func isActive, TimeSpan? ackTimeout = null) => + ActorOf(Props.Create(() => new ResyncTestActor( + _storage, _sfStorage, _replicationService, SiteRole, + NullLogger.Instance, CreateTestProbe().Ref, isActive, ackTimeout))); + /// Test message: drives directly, /// standing in for the MemberUp→TryTrackPeer path (a single-node TestKit cannot form a real peer). private sealed record TriggerPeerTracked; @@ -309,9 +504,10 @@ akka { public ResyncTestActor( SiteStorageService storage, StoreAndForwardStorage sfStorage, ReplicationService replicationService, string siteRole, - ILogger logger, IActorRef peerProbe, Func isActive) + ILogger logger, IActorRef peerProbe, Func isActive, + TimeSpan? ackTimeout = null) : base(storage, sfStorage, replicationService, siteRole, logger, - configFetcher: null, isActiveOverride: isActive) + configFetcher: null, isActiveOverride: isActive, resyncAckTimeout: ackTimeout) { _peerProbe = peerProbe; Receive(_ => OnPeerTracked()); diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/ResyncWireSerializationPinTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/ResyncWireSerializationPinTests.cs new file mode 100644 index 00000000..43560a14 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/ResyncWireSerializationPinTests.cs @@ -0,0 +1,95 @@ +using Akka.TestKit.Xunit2; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; +using ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors; +using ZB.MOM.WW.ScadaBridge.StoreAndForward; + +namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests; + +/// +/// Characterization pin for the chunked anti-entropy resync contract (review 02 round 2, +/// N2). (active→standby) and +/// (standby→active) ride intra-site Akka remoting on the default reflective-JSON wire +/// format. A rename/move, dropped setter, or non-default-constructible message would +/// silently break resync across a rolling upgrade and only surface as a divergent buffer +/// after a failover. These pin round-trip fidelity and type identity. (These messages are +/// NOT ClusterClient traffic, so they are intentionally absent from ClusterClientContractLockTests.) +/// +public class ResyncWireSerializationPinTests : TestKit +{ + private T RoundTrip(T message) + { + var serialization = Sys.Serialization; + var serializer = serialization.FindSerializerFor(message); + var bytes = serializer.ToBinary(message); + return (T)serialization.Deserialize(bytes, serializer.Identifier, message!.GetType()); + } + + private static StoreAndForwardMessage FullMessage() => new() + { + Id = Guid.NewGuid().ToString("N"), + Category = StoreAndForwardCategory.Notification, + Target = "Operators", + PayloadJson = "{\"notificationId\":\"abc\"}", + RetryCount = 4, + MaxRetries = 0, + RetryIntervalMs = 30000, + CreatedAt = DateTimeOffset.UtcNow, + LastAttemptAt = DateTimeOffset.UtcNow, + Status = StoreAndForwardMessageStatus.Parked, + LastError = "central rejected", + OriginInstanceName = "Plant.Pump3", + ExecutionId = Guid.NewGuid(), + SourceScript = "ScriptActor:MonitorSpeed", + ParentExecutionId = Guid.NewGuid(), + }; + + [Fact] + public void SfBufferSnapshotChunk_WithFullMessage_RoundTripsOnTheWire() + { + var message = FullMessage(); + var original = new SfBufferSnapshotChunk( + "resync-1", 2, 5, new List { message }, Truncated: true); + + var back = RoundTrip(original); + + Assert.Equal(original.ResyncId, back.ResyncId); + Assert.Equal(original.Sequence, back.Sequence); + Assert.Equal(original.TotalChunks, back.TotalChunks); + Assert.Equal(original.Truncated, back.Truncated); + var m = Assert.Single(back.Messages); + Assert.Equal(message.Id, m.Id); + Assert.Equal(message.Category, m.Category); + Assert.Equal(message.Target, m.Target); + Assert.Equal(message.PayloadJson, m.PayloadJson); + Assert.Equal(message.RetryCount, m.RetryCount); + Assert.Equal(message.MaxRetries, m.MaxRetries); + Assert.Equal(message.RetryIntervalMs, m.RetryIntervalMs); + Assert.Equal(message.CreatedAt, m.CreatedAt); + Assert.Equal(message.LastAttemptAt, m.LastAttemptAt); + Assert.Equal(message.Status, m.Status); + Assert.Equal(message.LastError, m.LastError); + Assert.Equal(message.OriginInstanceName, m.OriginInstanceName); + Assert.Equal(message.ExecutionId, m.ExecutionId); + Assert.Equal(message.SourceScript, m.SourceScript); + Assert.Equal(message.ParentExecutionId, m.ParentExecutionId); + } + + [Fact] + public void SfBufferResyncAck_RoundTripsOnTheWire() + { + var original = new SfBufferResyncAck("resync-1", 42); + + var back = RoundTrip(original); + + Assert.Equal(original.ResyncId, back.ResyncId); + Assert.Equal(original.RowCount, back.RowCount); + } + + // Type-identity pins: the reflective-JSON wire embeds CLR type manifests, so a + // rename/move of either type silently breaks resync across a rolling upgrade. + [Theory] + [InlineData(typeof(SfBufferSnapshotChunk), "ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors.SfBufferSnapshotChunk")] + [InlineData(typeof(SfBufferResyncAck), "ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors.SfBufferResyncAck")] + public void ResyncContract_TypeIdentity_IsPinned(Type type, string expectedFullName) => + Assert.Equal(expectedFullName, type.FullName); +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardOptionsValidatorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardOptionsValidatorTests.cs index 5ffb3063..346516fa 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardOptionsValidatorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardOptionsValidatorTests.cs @@ -47,4 +47,33 @@ public class StoreAndForwardOptionsValidatorTests Assert.True(result.Failed); Assert.Contains("SqliteDbPath", result.FailureMessage); } + + // ── R2 T9: sweep-tuning eager validation (N4) ── + + [Theory] + [InlineData(-1)] + [InlineData(-500)] + public void Validate_NegativeSweepBatchLimit_Fails(int limit) + { + var result = Validate(new StoreAndForwardOptions { SweepBatchLimit = limit }); + Assert.True(result.Failed); + Assert.Contains("SweepBatchLimit", result.FailureMessage); + } + + [Fact] + public void Validate_ZeroSweepBatchLimit_IsValidUnlimitedLegacy() + { + var result = Validate(new StoreAndForwardOptions { SweepBatchLimit = 0 }); + Assert.True(result.Succeeded, result.FailureMessage); + } + + [Theory] + [InlineData(0)] + [InlineData(-4)] + public void Validate_NonPositiveSweepTargetParallelism_Fails(int parallelism) + { + var result = Validate(new StoreAndForwardOptions { SweepTargetParallelism = parallelism }); + Assert.True(result.Failed); + Assert.Contains("SweepTargetParallelism", result.FailureMessage); + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs index 4facffa7..3c7a4c94 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/StoreAndForwardServiceTests.cs @@ -821,4 +821,67 @@ public class StoreAndForwardServiceTests : IAsyncLifetime, IDisposable } finally { await service.StopAsync(); } } + + // ── R2 T8: _sweepTask clobber (N3) ── + + [Fact] + public async Task TriggerSweep_WhileSweepInFlight_DoesNotClobberTheDrainHandle() + { + var service = CreateService(retryTimerInterval: TimeSpan.FromHours(1)); + var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem, async _ => + { + entered.TrySetResult(); + await release.Task; + return true; + }); + await service.StartAsync(); + try + { + await service.EnqueueAsync(StoreAndForwardCategory.ExternalSystem, "t", "{}", + attemptImmediateDelivery: false, retryInterval: TimeSpan.Zero); + + service.TriggerSweep(); // real sweep, blocked in the handler + await entered.Task.WaitAsync(TimeSpan.FromSeconds(5)); + service.TriggerSweep(); // redundant kick — pre-fix clobbers _sweepTask + + var handle = service.CurrentSweepTaskForTest; + Assert.NotNull(handle); + Assert.False(handle!.IsCompleted); // pre-fix: true (a completed no-op replaced the real sweep) + } + finally + { + release.TrySetResult(); + await service.StopAsync(); + } + } + + [Fact] + public async Task StopAsync_WaitsForTheRealInFlightSweep_EvenAfterARedundantTrigger() + { + var service = CreateService(retryTimerInterval: TimeSpan.FromHours(1)); + var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem, async _ => + { + entered.TrySetResult(); + await release.Task; + return true; + }); + await service.StartAsync(); + await service.EnqueueAsync(StoreAndForwardCategory.ExternalSystem, "t", "{}", + attemptImmediateDelivery: false, retryInterval: TimeSpan.Zero); + + service.TriggerSweep(); + await entered.Task.WaitAsync(TimeSpan.FromSeconds(5)); + service.TriggerSweep(); // the clobbering kick + + var stop = service.StopAsync(); + await Task.Delay(300); + Assert.False(stop.IsCompleted); // pre-fix: StopAsync already returned (awaited the no-op) + + release.TrySetResult(); + await stop.WaitAsync(TimeSpan.FromSeconds(5)); // drains the real sweep promptly once released + } }