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

feat(sql): SqlDriver shell wiring PollGroupEngine + probe + read-only discovery

Browse Source 
SqlDriver implements IDriver / ITagDiscovery / IReadable / ISubscribable /
IHostConnectivityProbe / IAsyncDisposable over the landed SqlPollReader,
mirroring ModbusDriver. IWritable is deliberately absent: v1 is read-only
structurally, and every discovered variable is SecurityClass=ViewOnly.

The shell classifies poll outcomes because nothing below it does. The reader
honours IReadable literally — it throws only when the database is unreachable
and Bad-codes everything else — so a frozen database returns all-BadTimeout
snapshots through a perfectly successful PollGroupEngine tick. Without
ObservePollOutcome the driver would report Healthy while every value was Bad.
Connection-class codes (BadTimeout / BadCommunicationError) degrade health and
report the host Stopped; authoring-class codes (unresolvable RawPath, absent
row, type mismatch) change nothing, so a tag typo never reports the database
down. It deliberately does NOT synthesise an exception to earn engine backoff:
the engine's exception path publishes nothing, which would cost clients the
Bad quality the reader went out of its way to produce.

Initialize builds the authored RawPath table first (pure, cannot fail; a
malformed TagConfig is logged and skipped) then verifies liveness over one
open-use-dispose connection bounded by wall clock as well as by token (the
R2-01 lesson) — failure records Faulted and rethrows so DriverInstanceActor
retries. The connection string is never logged: a credential-free
server/database Endpoint is the only rendering that reaches a log, LastError,
or the host status.

DriverTypeNames.Sql is NOT added here — DriverTypeNamesGuardTests asserts
bidirectional parity with registered factories, so the constant must land with
the factory (Task 11). SqlDriver.DriverTypeName carries the string until then.

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
2026-07-24 14:48:14 -04:00
parent 0efee7413a
commit 9b30bdeb7a
4 changed files with 1247 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriver.cs
+694
View File
@@ -0,0 +1,694 @@
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;
/// <summary>
/// The <c>Sql</c> driver instance: a <b>read-only</b> Equipment-kind driver that polls SQL tables/views
/// and publishes selected cells as OPC UA variables. Structurally the SQL analogue of
/// <c>ModbusDriver</c> — this type owns lifecycle, the authored RawPath table, health, host
/// connectivity, and the <see cref="PollGroupEngine"/> subscription overlay, while
/// <see cref="SqlPollReader"/> owns the read path.
/// <para><b>Read-only is structural, not configured.</b> <see cref="IWritable"/> is deliberately not
/// implemented, so no amount of config (and no <c>WriteOperate</c> role) can produce a write; every
/// discovered variable is <see cref="SecurityClassification.ViewOnly"/>. A future write feature is a
/// new capability, not a flag flip.</para>
/// <para><b>Health is classified here, because nothing below does it.</b> The reader honours
/// <see cref="IReadable"/> 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
/// <em>successfully</em>, so <see cref="PollGroupEngine"/> sees a clean tick, does not back off, and does
/// not call <see cref="HandlePollError"/>. Without <see cref="ObservePollOutcome"/> below, a driver whose
/// every value is <see cref="SqlStatusCodes.BadTimeout"/> would keep reporting
/// <see cref="DriverState.Healthy"/>. See that method for the exact rule and for why it does not
/// manufacture an exception to force backoff.</para>
/// </summary>
public sealed class SqlDriver
: IDriver, ITagDiscovery, IReadable, ISubscribable, IHostConnectivityProbe, IAsyncDisposable
{
/// <summary>
/// The driver-type string this driver registers under.
/// <para><b>Deliberately a local constant, not <c>DriverTypeNames.Sql</c>.</b>
/// <c>DriverTypeNamesGuardTests</c> 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 <c>DriverTypeNames.Sql</c> and repoints this constant at it).</para>
/// </summary>
public const string DriverTypeName = "Sql";
/// <summary>Shown for a connection string that names no recognisable server.</summary>
private const string UnknownEndpoint = "(unknown sql endpoint)";
/// <summary>Upper bound on the poll-loop failure backoff — the S7-proven 30 s cap adopted fleet-wide (05/STAB-8).</summary>
private static readonly TimeSpan PollBackoffCap = TimeSpan.FromSeconds(30);
/// <summary>Connection-string keys that name the server, in the order they are consulted.</summary>
private static readonly string[] ServerKeys =
["server", "data source", "datasource", "host", "address", "addr", "network address"];
/// <summary>Connection-string keys that name the database.</summary>
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;
/// <summary>
/// The resolved connection string — a <b>secret</b>. It is passed to the provider and to nothing
/// else: every log line, health message and host status carries <see cref="Endpoint"/> instead.
/// </summary>
private readonly string _connectionString;
private readonly ILogger<SqlDriver> _logger;
/// <summary>
/// The authored RawPath → definition table, held as an <b>immutable snapshot swapped atomically</b>
/// rather than as a mutated dictionary (the shape <c>ModbusDriver</c> uses).
/// <see cref="ReinitializeAsync"/> rebuilds this table while poll loops are reading it; clearing and
/// refilling a shared <see cref="Dictionary{TKey,TValue}"/> under that concurrency is a torn read at
/// best and an <see cref="InvalidOperationException"/> inside a poll at worst.
/// </summary>
private IReadOnlyDictionary<string, SqlTagDefinition> _tagsByRawPath =
new Dictionary<string, SqlTagDefinition>(StringComparer.Ordinal);
/// <summary>Resolves a read/subscribe RawPath to its authored definition — a single table hit, or a miss.</summary>
private readonly EquipmentTagRefResolver<SqlTagDefinition> _resolver;
/// <summary>
/// The read path. Built once, in the constructor, and <b>not</b> injected: its <c>resolve</c>
/// delegate must read this driver's live authored table, which does not exist before the driver does.
/// </summary>
private readonly SqlPollReader _reader;
/// <summary>Polled subscriptions. The driver supplies the reader + change bridge; the engine owns the loop.</summary>
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);
/// <summary>Occurs when a subscribed tag's value or quality changes.</summary>
public event EventHandler<DataChangeEventArgs>? OnDataChange;
/// <summary>Occurs when the database endpoint transitions between reachable and unreachable.</summary>
public event EventHandler<HostStatusChangedEventArgs>? OnHostStatusChanged;
// ---- ctor + identity ----
/// <summary>Initializes a new <c>Sql</c> driver instance.</summary>
/// <param name="options">The driver's typed configuration (the factory applies the documented defaults).</param>
/// <param name="driverInstanceId">The central config DB's identity for this driver instance.</param>
/// <param name="dialect">
/// The provider seam — identifier quoting, liveness SQL, row-limit syntax and column-type mapping.
/// </param>
/// <param name="connectionString">
/// The <b>already-resolved</b> connection string. The driver never resolves a
/// <c>connectionStringRef</c> itself and never logs this value.
/// </param>
/// <param name="factory">
/// Creates the provider's connections. Defaults to <paramref name="dialect"/>'s factory; passed
/// explicitly by tests that need to observe or delay connection creation.
/// </param>
/// <param name="logger">Optional; defaults to a no-op logger.</param>
/// <exception cref="ArgumentNullException">A required reference argument is null.</exception>
/// <exception cref="ArgumentException"><paramref name="driverInstanceId"/> or <paramref name="connectionString"/> is blank.</exception>
public SqlDriver(
SqlDriverOptions options,
string driverInstanceId,
ISqlDialect dialect,
string connectionString,
DbProviderFactory? factory = null,
ILogger<SqlDriver>? 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<SqlDriver>.Instance;
Endpoint = DescribeEndpoint(connectionString);
_resolver = new EquipmentTagRefResolver<SqlTagDefinition>(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);
}
/// <inheritdoc />
public string DriverInstanceId => _driverInstanceId;
/// <inheritdoc />
public string DriverType => DriverTypeName;
/// <summary>
/// The credential-free description of the database this instance polls — <c>server/database</c> 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.
/// </summary>
public string Endpoint { get; }
/// <summary>Host identifier surfaced through <see cref="IHostConnectivityProbe"/> — the endpoint description.</summary>
public string HostName => Endpoint;
/// <summary>Active polled-subscription count. Diagnostics + disposal assertions.</summary>
internal int ActiveSubscriptionCount => _poll.ActiveSubscriptionCount;
/// <summary>The current authored-tag snapshot. Read through a barrier — <see cref="BuildTagTable"/> swaps it.</summary>
private IReadOnlyDictionary<string, SqlTagDefinition> Tags => Volatile.Read(ref _tagsByRawPath);
/// <summary>The resolver's lookup: RawPath → authored definition, or null on a miss.</summary>
private SqlTagDefinition? Lookup(string rawPath) => Tags.GetValueOrDefault(rawPath);
// ---- IDriver lifecycle ----
/// <summary>
/// Builds the authored RawPath table and proves the database is reachable.
/// <para><paramref name="driverConfigJson"/> is <b>not</b> re-parsed here: the driver serves the
/// typed <see cref="SqlDriverOptions"/> it was constructed with, exactly as <c>ModbusDriver</c> does
/// (config parsing belongs to the factory, which builds a fresh instance).</para>
/// <para>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 <see cref="DriverState.Faulted"/> <b>and rethrows</b> —
/// <c>DriverInstanceActor</c> 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.</para>
/// </summary>
/// <param name="driverConfigJson">The driver configuration JSON (unused; see remarks).</param>
/// <param name="cancellationToken">Cancellation token for the operation.</param>
/// <returns>A task that represents the asynchronous operation.</returns>
/// <exception cref="InvalidOperationException">The database could not be reached.</exception>
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 — Endpoint is the credential-free rendering.
var message =
$"Sql driver '{_driverInstanceId}' could not reach {Endpoint}: {ex.Message} " +
$"Check the database is up, the network path is open, and the configured credentials are 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);
}
/// <inheritdoc />
public async Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
await ShutdownAsync(cancellationToken).ConfigureAwait(false);
await InitializeAsync(driverConfigJson, cancellationToken).ConfigureAwait(false);
}
/// <inheritdoc />
public async Task ShutdownAsync(CancellationToken cancellationToken)
{
var lastRead = ReadHealth().LastSuccessfulRead;
await TeardownAsync().ConfigureAwait(false);
WriteHealth(new DriverHealth(DriverState.Unknown, lastRead, null));
}
/// <inheritdoc />
public DriverHealth GetHealth() => ReadHealth();
/// <summary>No caches, no symbol table, no connection between polls — the footprint is the tag table.</summary>
/// <returns>Always zero.</returns>
public long GetMemoryFootprint() => 0;
/// <summary>Nothing optional to flush: the authored table is required for correctness.</summary>
/// <param name="cancellationToken">Cancellation token for the operation.</param>
/// <returns>A completed task.</returns>
public Task FlushOptionalCachesAsync(CancellationToken cancellationToken) => Task.CompletedTask;
// ---- ITagDiscovery ----
/// <summary>
/// Exactly one pass: the tag set is the authored one, fixed at Initialize, so re-running discovery
/// could only produce the same nodes.
/// </summary>
public DiscoveryRediscoverPolicy RediscoverPolicy => DiscoveryRediscoverPolicy.Once;
/// <summary>
/// False — <see cref="DiscoverAsync"/> 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: <c>Driver.Sql.Browser</c>.)
/// </summary>
public bool SupportsOnlineDiscovery => false;
/// <summary>
/// Streams the authored tags as read-only variables.
/// <para>An undeclared tag materialises as <see cref="DriverDataType.String"/> — the same fallback
/// <see cref="ISqlDialect.MapColumnType"/> 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 <c>"type"</c> on a numeric tag.</para>
/// </summary>
/// <param name="builder">The address-space builder to stream discovered nodes into.</param>
/// <param name="cancellationToken">Cancellation token for the operation.</param>
/// <returns>A completed task — discovery is synchronous over the authored table.</returns>
public Task DiscoverAsync(IAddressSpaceBuilder builder, CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(builder);
var folder = builder.Folder(DriverTypeName, DriverTypeName);
foreach (var tag in Tags.Values)
{
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 ----
/// <summary>
/// Reads a batch, delegating to <see cref="SqlPollReader"/> and classifying what came back
/// (<see cref="ObservePollOutcome"/>).
/// </summary>
/// <param name="fullReferences">The RawPaths to read.</param>
/// <param name="cancellationToken">Cancellation token for the operation.</param>
/// <returns>One snapshot per reference, at the same index.</returns>
/// <exception cref="DbException">The database could not be reached — the engine backs off on this.</exception>
public async Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(
IReadOnlyList<string> 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);
Degrade($"Sql driver '{_driverInstanceId}' could not reach {Endpoint}: {ex.Message}");
TransitionHostTo(HostState.Stopped);
throw;
}
}
// ---- ISubscribable (polling overlay via the shared engine) ----
/// <summary>
/// Registers a polled subscription.
/// <para>A non-positive <paramref name="publishingInterval"/> falls back to the configured
/// <see cref="SqlDriverOptions.DefaultPollInterval"/>; anything faster than
/// <see cref="PollGroupEngine.DefaultMinInterval"/> is floored by the engine.</para>
/// </summary>
/// <param name="fullReferences">The RawPaths to poll.</param>
/// <param name="publishingInterval">The requested publishing interval.</param>
/// <param name="cancellationToken">Cancellation token for the operation.</param>
/// <returns>The subscription handle.</returns>
public Task<ISubscriptionHandle> SubscribeAsync(
IReadOnlyList<string> fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken)
{
var interval = publishingInterval > TimeSpan.Zero ? publishingInterval : _options.DefaultPollInterval;
return Task.FromResult(_poll.Subscribe(fullReferences, interval));
}
/// <inheritdoc />
public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
{
_poll.Unsubscribe(handle);
return Task.CompletedTask;
}
// ---- IHostConnectivityProbe ----
/// <summary>
/// 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.
/// </summary>
/// <returns>A one-element list describing the configured database endpoint.</returns>
public IReadOnlyList<HostConnectivityStatus> GetHostStatuses()
{
lock (_probeLock)
return [new HostConnectivityStatus(HostName, _hostState, _hostStateChangedUtc)];
}
// ---- health + host-state classification ----
/// <summary>
/// Classifies one poll's snapshots into a health verdict and a host state.
/// <list type="bullet">
/// <item>Any Good snapshot ⇒ the database answered: <c>LastSuccessfulRead</c> advances and the
/// host is Running.</item>
/// <item>Any <b>connection-class</b> Bad snapshot (<see cref="SqlStatusCodes.BadTimeout"/> /
/// <see cref="SqlStatusCodes.BadCommunicationError"/>) ⇒ Degraded; when nothing at all was Good,
/// the host is Stopped.</item>
/// <item>Bad codes that are <b>authoring</b> 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.</item>
/// </list>
/// <para><b>Why this does not throw to force poll-engine backoff.</b> Backoff would require turning
/// an all-<c>BadTimeout</c> 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 <c>operationTimeout</c> and the reader never holds more than
/// <c>maxConcurrentGroups</c> 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.</para>
/// </summary>
/// <param name="snapshots">The snapshots the reader returned for one poll.</param>
private void ObservePollOutcome(IReadOnlyList<DataValueSnapshot> 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.
WriteHealth(new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null));
TransitionHostTo(HostState.Running);
}
/// <summary>True for the status codes that mean "the database did not answer", as opposed to "the data is not there".</summary>
private static bool IsConnectionClass(uint statusCode)
=> statusCode is SqlStatusCodes.BadTimeout or SqlStatusCodes.BadCommunicationError;
/// <summary>
/// Routes a poll-loop reader failure to the health surface (05/STAB-9). Deliberately does <b>not</b>
/// touch host state: the engine also reports its own contract violations here, which say nothing
/// about connectivity — the unreachable-database transition is raised by <see cref="ReadAsync"/>,
/// which knows the exception came from the reader.
/// </summary>
/// <param name="ex">The exception the poll engine caught.</param>
internal void HandlePollError(Exception ex)
{
_logger.LogWarning(ex, "Sql poll failed. Driver={DriverInstanceId} Endpoint={Endpoint}",
_driverInstanceId, Endpoint);
Degrade(ex.Message);
}
/// <summary>
/// Degrades health, preserving <c>LastSuccessfulRead</c> and never downgrading
/// <see cref="DriverState.Faulted"/> — a config-level verdict that only a successful read or an
/// operator reinitialize may clear.
/// </summary>
/// <param name="reason">The operator-facing reason, which must never carry the connection string.</param>
private void Degrade(string reason)
{
var current = ReadHealth();
if (current.State == DriverState.Faulted) return;
WriteHealth(new DriverHealth(DriverState.Degraded, current.LastSuccessfulRead, reason));
}
/// <summary>Advances <c>LastSuccessfulRead</c> without changing the current state or error.</summary>
private void TouchLastSuccessfulRead()
{
var current = ReadHealth();
WriteHealth(current with { LastSuccessfulRead = DateTime.UtcNow });
}
/// <summary>Barrier-protected read of the multi-threaded health field.</summary>
private DriverHealth ReadHealth() => Volatile.Read(ref _health);
/// <summary>Barrier-protected publish of a new health snapshot.</summary>
private void WriteHealth(DriverHealth value) => Volatile.Write(ref _health, value);
/// <summary>Records a host-state change and raises <see cref="OnHostStatusChanged"/> — on transitions only.</summary>
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 ----
/// <summary>
/// Maps every authored <see cref="RawTagEntry"/> through <see cref="SqlEquipmentTagParser.TryParse"/>
/// and publishes the result as one atomic snapshot. A blob that does not map is <b>logged and
/// skipped</b>, 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 <see cref="SqlStatusCodes.BadNodeIdUnknown"/> rather than someone else's row.
/// </summary>
private void BuildTagTable()
{
var table = new Dictionary<string, SqlTagDefinition>(_options.RawTags.Count, StringComparer.Ordinal);
foreach (var entry in _options.RawTags)
{
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);
}
Volatile.Write(ref _tagsByRawPath, table);
}
// ---- liveness ----
/// <summary>
/// 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.
/// <para><b>Bounded by wall-clock, not only by the token</b> (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, <c>DriverInstanceActor</c>'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.</para>
/// </summary>
/// <param name="cancellationToken">The caller's token.</param>
/// <returns>A task that completes when the database has answered.</returns>
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.");
}
}
/// <summary>Opens a connection and executes the dialect's cheapest proof-of-life statement.</summary>
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);
}
}
}
/// <summary>
/// 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.
/// </summary>
private static void Detach(Task work)
=> _ = work.ContinueWith(
static t => _ = t.Exception,
CancellationToken.None,
TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously,
TaskScheduler.Default);
// ---- endpoint description ----
/// <summary>
/// Renders a connection string as <c>server/database</c>, reading only those two keys.
/// <para>This is the <b>only</b> 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 <c>User ID</c> / <c>Password</c> / <c>Authentication</c>, which are never
/// read here. A string that cannot be parsed yields <see cref="UnknownEndpoint"/> — an unusable
/// description must not become a crash at construction.</para>
/// </summary>
/// <param name="connectionString">The resolved connection string.</param>
/// <returns>A credential-free description safe to log and display.</returns>
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;
}
}
/// <summary>Reads the first of <paramref name="keys"/> the builder carries; null when it carries none.</summary>
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 ----
/// <summary>
/// Performs the same teardown as <see cref="ShutdownAsync"/>, so a caller that only uses
/// <c>await using</c> does not leak the poll loops.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
public async ValueTask DisposeAsync() => await TeardownAsync().ConfigureAwait(false);
/// <summary>
/// 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
/// <c>ShutdownAsync</c>-then-<c>DisposeAsync</c> (and <see cref="ReinitializeAsync"/>) safe.
/// <para>There is no connection to close: the reader opens and disposes one per poll.</para>
/// </summary>
private async Task TeardownAsync()
{
await _poll.DisposeAsync().ConfigureAwait(false);
Volatile.Write(ref _tagsByRawPath, new Dictionary<string, SqlTagDefinition>(StringComparer.Ordinal));
_resolver.Clear(); // no-op in v3 (the resolver reads the live table by closure); kept for symmetry
}
}
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriverOptions.cs
+56
View File
@@ -0,0 +1,56 @@
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts;
namespace ZB.MOM.WW.OtOpcUa.Driver.Sql;
/// <summary>
/// The typed form of a <c>Sql</c> driver instance's configuration — what
/// <see cref="SqlDriverConfigDto"/> becomes once the factory has applied its defaults (design §5.1).
/// Mirrors <c>ModbusDriverOptions</c>: the driver instance is constructed with these, and its
/// <see cref="SqlDriver.InitializeAsync"/> serves them rather than re-parsing the config JSON.
/// <para><b>No connection string here.</b> The resolved connection string is a separate constructor
/// argument, because it is a secret and these options are safe to log, diff and hold in a snapshot.</para>
/// </summary>
public sealed class SqlDriverOptions
{
/// <summary>
/// The authored raw tags this driver serves. The deploy artifact hands each authored raw <c>Tag</c>
/// as a <see cref="RawTagEntry"/> (RawPath identity + driver <c>TagConfig</c> blob); the driver maps
/// each through <see cref="SqlEquipmentTagParser.TryParse"/> into its RawPath → definition table.
/// A SQL source has no tag-discovery protocol — the driver serves exactly these.
/// </summary>
public IReadOnlyList<RawTagEntry> RawTags { get; init; } = [];
/// <summary>
/// Poll interval used when a subscriber does not ask for one (design §4: <c>defaultPollInterval</c>,
/// default 5 s). A subscriber that names an interval gets that interval, floored by
/// <see cref="PollGroupEngine.DefaultMinInterval"/>.
/// </summary>
public TimeSpan DefaultPollInterval { get; init; } = TimeSpan.FromSeconds(5);
/// <summary>
/// The client-side wall-clock ceiling on one group query (design §8.3, default 15 s). A breach
/// publishes <see cref="SqlStatusCodes.BadTimeout"/> for that group's tags.
/// </summary>
public TimeSpan OperationTimeout { get; init; } = TimeSpan.FromSeconds(15);
/// <summary>
/// The ADO.NET <c>CommandTimeout</c> server-side backstop (default 10 s), also the bound on the
/// Initialize-time liveness check.
/// <para>Authoring must keep <see cref="OperationTimeout"/> strictly greater than this (design §8.3);
/// inverted, the client-side abort always fires first and masks the backstop. <b>That rule is
/// enforced by config validation in the factory, not here</b> — the reader deliberately stays usable
/// with the order inverted so the frozen-database tests can prove the client-side bound fires.</para>
/// </summary>
public TimeSpan CommandTimeout { get; init; } = TimeSpan.FromSeconds(10);
/// <summary>Cap on concurrently executing group queries — and therefore on concurrently open connections.</summary>
public int MaxConcurrentGroups { get; init; } = 4;
/// <summary>
/// When <see langword="true"/>, a present row whose value cell is NULL publishes
/// <see cref="SqlStatusCodes.Bad"/> instead of the default <see cref="SqlStatusCodes.Uncertain"/>.
/// It never governs an absent row, which is always <see cref="SqlStatusCodes.BadNoData"/>.
/// </summary>
public bool NullIsBad { get; init; }
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlDriverTests.cs
+490
View File
@@ -0,0 +1,490 @@
using System.Data.Common;
using System.Globalization;
using Microsoft.Data.Sqlite;
using Microsoft.Extensions.Logging;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Sql.Contracts;
namespace ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests;
/// <summary>
/// Proves the <see cref="SqlDriver"/> shell — the part <see cref="SqlPollReader"/> deliberately does not
/// own: lifecycle (initialize / reinitialize / dispose), the authored RawPath table, read-only discovery,
/// the poll-engine subscription overlay, and the health + host-connectivity surface.
/// <para><b>The health assertions are the point of this class.</b> The reader returns Bad-coded snapshots
/// rather than throwing for everything short of an unreachable server, so nothing below the shell degrades
/// health on its own: a frozen database yields all-<c>BadTimeout</c> snapshots and a perfectly successful
/// <see cref="PollGroupEngine"/> tick. If the shell does not classify what came back, the driver reports
/// <see cref="DriverState.Healthy"/> while every value is Bad — the failure mode these tests exist to
/// prevent, together with its inverse (a tag typo must NOT report the database down).</para>
/// </summary>
public sealed class SqlDriverTests
{
private const string DriverInstanceId = "sql-1";
// ---- discovery + initialize ----
[Fact]
public async Task Initialize_thenDiscover_streamsAuthoredTagsReadOnly()
{
using var fixture = new SqlitePollFixture();
await using var driver = NewDriver(fixture, KvEntry("Speed", SqlitePollFixture.PresentKey));
await driver.InitializeAsync(ConfigJson, CancellationToken.None);
driver.GetHealth().State.ShouldBe(DriverState.Healthy);
driver.DriverType.ShouldBe("Sql");
driver.DriverInstanceId.ShouldBe(DriverInstanceId);
((ITagDiscovery)driver).SupportsOnlineDiscovery.ShouldBeFalse();
((ITagDiscovery)driver).RediscoverPolicy.ShouldBe(DiscoveryRediscoverPolicy.Once);
var capture = new CapturingBuilder();
await ((ITagDiscovery)driver).DiscoverAsync(capture, CancellationToken.None);
capture.Variables.ShouldContain(v =>
v.Info.FullName == "Speed" && v.Info.SecurityClass == SecurityClassification.ViewOnly);
// v1 is read-only by construction, not by configuration.
driver.ShouldNotBeAssignableTo<IWritable>();
}
[Fact]
public async Task Initialize_anUnparseableTagConfig_isSkippedAndLogged_andTheRestStillServe()
{
using var fixture = new SqlitePollFixture();
var logger = new CapturingLogger();
await using var driver = NewDriver(
fixture,
logger,
KvEntry("Speed", SqlitePollFixture.PresentKey),
new RawTagEntry("Broken", "not json at all", WriteIdempotent: false));
// A single malformed blob must never fail the whole driver.
await driver.InitializeAsync(ConfigJson, CancellationToken.None);
driver.GetHealth().State.ShouldBe(DriverState.Healthy);
var capture = new CapturingBuilder();
await ((ITagDiscovery)driver).DiscoverAsync(capture, CancellationToken.None);
capture.Variables.Count.ShouldBe(1);
capture.Variables[0].Info.FullName.ShouldBe("Speed");
logger.Entries.ShouldContain(e => e.Level == LogLevel.Warning && e.Message.Contains("Broken"));
// …and the skipped tag resolves to nothing rather than to someone else's row.
var snapshots = await driver.ReadAsync(["Broken"], CancellationToken.None);
snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.BadNodeIdUnknown);
}
[Fact]
public async Task Initialize_whenTheDatabaseIsUnreachable_faultsWithAnActionableErrorAndNoCredentials()
{
using var fixture = new SqlitePollFixture();
await using var driver = NewDriver(
fixture, Unreachable(), CapturingLogger.Null, KvEntry("Speed", SqlitePollFixture.PresentKey));
// Mirrors ModbusDriver: the driver records Faulted AND rethrows, so DriverInstanceActor lands in
// Reconnecting and its retry-connect timer keeps trying — a database that is merely down must not
// seal as a silently-connected driver.
var thrown = await Should.ThrowAsync<Exception>(
async () => await driver.InitializeAsync(ConfigJson, CancellationToken.None));
var health = driver.GetHealth();
health.State.ShouldBe(DriverState.Faulted);
health.LastError.ShouldNotBeNullOrWhiteSpace();
// Actionable: it names the endpoint the operator has to go look at…
health.LastError!.ShouldContain(NoSuchDatabaseName);
thrown.Message.ShouldContain(NoSuchDatabaseName);
// …and never the credential-bearing connection string.
health.LastError.ShouldNotContain(SecretToken);
thrown.Message.ShouldNotContain(SecretToken);
driver.GetHostStatuses()[0].State.ShouldBe(HostState.Stopped);
driver.GetHostStatuses()[0].HostName.ShouldNotContain(SecretToken);
}
[Fact]
public async Task ReinitializeAsync_recoversFromFaulted_onceTheDatabaseIsReachableAgain()
{
using var fixture = new SqlitePollFixture();
// SQLite creates a missing FILE on open but cannot create a missing DIRECTORY — so this connection
// string is unreachable until the directory exists, and reachable the moment it does.
var directory = Path.Combine(Path.GetTempPath(), $"otopcua-sql-driver-{Guid.NewGuid():N}");
var connectionString = new SqliteConnectionStringBuilder
{
DataSource = Path.Combine(directory, "late.db"),
}.ToString();
await using var driver = NewDriver(
fixture, connectionString, CapturingLogger.Null, KvEntry("Speed", SqlitePollFixture.PresentKey));
await Should.ThrowAsync<Exception>(
async () => await driver.InitializeAsync(ConfigJson, CancellationToken.None));
driver.GetHealth().State.ShouldBe(DriverState.Faulted);
try
{
Directory.CreateDirectory(directory);
await driver.ReinitializeAsync(ConfigJson, CancellationToken.None);
driver.GetHealth().State.ShouldBe(DriverState.Healthy);
driver.GetHostStatuses()[0].State.ShouldBe(HostState.Running);
// The authored table survived the round trip — a recovered driver still serves its tags.
var capture = new CapturingBuilder();
await ((ITagDiscovery)driver).DiscoverAsync(capture, CancellationToken.None);
capture.Variables.Count.ShouldBe(1);
}
finally
{
SqliteConnection.ClearAllPools();
try { Directory.Delete(directory, recursive: true); } catch (IOException) { }
}
}
// ---- IReadable delegation ----
[Fact]
public async Task ReadAsync_delegatesToTheReader_valueForValue()
{
using var fixture = new SqlitePollFixture();
await using var driver = NewDriver(
fixture,
KvEntry("Speed", SqlitePollFixture.PresentKey),
KvEntry("Temp", SqlitePollFixture.NullValueKey),
KvEntry("Missing", SqlitePollFixture.AbsentKey));
await driver.InitializeAsync(ConfigJson, CancellationToken.None);
var reference = new SqlPollReader(
fixture.Factory, fixture.ConnectionString, new SqliteDialect(),
commandTimeout: TimeSpan.FromSeconds(10), operationTimeout: TimeSpan.FromSeconds(15),
maxConcurrentGroups: 4, nullIsBad: false,
resolve: rawPath => rawPath switch
{
"Speed" => KvDefinition("Speed", SqlitePollFixture.PresentKey),
"Temp" => KvDefinition("Temp", SqlitePollFixture.NullValueKey),
"Missing" => KvDefinition("Missing", SqlitePollFixture.AbsentKey),
_ => null,
});
string[] refs = ["Speed", "Temp", "Missing", "NotAuthored"];
var throughDriver = await driver.ReadAsync(refs, CancellationToken.None);
var throughReader = await reference.ReadAsync(refs, CancellationToken.None);
throughDriver.Count.ShouldBe(throughReader.Count);
for (var i = 0; i < throughDriver.Count; i++)
{
throughDriver[i].Value.ShouldBe(throughReader[i].Value);
throughDriver[i].StatusCode.ShouldBe(throughReader[i].StatusCode);
throughDriver[i].SourceTimestampUtc.ShouldBe(throughReader[i].SourceTimestampUtc);
}
}
// ---- the sustained-timeout decision ----
[Fact]
public async Task ReadAsync_whenEveryTagTimesOut_degradesHealthAndReportsTheHostStopped()
{
using var fixture = new SqlitePollFixture();
await using var driver = NewDriver(
fixture, fixture.ConnectionString, CapturingLogger.Null,
// Deliberately inverted (operationTimeout < commandTimeout) so the CLIENT-side bound fires while
// the server-side backstop would still be waiting — the reader's frozen-peer shape.
operationTimeout: TimeSpan.FromMilliseconds(500),
commandTimeout: TimeSpan.FromSeconds(30),
rawTags: [KvEntry("Speed", SqlitePollFixture.PresentKey)]);
await driver.InitializeAsync(ConfigJson, CancellationToken.None);
driver.GetHealth().State.ShouldBe(DriverState.Healthy);
// A held EXCLUSIVE transaction is this rig's frozen database (see SqlPollReaderTests).
using var locker = fixture.OpenNewConnection();
using (var begin = locker.CreateCommand())
{
begin.CommandText = "BEGIN EXCLUSIVE";
begin.ExecuteNonQuery();
}
try
{
var snapshots = await driver.ReadAsync(["Speed"], CancellationToken.None);
// The reader does exactly what it promises — Bad-coded snapshots, no exception…
snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.BadTimeout);
// …so the poll engine sees a clean tick, and ONLY the shell's own classification degrades health.
var health = driver.GetHealth();
health.State.ShouldBe(DriverState.Degraded);
health.LastError.ShouldNotBeNullOrWhiteSpace();
driver.GetHostStatuses()[0].State.ShouldBe(HostState.Stopped);
}
finally
{
using var rollback = locker.CreateCommand();
rollback.CommandText = "ROLLBACK";
rollback.ExecuteNonQuery();
}
}
[Fact]
public async Task ReadAsync_anUnresolvableRef_isNotAConnectivityFact_soHealthAndHostStand()
{
// The falsifiability control for the test above: a Bad-coded batch must degrade the driver only when
// the Bad codes are connection-class. An authoring typo reporting "the database is down" would send
// an operator to the wrong system entirely.
using var fixture = new SqlitePollFixture();
await using var driver = NewDriver(fixture, KvEntry("Speed", SqlitePollFixture.PresentKey));
await driver.InitializeAsync(ConfigJson, CancellationToken.None);
(await driver.ReadAsync(["Speed"], CancellationToken.None))[0].StatusCode.ShouldBe(SqlStatusCodes.Good);
var snapshots = await driver.ReadAsync(["TypoedTagName"], CancellationToken.None);
snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.BadNodeIdUnknown);
driver.GetHealth().State.ShouldBe(DriverState.Healthy);
driver.GetHostStatuses()[0].State.ShouldBe(HostState.Running);
}
// ---- ISubscribable (poll-engine overlay) ----
[Fact]
public async Task SubscribeAsync_deliversAnInitialChange_andUnsubscribeStopsThePolling()
{
using var fixture = new SqlitePollFixture();
await using var driver = NewDriver(fixture, KvEntry("Speed", SqlitePollFixture.PresentKey));
await driver.InitializeAsync(ConfigJson, CancellationToken.None);
var first = new TaskCompletionSource<DataChangeEventArgs>(
TaskCreationOptions.RunContinuationsAsynchronously);
var changes = 0;
driver.OnDataChange += (_, args) =>
{
Interlocked.Increment(ref changes);
first.TrySetResult(args);
};
var handle = await driver.SubscribeAsync(
["Speed"], TimeSpan.FromMilliseconds(100), CancellationToken.None);
var initial = await first.Task.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken);
initial.FullReference.ShouldBe("Speed");
initial.Snapshot.Value.ShouldBe(SqlitePollFixture.PresentValue);
initial.SubscriptionHandle.ShouldBe(handle);
await driver.UnsubscribeAsync(handle, CancellationToken.None);
driver.ActiveSubscriptionCount.ShouldBe(0);
// Unsubscribe awaits the loop task, so nothing may arrive after it returns.
var afterUnsubscribe = Volatile.Read(ref changes);
await Task.Delay(400, TestContext.Current.CancellationToken);
Volatile.Read(ref changes).ShouldBe(afterUnsubscribe);
}
// ---- disposal ----
[Fact]
public async Task DisposeAsync_stopsThePollEngine_andIsIdempotent()
{
using var fixture = new SqlitePollFixture();
var driver = NewDriver(fixture, KvEntry("Speed", SqlitePollFixture.PresentKey));
await driver.InitializeAsync(ConfigJson, CancellationToken.None);
var changes = 0;
driver.OnDataChange += (_, _) => Interlocked.Increment(ref changes);
_ = await driver.SubscribeAsync(["Speed"], TimeSpan.FromMilliseconds(100), CancellationToken.None);
await Task.Delay(300, TestContext.Current.CancellationToken);
await driver.DisposeAsync();
driver.ActiveSubscriptionCount.ShouldBe(0);
var afterDispose = Volatile.Read(ref changes);
await Task.Delay(400, TestContext.Current.CancellationToken);
Volatile.Read(ref changes).ShouldBe(afterDispose);
// Idempotent: an `await using` after an explicit ShutdownAsync/DisposeAsync must not throw.
await driver.ShutdownAsync(CancellationToken.None);
await Should.NotThrowAsync(async () => await driver.DisposeAsync());
}
// ---- host identity ----
[Fact]
public async Task GetHostStatuses_namesTheServerAndDatabase_butNeverTheCredentials()
{
using var fixture = new SqlitePollFixture();
const string connectionString =
"Server=sqlsrv01;Initial Catalog=MesStaging;User ID=svc_ot;Password=" + SecretToken + ";";
await using var driver = NewDriver(fixture, connectionString, CapturingLogger.Null);
var hosts = driver.GetHostStatuses();
hosts.Count.ShouldBe(1);
hosts[0].HostName.ShouldContain("sqlsrv01");
hosts[0].HostName.ShouldContain("MesStaging");
hosts[0].HostName.ShouldNotContain(SecretToken);
hosts[0].HostName.ShouldNotContain("svc_ot");
hosts[0].State.ShouldBe(HostState.Unknown); // nothing has been attempted yet
}
[Fact]
public async Task OnHostStatusChanged_firesOnceOnTheRunningTransition()
{
using var fixture = new SqlitePollFixture();
await using var driver = NewDriver(fixture, KvEntry("Speed", SqlitePollFixture.PresentKey));
var transitions = new List<HostStatusChangedEventArgs>();
driver.OnHostStatusChanged += (_, args) => { lock (transitions) { transitions.Add(args); } };
await driver.InitializeAsync(ConfigJson, CancellationToken.None);
// A second successful poll must not re-raise — the event is for transitions, not for ticks.
await driver.ReadAsync(["Speed"], CancellationToken.None);
lock (transitions)
{
transitions.Count.ShouldBe(1);
transitions[0].OldState.ShouldBe(HostState.Unknown);
transitions[0].NewState.ShouldBe(HostState.Running);
}
}
// ---- helpers ----
/// <summary>A distinctive password token: any assertion that finds it has found a credential leak.</summary>
private const string SecretToken = "hunter2-do-not-log";
/// <summary>The unreachable connection string's database file name, used as the actionable-error probe.</summary>
private const string NoSuchDatabaseName = "otopcua-sql-no-such-dir";
/// <summary>
/// The driver's <c>DriverConfig</c> JSON. The shell serves its typed options (the factory, Task 9,
/// owns parsing) exactly as <c>ModbusDriver</c> does, so this is deliberately inert.
/// </summary>
private const string ConfigJson = """{"provider":"SqlServer"}""";
private static SqlDriver NewDriver(SqlitePollFixture fixture, params RawTagEntry[] rawTags)
=> NewDriver(fixture, CapturingLogger.Null, rawTags);
private static SqlDriver NewDriver(
SqlitePollFixture fixture, CapturingLogger logger, params RawTagEntry[] rawTags)
=> NewDriver(fixture, fixture.ConnectionString, logger, rawTags);
private static SqlDriver NewDriver(
SqlitePollFixture fixture, string connectionString, CapturingLogger logger, params RawTagEntry[] rawTags)
=> NewDriver(
fixture, connectionString, logger,
operationTimeout: TimeSpan.FromSeconds(15), commandTimeout: TimeSpan.FromSeconds(10),
rawTags: rawTags);
/// <summary>
/// Builds the driver through its production constructor. That constructor <b>is</b> the injection
/// seam — dialect, provider factory and already-resolved connection string are all parameters — so no
/// test-only factory is needed on the product type (Task 9 resolves the same three from config).
/// </summary>
private static SqlDriver NewDriver(
SqlitePollFixture fixture,
string connectionString,
CapturingLogger logger,
TimeSpan operationTimeout,
TimeSpan commandTimeout,
IReadOnlyList<RawTagEntry> rawTags)
=> new(
new SqlDriverOptions
{
RawTags = rawTags,
OperationTimeout = operationTimeout,
CommandTimeout = commandTimeout,
},
DriverInstanceId,
new SqliteDialect(),
connectionString,
factory: fixture.Factory,
logger: logger);
/// <summary>A connection string whose database cannot be opened — the directory does not exist.</summary>
private static string Unreachable() => new SqliteConnectionStringBuilder
{
DataSource = Path.Combine(
Path.GetTempPath(), $"{NoSuchDatabaseName}-{Guid.NewGuid():N}", "db.sqlite"),
Password = SecretToken,
}.ToString();
/// <summary>One authored raw tag: a key-value <c>TagConfig</c> blob over the fixture's EAV table.</summary>
private static RawTagEntry KvEntry(string rawPath, string keyValue)
=> new(rawPath, string.Create(CultureInfo.InvariantCulture, $$"""
{
"driver": "Sql",
"model": "KeyValue",
"table": "{{SqlitePollFixture.KeyValueTable}}",
"keyColumn": "{{SqlitePollFixture.KeyColumn}}",
"keyValue": "{{keyValue}}",
"valueColumn": "{{SqlitePollFixture.ValueColumn}}",
"timestampColumn": "{{SqlitePollFixture.TimestampColumn}}"
}
"""), WriteIdempotent: false);
/// <summary>The same tag as <see cref="KvEntry"/>, already typed — for the reference reader.</summary>
private static SqlTagDefinition KvDefinition(string rawPath, string keyValue)
=> new(rawPath, SqlTagModel.KeyValue, SqlitePollFixture.KeyValueTable,
KeyColumn: SqlitePollFixture.KeyColumn, KeyValue: keyValue,
ValueColumn: SqlitePollFixture.ValueColumn,
TimestampColumn: SqlitePollFixture.TimestampColumn);
/// <summary>Records everything the driver streams into the address space.</summary>
private sealed class CapturingBuilder : IAddressSpaceBuilder
{
/// <summary>The folders created, in order.</summary>
public List<(string BrowseName, string DisplayName)> Folders { get; } = [];
/// <summary>The variables registered, in order.</summary>
public List<(string BrowseName, DriverAttributeInfo Info)> Variables { get; } = [];
public IAddressSpaceBuilder Folder(string browseName, string displayName)
{
Folders.Add((browseName, displayName));
return this;
}
public IVariableHandle Variable(string browseName, string displayName, DriverAttributeInfo attributeInfo)
{
Variables.Add((browseName, attributeInfo));
return new Handle(attributeInfo.FullName);
}
public void AddProperty(string browseName, DriverDataType dataType, object? value) { }
private sealed class Handle(string fullReference) : IVariableHandle
{
public string FullReference => fullReference;
public IAlarmConditionSink MarkAsAlarmCondition(AlarmConditionInfo info) => new Sink();
private sealed class Sink : IAlarmConditionSink
{
public void OnTransition(AlarmEventArgs args) { }
}
}
}
/// <summary>Captures the driver's log so "skipped and logged" can be asserted rather than assumed.</summary>
private sealed class CapturingLogger : ILogger<SqlDriver>
{
/// <summary>A logger that records nothing — for the tests that do not assert on logging.</summary>
public static CapturingLogger Null { get; } = new();
/// <summary>Every record written, level + rendered message.</summary>
public List<(LogLevel Level, string Message)> Entries { get; } = [];
public IDisposable BeginScope<TState>(TState state) where TState : notnull => NullScope.Instance;
public bool IsEnabled(LogLevel logLevel) => true;
public void Log<TState>(
LogLevel logLevel, EventId eventId, TState state, Exception? exception,
Func<TState, Exception?, string> formatter)
{
lock (Entries) Entries.Add((logLevel, formatter(state, exception)));
}
private sealed class NullScope : IDisposable
{
public static NullScope Instance { get; } = new();
public void Dispose() { }
}
}
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests.csproj
+7
View File
@@ -1,5 +1,12 @@
<Project Sdk="Microsoft.NET.Sdk"> <Project Sdk="Microsoft.NET.Sdk">
<!-- OTOPCUA0001 (arch-review 07/C-1): this suite invokes driver capability methods DIRECTLY to
exercise driver-level behavior — the analyzer's documented intentional case ("move the
suppression into a NoWarn for the test project"). Not a resilience-dispatch gap. -->
<PropertyGroup>
<NoWarn>$(NoWarn);OTOPCUA0001</NoWarn>
</PropertyGroup>
<PropertyGroup> <PropertyGroup>
<TargetFramework>net10.0</TargetFramework> <TargetFramework>net10.0</TargetFramework>
<Nullable>enable</Nullable> <Nullable>enable</Nullable>