using System.Data.Common;
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.
/// Deliberately a local constant, not DriverTypeNames.Sql.
/// DriverTypeNamesGuardTests asserts bidirectional parity between the constants and the
/// driver factories actually registered in the process, so the constant may only be added in the
/// same change that wires the factory (this task ships no factory — see the plan's Task 11, which
/// adds DriverTypeNames.Sql and repoints this constant at it).
///
public const string DriverTypeName = "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) ----
private readonly SqlDriverOptions _options;
private readonly string _driverInstanceId;
private readonly ISqlDialect _dialect;
private readonly 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 readonly 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);
/// Resolves a read/subscribe RawPath to its authored definition — a single table hit, or a miss.
private readonly EquipmentTagRefResolver _resolver;
///
/// 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 readonly 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.
/// 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)
{
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));
_options = options;
_driverInstanceId = driverInstanceId;
_dialect = dialect;
_factory = factory ?? dialect.Factory;
_connectionString = connectionString;
_logger = logger ?? NullLogger.Instance;
Endpoint = DescribeEndpoint(connectionString);
_resolver = new EquipmentTagRefResolver(Lookup);
_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);
_poll = new PollGroupEngine(
reader: ReadAsync,
onChange: (handle, tagRef, snapshot) =>
OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, tagRef, snapshot)),
onError: HandlePollError,
backoffCap: PollBackoffCap);
}
///
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; }
/// 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 resolver's lookup: RawPath → authored definition, or null on a miss.
private SqlTagDefinition? Lookup(string rawPath) => Tags.GetValueOrDefault(rawPath);
// ---- IDriver lifecycle ----
///
/// Builds the authored RawPath table and proves the database is reachable.
/// is not re-parsed here: the driver serves the
/// typed it was constructed with, exactly as ModbusDriver does
/// (config parsing belongs to the factory, which builds a fresh instance).
/// 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 driver configuration JSON (unused; see remarks).
/// Cancellation token for the operation.
/// A task that represents the asynchronous operation.
/// The database could not be reached.
public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
WriteHealth(new DriverHealth(DriverState.Initializing, null, null));
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);
}
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, for a provider whose text echoes credentials, re-open the C1 leak) and log
// the outage a second time. So skip the connection class ReadAsync owns; the only exceptions that
// reach past this guard are the engine's own contract-violation throws — an InvalidOperationException
// carrying no provider text — which nothing else classifies.
if (ex is DbException) return;
_logger.LogWarning(ex, "Sql poll failed. Driver={DriverInstanceId} Endpoint={Endpoint}",
_driverInstanceId, Endpoint);
Degrade(ex.Message);
}
///
/// 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);
}
// ---- 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 async Task VerifyLivenessAsync(CancellationToken cancellationToken)
{
var budget = _options.CommandTimeout;
var deadline = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
Task work;
try
{
deadline.CancelAfter(budget);
work = Task.Run(
async () =>
{
try { await PingAsync(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
{
await work.WaitAsync(budget, cancellationToken).ConfigureAwait(false);
}
catch (TimeoutException)
{
Detach(work);
throw new TimeoutException(
$"the liveness statement did not return within {(int)budget.TotalMilliseconds} ms.");
}
catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
{
// Our deadline fired and the provider DID honour the linked token.
Detach(work);
throw new TimeoutException(
$"the liveness statement did not return 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));
_resolver.Clear(); // no-op in v3 (the resolver reads the live table by closure); kept for symmetry
}
}