From fd5e023d085a1030a242f4ad4fa8999bbda58fa1 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 14 Aug 2026 23:52:25 -0400 Subject: [PATCH] =?UTF-8?q?fix(comms):=20review=20findings=20=E2=80=94=20c?= =?UTF-8?q?onsumer-based=20debug=20orphan=20net,=20foreign-cancel=20triad,?= =?UTF-8?q?=20honest=20onConnected,=20served-row-exact=20retirement,=20ful?= =?UTF-8?q?l-rate=20reconcile?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit F1 (HIGH) DebugStreamBridgeActor: the 5-minute orphan net measured the MAILBOX (SetReceiveTimeout), and once stream events were correctly marked INotInfluenceReceiveTimeout nothing recurring reset it — the snapshot lands once and GrpcStreamStable once — so every healthy session self-terminated at ~6 min with a false "Site disconnected". Replaced with a periodic self-tick (ConsumerLivenessCheckInterval, 30s) over a consumer-last-seen stamp renewed only by DebugStreamConsumerAlive, which DebugStreamService Tells on a shared timer to every session still in its registry (holding a session there IS "a consumer is attached" — both the Blazor view and the SignalR hub release it on dispose/disconnect, and it works headless). Reverting the wrapper was rejected: it would restore the quiet-instance orphan bug. F2 (MED) SiteStreamGrpcClient: the RpcException(Cancelled) filter now requires cts.IsCancellationRequested. A peer-originated / channel-dispose Cancelled fired none of onError/onCompleted/onConnected, leaving SiteAlarmAggregatorActor with _streamDown=false forever (IsLive stuck true, reconcile reopen guard never fired). F3 (MED) SiteStreamGrpcClient: a header TIMEOUT is no longer reported as connected — that shape is exactly what an unreachable site produces, and it cleared _streamDown, consumed _seedOnConnect and launched a full snapshot fan-out at a dead site. AwaitHeadersAsync returns bool; the first received event is the fallback connected signal, fired at most once from headers OR first event. F4 (LOW-MED) SqliteAuditWriter.MarkReconciledUpToAsync: the blanket below-cursor UPDATE retired late-stamped inserts that were never served (then age-purged — silent loss). The flip is now bounded by insertion order: a Pending row retires only if its rowid is at or below the high-water mark of rows this instance has served from ReadPendingSinceAsync (clamped on purge, since SQLite reuses rowids); Forwarded rows are exempt (central ACKed them over the push path). At-least-once is unchanged. F5 (LOW) Documented the liveness dependency (a served row never covered by a later cursor stays Pending forever; PurgeExpiredAsync never purges Pending) in ISiteAuditQueue + Component-AuditLog.md, and added a cheap site-health signal: SiteAuditBacklogReporter logs a rate-limited warning when the existing oldest-pending metric exceeds 24h. F6 (MED) SiteAlarmAggregatorActor: _fanoutSinceLastTick was armed by the reconcile's OWN fan-out, so steady state ran fan-out→skip→fan-out→skip — one reconcile per 2x interval (120s), halving the not-reporting refresh and the alarm reconcile backstop. The skip is now armed only by connect/failover-driven seeds (initial, _seedOnConnect, and a re-seed queued behind one). Tests: Communication.Tests 691 passed (+13), AuditLog.Tests 382 passed (+5). --- docs/requirements/Component-AuditLog.md | 42 ++++- docs/requirements/Component-Communication.md | 6 +- .../Site/SiteAuditBacklogReporter.cs | 44 +++++ .../Site/SqliteAuditWriter.cs | 86 ++++++++- .../Interfaces/Services/ISiteAuditQueue.cs | 23 +++ .../Actors/DebugStreamBridgeActor.cs | 119 ++++++++++--- .../Actors/SiteAlarmAggregatorActor.cs | 91 +++++++--- .../DebugStreamService.cs | 90 +++++++++- .../Grpc/SiteStreamGrpcClient.cs | 81 ++++++--- .../SiteAuditBacklogReporterCadenceTests.cs | 64 +++++++ .../Site/SqliteAuditWriterWriteTests.cs | 97 +++++++++++ .../DebugStreamServiceTests.cs | 90 ++++++++++ .../Grpc/DebugStreamBridgeActorTests.cs | 148 +++++++++++++++- .../Grpc/SiteAlarmAggregatorActorTests.cs | 93 ++++++++++ .../Grpc/SiteStreamGrpcClientTests.cs | 163 +++++++++++++++++- .../SiteStreamPullAuditEventsTests.cs | 39 +++++ 16 files changed, 1192 insertions(+), 84 deletions(-) diff --git a/docs/requirements/Component-AuditLog.md b/docs/requirements/Component-AuditLog.md index b9364d22..7beb92ac 100644 --- a/docs/requirements/Component-AuditLog.md +++ b/docs/requirements/Component-AuditLog.md @@ -261,7 +261,15 @@ room is a compliance violation, not a self-healing behavior. To bound that growth in practice, the site emits a `SiteAuditBacklog` health metric (pending row count, oldest pending age, bytes on disk); crossing operator-configured thresholds surfaces a warning on the relevant site tile in the Health -dashboard, mirroring the Store-and-Forward Engine's backlog metric. +dashboard, mirroring the Store-and-Forward Engine's backlog metric, and +`SiteAuditBacklogReporter` additionally logs a rate-limited warning once the +oldest pending row passes 24 h. + +The same invariant carries a liveness dependency worth stating explicitly: the +`Pending` floor clears only when central acknowledges the rows — a telemetry ack +or a reconciliation cursor that covers them — never on age. See *Reconciliation +pull* below for the served-row retirement rule that decides which rows can be +acknowledged by a cursor at all. Central is the durable home. Site SQLite is a write-buffer with a forwarding guarantee. @@ -367,6 +375,38 @@ it — a lagging drain is meant to surface as the stalled signal. The id tiebrea what makes that safe against a same-instant burst larger than one batch: the cursor advances on every tick even when the timestamp cannot. +**Only rows that were actually served may retire.** `OccurredAtUtc` is stamped by the +caller, so a row can be *inserted* after a batch was served yet carry a timestamp +*below* central's (by then advanced) cursor — a back-dated stamp, a clock nudge, a +write flushed late. A cursor flip that keyed on the timestamp alone retired exactly +those rows: never served, never servable again (the keyset read has moved past them), +and, being `Reconciled`, purged on age. That is silent audit loss, and it is a failure +mode the pre-WP2.3 explicit id-set flip could not produce, so `MarkReconciledUpToAsync` +carries a second, insertion-order bound: + +> A `Pending` row retires only if its insertion order is at or below the high-water +> mark of rows this site node has actually served from `ReadPendingSinceAsync` +> (SQLite `rowid`). A `Forwarded` row is exempt — central ACKED it over the telemetry +> push path, which is proof independent of the pull. + +The bound is per-process, so after a site-node restart the first pull retires only +`Forwarded` rows and the pull after it resumes normal retirement — conservative in the +safe direction (retirement is delayed, never a row lost). The gRPC handler's ordering +(retire, *then* read) is what keeps the bound from ever vouching for the batch it is +about to serve. + +**Liveness note (accepted, documented).** A row that was served but never covered by a +later cursor — central reconciliation stopped for good, or the bound reset over a +restart before the next cursor arrived — stays `Pending` indefinitely, and the site +retention purge never purges `Pending`. That is the hard `ForwardState` invariant +working as intended (an unacknowledged row is not droppable), but it means the site +store's floor depends on reconciliation actually running, not merely on the retention +window elapsing. It is observable rather than silent: `GetBacklogStatsAsync` reports +the pending count and oldest-pending instant on every site health report +(`SiteAuditBacklog`), and `SiteAuditBacklogReporter` logs a rate-limited warning once +the oldest pending row exceeds `StalePendingThreshold` (24 h) naming the drain and the +reconciliation pull as the things to check. + ### Central direct-write (central-originated events) Events originating at central never touch site SQLite. Inbound API writes one diff --git a/docs/requirements/Component-Communication.md b/docs/requirements/Component-Communication.md index b232c97e..77147d52 100644 --- a/docs/requirements/Component-Communication.md +++ b/docs/requirements/Component-Communication.md @@ -70,7 +70,8 @@ Both central and site clusters. Each side has communication actors that handle m - **Bridge-session hardening (WP2.3):** - The pre-snapshot buffer is **bounded (20 000 events, drop-oldest)** and its evictions counted (`scadabridge.central.debug_view.presnapshot_dropped`). It was previously unbounded, so a session whose snapshot never arrived grew without limit on the CENTRAL node. Dropping the oldest is correct here: the snapshot that ends the buffering phase is authoritative for anything that old. - A **hard snapshot deadline** (`DebugStreamBridgeActor.SnapshotTimeout`, 60 s) fails the session if no `DebugViewSnapshot` arrives. Nothing else ended a session wedged in the buffering phase — a lost site reply raises no gRPC error. - - **Stream events no longer influence the orphan receive timeout.** The gRPC callback wraps each event in an envelope marked `INotInfluenceReceiveTimeout`, so a busy site can no longer keep an abandoned session alive forever by feeding it events. The 5-minute timeout now measures session/consumer liveness, which is what it was for. + - **The orphan net measures the CONSUMER, not the mailbox.** A session self-terminates after `DebugStreamBridgeActor.ConsumerIdleTimeout` (5 min) without a sign of life from its consumer, checked by a periodic self-tick (`ConsumerLivenessCheckInterval`, 30 s) against a consumer-last-seen stamp. The stamp is renewed only by `DebugStreamConsumerAlive`, which `DebugStreamService` Tells to every session still in its registry on a shared 30 s timer — holding a session there IS what "a consumer is attached" means, since both consumers (the Blazor debug view and the SignalR hub) release it on dispose/disconnect. + This replaced an Akka `SetReceiveTimeout(5 min)`. That version measured the MAILBOX, which conflates site chatter with consumer liveness: a busy site kept an abandoned session alive forever. Marking stream events `INotInfluenceReceiveTimeout` fixed that but left the timeout with nothing recurring to reset it — the snapshot lands once and the stability tick once — so **every healthy session self-terminated ~6 min in and the operator was told "Site disconnected"**. Both failure directions are pinned by tests: streaming events with keepalives survive many windows; an orphaned session terminates while events keep arriving. ### 6.1 Aggregated Live Alarm Stream (Site → Central) @@ -80,7 +81,8 @@ 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 its site-wide stream is currently up**. Liveness rides every publish from the aggregator, so a stream that faults or ends gracefully drops `IsLive` immediately rather than leaving the page grafting a freezing snapshot over fresh poll data until the next reconcile (WP2.3). - **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. - **Alarms-only seed (WP2.3)**: the seed/reconcile fan-out sets `DebugSnapshotRequest.AlarmsOnly` (wire: `DebugSnapshotRequestDto.alarms_only`, field 3, additive), so the site builds and ships only the alarm half of the snapshot. The fan-out discarded every attribute row anyway, and an instance's attribute surface dwarfs its alarm set. A pre-WP2.3 site ignores the flag and returns the full snapshot, which reads identically. -- **Failover & drift**: a re-seed runs **once per successful (re)connect**, not once per reconnect ATTEMPT — the connect signal is `SiteStreamGrpcClient.SubscribeSiteAsync`'s `onConnected` callback, raised when the site's response headers arrive (the site flushes them as soon as its relay is attached, so no event can be missed after it). Fanning a whole-site snapshot out per retry meant N snapshots against a site that was, by definition of the retry, unreachable. A periodic **reconcile snapshot** (default 60s, **jittered** by `LiveAlarmCacheReconcileJitterFraction` so aggregators started together do not stampede one boundary) remains the drift backstop, but it is **skipped when a fan-out already ran in that window** and **publishes only when the snapshot actually changed the cache** (a diff, not an unconditional viewer fan-out). Staleness stays bounded at two intervals: the skip consumes its flag, so the next tick always fans out. A fan-out that fails as a whole now retries on its own **backoff** timer (`reconnectDelay` doubling, capped at 8× the reconcile interval) instead of waiting a full interval. `[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. +- **Failover & drift**: a re-seed runs **once per successful (re)connect**, not once per reconnect ATTEMPT — the connect signal is `SiteStreamGrpcClient.SubscribeSiteAsync`'s `onConnected` callback, raised when the site's response headers arrive (the site flushes them as soon as its relay is attached, so no event can be missed after it) or, for a peer that defers its headers, when the FIRST EVENT arrives — whichever comes first, and at most once. A header **timeout** is deliberately not a connect signal: an unreachable or wedged site produces exactly that shape, and reporting it as connected cleared `_streamDown`, consumed the pending re-seed and fanned a whole-site snapshot out at a site that never answered. Fanning a whole-site snapshot out per retry meant N snapshots against a site that was, by definition of the retry, unreachable. A periodic **reconcile snapshot** (default 60s, **jittered** by `LiveAlarmCacheReconcileJitterFraction` so aggregators started together do not stampede one boundary) remains the drift backstop, but it is **skipped when a CONNECT- OR FAILOVER-DRIVEN seed already ran in that window** and **publishes only when the snapshot actually changed the cache** (a diff, not an unconditional viewer fan-out). Staleness stays bounded at two intervals: the skip consumes its flag, so the next tick always fans out. A **tick's own fan-out never arms that skip** — when it did, steady state alternated fan-out/skip and the effective reconcile rate was one per TWO intervals (120 s by default), halving both the not-reporting refresh and the alarm reconcile backstop for no benefit: with no reconnect in play there is nothing duplicated to suppress. A fan-out that fails as a whole now retries on its own **backoff** timer (`reconnectDelay` doubling, capped at 8× the reconcile interval) instead of waiting a full interval. `[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. +- **Stream terminations are a triad, and every ending hits exactly one leg.** `SiteStreamGrpcClient.ConsumeStreamAsync` classifies a stream's end as a fault (`onError`), a graceful server-side end (`onCompleted`), or our own teardown (neither). Only a cancellation **we asked for** takes the silent leg — the `RpcException(Cancelled)` filter is guarded by `cts.IsCancellationRequested`. A **foreign** `Cancelled` (the peer cancelled, or the channel was disposed underneath us) falls through to `onError`; unguarded, it fired none of the three, so the aggregator kept `_streamDown = false` — `IsLive` stuck true and the reconcile tick's reopen, which only runs on a stream known to be down, never fired. - **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), `LiveAlarmCacheReconcileJitterFraction` (default 0.2 = up to +20% per tick; `0` disables). Stream channel sizing lives alongside the other gRPC limits: `GrpcInstanceStreamChannelCapacity` (default 1000) and `GrpcSiteAlarmStreamChannelCapacity` (default 20 000). - **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), and counter `scadabridge.site.alarm_cache.buffer_dropped` (deltas evicted from the aggregator's **bounded** 20 000-entry pre-seed buffer, drop-oldest — non-zero means a fan-out ran long enough for the delta storm behind it to exceed the cap; the fan-out's snapshot is authoritative for the evicted rows). - **Accepted limitations (arch review 02 round 2, N8):** diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SiteAuditBacklogReporter.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SiteAuditBacklogReporter.cs index ce1d1ec7..9350e1b1 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SiteAuditBacklogReporter.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SiteAuditBacklogReporter.cs @@ -45,12 +45,27 @@ public sealed class SiteAuditBacklogReporter : IHostedService, IDisposable /// internal static readonly TimeSpan DefaultRefreshInterval = TimeSpan.FromSeconds(30); + /// + /// Age at which the oldest still-Pending row is called out in the log. A Pending row is + /// one central has not acknowledged through EITHER path (telemetry push or reconciliation + /// pull), and PurgeExpiredAsync deliberately never purges Pending — so the site + /// store's floor depends on reconciliation actually running. Rows served by a pull but + /// never covered by a later cursor sit in exactly this state, which is why the age is + /// worth a signal rather than only a dashboard number. One day is comfortably longer than + /// any normal drain/reconcile outage and well inside the ~7-day site retention window. + /// + internal static readonly TimeSpan StalePendingThreshold = TimeSpan.FromHours(24); + + /// How often the stale-pending warning may repeat (it is a standing condition). + internal static readonly TimeSpan StalePendingWarnInterval = TimeSpan.FromHours(1); + private readonly ISiteAuditQueue _queue; private readonly ISiteHealthCollector _collector; private readonly ILogger _logger; private readonly TimeSpan _refreshInterval; private CancellationTokenSource? _cts; private Task? _loop; + private DateTime _lastStalePendingWarnUtc = DateTime.MinValue; /// Initializes a new instance of . /// The site audit queue used to probe the backlog count. @@ -151,6 +166,7 @@ public sealed class SiteAuditBacklogReporter : IHostedService, IDisposable { var snapshot = await _queue.GetBacklogStatsAsync(ct).ConfigureAwait(false); _collector.UpdateSiteAuditBacklog(snapshot); + WarnIfPendingIsStale(snapshot.OldestPendingUtc, snapshot.PendingCount); } catch (OperationCanceledException) { @@ -166,6 +182,34 @@ public sealed class SiteAuditBacklogReporter : IHostedService, IDisposable } } + /// + /// Logs a rate-limited warning when the oldest still-Pending audit row is older than + /// . Pending rows are exempt from the retention purge + /// by design, so a standing Pending backlog is the one site-store condition that does not + /// self-heal on age — it clears only when central acknowledges the rows (telemetry ack or + /// a reconciliation cursor that covers them). Internal so tests can drive it directly. + /// + /// Oldest pending row's occurrence instant, or null when none. + /// Number of rows currently pending. + internal void WarnIfPendingIsStale(DateTime? oldestPendingUtc, int pendingCount) + { + if (oldestPendingUtc is null) return; + + var age = DateTime.UtcNow - DateTime.SpecifyKind(oldestPendingUtc.Value, DateTimeKind.Utc); + if (age < StalePendingThreshold) return; + + var now = DateTime.UtcNow; + if (now - _lastStalePendingWarnUtc < StalePendingWarnInterval) return; + _lastStalePendingWarnUtc = now; + + _logger.LogWarning( + "Site audit backlog: oldest pending row is {AgeHours:F1}h old ({PendingCount} pending). " + + "Pending rows are never purged on age, so this backlog only clears when central " + + "acknowledges them — check the site→central audit telemetry drain and the central " + + "reconciliation pull.", + age.TotalHours, pendingCount); + } + /// Signals the polling loop to stop and waits for it to complete. /// Cancellation token (not used; the internal CTS governs shutdown). /// A task that represents the asynchronous operation. diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs index bc0cf237..98bf7423 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs @@ -792,9 +792,14 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable // ordering the query applies, so it is a strict "everything after this row". : "(fs.OccurredAtUtc > $since OR (fs.OccurredAtUtc = $since AND ae.EventId > $afterId))"; + // fs.rowid rides along as an 11th column purely to maintain + // _maxServedRowId — the insertion-order high-water mark that bounds + // MarkReconciledUpToAsync (see that method). It is never mapped onto + // the returned AuditEvent; rowid stays a storage-layer concept. cmd.CommandText = $""" SELECT ae.EventId, ae.OccurredAtUtc, ae.Actor, ae.Action, ae.Outcome, - ae.Category, ae.Target, ae.SourceNode, ae.CorrelationId, ae.DetailsJson + ae.Category, ae.Target, ae.SourceNode, ae.CorrelationId, ae.DetailsJson, + fs.rowid FROM audit_event ae JOIN audit_forward_state fs ON fs.EventId = ae.EventId WHERE fs.ForwardState IN ($pending, $forwarded) @@ -816,10 +821,47 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable } cmd.Parameters.AddWithValue("$limit", batchSize); - return Task.FromResult(ReadRows(cmd, batchSize)); + return Task.FromResult(ReadServedRows(cmd, batchSize)); } } + /// + /// Highest audit_forward_state.rowid this instance has ever SERVED from + /// — i.e. the insertion-order high-water mark of + /// rows central could possibly have received through the pull path. Bounds + /// so a row inserted after that point can never be + /// retired by a cursor that happens to sit above its timestamp. Monotonic (a later, smaller + /// batch never lowers it) except for the purge clamp. Written under _readLock (the + /// read path) and read under _writeLock (the flip/purge paths) — two different + /// locks, so the accesses go through for visibility. A stale read + /// can only make the bound smaller, i.e. more conservative, never unsafe. + /// + private long _maxServedRowId; + + /// + /// Executes a reconciliation-pull read whose projection carries fs.rowid as an + /// 11th column, materialising the canonical rows while advancing + /// . Serving a row is what makes it eligible for + /// cursor-proved retirement, so the two must happen in the same step. + /// + private IReadOnlyList ReadServedRows(SqliteCommand cmd, int capacityHint) + { + var rows = new List(Math.Min(capacityHint, 256)); + var maxRowId = Volatile.Read(ref _maxServedRowId); + using (var reader = cmd.ExecuteReader()) + { + while (reader.Read()) + { + rows.Add(MapRow(reader)); + var rowId = reader.GetInt64(10); + if (rowId > maxRowId) maxRowId = rowId; + } + } + + Volatile.Write(ref _maxServedRowId, maxRowId); + return rows; + } + /// /// Normalises a wire-supplied event id to the exact textual form stored in /// audit_event.EventId so the keyset comparison is apples-to-apples. An @@ -845,16 +887,40 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable ? "fs.OccurredAtUtc < $since" : "(fs.OccurredAtUtc < $since OR (fs.OccurredAtUtc = $since AND fs.EventId <= $afterId))"; + // INSERTION-ORDER BOUND — "only rows we actually served can retire". + // + // The timestamp cursor alone is not enough. OccurredAtUtc is stamped by the + // caller, so a row can be INSERTED after a batch was served yet carry a + // timestamp BELOW central's (by then advanced) cursor — a late-stamped insert. + // The blanket cursor UPDATE retired exactly those rows: never served, never + // servable again (ReadPendingSinceAsync's keyset has moved past them), and, + // being Reconciled, purged on age — silent audit loss, a failure mode the old + // explicit id-set flip could not produce. + // + // fs.rowid is insertion order, so bounding the flip at the highest rowid this + // instance has ever served restores the invariant exactly: rows inserted after + // that point are out of the flip's reach no matter where the cursor sits. A + // FORWARDED row is exempt from the bound — central ACKED it through the + // telemetry push path, which is receipt-proof independent of the pull. + // + // Bound state is per-process: after a restart it is 0 until this instance serves + // a batch, so the first pull retires only Forwarded rows and the pull after it + // resumes normal retirement. Conservative in the safe direction (delays + // retirement, never loses a row); the liveness note in MarkReconciledUpToAsync's + // interface doc + Component-AuditLog.md covers what a permanently halted + // reconciliation means for such rows. using var cmd = _connection.CreateCommand(); cmd.CommandText = $""" UPDATE audit_forward_state AS fs SET ForwardState = $reconciled WHERE fs.ForwardState IN ($pending, $forwarded) + AND (fs.ForwardState = $forwarded OR fs.rowid <= $maxServedRowId) AND {predicate}; """; cmd.Parameters.AddWithValue("$reconciled", AuditForwardState.Reconciled.ToString()); cmd.Parameters.AddWithValue("$pending", AuditForwardState.Pending.ToString()); cmd.Parameters.AddWithValue("$forwarded", AuditForwardState.Forwarded.ToString()); + cmd.Parameters.AddWithValue("$maxServedRowId", Volatile.Read(ref _maxServedRowId)); cmd.Parameters.AddWithValue("$since", EnsureUtc(sinceUtc).ToString( "o", System.Globalization.CultureInfo.InvariantCulture)); if (afterId is not null) @@ -1063,6 +1129,22 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable dropCmd.ExecuteNonQuery(); } + // Clamp the served-rows high-water mark to what is still in the table. + // SQLite hands a fresh insert MAX(rowid)+1, so a purge that empties the + // sidecar (or removes its top rows) makes rowids REUSABLE — a stale-high + // _maxServedRowId would then vouch for rows nobody has served. Cheap: + // MAX(rowid) is an index-less O(1) lookup. + if (purged > 0) + { + using var maxCmd = _connection.CreateCommand(); + maxCmd.Transaction = transaction; + maxCmd.CommandText = "SELECT IFNULL(MAX(rowid), 0) FROM audit_forward_state;"; + var liveMax = Convert.ToInt64(maxCmd.ExecuteScalar(), + System.Globalization.CultureInfo.InvariantCulture); + if (liveMax < Volatile.Read(ref _maxServedRowId)) + Volatile.Write(ref _maxServedRowId, liveMax); + } + transaction.Commit(); } catch diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/ISiteAuditQueue.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/ISiteAuditQueue.cs index da545ff5..be3fc2ff 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/ISiteAuditQueue.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/ISiteAuditQueue.cs @@ -143,6 +143,29 @@ public interface ISiteAuditQueue /// are proven received; rows at the boundary instant are left /// alone. Idempotent; already-Reconciled rows are untouched. /// + /// + /// Only-served-rows-retire. The cursor is a timestamp, and + /// is stamped by the caller, so a row can be + /// INSERTED after a batch was served yet carry a timestamp below the (by then advanced) + /// cursor. Implementations MUST NOT retire such a row: it was never served, the keyset + /// read has moved past it, and retiring it would make it purgeable — silent audit loss. + /// The SQLite implementation bounds the flip by insertion order (rowid high-water + /// mark of rows it has served), exempting rows already + /// + /// because central ACKED those through the telemetry push path. + /// + /// + /// Liveness dependency (documented, accepted). A row that WAS served but is never + /// covered by a later cursor — central reconciliation stopped for good, or the site node + /// restarted and its bound reset — stays + /// + /// indefinitely, and never purges Pending. That is the + /// retention invariant working as designed (an unacknowledged row is not droppable), but + /// it means the site store's floor is bounded by reconciliation actually running. The + /// backlog is observable: reports the pending count and + /// oldest-pending instant on every site health report, and the site reporter logs a + /// warning once the oldest pending row exceeds its stale threshold. + /// /// /// The cursor timestamp central has consumed up to (UTC). /// The last consumed at that instant, or null. diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/DebugStreamBridgeActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/DebugStreamBridgeActor.cs index dc83fb91..8d105103 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/DebugStreamBridgeActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/DebugStreamBridgeActor.cs @@ -47,6 +47,7 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers private const string ReconnectTimerKey = "grpc-reconnect"; private const string StabilityTimerKey = "grpc-stability"; private const string SnapshotTimerKey = "debug-snapshot-deadline"; + private const string ConsumerLivenessTimerKey = "debug-consumer-liveness"; /// Delay between gRPC reconnection attempts. internal static TimeSpan ReconnectDelay { get; set; } = TimeSpan.FromSeconds(5); @@ -69,6 +70,33 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers /// internal static TimeSpan StabilityWindow { get; set; } = TimeSpan.FromSeconds(60); + /// + /// Orphan window: how long the session may go without ANY sign of life from its CONSUMER + /// before it self-terminates. Renewed by , which + /// DebugStreamService Tells on a timer for every session still attached to a + /// consumer (Blazor debug view or the SignalR hub) — so it measures the consumer, never + /// the stream. Settable for tests. + /// + internal static TimeSpan ConsumerIdleTimeout { get; set; } = TimeSpan.FromMinutes(5); + + /// + /// How often the actor checks the consumer-last-seen stamp against + /// . A self-tick rather than SetReceiveTimeout: + /// the receive timeout measures the MAILBOX, which conflates site chatter with consumer + /// liveness — and once stream events were correctly excluded from it (via + /// ) nothing recurring reset it at all, so every healthy + /// session self-terminated one window after its snapshot with a false "Site disconnected". + /// Settable for tests. + /// + internal static TimeSpan ConsumerLivenessCheckInterval { get; set; } = TimeSpan.FromSeconds(30); + + /// + /// When the consumer was last known to be attached (UTC). Seeded in + /// so a session gets a full window to receive its first keepalive, then refreshed by every + /// . Actor-thread only. + /// + private DateTime _consumerLastSeenUtc = DateTime.UtcNow; + private int _retryCount; private bool _useNodeA = true; private bool _stopped; @@ -188,7 +216,7 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers // non-deployed instance — cancel it (and any buffered gap events are // discarded with the actor). No pass-through. // _stopped is set AFTER CleanupGrpc() to match the ordering in the - // DebugStreamTerminated and ReceiveTimeout handlers (cosmetic consistency). + // DebugStreamTerminated and consumer-liveness handlers (cosmetic consistency). CleanupGrpc(); _stopped = true; _preSnapshotBuffer.Clear(); @@ -217,8 +245,8 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers // Hard snapshot deadline (WP2.3). Nothing else ends a session stuck in the // buffering phase: the site's reply was lost, so no gRPC error fires, the stream - // keeps delivering events, and (with the wrapper above) they no longer even reset - // the orphan timeout. Fail the session so the consumer is told and can reopen. + // keeps delivering events, and stream traffic does not renew the orphan net (which + // measures the consumer). Fail the session so the consumer is told and can reopen. Receive(_ => { if (_stopped || _snapshotDelivered) return; @@ -234,21 +262,19 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers Context.Stop(Self); }); - // Domain events arriving via Self.Tell from the gRPC callback, wrapped so they do - // NOT influence the receive timeout (WP2.3): the orphan safety net exists to end a - // session whose CONSUMER is gone, and a busy site's event flood used to keep that - // net permanently reset — an abandoned session on a chatty instance never timed out. - // Receiving an event must not reset _retryCount either: a flapping stream that - // delivers a single event between failures would otherwise never trip MaxRetries. - // The retry budget is recovered only by GrpcStreamStable (a stream that has stayed - // up for StabilityWindow). Before the snapshot has been delivered, BUFFER (in arrival - // order) rather than deliver — these may be gap-window events; after the snapshot has - // been flushed, pass through directly (phase-dependent behavior). + // Domain events arriving via Self.Tell from the gRPC callback. Stream traffic never + // proves the CONSUMER is still there, so it deliberately does not touch the orphan + // net (which now measures the consumer keepalive, not the mailbox). Receiving an + // event must not reset _retryCount either: a flapping stream that delivers a single + // event between failures would otherwise never trip MaxRetries. The retry budget is + // recovered only by GrpcStreamStable (a stream that has stayed up for + // StabilityWindow). Before the snapshot has been delivered, BUFFER (in arrival order) + // rather than deliver — these may be gap-window events; after the snapshot has been + // flushed, pass through directly (phase-dependent behavior). Receive(wrapped => HandleStreamEvent(wrapped.Event)); // Unwrapped forms are still accepted (a direct Tell from a test or a future - // in-process producer); those DO influence the receive timeout, which is correct — - // they are not the high-volume stream path. + // in-process producer) and take the identical path. Receive(changed => HandleStreamEvent(changed)); Receive(changed => HandleStreamEvent(changed)); @@ -315,11 +341,32 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers Context.Stop(Self); }); - // Orphan safety net — if nobody stops us within 5 minutes, self-terminate - Context.SetReceiveTimeout(TimeSpan.FromMinutes(5)); - Receive(_ => + // Consumer keepalive: DebugStreamService Tells this on a timer for every session it + // still holds (i.e. still attached to a Blazor debug view / SignalR connection). It + // is the ONLY thing that renews the orphan window — deliberately, so neither a chatty + // site nor a silent one can influence it. + Receive(_ => { - _log.Warning("Debug stream for {0} timed out (orphaned session), stopping", _instanceUniqueName); + if (_stopped) return; + _consumerLastSeenUtc = DateTime.UtcNow; + }); + + // Orphan safety net, CONSUMER-measured (WP2.3 follow-up). A periodic self-tick + // compares the consumer-last-seen stamp against ConsumerIdleTimeout; the previous + // SetReceiveTimeout(5 min) measured the mailbox instead, and once stream events were + // (correctly) marked INotInfluenceReceiveTimeout nothing recurring reset it — the + // snapshot arrives once and GrpcStreamStable once, so EVERY healthy session died at + // ~6 minutes and the consumer was told "Site disconnected". + Receive(_ => + { + if (_stopped) return; + var idle = DateTime.UtcNow - _consumerLastSeenUtc; + if (idle < ConsumerIdleTimeout) return; + + _log.Warning( + "Debug stream for {0} has had no consumer activity for {1:F0}s (orphaned session), stopping", + _instanceUniqueName, idle.TotalSeconds); + Timers.Cancel(ConsumerLivenessTimerKey); CleanupGrpc(); SendUnsubscribe(); _stopped = true; @@ -507,6 +554,15 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers // Arm the hard snapshot deadline alongside the request. if (SnapshotTimeout > TimeSpan.Zero) Timers.StartSingleTimer(SnapshotTimerKey, new DebugSnapshotDeadline(), SnapshotTimeout); + + // Arm the consumer-liveness net. The stamp starts now, so the session always gets a + // full ConsumerIdleTimeout to see its first keepalive from DebugStreamService. + _consumerLastSeenUtc = DateTime.UtcNow; + if (ConsumerIdleTimeout > TimeSpan.Zero && ConsumerLivenessCheckInterval > TimeSpan.Zero) + { + Timers.StartPeriodicTimer( + ConsumerLivenessTimerKey, new ConsumerLivenessTick(), ConsumerLivenessCheckInterval); + } } /// @@ -546,7 +602,7 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers await client.SubscribeAsync( _correlationId, _instanceUniqueName, - // Wrapped: stream traffic must not reset the orphan receive timeout. + // Wrapped so the stream path is explicit (it never renews the orphan net). evt => self.Tell(new LiveDebugStreamEvent(evt)), ex => self.Tell(new GrpcStreamError(ex, generation)), () => self.Tell(new GrpcStreamCompleted(generation)), @@ -660,11 +716,26 @@ public record StopDebugStream; /// /// Envelope for a live gRPC stream event (AttributeValueChanged/ -/// AlarmStateChanged). Implements so a busy -/// site's event flood cannot keep resetting the orphan-session receive timeout — the timeout -/// measures consumer/session liveness, not site chatter (WP2.3). +/// AlarmStateChanged). Kept as a distinct envelope so the high-volume stream path is +/// explicit at the call site; the orphan net no longer keys off the mailbox at all (it +/// measures the consumer keepalive), so a busy site's event flood can neither hold a dead +/// session open nor — as briefly happened — be the only thing keeping a healthy one alive. /// -internal record LiveDebugStreamEvent(object Event) : INotInfluenceReceiveTimeout; +internal record LiveDebugStreamEvent(object Event); + +/// +/// Consumer keepalive: DebugStreamService Tells one of these to every bridge actor +/// whose session is still attached to a consumer, on +/// -scale cadence. Renewing +/// the consumer-last-seen stamp is its ONLY effect. +/// +public record DebugStreamConsumerAlive; + +/// +/// Internal self-tick that checks the consumer-last-seen stamp against +/// . +/// +internal record ConsumerLivenessTick; /// /// Internal message: the hard deadline for the initial DebugViewSnapshot expired. diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs index fc092ba2..e2e0b439 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs @@ -144,15 +144,38 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers private bool _seedOnConnect; /// - /// Set whenever a fan-out finishes (success or failure); consumed by the next reconcile - /// tick, which skips its own fan-out when it finds the flag set. That makes the - /// connect-driven seed and the periodic backstop mutually exclusive — the pair used to + /// Set when a CONNECT- OR FAILOVER-DRIVEN fan-out finishes (the initial seed, a re-seed + /// consumed by a successful (re)connect, or a re-seed queued behind one); consumed by the + /// next reconcile tick, which skips its own fan-out when it finds the flag set. That makes + /// the connect-driven seed and the periodic backstop mutually exclusive — the pair used to /// run BOTH, which is the "full unconditional snapshot every 60s" waste — while bounding /// staleness at two intervals (a tick can be skipped at most once in a row). + /// + /// A TICK-driven fan-out deliberately does NOT set it. It used to: the tick's own fan-out + /// completion armed the skip, so steady state ran fan-out → skip → fan-out → skip, i.e. + /// one reconcile per TWO intervals — halving the not-reporting refresh rate and the alarm + /// reconcile backstop for no reason (nothing was duplicated to suppress). + /// /// Actor-thread only. /// private bool _fanoutSinceLastTick; + /// + /// Whether the in-flight fan-out is connect/failover-driven and therefore makes the next + /// reconcile tick redundant (see ). Stable for the + /// fan-out's lifetime: never overwrites it while one is in + /// flight (a colliding request queues instead). Actor-thread only. + /// + private bool _fanoutSuppressesNextTick; + + /// + /// Same flag for a fan-out that was queued behind an in-flight one + /// () — OR-ed across colliding requests so a queued + /// connect/failover re-seed keeps its suppression even if a tick collided too. + /// Actor-thread only. + /// + private bool _queuedFanoutSuppressesNextTick; + /// /// True between issuing a stream open and its first outcome (connected / error / /// completed), so a reconcile tick never stacks a second open on an in-flight one. @@ -269,7 +292,9 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers Receive(_ => { if (_stopped) return; - StartFanout(isInitial: false); + // A backoff retry of a failed fan-out is not a connect/failover seed: it must + // not stand the next reconcile tick down. + StartFanout(isInitial: false, suppressesNextTick: false); }); // The site-wide stream is confirmed established (response headers received). This — @@ -356,7 +381,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers // Kick the initial seed fan-out. The connect callback for this first stream must // NOT seed again on top of it, so the flag starts cleared. _seedOnConnect = false; - StartFanout(isInitial: true); + StartFanout(isInitial: true, suppressesNextTick: true); // Periodic reconcile backstop. Single-shot and re-armed with fresh jitter each // tick (a periodic timer would lock every site to the same phase forever). @@ -398,9 +423,11 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers /// /// Periodic reconcile backstop: re-run the snapshot fan-out (corrects instance-set drift - /// + any missed delta) UNLESS one already completed inside this window — a reconnect-driven - /// seed makes the tick redundant, and running both was the "full unconditional snapshot - /// every 60s" waste (WP2.3). If the live stream was previously given up, self-heal it by + /// + any missed delta) UNLESS a CONNECT- OR FAILOVER-DRIVEN seed already completed inside + /// this window — that seed makes the tick redundant, and running both was the "full + /// unconditional snapshot every 60s" waste (WP2.3). A tick's OWN fan-out never arms that + /// skip: it used to, which made steady state alternate fan-out/skip and halved the + /// effective reconcile rate. If the live stream was previously given up, self-heal it by /// resetting the retry budget and reopening, so a sustained outage never permanently kills /// the live feed. /// @@ -441,7 +468,11 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers } if (!reopening) - StartFanout(isInitial: false); + { + // Tick-driven: deliberately does NOT set the skip flag, so the NEXT tick fans + // out too. Setting it here was the "one reconcile per two intervals" bug. + StartFanout(isInitial: false, suppressesNextTick: false); + } } // ── Seed / reconcile fan-out ──────────────────────────────────────────────── @@ -451,7 +482,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers /// Self.Tell. While in flight, live deltas buffer. A reconcile that arrives /// while a fan-out is already running is skipped (no stacking). /// - private void StartFanout(bool isInitial) + private void StartFanout(bool isInitial, bool suppressesNextTick) { if (_stopped) return; if (_fanoutInFlight) @@ -460,11 +491,16 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers // 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; + if (!isInitial) + { + _reseedQueued = true; + _queuedFanoutSuppressesNextTick |= suppressesNextTick; + } return; } _fanoutInFlight = true; + _fanoutSuppressesNextTick = suppressesNextTick; var self = Self; var ct = _lifetimeCts?.Token ?? CancellationToken.None; @@ -514,7 +550,9 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers var flushChanged = FlushBuffer(); _fanoutInFlight = false; - _fanoutSinceLastTick = true; + // Only a connect/failover-driven seed makes the next tick redundant — a tick's own + // fan-out must never suppress the tick after it (that halved the reconcile rate). + if (_fanoutSuppressesNextTick) _fanoutSinceLastTick = true; _consecutiveSeedFailures = 0; Timers.Cancel(SeedRetryTimerKey); var firstSeed = !_seeded; @@ -531,11 +569,21 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers Publish(); // A failover re-seed requested while this fan-out was in flight runs now (N7.1). - if (_reseedQueued) - { - _reseedQueued = false; - StartFanout(isInitial: false); - } + RunQueuedFanoutIfAny(); + } + + /// + /// Runs a fan-out that collided with an in-flight one, carrying the suppression flag it + /// was queued with so a queued connect/failover re-seed still stands the next tick down + /// (and a queued TICK fan-out still does not). + /// + private void RunQueuedFanoutIfAny() + { + if (!_reseedQueued) return; + _reseedQueued = false; + var suppresses = _queuedFanoutSuppressesNextTick; + _queuedFanoutSuppressesNextTick = false; + StartFanout(isInitial: false, suppressesNextTick: suppresses); } /// @@ -570,7 +618,9 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers // Don't lose deltas captured during the failed window — apply them pass-through // into the (possibly stale/empty) cache. The next reconcile re-seeds authoritatively. _fanoutInFlight = false; - _fanoutSinceLastTick = true; + // Same rule as the success path: only a connect/failover-driven fan-out stands the + // next reconcile tick down. + if (_fanoutSuppressesNextTick) _fanoutSinceLastTick = true; var flushChanged = FlushBuffer(dedupAgainstSeed: false); // Only publish if we already had a seed (so IsLive doesn't flip true on a @@ -585,8 +635,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers // A failover re-seed requested while this fan-out was in flight runs now (N7.1). if (_reseedQueued) { - _reseedQueued = false; - StartFanout(isInitial: false); + RunQueuedFanoutIfAny(); return; } @@ -794,7 +843,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers { _seedOnConnect = false; _log.Info("Site-alarm gRPC stream for {0} (re)connected; running one re-seed", _siteIdentifier); - StartFanout(isInitial: false); + StartFanout(isInitial: false, suppressesNextTick: true); } // Liveness recovered — republish so viewers stop falling back to polling even if the diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/DebugStreamService.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/DebugStreamService.cs index 454ab8d3..b467efce 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/DebugStreamService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/DebugStreamService.cs @@ -13,14 +13,34 @@ namespace ZB.MOM.WW.ScadaBridge.Communication; /// Manages debug stream sessions by creating DebugStreamBridgeActors that persist /// as subscribers on the site side. Both the Blazor debug view and the SignalR hub /// use this service to start/stop streams. +/// +/// Consumer keepalive. This service is the session registry, and holding a session +/// here IS what "a consumer is attached" means: the Blazor debug view and the SignalR hub +/// both release their session () on dispose/disconnect. A single +/// shared timer therefore Tells to every registered +/// bridge actor, which is the only thing that renews each actor's orphan window. A bridge +/// actor that outlives its registration — the leak the orphan net exists for — stops +/// receiving keepalives and self-terminates. The keepalive is independent of stream traffic +/// by design: neither a chatty nor a silent site can influence the orphan decision. +/// /// -public class DebugStreamService +public class DebugStreamService : IDisposable { + /// + /// Cadence of the shared consumer keepalive. Comfortably shorter than + /// so a few missed ticks (GC + /// pause, thread-pool starvation) can never orphan a live session. Settable for tests. + /// + internal static TimeSpan KeepaliveInterval { get; set; } = TimeSpan.FromSeconds(30); + private readonly CommunicationService _communicationService; private readonly IServiceProvider _serviceProvider; private readonly SiteStreamGrpcClientFactory _grpcClientFactory; private readonly ILogger _logger; private readonly ConcurrentDictionary _sessions = new(); + private readonly object _keepaliveLock = new(); + private Timer? _keepaliveTimer; + private bool _disposed; private ActorSystem? _actorSystem; /// @@ -135,6 +155,7 @@ public class DebugStreamService var bridgeActor = system.ActorOf(props, $"debug-stream-{sessionId}"); _sessions[sessionId] = bridgeActor; + EnsureKeepaliveTimer(); // Wait for the initial snapshot (with timeout) using var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(ct); @@ -184,6 +205,73 @@ public class DebugStreamService _logger.LogInformation("Debug stream {SessionId} stopped", sessionId); } } + + /// + /// Sends one keepalive round to every attached session. Exposed (internal) so tests can + /// drive the keepalive deterministically instead of waiting on the timer. + /// + internal void SendConsumerKeepalives() + { + foreach (var session in _sessions) + { + session.Value.Tell(new DebugStreamConsumerAlive()); + } + } + + /// + /// Starts the shared keepalive timer on first use. One timer for the whole service (not + /// one per session): the payload is a Tell per attached session, so a single periodic + /// callback is the cheapest shape that covers every consumer surface. + /// + private void EnsureKeepaliveTimer() + { + if (_keepaliveTimer is not null) return; + lock (_keepaliveLock) + { + if (_keepaliveTimer is not null || _disposed) return; + _keepaliveTimer = new Timer( + _ => + { + try + { + SendConsumerKeepalives(); + } + catch (Exception ex) + { + // A keepalive round must never take the timer down — a thrown Tell + // would silently stop renewing EVERY session's orphan window. + _logger.LogWarning(ex, "Debug stream consumer keepalive round failed"); + } + }, + null, + KeepaliveInterval, + KeepaliveInterval); + } + } + + /// Stops the keepalive timer. + public void Dispose() + { + Dispose(true); + GC.SuppressFinalize(this); + } + + /// Stops the keepalive timer. + /// True when called from . + protected virtual void Dispose(bool disposing) + { + if (_disposed) return; + lock (_keepaliveLock) + { + if (_disposed) return; + _disposed = true; + if (disposing) + { + _keepaliveTimer?.Dispose(); + _keepaliveTimer = null; + } + } + } } public record DebugStreamSession(string SessionId, DebugViewSnapshot InitialSnapshot); diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClient.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClient.cs index 667fd1d3..2789cabb 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClient.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClient.cs @@ -246,11 +246,12 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable /// /// Cancellation token to stop the subscription. /// - /// Optional callback invoked once when the site has ACCEPTED the subscription (response - /// headers received — the site writes them as soon as its relay actor is subscribed, so - /// no event can be missed after this point). The per-site aggregator uses it to run - /// exactly one re-seed per successful (re)connect instead of one per reconnect attempt. - /// Never invoked more than once per call, and never after . + /// Optional callback invoked once when the site has ACCEPTED the subscription — response + /// headers received (the site writes them as soon as its relay actor is subscribed, so no + /// event can be missed after this point), or the first event received if the peer defers + /// its headers. A header timeout is NOT treated as connected. The per-site aggregator uses + /// this to run exactly one re-seed per successful (re)connect instead of one per reconnect + /// attempt. Never invoked more than once per call, and never after . /// /// A task that represents the asynchronous operation. public virtual async Task SubscribeSiteAsync( @@ -311,11 +312,11 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable /// Invoked once if the stream faulted. /// Invoked once if the server ended the stream with OK. /// - /// Optional; invoked once when the server's response headers arrive — i.e. the site has - /// accepted the subscription and its relay actor is attached. Bounded by - /// so a peer that defers headers (a pre-WP2.3 site, - /// which only flushes them with its first event) still reports connected instead of - /// leaving the caller waiting for a signal that may never come on a quiet site. + /// Optional; invoked AT MOST ONCE when the site has demonstrably accepted the + /// subscription — either the server's response headers arrived (bounded by + /// ) or, for a peer that defers headers until its + /// first message, the first event was received. A header timeout alone is never + /// reported as connected: that is also exactly what an unreachable site looks like. /// /// A task that completes when the stream has ended and its outcome been reported. internal async Task ConsumeStreamAsync( @@ -328,27 +329,49 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable Action? onConnected = null) { var completedGracefully = false; + var connectedReported = false; + + // Fires onConnected AT MOST ONCE, from whichever proof of a live peer arrives + // first: the response headers, or (for a peer that defers headers until its first + // message) the first event. A header TIMEOUT is deliberately NOT such a proof — + // see AwaitHeadersAsync. + void ReportConnected() + { + if (connectedReported || onConnected is null) return; + connectedReported = true; + onConnected(); + } + try { using (var call = openCall()) { - if (onConnected is not null) + if (onConnected is not null && await AwaitHeadersAsync(call, cts.Token).ConfigureAwait(false)) { - await AwaitHeadersAsync(call, cts.Token).ConfigureAwait(false); - onConnected(); + ReportConnected(); } await foreach (var evt in call.ResponseStream.ReadAllAsync(cts.Token)) { + // Fallback connected signal: an event can only come from a peer that + // accepted the subscription, so it proves what the headers would have. + // Raised BEFORE the event is delivered so the consumer sees + // connected-then-event ordering. + ReportConnected(); onEvent(evt); } } completedGracefully = true; } - catch (RpcException ex) when (ex.StatusCode == StatusCode.Cancelled) + catch (RpcException ex) when (ex.StatusCode == StatusCode.Cancelled && cts.IsCancellationRequested) { - // Normal cancellation — not an error + // OUR OWN cancellation (Unsubscribe / reconnect / channel teardown we asked + // for) — not an error. The IsCancellationRequested guard matters: a Cancelled + // status can also originate at the PEER or from a channel disposed underneath + // us, and swallowing THAT fired none of onError/onCompleted/onConnected, so + // the consuming actor kept a dead stream marked live forever. Foreign + // Cancelled now falls through to the onError path below. } catch (OperationCanceledException) when (cts.IsCancellationRequested) { @@ -372,26 +395,37 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable } /// - /// How long to wait for response headers before treating the stream as connected anyway. - /// A peer that only flushes headers with its first message would otherwise hold the - /// connected signal — and with it the aggregator's re-seed — for as long as the site - /// happens to be quiet. + /// How long to wait for response headers before giving up on them as the connected + /// signal and falling back to the first received event. A peer that only flushes headers + /// with its first message would otherwise hold the connected signal — and with it the + /// aggregator's re-seed — for as long as the site happens to be quiet. /// internal static TimeSpan ConnectedHeaderTimeout { get; set; } = TimeSpan.FromSeconds(10); /// /// Awaits the call's response headers, bounded by . - /// A fault propagates (the caller reports it through onError like any other stream - /// fault); a timeout returns normally. On timeout the abandoned headers task is observed - /// so a later fault on it can never surface as an unobserved task exception. + /// Returns when the headers arrived (the site accepted the + /// subscription) and on timeout. A fault propagates (the caller + /// reports it through onError like any other stream fault); on timeout the + /// abandoned headers task is observed so a later fault on it can never surface as an + /// unobserved task exception. + /// + /// A timeout must NOT be reported as connected: an unreachable/wedged site produces + /// exactly that shape, and calling onConnected for it made the aggregator clear + /// _streamDown, consume its pending re-seed and fan a full snapshot out at a site + /// that never answered. The caller instead treats the FIRST RECEIVED EVENT as the + /// fallback connected signal — real proof of a live peer, and the quiet-site case the + /// timeout was added for is covered by the reconcile backstop. + /// /// - private static async Task AwaitHeadersAsync( + private static async Task AwaitHeadersAsync( AsyncServerStreamingCall call, CancellationToken ct) { var headers = call.ResponseHeadersAsync; try { await headers.WaitAsync(ConnectedHeaderTimeout, ct).ConfigureAwait(false); + return true; } catch (TimeoutException) { @@ -400,6 +434,7 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable CancellationToken.None, TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); + return false; } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SiteAuditBacklogReporterCadenceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SiteAuditBacklogReporterCadenceTests.cs index 4798aad9..daa0826d 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SiteAuditBacklogReporterCadenceTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SiteAuditBacklogReporterCadenceTests.cs @@ -1,4 +1,5 @@ using System.Diagnostics; +using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; using NSubstitute; @@ -115,4 +116,67 @@ public class SiteAuditBacklogReporterCadenceTests Assert.Equal(TimeSpan.FromSeconds(3), reporter.RefreshInterval); } + + // ----- Stale-Pending signal (review F5) ----- // + + [Fact] + public void StalePendingBacklog_IsWarned_ThenRateLimited() + { + // Pending rows are exempt from the retention purge by design, so a standing Pending + // backlog is the one site-store condition that never self-heals on age — it clears + // only when central acknowledges the rows. It is worth a log line, not just a number + // on the health report, and the warning must not spam every 30 s poll. + var logger = new CapturingLogger(); + var reporter = new SiteAuditBacklogReporter( + Substitute.For(), + Substitute.For(), + logger, + TimeSpan.FromHours(1), + null); + + var stale = DateTime.UtcNow - SiteAuditBacklogReporter.StalePendingThreshold - TimeSpan.FromHours(1); + reporter.WarnIfPendingIsStale(stale, pendingCount: 4321); + reporter.WarnIfPendingIsStale(stale, pendingCount: 4321); // same poll cycle-ish + + var warning = Assert.Single(logger.Entries, e => e.Level == LogLevel.Warning); + Assert.Contains("4321", warning.Message); + Assert.Contains("pending", warning.Message, StringComparison.OrdinalIgnoreCase); + } + + [Fact] + public void FreshOrEmptyPendingBacklog_IsNotWarned() + { + var logger = new CapturingLogger(); + var reporter = new SiteAuditBacklogReporter( + Substitute.For(), + Substitute.For(), + logger, + TimeSpan.FromHours(1), + null); + + reporter.WarnIfPendingIsStale(null, pendingCount: 0); // nothing pending + reporter.WarnIfPendingIsStale(DateTime.UtcNow.AddMinutes(-5), 12); // a normal drain lag + + Assert.DoesNotContain(logger.Entries, e => e.Level == LogLevel.Warning); + } + + /// Captures log entries so the stale-pending signal can be asserted. + private sealed class CapturingLogger : ILogger + { + public List<(LogLevel Level, Exception? Exception, string Message)> Entries { get; } = new(); + + public IDisposable? BeginScope(TState state) where TState : notnull => null; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log( + LogLevel logLevel, + EventId eventId, + TState state, + Exception? exception, + Func formatter) + { + Entries.Add((logLevel, exception, formatter(state, exception))); + } + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterWriteTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterWriteTests.cs index 73391a30..2f388a04 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterWriteTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterWriteTests.cs @@ -614,6 +614,10 @@ public class SqliteAuditWriterWriteTests await writer.WriteAsync(older); await writer.WriteAsync(boundary); + // Serve them first — retirement is bounded by what has actually been served, which + // is exactly the order the pull handler runs in (read, then next pull's cursor). + await writer.ReadPendingSinceAsync(DateTime.MinValue, batchSize: 100); + var flipped = await writer.MarkReconciledUpToAsync(instant, afterId: null); Assert.Equal(1, flipped); @@ -648,6 +652,99 @@ public class SqliteAuditWriterWriteTests second.Select(r => r.EventId).ToHashSet()); } + [Fact] + public async Task MarkReconciledUpToAsync_LateStampedInsertBelowTheCursor_IsNeverRetired() + { + // THE data-loss case. OccurredAtUtc is caller-stamped, so a row can be INSERTED + // after a batch was served yet carry a timestamp BELOW central's (by then advanced) + // cursor — a script that back-dates, a clock nudge, a queued write flushed late. + // The blanket cursor UPDATE retired exactly those rows: never served, never servable + // again (the keyset read has moved past them) and, being Reconciled, purged on age. + // The insertion-order (rowid) bound makes them unreachable by the flip. + var (writer, dataSource) = CreateWriter( + nameof(MarkReconciledUpToAsync_LateStampedInsertBelowTheCursor_IsNeverRetired)); + await using var _w = writer; + + var t0 = new DateTime(2026, 5, 20, 12, 0, 0, DateTimeKind.Utc); + var served1 = NewEvent(occurredAtUtc: t0); + var served2 = NewEvent(occurredAtUtc: t0.AddSeconds(20)); + await writer.WriteAsync(served1); + await writer.WriteAsync(served2); + + // Central pulls both and advances its cursor to the newest row. + var page = await writer.ReadPendingSinceAsync(DateTime.MinValue, batchSize: 100); + Assert.Equal(2, page.Count); + var cursorTime = t0.AddSeconds(20); + var cursorId = served2.EventId.ToString(); + + // A row lands NOW carrying a timestamp between the two served rows. + var lateStamped = NewEvent(occurredAtUtc: t0.AddSeconds(10)); + await writer.WriteAsync(lateStamped); + + var flipped = await writer.MarkReconciledUpToAsync(cursorTime, cursorId); + + // The two genuinely served rows retire; the late-stamped one does not. + Assert.Equal(2, flipped); + Assert.Equal(AuditForwardState.Reconciled.ToString(), ReadForwardState(dataSource, served1.EventId)); + Assert.Equal(AuditForwardState.Reconciled.ToString(), ReadForwardState(dataSource, served2.EventId)); + Assert.Equal(AuditForwardState.Pending.ToString(), ReadForwardState(dataSource, lateStamped.EventId)); + + // Still recoverable: a central that restarts (cursor resets to MinValue) re-serves it, + // and — being Pending — the retention purge can never drop it in the meantime. + var reread = await writer.ReadPendingSinceAsync(DateTime.MinValue, batchSize: 100); + Assert.Equal(lateStamped.EventId, Assert.Single(reread).EventId); + } + + [Fact] + public async Task MarkReconciledUpToAsync_ForwardedRowsRetire_EvenWithoutHavingBeenPulled() + { + // The bound applies to PENDING rows (nothing proves central saw them but the pull). + // A FORWARDED row was ACKED by central through the telemetry push path, so the cursor + // may retire it whether or not this node has ever served it in a pull — otherwise a + // site node that never serves a pull would accumulate acked rows forever. + var (writer, dataSource) = CreateWriter( + nameof(MarkReconciledUpToAsync_ForwardedRowsRetire_EvenWithoutHavingBeenPulled)); + await using var _w = writer; + + var t0 = new DateTime(2026, 5, 20, 12, 0, 0, DateTimeKind.Utc); + var pushed = NewEvent(occurredAtUtc: t0); + var neverShipped = NewEvent(occurredAtUtc: t0.AddSeconds(1)); + await writer.WriteAsync(pushed); + await writer.WriteAsync(neverShipped); + + // Central acked the first row over the telemetry drain — no pull involved. + await writer.MarkForwardedAsync(new[] { pushed.EventId }); + + var flipped = await writer.MarkReconciledUpToAsync(t0.AddSeconds(30), afterId: null); + + Assert.Equal(1, flipped); + Assert.Equal(AuditForwardState.Reconciled.ToString(), ReadForwardState(dataSource, pushed.EventId)); + Assert.Equal(AuditForwardState.Pending.ToString(), ReadForwardState(dataSource, neverShipped.EventId)); + } + + [Fact] + public async Task MarkReconciledUpToAsync_BeforeAnyPull_RetiresNothingPending() + { + // Bound state is per-process: after a site-node restart nothing has been served yet, + // so an incoming cursor retires no Pending row. Conservative in the safe direction — + // the rows stay servable and the next pull re-establishes the bound. + var (writer, dataSource) = CreateWriter(nameof(MarkReconciledUpToAsync_BeforeAnyPull_RetiresNothingPending)); + await using var _w = writer; + + var t0 = new DateTime(2026, 5, 20, 12, 0, 0, DateTimeKind.Utc); + var evt = NewEvent(occurredAtUtc: t0); + await writer.WriteAsync(evt); + + var flipped = await writer.MarkReconciledUpToAsync(t0.AddHours(1), afterId: null); + + Assert.Equal(0, flipped); + Assert.Equal(AuditForwardState.Pending.ToString(), ReadForwardState(dataSource, evt.EventId)); + + // …and the very next pull cycle retires it normally. + await writer.ReadPendingSinceAsync(DateTime.MinValue, batchSize: 100); + Assert.Equal(1, await writer.MarkReconciledUpToAsync(t0.AddHours(1), afterId: null)); + } + [Fact] public async Task ReadPendingSinceAsync_InvalidBatchSize_Throws() { diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/DebugStreamServiceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/DebugStreamServiceTests.cs index f91b0377..5a69a6af 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/DebugStreamServiceTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/DebugStreamServiceTests.cs @@ -16,6 +16,11 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Tests; /// /// Tests for DebugStreamService session lifecycle. /// +/// +/// Shares the DebugStreamStatics xUnit collection with DebugStreamBridgeActorTests +/// so the two classes never race on the actor's static test seams (see that class). +/// +[Collection("DebugStreamStatics")] public class DebugStreamServiceTests : TestKit { [Fact] @@ -74,4 +79,89 @@ public class DebugStreamServiceTests : TestKit Assert.Contains("Site1.Pump01", ex.Message); Assert.NotNull(ex.InnerException); } + + [Fact] + public async Task AttachedSession_IsKeptAliveByTheServiceKeepalive_AndDiesOnceDetached() + { + // The consumer half of the orphan net: holding a session in DebugStreamService IS + // "a consumer is attached", and the service's shared timer is what renews the bridge + // actor's window. Without this wiring every healthy session self-terminated one + // window after its snapshot (the old mailbox-based ReceiveTimeout had nothing + // recurring to reset it) and the consumer was told "Site disconnected". + var previousKeepalive = DebugStreamService.KeepaliveInterval; + var previousIdle = DebugStreamBridgeActor.ConsumerIdleTimeout; + var previousCheck = DebugStreamBridgeActor.ConsumerLivenessCheckInterval; + DebugStreamService.KeepaliveInterval = TimeSpan.FromMilliseconds(50); + DebugStreamBridgeActor.ConsumerIdleTimeout = TimeSpan.FromMilliseconds(400); + DebugStreamBridgeActor.ConsumerLivenessCheckInterval = TimeSpan.FromMilliseconds(50); + try + { + var instance = new Instance("Site1.Pump01") { Id = 7, SiteId = 3 }; + var site = new Site("Site One", "site-1") + { + Id = 3, + GrpcNodeAAddress = "http://localhost:5100", + GrpcNodeBAddress = "http://localhost:5200" + }; + + var instanceRepo = Substitute.For(); + instanceRepo.GetInstanceByIdAsync(7, Arg.Any()).Returns(instance); + var siteRepo = Substitute.For(); + siteRepo.GetSiteByIdAsync(3, Arg.Any()).Returns(site); + + var services = new ServiceCollection(); + services.AddScoped(_ => instanceRepo); + services.AddScoped(_ => siteRepo); + using var provider = services.BuildServiceProvider(); + + var commProbe = CreateTestProbe(); + var commService = new CommunicationService( + Options.Create(new CommunicationOptions()), + NullLogger.Instance); + commService.SetCommunicationActor(commProbe.Ref); + + // Mock gRPC factory: the real one would dial localhost:5100, fail, and trip the + // bridge actor's retry budget — a termination unrelated to the orphan net under + // test. The mock keeps the stream "up" so the only thing that can end this + // session is the consumer-liveness decision. + using var grpcFactory = new Grpc.MockSiteStreamGrpcClientFactory( + new Grpc.MockSiteStreamGrpcClient()); + using var service = new DebugStreamService( + commService, provider, grpcFactory, NullLogger.Instance); + service.SetActorSystem(Sys); + + var startTask = service.StartStreamAsync( + instanceId: 7, onEvent: _ => { }, onTerminated: () => { }); + + commProbe.ExpectMsg(TimeSpan.FromSeconds(5)); + var bridgeActor = commProbe.LastSender; + Watch(bridgeActor); + + // Resolve the snapshot so the session is fully established. + bridgeActor.Tell(new ZB.MOM.WW.ScadaBridge.Commons.Messages.DebugView.DebugViewSnapshot( + "Site1.Pump01", + new List(), + new List(), + DateTimeOffset.UtcNow)); + var session = await startTask; + + // Several orphan windows with NO traffic of any kind — only the service's + // keepalive timer. The session must survive. + await Task.Delay(TimeSpan.FromMilliseconds(1500)); + ExpectNoMsg(TimeSpan.FromMilliseconds(100)); + Assert.False(bridgeActor.IsNobody()); + + // Detach the consumer: the session leaves the registry, keepalives stop, and the + // actor is stopped (StopStream) — the orphan net is the backstop for the case + // where that explicit stop never happens. + service.StopStream(session.SessionId); + ExpectTerminated(bridgeActor, TimeSpan.FromSeconds(3)); + } + finally + { + DebugStreamService.KeepaliveInterval = previousKeepalive; + DebugStreamBridgeActor.ConsumerIdleTimeout = previousIdle; + DebugStreamBridgeActor.ConsumerLivenessCheckInterval = previousCheck; + } + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/DebugStreamBridgeActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/DebugStreamBridgeActorTests.cs index 0d122c1f..dc652363 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/DebugStreamBridgeActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/DebugStreamBridgeActorTests.cs @@ -11,6 +11,13 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Tests.Grpc; /// /// Tests for DebugStreamBridgeActor with gRPC streaming integration. /// +/// +/// Shares the DebugStreamStatics xUnit collection with DebugStreamServiceTests: +/// both tune the actor's static test seams (SnapshotTimeout, ConsumerIdleTimeout, +/// …), and xUnit parallelizes distinct classes by default, which would let one class's +/// try/finally restore clobber the other's window mid-test. +/// +[Collection("DebugStreamStatics")] public class DebugStreamBridgeActorTests : TestKit { private const string SiteId = "site-alpha"; @@ -1034,14 +1041,11 @@ public class DebugStreamBridgeActorTests : TestKit } [Fact] - public void StreamEvents_AreWrapped_SoTheyDoNotResetTheOrphanReceiveTimeout() + public void StreamEvents_StillReachTheConsumer_ThroughTheWrappedCallbackPath() { - // Structural pin for the fix: the gRPC callback wraps every event in an envelope - // marked INotInfluenceReceiveTimeout, so a busy site can no longer keep an - // abandoned session alive indefinitely by feeding it events. - Assert.True(typeof(Akka.Actor.INotInfluenceReceiveTimeout) - .IsAssignableFrom(typeof(LiveDebugStreamEvent))); - + // The gRPC callback wraps every event in LiveDebugStreamEvent; the wrapper must be + // transparent to delivery (it exists only to keep the high-volume path explicit — + // the orphan net keys off the consumer keepalive, not the mailbox). var ctx = CreateBridgeActor(); ctx.CommProbe.ExpectMsg(); AwaitCondition(() => ctx.MockGrpcClient.SubscribeCalls.Count == 1, TimeSpan.FromSeconds(3)); @@ -1052,8 +1056,6 @@ public class DebugStreamBridgeActorTests : TestKit new List(), DateTimeOffset.UtcNow)); - // An event delivered through the real gRPC callback path still reaches the - // consumer — the wrapper is transparent to delivery. var evt = new AttributeValueChanged( InstanceName, "Modules.IO", "Temperature", 42.5, "Good", DateTimeOffset.UtcNow); ctx.MockGrpcClient.SubscribeCalls[0].OnEvent(evt); @@ -1063,6 +1065,134 @@ public class DebugStreamBridgeActorTests : TestKit lock (ctx.ReceivedEvents) { return ctx.ReceivedEvents.OfType().Any(); } }, TimeSpan.FromSeconds(3)); } + + // ----- Orphan net: measures the CONSUMER, not the mailbox ----- // + + [Fact] + public void HealthySession_StreamingEvents_WithConsumerKeepalives_SurvivesWellPastTheOrphanWindow() + { + // THE regression this closes: with the orphan net armed off the mailbox + // (SetReceiveTimeout) and stream events correctly excluded from it, nothing recurring + // reset it — the snapshot lands once, GrpcStreamStable once — so a perfectly healthy + // session self-terminated one window later and the consumer was told + // "Site disconnected". Here the session streams events and receives the keepalive + // DebugStreamService sends while it is attached; it must live through MANY windows. + DebugStreamBridgeActor.ConsumerIdleTimeout = TimeSpan.FromMilliseconds(400); + DebugStreamBridgeActor.ConsumerLivenessCheckInterval = TimeSpan.FromMilliseconds(50); + try + { + var ctx = CreateBridgeActor(); + ctx.CommProbe.ExpectMsg(); + AwaitCondition(() => ctx.MockGrpcClient.SubscribeCalls.Count == 1, TimeSpan.FromSeconds(3)); + Watch(ctx.BridgeActor); + + ctx.BridgeActor.Tell(new DebugViewSnapshot( + InstanceName, + new List(), + new List(), + DateTimeOffset.UtcNow)); + + // ~5 orphan windows of pure stream traffic + consumer keepalives, and no other + // mailbox activity whatsoever (no reconnects, no snapshots, no stop). + var deadline = DateTime.UtcNow.AddSeconds(2); + var delivered = 0; + while (DateTime.UtcNow < deadline) + { + ctx.MockGrpcClient.SubscribeCalls[0].OnEvent(new AttributeValueChanged( + InstanceName, "Modules.IO", "Temperature", 20.0 + delivered, "Good", + DateTimeOffset.UtcNow)); + delivered++; + // What DebugStreamService's shared timer does for an attached session. + ctx.BridgeActor.Tell(new DebugStreamConsumerAlive()); + Thread.Sleep(100); + } + + ExpectNoMsg(TimeSpan.FromMilliseconds(100)); + Assert.False(ctx.TerminatedFlag[0]); + + // Still serving: the events all arrived and the actor is alive. + AwaitCondition(() => + { + lock (ctx.ReceivedEvents) { return ctx.ReceivedEvents.OfType().Count() == delivered; } + }, TimeSpan.FromSeconds(3)); + Assert.False(ctx.BridgeActor.IsNobody()); + } + finally + { + DebugStreamBridgeActor.ConsumerIdleTimeout = TimeSpan.FromMinutes(5); + DebugStreamBridgeActor.ConsumerLivenessCheckInterval = TimeSpan.FromSeconds(30); + } + } + + [Fact] + public void OrphanedSession_ConsumerGone_StillTerminates_EvenWhileEventsKeepArriving() + { + // The other half of the contract: site chatter must NOT hold an abandoned session + // open. No keepalive arrives (the consumer is gone), so the session terminates, + // unsubscribes from the site and reports termination — while events stream in. + DebugStreamBridgeActor.ConsumerIdleTimeout = TimeSpan.FromMilliseconds(400); + DebugStreamBridgeActor.ConsumerLivenessCheckInterval = TimeSpan.FromMilliseconds(50); + try + { + var ctx = CreateBridgeActor(); + ctx.CommProbe.ExpectMsg(); + AwaitCondition(() => ctx.MockGrpcClient.SubscribeCalls.Count == 1, TimeSpan.FromSeconds(3)); + Watch(ctx.BridgeActor); + + ctx.BridgeActor.Tell(new DebugViewSnapshot( + InstanceName, + new List(), + new List(), + DateTimeOffset.UtcNow)); + + // Keep the site chatty for longer than the orphan window — with no keepalive. + var deadline = DateTime.UtcNow.AddSeconds(1); + while (DateTime.UtcNow < deadline) + { + ctx.MockGrpcClient.SubscribeCalls[0].OnEvent(new AttributeValueChanged( + InstanceName, "Modules.IO", "Temperature", 42.5, "Good", DateTimeOffset.UtcNow)); + Thread.Sleep(50); + } + + ExpectTerminated(ctx.BridgeActor, TimeSpan.FromSeconds(3)); + Assert.True(ctx.TerminatedFlag[0]); + Assert.Contains("corr-1", ctx.MockGrpcClient.UnsubscribedCorrelationIds); + } + finally + { + DebugStreamBridgeActor.ConsumerIdleTimeout = TimeSpan.FromMinutes(5); + DebugStreamBridgeActor.ConsumerLivenessCheckInterval = TimeSpan.FromSeconds(30); + } + } + + [Fact] + public void ConsumerKeepalive_RenewsTheWindow_AfterANearMiss() + { + // A single late keepalive is enough to save a session — the stamp is refreshed, not + // a one-shot grace period. + DebugStreamBridgeActor.ConsumerIdleTimeout = TimeSpan.FromMilliseconds(500); + DebugStreamBridgeActor.ConsumerLivenessCheckInterval = TimeSpan.FromMilliseconds(50); + try + { + var ctx = CreateBridgeActor(); + ctx.CommProbe.ExpectMsg(); + Watch(ctx.BridgeActor); + + Thread.Sleep(350); // most of the window burnt + ctx.BridgeActor.Tell(new DebugStreamConsumerAlive()); // …then the consumer checks in + Thread.Sleep(350); // past the ORIGINAL deadline + + Assert.False(ctx.TerminatedFlag[0]); + + // …and once the keepalives stop, it does terminate. + ExpectTerminated(ctx.BridgeActor, TimeSpan.FromSeconds(3)); + } + finally + { + DebugStreamBridgeActor.ConsumerIdleTimeout = TimeSpan.FromMinutes(5); + DebugStreamBridgeActor.ConsumerLivenessCheckInterval = TimeSpan.FromSeconds(30); + } + } } /// 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 d5c85507..b427520f 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs @@ -720,6 +720,99 @@ public class SiteAlarmAggregatorActorTests : TestKit AwaitCondition(() => !sink.StreamLive, TimeSpan.FromSeconds(3)); } + [Fact] + public void ConsecutiveReconcileTicks_WithNoReconnect_EachRunAFanout() + { + // The bug: a tick's OWN fan-out completion armed the skip flag, so steady state ran + // fan-out → skip → fan-out → skip — one reconcile per TWO intervals, halving both the + // not-reporting refresh and the alarm reconcile backstop. Nothing was being + // duplicated: only a connect/failover-driven seed can make a tick redundant. + var (actor, seed, _, _) = CreateActor(reconcileInterval: TimeSpan.FromMinutes(10)); + AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); + seed.CompleteNext(); // initial seed — this one DOES stand a tick down + Thread.Sleep(150); + + actor.Tell(new RunReconcile()); // consumed by the initial seed's flag + Thread.Sleep(200); + Assert.Equal(1, seed.CallCount); + + // From here on, with no reconnect in between, EVERY tick must fan out. + for (var expected = 2; expected <= 5; expected++) + { + actor.Tell(new RunReconcile()); + AwaitCondition(() => seed.CallCount == expected, TimeSpan.FromSeconds(3)); + seed.CompleteNext(); + Thread.Sleep(100); + } + + Assert.Equal(5, seed.CallCount); + } + + [Fact] + public void ConnectDrivenReseed_StillStandsTheNextTickDown() + { + // The other half of the fix: the skip exists for the connect/failover seed, and that + // suppression must survive. A reconnect's re-seed still makes the very next tick a + // no-op, so a flapping stream cannot double the whole-site snapshot rate. + var (actor, seed, _, factory) = CreateActor( + reconcileInterval: TimeSpan.FromMinutes(10), autoConnect: false); + AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); + AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3)); + factory.ClientFor(GrpcNodeA).Subs[0].OnConnected!(); + seed.CompleteNext(); + Thread.Sleep(150); + + actor.Tell(new RunReconcile()); // consumed by the initial seed + actor.Tell(new RunReconcile()); // real tick fan-out + AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(3)); + seed.CompleteNext(); + Thread.Sleep(150); + + // Stream faults → reconnect on node B → connect-driven re-seed (call 3). + factory.ClientFor(GrpcNodeA).Subs[0].OnError(new Exception("site gone")); + AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 1, TimeSpan.FromSeconds(5)); + factory.ClientFor(GrpcNodeB).Subs[0].OnConnected!(); + AwaitCondition(() => seed.CallCount == 3, TimeSpan.FromSeconds(3)); + seed.CompleteNext(); + Thread.Sleep(150); + + // That re-seed covers this window: the next tick is skipped… + actor.Tell(new RunReconcile()); + Thread.Sleep(250); + Assert.Equal(3, seed.CallCount); + + // …and the one after it fans out again. + actor.Tell(new RunReconcile()); + AwaitCondition(() => seed.CallCount == 4, TimeSpan.FromSeconds(3)); + } + + [Fact] + public void ForeignCancelledStreamError_DropsLiveness_AndReopensOnTheOtherNode() + { + // End-to-end for the SiteStreamGrpcClient fix: a peer-originated + // RpcException(Cancelled) now reaches onError instead of being swallowed. Before the + // fix NONE of onError/onCompleted/onConnected fired, so the aggregator kept + // _streamDown=false — IsLive stuck true and the reconcile tick's reopen guard, which + // only fires when the stream is known down, never ran. + var (_, seed, sink, factory) = CreateActor( + reconcileInterval: TimeSpan.FromMinutes(10), autoConnect: false); + AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); + AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3)); + factory.ClientFor(GrpcNodeA).Subs[0].OnConnected!(); + seed.CompleteNext(); + AwaitCondition(() => sink.StreamLive, TimeSpan.FromSeconds(3)); + + factory.ClientFor(GrpcNodeA).Subs[0].OnError( + new global::Grpc.Core.RpcException(new global::Grpc.Core.Status( + global::Grpc.Core.StatusCode.Cancelled, "cancelled by peer"))); + + AwaitCondition(() => !sink.StreamLive, TimeSpan.FromSeconds(3)); + // …and the stream is reopened on the other node, where a connect earns one re-seed. + AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 1, TimeSpan.FromSeconds(5)); + factory.ClientFor(GrpcNodeB).Subs[0].OnConnected!(); + AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(3)); + } + [Fact] public void FailedSeed_IsRetried_WithBackoff_BeforeTheNextReconcile() { diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientTests.cs index cb21d41b..331ce023 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientTests.cs @@ -351,9 +351,170 @@ public class SiteStreamGrpcClientTests Assert.Equal(0, completed); } + // --- Foreign vs. own Cancelled (review F2) --- + + [Fact] + public async Task ConsumeStream_ForeignCancelled_InvokesOnError() + { + // A Cancelled status we did NOT ask for — the PEER cancelled, or the channel was + // disposed underneath us. It used to be swallowed by an unguarded + // `when (ex.StatusCode == StatusCode.Cancelled)` filter, firing none of + // onError/onCompleted/onConnected: the consuming aggregator kept _streamDown=false, + // so IsLive stayed true forever and the reconcile tick's reopen guard never fired. + var client = SiteStreamGrpcClient.CreateForTesting(); + var cts = new CancellationTokenSource(); // deliberately NOT cancelled + Exception? error = null; + var completed = 0; + + await client.ConsumeStreamAsync( + "corr-foreign-cancel", + cts, + () => FakeCall(new StubStreamReader( + new RpcException(new Status(StatusCode.Cancelled, "cancelled by peer")))), + _ => { }, + ex => error = ex, + () => completed++); + + var rpc = Assert.IsType(error); + Assert.Equal(StatusCode.Cancelled, rpc.StatusCode); + Assert.Equal(0, completed); + Assert.False(cts.IsCancellationRequested); + } + + [Fact] + public async Task ConsumeStream_OwnCancelledRpcException_InvokesNeitherCallback() + { + // The other side of the same filter: OUR cancellation surfacing as + // RpcException(Cancelled) is a teardown, so neither callback fires. + var client = SiteStreamGrpcClient.CreateForTesting(); + var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + Exception? error = null; + var completed = 0; + + await client.ConsumeStreamAsync( + "corr-own-cancel", + cts, + () => throw new RpcException(new Status(StatusCode.Cancelled, "we cancelled")), + _ => { }, + ex => error = ex, + () => completed++); + + Assert.Null(error); + Assert.Equal(0, completed); + } + + // --- onConnected is only ever raised on real proof of a live peer (review F3) --- + + [Fact] + public async Task ConsumeStream_HeadersArrive_InvokesOnConnectedOnce() + { + var client = SiteStreamGrpcClient.CreateForTesting(); + var cts = new CancellationTokenSource(); + var connected = 0; + + await client.ConsumeStreamAsync( + "corr-headers", + cts, + () => FakeCall( + new StubStreamReader( + new SiteStreamEvent { CorrelationId = "corr-headers" }, + new SiteStreamEvent { CorrelationId = "corr-headers" }), + Task.FromResult(new Metadata())), + _ => { }, + _ => { }, + () => { }, + () => connected++); + + // Once — the two events must not re-raise it. + Assert.Equal(1, connected); + } + + [Fact] + public async Task ConsumeStream_HeaderTimeoutOnADeadSite_NeverReportsConnected() + { + // The bug: a header TIMEOUT used to be reported as connected. An unreachable/wedged + // site produces exactly that shape, so the aggregator cleared _streamDown, consumed + // its pending re-seed and fanned a full snapshot out at a site that never answered. + var previous = SiteStreamGrpcClient.ConnectedHeaderTimeout; + SiteStreamGrpcClient.ConnectedHeaderTimeout = TimeSpan.FromMilliseconds(50); + try + { + var client = SiteStreamGrpcClient.CreateForTesting(); + var cts = new CancellationTokenSource(); + var connected = 0; + var neverArrives = new TaskCompletionSource(); + + // No headers AND no events: the site is dead. The stream ends (status OK) with + // no connected signal ever raised. + await client.ConsumeStreamAsync( + "corr-dead", + cts, + () => FakeCall(new StubStreamReader(), neverArrives.Task), + _ => { }, + _ => { }, + () => { }, + () => connected++); + + Assert.Equal(0, connected); + } + finally + { + SiteStreamGrpcClient.ConnectedHeaderTimeout = previous; + } + } + + [Fact] + public async Task ConsumeStream_HeaderTimeoutThenEvent_ReportsConnectedOnceFromTheEvent() + { + // The case the timeout was originally added for — a peer that defers its headers + // until the first message — is now covered by the event itself, which is real proof + // of a live peer. Connected must precede the event's delivery and fire only once. + var previous = SiteStreamGrpcClient.ConnectedHeaderTimeout; + SiteStreamGrpcClient.ConnectedHeaderTimeout = TimeSpan.FromMilliseconds(50); + try + { + var client = SiteStreamGrpcClient.CreateForTesting(); + var cts = new CancellationTokenSource(); + var connected = 0; + var connectedBeforeFirstEvent = false; + var events = 0; + var neverArrives = new TaskCompletionSource(); + + await client.ConsumeStreamAsync( + "corr-late-headers", + cts, + () => FakeCall( + new StubStreamReader( + new SiteStreamEvent { CorrelationId = "corr-late-headers" }, + new SiteStreamEvent { CorrelationId = "corr-late-headers" }), + neverArrives.Task), + _ => + { + if (events == 0) connectedBeforeFirstEvent = connected == 1; + events++; + }, + _ => { }, + () => { }, + () => connected++); + + Assert.Equal(2, events); + Assert.Equal(1, connected); + Assert.True(connectedBeforeFirstEvent); + } + finally + { + SiteStreamGrpcClient.ConnectedHeaderTimeout = previous; + } + } + private static AsyncServerStreamingCall FakeCall(StubStreamReader reader) => + FakeCall(reader, Task.FromResult(new Metadata())); + + private static AsyncServerStreamingCall FakeCall( + StubStreamReader reader, Task responseHeaders) => new(reader, - Task.FromResult(new Metadata()), + responseHeaders, () => Status.DefaultSuccess, () => new Metadata(), () => { }); diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteStreamPullAuditEventsTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteStreamPullAuditEventsTests.cs index ee3e3ad2..82c36243 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteStreamPullAuditEventsTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteStreamPullAuditEventsTests.cs @@ -243,6 +243,45 @@ public class SiteStreamPullAuditEventsTests : TestKit Assert.True(response.MoreAvailable); } + [Fact] + public async Task PullAuditEvents_RetiresBeforeItServes_SoThisBatchIsNeverSelfRetired() + { + // Ordering is load-bearing for the "only served rows retire" invariant. The queue + // bounds the cursor flip by insertion order — the high-water mark of rows it has + // SERVED — so the retire step must run BEFORE the read: retiring first can only ever + // reach rows served by an EARLIER pull, never the ones this call is about to serve + // (which would defeat at-least-once), and never a row inserted after them (the + // late-stamped insert that used to be silently retired and then age-purged). + var queue = Substitute.For(); + queue.ReadPendingSinceAsync( + Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns((IReadOnlyList)new[] { NewEvent() }); + + var server = CreateServer(); + server.SetSiteAuditQueue(queue); + + var cursorTime = DateTime.SpecifyKind(new DateTime(2026, 5, 20, 10, 0, 0), DateTimeKind.Utc); + await server.PullAuditEvents( + new PullAuditEventsRequest + { + SinceUtc = Timestamp.FromDateTime(cursorTime), + BatchSize = 100, + }, + NewContext()); + + Received.InOrder(() => + { + queue.MarkReconciledUpToAsync(cursorTime, null, Arg.Any()); + queue.ReadPendingSinceAsync(cursorTime, 100, null, Arg.Any()); + }); + + // Serving still flips nothing by itself — the next cursor is the only receipt. + await queue.DidNotReceive().MarkReconciledAsync( + Arg.Any>(), Arg.Any()); + await queue.DidNotReceive().MarkForwardedAsync( + Arg.Any>(), Arg.Any()); + } + [Fact] public async Task PullAuditEvents_MarkReconciledUpToThrows_ResponseStillReturned() {