diff --git a/docs/requirements/Component-AuditLog.md b/docs/requirements/Component-AuditLog.md index bcc4a850..b9364d22 100644 --- a/docs/requirements/Component-AuditLog.md +++ b/docs/requirements/Component-AuditLog.md @@ -343,21 +343,29 @@ mapping/unexpected fault collapses to empty without a second dial (the other node would hit the same fault). The same resolution + failover applies to Site Call Audit's `PullSiteCalls`. -**Cursor keyset.** The site side of the composite `(OccurredAtUtc, EventId)` -keyset now exists (WP2.3): `PullAuditEventsRequest.after_id` (additive field 3) -mirrors `PullSiteCallsRequest.after_id`, and `ISiteAuditQueue.ReadPendingSinceAsync` -switches from the inclusive `OccurredAtUtc >= since` to a strict composite -comparison when it is set, so a burst sharing one instant drains via the id -tiebreak instead of pinning the cursor. +**Cursor keyset.** The composite `(OccurredAtUtc, EventId)` keyset is wired end to +end. `PullAuditEventsRequest.after_id` (additive field 3) mirrors +`PullSiteCallsRequest.after_id`; `ISiteAuditQueue.ReadPendingSinceAsync` switches +from the inclusive `OccurredAtUtc >= since` to a strict composite comparison when it +is set, so a burst sharing one instant drains via the id tiebreak instead of pinning +the cursor; and `IPullAuditEventsClient` /`GrpcPullAuditEventsClient` / +`SiteAuditReconciliationActor` populate it — the actor's per-site watermark is the +`(timestamp, id)` pair of the highest row ingested, threaded back into the next pull, +mirroring `SiteCallAuditActor`'s `PullSiteCalls` cursor exactly. -> **Central still sends a bare timestamp (tracked follow-up).** -> `IPullAuditEventsClient`/`SiteAuditReconciliationActor` do not yet populate -> `after_id`, so today's cursor remains a single `sinceUtc` under the legacy -> inclusive read. Two consequences, both benign: a window whose rows all share one -> instant re-serves rather than advancing (idempotent on `EventId`, as before), and -> the rows **at** the cursor instant cannot be proven received, so they stay -> servable until a newer row moves the cursor. Wiring `after_id` at central makes -> the retirement exact; the site accepts it already. +That makes retirement **exact**: `MarkReconciledUpToAsync` is handed the id half, so +the rows **at** the cursor instant are proven received and retired, where a bare +timestamp could only prove receipt of rows strictly older than itself and left the +boundary rows servable until a newer row moved the cursor. `after_id` is null only on +the first pull for a site (and against a site that predates the field), which keeps +the legacy inclusive `>=` read as the compatible fallback. The cursor stays in-memory +and resets on singleton restart; idempotent `InsertIfNotExistsAsync` still absorbs any +re-pulled duplicates, so exactness is an optimization, not a correctness dependency. + +Unlike Site Call Audit, this actor issues ONE pull per tick rather than paging within +it — a lagging drain is meant to surface as the stalled signal. The id tiebreak is +what makes that safe against a same-instant burst larger than one batch: the cursor +advances on every tick even when the timestamp cannot. ### Central direct-write (central-originated events) diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogIngestActor.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogIngestActor.cs index e7b3f9a0..63eab907 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogIngestActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogIngestActor.cs @@ -58,15 +58,21 @@ public class AuditLogIngestActor : ReceiveActor /// SHORTER than the gRPC Ask that wraps it. /// /// - /// The path used to stack three identical 30 s budgets — the site's Ask - /// (CommunicationOptions.NotificationForwardTimeout), the central gRPC - /// handler's Ask (SiteStreamGrpcServer.AuditIngestAskTimeout) and the - /// ADO.NET command default — so they all expired at the same instant. The + /// The path used to stack three identical 30 s budgets — the site's Ask, the + /// central gRPC handler's Ask (SiteStreamGrpcServer.AuditIngestAskTimeout) + /// and the ADO.NET command default — so they all expired at the same instant. The /// caller therefore learned nothing except "it took 30 s": no partial ack, no /// distinction between a slow database and a wedged singleton. Making the /// innermost budget strictly smallest means a slow batch is abandoned by the /// actor FIRST, with the accepted-so-far ids still replied, while the outer /// Asks are still waiting. + /// + /// The full ladder is now strictly monotonic end to end: + /// CommunicationOptions.AuditForwardTimeout (35 s, the site-side forward Ask) > + /// SiteStreamGrpcServer.AuditIngestAskTimeout (30 s, the gRPC deadline AND central's + /// Ask of this singleton) > (20 s) > + /// (15 s). + /// /// internal static readonly TimeSpan IngestBudget = TimeSpan.FromSeconds(20); diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullAuditEventsClient.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullAuditEventsClient.cs index 7001f9df..3020e57f 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullAuditEventsClient.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullAuditEventsClient.cs @@ -73,6 +73,7 @@ public sealed class GrpcPullAuditEventsClient : IPullAuditEventsClient public async Task PullAsync( string siteId, DateTime sinceUtc, + string? afterId, int batchSize, CancellationToken ct) { @@ -93,6 +94,10 @@ public sealed class GrpcPullAuditEventsClient : IPullAuditEventsClient // EnsureUtc keeps Timestamp.FromDateTime happy (it requires UTC kind). SinceUtc = Timestamp.FromDateTime(EnsureUtc(sinceUtc)), BatchSize = batchSize, + // Composite-keyset tiebreak (proto field 3), mirroring PullSiteCalls exactly. + // proto3 has no nullable string — an unset/empty AfterId is the site's signal to + // keep the legacy inclusive-timestamp contract (also what a first pull sends). + AfterId = afterId ?? string.Empty, }; var (reply, transportFault) = await TryInvokeAsync(endpoint, request, siteId, ct) @@ -121,10 +126,13 @@ public sealed class GrpcPullAuditEventsClient : IPullAuditEventsClient // Map proto DTOs to canonical AuditEvent records and order oldest-first // (the wire is already ordered by the site queue, but the - // IPullAuditEventsClient contract is explicit, so sort defensively). + // IPullAuditEventsClient contract is explicit, so sort defensively). The EventId + // tiebreak matches the site's own composite ordering — ordinal over the "D" GUID + // text, which is what SQLite's BINARY collation compares. var events = reply.Events .Select(AuditEventDtoMapper.FromDto) .OrderBy(e => e.OccurredAtUtc) + .ThenBy(e => e.EventId.ToString(), StringComparer.Ordinal) .ToList(); return new PullAuditEventsResponse(events, reply.MoreAvailable); diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/IPullAuditEventsClient.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/IPullAuditEventsClient.cs index 52ebf083..e04357c7 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/IPullAuditEventsClient.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/IPullAuditEventsClient.cs @@ -39,12 +39,24 @@ public interface IPullAuditEventsClient /// /// The identifier of the site to pull audit events from. /// Only events with an OccurredAtUtc at or after this cursor time are returned. + /// + /// The composite-keyset tiebreak cursor, mirroring + /// . When non-null it is the + /// EventId ("D" GUID form) of the last row already consumed at + /// ; the site returns only rows strictly greater than the + /// composite (OccurredAtUtc, EventId) pair, so a burst sharing one exact instant + /// drains via the id tiebreak instead of pinning the inclusive-timestamp cursor — and the + /// site's MarkReconciledUpToAsync can retire the rows AT that instant, which a bare + /// timestamp can never prove received. Null on the first pull (or against a legacy site) + /// preserves the inclusive >= contract. + /// /// Maximum number of events to return per call. /// Cancellation token. /// A task that resolves to the next reconciliation batch with a MoreAvailable flag. Task PullAsync( string siteId, DateTime sinceUtc, + string? afterId, int batchSize, CancellationToken ct); } diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs index 11423553..a68f7ee5 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs @@ -76,14 +76,28 @@ public class SiteAuditReconciliationActor : ReceiveActor private readonly ILogger _logger; /// - /// Per-site reconciliation watermark — the highest - /// seen for that site on a previous - /// tick. Asking for OccurredAtUtc >= cursor rather than > - /// is the site contract (); - /// duplicate-with-same-timestamp rows are filtered out by the idempotent - /// repository write. + /// Per-site reconciliation watermark — the COMPOSITE + /// (OccurredAtUtc, EventId) of the highest row seen for that site on a previous tick, + /// mirroring SiteCallAuditActor's PullSiteCalls cursor. /// - private readonly Dictionary _cursors = new(); + /// + /// + /// The site's + /// + /// serves rows strictly after the pair when AfterId is set, and falls back to the + /// legacy inclusive OccurredAtUtc >= since when it is null (the first pull). + /// Sending the id half matters twice over: a burst sharing one exact instant drains via the + /// tiebreak instead of pinning the timestamp forever, and + /// + /// can retire the rows AT the cursor instant — a bare timestamp can only prove receipt of + /// rows strictly older than itself, so the boundary rows stayed servable forever. + /// + /// + /// Duplicates that a re-pull does produce are still filtered by the idempotent repository + /// write, so the cursor is an optimization for exactness, never a correctness crutch. + /// + /// + private readonly Dictionary _cursors = new(); /// /// Per-site count of consecutive non-draining cycles. Resets to zero on the @@ -239,20 +253,33 @@ public class SiteAuditReconciliationActor : ReceiveActor /// /// Issues one PullAuditEvents RPC against the site, ingests the /// returned rows idempotently into the central repository, and advances - /// the cursor based on the maximum - /// observed. The brief's "saturate until backlog clears" intent is met by - /// the natural cadence — each tick issues one pull, and a backed-up site + /// the composite (, ) + /// cursor to the maximum row observed. The brief's "saturate until backlog clears" intent is + /// met by the natural cadence — each tick issues one pull, and a backed-up site /// drains across consecutive ticks. The stalled signal (two non-draining /// ticks in a row) surfaces when that drain isn't keeping up. /// + /// + /// Unlike SiteCallAuditActor this does NOT page within a tick; one pull per tick is + /// deliberate (the stalled signal is how a lagging drain surfaces). The composite cursor is + /// what makes that safe against a same-instant burst larger than one batch: the id tiebreak + /// advances even when the timestamp cannot, so consecutive ticks make real progress instead + /// of re-serving the same window forever. + /// private async Task PullSiteAsync(SiteEntry site, IAuditLogRepository repository, Akka.Event.EventStream eventStream) { - var since = _cursors.TryGetValue(site.SiteId, out var c) ? c : DateTime.MinValue; + var cursor = _cursors.TryGetValue(site.SiteId, out var c) + ? c + : (Since: DateTime.MinValue, AfterId: (string?)null); + var since = cursor.Since; + var afterId = cursor.AfterId; + var response = await _client.PullAsync( - site.SiteId, since, _options.BatchSize, CancellationToken.None) + site.SiteId, since, afterId, _options.BatchSize, CancellationToken.None) .ConfigureAwait(false); var maxOccurred = since; + var maxAfterId = afterId; var hasUnresolvedFailure = false; var nowUtc = DateTime.UtcNow; foreach (var evt in response.Events) @@ -310,10 +337,18 @@ public class SiteAuditReconciliationActor : ReceiveActor } // Canonical OccurredAtUtc is a DateTimeOffset; the cursor is a UTC DateTime. + // Advance the COMPOSITE max, exactly as SiteCallAuditActor does: a greater + // timestamp wins; on a tie the greater EventId (ordinal over the "D" GUID text, + // matching the site's SQLite BINARY collation) wins. CompareOrdinal handles the + // null seed — CompareOrdinal(x, null) > 0 for any non-null x. var occurredUtc = evt.OccurredAtUtc.UtcDateTime; - if (advanceForThisRow && occurredUtc > maxOccurred) + var rowId = evt.EventId.ToString(); + if (advanceForThisRow && + (occurredUtc > maxOccurred || + (occurredUtc == maxOccurred && string.CompareOrdinal(rowId, maxAfterId) > 0))) { maxOccurred = occurredUtc; + maxAfterId = rowId; } } @@ -322,8 +357,11 @@ public class SiteAuditReconciliationActor : ReceiveActor // the whole batch next tick — successful rows are no-ops thanks to // InsertIfNotExistsAsync's idempotency, and the failing row gets // another attempt. Once it succeeds (or hits the permanent-abandon - // threshold) the cursor unblocks naturally. - _cursors[site.SiteId] = hasUnresolvedFailure ? since : maxOccurred; + // threshold) the cursor unblocks naturally. Both halves move together: holding the + // timestamp back while advancing the id would skip the very rows being retried. + _cursors[site.SiteId] = hasUnresolvedFailure + ? (since, afterId) + : (maxOccurred, maxAfterId); var nonDraining = response.MoreAvailable && response.Events.Count > 0; UpdateStalledState(site.SiteId, draining: !nonDraining, eventStream); diff --git a/src/ZB.MOM.WW.ScadaBridge.CentralUI/Components/Pages/Monitoring/AlarmSummary.razor b/src/ZB.MOM.WW.ScadaBridge.CentralUI/Components/Pages/Monitoring/AlarmSummary.razor index 6571d165..284486e2 100644 --- a/src/ZB.MOM.WW.ScadaBridge.CentralUI/Components/Pages/Monitoring/AlarmSummary.razor +++ b/src/ZB.MOM.WW.ScadaBridge.CentralUI/Components/Pages/Monitoring/AlarmSummary.razor @@ -331,8 +331,10 @@ return; } - // _notReporting is the poll's unique authority — the alarm-only live cache - // cannot compute it — so it is always refreshed. + // _notReporting always comes from this call, live or not. While the cache is + // serving the site the façade sources it from the aggregator's own fan-out (no + // second fan-out is run); while it is cold the poll computes it. Either way the + // shape and ordering are identical. _notReporting = result.NotReportingInstances; // While the cache is live, the live deltas own the row set: a poll whose fan-out diff --git a/src/ZB.MOM.WW.ScadaBridge.CentralUI/Services/AlarmSummaryService.cs b/src/ZB.MOM.WW.ScadaBridge.CentralUI/Services/AlarmSummaryService.cs index 6d22d117..1d33fc82 100644 --- a/src/ZB.MOM.WW.ScadaBridge.CentralUI/Services/AlarmSummaryService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.CentralUI/Services/AlarmSummaryService.cs @@ -154,9 +154,11 @@ public sealed class AlarmSummaryService : IAlarmSummaryService .ThenBy(r => r.Alarm.AlarmName, StringComparer.OrdinalIgnoreCase) .ToList(); - // The live cache is alarm-only, so it cannot enumerate "not reporting" - // instances — left empty here; the periodic poll (GetSiteAlarmsAsync) - // remains the authority for that list. + // Alarm rows alone cannot name the instances that failed to answer, so this pure + // flattening leaves the list empty. The names come from the fan-out that produced the + // snapshot: while the live cache is serving a site, SharedAlarmSummaryService pairs + // these rows with ISiteAlarmLiveCache.GetNotReportingInstances (the aggregator's own + // seed/reconcile result); while it is cold, GetSiteAlarmsAsync's poll supplies both. return new AlarmSummaryResult(orderedRows, Array.Empty()); } diff --git a/src/ZB.MOM.WW.ScadaBridge.CentralUI/Services/SharedAlarmSummaryService.cs b/src/ZB.MOM.WW.ScadaBridge.CentralUI/Services/SharedAlarmSummaryService.cs index 29a6359e..3457ded2 100644 --- a/src/ZB.MOM.WW.ScadaBridge.CentralUI/Services/SharedAlarmSummaryService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.CentralUI/Services/SharedAlarmSummaryService.cs @@ -1,6 +1,5 @@ using System.Collections.Concurrent; using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.Options; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming; using ZB.MOM.WW.ScadaBridge.Communication; @@ -18,12 +17,25 @@ namespace ZB.MOM.WW.ScadaBridge.CentralUI.Services; /// viewers are watching. /// /// -/// The freshness window follows the live cache. While -/// is true the page deliberately ignores the poll's -/// alarm rows (the live deltas own them — arch-review R2 N5), so the only thing the poll -/// still supplies is the not-reporting list, and the window widens to the aggregator's own -/// reconcile interval. While the cache is cold the poll is the page's full-rebuild safety -/// net, so the window stays just under the page's 15s tick and every tick gets fresh data. +/// While the live cache is serving the site there is NO fan-out at all. The page already +/// ignores the poll's alarm rows in that state (the live deltas own them — arch-review R2 N5), +/// and the aggregator's own seed/reconcile fan-out now publishes the not-reporting set through +/// — the one thing the poll used to +/// still supply. So the live path is answered entirely from the cache: rows flattened from +/// through the same +/// AlarmSummaryService.BuildFromLiveAlarmsCore the page's live subscription uses, plus the +/// cached not-reporting names. One fan-out per site per reconcile window, run by the aggregator, +/// instead of two. +/// +/// +/// The rows are returned rather than left empty on purpose: liveness can flip to false in the +/// window between this call and the page's own IsLive re-check, and the page would then +/// adopt whatever came back. Handing it the live snapshot makes that race land on last-known +/// state instead of a blank grid. +/// +/// +/// While the cache is cold the poll is the page's full-rebuild safety net, so it still fans out, +/// memoized with a window just under the page's 15s tick so every tick gets fresh data. /// /// public sealed class SharedAlarmSummaryService : IAlarmSummaryService @@ -36,7 +48,6 @@ public sealed class SharedAlarmSummaryService : IAlarmSummaryService private readonly IServiceScopeFactory _scopeFactory; private readonly ISiteAlarmLiveCache _liveCache; - private readonly TimeSpan _liveCacheTtl; private readonly Func? _clock; private readonly ConcurrentDictionary> _bySite = new(); @@ -44,36 +55,34 @@ public sealed class SharedAlarmSummaryService : IAlarmSummaryService /// /// Initializes the shared alarm summary façade. /// + /// + /// There is no live-cache freshness window any more: the live path reads the aggregator's + /// cache directly and never fans out, so the only window left is . + /// The reconcile interval this ctor used to read from CommunicationOptions is now + /// implicit — it IS the cadence at which the aggregator refreshes what the live path returns. + /// /// Opens a fresh DI scope per fan-out (fresh repositories, off any circuit scope). - /// The shared live alarm cache, consulted only for its per-site liveness. - /// Communication options; supplies the aggregator reconcile interval. + /// The shared live alarm cache: liveness, alarm rows, and not-reporting names. public SharedAlarmSummaryService( IServiceScopeFactory scopeFactory, - ISiteAlarmLiveCache liveCache, - IOptions options) - : this(scopeFactory, liveCache, (options ?? throw new ArgumentNullException(nameof(options))) - .Value.LiveAlarmCacheReconcileInterval, clock: null) + ISiteAlarmLiveCache liveCache) + : this(scopeFactory, liveCache, clock: null) { } /// - /// Test seam: same façade with an explicit live-cache window and clock. + /// Test seam: same façade with an explicit clock. /// /// Opens a fresh DI scope per fan-out. /// The shared live alarm cache. - /// Freshness window used while the live cache is serving the site. /// Clock used for freshness. internal SharedAlarmSummaryService( IServiceScopeFactory scopeFactory, ISiteAlarmLiveCache liveCache, - TimeSpan liveCacheTtl, Func? clock) { _scopeFactory = scopeFactory ?? throw new ArgumentNullException(nameof(scopeFactory)); _liveCache = liveCache ?? throw new ArgumentNullException(nameof(liveCache)); - // Never shorter than the cold window — a misconfigured reconcile interval must not - // silently make the memo useless. - _liveCacheTtl = liveCacheTtl > ColdCacheTtl ? liveCacheTtl : ColdCacheTtl; _clock = clock; } @@ -81,17 +90,25 @@ public sealed class SharedAlarmSummaryService : IAlarmSummaryService public Task GetSiteAlarmsAsync( int siteId, CancellationToken cancellationToken = default) { + // Live: answer from the shared aggregator's cache and skip the fan-out entirely. Not + // memoized — there is nothing to amortize, and reading the cache directly keeps the + // answer as fresh as the aggregator itself. + if (_liveCache.IsLive(siteId)) + { + var live = AlarmSummaryService.BuildFromLiveAlarmsCore(_liveCache.GetCurrentAlarms(siteId)); + return Task.FromResult(new AlarmSummaryResult( + live.Alarms, _liveCache.GetNotReportingInstances(siteId))); + } + var memo = _bySite.GetOrAdd( siteId, _ => new SingleFlightMemo(ColdCacheTtl, _clock)); - var ttl = _liveCache.IsLive(siteId) ? _liveCacheTtl : ColdCacheTtl; - return memo.GetAsync( () => FanOutAsync(siteId), forceRefresh: false, cancellationToken, - ttlOverride: ttl); + ttlOverride: ColdCacheTtl); } /// diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Health/HeartbeatMessage.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Health/HeartbeatMessage.cs index 893e4718..7407c909 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Health/HeartbeatMessage.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Health/HeartbeatMessage.cs @@ -1,7 +1,27 @@ namespace ZB.MOM.WW.ScadaBridge.Commons.Messages.Health; +/// +/// Application-level heartbeat a site node sends to central on its heartbeat interval. +/// Central's health aggregator stamps the site's last-seen time from it, and the offline +/// timeout fires when they stop landing. +/// +/// The reporting site's identifier. +/// The reporting node's hostname. +/// Whether the reporting node is the site's active (oldest-Up) node. +/// When the heartbeat was produced (UTC). +/// +/// true marks a heartbeat that carries NO liveness meaning and that central must skip for +/// all health bookkeeping — it neither stamps a last-seen time nor replicates to the peer node. +/// The only producer today is CentralChannelProvider's failback probe, which reuses the +/// heartbeat RPC purely to ask "does the preferred central endpoint answer again?"; it is emitted +/// by a SITE's transport layer, not by any node's heartbeat timer, and counting it as liveness +/// would keep a site looking alive on the health dashboard from the probe alone. Additive with a +/// false default (and proto field 5 on HeartbeatDto), so every real heartbeat — +/// including one from a node that predates the flag — is unaffected. +/// public record HeartbeatMessage( string SiteId, string NodeHostname, bool IsActive, - DateTimeOffset Timestamp); + DateTimeOffset Timestamp, + bool Synthetic = false); diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs index b2cbc71f..ca63ca68 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs @@ -67,36 +67,6 @@ public class CentralCommunicationActor : ReceiveActor /// private IActorRef? _notificationOutboxProxy; - /// - /// Proxy for the central AuditLogIngestActor cluster - /// singleton. Set via — the Host creates the - /// singleton proxy after this actor and registers it (mirrors - /// ). Null until registration completes; - /// an audit ingest command arriving before then is answered with an empty - /// reply so the site keeps its rows Pending and retries. - /// - /// Once registered, the handler Asks this proxy and pipes the reply straight - /// back to the caller. On an Ask timeout or a faulted reply, PipeTo forwards a - /// to the caller — the fault propagates rather - /// than being swallowed. This differs from the gRPC handler - /// (SiteStreamGrpcServer), which catches the exception and returns an - /// empty ack; here the faulted Ask is the transient signal the site relies on - /// (see ). - /// - private IActorRef? _auditIngestProxy; - - /// - /// Default Ask timeout for routing audit ingest commands to the - /// Effective Ask timeout for audit ingest routing. Defaults to - /// (30 s) — the two - /// audit-ingest entry points (the site stream server and the control plane) share one source of truth - /// for the timeout. Overridable via the constructor so tests can exercise the - /// timeout/fault path without waiting 30 s. When the window is exceeded the Ask - /// faults and that fault is piped back to the caller as a - /// (see ). - /// - private readonly TimeSpan _auditIngestAskTimeout; - /// /// DistributedPubSub topic used to fan health reports out to the peer /// central node so both per-node aggregators stay in sync. See @@ -111,12 +81,10 @@ public class CentralCommunicationActor : ReceiveActor /// /// DI service provider for scoped repository and aggregator access. /// The central→site command transport to route every through. - /// Optional override for the audit-ingest Ask timeout (test hook). public CentralCommunicationActor( IServiceProvider serviceProvider, - ISiteCommandTransport transport, - TimeSpan? auditIngestAskTimeout = null) - : this(serviceProvider, auditIngestAskTimeout) + ISiteCommandTransport transport) + : this(serviceProvider) { _transport = transport ?? throw new ArgumentNullException(nameof(transport)); } @@ -125,13 +93,9 @@ public class CentralCommunicationActor : ReceiveActor /// is assigned by the delegating public constructor before any message /// is dispatched. /// DI service provider. - /// Optional audit-ingest Ask timeout override. - private CentralCommunicationActor( - IServiceProvider serviceProvider, - TimeSpan? auditIngestAskTimeout) + private CentralCommunicationActor(IServiceProvider serviceProvider) { _serviceProvider = serviceProvider; - _auditIngestAskTimeout = auditIngestAskTimeout ?? Grpc.SiteStreamGrpcServer.AuditIngestAskTimeout; // Site address cache loaded from database Receive(HandleSiteAddressCacheLoaded); @@ -176,24 +140,11 @@ public class CentralCommunicationActor : ReceiveActor // so the NotificationStatusResponse routes back to the querying site. Receive(HandleNotificationStatusQuery); - // Audit Log: the Host registers the AuditLogIngestActor singleton - // proxy after this actor is created (the proxy cannot exist before this - // actor's construction). - Receive(msg => - { - _auditIngestProxy = msg.AuditIngestActor; - _log.Info("Registered audit ingest proxy"); - }); - - // Audit Log site→central ingest: a site forwards a batch of audit - // events to the central cluster. Ask the ingest proxy - // and pipe the IngestAuditEventsReply back to the original Sender (the - // site's transport path) so the site can flip its rows to Forwarded. - Receive(HandleIngestAuditEvents); - - // Audit Log combined-telemetry ingest: routes to the same proxy - // the same way; the proxy replies with an IngestCachedTelemetryReply. - Receive(HandleIngestCachedTelemetry); + // Audit Log site→central ingest is NOT relayed here. The central-hosted + // CentralControlGrpcService Asks the audit-log-ingest singleton proxy directly + // (as SiteStreamGrpcServer always has), so this actor no longer sits in that + // path — the relay was a second hop with an identical 30 s Ask timeout that + // could only add latency. Do not reintroduce it. // Startup reconciliation: a site node forwards its local deployed inventory on // startup. Resolve the scoped ReconcileService, diff the @@ -239,51 +190,6 @@ public class CentralCommunicationActor : ReceiveActor _notificationOutboxProxy.Forward(msg); } - private void HandleIngestAuditEvents(IngestAuditEventsCommand msg) - { - if (_auditIngestProxy == null) - { - // No ingest proxy registered yet (host startup race). Reply with an - // empty IngestAuditEventsReply so the site keeps its rows Pending and - // retries — the same behaviour as the gRPC handler's wiring-race path. - _log.Warning( - "Cannot route IngestAuditEventsCommand ({0} events) — audit ingest not available", - msg.Events.Count); - Sender.Tell(new IngestAuditEventsReply(Array.Empty())); - return; - } - - // Capture Sender before the async/PipeTo — Akka resets Sender between - // dispatches. The reply is piped straight back to the calling site node. - // On an Ask timeout or a faulted reply, PipeTo delivers a Status.Failure to - // replyTo: the fault propagates to the caller rather than being swallowed. - // The site's own Ask through this path then faults, and the site drain loop - // treats that as a transient failure — rows stay Pending and are retried on - // the next tick. (The gRPC handler instead returns an empty ack on fault; - // propagating the fault here is the cleaner transient signal.) - var replyTo = Sender; - _log.Debug("Routing IngestAuditEventsCommand ({0} events) to the audit ingest actor", msg.Events.Count); - _auditIngestProxy.Ask(msg, _auditIngestAskTimeout) - .PipeTo(replyTo); - } - - private void HandleIngestCachedTelemetry(IngestCachedTelemetryCommand msg) - { - if (_auditIngestProxy == null) - { - _log.Warning( - "Cannot route IngestCachedTelemetryCommand ({0} entries) — audit ingest not available", - msg.Entries.Count); - Sender.Tell(new IngestCachedTelemetryReply(Array.Empty())); - return; - } - - var replyTo = Sender; - _log.Debug("Routing IngestCachedTelemetryCommand ({0} entries) to the audit ingest actor", msg.Entries.Count); - _auditIngestProxy.Ask(msg, _auditIngestAskTimeout) - .PipeTo(replyTo); - } - /// /// Startup reconciliation (site→central): resolve the scoped /// in a DI scope, diff the node's reported inventory @@ -291,7 +197,7 @@ public class CentralCommunicationActor : ReceiveActor /// back to the site node's transport path. The actor stays thin — all the diff /// and staging logic lives in the service. Mirrors the DB-access pattern used by /// (Task.Run + CreateScope + PipeTo) and the - /// Sender-preservation pattern of . + /// Sender-preservation pattern of . /// /// On a faulted task PipeTo delivers a to the node; its /// Ask faults and it simply retries reconcile on the next startup — reconcile is @@ -328,6 +234,21 @@ public class CentralCommunicationActor : ReceiveActor private void HandleHeartbeat(HeartbeatMessage heartbeat) { + // Synthetic heartbeats carry no liveness meaning — today only + // CentralChannelProvider's failback probe, which reuses this RPC to ask whether the + // preferred central endpoint answers again. It is emitted by a site's TRANSPORT layer, + // not by any node's heartbeat timer, so stamping it would let a site whose real + // heartbeats had stopped keep looking alive on the health dashboard for as long as its + // transport kept probing. Dropped before the local mark AND before the peer fan-out, + // so neither central node's aggregator ever sees it. + if (heartbeat.Synthetic) + { + _log.Debug( + "Ignoring synthetic heartbeat from site {0} (node '{1}') — probe traffic, not liveness", + heartbeat.SiteId, heartbeat.NodeHostname); + return; + } + MarkHeartbeatLocally(heartbeat); // Fan the heartbeat out to the peer central node so BOTH aggregators mark @@ -350,9 +271,17 @@ public class CentralCommunicationActor : ReceiveActor /// /// Marks a site heartbeat on the local aggregator without re-broadcasting. /// Used for both site-originated heartbeats and peer-replicated ones. + /// A synthetic heartbeat is never marked — already drops it + /// before the fan-out, so a replica can only carry one if a peer node predates the flag; + /// the guard here is belt-and-braces on the last hop before the aggregator. /// private void MarkHeartbeatLocally(HeartbeatMessage heartbeat) { + if (heartbeat.Synthetic) + { + return; + } + var aggregator = _serviceProvider.GetService(); aggregator?.MarkHeartbeat(heartbeat.SiteId, heartbeat.Timestamp); } @@ -623,13 +552,7 @@ public record DebugStreamTerminated(string SiteId, string CorrelationId); /// public record RegisterNotificationOutbox(IActorRef OutboxProxy); -/// -/// Registers the central AuditLogIngestActor singleton proxy with the -/// so site-forwarded -/// and -/// messages can be routed to it. Sent by the Host after the audit-ingest -/// singleton proxy is created. Lives here (not in Commons) because -/// ZB.MOM.WW.ScadaBridge.Commons has no Akka package reference and cannot hold an -/// field. -/// -public sealed record RegisterAuditIngest(IActorRef AuditIngestActor); +// NOTE: there is deliberately no RegisterAuditIngest counterpart. The audit-log-ingest +// singleton proxy goes STRAIGHT to the two gRPC servers that need it +// (CentralControlGrpcService.SetAuditIngestActor / SiteStreamGrpcServer.SetAuditIngestActor); +// routing audit batches through this actor was a redundant hop and was removed. diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/CentralControlGrpc/CentralControl.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/CentralControlGrpc/CentralControl.cs index e8675699..f40a9138 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/CentralControlGrpc/CentralControl.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/CentralControlGrpc/CentralControl.cs @@ -114,33 +114,34 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { "c0R0bzoCOAEaQgogU3RvcmVBbmRGb3J3YXJkQnVmZmVyRGVwdGhzRW50cnkS", "CwoDa2V5GAEgASgJEg0KBXZhbHVlGAIgASgFOgI4ASJjChZTaXRlSGVhbHRo", "UmVwb3J0QWNrRHRvEg8KB3NpdGVfaWQYASABKAkSFwoPc2VxdWVuY2VfbnVt", - "YmVyGAIgASgDEhAKCGFjY2VwdGVkGAMgASgIEg0KBWVycm9yGAQgASgJIngK", - "DEhlYXJ0YmVhdER0bxIPCgdzaXRlX2lkGAEgASgJEhUKDW5vZGVfaG9zdG5h", - "bWUYAiABKAkSEQoJaXNfYWN0aXZlGAMgASgIEi0KCXRpbWVzdGFtcBgEIAEo", - "CzIaLmdvb2dsZS5wcm90b2J1Zi5UaW1lc3RhbXAqvQEKFENvbm5lY3Rpb25I", - "ZWFsdGhFbnVtEiEKHUNPTk5FQ1RJT05fSEVBTFRIX1VOU1BFQ0lGSUVEEAAS", - "HwobQ09OTkVDVElPTl9IRUFMVEhfQ09OTkVDVEVEEAESIgoeQ09OTkVDVElP", - "Tl9IRUFMVEhfRElTQ09OTkVDVEVEEAISIAocQ09OTkVDVElPTl9IRUFMVEhf", - "Q09OTkVDVElORxADEhsKF0NPTk5FQ1RJT05fSEVBTFRIX0VSUk9SEAQyoQYK", - "FUNlbnRyYWxDb250cm9sU2VydmljZRKDAQoSU3VibWl0Tm90aWZpY2F0aW9u", - "EjQuc2NhZGFicmlkZ2UuY2VudHJhbGNvbnRyb2wudjEuTm90aWZpY2F0aW9u", - "U3VibWl0RHRvGjcuc2NhZGFicmlkZ2UuY2VudHJhbGNvbnRyb2wudjEuTm90", - "aWZpY2F0aW9uU3VibWl0QWNrRHRvEpIBChdRdWVyeU5vdGlmaWNhdGlvblN0", - "YXR1cxI5LnNjYWRhYnJpZGdlLmNlbnRyYWxjb250cm9sLnYxLk5vdGlmaWNh", - "dGlvblN0YXR1c1F1ZXJ5RHRvGjwuc2NhZGFicmlkZ2UuY2VudHJhbGNvbnRy", - "b2wudjEuTm90aWZpY2F0aW9uU3RhdHVzUmVzcG9uc2VEdG8SRwoRSW5nZXN0", - "QXVkaXRFdmVudHMSGy5zaXRlc3RyZWFtLkF1ZGl0RXZlbnRCYXRjaBoVLnNp", - "dGVzdHJlYW0uSW5nZXN0QWNrElAKFUluZ2VzdENhY2hlZFRlbGVtZXRyeRIg", - "LnNpdGVzdHJlYW0uQ2FjaGVkVGVsZW1ldHJ5QmF0Y2gaFS5zaXRlc3RyZWFt", - "LkluZ2VzdEFjaxKAAQoNUmVjb25jaWxlU2l0ZRI2LnNjYWRhYnJpZGdlLmNl", - "bnRyYWxjb250cm9sLnYxLlJlY29uY2lsZVNpdGVSZXF1ZXN0RHRvGjcuc2Nh", - "ZGFicmlkZ2UuY2VudHJhbGNvbnRyb2wudjEuUmVjb25jaWxlU2l0ZVJlc3Bv", - "bnNlRHRvEn0KEFJlcG9ydFNpdGVIZWFsdGgSMi5zY2FkYWJyaWRnZS5jZW50", - "cmFsY29udHJvbC52MS5TaXRlSGVhbHRoUmVwb3J0RHRvGjUuc2NhZGFicmlk", - "Z2UuY2VudHJhbGNvbnRyb2wudjEuU2l0ZUhlYWx0aFJlcG9ydEFja0R0bxJQ", - "CglIZWFydGJlYXQSKy5zY2FkYWJyaWRnZS5jZW50cmFsY29udHJvbC52MS5I", - "ZWFydGJlYXREdG8aFi5nb29nbGUucHJvdG9idWYuRW1wdHlCK6oCKFpCLk1P", - "TS5XVy5TY2FkYUJyaWRnZS5Db21tdW5pY2F0aW9uLkdycGNiBnByb3RvMw==")); + "YmVyGAIgASgDEhAKCGFjY2VwdGVkGAMgASgIEg0KBWVycm9yGAQgASgJIosB", + "CgxIZWFydGJlYXREdG8SDwoHc2l0ZV9pZBgBIAEoCRIVCg1ub2RlX2hvc3Ru", + "YW1lGAIgASgJEhEKCWlzX2FjdGl2ZRgDIAEoCBItCgl0aW1lc3RhbXAYBCAB", + "KAsyGi5nb29nbGUucHJvdG9idWYuVGltZXN0YW1wEhEKCXN5bnRoZXRpYxgF", + "IAEoCCq9AQoUQ29ubmVjdGlvbkhlYWx0aEVudW0SIQodQ09OTkVDVElPTl9I", + "RUFMVEhfVU5TUEVDSUZJRUQQABIfChtDT05ORUNUSU9OX0hFQUxUSF9DT05O", + "RUNURUQQARIiCh5DT05ORUNUSU9OX0hFQUxUSF9ESVNDT05ORUNURUQQAhIg", + "ChxDT05ORUNUSU9OX0hFQUxUSF9DT05ORUNUSU5HEAMSGwoXQ09OTkVDVElP", + "Tl9IRUFMVEhfRVJST1IQBDKhBgoVQ2VudHJhbENvbnRyb2xTZXJ2aWNlEoMB", + "ChJTdWJtaXROb3RpZmljYXRpb24SNC5zY2FkYWJyaWRnZS5jZW50cmFsY29u", + "dHJvbC52MS5Ob3RpZmljYXRpb25TdWJtaXREdG8aNy5zY2FkYWJyaWRnZS5j", + "ZW50cmFsY29udHJvbC52MS5Ob3RpZmljYXRpb25TdWJtaXRBY2tEdG8SkgEK", + "F1F1ZXJ5Tm90aWZpY2F0aW9uU3RhdHVzEjkuc2NhZGFicmlkZ2UuY2VudHJh", + "bGNvbnRyb2wudjEuTm90aWZpY2F0aW9uU3RhdHVzUXVlcnlEdG8aPC5zY2Fk", + "YWJyaWRnZS5jZW50cmFsY29udHJvbC52MS5Ob3RpZmljYXRpb25TdGF0dXNS", + "ZXNwb25zZUR0bxJHChFJbmdlc3RBdWRpdEV2ZW50cxIbLnNpdGVzdHJlYW0u", + "QXVkaXRFdmVudEJhdGNoGhUuc2l0ZXN0cmVhbS5Jbmdlc3RBY2sSUAoVSW5n", + "ZXN0Q2FjaGVkVGVsZW1ldHJ5EiAuc2l0ZXN0cmVhbS5DYWNoZWRUZWxlbWV0", + "cnlCYXRjaBoVLnNpdGVzdHJlYW0uSW5nZXN0QWNrEoABCg1SZWNvbmNpbGVT", + "aXRlEjYuc2NhZGFicmlkZ2UuY2VudHJhbGNvbnRyb2wudjEuUmVjb25jaWxl", + "U2l0ZVJlcXVlc3REdG8aNy5zY2FkYWJyaWRnZS5jZW50cmFsY29udHJvbC52", + "MS5SZWNvbmNpbGVTaXRlUmVzcG9uc2VEdG8SfQoQUmVwb3J0U2l0ZUhlYWx0", + "aBIyLnNjYWRhYnJpZGdlLmNlbnRyYWxjb250cm9sLnYxLlNpdGVIZWFsdGhS", + "ZXBvcnREdG8aNS5zY2FkYWJyaWRnZS5jZW50cmFsY29udHJvbC52MS5TaXRl", + "SGVhbHRoUmVwb3J0QWNrRHRvElAKCUhlYXJ0YmVhdBIrLnNjYWRhYnJpZGdl", + "LmNlbnRyYWxjb250cm9sLnYxLkhlYXJ0YmVhdER0bxoWLmdvb2dsZS5wcm90", + "b2J1Zi5FbXB0eUIrqgIoWkIuTU9NLldXLlNjYWRhQnJpZGdlLkNvbW11bmlj", + "YXRpb24uR3JwY2IGcHJvdG8z")); descriptor = pbr::FileDescriptor.FromGeneratedCode(descriptorData, new pbr::FileDescriptor[] { global::Google.Protobuf.WellKnownTypes.EmptyReflection.Descriptor, global::Google.Protobuf.WellKnownTypes.TimestampReflection.Descriptor, global::Google.Protobuf.WellKnownTypes.WrappersReflection.Descriptor, global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SitestreamReflection.Descriptor, }, new pbr::GeneratedClrTypeInfo(new[] {typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.ConnectionHealthEnum), }, null, new pbr::GeneratedClrTypeInfo[] { @@ -160,7 +161,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.NodeStatusListDto), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.NodeStatusListDto.Parser, new[]{ "Nodes" }, null, null, null, null), new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteHealthReportDto), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteHealthReportDto.Parser, new[]{ "SiteId", "SequenceNumber", "ReportTimestamp", "DataConnectionStatuses", "TagResolutionCounts", "ScriptErrorCount", "AlarmEvaluationErrorCount", "StoreAndForwardBufferDepths", "DeadLetterCount", "DeployedInstanceCount", "EnabledInstanceCount", "DisabledInstanceCount", "NodeRole", "NodeHostname", "DataConnectionEndpoints", "DataConnectionTagQuality", "ParkedMessageCount", "ClusterNodes", "SiteAuditWriteFailures", "AuditRedactionFailure", "SiteAuditBacklog", "SiteEventLogWriteFailures", "OldestParkedMessageAgeSeconds", "ScriptQueueDepth", "ScriptBusyThreads", "ScriptOldestBusyAgeSeconds", "LocalDbReplicationConnected", "LocalDbOplogBacklog" }, null, null, null, new pbr::GeneratedClrTypeInfo[] { null, null, null, }), new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteHealthReportAckDto), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteHealthReportAckDto.Parser, new[]{ "SiteId", "SequenceNumber", "Accepted", "Error" }, null, null, null, null), - new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.HeartbeatDto), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.HeartbeatDto.Parser, new[]{ "SiteId", "NodeHostname", "IsActive", "Timestamp" }, null, null, null, null) + new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.HeartbeatDto), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.HeartbeatDto.Parser, new[]{ "SiteId", "NodeHostname", "IsActive", "Timestamp", "Synthetic" }, null, null, null, null) })); } #endregion @@ -5826,6 +5827,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { nodeHostname_ = other.nodeHostname_; isActive_ = other.isActive_; timestamp_ = other.timestamp_ != null ? other.timestamp_.Clone() : null; + synthetic_ = other.synthetic_; _unknownFields = pb::UnknownFieldSet.Clone(other._unknownFields); } @@ -5883,6 +5885,27 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { } } + /// Field number for the "synthetic" field. + public const int SyntheticFieldNumber = 5; + private bool synthetic_; + /// + /// Additive (field 5): true marks a SYNTHETIC heartbeat that carries no liveness + /// meaning — today only CentralChannelProvider's failback probe, which reuses this + /// RPC purely to test whether the preferred central endpoint answers again. Central + /// must skip ALL liveness/health bookkeeping for such a heartbeat: it does not come + /// from a site node's heartbeat timer, and treating it as one would keep a site + /// looking alive on the health dashboard from the probe alone. proto3 defaults it to + /// false, so a pre-existing site's heartbeat is a real one, as before. + /// + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public bool Synthetic { + get { return synthetic_; } + set { + synthetic_ = value; + } + } + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] public override bool Equals(object other) { @@ -5902,6 +5925,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { if (NodeHostname != other.NodeHostname) return false; if (IsActive != other.IsActive) return false; if (!object.Equals(Timestamp, other.Timestamp)) return false; + if (Synthetic != other.Synthetic) return false; return Equals(_unknownFields, other._unknownFields); } @@ -5913,6 +5937,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { if (NodeHostname.Length != 0) hash ^= NodeHostname.GetHashCode(); if (IsActive != false) hash ^= IsActive.GetHashCode(); if (timestamp_ != null) hash ^= Timestamp.GetHashCode(); + if (Synthetic != false) hash ^= Synthetic.GetHashCode(); if (_unknownFields != null) { hash ^= _unknownFields.GetHashCode(); } @@ -5947,6 +5972,10 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { output.WriteRawTag(34); output.WriteMessage(Timestamp); } + if (Synthetic != false) { + output.WriteRawTag(40); + output.WriteBool(Synthetic); + } if (_unknownFields != null) { _unknownFields.WriteTo(output); } @@ -5973,6 +6002,10 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { output.WriteRawTag(34); output.WriteMessage(Timestamp); } + if (Synthetic != false) { + output.WriteRawTag(40); + output.WriteBool(Synthetic); + } if (_unknownFields != null) { _unknownFields.WriteTo(ref output); } @@ -5995,6 +6028,9 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { if (timestamp_ != null) { size += 1 + pb::CodedOutputStream.ComputeMessageSize(Timestamp); } + if (Synthetic != false) { + size += 1 + 1; + } if (_unknownFields != null) { size += _unknownFields.CalculateSize(); } @@ -6022,6 +6058,9 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { } Timestamp.MergeFrom(other.Timestamp); } + if (other.Synthetic != false) { + Synthetic = other.Synthetic; + } _unknownFields = pb::UnknownFieldSet.MergeFrom(_unknownFields, other._unknownFields); } @@ -6060,6 +6099,10 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { input.ReadMessage(Timestamp); break; } + case 40: { + Synthetic = input.ReadBool(); + break; + } } } #endif @@ -6098,6 +6141,10 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { input.ReadMessage(Timestamp); break; } + case 40: { + Synthetic = input.ReadBool(); + break; + } } } } diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs index 33073cdb..b8ff2a20 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptions.cs @@ -47,6 +47,36 @@ public class CommunicationOptions /// public TimeSpan NotificationForwardTimeout { get; set; } = TimeSpan.FromSeconds(30); + /// + /// Audit Log: timeout for the SITE-side Ask that forwards one audit-telemetry batch + /// (IngestAuditEventsCommand / IngestCachedTelemetryCommand) through the site's + /// SiteCommunicationActor and awaits central's ack. Deliberately the LONGEST rung of the + /// ingest timeout ladder. + /// + /// + /// + /// The ladder is strictly monotonic, outermost first: + /// AuditForwardTimeout (35 s, this option) > + /// SiteStreamGrpcServer.AuditIngestAskTimeout (30 s — both the gRPC call deadline and + /// central's own Ask of the ingest singleton) > + /// AuditLogIngestActor.IngestBudget (20 s) > + /// AuditLogIngestActor.IngestSqlCommandTimeout (15 s). + /// + /// + /// Strictness matters: it used to reuse (30 s), the + /// same value as the rung below it, so a slow-but-succeeding central write could be acked to a + /// caller whose Ask had already expired — the drain loop would treat the batch as unsent and + /// re-ship it. Central dedups on EventId, so the duplicate was harmless but the retry + /// traffic and the "stalled" telemetry signal were not. A timeout here is still transient: the + /// rows stay Pending and drain on the next tick. + /// + /// + /// enforces the outermost inequality — this value + /// must be strictly greater than the 30 s gRPC/central Ask rung. + /// + /// + public TimeSpan AuditForwardTimeout { get; set; } = TimeSpan.FromSeconds(35); + /// /// Preshared key authenticating this node's gRPC control plane — the site↔central /// boundary. On a site node this is the key its inbound gate diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptionsValidator.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptionsValidator.cs index 00f70888..7bee6e27 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptionsValidator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/CommunicationOptionsValidator.cs @@ -39,6 +39,15 @@ public sealed class CommunicationOptionsValidator : OptionsValidatorBase TimeSpan.Zero, $"ScadaBridge:Communication:NotificationForwardTimeout must be a positive duration (was {options.NotificationForwardTimeout})."); + // Outermost rung of the audit-ingest timeout ladder. It must be strictly greater than the + // gRPC deadline / central Ask beneath it, or a slow-but-succeeding central write gets acked + // to a caller that already gave up and re-sent the batch. See CommunicationOptions. + builder.RequireThat( + options.AuditForwardTimeout > Grpc.SiteStreamGrpcServer.AuditIngestAskTimeout, + $"ScadaBridge:Communication:AuditForwardTimeout must be strictly greater than the " + + $"central audit-ingest Ask timeout ({Grpc.SiteStreamGrpcServer.AuditIngestAskTimeout}) " + + $"so the ingest timeout ladder stays monotonic (was {options.AuditForwardTimeout})."); + builder.RequireThat(options.GrpcKeepAlivePingDelay > TimeSpan.Zero, $"ScadaBridge:Communication:GrpcKeepAlivePingDelay must be a positive duration (was {options.GrpcKeepAlivePingDelay})."); diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralChannelProvider.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralChannelProvider.cs index 9b8f96fd..ff407971 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralChannelProvider.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralChannelProvider.cs @@ -38,6 +38,15 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc; /// public sealed class CentralChannelProvider : IDisposable { + /// + /// Hostname stamped on the failback probe's heartbeat, so a central-side log line names it + /// for what it is. It is a LABEL only — the machine-readable contract is + /// + /// (HeartbeatDto.synthetic, proto field 5), which central keys its skip on. Never + /// filter on this string. + /// + public const string SyntheticProbeHostname = "failback-probe"; + private static readonly TimeSpan DefaultBackoffBase = TimeSpan.FromSeconds(1); private static readonly TimeSpan DefaultBackoffCap = TimeSpan.FromSeconds(60); private static readonly TimeSpan DefaultProbeDeadline = TimeSpan.FromSeconds(5); @@ -202,9 +211,16 @@ public sealed class CentralChannelProvider : IDisposable new HeartbeatDto { SiteId = _siteId, - NodeHostname = "failback-probe", + NodeHostname = SyntheticProbeHostname, IsActive = false, Timestamp = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow), + // Explicit contract, not a hostname convention: central skips ALL + // liveness/health bookkeeping for a synthetic heartbeat. Without it the + // probe would stamp the site's last-seen time on the central health + // aggregator every backoff tick, so a site whose real heartbeats had + // stopped could still look alive purely because its transport layer was + // probing for failback. + Synthetic = true, }, deadline: DateTime.UtcNow.Add(_probeDeadline)).ConfigureAwait(false); diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralControlDtoMapper.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralControlDtoMapper.cs index 2393cff1..d675845b 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralControlDtoMapper.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralControlDtoMapper.cs @@ -695,6 +695,7 @@ public static class CentralControlDtoMapper NodeHostname = msg.NodeHostname, IsActive = msg.IsActive, Timestamp = Timestamp.FromDateTimeOffset(msg.Timestamp), + Synthetic = msg.Synthetic, }; } @@ -709,7 +710,8 @@ public static class CentralControlDtoMapper SiteId: dto.SiteId, NodeHostname: dto.NodeHostname, IsActive: dto.IsActive, - Timestamp: dto.Timestamp.ToDateTimeOffset()); + Timestamp: dto.Timestamp.ToDateTimeOffset(), + Synthetic: dto.Synthetic); } /// Projects a onto its wire enum. diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralControlGrpcService.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralControlGrpcService.cs index e0b821bf..bf74635d 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralControlGrpcService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralControlGrpcService.cs @@ -33,13 +33,23 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc; /// host's Akka bootstrap, not by the container. /// /// +/// The two audit-ingest RPCs are the exception: they Ask the ingest singleton DIRECTLY. +/// CentralCommunicationActor used to relay them — an extra hop that re-Asked the +/// same audit-log-ingest proxy with the same 30 s +/// , so the inner Ask expired at the same +/// instant as the outer one and could only add latency plus a +/// repackaging. The proxy now arrives through +/// , exactly as has always +/// taken it, and the relay handlers were deleted. +/// +/// /// Fault semantics deliberately differ from 's ingest /// RPCs. That server answers a failed audit ingest with an EMPTY IngestAck; this one /// fails the call with a non-OK status. Both leave the site's rows Pending for the next /// drain, so the outcome is the same — but this service replaced the ClusterClient path, whose -/// documented behaviour was to propagate the fault (CentralCommunicationActor's -/// HandleIngestAuditEvents pipes a Status.Failure back), and preserving that keeps -/// a lost batch visible as a failure rather than as a successful call that acked nothing. +/// documented behaviour was to propagate the fault, and preserving that keeps a lost batch +/// visible as a failure rather than as a successful call that acked nothing. A proxy that has +/// not been handed over yet is NOT a fault and still answers with an empty ack. /// /// /// Status mapping, and why it is not uniform. A site transport may safely re-send a call @@ -66,6 +76,14 @@ public sealed class CentralControlGrpcService : CentralControlService.CentralCon // Kestrel's thread pool. private volatile IActorRef? _central; + // Central-side audit-ingest singleton proxy (audit-log-ingest). Handed over by the host + // AFTER the singleton starts, so it arrives strictly later than _central. The two ingest + // RPCs Ask THIS proxy directly rather than relaying through CentralCommunicationActor — + // see the class remarks. Null until registration completes; an ingest arriving before then + // gets an empty ack (leave the site's rows Pending, retry next drain), which is exactly + // what the removed relay replied in the same window. + private volatile IActorRef? _auditIngest; + /// /// Creates the service. This must remain the only public constructor — see /// for how the actor arrives, and the Host's @@ -105,9 +123,32 @@ public sealed class CentralControlGrpcService : CentralControlService.CentralCon _central = centralCommunicationActor; } + /// + /// Hands the central AuditLogIngestActor singleton proxy to the service so the two + /// ingest RPCs can Ask it DIRECTLY. Mirrors + /// — the site-stream server has always + /// dispatched straight to the singleton, and this service now matches it. + /// + /// + /// Removing the CentralCommunicationActor relay removes an actor hop, a second Ask, + /// and a second timeout from every audit batch: the relay's own Ask was armed with the same + /// 30 s as this service's, so the two + /// expired together and the inner one could only ever add latency and a + /// repackaging on the way out. + /// + /// The audit-log-ingest singleton proxy. + public void SetAuditIngestActor(IActorRef auditIngestProxy) + { + ArgumentNullException.ThrowIfNull(auditIngestProxy); + _auditIngest = auditIngestProxy; + } + /// Exposed for wiring assertions in tests. internal bool IsReady => _central is not null; + /// Exposed for wiring assertions in tests. + internal bool IsAuditIngestBound => _auditIngest is not null; + /// public override async Task SubmitNotification( NotificationSubmitDto request, ServerCallContext context) @@ -147,9 +188,16 @@ public sealed class CentralControlGrpcService : CentralControlService.CentralCon return new IngestAck(); } - var central = RequireReady(context); + // Node-readiness gate first (the actor system must be up), then the ingest proxy. + RequireReady(context); + var ingest = RequireAuditIngest(request.Events.Count, nameof(IngestAuditEvents)); + if (ingest is null) + { + return new IngestAck(); + } + var reply = await AskAsync( - central, + ingest, CentralControlDtoMapper.FromDto(request), SiteStreamGrpcServer.AuditIngestAskTimeout, context).ConfigureAwait(false); @@ -166,9 +214,15 @@ public sealed class CentralControlGrpcService : CentralControlService.CentralCon return new IngestAck(); } - var central = RequireReady(context); + RequireReady(context); + var ingest = RequireAuditIngest(request.Packets.Count, nameof(IngestCachedTelemetry)); + if (ingest is null) + { + return new IngestAck(); + } + var reply = await AskAsync( - central, + ingest, CentralControlDtoMapper.FromDto(request), SiteStreamGrpcServer.AuditIngestAskTimeout, context).ConfigureAwait(false); @@ -176,6 +230,30 @@ public sealed class CentralControlGrpcService : CentralControlService.CentralCon return CentralControlDtoMapper.ToIngestAck(reply.AcceptedEventIds); } + /// + /// Returns the audit-ingest singleton proxy, or null when the host has not handed it + /// over yet (it starts moments after ). A null answer means the caller + /// must reply with an EMPTY — nothing accepted, rows stay + /// Pending on the site and drain on the next tick. That is deliberately NOT + /// : readiness here is intentionally narrow (see + /// ), and this is the exact reply the removed + /// CentralCommunicationActor relay produced in the same window. + /// + private IActorRef? RequireAuditIngest(int itemCount, string rpc) + { + var ingest = _auditIngest; + if (ingest is not null) + { + return ingest; + } + + _logger.LogWarning( + "{Rpc} received {Count} item(s) before SetAuditIngestActor was called; returning an " + + "empty ack so the site keeps its rows Pending and retries.", + rpc, itemCount); + return null; + } + /// public override async Task ReconcileSite( ReconcileSiteRequestDto request, ServerCallContext context) diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/ISiteAlarmLiveCache.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/ISiteAlarmLiveCache.cs index 5d094a47..09042357 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/ISiteAlarmLiveCache.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/ISiteAlarmLiveCache.cs @@ -62,4 +62,29 @@ public interface ISiteAlarmLiveCache /// The numeric site id. /// true while a living aggregator has seeded and published; reverts to false if the aggregator terminates (R2 N6) — a frozen snapshot is never reported as live. bool IsLive(int siteId); + + /// + /// The Enabled instances on the site that failed to answer the aggregator's most recent + /// snapshot fan-out (seed or reconcile) — i.e. that reported InstanceNotFound, timed + /// out, or faulted. Deterministically ordered by instance name, and empty when the site has + /// no aggregator or has not yet completed a fan-out. + /// + /// + /// + /// Published alongside the alarm snapshot on every fan-out, because the aggregator already + /// knows it: the seed/reconcile is the SAME per-instance debug-snapshot fan-out the Alarm + /// Summary poll used to run a second time purely to compute this list. Exposing it here lets + /// the UI-side poll be skipped entirely while is true, which is the + /// whole point — one fan-out per site per reconcile window, not two. + /// + /// + /// It is refreshed at the aggregator's fan-out cadence (seed, then each reconcile), NOT on + /// every stream delta: an instance going silent is precisely what the stream cannot tell you. + /// Callers must therefore treat it as reconcile-fresh, not delta-fresh — which matches the + /// poll's own 15 s freshness, so nothing on screen changes. + /// + /// + /// The numeric site id. + /// The not-reporting instance unique names (possibly empty), never null. + IReadOnlyList GetNotReportingInstances(int siteId); } diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Protos/central_control.proto b/src/ZB.MOM.WW.ScadaBridge.Communication/Protos/central_control.proto index 97d4daf1..7faa1170 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Protos/central_control.proto +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Protos/central_control.proto @@ -244,4 +244,12 @@ message HeartbeatDto { string node_hostname = 2; bool is_active = 3; google.protobuf.Timestamp timestamp = 4; + // Additive (field 5): true marks a SYNTHETIC heartbeat that carries no liveness + // meaning — today only CentralChannelProvider's failback probe, which reuses this + // RPC purely to test whether the preferred central endpoint answers again. Central + // must skip ALL liveness/health bookkeeping for such a heartbeat: it does not come + // from a site node's heartbeat timer, and treating it as one would keep a site + // looking alive on the health dashboard from the probe alone. proto3 defaults it to + // false, so a pre-existing site's heartbeat is a real one, as before. + bool synthetic = 5; } diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs index c71071ee..49d55bf6 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs @@ -30,6 +30,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication; public sealed class SiteAlarmLiveCacheService : ISiteAlarmLiveCache { private static readonly IReadOnlyList Empty = Array.Empty(); + private static readonly IReadOnlyList EmptyNames = Array.Empty(); private readonly IServiceProvider _serviceProvider; private readonly CommunicationService _communicationService; @@ -135,6 +136,13 @@ public sealed class SiteAlarmLiveCacheService : ISiteAlarmLiveCache return _sites.TryGetValue(siteId, out var entry) && entry.HasPublished && entry.StreamLive; } + /// + public IReadOnlyList GetNotReportingInstances(int siteId) + { + lock (_lock) + return _sites.TryGetValue(siteId, out var entry) ? entry.NotReporting : EmptyNames; + } + // ── Subscriber teardown ───────────────────────────────────────────────────── private void Unsubscribe(Subscription subscription) @@ -332,6 +340,15 @@ public sealed class SiteAlarmLiveCacheService : ISiteAlarmLiveCache /// contributes no rows rather than /// failing the whole seed (mirrors AlarmSummaryService). Re-enumerates instances /// each call so a reconcile picks up enable/disable/deploy/delete drift. + /// + /// It also records WHICH instances failed to answer into + /// , published through + /// . That set was previously + /// computed and thrown away here, forcing the Alarm Summary page to run a second, identical + /// fan-out through AlarmSummaryService purely to rebuild it. Same classification as + /// that service (InstanceNotFound or any non-cancellation fault), same + /// ordinal-ignore-case ordering, so the page renders identically from either source. + /// /// private async Task> FanOutSnapshotsAsync(int siteId, CancellationToken ct) { @@ -343,7 +360,10 @@ public sealed class SiteAlarmLiveCacheService : ISiteAlarmLiveCache var siteRepo = scope.ServiceProvider.GetRequiredService(); var site = await siteRepo.GetSiteByIdAsync(siteId, ct); if (site is null) + { + PublishNotReporting(siteId, EmptyNames); return Empty; + } siteIdentifier = site.SiteIdentifier; var instanceRepo = scope.ServiceProvider.GetRequiredService(); @@ -355,25 +375,48 @@ public sealed class SiteAlarmLiveCacheService : ISiteAlarmLiveCache } if (enabledInstanceNames.Count == 0) + { + PublishNotReporting(siteId, EmptyNames); return Empty; + } var rows = new ConcurrentBag(); + var notReporting = new ConcurrentBag(); using var gate = new SemaphoreSlim( Math.Max(1, _options.LiveAlarmCacheSeedConcurrency), Math.Max(1, _options.LiveAlarmCacheSeedConcurrency)); var fetches = enabledInstanceNames.Select(name => - FetchInstanceSnapshotAsync(siteIdentifier, name, gate, rows, ct)); + FetchInstanceSnapshotAsync(siteIdentifier, name, gate, rows, notReporting, ct)); await Task.WhenAll(fetches); + PublishNotReporting( + siteId, + notReporting.OrderBy(n => n, StringComparer.OrdinalIgnoreCase).ToList()); + return rows.ToList(); } + /// + /// Stores the fan-out's not-reporting set on the site entry. Silently no-ops when the entry + /// has been torn down mid-fan-out (last viewer left), matching every other late-write path + /// in this service. + /// + private void PublishNotReporting(int siteId, IReadOnlyList notReporting) + { + lock (_lock) + { + if (_sites.TryGetValue(siteId, out var entry)) + entry.NotReporting = notReporting; + } + } + private async Task FetchInstanceSnapshotAsync( string siteIdentifier, string instanceUniqueName, SemaphoreSlim gate, ConcurrentBag rows, + ConcurrentBag notReporting, CancellationToken ct) { await gate.WaitAsync(ct); @@ -387,7 +430,10 @@ public sealed class SiteAlarmLiveCacheService : ISiteAlarmLiveCache instanceUniqueName, Guid.NewGuid().ToString("N"), AlarmsOnly: true); var snapshot = await _communicationService.RequestDebugSnapshotAsync(siteIdentifier, request, ct); if (snapshot.InstanceNotFound) + { + notReporting.Add(instanceUniqueName); return; + } foreach (var alarm in snapshot.AlarmStates) rows.Add(alarm); @@ -398,7 +444,9 @@ public sealed class SiteAlarmLiveCacheService : ISiteAlarmLiveCache } catch { - // Degrade this one instance to "no rows" rather than failing the whole seed. + // Degrade this one instance to "no rows, not reporting" rather than failing the + // whole seed. Same rule as AlarmSummaryService.FetchInstanceAsync. + notReporting.Add(instanceUniqueName); } finally { @@ -474,6 +522,9 @@ public sealed class SiteAlarmLiveCacheService : ISiteAlarmLiveCache entry.HasPublished = false; entry.StreamLive = false; entry.Current = Empty; + // A dead aggregator's last fan-out result is stale too — clear it so the page's + // poll fallback (now authoritative again) owns the not-reporting list outright. + entry.NotReporting = EmptyNames; if (entry.Subscribers.Count > 0 && !entry.Starting) { @@ -509,6 +560,13 @@ public sealed class SiteAlarmLiveCacheService : ISiteAlarmLiveCache /// True once the aggregator has seeded and published at least once. public bool HasPublished { get; set; } + /// + /// Enabled instances that failed to answer the most recent snapshot fan-out, ordered by + /// name. Refreshed on every seed/reconcile (see FanOutSnapshotsAsync) and cleared + /// when liveness resets, so a dead aggregator never leaves a stale list behind. + /// + public IReadOnlyList NotReporting { get; set; } = EmptyNames; + /// /// Liveness of the aggregator's site-wide gRPC stream as of the last publish. Both /// this and must hold for IsLive: a published cache diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs index 66d6996d..985bb4e6 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs @@ -534,12 +534,12 @@ akka {{ _logger.LogInformation("NotificationOutbox singleton created and registered with CentralCommunicationActor"); // Audit Log — central singleton mirrors the Notification Outbox - // pattern. The IngestAuditEvents gRPC handler lives on SiteStreamGrpcServer - // (Communication.Grpc); a central node hosting that server (reconciliation - // path) hands the proxy in via SetAuditIngestActor below. When the gRPC - // server is not registered (current central topology), the host still - // brings the singleton up so an in-process test (or a future - // direct caller) can Ask the proxy without further wiring. + // pattern. TWO gRPC servers can carry an IngestAuditEvents call and both Ask this + // proxy directly: the central-hosted CentralControlGrpcService (the production + // site→central path) and SiteStreamGrpcServer (Communication.Grpc). Each is handed + // the proxy below; either may be absent on a given node and the wiring no-ops. Even + // with neither bound the host still brings the singleton up so an in-process test + // (or a future direct caller) can Ask the proxy without further wiring. // IAuditLogRepository is a SCOPED EF Core service, so the singleton // actor takes the root IServiceProvider and creates a fresh scope per // message (mirroring NotificationOutboxActor). Pre-resolving the @@ -555,10 +555,11 @@ akka {{ auditIngestLogger)), _logger); - // Hand the audit-ingest proxy to the CentralCommunicationActor so audit - // ingest commands forwarded by sites are routed to the - // singleton. Mirrors the RegisterNotificationOutbox wiring above. - centralCommActor.Tell(new RegisterAuditIngest(auditIngest.Proxy)); + // Hand the audit-ingest proxy to the central-hosted gRPC control plane so its two + // ingest RPCs Ask the singleton DIRECTLY. There is deliberately no + // CentralCommunicationActor relay any more: it re-Asked this same proxy with the same + // 30 s timeout, so it could only add a hop and latency to every audit batch. + centralControlGrpc?.SetAuditIngestActor(auditIngest.Proxy); // Hand the proxy to the SiteStreamGrpcServer (if registered on this node) // so the IngestAuditEvents RPC routes incoming site batches to the singleton. @@ -568,7 +569,9 @@ akka {{ var grpcServer = _serviceProvider.GetService(); grpcServer?.SetAuditIngestActor(auditIngest.Proxy); _logger.LogInformation( - "AuditLogIngestActor singleton created (gRPC server bound: {GrpcBound})", + "AuditLogIngestActor singleton created (control-plane bound: {ControlBound}, " + + "site-stream server bound: {GrpcBound})", + centralControlGrpc is not null, grpcServer is not null); // Subscribe the per-site stalled @@ -1066,13 +1069,16 @@ akka {{ // not the DI-resolved NoOpSiteStreamAuditClient. The NoOp default stays // correct for central/test composition roots (no SiteCommunicationActor); // a site role wires the real client here so the - // SQLite Pending backlog actually drains to central. The forward Ask - // reuses NotificationForwardTimeout — the same site→central command - // forward bound notifications already use over this transport. + // SQLite Pending backlog actually drains to central. The forward Ask uses + // AuditForwardTimeout (35 s) — the OUTERMOST rung of the ingest timeout + // ladder (35 > 30 gRPC/central Ask > 20 actor budget > 15 SQL), so a + // slow-but-succeeding central write is never acked to a caller that has + // already given up and re-sent. It used to reuse NotificationForwardTimeout + // (30 s), which tied it with the rung below. ZB.MOM.WW.ScadaBridge.AuditLog.Site.Telemetry.ISiteStreamAuditClient siteAuditClient = new ZB.MOM.WW.ScadaBridge.AuditLog.Site.Telemetry.SiteCommunicationAuditClient( siteCommActor, - _communicationOptions.NotificationForwardTimeout); + _communicationOptions.AuditForwardTimeout); var siteAuditLogger = _serviceProvider.GetRequiredService() .CreateLogger(); diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/GrpcPullAuditEventsClientTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/GrpcPullAuditEventsClientTests.cs index 269407a6..61cba883 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/GrpcPullAuditEventsClientTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/GrpcPullAuditEventsClientTests.cs @@ -117,7 +117,7 @@ public class GrpcPullAuditEventsClientTests invoker, NullLogger.Instance); - var result = await sut.PullAsync("site-a", BaseTime, batchSize: 256, CancellationToken.None); + var result = await sut.PullAsync("site-a", BaseTime, afterId: null, batchSize: 256, ct: CancellationToken.None); // Endpoint resolution + request shaping. Assert.Equal("http://site-a:8083", invoker.Endpoint); @@ -141,7 +141,7 @@ public class GrpcPullAuditEventsClientTests invoker, NullLogger.Instance); - var result = await sut.PullAsync("site-a", BaseTime, batchSize: 256, CancellationToken.None); + var result = await sut.PullAsync("site-a", BaseTime, afterId: null, batchSize: 256, ct: CancellationToken.None); Assert.Empty(result.Events); Assert.False(result.MoreAvailable); @@ -161,7 +161,7 @@ public class GrpcPullAuditEventsClientTests NullLogger.Instance); // MUST NOT throw — per the IPullAuditEventsClient contract. - var result = await sut.PullAsync("site-a", BaseTime, batchSize: 256, CancellationToken.None); + var result = await sut.PullAsync("site-a", BaseTime, afterId: null, batchSize: 256, ct: CancellationToken.None); Assert.Empty(result.Events); Assert.False(result.MoreAvailable); @@ -178,7 +178,7 @@ public class GrpcPullAuditEventsClientTests invoker, NullLogger.Instance); - var result = await sut.PullAsync("site-a", BaseTime, batchSize: 256, CancellationToken.None); + var result = await sut.PullAsync("site-a", BaseTime, afterId: null, batchSize: 256, ct: CancellationToken.None); Assert.Empty(result.Events); Assert.False(result.MoreAvailable); @@ -197,7 +197,7 @@ public class GrpcPullAuditEventsClientTests invoker, NullLogger.Instance); - var result = await sut.PullAsync("site-a", BaseTime, batchSize: 256, CancellationToken.None); + var result = await sut.PullAsync("site-a", BaseTime, afterId: null, batchSize: 256, ct: CancellationToken.None); Assert.Empty(result.Events); Assert.False(result.MoreAvailable); @@ -221,7 +221,7 @@ public class GrpcPullAuditEventsClientTests NullLogger.Instance); // MUST NOT throw — must dial successfully. - var result = await sut.PullAsync("site-a", minUnspecified, batchSize: 256, CancellationToken.None); + var result = await sut.PullAsync("site-a", minUnspecified, afterId: null, batchSize: 256, ct: CancellationToken.None); Assert.Equal(1, invoker.CallCount); Assert.Equal("http://site-a:8083", invoker.Endpoint); @@ -252,7 +252,7 @@ public class GrpcPullAuditEventsClientTests invoker, NullLogger.Instance); - var result = await sut.PullAsync("site-a", BaseTime, batchSize: 256, CancellationToken.None); + var result = await sut.PullAsync("site-a", BaseTime, afterId: null, batchSize: 256, ct: CancellationToken.None); Assert.Equal(new[] { "http://node-a:8083", "http://node-b:8083" }, invoker.Dialed); var evt = Assert.Single(result.Events); @@ -271,7 +271,7 @@ public class GrpcPullAuditEventsClientTests invoker, NullLogger.Instance); - var result = await sut.PullAsync("site-a", BaseTime, batchSize: 256, CancellationToken.None); + var result = await sut.PullAsync("site-a", BaseTime, afterId: null, batchSize: 256, ct: CancellationToken.None); Assert.Empty(result.Events); Assert.Equal(new[] { "http://node-a:8083" }, invoker.Dialed); diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/SiteAuditReconciliationActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/SiteAuditReconciliationActorTests.cs index bb4b9eef..411cc2c4 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/SiteAuditReconciliationActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/SiteAuditReconciliationActorTests.cs @@ -194,7 +194,7 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture private sealed class ScriptedPullClient : IPullAuditEventsClient { - public List<(string SiteId, DateTime SinceUtc, int BatchSize)> Calls { get; } = new(); + public List<(string SiteId, DateTime SinceUtc, string? AfterId, int BatchSize)> Calls { get; } = new(); private readonly Dictionary> _scripted = new(); private readonly Dictionary _throwOnSite = new(); @@ -211,9 +211,9 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture PullAsync( - string siteId, DateTime sinceUtc, int batchSize, CancellationToken ct) + string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct) { - Calls.Add((siteId, sinceUtc, batchSize)); + Calls.Add((siteId, sinceUtc, afterId, batchSize)); if (_throwOnSite.TryGetValue(siteId, out var ex)) { throw ex; @@ -425,6 +425,104 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture Assert.True(client.Calls.Count >= 2, + $"need at least 2 pulls, got {client.Calls.Count}"), + duration: TimeSpan.FromSeconds(5), + interval: TimeSpan.FromMilliseconds(50)); + + Assert.Equal(t, client.Calls[1].SinceUtc); + Assert.Equal(high.EventId.ToString(), client.Calls[1].AfterId); + } + + [Fact] + public void BothCursorHalves_AreHeldBack_WhileARowIsStillBeingRetried() + { + // A held-back cursor must hold BOTH halves: advancing the id while pinning the + // timestamp would skip past the very rows being retried. + var sites = new StaticEnumerator(new SiteEntry("siteA", "http://siteA:8083")); + var t = new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc); + var evt = NewEvent("siteA", t); + + var client = new ScriptedPullClient().Script("siteA", + new PullAuditEventsResponse(new[] { evt }, MoreAvailable: false)); + var repo = new AlwaysThrowingRepo(); + + CreateActor(sites, client, repo, FastTickOptions()); + + AwaitAssert(() => Assert.True(client.Calls.Count >= 2, + $"need at least 2 pulls, got {client.Calls.Count}"), + duration: TimeSpan.FromSeconds(5), + interval: TimeSpan.FromMilliseconds(50)); + + Assert.Equal(DateTime.MinValue, client.Calls[1].SinceUtc); + Assert.Null(client.Calls[1].AfterId); + } + + /// Repository whose every insert throws, so the retry hold-back path is taken. + private sealed class AlwaysThrowingRepo : IAuditLogRepository + { + public Task InsertIfNotExistsAsync(AuditEvent evt, CancellationToken ct = default) => + throw new InvalidOperationException("central insert failed"); + + public Task> QueryAsync( + AuditLogQueryFilter filter, AuditLogPaging paging, CancellationToken ct = default) + => throw new NotSupportedException(); + + public Task SwitchOutPartitionAsync( + DateTime monthBoundary, TimeSpan? commandTimeout = null, CancellationToken ct = default) + => throw new NotSupportedException(); + + public Task PurgeChannelOlderThanAsync( + string channel, DateTime threshold, int batchSize, TimeSpan? commandTimeout = null, + CancellationToken ct = default) + => throw new NotSupportedException(); + + public Task BackfillSourceNodeAsync( + string sentinel, DateTime before, int batchSize, CancellationToken ct = default) + => throw new NotSupportedException(); + + public Task> GetPartitionBoundariesOlderThanAsync( + DateTime threshold, CancellationToken ct = default) + => throw new NotSupportedException(); + + public Task GetKpiSnapshotAsync( + TimeSpan window, DateTime? nowUtc = null, CancellationToken ct = default) + => throw new NotSupportedException(); + + public Task> GetExecutionTreeAsync( + Guid executionId, CancellationToken ct = default) + => throw new NotSupportedException(); + + public Task> GetDistinctSourceNodesAsync(CancellationToken ct = default) + => throw new NotSupportedException(); } // --------------------------------------------------------------------- diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Integration/OutageReconciliationTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Integration/OutageReconciliationTests.cs index 075fddc6..2104c9a1 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Integration/OutageReconciliationTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Integration/OutageReconciliationTests.cs @@ -85,7 +85,7 @@ public class OutageReconciliationTests : TestKit, IClassFixture PullAsync( - string siteId, DateTime sinceUtc, int batchSize, CancellationToken ct) + string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct) { CallCount++; @@ -94,16 +94,16 @@ public class OutageReconciliationTests : TestKit, IClassFixture= read contract and only rows strictly - // older than it are provably received. + // losing them. The actor now sends the composite (timestamp, id) cursor, + // so retirement is exact: the rows AT the cursor instant are proven + // received too, where a bare timestamp could only prove strictly-older ones. if (sinceUtc > DateTime.MinValue) { - await _siteQueue.MarkReconciledUpToAsync(sinceUtc, null, ct).ConfigureAwait(false); + await _siteQueue.MarkReconciledUpToAsync(sinceUtc, afterId, ct).ConfigureAwait(false); } var rows = await _siteQueue - .ReadPendingSinceAsync(sinceUtc, batchSize, afterId: null, ct) + .ReadPendingSinceAsync(sinceUtc, batchSize, afterId, ct) .ConfigureAwait(false); // MoreAvailable is true iff the read filled the batch — the actor diff --git a/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Monitoring/AlarmSummaryRenderTests.cs b/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Monitoring/AlarmSummaryRenderTests.cs index 1c117079..802ef522 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Monitoring/AlarmSummaryRenderTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Monitoring/AlarmSummaryRenderTests.cs @@ -331,6 +331,14 @@ public class AlarmSummaryRenderTests : BunitContext public bool IsLive(int siteId) => _live; + /// + /// The aggregator's own fan-out result. Settable so a test can prove the page renders + /// the not-reporting list sourced from the live cache while it is serving the site. + /// + public IReadOnlyList NotReporting { get; set; } = Array.Empty(); + + public IReadOnlyList GetNotReportingInstances(int siteId) => NotReporting; + public void PushAlarms(IReadOnlyList alarms) { _current = alarms; diff --git a/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Monitoring/AlarmSummaryVirtualizeTests.cs b/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Monitoring/AlarmSummaryVirtualizeTests.cs index a51d9c9b..bd620106 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Monitoring/AlarmSummaryVirtualizeTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Monitoring/AlarmSummaryVirtualizeTests.cs @@ -105,6 +105,8 @@ public class AlarmSummaryVirtualizeTests : BunitContext public bool IsLive(int siteId) => false; + public IReadOnlyList GetNotReportingInstances(int siteId) => Array.Empty(); + private sealed class NoOp : IDisposable { public void Dispose() { } diff --git a/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Services/SharedAlarmSummaryServiceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Services/SharedAlarmSummaryServiceTests.cs index 83c4873a..66da08e5 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Services/SharedAlarmSummaryServiceTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Services/SharedAlarmSummaryServiceTests.cs @@ -56,13 +56,13 @@ public class SharedAlarmSummaryServiceTests : IDisposable _provider = services.BuildServiceProvider(); } - private SharedAlarmSummaryService CreateSut(TimeSpan liveCacheTtl) => - new(_provider.GetRequiredService(), _liveCache, liveCacheTtl, () => _now); + private SharedAlarmSummaryService CreateSut() => + new(_provider.GetRequiredService(), _liveCache, () => _now); [Fact] public async Task ConcurrentCircuits_ShareOneFanOut() { - var sut = CreateSut(TimeSpan.FromSeconds(60)); + var sut = CreateSut(); var results = await Task.WhenAll(Enumerable.Range(0, 8).Select(_ => sut.GetSiteAlarmsAsync(SiteId))); @@ -73,7 +73,7 @@ public class SharedAlarmSummaryServiceTests : IDisposable [Fact] public async Task ColdLiveCache_RefreshesWithinThePageTick() { - var sut = CreateSut(TimeSpan.FromSeconds(60)); + var sut = CreateSut(); _liveCache.Live = false; await sut.GetSiteAlarmsAsync(SiteId); @@ -86,22 +86,51 @@ public class SharedAlarmSummaryServiceTests : IDisposable } [Fact] - public async Task LiveCacheServing_WidensTheWindowToTheReconcileInterval() + public async Task LiveCacheServing_SkipsTheFanOutEntirely() { - var sut = CreateSut(TimeSpan.FromSeconds(60)); + // The aggregator's seed/reconcile already ran this exact fan-out and publishes both + // halves of the answer, so the façade must not run a second one — not now, not after + // any elapsed window (arch-review phase-2 residual #4). + var sut = CreateSut(); + _liveCache.Live = true; + _liveCache.NotReporting = new[] { "inst-silent" }; + _liveCache.Current = new[] + { + new AlarmStateChanged("inst-a", "A-alarm", AlarmState.Active, 500, T0), + }; + + var first = await sut.GetSiteAlarmsAsync(SiteId); + _now = T0.AddSeconds(61); + var second = await sut.GetSiteAlarmsAsync(SiteId); + + await _instanceRepo.DidNotReceive().GetInstancesBySiteIdAsync(SiteId, Arg.Any()); + await _snapshotClient.DidNotReceive().GetSnapshotAsync( + Arg.Any(), Arg.Any(), Arg.Any()); + + // Both halves come from the cache: the rows so a liveness flip mid-call lands on + // last-known state instead of a blank grid, and the not-reporting names the page shows. + foreach (var result in new[] { first, second }) + { + Assert.Equal("A-alarm", Assert.Single(result.Alarms).Alarm.AlarmName); + Assert.Equal("inst-silent", Assert.Single(result.NotReportingInstances)); + } + } + + [Fact] + public async Task LiveCacheGoingCold_FallsBackToTheFanOut() + { + var sut = CreateSut(); _liveCache.Live = true; await sut.GetSiteAlarmsAsync(SiteId); - _now = T0.AddSeconds(30); - await sut.GetSiteAlarmsAsync(SiteId); + await _instanceRepo.DidNotReceive().GetInstancesBySiteIdAsync(SiteId, Arg.Any()); + + // Aggregator died / stream degraded → the poll is the page's rebuild path again. + _liveCache.Live = false; + var result = await sut.GetSiteAlarmsAsync(SiteId); - // Live deltas own the rows; only the not-reporting list still comes from the - // fan-out, so a 30s-old answer is fine and costs no second fan-out. await _instanceRepo.Received(1).GetInstancesBySiteIdAsync(SiteId, Arg.Any()); - - _now = T0.AddSeconds(61); - await sut.GetSiteAlarmsAsync(SiteId); - await _instanceRepo.Received(2).GetInstancesBySiteIdAsync(SiteId, Arg.Any()); + Assert.Single(result.Alarms); } [Fact] @@ -113,7 +142,7 @@ public class SharedAlarmSummaryServiceTests : IDisposable _instanceRepo.GetInstancesBySiteIdAsync(otherSite, Arg.Any()) .Returns(new List()); - var sut = CreateSut(TimeSpan.FromSeconds(60)); + var sut = CreateSut(); await sut.GetSiteAlarmsAsync(SiteId); await sut.GetSiteAlarmsAsync(otherSite); @@ -125,7 +154,7 @@ public class SharedAlarmSummaryServiceTests : IDisposable [Fact] public void PureMethods_MatchTheDirectImplementation() { - var sut = CreateSut(TimeSpan.FromSeconds(60)); + var sut = CreateSut(); var direct = new AlarmSummaryService(_instanceRepo, _siteRepo, _snapshotClient); var alarms = new List { @@ -149,18 +178,24 @@ public class SharedAlarmSummaryServiceTests : IDisposable public void Dispose() => _provider.Dispose(); - /// Liveness-only stub — the façade consults nothing else on the live cache. + /// + /// Read-side stub: liveness, the published alarm snapshot, and the aggregator's + /// not-reporting set — the three things the façade reads while the cache is serving a site. + /// private sealed class FakeLiveCache : ISiteAlarmLiveCache { public bool Live { get; set; } + public IReadOnlyList Current { get; set; } = Array.Empty(); + public IReadOnlyList NotReporting { get; set; } = Array.Empty(); public IDisposable Subscribe(int siteId, Action onChanged) => new NoOp(); - public IReadOnlyList GetCurrentAlarms(int siteId) => - Array.Empty(); + public IReadOnlyList GetCurrentAlarms(int siteId) => Current; public bool IsLive(int siteId) => Live; + public IReadOnlyList GetNotReportingInstances(int siteId) => NotReporting; + private sealed class NoOp : IDisposable { public void Dispose() { } diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorAuditTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorAuditTests.cs deleted file mode 100644 index 8cc6ab5f..00000000 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorAuditTests.cs +++ /dev/null @@ -1,151 +0,0 @@ -using Akka.Actor; -using Akka.TestKit; -using Akka.TestKit.Xunit2; -using Microsoft.Extensions.DependencyInjection; -using NSubstitute; -using ZB.MOM.WW.Audit; -using ZB.MOM.WW.ScadaBridge.Commons.Entities.Audit; -using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; -using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit; -using ZB.MOM.WW.ScadaBridge.Commons.Types; -using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit; -using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; -using ZB.MOM.WW.ScadaBridge.Communication.Actors; - -namespace ZB.MOM.WW.ScadaBridge.Communication.Tests; - -/// -/// Tests for the Audit Log (#23) site→central ingest routing on -/// . A site delivers -/// / -/// to the actor, which forwards to the registered -/// AuditLogIngestActor proxy and routes the reply back to the site. -/// Mirrors the NotificationSubmit / RegisterNotificationOutbox pattern. -/// -public class CentralCommunicationActorAuditTests : TestKit -{ - public CentralCommunicationActorAuditTests() : base(@"akka.loglevel = DEBUG") { } - - private IActorRef CreateActor(TimeSpan? auditIngestAskTimeout = null) - { - var mockRepo = Substitute.For(); - mockRepo.GetAllSitesAsync(Arg.Any()) - .Returns(new List()); - - var services = new ServiceCollection(); - services.AddScoped(_ => mockRepo); - var sp = services.BuildServiceProvider(); - - var transport = Substitute.For(); - return Sys.ActorOf(Props.Create(() => - new CentralCommunicationActor(sp, transport, auditIngestAskTimeout))); - } - - // C3 (Task 2.5): canonical ZB.MOM.WW.Audit.AuditEvent via the shared factory. - private static AuditEvent SampleAuditEvent() => - ScadaBridgeAuditEventFactory.Create( - channel: AuditChannel.ApiOutbound, - kind: AuditKind.ApiCall, - status: AuditStatus.Delivered); - - private static SiteCall SampleSiteCall() => new() - { - TrackedOperationId = TrackedOperationId.New(), - Channel = "OutboundApi", - Target = "ExternalSystemA", - SourceSite = "site1", - Status = "Delivered", - RetryCount = 0, - CreatedAtUtc = DateTime.UtcNow, - UpdatedAtUtc = DateTime.UtcNow, - IngestedAtUtc = DateTime.UtcNow, - }; - - [Fact] - public void IngestAuditEventsCommand_WithRegisteredProxy_ForwardsAndRoutesReplyToSender() - { - var actor = CreateActor(); - var auditProbe = CreateTestProbe(); - actor.Tell(new RegisterAuditIngest(auditProbe.Ref)); - - var evt = SampleAuditEvent(); - var cmd = new IngestAuditEventsCommand(new[] { evt }); - actor.Tell(cmd); - - // The audit-ingest proxy receives the command, with the original site - // sender preserved (Forward semantics). - auditProbe.ExpectMsg(cmd); - - // When the proxy replies, the actor routes it back to the original sender. - var reply = new IngestAuditEventsReply(new[] { evt.EventId }); - auditProbe.Reply(reply); - - var received = ExpectMsg(); - Assert.Equal(new[] { evt.EventId }, received.AcceptedEventIds); - } - - [Fact] - public void IngestAuditEventsCommand_WithNoProxyRegistered_RepliesEmptyAcceptedEventIds() - { - var actor = CreateActor(); - - actor.Tell(new IngestAuditEventsCommand(new[] { SampleAuditEvent() })); - - var reply = ExpectMsg(); - Assert.Empty(reply.AcceptedEventIds); - } - - [Fact] - public void IngestAuditEventsCommand_WhenProxyNeverReplies_PipesStatusFailureToSender() - { - // A short test-only Ask timeout (constructor seam) keeps the test fast — - // production uses the 30 s default. - var actor = CreateActor(auditIngestAskTimeout: TimeSpan.FromMilliseconds(200)); - var auditProbe = CreateTestProbe(); - actor.Tell(new RegisterAuditIngest(auditProbe.Ref)); - - var cmd = new IngestAuditEventsCommand(new[] { SampleAuditEvent() }); - actor.Tell(cmd); - - // The proxy receives the command but deliberately never replies. - auditProbe.ExpectMsg(cmd); - - // The Ask times out; PipeTo forwards the faulted task as a Status.Failure - // to the original sender. This is the real transient signal the site's - // own Ask faults on — it is NOT swallowed into an empty ack. - var failure = ExpectMsg(); - Assert.IsType(failure.Cause); - } - - [Fact] - public void IngestCachedTelemetryCommand_WithRegisteredProxy_ForwardsAndRoutesReplyToSender() - { - var actor = CreateActor(); - var auditProbe = CreateTestProbe(); - actor.Tell(new RegisterAuditIngest(auditProbe.Ref)); - - var entry = new CachedTelemetryEntry(SampleAuditEvent(), SampleSiteCall()); - var cmd = new IngestCachedTelemetryCommand(new[] { entry }); - actor.Tell(cmd); - - auditProbe.ExpectMsg(cmd); - - var reply = new IngestCachedTelemetryReply(new[] { entry.Audit.EventId }); - auditProbe.Reply(reply); - - var received = ExpectMsg(); - Assert.Equal(new[] { entry.Audit.EventId }, received.AcceptedEventIds); - } - - [Fact] - public void IngestCachedTelemetryCommand_WithNoProxyRegistered_RepliesEmptyAcceptedEventIds() - { - var actor = CreateActor(); - - var entry = new CachedTelemetryEntry(SampleAuditEvent(), SampleSiteCall()); - actor.Tell(new IngestCachedTelemetryCommand(new[] { entry })); - - var reply = ExpectMsg(); - Assert.Empty(reply.AcceptedEventIds); - } -} diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorClientLifecycleTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorClientLifecycleTests.cs index 0799ac3d..1cf67ab9 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorClientLifecycleTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorClientLifecycleTests.cs @@ -37,7 +37,7 @@ public class CentralCommunicationActorClientLifecycleTests : TestKit var transport = Substitute.For(); var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor( - provider, transport, (TimeSpan?)null))); + provider, transport))); // Trigger the refresh (also fires at PreStart, but drive it explicitly so // the assertion is deterministic). The load runs on a detached task and diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorReconcileTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorReconcileTests.cs index 423fbed1..8a872ca0 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorReconcileTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorReconcileTests.cs @@ -63,7 +63,7 @@ public class CentralCommunicationActorReconcileTests : TestKit var sp = services.BuildServiceProvider(); var transport = Substitute.For(); - var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, transport, (TimeSpan?)null))); + var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, transport))); // Node B is missing inst-B entirely → it should come back as a gap item. actor.Tell(new ReconcileSiteRequest( diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorTests.cs index 8b7993ab..21da6f91 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorTests.cs @@ -40,7 +40,7 @@ public class CentralCommunicationActorTests : TestKit var sp = services.BuildServiceProvider(); var transport = Substitute.For(); - var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, transport, (TimeSpan?)null))); + var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, transport))); return (actor, mockRepo); } @@ -60,7 +60,7 @@ public class CentralCommunicationActorTests : TestKit var transport = Substitute.For(); var centralActor = Sys.ActorOf( - Props.Create(() => new CentralCommunicationActor(sp, transport, (TimeSpan?)null))); + Props.Create(() => new CentralCommunicationActor(sp, transport))); var timestamp = DateTimeOffset.UtcNow; centralActor.Tell(new HeartbeatMessage("site1", "host1", true, timestamp)); @@ -87,7 +87,7 @@ public class CentralCommunicationActorTests : TestKit var transport = Substitute.For(); var centralActor = Sys.ActorOf( - Props.Create(() => new CentralCommunicationActor(sp, transport, (TimeSpan?)null))); + Props.Create(() => new CentralCommunicationActor(sp, transport))); var ts = DateTimeOffset.UtcNow; centralActor.Tell(new SiteHeartbeatReplica(new HeartbeatMessage("site-1", "host-a", true, ts))); @@ -116,7 +116,7 @@ public class CentralCommunicationActorTests : TestKit // The fix logs a Warning carrying the InvalidOperationException as the cause. EventFilter.Warning(contains: "Failed to load site addresses from the database").ExpectOne(() => { - Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, transport, (TimeSpan?)null))); + Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, transport))); }); } diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorTransportTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorTransportTests.cs index a7a4770f..52ce262b 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorTransportTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorTransportTests.cs @@ -40,7 +40,7 @@ public class CentralCommunicationActorTransportTests : TestKit var sp = services.BuildServiceProvider(); var transport = Substitute.For(); - var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, transport, (TimeSpan?)null))); + var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, transport))); return (actor, transport, repo); } diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsValidatorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsValidatorTests.cs index fb3edbe9..8df4e45b 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsValidatorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CommunicationOptionsValidatorTests.cs @@ -29,6 +29,59 @@ public class CommunicationOptionsValidatorTests Assert.Contains("DeploymentTimeout", result.FailureMessage); } + // ── Audit-ingest timeout ladder (arch-review phase-2 residual #2) ──────────── + + [Fact] + public void TheAuditIngestTimeoutLadder_IsStrictlyMonotonic_EndToEnd() + { + // 35 (site forward Ask) > 30 (gRPC deadline AND central's Ask of the ingest singleton) + // > 20 (actor budget) > 15 (SQL command). Ties are the bug this closes: the site Ask + // used to reuse NotificationForwardTimeout (30 s), so a slow-but-succeeding central + // write could be acked to a caller that had already given up and re-sent the batch. + var options = new CommunicationOptions(); + + Assert.Equal(TimeSpan.FromSeconds(35), options.AuditForwardTimeout); + Assert.True(options.AuditForwardTimeout > ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteStreamGrpcServer.AuditIngestAskTimeout); + Assert.Equal(TimeSpan.FromSeconds(30), ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteStreamGrpcServer.AuditIngestAskTimeout); + } + + [Fact] + public void AuditForwardTimeout_EqualToTheAskTimeout_IsRejected() + { + // Equality is precisely the pre-fix state, so the validator must refuse it, not just + // refuse something smaller. + var result = Validate(new CommunicationOptions + { + AuditForwardTimeout = ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteStreamGrpcServer.AuditIngestAskTimeout, + }); + + Assert.True(result.Failed); + Assert.Contains("AuditForwardTimeout", result.FailureMessage); + } + + [Fact] + public void AuditForwardTimeout_ShorterThanTheAskTimeout_IsRejected() + { + var result = Validate(new CommunicationOptions + { + AuditForwardTimeout = TimeSpan.FromSeconds(5), + }); + + Assert.True(result.Failed); + Assert.Contains("AuditForwardTimeout", result.FailureMessage); + } + + [Fact] + public void AuditForwardTimeout_IsStillConfigurableUpwards() + { + var result = Validate(new CommunicationOptions + { + AuditForwardTimeout = TimeSpan.FromMinutes(2), + }); + + Assert.True(result.Succeeded, result.FailureMessage); + } + [Fact] public void NonPositiveGrpcMaxConcurrentStreams_IsRejected() { diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/CentralControlGrpcServiceAuditIngestTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/CentralControlGrpcServiceAuditIngestTests.cs new file mode 100644 index 00000000..c937ca24 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/CentralControlGrpcServiceAuditIngestTests.cs @@ -0,0 +1,194 @@ +using Akka.Actor; +using Akka.TestKit.Xunit2; +using Grpc.Core; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using NSubstitute; +using ZB.MOM.WW.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Entities.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Types; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; +using ZB.MOM.WW.ScadaBridge.Communication.Actors; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; +using Microsoft.Extensions.DependencyInjection; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Tests.Grpc; + +/// +/// Audit Log (#23) site→central ingest routing on . +/// +/// The service Asks the audit-log-ingest singleton proxy DIRECTLY. It used to relay +/// through , which re-Asked the same proxy with the same +/// 30 s — a second hop whose inner Ask +/// expired at the same instant as the outer one, so it could only add latency. These tests pin +/// the direct dispatch, the wiring-race reply, and the fact that the relay is gone. +/// +/// +public class CentralControlGrpcServiceAuditIngestTests : TestKit +{ + private static ServerCallContext NewContext(CancellationToken ct = default) + { + var context = Substitute.For(); + context.CancellationToken.Returns(ct); + return context; + } + + private static CentralControlGrpcService CreateService() => new( + NullLogger.Instance, + Options.Create(new CommunicationOptions())); + + [Fact] + public async Task IngestAuditEvents_AsksTheIngestProxy_AndNeverTheCommunicationActor() + { + var ingest = CreateTestProbe(); + var control = CreateTestProbe(); + var service = CreateService(); + service.SetReady(control.Ref); + service.SetAuditIngestActor(ingest.Ref); + + var evt = SampleAuditEvent(); + var batch = new AuditEventBatch(); + batch.Events.Add(AuditEventDtoMapper.ToDto(evt)); + + var call = service.IngestAuditEvents(batch, NewContext()); + + var received = ingest.ExpectMsg(); + Assert.Equal(evt.EventId, Assert.Single(received.Events).EventId); + ingest.Reply(new IngestAuditEventsReply(new[] { evt.EventId })); + + var ack = await call; + Assert.Equal(evt.EventId.ToString(), Assert.Single(ack.AcceptedEventIds)); + + // The old relay hop is gone: the control-plane actor sees nothing at all. + control.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); + } + + [Fact] + public async Task IngestCachedTelemetry_AsksTheIngestProxy_AndNeverTheCommunicationActor() + { + var ingest = CreateTestProbe(); + var control = CreateTestProbe(); + var service = CreateService(); + service.SetReady(control.Ref); + service.SetAuditIngestActor(ingest.Ref); + + var evt = SampleAuditEvent(); + var batch = new CachedTelemetryBatch(); + batch.Packets.Add(new CachedTelemetryPacket + { + AuditEvent = AuditEventDtoMapper.ToDto(evt), + Operational = SiteCallDtoMapper.ToDto(SampleSiteCall()), + }); + + var call = service.IngestCachedTelemetry(batch, NewContext()); + + var received = ingest.ExpectMsg(); + Assert.Single(received.Entries); + ingest.Reply(new IngestCachedTelemetryReply(new[] { evt.EventId })); + + var ack = await call; + Assert.Equal(evt.EventId.ToString(), Assert.Single(ack.AcceptedEventIds)); + control.ExpectNoMsg(TimeSpan.FromMilliseconds(200)); + } + + [Fact] + public async Task IngestAuditEvents_BeforeTheIngestProxyIsWired_ReturnsAnEmptyAck() + { + // The singleton starts moments after SetReady, so this window is real. An empty ack + // (NOT a fault, NOT Unavailable) leaves the site's rows Pending for the next drain — + // byte-for-byte what the removed relay replied when its proxy was still null. + var service = CreateService(); + service.SetReady(CreateTestProbe().Ref); + + var batch = new AuditEventBatch(); + batch.Events.Add(AuditEventDtoMapper.ToDto(SampleAuditEvent())); + + var ack = await service.IngestAuditEvents(batch, NewContext()); + + Assert.Empty(ack.AcceptedEventIds); + } + + [Fact] + public async Task IngestCachedTelemetry_BeforeTheIngestProxyIsWired_ReturnsAnEmptyAck() + { + var service = CreateService(); + service.SetReady(CreateTestProbe().Ref); + + var batch = new CachedTelemetryBatch(); + batch.Packets.Add(new CachedTelemetryPacket + { + AuditEvent = AuditEventDtoMapper.ToDto(SampleAuditEvent()), + Operational = SiteCallDtoMapper.ToDto(SampleSiteCall()), + }); + + var ack = await service.IngestCachedTelemetry(batch, NewContext()); + + Assert.Empty(ack.AcceptedEventIds); + } + + [Fact] + public void SetAuditIngestActor_IsIndependentOfSetReady() + { + var service = CreateService(); + Assert.False(service.IsAuditIngestBound); + + service.SetAuditIngestActor(CreateTestProbe().Ref); + + Assert.True(service.IsAuditIngestBound); + Assert.False(service.IsReady); + } + + [Fact] + public void CentralCommunicationActor_NoLongerRelaysIngestCommands() + { + // Regression pin for the removed hop: the actor has no ingest receive at all, so an + // ingest command reaching it is an unhandled message with no reply — not a silent + // second path that could drift from the direct one. + var actor = CreateCentralCommunicationActor(); + + actor.Tell(new IngestAuditEventsCommand(new[] { SampleAuditEvent() }), TestActor); + actor.Tell( + new IngestCachedTelemetryCommand( + new[] { new CachedTelemetryEntry(SampleAuditEvent(), SampleSiteCall()) }), + TestActor); + + ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + } + + private IActorRef CreateCentralCommunicationActor() + { + var siteRepo = Substitute.For(); + siteRepo.GetAllSitesAsync(Arg.Any()) + .Returns(new List()); + + var services = new ServiceCollection(); + services.AddScoped(_ => siteRepo); + var sp = services.BuildServiceProvider(); + + var transport = Substitute.For(); + return Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, transport))); + } + + private static AuditEvent SampleAuditEvent() => + ScadaBridgeAuditEventFactory.Create( + channel: AuditChannel.ApiOutbound, + kind: AuditKind.ApiCall, + status: AuditStatus.Delivered, + sourceSiteId: "site-a"); + + private static SiteCall SampleSiteCall() => new() + { + TrackedOperationId = TrackedOperationId.New(), + Channel = "OutboundApi", + Target = "ExternalSystemA", + SourceSite = "site-a", + Status = "Delivered", + RetryCount = 0, + CreatedAtUtc = DateTime.UtcNow, + UpdatedAtUtc = DateTime.UtcNow, + IngestedAtUtc = DateTime.UtcNow, + }; +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/HealthReportAckTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/HealthReportAckTests.cs index 1016d89c..31067de0 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/HealthReportAckTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/HealthReportAckTests.cs @@ -55,7 +55,7 @@ public class HealthReportAckTests : TestKit var sp = services.BuildServiceProvider(); var transport = Substitute.For(); - var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, transport, (TimeSpan?)null))); + var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, transport))); actor.Tell(SampleReport(seq: 3)); var ack = ExpectMsg(); diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheServiceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheServiceTests.cs index 0c94c776..8735009e 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheServiceTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheServiceTests.cs @@ -81,7 +81,7 @@ public class SiteAlarmLiveCacheServiceTests : TestKit } private SiteAlarmLiveCacheService CreateService(TimeSpan linger, out CountingFactory factory, - int maxSubscribersPerSite = 200) + int maxSubscribersPerSite = 200, IReadOnlyList? enabledInstances = null) { // Site with gRPC addresses, and NO enabled instances → the seed fan-out returns // empty immediately (so IsLive flips true fast without any snapshot Asks). @@ -97,7 +97,7 @@ public class SiteAlarmLiveCacheServiceTests : TestKit var instanceRepo = Substitute.For(); instanceRepo.GetInstancesBySiteIdAsync(SiteId, Arg.Any()) - .Returns(new List()); + .Returns((IReadOnlyList)(enabledInstances ?? new List())); var services = new ServiceCollection(); services.AddScoped(_ => siteRepo); @@ -121,6 +121,45 @@ public class SiteAlarmLiveCacheServiceTests : TestKit return service; } + [Fact] + public void Seed_FanOut_Publishes_The_Instances_That_Failed_To_Answer() + { + // Arch-review phase-2 residual #4: the seed/reconcile fan-out already knows which + // Enabled instances failed to answer and used to discard it, forcing the Alarm Summary + // page to run a SECOND identical fan-out purely to rebuild that list. It is now + // published alongside the snapshot. + var instances = new List + { + new("inst-b") { Id = 2, SiteId = SiteId, State = InstanceState.Enabled }, + new("inst-a") { Id = 1, SiteId = SiteId, State = InstanceState.Enabled }, + // Disabled instances are never fanned out, so they can never be "not reporting". + new("inst-off") { Id = 3, SiteId = SiteId, State = InstanceState.Disabled }, + }; + + // The CommunicationService has no site actor wired, so every snapshot Ask faults — + // which is exactly the "instance did not answer" case. + var service = CreateService(TimeSpan.FromMilliseconds(200), out _, enabledInstances: instances); + + using var sub = service.Subscribe(SiteId, () => { }); + AwaitCondition(() => service.IsLive(SiteId), TimeSpan.FromSeconds(5)); + + AwaitCondition( + () => service.GetNotReportingInstances(SiteId).Count == 2, + TimeSpan.FromSeconds(5)); + + // Ordered by name (ordinal-ignore-case), matching AlarmSummaryService's poll output so + // the page renders identically whichever source supplied the list. + Assert.Equal(new[] { "inst-a", "inst-b" }, service.GetNotReportingInstances(SiteId)); + } + + [Fact] + public void NotReporting_Is_Empty_For_An_Unknown_Site() + { + var service = CreateService(TimeSpan.FromMilliseconds(200), out _); + + Assert.Empty(service.GetNotReportingInstances(SiteId)); + } + [Fact] public void First_Subscriber_Starts_One_Aggregator_Shared_By_Multiple_Viewers() { diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SyntheticHeartbeatTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SyntheticHeartbeatTests.cs new file mode 100644 index 00000000..e23b45f1 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SyntheticHeartbeatTests.cs @@ -0,0 +1,201 @@ +using System.Buffers.Binary; +using Akka.Actor; +using Akka.TestKit.Xunit2; +using Google.Protobuf.WellKnownTypes; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; +using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Health; +using ZB.MOM.WW.ScadaBridge.Communication.Actors; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; +using ZB.MOM.WW.ScadaBridge.HealthMonitoring; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Tests; + +/// +/// Arch-review phase-2 residual #3: 's failback probe reuses +/// the Heartbeat RPC to ask "does the preferred central endpoint answer again?". It is +/// emitted by a site's TRANSPORT layer, not by any node's heartbeat timer, so central must not +/// count it as liveness — otherwise a site whose real heartbeats had stopped keeps looking alive +/// on the health dashboard for as long as its transport keeps probing. +/// +/// The contract is the explicit additive Synthetic flag (HeartbeatDto.synthetic, +/// proto field 5), NOT the failback-probe hostname, which is a log label only. +/// +/// +public class SyntheticHeartbeatTests : TestKit +{ + private static readonly DateTimeOffset T0 = + new(2026, 8, 14, 9, 0, 0, TimeSpan.Zero); + + // ── The consumer: central skips liveness bookkeeping for a synthetic heartbeat ─── + + [Fact] + public void SyntheticHeartbeat_DoesNotMarkTheHealthAggregator() + { + var (actor, aggregator) = CreateCentralActor(); + + actor.Tell(new HeartbeatMessage("site-1", CentralChannelProvider.SyntheticProbeHostname, + IsActive: false, Timestamp: T0, Synthetic: true)); + + // Give the actor a real chance to (wrongly) mark before asserting the negative. + ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + aggregator.DidNotReceiveWithAnyArgs().MarkHeartbeat(default!, default); + } + + [Fact] + public void RealHeartbeat_StillMarksTheHealthAggregator() + { + // The guard must be keyed on the flag alone — an ordinary heartbeat is unaffected, + // including one from a node that predates the field (proto3 defaults it to false). + var (actor, aggregator) = CreateCentralActor(); + + actor.Tell(new HeartbeatMessage("site-1", "node-a", IsActive: true, Timestamp: T0)); + + AwaitAssert(() => aggregator.Received(1).MarkHeartbeat("site-1", T0)); + } + + [Fact] + public void SyntheticHeartbeatReplica_IsAlsoSkipped() + { + // Belt-and-braces on the last hop before the aggregator: a peer central node that + // predates the flag could still replicate one. + var (actor, aggregator) = CreateCentralActor(); + + actor.Tell(new SiteHeartbeatReplica(new HeartbeatMessage( + "site-1", CentralChannelProvider.SyntheticProbeHostname, + IsActive: false, Timestamp: T0, Synthetic: true))); + + ExpectNoMsg(TimeSpan.FromMilliseconds(300)); + aggregator.DidNotReceiveWithAnyArgs().MarkHeartbeat(default!, default); + } + + // ── The wire contract: the flag survives the round trip ───────────────────────── + + [Theory] + [InlineData(true)] + [InlineData(false)] + public void TheSyntheticFlag_RoundTripsThroughTheDto(bool synthetic) + { + var msg = new HeartbeatMessage("site-1", "node-a", IsActive: true, Timestamp: T0, + Synthetic: synthetic); + + var back = CentralControlDtoMapper.FromDto(CentralControlDtoMapper.ToDto(msg)); + + Assert.Equal(synthetic, back.Synthetic); + Assert.Equal(msg with { Synthetic = synthetic }, back); + } + + [Fact] + public void ADtoFromAnOlderSite_DefaultsToNotSynthetic() + { + // proto3 default: a peer that never sets field 5 sends a REAL heartbeat, as before. + var dto = new HeartbeatDto + { + SiteId = "site-1", + NodeHostname = "node-a", + IsActive = true, + Timestamp = Timestamp.FromDateTimeOffset(T0), + }; + + Assert.False(CentralControlDtoMapper.FromDto(dto).Synthetic); + } + + // ── The producer: the failback probe marks itself synthetic ───────────────────── + + [Fact] + public async Task TheFailbackProbe_MarksItsHeartbeatSynthetic() + { + // Two endpoints so a flip is possible; the capture handler answers nothing useful, so + // the probe faults and re-arms — we only care about the request it put on the wire. + var capture = new HeartbeatCapturingHandler(); + using var provider = new CentralChannelProvider( + new[] { "http://central-a:8083", "http://central-b:8083" }, + new FixedPskProvider("k"), + "site-1", + new CommunicationOptions(), + NullLogger.Instance, + handlerFactory: _ => capture, + probeDeadline: TimeSpan.FromSeconds(2), + backoffBase: TimeSpan.FromMilliseconds(20), + backoffCap: TimeSpan.FromMilliseconds(50)); + + // Off the preferred endpoint → the background failback probe arms. + provider.ReportUnavailable(0); + + var probe = await capture.WaitForHeartbeatAsync(TimeSpan.FromSeconds(10)); + + Assert.True(probe.Synthetic); + Assert.Equal("site-1", probe.SiteId); + // The hostname is a human-readable label that rides ALONGSIDE the flag; central keys + // its skip on the flag, never on this string. + Assert.Equal(CentralChannelProvider.SyntheticProbeHostname, probe.NodeHostname); + Assert.False(probe.IsActive); + } + + private (IActorRef Actor, ICentralHealthAggregator Aggregator) CreateCentralActor() + { + var siteRepo = Substitute.For(); + siteRepo.GetAllSitesAsync(Arg.Any()).Returns(new List()); + + var aggregator = Substitute.For(); + + var services = new ServiceCollection(); + services.AddScoped(_ => siteRepo); + services.AddSingleton(aggregator); + var sp = services.BuildServiceProvider(); + + var actor = Sys.ActorOf(Props.Create(() => + new CentralCommunicationActor(sp, Substitute.For()))); + return (actor, aggregator); + } + + private sealed class FixedPskProvider(string key) : ISitePskProvider + { + public ValueTask GetAsync(string siteId, CancellationToken ct) => new(key); + public void Invalidate(string siteId) { } + } + + /// + /// Captures the first Heartbeat request body and decodes the length-prefixed gRPC + /// frame back into a . The response is deliberately a bare 500 so + /// the probe treats the endpoint as still down; the provider swallows that and re-arms. + /// + private sealed class HeartbeatCapturingHandler : HttpMessageHandler + { + private readonly TaskCompletionSource _captured = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + public async Task WaitForHeartbeatAsync(TimeSpan timeout) + { + var completed = await Task.WhenAny(_captured.Task, Task.Delay(timeout)); + Assert.Same(_captured.Task, completed); + return await _captured.Task; + } + + protected override async Task SendAsync( + HttpRequestMessage request, CancellationToken cancellationToken) + { + if (request.Content is not null && + request.RequestUri?.AbsolutePath.EndsWith("/Heartbeat", StringComparison.Ordinal) == true) + { + var body = await request.Content.ReadAsByteArrayAsync(cancellationToken); + // gRPC frame: 1 compression byte + 4-byte big-endian length + payload. + if (body.Length >= 5) + { + var length = BinaryPrimitives.ReadInt32BigEndian(body.AsSpan(1, 4)); + _captured.TrySetResult( + HeartbeatDto.Parser.ParseFrom(body.AsSpan(5, length).ToArray())); + } + } + + return new HttpResponseMessage(System.Net.HttpStatusCode.InternalServerError) + { + Version = request.Version, + }; + } + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CentralControlEndToEndTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CentralControlEndToEndTests.cs index 63d806aa..dc519e3f 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CentralControlEndToEndTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CentralControlEndToEndTests.cs @@ -62,6 +62,10 @@ public class CentralControlEndToEndTests : IAsyncLifetime NullLogger.Instance, Options.Create(new CommunicationOptions())); _service.SetReady(stub); + // The two ingest RPCs Ask the audit-ingest singleton DIRECTLY (no CentralCommunicationActor + // relay any more), so the service needs its own proxy handed over. Reusing the same stub + // keeps one actor answering both shapes. + _service.SetAuditIngestActor(stub); var pskProvider = new MapPskProvider(new Dictionary { @@ -174,7 +178,7 @@ public class CentralControlEndToEndTests : IAsyncLifetime // ---- One RPC per shape: unary (above) + the ingest bridge ---- [Fact] - public async Task IngestAuditEvents_DecodesTheBatch_AsksTheActor_EncodesTheAck() + public async Task IngestAuditEvents_DecodesTheBatch_AsksTheIngestProxy_EncodesTheAck() { var client = Client(SiteAKey, SiteA); @@ -185,12 +189,13 @@ public class CentralControlEndToEndTests : IAsyncLifetime var ack = await client.IngestAuditEventsAsync(batch); // The stub actor accepts one deterministic id per non-empty batch; its presence proves - // the full DTO→command→Ask→reply→ack bridge ran, gated call and all. + // the full DTO→command→Ask→reply→ack bridge ran straight to the ingest proxy, gated + // call and all. Assert.Contains(AcceptedId.ToString(), ack.AcceptedEventIds); } [Fact] - public async Task IngestAuditEvents_EmptyBatch_ShortCircuits_WithoutAskingTheActor() + public async Task IngestAuditEvents_EmptyBatch_ShortCircuits_WithoutAskingTheIngestProxy() { // Even the empty-batch fast path is behind the gate — it still needs a valid key. var client = Client(SiteAKey, SiteA); @@ -238,9 +243,9 @@ public class CentralControlEndToEndTests : IAsyncLifetime } /// - /// Minimal stand-in for CentralCommunicationActor: answers the two RPC shapes this - /// test exercises. Replies straight to the Ask's temp sender, exactly as the real actor's - /// Forward/PipeTo paths do. + /// Minimal stand-in for both actors the service Asks — CentralCommunicationActor for + /// the unary control RPCs and the audit-log-ingest singleton for the ingest ones. + /// Replies straight to the Ask's temp sender, exactly as the real actors do. /// private sealed class StubCentralActor : ReceiveActor { diff --git a/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/AuditLog/SiteAuditPushFlowTests.cs b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/AuditLog/SiteAuditPushFlowTests.cs index f77ec70f..abd62283 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/AuditLog/SiteAuditPushFlowTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/AuditLog/SiteAuditPushFlowTests.cs @@ -59,12 +59,24 @@ public class SiteAuditPushFlowTests : TestKit private sealed class BridgeCentralTransport : ICentralTransport { private readonly IActorRef _central; - public BridgeCentralTransport(IActorRef central) => _central = central; + private readonly IActorRef _auditIngest; + + /// Stands in for the central control plane's non-audit RPCs. + /// + /// The central audit-ingest singleton. The two ingest RPCs go straight here, mirroring + /// CentralControlGrpcService, which Asks the proxy directly rather than relaying + /// through CentralCommunicationActor. + /// + public BridgeCentralTransport(IActorRef central, IActorRef auditIngest) + { + _central = central; + _auditIngest = auditIngest; + } public void SubmitNotification(NotificationSubmit message, IActorRef replyTo) => _central.Tell(message, replyTo); public void QueryNotificationStatus(NotificationStatusQuery message, IActorRef replyTo) => _central.Tell(message, replyTo); - public void IngestAuditEvents(IngestAuditEventsCommand message, IActorRef replyTo) => _central.Tell(message, replyTo); - public void IngestCachedTelemetry(IngestCachedTelemetryCommand message, IActorRef replyTo) => _central.Tell(message, replyTo); + public void IngestAuditEvents(IngestAuditEventsCommand message, IActorRef replyTo) => _auditIngest.Tell(message, replyTo); + public void IngestCachedTelemetry(IngestCachedTelemetryCommand message, IActorRef replyTo) => _auditIngest.Tell(message, replyTo); public void ReconcileSite(ReconcileSiteRequest message, IActorRef replyTo) => _central.Tell(message, replyTo); public void ReportSiteHealth(SiteHealthReport message, IActorRef replyTo) => _central.Tell(message, replyTo); public void SendHeartbeat(HeartbeatMessage message, IActorRef self) => _central.Tell(message, self); @@ -144,9 +156,11 @@ public class SiteAuditPushFlowTests : TestKit centralRepo, NullLogger.Instance))); - // Real CentralCommunicationActor. Its periodic site-address refresh - // resolves an ISiteRepository from this provider; an empty result keeps - // the refresh a clean no-op and never touches the audit-ingest path. + // Real CentralCommunicationActor. It is NOT on the audit path any more (the + // central-hosted CentralControlGrpcService Asks the ingest singleton directly), but the + // bridge transport still routes notifications/health/reconcile through it, so it is + // constructed exactly as production does. Its periodic site-address refresh resolves an + // ISiteRepository from this provider; an empty result keeps the refresh a clean no-op. var siteRepo = Substitute.For(); siteRepo.GetAllSitesAsync().Returns(Array.Empty()); var centralServices = new ServiceCollection(); @@ -155,9 +169,7 @@ public class SiteAuditPushFlowTests : TestKit var centralCommActor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor( centralProvider, - Substitute.For(), - TimeSpan.FromSeconds(5)))); - centralCommActor.Tell(new RegisterAuditIngest(ingestActor)); + Substitute.For()))); // ── Site side ───────────────────────────────────────────────────── // Real SqliteAuditWriter on a file-backed SQLite db (the site hot-path @@ -177,7 +189,7 @@ public class SiteAuditPushFlowTests : TestKit CreateTestProbe().Ref, // deployment-manager proxy is unused here null, null, - new BridgeCentralTransport(centralCommActor)))); + new BridgeCentralTransport(centralCommActor, ingestActor)))); // The production site audit push client — the unit under integration. var auditClient = new SiteCommunicationAuditClient(