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

fix(sql): report an ambiguous wide-row selector; pin the zombie-slot bound

Browse Source 
Three review findings against SqlPollReader (967e5140).

1. WideRow's where-pair form emits no ORDER BY and no row limit, so a
   selector that is not actually unique made the reader publish an
   arbitrary, storage-order-dependent row — silently, unlike IndexByKey's
   duplicate-key warning. Row selection moves out of MapMember into a new
   SelectWideRow, which is called once per group and emits the same
   rate-limited contract warning (naming table, selector and row count).

   The row-limit half of the suggested fix was deliberately NOT taken: a
   TOP 1 with no ORDER BY is not deterministic either, and it would
   destroy the Rows.Count > 1 evidence the warning depends on, turning a
   loud misconfiguration into a permanently silent one. Rationale is in
   SelectWideRow's docs. No SQL changed, so no planner golden string moved.

2. The "a frozen database can never accumulate more than
   maxConcurrentGroups connections" claim was asserted in the docs and
   untested. ReadAsync_aTimedOutGroupKeepsItsConcurrencySlotUntilTheQuery-
   TrulyFinishes wedges a group behind an EXCLUSIVE transaction, times it
   out, and proves the slot is still held (a second poll on a cap of 1
   times out on the gate without ever reaching the factory) and is handed
   back only once the abandoned query truly finishes. Moving the release
   from the work to the waiter fails this test and no other.

3. Task.Run thread-pool starvation: documented, not changed.
   Microsoft.Data.SqlClient's async path parks no pool thread on a frozen
   server, so the driver cannot starve itself; LongRunning would cost a
   dedicated OS thread per group per poll forever to insure against a
   provider this driver does not ship. RunGroupAsync now records the
   decision and what a BadTimeout does and does not mean, for the
   blackhole gate.

Also: the timeout warning now names the group's table, so a BadTimeout
storm identifies which source is frozen.

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
2026-07-24 14:57:14 -04:00
parent 861a1d1df0
commit e3e3f5fb04
2 changed files with 254 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlPollReaderTests.cs
+179 -8
View File
@@ -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);
/// <summary>
/// Adds a second <see cref="SqlitePollFixture.WideRowTable"/> row for the station the wide-row tests
/// select, breaking that model's one-row-per-selector contract.
/// </summary>
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')
"""));
/// <summary>Runs one statement on a caller-owned connection.</summary>
private static void Execute(SqliteConnection connection, string sql)
{
using var command = connection.CreateCommand();
command.CommandText = sql;
command.ExecuteNonQuery();
}
/// <summary>
/// Polls <paramref name="condition"/> until it holds or <paramref name="limit"/> elapses. Bounded on
/// purpose: a test that asserts an abandoned task eventually finishes must fail, not hang, when it
/// does not.
/// </summary>
private static async Task<bool> WaitUntilAsync(Func<bool> condition, TimeSpan limit)
{
var clock = Stopwatch.StartNew();
while (clock.Elapsed < limit)
{
if (condition()) return true;
await Task.Delay(25);
}
return condition();
}
/// <summary>The driver's RawPath→definition table, standing in for <c>EquipmentTagRefResolver</c>.</summary>
private static Func<string, SqlTagDefinition?> Resolver(params SqlTagDefinition[] tags)
{
@@ -432,6 +574,35 @@ public sealed class SqlPollReaderTests
RowSelectorValue: selectorValue,
RowSelectorTopByTimestamp: topByTimestamp);
/// <summary>
/// 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.
/// </summary>
private sealed class RecordingLogger : ILogger
{
/// <summary>Every record written, level + rendered message.</summary>
public List<(LogLevel Level, string Message)> Entries { get; } = [];
public IDisposable BeginScope<TState>(TState state) where TState : notnull => Scope.Instance;
public bool IsEnabled(LogLevel logLevel) => true;
public void Log<TState>(
LogLevel logLevel, EventId eventId, TState state, Exception? exception,
Func<TState, Exception?, string> formatter)
{
lock (Entries) Entries.Add((logLevel, formatter(state, exception)));
}
private sealed class Scope : IDisposable
{
public static Scope Instance { get; } = new();
public void Dispose() { }
}
}
/// <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