using System.Data.Common;
using System.Diagnostics.CodeAnalysis;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts;
namespace ZB.MOM.WW.OtOpcUa.Driver.Sql;
///
/// The Sql driver instance: a read-only Equipment-kind driver that polls SQL tables/views
/// and publishes selected cells as OPC UA variables. Structurally the SQL analogue of
/// ModbusDriver — this type owns lifecycle, the authored RawPath table, health, host
/// connectivity, and the subscription overlay, while
/// owns the read path.
/// Read-only is structural, not configured. is deliberately not
/// implemented, so no amount of config (and no WriteOperate role) can produce a write; every
/// discovered variable is . A future write feature is a
/// new capability, not a flag flip.
/// Health is classified here, because nothing below does it. The reader honours
/// literally: it throws only when the database cannot be reached, and turns
/// every other failure — including a frozen database — into Bad-coded snapshots. Those return
/// successfully, so sees a clean tick, does not back off, and does
/// not call . Without below, a driver whose
/// every value is would keep reporting
/// . See that method for the exact rule and for why it does not
/// manufacture an exception to force backoff.
///
public sealed class SqlDriver
: IDriver, ITagDiscovery, IReadable, ISubscribable, IHostConnectivityProbe, IAsyncDisposable
{
///
/// The driver-type string this driver registers under — , the single
/// source of truth the deploy pipeline stores in DriverInstance.DriverType and every dispatch
/// map keys by. Chained off the shared constant (Task 11 wired the factory + probe, so
/// DriverTypeNamesGuardTests' bidirectional-parity check is satisfied).
///
public const string DriverTypeName = DriverTypeNames.Sql;
/// Shown for a connection string that names no recognisable server.
private const string UnknownEndpoint = "(unknown sql endpoint)";
/// Upper bound on the poll-loop failure backoff — the S7-proven 30 s cap adopted fleet-wide (05/STAB-8).
private static readonly TimeSpan PollBackoffCap = TimeSpan.FromSeconds(30);
/// Connection-string keys that name the server, in the order they are consulted.
private static readonly string[] ServerKeys =
["server", "data source", "datasource", "host", "address", "addr", "network address"];
/// Connection-string keys that name the database.
private static readonly string[] DatabaseKeys = ["database", "initial catalog"];
// ---- instance fields (grouped at top for auditability) ----
/// Config-derived, so mutable — see .
private SqlDriverOptions _options;
private readonly string _driverInstanceId;
/// Config-derived, so mutable: swaps it with the connection string
/// and options together on a reinitialize (#516).
private ISqlDialect _dialect;
private DbProviderFactory _factory;
///
/// The resolved connection string — a secret. It is passed to the provider and to nothing
/// else: every log line, health message and host status carries instead.
///
private string _connectionString;
private readonly ILogger _logger;
///
/// The authored RawPath → definition table, held as an immutable snapshot swapped atomically
/// rather than as a mutated dictionary (the shape ModbusDriver uses).
/// rebuilds this table while poll loops are reading it; clearing and
/// refilling a shared under that concurrency is a torn read at
/// best and an inside a poll at worst.
///
private IReadOnlyDictionary _tagsByRawPath =
new Dictionary(StringComparer.Ordinal);
///
/// The subset of that survived the design §8.1 catalog gate, with every
/// identifier rewritten to the catalog's own spelling. This — not the authored table — is what a
/// read or a poll resolves against, so an identifier that never appeared in the catalog can never
/// reach a query.
/// Why the two tables are separate. A tag the gate rejected must still exist as a
/// node: §8.1's specified outcome is that it publishes
/// , and a status code can only be published by a node
/// that is there. Dropping rejected tags from the authored table too would delete the node instead,
/// turning a diagnosable Bad quality into a silently missing address-space entry — a strictly worse
/// failure for the operator trying to find their typo.
/// Swapped atomically, for the same reason the authored table is.
///
private IReadOnlyDictionary _polledByRawPath =
new Dictionary(StringComparer.Ordinal);
/// Resolves a read/subscribe RawPath to its catalog-validated definition, or a miss.
private readonly EquipmentTagRefResolver _resolver;
/// The factory the CALLER passed, or null to take the dialect's. Kept separate from
/// so a re-derived dialect cannot silently displace a test's injected factory.
private readonly DbProviderFactory? _explicitFactory;
/// Re-parses a DriverConfig into every config-derived dependency. Null when the driver
/// was constructed directly, which keeps the constructor-supplied trio authoritative.
private readonly Func? _rebind;
///
/// The read path. Built once, in the constructor, and not injected: its resolve
/// delegate must read this driver's live authored table, which does not exist before the driver does.
///
private SqlPollReader _reader;
/// Polled subscriptions. The driver supplies the reader + change bridge; the engine owns the loop.
private readonly PollGroupEngine _poll;
// Single logical host: one driver instance dials one connection string. HostName is the endpoint
// description (server[/database]) so the Admin UI shows operators where to look.
private readonly object _probeLock = new();
private HostState _hostState = HostState.Unknown;
private DateTime _hostStateChangedUtc = DateTime.UtcNow;
private DriverHealth _health = new(DriverState.Unknown, null, null);
/// Occurs when a subscribed tag's value or quality changes.
public event EventHandler? OnDataChange;
/// Occurs when the database endpoint transitions between reachable and unreachable.
public event EventHandler? OnHostStatusChanged;
// ---- ctor + identity ----
/// Initializes a new Sql driver instance.
/// The driver's typed configuration (the factory applies the documented defaults).
/// The central config DB's identity for this driver instance.
///
/// The provider seam — identifier quoting, liveness SQL, row-limit syntax and column-type mapping.
///
///
/// The already-resolved connection string. The driver never resolves a
/// connectionStringRef itself and never logs this value.
///
///
/// Creates the provider's connections. Defaults to 's factory; passed
/// explicitly by tests that need to observe or delay connection creation.
///
/// Optional; defaults to a no-op logger.
///
/// Optional re-parse of a DriverConfig into every config-derived dependency, supplied by the
/// factory so can adopt a changed config. Null (direct construction —
/// every test) keeps the constructor-supplied options, dialect and connection string.
///
/// A required reference argument is null.
/// or is blank.
public SqlDriver(
SqlDriverOptions options,
string driverInstanceId,
ISqlDialect dialect,
string connectionString,
DbProviderFactory? factory = null,
ILogger? logger = null,
Func? rebind = null)
{
ArgumentNullException.ThrowIfNull(options);
ArgumentNullException.ThrowIfNull(dialect);
ArgumentException.ThrowIfNullOrWhiteSpace(driverInstanceId);
if (string.IsNullOrWhiteSpace(connectionString))
throw new ArgumentException("A Sql driver needs a resolved connection string.", nameof(connectionString));
_driverInstanceId = driverInstanceId;
_logger = logger ?? NullLogger.Instance;
_explicitFactory = factory;
_rebind = rebind;
_resolver = new EquipmentTagRefResolver(Lookup);
// Everything below this line is derived from config and is re-derived wholesale on a reinitialize.
ApplyBinding(options, dialect, connectionString);
_poll = new PollGroupEngine(
reader: ReadAsync,
onChange: (handle, tagRef, snapshot) =>
OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, tagRef, snapshot)),
onError: HandlePollError,
backoffCap: PollBackoffCap);
}
///
/// Adopts a config-derived trio — options, dialect, connection string — and rebuilds everything
/// downstream of it (the provider factory, the endpoint description, and the poll reader, which
/// captures all three). Called from the constructor and again from
/// when a changed config arrives.
/// All-or-nothing on purpose. The reason Sql was excluded from the first #516 pass is
/// that adopting options alone would poll a NEW tag set through the OLD connection and dialect. Doing
/// it in one method means a future field derived from config gets added here rather than being
/// forgotten on the reinit path.
/// Not thread-safe by design: only the init path calls it, before any poll group is running.
/// The poll engine and the tag resolver survive across a rebind — the engine holds a delegate to
/// ReadAsync and the resolver reads the live tag table, so neither captures the old trio.
///
/// The typed options to adopt.
/// The provider seam to adopt.
/// The resolved connection string to adopt. Never logged.
[MemberNotNull(nameof(_options), nameof(_dialect), nameof(_factory), nameof(_connectionString), nameof(_reader))]
private void ApplyBinding(SqlDriverOptions options, ISqlDialect dialect, string connectionString)
{
_options = options;
_dialect = dialect;
_factory = _explicitFactory ?? dialect.Factory;
_connectionString = connectionString;
Endpoint = DescribeEndpoint(connectionString);
_reader = new SqlPollReader(
_factory,
connectionString,
dialect,
commandTimeout: options.CommandTimeout,
operationTimeout: options.OperationTimeout,
maxConcurrentGroups: options.MaxConcurrentGroups,
nullIsBad: options.NullIsBad,
resolve: rawPath => _resolver.TryResolve(rawPath, out var definition) ? definition : null,
logger: _logger);
}
/// True when the supplied DriverConfig JSON carries a real body. The bootstrapper always passes
/// a populated document; some unit tests pass "{}" or an empty string to exercise lifecycle shape
/// without a config — those keep the constructor-supplied trio.
private static bool HasConfigBody(string? driverConfigJson)
{
if (string.IsNullOrWhiteSpace(driverConfigJson)) return false;
var trimmed = driverConfigJson.Trim();
return trimmed is not "{}" and not "[]";
}
///
public string DriverInstanceId => _driverInstanceId;
///
public string DriverType => DriverTypeName;
///
/// The credential-free description of the database this instance polls — server/database where
/// the connection string names both. This is the only rendering of the connection string that may
/// appear in a log, a health message, or the Admin UI.
///
public string Endpoint { get; private set; } = "";
/// Host identifier surfaced through — the endpoint description.
public string HostName => Endpoint;
/// Active polled-subscription count. Diagnostics + disposal assertions.
internal int ActiveSubscriptionCount => _poll.ActiveSubscriptionCount;
/// The current authored-tag snapshot. Read through a barrier — swaps it.
private IReadOnlyDictionary Tags => Volatile.Read(ref _tagsByRawPath);
///
/// The catalog-validated snapshot. Read through a barrier — swaps it.
///
private IReadOnlyDictionary PolledTags => Volatile.Read(ref _polledByRawPath);
///
/// The resolver's lookup: RawPath → catalog-validated definition, or null on a miss.
/// Deliberately reads rather than : a tag the gate
/// rejected must miss here so the reader publishes
/// , exactly as design §8.1 specifies.
///
private SqlTagDefinition? Lookup(string rawPath) => PolledTags.GetValueOrDefault(rawPath);
// ---- IDriver lifecycle ----
///
/// Builds the authored RawPath table and proves the database is reachable.
/// is re-parsed here when the driver was built by
/// the factory (Gitea #516). The original comment on this method claimed it was not, on the premise
/// that "config parsing belongs to the factory, which builds a fresh instance" — that was false when
/// written, because the host reinitialized the EXISTING child in place and every config edit was
/// silently discarded while the deployment still sealed green.
/// The re-parse is all-or-nothing via : options, the
/// and the resolved connection string are re-derived by one
/// ParseBinding call and adopted together. That is what makes it safe here — adopting options
/// alone would poll a NEW tag set through the OLD database, which is why this driver was excluded
/// from the first #516 pass. It happens BEFORE BuildTagTable, so the tag table and the
/// connection it is polled over always come from the same revision.
/// A driver constructed directly (every unit test) gets no rebinder and keeps its
/// constructor-supplied trio, so "{}"-passing lifecycle tests are unaffected.
/// DriverSpawnPlanner's stop + respawn remains the outer guarantee.
/// The table is built first because it is pure and cannot fail; the liveness check is the only
/// I/O, and on failure this method records and rethrows —
/// DriverInstanceActor reads a throw as InitializeFailed and lands in Reconnecting with its
/// retry timer running, which is exactly the recovery a database that is merely down needs.
/// The liveness check is then followed by the design §8.1 catalog gate
/// (), which is the second and last piece of I/O. It runs after
/// liveness because it needs a working connection, and before the driver reports Healthy because the
/// tag table it publishes must be the validated one — a poll must never see an unvalidated
/// identifier.
///
/// The driver configuration JSON (unused; see remarks).
/// Cancellation token for the operation.
/// A task that represents the asynchronous operation.
/// The database could not be reached, or its catalog could not be read.
public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
WriteHealth(new DriverHealth(DriverState.Initializing, null, null));
// #516: adopt a changed config BEFORE the tag table is built, so the table and the connection it
// will be polled over come from the same revision. ApplyBinding takes the options, dialect and
// connection string as one unit; adopting options alone would poll a NEW tag set through the OLD
// database, which is why this driver was excluded from the first pass. A null rebinder (direct
// construction) or an empty/placeholder document keeps the constructor-supplied trio.
if (_rebind is not null && HasConfigBody(driverConfigJson))
{
var binding = _rebind(driverConfigJson);
ApplyBinding(binding.Options, binding.Dialect, binding.ConnectionString);
}
BuildTagTable();
try
{
await VerifyLivenessAsync(cancellationToken).ConfigureAwait(false);
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
// The caller tore the initialization down; not a database verdict.
WriteHealth(new DriverHealth(DriverState.Unknown, null, null));
throw;
}
catch (Exception ex)
{
// Never interpolate _connectionString OR ex.Message onto the operator surface. The driver is
// dialect-agnostic over an arbitrary DbProviderFactory, so it cannot rely on any one provider's
// message discipline — Microsoft.Data.SqlClient's connection-string parser, for one, echoes an
// unrecognised keyword lower-cased, which a value-based password redactor misses entirely. Only
// the exception TYPE reaches LastError/host status; the full exception object goes to the log
// SINK via the structured logger's exception parameter (below), never into a rendered string.
var message =
$"Sql driver '{_driverInstanceId}' could not reach {Endpoint}: {ex.GetType().Name}. " +
$"Check the database is reachable and the connection string is valid.";
_logger.LogError(ex, "Sql driver {DriverInstanceId} liveness check against {Endpoint} failed.",
_driverInstanceId, Endpoint);
WriteHealth(new DriverHealth(DriverState.Faulted, null, message));
TransitionHostTo(HostState.Stopped);
throw new InvalidOperationException(message, ex);
}
// Separate try from liveness so the operator surface names the stage that actually failed: "could
// not reach the database" and "reached it but could not read its catalog" send an operator to
// different places, and the second is usually a GRANT rather than an outage.
try
{
await ApplyCatalogGateAsync(cancellationToken).ConfigureAwait(false);
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
WriteHealth(new DriverHealth(DriverState.Unknown, null, null));
throw;
}
catch (Exception ex)
{
// Same discipline as above: exception TYPE only on the operator surface, full exception to the
// log sink. A catalog read that fails is NOT evidence the tags are wrong — it is the absence of
// evidence — so this faults the driver (and retries) rather than rejecting every tag and serving
// a confidently-empty address space.
var message =
$"Sql driver '{_driverInstanceId}' reached {Endpoint} but could not read its catalog to " +
$"validate authored tables/columns: {ex.GetType().Name}. Check that the connecting principal " +
"can read the catalog views.";
_logger.LogError(ex,
"Sql driver {DriverInstanceId} catalog validation against {Endpoint} failed.",
_driverInstanceId, Endpoint);
WriteHealth(new DriverHealth(DriverState.Faulted, null, message));
TransitionHostTo(HostState.Stopped);
throw new InvalidOperationException(message, ex);
}
WriteHealth(new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null));
TransitionHostTo(HostState.Running);
_logger.LogInformation(
"Sql driver {DriverInstanceId} initialized against {Endpoint} with {TagCount} authored tag(s).",
_driverInstanceId, Endpoint, Tags.Count);
}
///
public async Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
await ShutdownAsync(cancellationToken).ConfigureAwait(false);
await InitializeAsync(driverConfigJson, cancellationToken).ConfigureAwait(false);
}
///
public async Task ShutdownAsync(CancellationToken cancellationToken)
{
var lastRead = ReadHealth().LastSuccessfulRead;
await TeardownAsync().ConfigureAwait(false);
WriteHealth(new DriverHealth(DriverState.Unknown, lastRead, null));
}
///
public DriverHealth GetHealth() => ReadHealth();
/// No caches, no symbol table, no connection between polls — the footprint is the tag table.
/// Always zero.
public long GetMemoryFootprint() => 0;
/// Nothing optional to flush: the authored table is required for correctness.
/// Cancellation token for the operation.
/// A completed task.
public Task FlushOptionalCachesAsync(CancellationToken cancellationToken) => Task.CompletedTask;
// ---- ITagDiscovery ----
///
/// Exactly one pass: the tag set is the authored one, fixed at Initialize, so re-running discovery
/// could only produce the same nodes.
///
public DiscoveryRediscoverPolicy RediscoverPolicy => DiscoveryRediscoverPolicy.Once;
///
/// False — replays authored tags rather than enumerating the backend, so
/// the universal discovery browser must not treat this driver as browsable. (Live schema browsing is
/// a separate surface: Driver.Sql.Browser.)
///
public bool SupportsOnlineDiscovery => false;
///
/// Streams the authored tags as read-only variables.
/// An undeclared tag materialises as — the same fallback
/// uses for an unrecognised column type — because a column's
/// real type is only known once a poll has returned result-set metadata, which has not happened at
/// discovery time. Author "type" on a numeric tag.
///
/// The address-space builder to stream discovered nodes into.
/// Cancellation token for the operation.
/// A completed task — discovery is synchronous over the authored table.
public Task DiscoverAsync(IAddressSpaceBuilder builder, CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(builder);
var folder = builder.Folder(DriverTypeName, DriverTypeName);
foreach (var tag in Tags.Values)
{
if (tag.DeclaredType is null)
{
// Discovery cannot infer a column's real type — that needs live result-set metadata, which
// only a poll returns — so an omitted-type tag materialises as String (the same fallback
// ISqlDialect.MapColumnType uses) while the reader coerces and publishes the column's actual,
// possibly numeric, CLR value against that String node. Nothing else signals the operator to
// this declared-vs-published mismatch, so warn and point them at the fix.
_logger.LogWarning(
"Sql tag '{Tag}' has no authored \"type\" and is discovered as String; a numeric column " +
"will then publish numeric values against a String node. Author an explicit \"type\" on " +
"numeric tags. Driver={DriverInstanceId}",
tag.Name, _driverInstanceId);
}
folder.Variable(tag.Name, tag.Name, new DriverAttributeInfo(
FullName: tag.Name,
DriverDataType: tag.DeclaredType ?? DriverDataType.String,
IsArray: false,
ArrayDim: null,
// v1 is read-only: no tag, and no configuration, can widen this.
SecurityClass: SecurityClassification.ViewOnly,
IsHistorized: false,
IsAlarm: false,
WriteIdempotent: false));
}
return Task.CompletedTask;
}
// ---- IReadable ----
///
/// Reads a batch, delegating to and classifying what came back
/// ().
///
/// The RawPaths to read.
/// Cancellation token for the operation.
/// One snapshot per reference, at the same index.
/// The database could not be reached — the engine backs off on this.
public async Task> ReadAsync(
IReadOnlyList fullReferences, CancellationToken cancellationToken)
{
// Guarded here rather than inside the try, so a caller's bug cannot be misread as "the database is
// unreachable" and degrade a perfectly healthy driver.
ArgumentNullException.ThrowIfNull(fullReferences);
try
{
var snapshots = await _reader.ReadAsync(fullReferences, cancellationToken).ConfigureAwait(false);
ObservePollOutcome(snapshots);
return snapshots;
}
catch (OperationCanceledException)
{
// Teardown, not a database verdict — leave health and host state exactly as they were.
throw;
}
catch (Exception ex)
{
// The reader throws for one reason only: opening a connection failed (IReadable's contract).
_logger.LogWarning(ex, "Sql driver {DriverInstanceId} could not reach {Endpoint} during a read.",
_driverInstanceId, Endpoint);
// Type name only — never ex.Message (see InitializeAsync for why); the full exception is already
// on the log sink via the LogWarning exception parameter above.
Degrade($"Sql driver '{_driverInstanceId}' could not reach {Endpoint}: {ex.GetType().Name}.");
TransitionHostTo(HostState.Stopped);
throw;
}
}
// ---- ISubscribable (polling overlay via the shared engine) ----
///
/// Registers a polled subscription.
/// A non-positive falls back to the configured
/// ; anything faster than
/// is floored by the engine.
///
/// The RawPaths to poll.
/// The requested publishing interval.
/// Cancellation token for the operation.
/// The subscription handle.
public Task SubscribeAsync(
IReadOnlyList fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken)
{
var interval = publishingInterval > TimeSpan.Zero ? publishingInterval : _options.DefaultPollInterval;
return Task.FromResult(_poll.Subscribe(fullReferences, interval));
}
///
public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
{
_poll.Unsubscribe(handle);
return Task.CompletedTask;
}
// ---- IHostConnectivityProbe ----
///
/// The single logical host this instance dials. There is no background probe loop: the state is a
/// by-product of the Initialize-time liveness check and of every poll, so the report is always a
/// statement about traffic that actually happened rather than about a synthetic ping.
///
/// A one-element list describing the configured database endpoint.
public IReadOnlyList GetHostStatuses()
{
lock (_probeLock)
return [new HostConnectivityStatus(HostName, _hostState, _hostStateChangedUtc)];
}
// ---- health + host-state classification ----
///
/// Classifies one poll's snapshots into a health verdict and a host state.
///
/// - Any Good snapshot ⇒ the database answered: LastSuccessfulRead advances and the
/// host is Running.
/// - Any connection-class Bad snapshot ( /
/// ) ⇒ Degraded; when nothing at all was Good,
/// the host is Stopped.
/// - Bad codes that are authoring facts — an unresolvable RawPath, an absent row, a
/// type mismatch, a rejected definition — change nothing. A tag typo must never report the
/// database as down and send an operator to the wrong system.
///
/// Why this does not throw to force poll-engine backoff. Backoff would require turning
/// an all-BadTimeout batch into an exception, and the engine's exception path publishes
/// nothing — clients would lose the very BadTimeout quality the reader went out of its way to
/// produce, and would sit on stale Good values instead. Nor is hammering a real risk: each group is
/// already bounded by operationTimeout and the reader never holds more than
/// maxConcurrentGroups connections however long the database stays frozen. So a frozen
/// database is reported (Degraded + host Stopped + Bad-quality values) at the configured poll
/// cadence rather than backed off from.
///
/// The snapshots the reader returned for one poll.
internal void ObservePollOutcome(IReadOnlyList snapshots)
{
if (snapshots.Count == 0) return;
var good = 0;
var unreachable = 0;
foreach (var snapshot in snapshots)
{
if (SqlStatusCodes.IsGood(snapshot.StatusCode)) good++;
else if (IsConnectionClass(snapshot.StatusCode)) unreachable++;
}
if (unreachable > 0)
{
Degrade(
$"{unreachable} of {snapshots.Count} Sql tag(s) timed out or failed communication against " +
$"{Endpoint} on the last poll.");
if (good > 0)
{
// Partial: something answered, so the endpoint is up — but the driver is not healthy.
TouchLastSuccessfulRead();
TransitionHostTo(HostState.Running);
}
else
{
TransitionHostTo(HostState.Stopped);
}
return;
}
if (good == 0) return; // authoring-class Bad only — not a statement about the database.
// Route through the Faulted guard, not a raw WriteHealth: a poll still in flight when a concurrent
// ReinitializeAsync has recorded Faulted (the engine's ~5 s teardown wait is shorter than the 15 s
// default operationTimeout, so a stale poll can still complete afterwards) must not flip Faulted back
// to Healthy with no real recovery. Only a successful (re)initialize clears Faulted.
SetHealthUnlessFaulted(new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null));
TransitionHostTo(HostState.Running);
}
/// True for the status codes that mean "the database did not answer", as opposed to "the data is not there".
private static bool IsConnectionClass(uint statusCode)
=> statusCode is SqlStatusCodes.BadTimeout or SqlStatusCodes.BadCommunicationError;
///
/// Routes a poll-loop reader failure to the health surface (05/STAB-9). Deliberately does not
/// touch host state: the engine also reports its own contract violations here, which say nothing
/// about connectivity — the unreachable-database transition is raised by ,
/// which knows the exception came from the reader.
///
/// The exception the poll engine caught.
internal void HandlePollError(Exception ex)
{
// A subscribed-read outage surfaces as a DbException the reader threw, which ReadAsync has ALREADY
// classified — good, endpoint-bearing LastError + host Stopped — before rethrowing. The poll engine
// then routes that same exception here. Re-degrading would overwrite the good LastError with a worse,
// guidance-free one and log the outage a second time, so skip the connection class ReadAsync owns.
if (ex is DbException) return;
// Whatever reaches here — an engine contract-violation, or a non-DbException the provider raised
// (e.g. an ArgumentException from a malformed connection string, which the parser echoes with
// credential fragments) — the operator surface carries only the exception TYPE, never ex.Message.
// The full exception, with its text, still reaches the log SINK via the exception parameter. This is
// the same unconditional rule InitializeAsync/ReadAsync follow; do not weaken it to a type check.
_logger.LogWarning(ex, "Sql poll failed. Driver={DriverInstanceId} Endpoint={Endpoint}",
_driverInstanceId, Endpoint);
Degrade($"Sql driver '{_driverInstanceId}' poll failed: {ex.GetType().Name}.");
}
///
/// Degrades health, preserving LastSuccessfulRead and never downgrading
/// — a config-level verdict that only a successful read or an
/// operator reinitialize may clear.
///
/// The operator-facing reason, which must never carry the connection string.
private void Degrade(string reason)
{
var current = ReadHealth();
SetHealthUnlessFaulted(new DriverHealth(DriverState.Degraded, current.LastSuccessfulRead, reason));
}
///
/// Publishes unless the driver is already
/// — a liveness/config verdict that only a successful (re)initialize may clear. Every non-Initialize
/// health write routes through here: and the poll's Healthy verdict in
/// , so a late in-flight poll cannot un-fault a driver a concurrent
/// just faulted.
///
/// The health snapshot to publish when the driver is not Faulted.
private void SetHealthUnlessFaulted(DriverHealth value)
{
if (ReadHealth().State == DriverState.Faulted) return;
WriteHealth(value);
}
/// Advances LastSuccessfulRead without changing the current state or error.
private void TouchLastSuccessfulRead()
{
var current = ReadHealth();
WriteHealth(current with { LastSuccessfulRead = DateTime.UtcNow });
}
/// Barrier-protected read of the multi-threaded health field.
private DriverHealth ReadHealth() => Volatile.Read(ref _health);
/// Barrier-protected publish of a new health snapshot.
private void WriteHealth(DriverHealth value) => Volatile.Write(ref _health, value);
/// Records a host-state change and raises — on transitions only.
private void TransitionHostTo(HostState newState)
{
HostState old;
lock (_probeLock)
{
old = _hostState;
if (old == newState) return;
_hostState = newState;
_hostStateChangedUtc = DateTime.UtcNow;
}
OnHostStatusChanged?.Invoke(this, new HostStatusChangedEventArgs(HostName, old, newState));
}
// ---- authored tag table ----
///
/// Maps every authored through
/// and publishes the result as one atomic snapshot. A blob that does not map is logged and
/// skipped, never thrown: one malformed tag must not take the whole driver — and therefore every
/// other tag on that database — down with it. The skipped RawPath then resolves to nothing, so it
/// publishes rather than someone else's row.
///
private void BuildTagTable()
{
var table = new Dictionary(_options.RawTags.Count, StringComparer.Ordinal);
foreach (var entry in _options.RawTags)
{
try
{
if (SqlEquipmentTagParser.TryParse(entry.TagConfig, entry.RawPath, out var definition))
table[entry.RawPath] = definition;
else
_logger.LogWarning(
"Sql tag config did not map to a definition; skipping. Driver={DriverInstanceId} RawPath={RawPath}",
_driverInstanceId, entry.RawPath);
}
catch (Exception ex)
{
// TryParse is contracted never to throw, but BuildTagTable runs BEFORE InitializeAsync's
// try/catch, so a stray throw here (a parser bug on one blob) would escape Initialize and
// strand health at Initializing across every DriverInstanceActor retry. Skip the offending
// tag — the same "one bad tag must not take the whole driver down" rule the malformed-blob
// branch above follows — rather than fault the driver over a single entry.
_logger.LogError(
ex, "Sql tag config threw while parsing; skipping. Driver={DriverInstanceId} RawPath={RawPath}",
_driverInstanceId, entry.RawPath);
}
}
Volatile.Write(ref _tagsByRawPath, table);
// Nothing is pollable until the catalog gate has validated it. Clearing here (rather than leaving a
// previous generation's validated table in place) means a Reinitialize whose gate then fails cannot
// keep polling the OLD identifiers against a database whose schema may be exactly what changed.
Volatile.Write(ref _polledByRawPath, new Dictionary(StringComparer.Ordinal));
}
// ---- catalog gate (design §8.1) ----
///
/// Validates every authored identifier against the live catalog and republishes the tag table with
/// catalog spellings, dropping the tags that do not resolve (design §8.1).
///
///
/// Why this must run before the driver reports Healthy. The table this swaps in is what
/// every poll resolves against. Running it later — or in the background — would leave a window in
/// which an unvalidated identifier could be quoted into a query, which is the entire hole the gate
/// exists to close.
/// A dropped tag is not a driver fault. It disappears from the table, so its RawPath
/// resolves to nothing and it publishes — one operator
/// typo must not stop the other tags on that database, the same rule
/// follows for a malformed blob. Each drop is logged at Warning with the
/// tag, the field and the reason, because a node that silently goes Bad with nothing in the log is
/// unsupportable.
/// No tags ⇒ no I/O. A driver with nothing authored has nothing to validate, and issuing
/// catalog queries to prove that would be a round-trip that can only fail.
/// Bounded by the same wall-clock discipline as the liveness check (R2-01 / STAB-14): the work
/// runs on the thread pool so a provider implementing the async path synchronously blocks a pool
/// thread rather than wedging Initialize forever with no retry.
///
/// The caller's token.
/// A task that completes when the tag table has been validated and republished.
private async Task ApplyCatalogGateAsync(CancellationToken cancellationToken)
{
var authored = Tags;
if (authored.Count == 0) return;
var tables = authored.Values
.Select(definition => definition.Table)
.Where(table => !string.IsNullOrWhiteSpace(table))
.Distinct(StringComparer.OrdinalIgnoreCase)
.ToArray();
var catalog = await RunBoundedAsync(
token => LoadCatalogAsync(tables, token),
_options.OperationTimeout,
"the catalog queries did not return",
cancellationToken).ConfigureAwait(false);
var result = SqlCatalogGate.Apply(authored.Values, catalog, _dialect);
foreach (var rejection in result.Rejected)
{
_logger.LogWarning(
"Sql tag '{Tag}' rejected by the catalog gate: {Field} — {Reason}. It will publish "
+ "BadNodeIdUnknown until the tag or the database is corrected. Driver={DriverInstanceId}",
rejection.RawPath, rejection.Field, rejection.Reason, _driverInstanceId);
}
var validated = new Dictionary(result.Accepted.Count, StringComparer.Ordinal);
foreach (var definition in result.Accepted) validated[definition.Name] = definition;
// Only the POLLED table is replaced. The authored table is left intact so every authored tag keeps
// its node and a rejected one publishes BadNodeIdUnknown rather than vanishing (see _polledByRawPath).
Volatile.Write(ref _polledByRawPath, validated);
_logger.LogInformation(
"Sql driver {DriverInstanceId} catalog gate accepted {Accepted} of {Total} authored tag(s) "
+ "across {Tables} table(s) on {Endpoint}.",
_driverInstanceId, result.Accepted.Count, authored.Count, tables.Length, Endpoint);
}
/// Opens one connection and reads the catalog slice the authored tables need.
private async Task LoadCatalogAsync(
IReadOnlyCollection authoredTables, CancellationToken cancellationToken)
{
var connection = _factory.CreateConnection()
?? throw new InvalidOperationException(
$"the {_dialect.Provider} provider factory returned no connection.");
await using (connection.ConfigureAwait(false))
{
connection.ConnectionString = _connectionString;
await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
return await SqlCatalogLoader
.LoadAsync(connection, _dialect, authoredTables, _options.CommandTimeout, cancellationToken)
.ConfigureAwait(false);
}
}
// ---- liveness ----
///
/// Opens ONE connection, runs the dialect's liveness statement, and disposes it. No connection is
/// held between this check and the first poll — the reader opens, uses and disposes per poll, so a
/// long-lived connection here would only be a second thing to keep alive.
/// Bounded by wall-clock, not only by the token (the R2-01 / STAB-14 lesson): some ADO.NET
/// providers implement the async path synchronously, and a wedged socket can hang inside the
/// provider's own cancellation handshake. If Initialize hung there, DriverInstanceActor's init
/// task would never complete and the driver would sit in Connecting forever with no retry. The work
/// runs on the thread pool so a synchronous provider blocks a pool thread rather than the caller, and
/// an abandoned attempt still owns and disposes its own connection.
///
/// The caller's token.
/// A task that completes when the database has answered.
private Task VerifyLivenessAsync(CancellationToken cancellationToken) =>
RunBoundedAsync(
async token => { await PingAsync(token).ConfigureAwait(false); return true; },
_options.CommandTimeout,
"the liveness statement did not return",
cancellationToken);
///
/// Runs under a hard wall-clock , on the thread pool,
/// converting a breach into a worded from .
///
///
/// The R2-01 / STAB-14 shape, shared by the two pieces of I/O Initialize performs. It exists
/// because a token is not a deadline: some ADO.NET providers implement the async path synchronously,
/// and a wedged socket can hang inside the provider's own cancellation handshake. Without an
/// independent wall clock, DriverInstanceActor's init task would never complete and the driver
/// would sit in Connecting forever with no retry — the exact frozen-peer wedge the S7 driver shipped.
/// Both the linked CTS deadline and an outer wait are used, because
/// each covers a case the other does not: the CTS handles a provider that does honour
/// cancellation, and the outer wait handles one that does not. An abandoned attempt keeps ownership of
/// its own connection and disposes it; observes its eventual fault.
///
/// The work's result type.
/// The operation to bound. Receives the deadline-linked token.
/// The wall-clock allowance.
/// Sentence fragment naming the operation, e.g. "the catalog queries did not return".
/// The caller's token.
/// The work's result.
/// The budget elapsed first.
private static async Task RunBoundedAsync(
Func> work,
TimeSpan budget,
string what,
CancellationToken cancellationToken)
{
var deadline = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
Task running;
try
{
deadline.CancelAfter(budget);
running = Task.Run(
async () =>
{
try { return await work(deadline.Token).ConfigureAwait(false); }
finally { deadline.Dispose(); }
},
CancellationToken.None);
}
catch
{
// Ownership of the CTS transfers to the work task; release it only if that task never started.
deadline.Dispose();
throw;
}
try
{
return await running.WaitAsync(budget, cancellationToken).ConfigureAwait(false);
}
catch (TimeoutException)
{
Detach(running);
throw new TimeoutException($"{what} within {(int)budget.TotalMilliseconds} ms.");
}
catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
{
// Our deadline fired and the provider DID honour the linked token.
Detach(running);
throw new TimeoutException($"{what} within {(int)budget.TotalMilliseconds} ms.");
}
}
/// Opens a connection and executes the dialect's cheapest proof-of-life statement.
private async Task PingAsync(CancellationToken cancellationToken)
{
var connection = _factory.CreateConnection()
?? throw new InvalidOperationException(
$"the {_dialect.Provider} provider factory returned no connection.");
await using (connection.ConfigureAwait(false))
{
connection.ConnectionString = _connectionString;
await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
var command = connection.CreateCommand();
await using (command.ConfigureAwait(false))
{
command.CommandText = _dialect.LivenessSql;
// ADO.NET reads CommandTimeout = 0 as "wait forever" — the one value it must never become.
command.CommandTimeout = Math.Max(1, (int)Math.Ceiling(_options.CommandTimeout.TotalSeconds));
await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false);
}
}
}
///
/// Observes an abandoned liveness attempt's eventual fault so it never surfaces as an unobserved
/// task exception. The task still owns — and disposes — its own connection.
/// TODO(M2): a verbatim duplicate of SqlPollReader.Detach; both live in Driver.Sql and
/// could hoist to one internal helper. Left in place here to keep this change off the reader.
///
private static void Detach(Task work)
=> _ = work.ContinueWith(
static t => _ = t.Exception,
CancellationToken.None,
TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously,
TaskScheduler.Default);
// ---- endpoint description ----
///
/// Renders a connection string as server/database, reading only those two keys.
/// This is the only function permitted to look at the connection string outside the
/// provider call, and it exists so that no other code is ever tempted to log the string itself:
/// credentials live in User ID / Password / Authentication, which are never
/// read here. A string that cannot be parsed yields — an unusable
/// description must not become a crash at construction.
///
/// The resolved connection string.
/// A credential-free description safe to log and display.
private static string DescribeEndpoint(string connectionString)
{
try
{
var builder = new DbConnectionStringBuilder { ConnectionString = connectionString };
var server = FirstValue(builder, ServerKeys);
if (string.IsNullOrWhiteSpace(server)) return UnknownEndpoint;
var database = FirstValue(builder, DatabaseKeys);
return string.IsNullOrWhiteSpace(database) ? server : $"{server}/{database}";
}
catch (ArgumentException)
{
return UnknownEndpoint;
}
}
/// Reads the first of the builder carries; null when it carries none.
private static string? FirstValue(DbConnectionStringBuilder builder, string[] keys)
{
foreach (var key in keys)
{
// DbConnectionStringBuilder's key comparison is case-insensitive.
if (builder.TryGetValue(key, out var value) && value is not null)
{
var text = value.ToString();
if (!string.IsNullOrWhiteSpace(text)) return text;
}
}
return null;
}
// ---- teardown ----
///
/// Performs the same teardown as , so a caller that only uses
/// await using does not leak the poll loops.
///
/// A task that represents the asynchronous operation.
public async ValueTask DisposeAsync() => await TeardownAsync().ConfigureAwait(false);
///
/// Shared teardown: stops every polled subscription and drops the authored table. Idempotent — safe
/// to call any number of times, in any order, which is what makes
/// ShutdownAsync-then-DisposeAsync (and ) safe.
/// There is no connection to close: the reader opens and disposes one per poll.
///
private async Task TeardownAsync()
{
await _poll.DisposeAsync().ConfigureAwait(false);
Volatile.Write(ref _tagsByRawPath, new Dictionary(StringComparer.Ordinal));
Volatile.Write(ref _polledByRawPath, new Dictionary(StringComparer.Ordinal));
_resolver.Clear(); // no-op in v3 (the resolver reads the live table by closure); kept for symmetry
}
}