dohertj2/ScadaBridge
1
0
Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix(audit-log): reconciliation pulls fail over to site NodeB when NodeA is unreachable

Browse Source 
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.
This commit is contained in:
Joseph Doherty
2026-07-09 08:57:06 -04:00
parent 0385942c3f
commit bfd2cc7e6f
8 changed files with 382 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullAuditEventsClient.cs
+75 -39
View File
@@ -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<string?> ResolveEndpointAsync(string siteId, CancellationToken ct)
/// <summary>
/// Issues one pull against <paramref name="endpoint"/>, classifying faults.
/// Returns <c>(reply, false)</c> on success; <c>(null, true)</c> on a
/// tolerable TRANSPORT fault (the failover-eligible set:
/// <see cref="StatusCode.Unavailable"/> / <see cref="StatusCode.DeadlineExceeded"/> /
/// <see cref="StatusCode.Cancelled"/> / <see cref="HttpRequestException"/> /
/// <c>SocketException</c> / <see cref="OperationCanceledException"/>); and
/// <c>(null, false)</c> on any other (mapping/unexpected) fault, which
/// collapses to empty WITHOUT a fallback dial.
/// </summary>
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 =
src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullSiteCallsClient.cs
+74 -36
View File
@@ -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<string?> ResolveEndpointAsync(string siteId, CancellationToken ct)
/// <summary>
/// Issues one pull against <paramref name="endpoint"/>, classifying faults.
/// Returns <c>(reply, false)</c> on success; <c>(null, true)</c> on a
/// tolerable TRANSPORT fault (the failover-eligible set:
/// <see cref="StatusCode.Unavailable"/> / <see cref="StatusCode.DeadlineExceeded"/> /
/// <see cref="StatusCode.Cancelled"/> / <see cref="HttpRequestException"/> /
/// <c>SocketException</c> / <see cref="OperationCanceledException"/>); and
/// <c>(null, false)</c> on any other (mapping/unexpected) fault, which
/// collapses to empty WITHOUT a fallback dial.
/// </summary>
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 =
src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/ISiteEnumerator.cs
+22 -11
View File
@@ -8,13 +8,13 @@ namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central;
/// </summary>
/// <remarks>
/// The production implementation wraps <c>ISiteRepository.GetAllSitesAsync</c>
/// and projects each <c>Site</c> to a <see cref="SiteEntry"/> using the
/// site's configured <c>GrpcNodeAAddress</c>. This is a NodeA-only first cut:
/// sites with a blank <c>GrpcNodeAAddress</c> 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 <c>SiteEnumerator.cs</c>).
/// and projects each <c>Site</c> to a <see cref="SiteEntry"/>. The primary dial
/// target is <c>GrpcNodeAAddress</c> when set, otherwise <c>GrpcNodeBAddress</c>;
/// when both a distinct NodeA and NodeB address exist, NodeB rides along as
/// <see cref="SiteEntry.FallbackGrpcEndpoint"/> 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.
/// </remarks>
public interface ISiteEnumerator
{
@@ -30,8 +30,19 @@ public interface ISiteEnumerator
/// <summary>
/// One reconciliation target: the site identifier the actor uses as the
/// cursor key and the gRPC endpoint <see cref="IPullAuditEventsClient"/> dials
/// to issue the pull. Endpoint is the bare authority (e.g. <c>http://siteA:8083</c>);
/// 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. <c>http://siteA:8083</c>); transport
/// selection (TLS, keepalive, etc.) is the client's concern.
/// </summary>
public sealed record SiteEntry(string SiteId, string GrpcEndpoint);
/// <param name="SiteId">The site identifier used as the reconciliation cursor key.</param>
/// <param name="GrpcEndpoint">The primary gRPC authority to dial (NodeA where configured, else NodeB).</param>
/// <param name="FallbackGrpcEndpoint">
/// 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.
/// </param>
public sealed record SiteEntry(
string SiteId,
string GrpcEndpoint,
string? FallbackGrpcEndpoint = null);
src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteEnumerator.cs
+27 -14
View File
@@ -21,17 +21,18 @@ namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central;
/// per-tick scope pattern in <see cref="SiteAuditReconciliationActor.OnTickAsync"/>.
/// </para>
/// <para>
/// <b>Blank-address skip.</b> Sites with no <c>GrpcNodeAAddress</c> 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
/// <b>Blank-address skip.</b> A site is skipped only when BOTH
/// <c>GrpcNodeAAddress</c> and <c>GrpcNodeBAddress</c> are blank: the
/// reconciliation pull cannot dial it, but absence of any address is a
/// configuration decision, not a runtime error (per the
/// <see cref="ISiteEnumerator"/> contract).
/// </para>
/// <para>
/// <b>NodeA-only first cut.</b> This implementation always uses NodeA's gRPC
/// address. NodeA/NodeB failover endpoint selection (dial NodeB when NodeA is
/// unreachable) is a follow-up — the <see cref="SiteEntry"/> shape already
/// carries a single endpoint, so failover will live in the puller/client, not
/// here.
/// <b>NodeA/NodeB failover (Task 18).</b> 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
/// <see cref="SiteEntry.FallbackGrpcEndpoint"/> so the pull clients can fail over
/// to it once per call when the primary raises a transport fault.
/// </para>
/// </remarks>
public sealed class SiteEnumerator : ISiteEnumerator
@@ -60,16 +61,28 @@ public sealed class SiteEnumerator : ISiteEnumerator
var entries = new List<SiteEntry>(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;