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

feat(sql): SqlPollReader — grouped read, bounded deadline, ordered slice-back

Browse Source 
Executes a poll pass: resolve refs -> SqlGroupPlanner.Plan -> one command per
group -> slice the result set back to per-tag DataValueSnapshots, positionally
aligned with the caller's reference list (design §3.2).

The frozen-peer contract (design §8.3, the R2-01 S7 lesson) is the core of it.
CommandTimeout is only the server-side backstop; the client-side bound is a
linked CancelAfter PLUS Task.WaitAsync, because a linked token bounds only a
provider that honours it and ADO.NET providers vary. The whole per-group
operation — waiting for a concurrency slot, OpenAsync, and the query — runs off
one operationTimeout budget, so ReadAsync returns on time regardless of what the
database is doing. A breach Bad-codes that group (BadTimeout); caller
cancellation propagates instead, since nobody consumes a torn-down poll.

Also: absent row (BadNoData) stays distinct from a NULL cell (Uncertain, or Bad
under nullIsBad); duplicate plan members are all fed (two tags on one keyValue
bind one parameter but stay two members); the concurrency slot is released by the
work, not the waiter, so a frozen database can never hold more than
maxConcurrentGroups connections; the whole call throws only when the connection
itself cannot open, which is what earns PollGroupEngine's backoff.

SqlStatusCodes is driver-local, matching every sibling driver's own table — there
is no shared helper in Core.Abstractions and drivers do not reference the OPC UA
SDK. Values were read off Opc.Ua.StatusCodes rather than copied from a sibling,
because two of those tables carry transcription errors.

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
2026-07-24 14:26:29 -04:00
parent 50f88b938f
commit 967e5140f1
2 changed files with 1250 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs
+770
View File
@@ -0,0 +1,770 @@
using System.Data.Common;
using System.Diagnostics;
using System.Globalization;
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>
/// Executes one poll pass: resolves each requested reference to a <see cref="SqlTagDefinition"/>, folds
/// the definitions into the minimum number of queries via <see cref="SqlGroupPlanner"/>, runs one
/// command per group under a bounded deadline, and slices each result set back to per-tag
/// <see cref="DataValueSnapshot"/>s — <b>positionally aligned with the caller's reference list</b>
/// (design §3.2). Structurally the SQL analogue of <c>ModbusDriver.ReadCoalescedAsync</c>: group → one
/// round-trip → slice back.
/// <para><b>N in, N out, in order.</b> The returned list always has exactly one entry per requested
/// reference, at the same index, whatever happened. Per-tag failures are Bad-coded snapshots, never
/// exceptions (the <see cref="IReadable"/> contract) — see <see cref="SqlStatusCodes"/> for which code
/// means what. This matters more here than in a point-to-point driver: one result set feeds many tags,
/// so a slicing defect does not fail loudly — it publishes <em>one tag's value onto another tag's
/// node</em>.</para>
/// <para><b>The whole call throws only when the database is unreachable</b> — that is, when opening a
/// connection fails. That surfaces to <see cref="PollGroupEngine"/> as a poll failure and earns the
/// capped-exponential backoff; the driver's <c>IHostConnectivityProbe</c> is what scopes Bad quality
/// across the subtree during an outage, so nothing is lost by not manufacturing per-tag snapshots for
/// an outage. A query that fails <em>after</em> the connection opened is the group's problem, not the
/// server's, and Bad-codes that group only.</para>
/// <para><b>Deadlines (design §8.3, the R2-01 frozen-peer lesson).</b> Two independent mechanisms, both
/// required. <c>CommandTimeout</c> is the <em>server-side</em> backstop and bounds nothing on a wedged
/// socket. The wall-clock bound is client-side and is what actually guarantees this method returns: see
/// <see cref="RunGroupAsync"/>. A frozen database yields <see cref="SqlStatusCodes.BadTimeout"/>
/// snapshots — it must never wedge the poll thread.</para>
/// <para><b>Not a driver.</b> This type owns the read path only; it deliberately holds no health state,
/// no subscription state and no connection between polls, so the driver shell can wrap it in
/// <see cref="PollGroupEngine"/> and own those concerns.</para>
/// </summary>
public sealed class SqlPollReader
{
/// <summary>Minimum spacing between "your source violates the query contract" warnings, per reader.</summary>
private static readonly TimeSpan ContractWarningInterval = TimeSpan.FromMinutes(1);
private readonly DbProviderFactory _factory;
private readonly string _connectionString;
private readonly ISqlDialect _dialect;
private readonly TimeSpan _operationTimeout;
private readonly int _commandTimeoutSeconds;
private readonly bool _nullIsBad;
private readonly Func<string, SqlTagDefinition?> _resolve;
private readonly ILogger _logger;
/// <summary>
/// Caps concurrently executing group queries — and therefore concurrently held connections
/// (design §8.4). Never disposed: a <see cref="SemaphoreSlim"/> only needs disposal once its
/// <c>AvailableWaitHandle</c> has been touched, and not disposing it also removes the hazard of a
/// timed-out-but-still-running group releasing into a disposed semaphore.
/// </summary>
private readonly SemaphoreSlim _gate;
private long _lastContractWarningTicks;
/// <summary>Initializes a new instance of the <see cref="SqlPollReader"/> class.</summary>
/// <param name="factory">
/// Creates the provider's connections. Passed explicitly rather than taken from
/// <paramref name="dialect"/> so the driver can hand in an instrumented or pre-configured factory.
/// </param>
/// <param name="connectionString">
/// The resolved connection string (never the authored <c>connectionStringRef</c> — secrets are
/// resolved by the driver at Initialize, design §8.2).
/// </param>
/// <param name="dialect">Supplies identifier quoting, row-limit syntax, and column-type mapping.</param>
/// <param name="commandTimeout">
/// The ADO.NET <c>CommandTimeout</c> backstop. Rounded <b>up</b> to whole seconds and floored at 1,
/// because ADO.NET reads <c>CommandTimeout = 0</c> as "wait forever" — the one value this option
/// must never silently become.
/// </param>
/// <param name="operationTimeout">
/// The wall-clock ceiling on one group's whole operation — waiting for a concurrency slot, opening
/// the connection, and running the query. A breach Bad-codes that group with
/// <see cref="SqlStatusCodes.BadTimeout"/>.
/// <para>Authoring should keep this strictly greater than <paramref name="commandTimeout"/>
/// (design §8.3): inverted, the client-side abort always fires first and masks the server-side
/// backstop. That rule is <b>not</b> enforced here — it belongs to config validation, and this type
/// stays usable for the tests that must deliberately invert it.</para>
/// </param>
/// <param name="maxConcurrentGroups">Maximum group queries — and connections — in flight at once.</param>
/// <param name="nullIsBad">
/// <see langword="true"/> publishes <see cref="SqlStatusCodes.Bad"/> for a NULL value cell instead
/// of the default <see cref="SqlStatusCodes.Uncertain"/>. It governs a <em>present</em> row with a
/// NULL cell only — an absent row is always <see cref="SqlStatusCodes.BadNoData"/>.
/// </param>
/// <param name="resolve">
/// RawPath → tag definition; <see langword="null"/> on a miss. Shaped to match
/// <see cref="EquipmentTagRefResolver{TDef}"/>'s lookup so the driver can pass its authored table
/// straight through.
/// </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="connectionString"/> is blank.</exception>
/// <exception cref="ArgumentOutOfRangeException">A timeout is non-positive, or the cap is below 1.</exception>
public SqlPollReader(
DbProviderFactory factory,
string connectionString,
ISqlDialect dialect,
TimeSpan commandTimeout,
TimeSpan operationTimeout,
int maxConcurrentGroups,
bool nullIsBad,
Func<string, SqlTagDefinition?> resolve,
ILogger? logger = null)
{
ArgumentNullException.ThrowIfNull(factory);
ArgumentNullException.ThrowIfNull(dialect);
ArgumentNullException.ThrowIfNull(resolve);
if (string.IsNullOrWhiteSpace(connectionString))
throw new ArgumentException("A Sql poll reader needs a connection string.", nameof(connectionString));
ArgumentOutOfRangeException.ThrowIfLessThanOrEqual(commandTimeout, TimeSpan.Zero);
ArgumentOutOfRangeException.ThrowIfLessThanOrEqual(operationTimeout, TimeSpan.Zero);
ArgumentOutOfRangeException.ThrowIfLessThan(maxConcurrentGroups, 1);
_factory = factory;
_connectionString = connectionString;
_dialect = dialect;
_operationTimeout = operationTimeout;
_commandTimeoutSeconds = Math.Max(1, (int)Math.Ceiling(commandTimeout.TotalSeconds));
_nullIsBad = nullIsBad;
_resolve = resolve;
_logger = logger ?? NullLogger.Instance;
_gate = new SemaphoreSlim(maxConcurrentGroups, maxConcurrentGroups);
}
/// <summary>
/// Reads every reference in one poll pass. Matches <see cref="IReadable.ReadAsync"/>'s shape so the
/// driver shell can delegate to it directly.
/// </summary>
/// <param name="fullReferences">The RawPaths to read. Empty yields an empty result and touches no connection.</param>
/// <param name="cancellationToken">
/// The caller's token. Its cancellation <b>propagates</b> as an
/// <see cref="OperationCanceledException"/> — the engine asked the poll to stop, and nobody is
/// waiting for the snapshots. This is deliberately asymmetric with an <c>operationTimeout</c>
/// breach, which is a data-quality fact clients must see and so becomes
/// <see cref="SqlStatusCodes.BadTimeout"/> snapshots.
/// </param>
/// <returns>One snapshot per reference, at the same index.</returns>
/// <exception cref="ArgumentNullException"><paramref name="fullReferences"/> is null.</exception>
/// <exception cref="OperationCanceledException"><paramref name="cancellationToken"/> was cancelled.</exception>
/// <exception cref="DbException">The database could not be reached (opening a connection failed).</exception>
public async Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(
IReadOnlyList<string> fullReferences, CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(fullReferences);
var results = new DataValueSnapshot[fullReferences.Count];
if (results.Length == 0) return results;
var readAt = DateTime.UtcNow;
// Resolve first. A miss is this tag's own problem (BadNodeIdUnknown) and must not reach the planner,
// whose members are the sole thing the slice-back walks.
var definitions = new List<SqlTagDefinition>(fullReferences.Count);
var slots = new Dictionary<SqlTagDefinition, List<int>>();
for (var i = 0; i < fullReferences.Count; i++)
{
var definition = fullReferences[i] is { } reference ? _resolve(reference) : null;
if (definition is null)
{
results[i] = new DataValueSnapshot(null, SqlStatusCodes.BadNodeIdUnknown, null, readAt);
continue;
}
definitions.Add(definition);
if (!slots.TryGetValue(definition, out var positions))
slots[definition] = positions = [];
positions.Add(i);
}
if (definitions.Count > 0)
await ReadGroupsAsync(definitions, slots, results, readAt, cancellationToken).ConfigureAwait(false);
// Fail-safe. Nothing above should leave a hole, but "N in, N out" is a contract the caller indexes
// blind — a null here would be a NullReferenceException in the publish fan-out, far from its cause.
for (var i = 0; i < results.Length; i++)
results[i] ??= new DataValueSnapshot(null, SqlStatusCodes.BadInternalError, null, readAt);
return results;
}
/// <summary>Plans the resolved definitions, runs every group concurrently, and slices the results back.</summary>
private async Task ReadGroupsAsync(
List<SqlTagDefinition> definitions,
Dictionary<SqlTagDefinition, List<int>> slots,
DataValueSnapshot[] results,
DateTime readAt,
CancellationToken cancellationToken)
{
IReadOnlyList<SqlQueryPlan> plans;
try
{
plans = SqlGroupPlanner.Plan(definitions, _dialect);
}
catch (Exception ex) when (ex is ArgumentException or NotSupportedException)
{
// A definition with a broken invariant — SqlEquipmentTagParser rejects all of these up front, so
// reaching here means a definition was built past the parser. It is an authoring fault, not a
// database fault: Bad-code the planned tags rather than throwing and triggering connection backoff.
_logger.LogError(ex, "Sql poll planning failed for {Count} tag(s); publishing BadConfigurationError.",
definitions.Count);
foreach (var definition in definitions)
{
if (!slots.TryGetValue(definition, out var positions)) continue;
var snapshot = new DataValueSnapshot(null, SqlStatusCodes.BadConfigurationError, null, readAt);
foreach (var position in positions) results[position] = snapshot;
}
return;
}
// Groups run concurrently; _gate — not the task count — is what bounds open connections. Each group
// carries its own deadline, so the whole call is bounded by ~operationTimeout rather than by
// plans.Count × operationTimeout.
var groups = new Task<GroupResult>[plans.Count];
for (var g = 0; g < plans.Count; g++)
groups[g] = RunGroupAsync(plans[g], cancellationToken);
// WhenAll waits for every group before surfacing a fault, so a thrown "database unreachable" never
// leaves a sibling group running against a connection nobody is watching.
var outcomes = await Task.WhenAll(groups).ConfigureAwait(false);
for (var g = 0; g < plans.Count; g++)
Slice(plans[g], outcomes[g], slots, results, readAt);
}
/// <summary>
/// Runs one group's query under a hard wall-clock ceiling.
/// <para><b>Why the ceiling is <c>Task.WaitAsync</c> and not just a linked
/// <c>CancelAfter</c>.</b> A linked token only bounds a
/// provider that <em>honours</em> it. The S7 R2-01 finding was exactly a provider whose async path
/// ignored its deadline, and ADO.NET has the same shape: some providers implement
/// <c>ExecuteReaderAsync</c> synchronously, and even a genuinely async one can hang inside its own
/// cancellation handshake against a frozen socket. Both are used here: the linked token so a
/// well-behaved provider unwinds cleanly and releases its connection, and <c>WaitAsync</c> so
/// control returns at the deadline regardless.</para>
/// <para>The work is started on the thread pool so a provider that blocks synchronously blocks a
/// pool thread rather than the caller — without that, a synchronous provider never yields the task
/// there would be to bound.</para>
/// <para><b>The concurrency slot is released by the work, not by the waiter.</b> A timed-out group
/// keeps its slot until it truly finishes, so a frozen database can never accumulate more than
/// <c>maxConcurrentGroups</c> connections however long it stays frozen. The slot wait is itself
/// inside the deadline, so a poll behind a wedged group still returns on time — as BadTimeout.</para>
/// </summary>
private async Task<GroupResult> RunGroupAsync(SqlQueryPlan plan, CancellationToken cancellationToken)
{
var clock = Stopwatch.StartNew();
if (!await _gate.WaitAsync(Remaining(clock), cancellationToken).ConfigureAwait(false))
return GroupResult.TimedOut;
var handedOff = false;
var deadline = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
Task<GroupResult> work;
try
{
var budget = Remaining(clock);
if (budget <= TimeSpan.Zero) return GroupResult.TimedOut;
deadline.CancelAfter(budget);
work = Task.Run(
async () =>
{
try
{
return await QueryAsync(plan, deadline.Token).ConfigureAwait(false);
}
finally
{
_gate.Release();
deadline.Dispose();
}
},
CancellationToken.None);
handedOff = true;
}
finally
{
// Ownership of both the slot and the CTS transfers to the work task; release them here only on
// the paths where that task was never created.
if (!handedOff)
{
_gate.Release();
deadline.Dispose();
}
}
try
{
return await work.WaitAsync(Remaining(clock), cancellationToken).ConfigureAwait(false);
}
catch (TimeoutException)
{
// Our wall-clock bound fired while the provider was still inside the call.
return TimedOut(plan, work);
}
catch (OperationCanceledException)
{
// Either the caller pulled the plug, or our deadline fired and the provider DID honour the
// linked token. Both leave the work running with nobody to observe its fault.
if (!cancellationToken.IsCancellationRequested) return TimedOut(plan, work);
Detach(work);
throw;
}
}
/// <summary>How much of this group's <c>operationTimeout</c> budget is left; never negative.</summary>
private TimeSpan Remaining(Stopwatch clock)
{
var remaining = _operationTimeout - clock.Elapsed;
return remaining > TimeSpan.Zero ? remaining : TimeSpan.Zero;
}
/// <summary>Records the deadline breach and detaches the abandoned work.</summary>
private GroupResult TimedOut(SqlQueryPlan plan, Task work)
{
_logger.LogWarning(
"Sql poll group ({Model}, {Members} tag(s)) exceeded the {Timeout} ms operation timeout; " +
"publishing BadTimeout.", plan.Model, plan.Members.Count, (int)_operationTimeout.TotalMilliseconds);
Detach(work);
return GroupResult.TimedOut;
}
/// <summary>
/// Observes an abandoned group's eventual fault so it never surfaces as an unobserved task
/// exception. The task still owns its connection and still releases the concurrency slot when it
/// finally completes — that is what keeps a frozen database from accumulating connections.
/// </summary>
private static void Detach(Task work)
=> _ = work.ContinueWith(
static t => _ = t.Exception,
CancellationToken.None,
TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously,
TaskScheduler.Default);
/// <summary>
/// Opens a connection, executes the plan's single command, and materialises the result set.
/// <para>The rows are read into memory <b>before</b> the reader is disposed — the slice-back needs
/// random access across members, and holding a reader (and therefore a connection) open across that
/// work is exactly how a poll loop exhausts a pool.</para>
/// </summary>
private async Task<GroupResult> QueryAsync(SqlQueryPlan plan, 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;
// Deliberately OUTSIDE the catch below: a failed open means the database is unreachable, which
// is the one condition IReadable says the whole call may throw on (→ PollGroupEngine backoff).
await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
try
{
var command = connection.CreateCommand();
await using (command.ConfigureAwait(false))
{
command.CommandText = plan.SqlText;
command.CommandTimeout = _commandTimeoutSeconds;
for (var i = 0; i < plan.ParameterNames.Count; i++)
{
var parameter = command.CreateParameter();
// Bound by the plan's own index pairing — never by re-deriving the marker convention.
parameter.ParameterName = plan.ParameterNames[i];
parameter.Value = plan.Parameters[i] ?? DBNull.Value;
command.Parameters.Add(parameter);
}
var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
await using (reader.ConfigureAwait(false))
{
return await MaterialiseAsync(plan, reader, cancellationToken).ConfigureAwait(false);
}
}
}
catch (OperationCanceledException)
{
throw;
}
catch (Exception ex)
{
// The connection opened, so the server is up — this query is what failed (an identifier that
// is not in the catalog, a lock, a permission). That is this group's problem alone.
_logger.LogWarning(
ex, "Sql poll group ({Model}, {Members} tag(s)) failed: {Sql}",
plan.Model, plan.Members.Count, plan.SqlText);
return GroupResult.Failed;
}
}
}
/// <summary>
/// Reads every row into memory, projected onto <see cref="SqlQueryPlan.SelectedColumns"/>. Row
/// fetching stays on the async path so a result set that streams slowly still honours the deadline
/// token mid-read, rather than only at the ExecuteReader boundary.
/// </summary>
private async Task<GroupResult> MaterialiseAsync(
SqlQueryPlan plan, DbDataReader reader, CancellationToken cancellationToken)
{
var selected = plan.SelectedColumns;
var ordinals = new int[selected.Count];
var typeNames = new string[selected.Count];
var columnIndex = new Dictionary<string, int>(selected.Count, StringComparer.Ordinal);
for (var c = 0; c < selected.Count; c++)
{
// GetOrdinal rather than the position in SelectedColumns: the list is documented to be in
// ordinal order, but resolving it against the live result set is what makes a wrong slice
// impossible rather than merely unlikely.
ordinals[c] = reader.GetOrdinal(selected[c]);
typeNames[c] = SafeTypeName(reader, ordinals[c]);
columnIndex[selected[c]] = c;
}
var rows = new List<object?[]>();
while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
{
var row = new object?[selected.Count];
for (var c = 0; c < selected.Count; c++)
row[c] = reader.IsDBNull(ordinals[c]) ? null : reader.GetValue(ordinals[c]);
rows.Add(row);
}
return GroupResult.Ok(rows, columnIndex, typeNames);
}
/// <summary>Reads a column's provider type name, falling back to the dialect's unknown-type default.</summary>
private static string SafeTypeName(DbDataReader reader, int ordinal)
{
try
{
return reader.GetDataTypeName(ordinal) ?? string.Empty;
}
catch (Exception ex) when (ex is NotSupportedException or InvalidOperationException
or IndexOutOfRangeException)
{
// A provider that declines to report a type name must not fail the poll — the tag's declared
// type (or the dialect's String fallback) covers it.
return string.Empty;
}
}
/// <summary>Maps one group's outcome onto every slot its members occupy in the caller's list.</summary>
private void Slice(
SqlQueryPlan plan,
GroupResult outcome,
Dictionary<SqlTagDefinition, List<int>> slots,
DataValueSnapshot[] results,
DateTime readAt)
{
// KeyValue indexes rows by the key column; WideRow's members all read the one selected row, so the
// where-column is not even in the result set (design §5.3).
var byKey = outcome.Succeeded && plan.Model == SqlTagModel.KeyValue
? IndexByKey(plan, outcome)
: null;
foreach (var member in plan.Members)
{
// Members may repeat (two tags on one keyValue, or the same ref twice) — every occurrence is fed.
if (!slots.TryGetValue(member, out var positions)) continue;
var snapshot = outcome.Succeeded
? MapMember(plan, member, outcome, byKey, readAt)
: new DataValueSnapshot(null, outcome.FailureStatus, null, readAt);
foreach (var position in positions) results[position] = snapshot;
}
}
/// <summary>
/// Builds the key → row index a key-value plan slices through. Later rows win, so a source that
/// breaks the one-row-per-key contract (design §3.6) still yields a deterministic value; the
/// violation is reported through a rate-limited warning rather than silently.
/// </summary>
private Dictionary<string, object?[]> IndexByKey(SqlQueryPlan plan, GroupResult outcome)
{
var index = new Dictionary<string, object?[]>(StringComparer.Ordinal);
if (plan.KeyColumn is null || !outcome.ColumnIndex.TryGetValue(plan.KeyColumn, out var keyAt))
return index;
var duplicated = 0;
foreach (var row in outcome.Rows)
{
// A NULL key cell indexes nothing: Convert.ToString would fold it to "", which would then be
// matched by a tag whose keyValue is legitimately the empty string.
if (row[keyAt] is not { } keyCell) continue;
// Stringified because the bound keyValue is authored text while the column may be any type
// (an INTEGER station id, say) — the provider coerces on the way in, so match on the way out.
var key = Convert.ToString(keyCell, CultureInfo.InvariantCulture);
if (key is null) continue;
if (!index.TryAdd(key, row))
{
index[key] = row;
duplicated++;
}
}
if (duplicated > 0 && ShouldWarnAboutContract())
{
_logger.LogWarning(
"Sql poll source '{Table}' returned more than one row for {Count} key(s) in one poll; the " +
"last row wins. The key-value model requires one row per key — point the tag at a " +
"current-values table or an operator-authored latest-per-key view.",
plan.Members[0].Table, duplicated);
}
return index;
}
/// <summary>Maps one member's cell to a snapshot: value, quality, and source timestamp.</summary>
private DataValueSnapshot MapMember(
SqlQueryPlan plan,
SqlTagDefinition member,
GroupResult outcome,
Dictionary<string, object?[]>? byKey,
DateTime readAt)
{
object?[]? row;
string? valueColumn;
string? timestampColumn;
if (plan.Model == SqlTagModel.KeyValue)
{
valueColumn = plan.ValueColumn;
timestampColumn = plan.TimestampColumn;
row = member.KeyValue is { } key && byKey is not null ? byKey.GetValueOrDefault(key) : null;
}
else
{
valueColumn = member.ColumnName;
// A wide-row plan's members may legitimately carry DIFFERENT timestamp columns (all of them are
// selected), which is why the plan-level TimestampColumn is null for this model by design.
timestampColumn = member.TimestampColumn;
row = outcome.Rows.Count > 0 ? outcome.Rows[^1] : null;
}
// No row: the key was deleted, or the row selector matched nothing. Distinct from a NULL cell.
if (row is null) return new DataValueSnapshot(null, SqlStatusCodes.BadNoData, null, readAt);
// A timestamp that will not parse falls back to the poll clock rather than poisoning a good value.
var sourceTimestamp = ReadTimestamp(outcome, row, timestampColumn) ?? readAt;
if (valueColumn is null || !outcome.ColumnIndex.TryGetValue(valueColumn, out var valueAt))
{
// The planner selects every member's value column, so this is unreachable short of plan/result
// drift — loud rather than silently publishing someone else's cell.
_logger.LogError(
"Sql tag '{Tag}' reads column '{Column}', which the executed plan did not select.",
member.Name, valueColumn);
return new DataValueSnapshot(null, SqlStatusCodes.BadInternalError, sourceTimestamp, readAt);
}
var cell = row[valueAt];
if (cell is null)
{
return new DataValueSnapshot(
null, _nullIsBad ? SqlStatusCodes.Bad : SqlStatusCodes.Uncertain, sourceTimestamp, readAt);
}
var target = member.DeclaredType ?? _dialect.MapColumnType(outcome.TypeNames[valueAt]);
if (!TryCoerce(cell, target, out var value))
{
_logger.LogWarning(
"Sql tag '{Tag}' could not coerce a {Actual} cell to its {Declared} type.",
member.Name, cell.GetType().Name, target);
return new DataValueSnapshot(null, SqlStatusCodes.BadTypeMismatch, sourceTimestamp, readAt);
}
return new DataValueSnapshot(value, SqlStatusCodes.Good, sourceTimestamp, readAt);
}
/// <summary>Reads a row's source timestamp, or null when there is no column, no value, or no parse.</summary>
private static DateTime? ReadTimestamp(GroupResult outcome, object?[] row, string? timestampColumn)
{
if (string.IsNullOrWhiteSpace(timestampColumn)) return null;
if (!outcome.ColumnIndex.TryGetValue(timestampColumn, out var at)) return null;
if (row[at] is not { } cell) return null;
return TryCoerce(cell, DriverDataType.DateTime, out var value) ? (DateTime?)value : null;
}
/// <summary>
/// Coerces a provider cell to the tag's effective <see cref="DriverDataType"/>, so the published CLR
/// type matches the type the address space declared for the node. Never throws.
/// </summary>
private static bool TryCoerce(object cell, DriverDataType target, out object? value)
{
try
{
value = target switch
{
DriverDataType.Boolean => Convert.ToBoolean(cell, CultureInfo.InvariantCulture),
DriverDataType.Int16 => Convert.ToInt16(cell, CultureInfo.InvariantCulture),
DriverDataType.Int32 => Convert.ToInt32(cell, CultureInfo.InvariantCulture),
DriverDataType.Int64 => Convert.ToInt64(cell, CultureInfo.InvariantCulture),
DriverDataType.UInt16 => Convert.ToUInt16(cell, CultureInfo.InvariantCulture),
DriverDataType.UInt32 => Convert.ToUInt32(cell, CultureInfo.InvariantCulture),
DriverDataType.UInt64 => Convert.ToUInt64(cell, CultureInfo.InvariantCulture),
DriverDataType.Float32 => Convert.ToSingle(cell, CultureInfo.InvariantCulture),
// decimal/numeric collapse here, with the documented precision caveat (design §8.6).
DriverDataType.Float64 => Convert.ToDouble(cell, CultureInfo.InvariantCulture),
DriverDataType.DateTime => ToUtc(cell),
_ => Convert.ToString(cell, CultureInfo.InvariantCulture) ?? string.Empty,
};
return true;
}
catch (Exception ex) when (ex is InvalidCastException or FormatException
or OverflowException or ArgumentException)
{
value = null;
return false;
}
}
/// <summary>
/// Normalises a timestamp cell to UTC. A naive <c>datetime</c> (no offset — the common SQL Server
/// shape) is <b>assumed</b> to already be UTC rather than reinterpreted through the server's local
/// zone, which would silently shift every source timestamp by the host's offset.
/// </summary>
private static DateTime ToUtc(object cell) => cell switch
{
DateTime { Kind: DateTimeKind.Utc } utc => utc,
DateTime { Kind: DateTimeKind.Local } local => local.ToUniversalTime(),
DateTime unspecified => DateTime.SpecifyKind(unspecified, DateTimeKind.Utc),
DateTimeOffset offset => offset.UtcDateTime,
string text => DateTime.Parse(
text, CultureInfo.InvariantCulture,
DateTimeStyles.AssumeUniversal | DateTimeStyles.AdjustToUniversal),
_ => DateTime.SpecifyKind(
Convert.ToDateTime(cell, CultureInfo.InvariantCulture), DateTimeKind.Utc),
};
/// <summary>Rate-limits the source-contract warning so a misconfigured table cannot flood the log.</summary>
private bool ShouldWarnAboutContract()
{
var now = Stopwatch.GetTimestamp();
var last = Interlocked.Read(ref _lastContractWarningTicks);
if (last != 0 && Stopwatch.GetElapsedTime(last, now) < ContractWarningInterval) return false;
return Interlocked.CompareExchange(ref _lastContractWarningTicks, now, last) == last;
}
/// <summary>One group's outcome: either a materialised result set, or the status its members inherit.</summary>
private sealed class GroupResult
{
private static readonly IReadOnlyList<object?[]> NoRows = [];
private static readonly IReadOnlyDictionary<string, int> NoColumns =
new Dictionary<string, int>(StringComparer.Ordinal);
private GroupResult(
uint failureStatus,
IReadOnlyList<object?[]> rows,
IReadOnlyDictionary<string, int> columnIndex,
IReadOnlyList<string> typeNames)
{
FailureStatus = failureStatus;
Rows = rows;
ColumnIndex = columnIndex;
TypeNames = typeNames;
}
/// <summary>The group exceeded its wall-clock deadline (design §8.3).</summary>
public static GroupResult TimedOut { get; } =
new(SqlStatusCodes.BadTimeout, NoRows, NoColumns, []);
/// <summary>The connection was fine but the query was not.</summary>
public static GroupResult Failed { get; } =
new(SqlStatusCodes.BadCommunicationError, NoRows, NoColumns, []);
/// <summary>The status every member inherits when <see cref="Succeeded"/> is false.</summary>
public uint FailureStatus { get; }
/// <summary>The materialised rows, projected onto the plan's selected columns.</summary>
public IReadOnlyList<object?[]> Rows { get; }
/// <summary>Selected column name → its position within each row array.</summary>
public IReadOnlyDictionary<string, int> ColumnIndex { get; }
/// <summary>Each selected column's provider type name, for dialect type inference.</summary>
public IReadOnlyList<string> TypeNames { get; }
/// <summary>True when the query ran; a zero-row result set is still a success.</summary>
public bool Succeeded => FailureStatus == SqlStatusCodes.Good;
/// <summary>A completed result set.</summary>
/// <param name="rows">The materialised rows.</param>
/// <param name="columnIndex">Selected column name → row-array position.</param>
/// <param name="typeNames">Each selected column's provider type name.</param>
/// <returns>A successful outcome.</returns>
public static GroupResult Ok(
IReadOnlyList<object?[]> rows,
IReadOnlyDictionary<string, int> columnIndex,
IReadOnlyList<string> typeNames)
=> new(SqlStatusCodes.Good, rows, columnIndex, typeNames);
}
}
/// <summary>
/// The OPC UA status codes the Sql driver publishes, and the quality-class predicates its tests read
/// them back with.
/// <para><b>Why a driver-local table rather than a shared one.</b> Every driver in this tree carries its
/// own (<c>TwinCATStatusMapper</c>, <c>AbCipStatusMapper</c>, <c>FocasStatusMapper</c>,
/// <c>AbLegacyStatusMapper</c>, Galaxy's <c>StatusCodeMap</c>) — there is no shared helper in
/// <c>Core.Abstractions</c>, because drivers deliberately do not reference the OPC UA SDK and
/// <see cref="DataValueSnapshot.StatusCode"/> is a bare <see cref="uint"/>. The values below were read
/// off <c>Opc.Ua.StatusCodes</c> rather than copied from a sibling driver, since two of those tables
/// carry transcription errors.</para>
/// </summary>
public static class SqlStatusCodes
{
/// <summary>The value is good. (<c>Good</c>)</summary>
public const uint Good = 0x00000000u;
/// <summary>Quality is uncertain with no more specific reason — a NULL cell under <c>nullIsBad=false</c>.</summary>
public const uint Uncertain = 0x40000000u;
/// <summary>Quality is bad with no more specific reason — a NULL cell under <c>nullIsBad=true</c>.</summary>
public const uint Bad = 0x80000000u;
/// <summary>
/// No row was returned for this tag — the key is absent from the result set, or the wide-row
/// selector matched nothing. <b>Deliberately distinct from a NULL cell:</b> the row is gone, versus
/// the row is there and the value is not.
/// </summary>
public const uint BadNoData = 0x809B0000u;
/// <summary>The group exceeded its client-side wall-clock deadline (design §8.3).</summary>
public const uint BadTimeout = 0x800A0000u;
/// <summary>The reference resolves to no authored tag.</summary>
public const uint BadNodeIdUnknown = 0x80340000u;
/// <summary>The query failed after the connection opened.</summary>
public const uint BadCommunicationError = 0x80050000u;
/// <summary>The cell could not be coerced to the tag's effective data type.</summary>
public const uint BadTypeMismatch = 0x80740000u;
/// <summary>A tag definition the planner rejected — an authoring fault, not a database fault.</summary>
public const uint BadConfigurationError = 0x80890000u;
/// <summary>A defect in the reader itself; never expected in the field.</summary>
public const uint BadInternalError = 0x80020000u;
/// <summary>The quality-class mask — the top two bits of an OPC UA status code.</summary>
private const uint SeverityMask = 0xC0000000u;
/// <summary>True when <paramref name="statusCode"/> is in the Good quality class.</summary>
/// <param name="statusCode">The status code to classify.</param>
/// <returns>True when the code's severity bits are Good.</returns>
public static bool IsGood(uint statusCode) => (statusCode & SeverityMask) == 0u;
/// <summary>True when <paramref name="statusCode"/> is in the Uncertain quality class.</summary>
/// <param name="statusCode">The status code to classify.</param>
/// <returns>True when the code's severity bits are Uncertain.</returns>
public static bool IsUncertain(uint statusCode) => (statusCode & SeverityMask) == Uncertain;
/// <summary>True when <paramref name="statusCode"/> is in the Bad quality class.</summary>
/// <param name="statusCode">The status code to classify.</param>
/// <returns>True when the code's severity bits are Bad.</returns>
public static bool IsBad(uint statusCode) => (statusCode & SeverityMask) >= Bad;
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlPollReaderTests.cs
+480
View File
@@ -0,0 +1,480 @@
using System.Data;
using System.Data.Common;
using System.Diagnostics;
using Microsoft.Data.Sqlite;
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 <see cref="SqlPollReader"/> against a real database (<see cref="SqlitePollFixture"/>): the
/// N-in/N-out ordering contract, the slice-back of one result set to many tags, the three distinct
/// no-value outcomes (absent row / NULL cell / unresolvable ref), and — the part that cannot be proven
/// by reading the code — that a query which refuses to return <b>does not wedge the caller</b>.
/// <para><b>Each test owns its own fixture.</b> The fixture is a temp file, so a fresh one per test is
/// cheap and buys total isolation — which matters here because two tests deliberately mutate the
/// database (an extra duplicate row; an EXCLUSIVE transaction) in ways a shared fixture would leak into
/// every other test in the class.</para>
/// </summary>
public sealed class SqlPollReaderTests
{
private static readonly TimeSpan CommandTimeout = TimeSpan.FromSeconds(10);
private static readonly TimeSpan OperationTimeout = TimeSpan.FromSeconds(15);
// ---- the plan's headline contract: N in, N out, in input order ----
[Fact]
public async Task ReadAsync_returnsSnapshotsInInputOrder_absentKeyIsBadNoData()
{
using var fixture = new SqlitePollFixture();
var reader = Reader(fixture,
KvTag("Speed", SqlitePollFixture.PresentKey),
KvTag("Missing", SqlitePollFixture.AbsentKey),
KvTag("Temp", SqlitePollFixture.NullValueKey));
var snapshots = await reader.ReadAsync(["Speed", "Missing", "Temp"], CancellationToken.None);
snapshots.Count.ShouldBe(3);
// [0] present row, present cell — REAL infers Float64, so the published CLR type is double.
snapshots[0].Value.ShouldBe(SqlitePollFixture.PresentValue);
snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.Good);
snapshots[0].SourceTimestampUtc.ShouldBe(
new DateTime(2026, 7, 24, 10, 0, 0, DateTimeKind.Utc)); // from sample_ts, not the poll clock
// [1] no row at all — NOT the same thing as a NULL cell.
snapshots[1].StatusCode.ShouldBe(SqlStatusCodes.BadNoData);
snapshots[1].Value.ShouldBeNull();
// [2] row present, value cell NULL, nullIsBad = false ⇒ Uncertain (and the row's timestamp survives).
snapshots[2].Value.ShouldBeNull();
SqlStatusCodes.IsUncertain(snapshots[2].StatusCode).ShouldBeTrue();
snapshots[2].SourceTimestampUtc.ShouldBe(
new DateTime(2026, 7, 24, 10, 0, 1, DateTimeKind.Utc));
}
[Fact]
public async Task ReadAsync_withNullIsBad_publishesBadForANullCell_butStillBadNoDataForAnAbsentRow()
{
using var fixture = new SqlitePollFixture();
var reader = Reader(fixture, nullIsBad: true,
tags: [KvTag("Temp", SqlitePollFixture.NullValueKey), KvTag("Missing", SqlitePollFixture.AbsentKey)]);
var snapshots = await reader.ReadAsync(["Temp", "Missing"], CancellationToken.None);
snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.Bad);
SqlStatusCodes.IsBad(snapshots[0].StatusCode).ShouldBeTrue();
// The nullIsBad switch governs a NULL cell only — an absent row keeps its own, more specific code.
snapshots[1].StatusCode.ShouldBe(SqlStatusCodes.BadNoData);
}
[Fact]
public async Task ReadAsync_anUnresolvableRef_isBadNodeIdUnknown_andItsNeighboursStillRead()
{
using var fixture = new SqlitePollFixture();
var reader = Reader(fixture,
KvTag("Speed", SqlitePollFixture.PresentKey),
KvTag("Pressure", SqlitePollFixture.SecondPresentKey));
var snapshots = await reader.ReadAsync(
["Speed", "NotAuthored", "Pressure"], CancellationToken.None);
snapshots.Count.ShouldBe(3);
snapshots[0].Value.ShouldBe(SqlitePollFixture.PresentValue);
snapshots[1].StatusCode.ShouldBe(SqlStatusCodes.BadNodeIdUnknown);
// The hole must not shift the tail — this is the ordering contract's real failure mode.
snapshots[2].Value.ShouldBe(SqlitePollFixture.SecondPresentValue);
}
[Fact]
public async Task ReadAsync_twoTagsSharingOneKeyValue_bothReceiveTheValue()
{
// SqlQueryPlan.Members keeps duplicates: these two tags bind ONE parameter but stay TWO members,
// and a reader that indexed members by key into a 1:1 map would feed only one of them.
using var fixture = new SqlitePollFixture();
var reader = Reader(fixture,
KvTag("SpeedA", SqlitePollFixture.PresentKey),
KvTag("SpeedB", SqlitePollFixture.PresentKey));
var snapshots = await reader.ReadAsync(["SpeedA", "SpeedB"], CancellationToken.None);
snapshots[0].Value.ShouldBe(SqlitePollFixture.PresentValue);
snapshots[1].Value.ShouldBe(SqlitePollFixture.PresentValue);
snapshots[1].StatusCode.ShouldBe(SqlStatusCodes.Good);
}
[Fact]
public async Task ReadAsync_theSameRefTwice_feedsBothSlots()
{
using var fixture = new SqlitePollFixture();
var reader = Reader(fixture, KvTag("Speed", SqlitePollFixture.PresentKey));
var snapshots = await reader.ReadAsync(["Speed", "Speed"], CancellationToken.None);
snapshots.Count.ShouldBe(2);
snapshots[0].Value.ShouldBe(SqlitePollFixture.PresentValue);
snapshots[1].Value.ShouldBe(SqlitePollFixture.PresentValue);
}
[Fact]
public async Task ReadAsync_withNoRefs_returnsAnEmptyList_andOpensNoConnection()
{
using var fixture = new SqlitePollFixture();
var factory = new TrackingFactory();
var reader = new SqlPollReader(
factory, fixture.ConnectionString, new SqliteDialect(),
commandTimeout: CommandTimeout, operationTimeout: OperationTimeout,
maxConcurrentGroups: 4, nullIsBad: false, resolve: _ => null);
(await reader.ReadAsync([], CancellationToken.None)).ShouldBeEmpty();
factory.Created.ShouldBe(0);
}
// ---- wide row ----
[Fact]
public async Task ReadAsync_wideRow_slicesEachColumnToItsOwnMember()
{
using var fixture = new SqlitePollFixture();
var reader = Reader(fixture,
WideTag("Oven", "oven_temp", selectorValue: SqlitePollFixture.PresentStation),
WideTag("Pressure", "pressure", selectorValue: SqlitePollFixture.PresentStation));
var snapshots = await reader.ReadAsync(["Oven", "Pressure"], CancellationToken.None);
// One row, two tags — the wrong-slice defect would cross these two values over.
snapshots[0].Value.ShouldBe(SqlitePollFixture.PresentStationOvenTemp);
snapshots[1].Value.ShouldBe(SqlitePollFixture.PresentStationPressure);
snapshots[0].SourceTimestampUtc.ShouldBe(new DateTime(2026, 7, 24, 10, 0, 0, DateTimeKind.Utc));
}
[Fact]
public async Task ReadAsync_wideRow_absentRowIsBadNoData_andANullCellIsUncertain()
{
using var fixture = new SqlitePollFixture();
var reader = Reader(fixture,
WideTag("Gone", "oven_temp", selectorValue: SqlitePollFixture.AbsentStation),
WideTag("NullOven", "oven_temp", selectorValue: SqlitePollFixture.NullOvenTempStation),
WideTag("NullOvenPressure", "pressure", selectorValue: SqlitePollFixture.NullOvenTempStation));
var snapshots = await reader.ReadAsync(
["Gone", "NullOven", "NullOvenPressure"], CancellationToken.None);
snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.BadNoData); // no row matched the selector
SqlStatusCodes.IsUncertain(snapshots[1].StatusCode).ShouldBeTrue(); // row matched, cell NULL
snapshots[1].Value.ShouldBeNull();
snapshots[2].Value.ShouldBe(1.6); // its neighbour on the SAME row is unaffected
}
[Fact]
public async Task ReadAsync_wideRow_topByTimestamp_readsTheNewestRow()
{
using var fixture = new SqlitePollFixture();
var reader = Reader(fixture,
WideTag("Newest", "oven_temp", topByTimestamp: SqlitePollFixture.TimestampColumn));
var snapshots = await reader.ReadAsync(["Newest"], CancellationToken.None);
// Station 8, not the first-inserted station 7 — a lost ORDER BY / row limit shows up here.
snapshots[0].Value.ShouldBe(SqlitePollFixture.NewestStationOvenTemp);
snapshots[0].Value.ShouldNotBe(SqlitePollFixture.PresentStationOvenTemp);
}
// ---- type + timestamp mapping ----
[Fact]
public async Task ReadAsync_coercesTheCellToTheTagsDeclaredType()
{
using var fixture = new SqlitePollFixture();
var reader = Reader(fixture,
KvTag("AsInt", SqlitePollFixture.PresentKey, DriverDataType.Int32),
KvTag("AsText", SqlitePollFixture.PresentKey, DriverDataType.String),
KvTag("Inferred", SqlitePollFixture.PresentKey));
var snapshots = await reader.ReadAsync(["AsInt", "AsText", "Inferred"], CancellationToken.None);
snapshots[0].Value.ShouldBe(42); // int, not double — the declared type wins
snapshots[1].Value.ShouldBe("42"); // string
snapshots[2].Value.ShouldBe(42.0); // no declaration ⇒ dialect-inferred from REAL
}
[Fact]
public async Task ReadAsync_withNoTimestampColumn_stampsThePollClock()
{
using var fixture = new SqlitePollFixture();
var before = DateTime.UtcNow;
var reader = Reader(fixture, KvTag("Speed", SqlitePollFixture.PresentKey, timestampColumn: null));
var snapshots = await reader.ReadAsync(["Speed"], CancellationToken.None);
snapshots[0].Value.ShouldBe(SqlitePollFixture.PresentValue);
snapshots[0].SourceTimestampUtc.ShouldNotBeNull();
snapshots[0].SourceTimestampUtc!.Value.ShouldBeGreaterThanOrEqualTo(before.AddSeconds(-1));
snapshots[0].SourceTimestampUtc!.Value.ShouldBe(snapshots[0].ServerTimestampUtc);
}
[Fact]
public async Task ReadAsync_whenASourceViolatesOneRowPerKey_takesTheLastRowDeterministically()
{
using var fixture = new SqlitePollFixture();
fixture.Execute(
$"INSERT INTO {SqlitePollFixture.KeyValueTable} " +
$"({SqlitePollFixture.KeyColumn}, {SqlitePollFixture.ValueColumn}, {SqlitePollFixture.TimestampColumn}) " +
$"VALUES ('{SqlitePollFixture.PresentKey}', 99.0, '2026-07-24T10:00:09Z')");
var reader = Reader(fixture, KvTag("Speed", SqlitePollFixture.PresentKey));
var snapshots = await reader.ReadAsync(["Speed"], CancellationToken.None);
// Design §3.6: the contract is one row per key; when a source breaks it the reader is at least
// deterministic — last occurrence in reader order — rather than silently arbitrary.
snapshots[0].Value.ShouldBe(99.0);
}
// ---- the frozen-peer contract ----
[Fact]
public async Task ReadAsync_whenTheQueryCannotComplete_surfacesBadTimeoutWithinTheDeadline()
{
using var fixture = new SqlitePollFixture();
// A held EXCLUSIVE transaction is this rig's frozen peer: the SELECT below cannot proceed and
// Microsoft.Data.Sqlite's busy-retry loop is fully SYNCHRONOUS — it neither returns nor honours a
// cancellation token until CommandTimeout expires. That is precisely the S7 R2-01 shape (an async
// API that ignores its deadline), so only a real wall-clock bound can end this call early.
using var locker = fixture.OpenNewConnection();
using (var begin = locker.CreateCommand())
{
begin.CommandText = "BEGIN EXCLUSIVE";
begin.ExecuteNonQuery();
}
try
{
// Deliberately inverted against the authoring rule (operationTimeout > commandTimeout): the
// point is to prove the CLIENT-side bound fires while the server-side backstop would still wait.
var reader = new SqlPollReader(
fixture.Factory, fixture.ConnectionString, new SqliteDialect(),
commandTimeout: TimeSpan.FromSeconds(30), operationTimeout: TimeSpan.FromMilliseconds(500),
maxConcurrentGroups: 4, nullIsBad: false,
resolve: Resolver(KvTag("Speed", SqlitePollFixture.PresentKey)));
var clock = Stopwatch.StartNew();
var snapshots = await reader.ReadAsync(["Speed"], CancellationToken.None);
clock.Stop();
snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.BadTimeout);
snapshots[0].Value.ShouldBeNull();
// Without the wall-clock bound this returns at CommandTimeout (30 s), not at 0.5 s.
clock.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(10));
}
finally
{
using var rollback = locker.CreateCommand();
rollback.CommandText = "ROLLBACK";
rollback.ExecuteNonQuery();
}
}
[Fact]
public async Task ReadAsync_whenTheCallerCancels_propagatesTheCancellation()
{
// Deliberate asymmetry: OUR deadline expiring is a data-quality outcome (BadTimeout snapshots, so
// clients see the staleness); the CALLER cancelling is the engine tearing the poll down, and must
// propagate rather than be laundered into a snapshot nobody will consume.
using var fixture = new SqlitePollFixture();
var reader = Reader(fixture, KvTag("Speed", SqlitePollFixture.PresentKey));
using var cancelled = new CancellationTokenSource();
await cancelled.CancelAsync();
await Should.ThrowAsync<OperationCanceledException>(
async () => await reader.ReadAsync(["Speed"], cancelled.Token));
}
[Fact]
public async Task ReadAsync_whenTheDatabaseCannotBeOpened_throwsSoTheEngineBacksOff()
{
// IReadable's contract: per-tag failures are Bad-coded snapshots, but an unreachable driver throws.
using var fixture = new SqlitePollFixture();
var unreachable = new SqliteConnectionStringBuilder
{
DataSource = Path.Combine(
Path.GetTempPath(), $"otopcua-sql-no-such-dir-{Guid.NewGuid():N}", "db.sqlite"),
}.ToString();
var reader = new SqlPollReader(
fixture.Factory, unreachable, new SqliteDialect(),
commandTimeout: CommandTimeout, operationTimeout: OperationTimeout,
maxConcurrentGroups: 4, nullIsBad: false,
resolve: Resolver(KvTag("Speed", SqlitePollFixture.PresentKey)));
await Should.ThrowAsync<DbException>(
async () => await reader.ReadAsync(["Speed"], CancellationToken.None));
}
// ---- connection lifecycle ----
[Theory]
[InlineData(1)]
[InlineData(3)]
public async Task ReadAsync_neverHoldsMoreConnectionsThanMaxConcurrentGroups(int maxConcurrentGroups)
{
using var fixture = new SqlitePollFixture();
// Three distinct group keys — key-value, wide-row where-pair, wide-row topByTimestamp.
var tags = new[]
{
KvTag("Speed", SqlitePollFixture.PresentKey),
WideTag("Oven", "oven_temp", selectorValue: SqlitePollFixture.PresentStation),
WideTag("Newest", "oven_temp", topByTimestamp: SqlitePollFixture.TimestampColumn),
};
// The delay makes the groups genuinely overlap; without it a SQLite query is too fast for
// concurrency to be observable at all and the cap assertion below would pass vacuously.
var factory = new TrackingFactory(TimeSpan.FromMilliseconds(150));
var reader = new SqlPollReader(
factory, fixture.ConnectionString, new SqliteDialect(),
commandTimeout: CommandTimeout, operationTimeout: TimeSpan.FromSeconds(30),
maxConcurrentGroups: maxConcurrentGroups, nullIsBad: false, resolve: Resolver(tags));
var snapshots = await reader.ReadAsync(["Speed", "Oven", "Newest"], CancellationToken.None);
snapshots.ShouldAllBe(s => s.StatusCode == SqlStatusCodes.Good);
factory.Created.ShouldBe(3);
factory.Peak.ShouldBe(maxConcurrentGroups); // == 3 for the uncapped run proves the overlap is real
factory.Live.ShouldBe(0); // every connection closed — no leak under the poll loop
}
[Fact]
public async Task ReadAsync_repeatedPolls_leakNoConnections()
{
using var fixture = new SqlitePollFixture();
var factory = new TrackingFactory();
var reader = new SqlPollReader(
factory, fixture.ConnectionString, new SqliteDialect(),
commandTimeout: CommandTimeout, operationTimeout: OperationTimeout,
maxConcurrentGroups: 4, nullIsBad: false,
resolve: Resolver(KvTag("Speed", SqlitePollFixture.PresentKey)));
for (var poll = 0; poll < 5; poll++)
{
var snapshots = await reader.ReadAsync(["Speed"], CancellationToken.None);
snapshots[0].Value.ShouldBe(SqlitePollFixture.PresentValue);
}
factory.Created.ShouldBe(5);
factory.Live.ShouldBe(0);
}
// ---- argument guards ----
[Fact]
public void Constructor_rejectsAnUnusableTimeoutOrConcurrencyCap()
{
var dialect = new SqliteDialect();
Should.Throw<ArgumentOutOfRangeException>(() => new SqlPollReader(
dialect.Factory, "Data Source=x", dialect, TimeSpan.Zero, OperationTimeout, 4, false, _ => null));
Should.Throw<ArgumentOutOfRangeException>(() => new SqlPollReader(
dialect.Factory, "Data Source=x", dialect, CommandTimeout, TimeSpan.Zero, 4, false, _ => null));
Should.Throw<ArgumentOutOfRangeException>(() => new SqlPollReader(
dialect.Factory, "Data Source=x", dialect, CommandTimeout, OperationTimeout, 0, false, _ => null));
Should.Throw<ArgumentException>(() => new SqlPollReader(
dialect.Factory, " ", dialect, CommandTimeout, OperationTimeout, 4, false, _ => null));
}
[Fact]
public async Task ReadAsync_rejectsANullReferenceList()
=> await Should.ThrowAsync<ArgumentNullException>(async () =>
{
using var fixture = new SqlitePollFixture();
await Reader(fixture).ReadAsync(null!, CancellationToken.None);
});
// ---- helpers ----
private static SqlPollReader Reader(SqlitePollFixture fixture, params SqlTagDefinition[] tags)
=> Reader(fixture, nullIsBad: false, tags: tags);
private static SqlPollReader Reader(
SqlitePollFixture fixture, bool nullIsBad, params SqlTagDefinition[] tags)
=> new(fixture.Factory, fixture.ConnectionString, new SqliteDialect(),
commandTimeout: CommandTimeout, operationTimeout: OperationTimeout,
maxConcurrentGroups: 4, nullIsBad: nullIsBad, resolve: Resolver(tags));
/// <summary>The driver's RawPath→definition table, standing in for <c>EquipmentTagRefResolver</c>.</summary>
private static Func<string, SqlTagDefinition?> Resolver(params SqlTagDefinition[] tags)
{
var table = tags.ToDictionary(t => t.Name, StringComparer.Ordinal);
return rawPath => table.GetValueOrDefault(rawPath);
}
private static SqlTagDefinition KvTag(
string name,
string keyValue,
DriverDataType? declaredType = null,
string? timestampColumn = SqlitePollFixture.TimestampColumn)
=> new(name, SqlTagModel.KeyValue, SqlitePollFixture.KeyValueTable,
KeyColumn: SqlitePollFixture.KeyColumn, KeyValue: keyValue,
ValueColumn: SqlitePollFixture.ValueColumn,
TimestampColumn: timestampColumn,
DeclaredType: declaredType);
private static SqlTagDefinition WideTag(
string name,
string columnName,
string? selectorValue = null,
string? topByTimestamp = null,
string? timestampColumn = SqlitePollFixture.TimestampColumn)
=> new(name, SqlTagModel.WideRow, SqlitePollFixture.WideRowTable,
ColumnName: columnName,
TimestampColumn: timestampColumn,
RowSelectorColumn: selectorValue is null ? null : SqlitePollFixture.WideRowSelectorColumn,
RowSelectorValue: selectorValue,
RowSelectorTopByTimestamp: topByTimestamp);
/// <summary>
/// A <see cref="DbProviderFactory"/> over SQLite that counts how many connections the reader has
/// created and how many are alive at once — the only way to observe the connection lifecycle from
/// outside, since <see cref="SqlPollReader"/> deliberately owns its connections end to end.
/// <para>The optional <c>createDelay</c> widens the window each connection is alive so that
/// concurrent group execution is actually observable against a database whose queries would
/// otherwise finish in microseconds.</para>
/// </summary>
private sealed class TrackingFactory(TimeSpan createDelay = default) : DbProviderFactory
{
private readonly object _sync = new();
private int _live;
private int _peak;
private int _created;
/// <summary>How many connections the reader has asked the factory for.</summary>
public int Created { get { lock (_sync) { return _created; } } }
/// <summary>How many created connections have not yet closed.</summary>
public int Live { get { lock (_sync) { return _live; } } }
/// <summary>The high-water mark of <see cref="Live"/>.</summary>
public int Peak { get { lock (_sync) { return _peak; } } }
public override DbConnection CreateConnection()
{
var connection = SqliteFactory.Instance.CreateConnection()!;
connection.StateChange += OnStateChange;
lock (_sync)
{
_created++;
_live++;
if (_live > _peak) _peak = _live;
}
if (createDelay > TimeSpan.Zero) Thread.Sleep(createDelay);
return connection;
}
private void OnStateChange(object sender, StateChangeEventArgs e)
{
if (e.CurrentState != ConnectionState.Closed && e.CurrentState != ConnectionState.Broken) return;
lock (_sync) { _live--; }
}
}
}