feat(localdb): fail-closed auth on the sync endpoint + replication health signal

Tasks 8 and 9 of the LocalDb Phase 1 adoption plan.

Task 8 - LocalDbSyncAuthInterceptor. The replication library's LocalDbSyncService
verifies nothing; inbound auth is explicitly the host's job. Without this,
anything able to reach a site node's gRPC port could stream arbitrary rows into
the consolidated site database - including OperationTracking, which central
reconciles from.

Scoped strictly to /localdb_sync.v1.LocalDbSync/; SiteStream shares the same
AddGrpc pipeline and passes through untouched. Fail-closed: with no
LocalDb:Replication:ApiKey configured NO sync stream is accepted, authenticated
or not. That is deliberate - "no key" is the default every site node ships with,
so treating it as "no auth required" would expose the endpoint on precisely the
most common configuration. Comparison is FixedTimeEquals over UTF-8 bytes.

All four server handler shapes are gated, not just unary: the sync RPC is a
bidirectional stream, so gating only the unary path would leave the real endpoint
open while every unary test still passed. There is a test for that.

Deviation from the plan: it specified Grpc.Core.Testing for the fake
ServerCallContext. That type ships in the retired native Grpc.Core package and
does not exist on the grpc-dotnet stack this solution uses; a minimal
FakeServerCallContext in the test file was the better trade than adding a dead
dependency.

Task 9 - ISyncStatus onto the site health report as LocalDbReplicationConnected
and LocalDbOplogBacklog, via a delegate-seam hosted service following the
AddSiteEventLogHealthMetricsBridge precedent (HealthMonitoring takes no reference
on the replication library). Both are additive init properties, so the
Akka-remoted SiteHealthReport constructor signature is untouched.

Both fields are nullable and the distinction is load-bearing:
  - null = the reporter has not run / replication is not wired ("no data");
  - false/0 = a real reading. On a node with no peer that IS the healthy
    default-OFF state, not an outage.
OplogBacklog is passed through nullable end-to-end because ISyncStatus returns
null when the poll fails - flattening it to 0 would report a replication pair
that cannot read its own oplog as perfectly healthy. The collector stores both
values as one tuple and CollectReport reads it once, so a torn read cannot pair a
fresh Connected with a stale backlog.

Verified: build 0 warnings; Host 307/307 (8 interceptor + 3 health tests new),
HealthMonitoring 97/97, Commons 684/684.

Claude-Session: https://claude.ai/code/session_01BL2Vu1ESDQ9SCN4gVKkdts
This commit is contained in:
Joseph Doherty
2026-07-19 09:29:22 -04:00
parent e62b076f2e
commit 59c695191c
10 changed files with 637 additions and 5 deletions
@@ -141,6 +141,25 @@ public interface ISiteHealthCollector
// SiteHealthCollector overrides this with the Interlocked.Exchange store.
}
/// <summary>
/// Replace the latest LocalDb replication status (peer-session connectivity and
/// unacked oplog backlog) used by the next <see cref="CollectReport"/> call.
/// Refreshed periodically by the <c>LocalDbReplicationStatusReporter</c> hosted
/// service. Point-in-time: values are NOT reset on <see cref="CollectReport"/>.
/// </summary>
/// <param name="connected">
/// Whether a sync session is currently running. <see langword="false"/> is the
/// normal state on a node with no peer configured — replication ships default-OFF.
/// </param>
/// <param name="oplogBacklog">
/// Unacked oplog entries, or <see langword="null"/> when unknown. Pass null through
/// unchanged: a failed poll rendered as 0 would report a broken pair as healthy.
/// </param>
void SetLocalDbReplicationStatus(bool connected, long? oplogBacklog)
{
// Default no-op so test fakes do not need to be updated.
}
/// <summary>
/// Replace the latest script-execution-scheduler gauges (queue depth, busy
/// thread count, and age in seconds of the oldest in-flight script) used by
@@ -0,0 +1,130 @@
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
namespace ZB.MOM.WW.ScadaBridge.HealthMonitoring;
/// <summary>
/// Site-side hosted service that periodically reads LocalDb replication status and pushes
/// it into <see cref="ISiteHealthCollector"/>, so the next
/// <see cref="ISiteHealthCollector.CollectReport"/> emits fresh
/// <c>LocalDbReplicationConnected</c> / <c>LocalDbOplogBacklog</c> fields.
/// </summary>
/// <remarks>
/// <para>
/// <b>Why delegates and not <c>ISyncStatus</c> directly.</b> Same reasoning as
/// <see cref="SiteEventLogFailureCountReporter"/>: HealthMonitoring does not take a
/// reference on the replication library. The Host site wiring captures the two reads as
/// lambdas at registration time; this service only moves numbers.
/// </para>
/// <para>
/// <b>Null backlog is preserved, never coerced to zero.</b> <c>ISyncStatus.OplogBacklog</c>
/// is nullable precisely so a failed poll reads as "unknown" — rendering it as 0 would
/// report a replication pair that cannot read its own oplog as perfectly healthy.
/// </para>
/// <para>
/// <b>Cadence.</b> 30 s, matching <see cref="SiteEventLogFailureCountReporter"/> and
/// <c>SiteAuditBacklogReporter</c>. Any exception during a probe is logged and swallowed;
/// the next tick retries.
/// </para>
/// </remarks>
public sealed class LocalDbReplicationStatusReporter : IHostedService, IDisposable
{
/// <summary>Default poll cadence, matching the other site health bridges.</summary>
internal static readonly TimeSpan DefaultRefreshInterval = TimeSpan.FromSeconds(30);
private readonly Func<bool> _connectedProvider;
private readonly Func<long?> _oplogBacklogProvider;
private readonly ISiteHealthCollector _collector;
private readonly ILogger<LocalDbReplicationStatusReporter> _logger;
private readonly TimeSpan _refreshInterval;
private CancellationTokenSource? _cts;
private Task? _loop;
/// <summary>Initializes a new instance of <see cref="LocalDbReplicationStatusReporter"/>.</summary>
/// <param name="connectedProvider">Reads whether a peer sync session is currently running.</param>
/// <param name="oplogBacklogProvider">Reads the unacked oplog backlog, or null when unknown.</param>
/// <param name="collector">The site health collector receiving the snapshot.</param>
/// <param name="logger">Logger instance.</param>
/// <param name="refreshInterval">Poll interval override; defaults to 30 s.</param>
public LocalDbReplicationStatusReporter(
Func<bool> connectedProvider,
Func<long?> oplogBacklogProvider,
ISiteHealthCollector collector,
ILogger<LocalDbReplicationStatusReporter> logger,
TimeSpan? refreshInterval = null)
{
_connectedProvider = connectedProvider ?? throw new ArgumentNullException(nameof(connectedProvider));
_oplogBacklogProvider = oplogBacklogProvider ?? throw new ArgumentNullException(nameof(oplogBacklogProvider));
_collector = collector ?? throw new ArgumentNullException(nameof(collector));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
_refreshInterval = refreshInterval ?? DefaultRefreshInterval;
}
/// <summary>Starts the polling loop, probing once immediately before entering the timed cycle.</summary>
/// <param name="ct">Cancellation token signalling host shutdown.</param>
/// <returns>A task that represents the asynchronous operation.</returns>
public Task StartAsync(CancellationToken ct)
{
var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
_cts = cts;
_loop = RunAsync(cts.Token);
return Task.CompletedTask;
}
/// <summary>Stops the polling loop.</summary>
/// <param name="ct">Cancellation token for the stop operation.</param>
/// <returns>A task that represents the asynchronous operation.</returns>
public async Task StopAsync(CancellationToken ct)
{
if (_cts is null) return;
await _cts.CancelAsync().ConfigureAwait(false);
if (_loop is not null)
{
// Await the loop so the reporter is quiescent before the host disposes the
// collector out from under it.
try { await _loop.ConfigureAwait(false); }
catch (OperationCanceledException) { /* expected on shutdown */ }
}
}
private async Task RunAsync(CancellationToken ct)
{
while (!ct.IsCancellationRequested)
{
Probe();
try
{
await Task.Delay(_refreshInterval, ct).ConfigureAwait(false);
}
catch (OperationCanceledException)
{
return;
}
}
}
/// <summary>
/// Reads both providers and pushes one snapshot onto the collector; never throws.
/// Public so a test can drive a single deterministic probe instead of starting the
/// service and waiting out the 30 s cadence.
/// </summary>
public void Probe()
{
try
{
_collector.SetLocalDbReplicationStatus(
_connectedProvider(), _oplogBacklogProvider());
}
catch (Exception ex)
{
// Health reporting must never take the site node down. The previous snapshot
// stays on the collector and the next tick retries.
_logger.LogWarning(ex, "Failed to read LocalDb replication status for the site health report.");
}
}
/// <inheritdoc />
public void Dispose() => _cts?.Dispose();
}
@@ -21,6 +21,12 @@ public static class ServiceCollectionExtensions
/// </summary>
private sealed class SiteEventLogHealthMetricsBridgeMarker { }
/// <summary>
/// Sentinel marker for <see cref="AddLocalDbReplicationHealthBridge"/>'s idempotency
/// guard — same rationale as <see cref="SiteEventLogHealthMetricsBridgeMarker"/>.
/// </summary>
private sealed class LocalDbReplicationHealthBridgeMarker { }
/// <summary>
/// Register site-side health monitoring services (metric collection + periodic reporting).
/// Call this on site nodes only. For central, call AddCentralHealthAggregation() instead.
@@ -153,6 +159,46 @@ public static class ServiceCollectionExtensions
return services;
}
/// <summary>
/// Bridge LocalDb replication status (peer connectivity + unacked oplog backlog) onto
/// the site health report. Must be called AFTER <c>AddSiteHealthMonitoring</c>
/// (registers <see cref="ISiteHealthCollector"/>) and after the replication engine is
/// registered. Idempotent via a marker sentinel.
/// </summary>
/// <remarks>
/// Delegates rather than an <c>ISyncStatus</c> parameter keep HealthMonitoring free of
/// a reference on the replication library, matching
/// <see cref="AddSiteEventLogHealthMetricsBridge"/>. Pass the backlog through as
/// nullable — coercing null to 0 would report an unreadable oplog as a healthy one.
/// </remarks>
/// <param name="services">The service collection to register into.</param>
/// <param name="connectedProvider">Given the root provider, returns a reader for "a sync session is running".</param>
/// <param name="oplogBacklogProvider">Given the root provider, returns a reader for the unacked backlog (null = unknown).</param>
/// <returns>The same <see cref="IServiceCollection"/> for chaining.</returns>
public static IServiceCollection AddLocalDbReplicationHealthBridge(
this IServiceCollection services,
Func<IServiceProvider, Func<bool>> connectedProvider,
Func<IServiceProvider, Func<long?>> oplogBacklogProvider)
{
ArgumentNullException.ThrowIfNull(services);
ArgumentNullException.ThrowIfNull(connectedProvider);
ArgumentNullException.ThrowIfNull(oplogBacklogProvider);
if (services.Any(d => d.ServiceType == typeof(LocalDbReplicationHealthBridgeMarker)))
{
return services;
}
services.AddSingleton<LocalDbReplicationHealthBridgeMarker>();
services.AddHostedService(sp => new LocalDbReplicationStatusReporter(
connectedProvider(sp),
oplogBacklogProvider(sp),
sp.GetRequiredService<ISiteHealthCollector>(),
sp.GetRequiredService<ILogger<LocalDbReplicationStatusReporter>>()));
return services;
}
/// <summary>
/// Register the <see cref="HealthMonitoringOptionsValidator"/>
/// so a misconfigured <c>ScadaBridge:HealthMonitoring</c> section (zero/negative
@@ -18,6 +18,10 @@ public class SiteHealthCollector : ISiteHealthCollector
private int _auditRedactionFailures;
private volatile SiteAuditBacklogSnapshot? _siteAuditBacklog;
private long _siteEventLogWriteFailures;
// One volatile tuple rather than two independent fields: a torn read that paired a
// fresh Connected with a stale backlog would be indistinguishable from a real state.
// Null = the reporter has not yet run (or replication is not wired).
private volatile Tuple<bool, long?>? _localDbReplicationStatus;
private readonly ConcurrentDictionary<string, ConnectionHealth> _connectionStatuses = new();
private readonly ConcurrentDictionary<string, TagResolutionStatus> _tagResolutionCounts = new();
private readonly ConcurrentDictionary<string, string> _connectionEndpoints = new();
@@ -94,6 +98,12 @@ public class SiteHealthCollector : ISiteHealthCollector
Interlocked.Exchange(ref _siteEventLogWriteFailures, count);
}
/// <inheritdoc />
public void SetLocalDbReplicationStatus(bool connected, long? oplogBacklog)
{
_localDbReplicationStatus = Tuple.Create(connected, oplogBacklog);
}
/// <inheritdoc />
public void UpdateConnectionHealth(string connectionName, ConnectionHealth health)
{
@@ -219,6 +229,9 @@ public class SiteHealthCollector : ISiteHealthCollector
var deadLetters = Interlocked.Exchange(ref _deadLetterCount, 0);
var siteAuditWriteFailures = Interlocked.Exchange(ref _siteAuditWriteFailures, 0);
var auditRedactionFailures = Interlocked.Exchange(ref _auditRedactionFailures, 0);
// Single read of the volatile tuple — two reads could straddle a reporter tick and
// pair a fresh Connected with a stale backlog.
var localDbReplication = _localDbReplicationStatus;
// Snapshot current connection and tag resolution state
var connectionStatuses = new Dictionary<string, ConnectionHealth>(_connectionStatuses);
@@ -259,7 +272,12 @@ public class SiteHealthCollector : ISiteHealthCollector
{
ScriptQueueDepth = Interlocked.CompareExchange(ref _scriptQueueDepth, 0, 0),
ScriptBusyThreads = Interlocked.CompareExchange(ref _scriptBusyThreads, 0, 0),
ScriptOldestBusyAgeSeconds = ReadScriptOldestBusyAgeSeconds()
ScriptOldestBusyAgeSeconds = ReadScriptOldestBusyAgeSeconds(),
// Both fields come from the ONE snapshot read above. Null (the reporter has
// not run) leaves both report fields null — "no data", not "disconnected
// with an empty backlog".
LocalDbReplicationConnected = localDbReplication?.Item1,
LocalDbOplogBacklog = localDbReplication?.Item2
};
}
}