diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs index 2c4155ea..ee1dce0f 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs @@ -84,7 +84,13 @@ public sealed class SqlPollReader /// 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. + /// + /// Maximum group queries — and connections — in flight at once. A timed-out group keeps its slot + /// until it truly finishes, so this is a hard ceiling on connections even against a frozen database. + /// See for why a high value is safe under + /// Microsoft.Data.SqlClient but wants sizing against thread-pool headroom under a provider + /// that blocks synchronously. + /// /// /// publishes for a NULL value cell instead /// of the default . It governs a present row with a @@ -244,6 +250,23 @@ public sealed class SqlPollReader /// 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. + /// Thread-pool scheduling and what a BadTimeout therefore means. The group's clock + /// starts before Task.Run, so in principle a group could burn budget queueing for a pool + /// thread rather than waiting on the database. TaskCreationOptions.LongRunning (a dedicated + /// OS thread) was considered and rejected: it would create and tear down one thread per group + /// per poll, forever, on a driver whose whole job is to poll on a short cadence — a permanent cost + /// paid against a hazard that cannot arise with the provider this driver actually ships. + /// Microsoft.Data.SqlClient's async path is genuinely asynchronous: the delegate reaches its + /// first real await within microseconds and returns the pool thread, so a frozen SQL Server parks + /// zero pool threads however long it stays frozen, and the driver cannot starve itself no + /// matter how high maxConcurrentGroups goes. The queueing scenario needs a provider that + /// degrades to synchronous blocking — which is exactly what the SQLite test rig exploits on purpose, + /// and is not a shipped configuration. Consequence for outage diagnosis (e.g. blackhole-testing a + /// paused SQL Server): under Microsoft.Data.SqlClient a + /// means the database did not answer inside + /// operationTimeout — it cannot be an artefact of this driver's own pool pressure. Under a + /// synchronously-blocking provider it may also include queueing delay, so size + /// maxConcurrentGroups below the process's readily-available worker count there. /// 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 @@ -321,9 +344,13 @@ public sealed class SqlPollReader /// Records the deadline breach and detaches the abandoned work. private GroupResult TimedOut(SqlQueryPlan plan, Task work) { + // The table is named for the same reason the contract warnings name it: under a partial outage this + // line repeats every poll, and "which source is frozen" is the only question the operator has. _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); + "Sql poll group ({Model}, '{Table}', {Members} tag(s)) exceeded the {Timeout} ms operation " + + "timeout; publishing BadTimeout.", + plan.Model, plan.Members[0].Table, plan.Members.Count, + (int)_operationTimeout.TotalMilliseconds); Detach(work); return GroupResult.TimedOut; @@ -459,10 +486,14 @@ public sealed class SqlPollReader 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). + // where-column is not even in the result set (design §5.3). Both selections happen ONCE per group — + // they are group-wide facts, and doing them per member would report a contract violation N times. var byKey = outcome.Succeeded && plan.Model == SqlTagModel.KeyValue ? IndexByKey(plan, outcome) : null; + var wideRow = outcome.Succeeded && plan.Model == SqlTagModel.WideRow + ? SelectWideRow(plan, outcome) + : null; foreach (var member in plan.Members) { @@ -470,7 +501,7 @@ public sealed class SqlPollReader if (!slots.TryGetValue(member, out var positions)) continue; var snapshot = outcome.Succeeded - ? MapMember(plan, member, outcome, byKey, readAt) + ? MapMember(plan, member, outcome, byKey, wideRow, readAt) : new DataValueSnapshot(null, outcome.FailureStatus, null, readAt); foreach (var position in positions) results[position] = snapshot; @@ -518,12 +549,50 @@ public sealed class SqlPollReader return index; } + /// + /// Picks the single row every member of a wide-row plan reads, and reports an ambiguous selector. + /// Last row wins — the same tie-break applies, so both models + /// degrade the same way. It is deterministic only within one result set: the where-pair form + /// emits no ORDER BY, so across polls the "last" row is whatever the server returned last, + /// i.e. storage order. That is why more than one row is a reported contract violation and not merely + /// a tie to break — the model's premise (design §5.3) is that the selector identifies one row. + /// Why the where-pair form is deliberately NOT given the dialect's single-row limit. A + /// TOP 1 with no ORDER BY is not deterministic either — it returns whichever row the + /// plan happens to produce first, so it would trade one arbitrary row for a differently arbitrary + /// one. Worse, it would destroy the only evidence this method has that the selector is ambiguous + /// (Rows.Count > 1), converting a loud misconfiguration into a permanently silent one — + /// precisely the failure mode this class's summary calls its top risk. The topByTimestamp + /// form keeps its row limit because there the ORDER BY makes "the first row" a defined + /// answer. The accepted cost is that an ambiguous selector materialises every matching row for one + /// poll; the warning is what gets it fixed. + /// + private object?[]? SelectWideRow(SqlQueryPlan plan, GroupResult outcome) + { + if (outcome.Rows.Count == 0) return null; + if (outcome.Rows.Count > 1 && ShouldWarnAboutContract()) + { + var first = plan.Members[0]; + var selector = string.IsNullOrWhiteSpace(first.RowSelectorColumn) + ? string.Concat("topByTimestamp ", first.RowSelectorTopByTimestamp) + : string.Concat(first.RowSelectorColumn, " = ", first.RowSelectorValue); + _logger.LogWarning( + "Sql poll source '{Table}' returned {Rows} rows for the wide-row selector '{Selector}' in " + + "one poll; the last row read wins, and the query carries no ORDER BY, so which row that is " + + "depends on storage order. The wide-row model requires a selector matching exactly one row " + + "— narrow the whereColumn/whereValue pair, or point the tag at a latest-row view.", + first.Table, outcome.Rows.Count, selector); + } + + return outcome.Rows[^1]; + } + /// Maps one member's cell to a snapshot: value, quality, and source timestamp. private DataValueSnapshot MapMember( SqlQueryPlan plan, SqlTagDefinition member, GroupResult outcome, Dictionary? byKey, + object?[]? wideRow, DateTime readAt) { object?[]? row; @@ -541,7 +610,7 @@ public sealed class SqlPollReader // 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; + row = wideRow; } // No row: the key was deleted, or the row selector matched nothing. Distinct from a NULL cell. 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 index 79897f3e..cc203d9d 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlPollReaderTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlPollReaderTests.cs @@ -1,7 +1,9 @@ using System.Data; using System.Data.Common; using System.Diagnostics; +using System.Globalization; using Microsoft.Data.Sqlite; +using Microsoft.Extensions.Logging; using Shouldly; using Xunit; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; @@ -184,6 +186,56 @@ public sealed class SqlPollReaderTests snapshots[0].Value.ShouldNotBe(SqlitePollFixture.PresentStationOvenTemp); } + [Fact] + public async Task ReadAsync_wideRow_whenTheSelectorMatchesManyRows_warnsInsteadOfSilentlyPickingOne() + { + using var fixture = new SqlitePollFixture(); + // A second row for the SAME station. The where-pair form emits no ORDER BY and no row limit, so + // which of the two the reader publishes is decided by physical storage order — an authoring mistake + // (a selector column that is not actually unique) that must not pass silently. This is the WideRow + // counterpart of ReadAsync_whenASourceViolatesOneRowPerKey_...'s duplicate-key warning. + AddSecondRowForPresentStation(fixture); + var logger = new RecordingLogger(); + var reader = Reader(fixture, logger, + WideTag("Oven", "oven_temp", selectorValue: SqlitePollFixture.PresentStation)); + + var snapshots = await reader.ReadAsync(["Oven"], CancellationToken.None); + + // Still a value — the reader degrades to "last row wins", exactly as the key-value model does. + snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.Good); + + var warning = logger.Entries.ShouldHaveSingleItem(); + warning.Level.ShouldBe(LogLevel.Warning); + // Names the table AND the selector, so an operator can find the offending tag from the log alone. + warning.Message.ShouldContain(SqlitePollFixture.WideRowTable); + warning.Message.ShouldContain(SqlitePollFixture.WideRowSelectorColumn); + warning.Message.ShouldContain(SqlitePollFixture.PresentStation); + } + + [Fact] + public async Task ReadAsync_wideRow_theAmbiguousSelectorWarningIsRateLimited_andSilentWhenUnambiguous() + { + using var fixture = new SqlitePollFixture(); + var logger = new RecordingLogger(); + + // A selector that matches exactly one row is the normal case and must log nothing at all — + // without this leg a warning that always fires would still pass the test above. + var clean = Reader(fixture, logger, + WideTag("Oven", "oven_temp", selectorValue: SqlitePollFixture.PresentStation)); + await clean.ReadAsync(["Oven"], CancellationToken.None); + logger.Entries.ShouldBeEmpty(); + + AddSecondRowForPresentStation(fixture); + var ambiguous = Reader(fixture, logger, + WideTag("Oven", "oven_temp", selectorValue: SqlitePollFixture.PresentStation)); + await ambiguous.ReadAsync(["Oven"], CancellationToken.None); + await ambiguous.ReadAsync(["Oven"], CancellationToken.None); + + // A poll loop hits this every cycle; the rate limit is what stops a misconfigured table flooding + // the log. Two polls, one warning. + logger.Entries.Count.ShouldBe(1); + } + // ---- type + timestamp mapping ---- [Fact] @@ -246,11 +298,7 @@ public sealed class SqlPollReaderTests // 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(); - } + Execute(locker, "BEGIN EXCLUSIVE"); try { @@ -273,12 +321,63 @@ public sealed class SqlPollReaderTests } finally { - using var rollback = locker.CreateCommand(); - rollback.CommandText = "ROLLBACK"; - rollback.ExecuteNonQuery(); + Execute(locker, "ROLLBACK"); } } + [Fact] + public async Task ReadAsync_aTimedOutGroupKeepsItsConcurrencySlotUntilTheQueryTrulyFinishes() + { + // The reason the concurrency slot is released by the WORK and not by the waiter: a timed-out group + // is still holding a connection, so handing its slot back at the deadline would let the next poll + // open another one — and a database that stays frozen would accumulate one connection per poll pass + // forever. This test is the only thing that distinguishes the two designs; the concurrency-cap test + // above never times a group out, and the BadTimeout test above never counts connections. + using var fixture = new SqlitePollFixture(); + var factory = new TrackingFactory(); + var reader = new SqlPollReader( + factory, fixture.ConnectionString, new SqliteDialect(), + // commandTimeout deliberately far beyond the test's own horizon: the wedged query must stay + // wedged for as long as the lock is held, so nothing but the test releases the slot. + commandTimeout: TimeSpan.FromSeconds(30), operationTimeout: TimeSpan.FromMilliseconds(500), + maxConcurrentGroups: 1, nullIsBad: false, + resolve: Resolver(KvTag("Speed", SqlitePollFixture.PresentKey))); + + var locker = fixture.OpenNewConnection(); + Execute(locker, "BEGIN EXCLUSIVE"); // outside the try: a ROLLBACK with no open transaction throws + try + { + // Poll 1 wedges against the lock and times out, abandoning a still-running query. + var first = await reader.ReadAsync(["Speed"], CancellationToken.None); + first[0].StatusCode.ShouldBe(SqlStatusCodes.BadTimeout); + factory.Created.ShouldBe(1); + factory.Live.ShouldBe(1); // the zombie still owns its connection + + // (a) The slot is STILL held. maxConcurrentGroups is 1, so poll 2 cannot even reach the + // factory: it times out waiting on the gate. Created staying at 1 is the load-bearing + // assertion — release-from-the-waiter would make this 2. + var second = await reader.ReadAsync(["Speed"], CancellationToken.None); + second[0].StatusCode.ShouldBe(SqlStatusCodes.BadTimeout); + factory.Created.ShouldBe(1); + factory.Live.ShouldBe(1); + } + finally + { + Execute(locker, "ROLLBACK"); + locker.Dispose(); + } + + // (b) With the lock gone the zombie's query completes (or faults on its own cancelled deadline), + // closes its connection, and only then hands the slot back. + (await WaitUntilAsync(() => factory.Live == 0, TimeSpan.FromSeconds(30))) + .ShouldBeTrue("the abandoned group never released its connection"); + + var third = await reader.ReadAsync(["Speed"], CancellationToken.None); + third[0].Value.ShouldBe(SqlitePollFixture.PresentValue); + factory.Created.ShouldBe(2); // the slot was reusable exactly once the zombie finished + factory.Live.ShouldBe(0); + } + [Fact] public async Task ReadAsync_whenTheCallerCancels_propagatesTheCancellation() { @@ -401,6 +500,49 @@ public sealed class SqlPollReaderTests commandTimeout: CommandTimeout, operationTimeout: OperationTimeout, maxConcurrentGroups: 4, nullIsBad: nullIsBad, resolve: Resolver(tags)); + private static SqlPollReader Reader( + SqlitePollFixture fixture, ILogger logger, params SqlTagDefinition[] tags) + => new(fixture.Factory, fixture.ConnectionString, new SqliteDialect(), + commandTimeout: CommandTimeout, operationTimeout: OperationTimeout, + maxConcurrentGroups: 4, nullIsBad: false, resolve: Resolver(tags), logger: logger); + + /// + /// Adds a second row for the station the wide-row tests + /// select, breaking that model's one-row-per-selector contract. + /// + private static void AddSecondRowForPresentStation(SqlitePollFixture fixture) + => fixture.Execute(string.Create(CultureInfo.InvariantCulture, $""" + INSERT INTO {SqlitePollFixture.WideRowTable} + ({SqlitePollFixture.WideRowSelectorColumn}, oven_temp, pressure, + {SqlitePollFixture.TimestampColumn}) + VALUES ({SqlitePollFixture.PresentStation}, 999.0, 9.9, '2026-07-24T10:00:09Z') + """)); + + /// Runs one statement on a caller-owned connection. + private static void Execute(SqliteConnection connection, string sql) + { + using var command = connection.CreateCommand(); + command.CommandText = sql; + command.ExecuteNonQuery(); + } + + /// + /// Polls until it holds or elapses. Bounded on + /// purpose: a test that asserts an abandoned task eventually finishes must fail, not hang, when it + /// does not. + /// + private static async Task WaitUntilAsync(Func condition, TimeSpan limit) + { + var clock = Stopwatch.StartNew(); + while (clock.Elapsed < limit) + { + if (condition()) return true; + await Task.Delay(25); + } + + return condition(); + } + /// The driver's RawPath→definition table, standing in for EquipmentTagRefResolver. private static Func Resolver(params SqlTagDefinition[] tags) { @@ -432,6 +574,35 @@ public sealed class SqlPollReaderTests RowSelectorValue: selectorValue, RowSelectorTopByTimestamp: topByTimestamp); + /// + /// Captures the reader's log so "reported, not silent" can be asserted rather than assumed — the + /// contract-violation warnings are the reader's only signal that a source is misauthored, so they are + /// behaviour, not diagnostics. + /// + private sealed class RecordingLogger : ILogger + { + /// Every record written, level + rendered message. + public List<(LogLevel Level, string Message)> Entries { get; } = []; + + public IDisposable BeginScope(TState state) where TState : notnull => Scope.Instance; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log( + LogLevel logLevel, EventId eventId, TState state, Exception? exception, + Func formatter) + { + lock (Entries) Entries.Add((logLevel, formatter(state, exception))); + } + + private sealed class Scope : IDisposable + { + public static Scope Instance { get; } = new(); + + public void Dispose() { } + } + } + /// /// 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