From bfd2cc7e6f1ed64cc40c49a0e3347ef5e1f8391d Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 9 Jul 2026 08:57:06 -0400 Subject: [PATCH] fix(audit-log): reconciliation pulls fail over to site NodeB when NodeA is unreachable MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SiteEntry gains an additive FallbackGrpcEndpoint; SiteEnumerator now picks NodeA else NodeB as primary (a NodeB-only site is no longer skipped) and carries a distinct NodeB as the fallback. Both GrpcPull{AuditEvents,SiteCalls} clients dial the primary first and, on a transport fault (Unavailable/DeadlineExceeded/ Cancelled/HttpRequestException/SocketException/OperationCanceled), fail over to NodeB once before collapsing to empty — a mapping/unexpected fault does NOT fail over. Keeps the reconciliation loss-recovery net available during a NodeA outage. --- docs/requirements/Component-AuditLog.md | 12 ++ .../Central/GrpcPullAuditEventsClient.cs | 114 ++++++++++++------ .../Central/GrpcPullSiteCallsClient.cs | 110 +++++++++++------ .../Central/ISiteEnumerator.cs | 33 +++-- .../Central/SiteEnumerator.cs | 41 ++++--- .../Central/GrpcPullAuditEventsClientTests.cs | 64 ++++++++++ .../Central/GrpcPullSiteCallsClientTests.cs | 70 +++++++++++ .../Central/SiteEnumeratorTests.cs | 38 ++++++ 8 files changed, 382 insertions(+), 100 deletions(-) diff --git a/docs/requirements/Component-AuditLog.md b/docs/requirements/Component-AuditLog.md index b2f14675..ee470dc2 100644 --- a/docs/requirements/Component-AuditLog.md +++ b/docs/requirements/Component-AuditLog.md @@ -279,6 +279,18 @@ non-draining (e.g., telemetry actor wedged), central issues a are flipped to `ForwardState = 'Reconciled'` site-side. Same self-healing pattern as Site Call Audit's reconciliation of `SiteCalls`. +**Endpoint resolution & NodeB failover.** Each pull dials the site's `NodeA` +gRPC address first; if `NodeA` is blank the site's `NodeB` address becomes the +primary (a NodeB-only site is no longer skipped — only a site with *both* +addresses blank is). When a distinct `NodeB` address also exists it is carried +as a fallback: a transport fault against the primary (site offline, deadline +exceeded, cancelled, connection error) fails over to `NodeB` ONCE before +collapsing to an empty batch. So during a NodeA outage the loss-recovery net +stays available. The failover is scoped to transport faults — a +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 (tracked follow-up).** The `PullAuditEvents` cursor is still a > single `sinceUtc` timestamp. Site Call Audit's pull now uses a composite > `(UpdatedAtUtc, TrackedOperationId)` keyset to avoid a single-timestamp pin diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullAuditEventsClient.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullAuditEventsClient.cs index 6d7d3297..d85731c4 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullAuditEventsClient.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullAuditEventsClient.cs @@ -76,7 +76,7 @@ public sealed class GrpcPullAuditEventsClient : IPullAuditEventsClient int batchSize, CancellationToken ct) { - var endpoint = await ResolveEndpointAsync(siteId, ct).ConfigureAwait(false); + var (endpoint, fallback) = await ResolveEndpointsAsync(siteId, ct).ConfigureAwait(false); if (endpoint is null) { // No gRPC address registered for the site — absence of an address is @@ -95,44 +95,27 @@ public sealed class GrpcPullAuditEventsClient : IPullAuditEventsClient BatchSize = batchSize, }; - ProtoPullResponse reply; - try + var (reply, transportFault) = await TryInvokeAsync(endpoint, request, siteId, ct) + .ConfigureAwait(false); + + // NodeB failover (Task 18): a transport fault against the primary (NodeA) + // dials the fallback (NodeB) ONCE before collapsing to empty, so a NodeA + // outage doesn't take the reconciliation loss-recovery net offline. Only + // the transport-fault set triggers this (not a mapping/unexpected fault), + // and not when the caller's token is already cancelled (host shutdown). + if (reply is null && transportFault + && !string.IsNullOrWhiteSpace(fallback) + && !string.Equals(fallback, endpoint, StringComparison.Ordinal) + && !ct.IsCancellationRequested) { - reply = await _invoker.InvokeAsync(endpoint, request, ct).ConfigureAwait(false); + _logger.LogInformation( + "PullAuditEvents failing over from primary {Primary} to NodeB {Fallback} for site {SiteId}.", + endpoint, fallback, siteId); + (reply, _) = await TryInvokeAsync(fallback, request, siteId, ct).ConfigureAwait(false); } - catch (RpcException ex) when (IsTolerable(ex.StatusCode)) + + if (reply is null) { - _logger.LogDebug(ex, - "PullAuditEvents tolerable transport fault for site {SiteId} ({Endpoint}): {Status}. Returning empty batch.", - siteId, endpoint, ex.StatusCode); - return Empty; - } - catch (Exception ex) when (ex is HttpRequestException or System.Net.Sockets.SocketException) - { - _logger.LogDebug(ex, - "PullAuditEvents connection-layer fault for site {SiteId} ({Endpoint}). Returning empty batch.", - siteId, endpoint); - return Empty; - } - catch (OperationCanceledException) - { - // Reconciliation tick was cancelled — either the caller's token - // (host shutdown / scope dispose) or an internal gRPC deadline / - // linked-CTS cancellation. Both are tolerable for a best-effort - // pull; collapse to empty rather than letting an internal - // cancellation land noisily in the catch-all below. - return Empty; - } - catch (Exception ex) - { - // Any other fault (e.g. a malformed reply that fails DTO mapping - // below would actually surface here only if mapping moved inline, - // but a non-RpcException transport fault wrapper lands here too). - // Audit reconciliation is best-effort; swallow to empty rather than - // throw — the actor's per-site guard would only re-catch it. - _logger.LogWarning(ex, - "PullAuditEvents unexpected fault for site {SiteId} ({Endpoint}). Returning empty batch.", - siteId, endpoint); return Empty; } @@ -147,7 +130,60 @@ public sealed class GrpcPullAuditEventsClient : IPullAuditEventsClient return new PullAuditEventsResponse(events, reply.MoreAvailable); } - private async Task ResolveEndpointAsync(string siteId, CancellationToken ct) + /// + /// Issues one pull against , classifying faults. + /// Returns (reply, false) on success; (null, true) on a + /// tolerable TRANSPORT fault (the failover-eligible set: + /// / / + /// / / + /// SocketException / ); and + /// (null, false) on any other (mapping/unexpected) fault, which + /// collapses to empty WITHOUT a fallback dial. + /// + private async Task<(ProtoPullResponse? Reply, bool TransportFault)> TryInvokeAsync( + string endpoint, ProtoPullRequest request, string siteId, CancellationToken ct) + { + try + { + var reply = await _invoker.InvokeAsync(endpoint, request, ct).ConfigureAwait(false); + return (reply, false); + } + catch (RpcException ex) when (IsTolerable(ex.StatusCode)) + { + _logger.LogDebug(ex, + "PullAuditEvents tolerable transport fault for site {SiteId} ({Endpoint}): {Status}.", + siteId, endpoint, ex.StatusCode); + return (null, true); + } + catch (Exception ex) when (ex is HttpRequestException or System.Net.Sockets.SocketException) + { + _logger.LogDebug(ex, + "PullAuditEvents connection-layer fault for site {SiteId} ({Endpoint}).", + siteId, endpoint); + return (null, true); + } + catch (OperationCanceledException) + { + // Reconciliation tick was cancelled — either the caller's token + // (host shutdown / scope dispose) or an internal gRPC deadline / + // linked-CTS cancellation. Tolerable for a best-effort pull. + return (null, true); + } + catch (Exception ex) + { + // Any other fault (e.g. a malformed reply that fails DTO mapping). + // Audit reconciliation is best-effort; swallow to empty rather than + // throw — and do NOT fail over, since a second node would hit the + // same non-transport fault. + _logger.LogWarning(ex, + "PullAuditEvents unexpected fault for site {SiteId} ({Endpoint}). Returning empty batch.", + siteId, endpoint); + return (null, false); + } + } + + private async Task<(string? Primary, string? Fallback)> ResolveEndpointsAsync( + string siteId, CancellationToken ct) { var sites = await _sites.EnumerateAsync(ct).ConfigureAwait(false); foreach (var site in sites) @@ -155,10 +191,10 @@ public sealed class GrpcPullAuditEventsClient : IPullAuditEventsClient if (string.Equals(site.SiteId, siteId, StringComparison.Ordinal) && !string.IsNullOrWhiteSpace(site.GrpcEndpoint)) { - return site.GrpcEndpoint; + return (site.GrpcEndpoint, site.FallbackGrpcEndpoint); } } - return null; + return (null, null); } private static readonly PullAuditEventsResponse Empty = diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullSiteCallsClient.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullSiteCallsClient.cs index 6d82ad24..d923af69 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullSiteCallsClient.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullSiteCallsClient.cs @@ -84,7 +84,7 @@ public sealed class GrpcPullSiteCallsClient : IPullSiteCallsClient int batchSize, CancellationToken ct) { - var endpoint = await ResolveEndpointAsync(siteId, ct).ConfigureAwait(false); + var (endpoint, fallback) = await ResolveEndpointsAsync(siteId, ct).ConfigureAwait(false); if (endpoint is null) { // No gRPC address registered for the site — a configuration decision @@ -107,41 +107,27 @@ public sealed class GrpcPullSiteCallsClient : IPullSiteCallsClient AfterId = afterId ?? string.Empty, }; - ProtoPullResponse reply; - try + var (reply, transportFault) = await TryInvokeAsync(endpoint, request, siteId, ct) + .ConfigureAwait(false); + + // NodeB failover (Task 18): a transport fault against the primary (NodeA) + // dials the fallback (NodeB) ONCE before collapsing to empty, so a NodeA + // outage doesn't take the reconciliation loss-recovery net offline. Only + // the transport-fault set triggers this (not a mapping/unexpected fault), + // and not when the caller's token is already cancelled (host shutdown). + if (reply is null && transportFault + && !string.IsNullOrWhiteSpace(fallback) + && !string.Equals(fallback, endpoint, StringComparison.Ordinal) + && !ct.IsCancellationRequested) { - reply = await _invoker.InvokeAsync(endpoint, request, ct).ConfigureAwait(false); + _logger.LogInformation( + "PullSiteCalls failing over from primary {Primary} to NodeB {Fallback} for site {SiteId}.", + endpoint, fallback, siteId); + (reply, _) = await TryInvokeAsync(fallback, request, siteId, ct).ConfigureAwait(false); } - catch (RpcException ex) when (IsTolerable(ex.StatusCode)) + + if (reply is null) { - _logger.LogDebug(ex, - "PullSiteCalls tolerable transport fault for site {SiteId} ({Endpoint}): {Status}. Returning empty batch.", - siteId, endpoint, ex.StatusCode); - return Empty; - } - catch (Exception ex) when (ex is HttpRequestException or System.Net.Sockets.SocketException) - { - _logger.LogDebug(ex, - "PullSiteCalls connection-layer fault for site {SiteId} ({Endpoint}). Returning empty batch.", - siteId, endpoint); - return Empty; - } - catch (OperationCanceledException) - { - // Reconciliation tick cancelled — caller token (host shutdown) or an - // internal gRPC deadline / linked-CTS cancellation. Both tolerable for - // a best-effort pull; collapse to empty rather than landing noisily in - // the catch-all below. - return Empty; - } - catch (Exception ex) - { - // Any other fault. Reconciliation is best-effort; swallow to empty - // rather than throw — the (future) actor's per-site guard would only - // re-catch it. - _logger.LogWarning(ex, - "PullSiteCalls unexpected fault for site {SiteId} ({Endpoint}). Returning empty batch.", - siteId, endpoint); return Empty; } @@ -173,7 +159,59 @@ public sealed class GrpcPullSiteCallsClient : IPullSiteCallsClient return new PullSiteCallsResponse(siteCalls, reply.MoreAvailable); } - private async Task ResolveEndpointAsync(string siteId, CancellationToken ct) + /// + /// Issues one pull against , classifying faults. + /// Returns (reply, false) on success; (null, true) on a + /// tolerable TRANSPORT fault (the failover-eligible set: + /// / / + /// / / + /// SocketException / ); and + /// (null, false) on any other (mapping/unexpected) fault, which + /// collapses to empty WITHOUT a fallback dial. + /// + private async Task<(ProtoPullResponse? Reply, bool TransportFault)> TryInvokeAsync( + string endpoint, ProtoPullRequest request, string siteId, CancellationToken ct) + { + try + { + var reply = await _invoker.InvokeAsync(endpoint, request, ct).ConfigureAwait(false); + return (reply, false); + } + catch (RpcException ex) when (IsTolerable(ex.StatusCode)) + { + _logger.LogDebug(ex, + "PullSiteCalls tolerable transport fault for site {SiteId} ({Endpoint}): {Status}.", + siteId, endpoint, ex.StatusCode); + return (null, true); + } + catch (Exception ex) when (ex is HttpRequestException or System.Net.Sockets.SocketException) + { + _logger.LogDebug(ex, + "PullSiteCalls connection-layer fault for site {SiteId} ({Endpoint}).", + siteId, endpoint); + return (null, true); + } + catch (OperationCanceledException) + { + // Reconciliation tick cancelled — caller token (host shutdown) or an + // internal gRPC deadline / linked-CTS cancellation. Tolerable for a + // best-effort pull. + return (null, true); + } + catch (Exception ex) + { + // Any other fault. Reconciliation is best-effort; swallow to empty + // rather than throw — and do NOT fail over, since a second node would + // hit the same non-transport fault. + _logger.LogWarning(ex, + "PullSiteCalls unexpected fault for site {SiteId} ({Endpoint}). Returning empty batch.", + siteId, endpoint); + return (null, false); + } + } + + private async Task<(string? Primary, string? Fallback)> ResolveEndpointsAsync( + string siteId, CancellationToken ct) { var sites = await _sites.EnumerateAsync(ct).ConfigureAwait(false); foreach (var site in sites) @@ -181,10 +219,10 @@ public sealed class GrpcPullSiteCallsClient : IPullSiteCallsClient if (string.Equals(site.SiteId, siteId, StringComparison.Ordinal) && !string.IsNullOrWhiteSpace(site.GrpcEndpoint)) { - return site.GrpcEndpoint; + return (site.GrpcEndpoint, site.FallbackGrpcEndpoint); } } - return null; + return (null, null); } private static readonly PullSiteCallsResponse Empty = diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/ISiteEnumerator.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/ISiteEnumerator.cs index 25a5a4c7..94686ab3 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/ISiteEnumerator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/ISiteEnumerator.cs @@ -8,13 +8,13 @@ namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central; /// /// /// The production implementation wraps ISiteRepository.GetAllSitesAsync -/// and projects each Site to a using the -/// site's configured GrpcNodeAAddress. This is a NodeA-only first cut: -/// sites with a blank GrpcNodeAAddress are silently SKIPPED — the -/// reconciliation pull cannot reach them, but absence of an address is a -/// configuration decision, not a runtime error. NodeB-fallback endpoint -/// selection (dial NodeB when NodeA is unset/unreachable) is a follow-up -/// (mirrors the comment in SiteEnumerator.cs). +/// and projects each Site to a . The primary dial +/// target is GrpcNodeAAddress when set, otherwise GrpcNodeBAddress; +/// when both a distinct NodeA and NodeB address exist, NodeB rides along as +/// so the reconciliation pull can +/// fail over per call (Task 18). A site is skipped only when BOTH addresses are +/// blank — the reconciliation pull cannot reach it, but absence of any address +/// is a configuration decision, not a runtime error. /// public interface ISiteEnumerator { @@ -30,8 +30,19 @@ public interface ISiteEnumerator /// /// One reconciliation target: the site identifier the actor uses as the -/// cursor key and the gRPC endpoint dials -/// to issue the pull. Endpoint is the bare authority (e.g. http://siteA:8083); -/// transport selection (TLS, keepalive, etc.) is the client's concern. +/// cursor key and the gRPC endpoint the pull clients dial to issue the pull. +/// Endpoints are bare authorities (e.g. http://siteA:8083); transport +/// selection (TLS, keepalive, etc.) is the client's concern. /// -public sealed record SiteEntry(string SiteId, string GrpcEndpoint); +/// The site identifier used as the reconciliation cursor key. +/// The primary gRPC authority to dial (NodeA where configured, else NodeB). +/// +/// Optional secondary gRPC authority (NodeB) dialed once when the primary raises +/// a transport fault during a pull (Task 18). Null when no distinct second +/// address is configured. Additive — absent value preserves the prior +/// single-endpoint behaviour. +/// +public sealed record SiteEntry( + string SiteId, + string GrpcEndpoint, + string? FallbackGrpcEndpoint = null); diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteEnumerator.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteEnumerator.cs index 357ee4ee..50af3d0e 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteEnumerator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteEnumerator.cs @@ -21,17 +21,18 @@ namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central; /// per-tick scope pattern in . /// /// -/// Blank-address skip. Sites with no GrpcNodeAAddress configured -/// are silently skipped: the reconciliation pull cannot dial them, but absence -/// of an address is a configuration decision, not a runtime error (per the +/// Blank-address skip. A site is skipped only when BOTH +/// GrpcNodeAAddress and GrpcNodeBAddress are blank: the +/// reconciliation pull cannot dial it, but absence of any address is a +/// configuration decision, not a runtime error (per the /// contract). /// /// -/// NodeA-only first cut. This implementation always uses NodeA's gRPC -/// address. NodeA/NodeB failover endpoint selection (dial NodeB when NodeA is -/// unreachable) is a follow-up — the shape already -/// carries a single endpoint, so failover will live in the puller/client, not -/// here. +/// NodeA/NodeB failover (Task 18). The primary dial target is NodeA when +/// configured, otherwise NodeB (a site with only NodeB is no longer skipped). +/// When a distinct NodeB address also exists it rides along as +/// so the pull clients can fail over +/// to it once per call when the primary raises a transport fault. /// /// public sealed class SiteEnumerator : ISiteEnumerator @@ -60,16 +61,28 @@ public sealed class SiteEnumerator : ISiteEnumerator var entries = new List(sites.Count); foreach (var site in sites) { - // First cut: NodeA's gRPC address is the dial target. NodeA/NodeB - // failover endpoint selection is a follow-up. - if (string.IsNullOrWhiteSpace(site.GrpcNodeAAddress)) + // Primary dial target: NodeA when configured, otherwise NodeB (Task 18) — + // a site with only a NodeB address is no longer skipped. + var nodeA = site.GrpcNodeAAddress; + var nodeB = site.GrpcNodeBAddress; + var primary = !string.IsNullOrWhiteSpace(nodeA) ? nodeA : nodeB; + + if (string.IsNullOrWhiteSpace(primary)) { + // Both addresses blank — nothing to dial (a configuration + // decision, not a runtime error). continue; } - // The IsNullOrWhiteSpace guard above proves GrpcNodeAAddress is - // non-null here; explicit null-forgiving for clarity. - entries.Add(new SiteEntry(site.SiteIdentifier, site.GrpcNodeAAddress!)); + // NodeB rides along as the failover endpoint only when it is a + // distinct, non-blank address (i.e. NodeA is the primary and NodeB + // differs); when NodeB IS the primary there is no separate fallback. + var fallback = + !string.IsNullOrWhiteSpace(nodeB) && !string.Equals(nodeB, primary, StringComparison.Ordinal) + ? nodeB + : null; + + entries.Add(new SiteEntry(site.SiteIdentifier, primary!, fallback)); } return entries; 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 7664b36c..e76620b3 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/GrpcPullAuditEventsClientTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/GrpcPullAuditEventsClientTests.cs @@ -71,6 +71,26 @@ public class GrpcPullAuditEventsClientTests } } + /// + /// Per-endpoint fake invoker (Task 18): routes dials by endpoint string so a + /// NodeA-down / NodeB-up failover can be scripted. Records dialed endpoints. + /// + private sealed class PerEndpointInvoker : GrpcPullAuditEventsClient.IPullAuditEventsInvoker + { + private readonly Dictionary> _byEndpoint; + public List Dialed { get; } = new(); + + public PerEndpointInvoker(Dictionary> byEndpoint) => + _byEndpoint = byEndpoint; + + public Task InvokeAsync( + string endpoint, ProtoPullRequest request, CancellationToken ct) + { + Dialed.Add(endpoint); + return Task.FromResult(_byEndpoint[endpoint]()); + } + } + private static AuditEventDto Dto(Guid id, DateTime occurredAtUtc) => AuditEventDtoMapper.ToDto(ScadaBridgeAuditEventFactory.Create( eventId: id, @@ -212,4 +232,48 @@ public class GrpcPullAuditEventsClientTests Assert.Empty(result.Events); Assert.False(result.MoreAvailable); } + + [Fact] + public async Task PullAsync_fails_over_to_NodeB_when_primary_raises_transport_fault() + { + // Task 18: NodeA (primary) is unreachable, NodeB (fallback) answers — the + // pull must return NodeB's events rather than collapsing to empty. + var id = Guid.NewGuid(); + var fallbackResponse = new ProtoPullResponse { MoreAvailable = false }; + fallbackResponse.Events.Add(Dto(id, BaseTime)); + + var invoker = new PerEndpointInvoker(new Dictionary> + { + ["http://node-a:8083"] = () => throw new RpcException(new Status(StatusCode.Unavailable, "node A down")), + ["http://node-b:8083"] = () => fallbackResponse, + }); + var sut = new GrpcPullAuditEventsClient( + new StaticEnumerator(new SiteEntry("site-a", "http://node-a:8083", "http://node-b:8083")), + invoker, + NullLogger.Instance); + + var result = await sut.PullAsync("site-a", BaseTime, batchSize: 256, CancellationToken.None); + + Assert.Equal(new[] { "http://node-a:8083", "http://node-b:8083" }, invoker.Dialed); + var evt = Assert.Single(result.Events); + Assert.Equal(id, evt.EventId); + } + + [Fact] + public async Task PullAsync_does_not_fail_over_when_no_fallback_configured() + { + var invoker = new PerEndpointInvoker(new Dictionary> + { + ["http://node-a:8083"] = () => throw new RpcException(new Status(StatusCode.Unavailable, "node A down")), + }); + var sut = new GrpcPullAuditEventsClient( + new StaticEnumerator(new SiteEntry("site-a", "http://node-a:8083")), // no fallback + invoker, + NullLogger.Instance); + + var result = await sut.PullAsync("site-a", BaseTime, batchSize: 256, 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/GrpcPullSiteCallsClientTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/GrpcPullSiteCallsClientTests.cs index 7d0e1da9..85b37abd 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/GrpcPullSiteCallsClientTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/GrpcPullSiteCallsClientTests.cs @@ -63,6 +63,28 @@ public class GrpcPullSiteCallsClientTests } } + /// + /// Per-endpoint fake invoker (Task 18): dials are routed by endpoint string so + /// a NodeA-down / NodeB-up failover can be scripted. Records every dialed + /// endpoint in order. + /// + private sealed class PerEndpointInvoker : GrpcPullSiteCallsClient.IPullSiteCallsInvoker + { + private readonly Dictionary> _byEndpoint; + public List Dialed { get; } = new(); + + public PerEndpointInvoker(Dictionary> byEndpoint) => + _byEndpoint = byEndpoint; + + public Task InvokeAsync( + string endpoint, ProtoPullRequest request, CancellationToken ct) + { + Dialed.Add(endpoint); + // The Func may throw (transport fault) — the client's try/catch handles it. + return Task.FromResult(_byEndpoint[endpoint]()); + } + } + // The site leaves SourceSite empty (it is not a tracking-store column); the // client re-stamps it from the dialed siteId. Mint DTOs with empty SourceSite // to prove that re-stamp. @@ -248,4 +270,52 @@ public class GrpcPullSiteCallsClientTests Assert.Empty(result.SiteCalls); Assert.False(result.MoreAvailable); } + + [Fact] + public async Task PullAsync_fails_over_to_NodeB_when_primary_raises_transport_fault() + { + // Task 18: NodeA (primary) is unreachable, NodeB (fallback) answers — the + // pull must return NodeB's rows rather than collapsing to empty. + var id = Guid.NewGuid(); + var fallbackResponse = new ProtoPullResponse(); + fallbackResponse.Operationals.Add(Dto(id, BaseTime.AddMinutes(1))); + + var invoker = new PerEndpointInvoker(new Dictionary> + { + ["http://node-a:8083"] = () => throw new RpcException(new Status(StatusCode.Unavailable, "node A down")), + ["http://node-b:8083"] = () => fallbackResponse, + }); + var sut = new GrpcPullSiteCallsClient( + new StaticEnumerator(new SiteEntry("site-a", "http://node-a:8083", "http://node-b:8083")), + invoker, + NullLogger.Instance); + + var result = await sut.PullAsync("site-a", BaseTime, afterId: null, batchSize: 256, CancellationToken.None); + + // Primary dialed first, then the NodeB fallback once. + Assert.Equal(new[] { "http://node-a:8083", "http://node-b:8083" }, invoker.Dialed); + var row = Assert.Single(result.SiteCalls); + Assert.Equal(id, row.TrackedOperationId.Value); + Assert.Equal("site-a", row.SourceSite); + } + + [Fact] + public async Task PullAsync_does_not_fail_over_when_no_fallback_configured() + { + // A transport fault with no NodeB fallback collapses to empty and dials + // only the primary once (no phantom second dial). + var invoker = new PerEndpointInvoker(new Dictionary> + { + ["http://node-a:8083"] = () => throw new RpcException(new Status(StatusCode.Unavailable, "node A down")), + }); + var sut = new GrpcPullSiteCallsClient( + new StaticEnumerator(new SiteEntry("site-a", "http://node-a:8083")), // no fallback + invoker, + NullLogger.Instance); + + var result = await sut.PullAsync("site-a", BaseTime, afterId: null, batchSize: 256, CancellationToken.None); + + Assert.Empty(result.SiteCalls); + Assert.Equal(new[] { "http://node-a:8083" }, invoker.Dialed); + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/SiteEnumeratorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/SiteEnumeratorTests.cs index d5a8951b..81febddb 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/SiteEnumeratorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/SiteEnumeratorTests.cs @@ -76,6 +76,44 @@ public class SiteEnumeratorTests Assert.Empty(result); } + [Fact] + public async Task EnumerateAsync_BothAddresses_ProjectsNodeBAsFallback() + { + // Task 18: a site with both NodeA and a DISTINCT NodeB yields + // SiteEntry(primary=NodeA, FallbackGrpcEndpoint=NodeB). + var repository = Substitute.For(); + repository.GetAllSitesAsync(Arg.Any()).Returns(new List + { + SiteWith("site-ab", grpcNodeA: "http://site-a:8083", grpcNodeB: "http://site-b:8083"), + }); + + var enumerator = new SiteEnumerator(BuildProvider(repository)); + + var entry = Assert.Single(await enumerator.EnumerateAsync()); + Assert.Equal("site-ab", entry.SiteId); + Assert.Equal("http://site-a:8083", entry.GrpcEndpoint); + Assert.Equal("http://site-b:8083", entry.FallbackGrpcEndpoint); + } + + [Fact] + public async Task EnumerateAsync_OnlyNodeB_IsNotSkipped_AndHasNoFallback() + { + // Task 18: a site with a blank NodeA but a valid NodeB is no longer + // skipped — NodeB becomes the primary and there is no separate fallback. + var repository = Substitute.For(); + repository.GetAllSitesAsync(Arg.Any()).Returns(new List + { + SiteWith("site-bonly", grpcNodeA: " ", grpcNodeB: "http://site-b:8083"), + }); + + var enumerator = new SiteEnumerator(BuildProvider(repository)); + + var entry = Assert.Single(await enumerator.EnumerateAsync()); + Assert.Equal("site-bonly", entry.SiteId); + Assert.Equal("http://site-b:8083", entry.GrpcEndpoint); + Assert.Null(entry.FallbackGrpcEndpoint); + } + [Fact] public async Task EnumerateAsync_ReturnsEmpty_WhenNoSites() {