diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs new file mode 100644 index 00000000..2c4155ea --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs @@ -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; + +/// +/// Executes one poll pass: resolves each requested reference to a , folds +/// the definitions into the minimum number of queries via , runs one +/// command per group under a bounded deadline, and slices each result set back to per-tag +/// s — positionally aligned with the caller's reference list +/// (design §3.2). Structurally the SQL analogue of ModbusDriver.ReadCoalescedAsync: group → one +/// round-trip → slice back. +/// N in, N out, in order. 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 contract) — see 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 one tag's value onto another tag's +/// node. +/// The whole call throws only when the database is unreachable — that is, when opening a +/// connection fails. That surfaces to as a poll failure and earns the +/// capped-exponential backoff; the driver's IHostConnectivityProbe 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 after the connection opened is the group's problem, not the +/// server's, and Bad-codes that group only. +/// Deadlines (design §8.3, the R2-01 frozen-peer lesson). Two independent mechanisms, both +/// required. CommandTimeout is the server-side backstop and bounds nothing on a wedged +/// socket. The wall-clock bound is client-side and is what actually guarantees this method returns: see +/// . A frozen database yields +/// snapshots — it must never wedge the poll thread. +/// Not a driver. 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 +/// and own those concerns. +/// +public sealed class SqlPollReader +{ + /// Minimum spacing between "your source violates the query contract" warnings, per reader. + 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 _resolve; + private readonly ILogger _logger; + + /// + /// Caps concurrently executing group queries — and therefore concurrently held connections + /// (design §8.4). Never disposed: a only needs disposal once its + /// AvailableWaitHandle has been touched, and not disposing it also removes the hazard of a + /// timed-out-but-still-running group releasing into a disposed semaphore. + /// + private readonly SemaphoreSlim _gate; + + private long _lastContractWarningTicks; + + /// Initializes a new instance of the class. + /// + /// Creates the provider's connections. Passed explicitly rather than taken from + /// so the driver can hand in an instrumented or pre-configured factory. + /// + /// + /// The resolved connection string (never the authored connectionStringRef — secrets are + /// resolved by the driver at Initialize, design §8.2). + /// + /// Supplies identifier quoting, row-limit syntax, and column-type mapping. + /// + /// The ADO.NET CommandTimeout backstop. Rounded up to whole seconds and floored at 1, + /// because ADO.NET reads CommandTimeout = 0 as "wait forever" — the one value this option + /// must never silently become. + /// + /// + /// 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 + /// . + /// Authoring should keep this strictly greater than + /// (design §8.3): inverted, the client-side abort always fires first and masks the server-side + /// backstop. That rule is not enforced here — it belongs to config validation, and this type + /// stays usable for the tests that must deliberately invert it. + /// + /// Maximum group queries — and connections — in flight at once. + /// + /// publishes for a NULL value cell instead + /// of the default . It governs a present row with a + /// NULL cell only — an absent row is always . + /// + /// + /// RawPath → tag definition; on a miss. Shaped to match + /// 's lookup so the driver can pass its authored table + /// straight through. + /// + /// Optional; defaults to a no-op logger. + /// A required reference argument is null. + /// is blank. + /// A timeout is non-positive, or the cap is below 1. + public SqlPollReader( + DbProviderFactory factory, + string connectionString, + ISqlDialect dialect, + TimeSpan commandTimeout, + TimeSpan operationTimeout, + int maxConcurrentGroups, + bool nullIsBad, + Func 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); + } + + /// + /// Reads every reference in one poll pass. Matches 's shape so the + /// driver shell can delegate to it directly. + /// + /// The RawPaths to read. Empty yields an empty result and touches no connection. + /// + /// The caller's token. Its cancellation propagates as an + /// — the engine asked the poll to stop, and nobody is + /// waiting for the snapshots. This is deliberately asymmetric with an operationTimeout + /// breach, which is a data-quality fact clients must see and so becomes + /// snapshots. + /// + /// One snapshot per reference, at the same index. + /// is null. + /// was cancelled. + /// The database could not be reached (opening a connection failed). + public async Task> ReadAsync( + IReadOnlyList 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(fullReferences.Count); + var slots = new Dictionary>(); + 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; + } + + /// Plans the resolved definitions, runs every group concurrently, and slices the results back. + private async Task ReadGroupsAsync( + List definitions, + Dictionary> slots, + DataValueSnapshot[] results, + DateTime readAt, + CancellationToken cancellationToken) + { + IReadOnlyList 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[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); + } + + /// + /// Runs one group's query under a hard wall-clock ceiling. + /// Why the ceiling is Task.WaitAsync and not just a linked + /// CancelAfter. A linked token only bounds a + /// provider that honours 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 + /// ExecuteReaderAsync 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 WaitAsync so + /// control returns at the deadline regardless. + /// 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. + /// The concurrency slot is released by the work, not by the waiter. A timed-out group + /// keeps its slot until it truly finishes, so a frozen database can never accumulate more than + /// maxConcurrentGroups 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. + /// + private async Task 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 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; + } + } + + /// How much of this group's operationTimeout budget is left; never negative. + private TimeSpan Remaining(Stopwatch clock) + { + var remaining = _operationTimeout - clock.Elapsed; + return remaining > TimeSpan.Zero ? remaining : TimeSpan.Zero; + } + + /// Records the deadline breach and detaches the abandoned work. + 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; + } + + /// + /// 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. + /// + private static void Detach(Task work) + => _ = work.ContinueWith( + static t => _ = t.Exception, + CancellationToken.None, + TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously, + TaskScheduler.Default); + + /// + /// Opens a connection, executes the plan's single command, and materialises the result set. + /// The rows are read into memory before 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. + /// + private async Task 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; + } + } + } + + /// + /// Reads every row into memory, projected onto . 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. + /// + private async Task 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(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(); + 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); + } + + /// Reads a column's provider type name, falling back to the dialect's unknown-type default. + 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; + } + } + + /// Maps one group's outcome onto every slot its members occupy in the caller's list. + private void Slice( + SqlQueryPlan plan, + GroupResult outcome, + Dictionary> 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; + } + } + + /// + /// 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. + /// + private Dictionary IndexByKey(SqlQueryPlan plan, GroupResult outcome) + { + var index = new Dictionary(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; + } + + /// Maps one member's cell to a snapshot: value, quality, and source timestamp. + private DataValueSnapshot MapMember( + SqlQueryPlan plan, + SqlTagDefinition member, + GroupResult outcome, + Dictionary? 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); + } + + /// Reads a row's source timestamp, or null when there is no column, no value, or no parse. + 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; + } + + /// + /// Coerces a provider cell to the tag's effective , so the published CLR + /// type matches the type the address space declared for the node. Never throws. + /// + 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; + } + } + + /// + /// Normalises a timestamp cell to UTC. A naive datetime (no offset — the common SQL Server + /// shape) is assumed 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. + /// + 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), + }; + + /// Rate-limits the source-contract warning so a misconfigured table cannot flood the log. + 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; + } + + /// One group's outcome: either a materialised result set, or the status its members inherit. + private sealed class GroupResult + { + private static readonly IReadOnlyList NoRows = []; + private static readonly IReadOnlyDictionary NoColumns = + new Dictionary(StringComparer.Ordinal); + + private GroupResult( + uint failureStatus, + IReadOnlyList rows, + IReadOnlyDictionary columnIndex, + IReadOnlyList typeNames) + { + FailureStatus = failureStatus; + Rows = rows; + ColumnIndex = columnIndex; + TypeNames = typeNames; + } + + /// The group exceeded its wall-clock deadline (design §8.3). + public static GroupResult TimedOut { get; } = + new(SqlStatusCodes.BadTimeout, NoRows, NoColumns, []); + + /// The connection was fine but the query was not. + public static GroupResult Failed { get; } = + new(SqlStatusCodes.BadCommunicationError, NoRows, NoColumns, []); + + /// The status every member inherits when is false. + public uint FailureStatus { get; } + + /// The materialised rows, projected onto the plan's selected columns. + public IReadOnlyList Rows { get; } + + /// Selected column name → its position within each row array. + public IReadOnlyDictionary ColumnIndex { get; } + + /// Each selected column's provider type name, for dialect type inference. + public IReadOnlyList TypeNames { get; } + + /// True when the query ran; a zero-row result set is still a success. + public bool Succeeded => FailureStatus == SqlStatusCodes.Good; + + /// A completed result set. + /// The materialised rows. + /// Selected column name → row-array position. + /// Each selected column's provider type name. + /// A successful outcome. + public static GroupResult Ok( + IReadOnlyList rows, + IReadOnlyDictionary columnIndex, + IReadOnlyList typeNames) + => new(SqlStatusCodes.Good, rows, columnIndex, typeNames); + } +} + +/// +/// The OPC UA status codes the Sql driver publishes, and the quality-class predicates its tests read +/// them back with. +/// Why a driver-local table rather than a shared one. Every driver in this tree carries its +/// own (TwinCATStatusMapper, AbCipStatusMapper, FocasStatusMapper, +/// AbLegacyStatusMapper, Galaxy's StatusCodeMap) — there is no shared helper in +/// Core.Abstractions, because drivers deliberately do not reference the OPC UA SDK and +/// is a bare . The values below were read +/// off Opc.Ua.StatusCodes rather than copied from a sibling driver, since two of those tables +/// carry transcription errors. +/// +public static class SqlStatusCodes +{ + /// The value is good. (Good) + public const uint Good = 0x00000000u; + + /// Quality is uncertain with no more specific reason — a NULL cell under nullIsBad=false. + public const uint Uncertain = 0x40000000u; + + /// Quality is bad with no more specific reason — a NULL cell under nullIsBad=true. + public const uint Bad = 0x80000000u; + + /// + /// No row was returned for this tag — the key is absent from the result set, or the wide-row + /// selector matched nothing. Deliberately distinct from a NULL cell: the row is gone, versus + /// the row is there and the value is not. + /// + public const uint BadNoData = 0x809B0000u; + + /// The group exceeded its client-side wall-clock deadline (design §8.3). + public const uint BadTimeout = 0x800A0000u; + + /// The reference resolves to no authored tag. + public const uint BadNodeIdUnknown = 0x80340000u; + + /// The query failed after the connection opened. + public const uint BadCommunicationError = 0x80050000u; + + /// The cell could not be coerced to the tag's effective data type. + public const uint BadTypeMismatch = 0x80740000u; + + /// A tag definition the planner rejected — an authoring fault, not a database fault. + public const uint BadConfigurationError = 0x80890000u; + + /// A defect in the reader itself; never expected in the field. + public const uint BadInternalError = 0x80020000u; + + /// The quality-class mask — the top two bits of an OPC UA status code. + private const uint SeverityMask = 0xC0000000u; + + /// True when is in the Good quality class. + /// The status code to classify. + /// True when the code's severity bits are Good. + public static bool IsGood(uint statusCode) => (statusCode & SeverityMask) == 0u; + + /// True when is in the Uncertain quality class. + /// The status code to classify. + /// True when the code's severity bits are Uncertain. + public static bool IsUncertain(uint statusCode) => (statusCode & SeverityMask) == Uncertain; + + /// True when is in the Bad quality class. + /// The status code to classify. + /// True when the code's severity bits are Bad. + public static bool IsBad(uint statusCode) => (statusCode & SeverityMask) >= Bad; +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlPollReaderTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlPollReaderTests.cs new file mode 100644 index 00000000..79897f3e --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlPollReaderTests.cs @@ -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; + +/// +/// Proves against a real database (): 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 does not wedge the caller. +/// Each test owns its own fixture. 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. +/// +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( + 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( + 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(() => new SqlPollReader( + dialect.Factory, "Data Source=x", dialect, TimeSpan.Zero, OperationTimeout, 4, false, _ => null)); + Should.Throw(() => new SqlPollReader( + dialect.Factory, "Data Source=x", dialect, CommandTimeout, TimeSpan.Zero, 4, false, _ => null)); + Should.Throw(() => new SqlPollReader( + dialect.Factory, "Data Source=x", dialect, CommandTimeout, OperationTimeout, 0, false, _ => null)); + Should.Throw(() => new SqlPollReader( + dialect.Factory, " ", dialect, CommandTimeout, OperationTimeout, 4, false, _ => null)); + } + + [Fact] + public async Task ReadAsync_rejectsANullReferenceList() + => await Should.ThrowAsync(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)); + + /// The driver's RawPath→definition table, standing in for EquipmentTagRefResolver. + private static Func 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); + + /// + /// A 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 deliberately owns its connections end to end. + /// The optional createDelay 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. + /// + private sealed class TrackingFactory(TimeSpan createDelay = default) : DbProviderFactory + { + private readonly object _sync = new(); + private int _live; + private int _peak; + private int _created; + + /// How many connections the reader has asked the factory for. + public int Created { get { lock (_sync) { return _created; } } } + + /// How many created connections have not yet closed. + public int Live { get { lock (_sync) { return _live; } } } + + /// The high-water mark of . + 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--; } + } + } +}