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

merge: SQL driver health review findings resolved (Driver.Sql-004/-005/-006/-008/-010)

Browse Source 
# Conflicts:
#	docs/drivers/Sql.md
This commit is contained in:
Joseph Doherty
2026-08-10 08:11:40 -04:00
parent ead8a97511 bb83099300
commit b8acabfa5f
5 changed files with 497 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlDriverTests.cs
+288 -9
View File
@@ -224,6 +224,33 @@ public sealed class SqlDriverTests
}
}
[Fact]
public async Task ReadAsync_whenTheQueryFailsAfterTheConnectionOpened_degradesButLeavesTheEndpointUp()
{
// Driver.Sql-005. BadCommunicationError is minted by GroupResult.Failed, which the reader raises only
// AFTER the connection opened ("the connection opened, so the server is up — this query is what
// failed"). Classifying it as connection-class reported the whole database endpoint Stopped over one
// group failing on a dropped table, a revoked SELECT or an over-long IN list, sending the operator at
// the network instead of at the schema. The driver must still degrade — every tag in that group is
// Bad — but the endpoint is demonstrably up.
using var fixture = new SqlitePollFixture();
await using var driver = NewDriver(fixture, KvEntry("Speed", SqlitePollFixture.PresentKey));
await driver.InitializeAsync(ConfigJson, CancellationToken.None);
driver.GetHostStatuses()[0].State.ShouldBe(HostState.Running);
// The schema drifts out from under a driver whose catalog gate already passed.
fixture.Execute($"DROP TABLE {SqlitePollFixture.KeyValueTable}");
var snapshots = await driver.ReadAsync(["Speed"], CancellationToken.None);
snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.BadCommunicationError);
var health = driver.GetHealth();
health.State.ShouldBe(DriverState.Degraded);
health.LastError.ShouldNotBeNull();
health.LastError.ShouldContain("quer"); // names the query, not the network
driver.GetHostStatuses()[0].State.ShouldBe(HostState.Running); // RED pre-fix: Stopped
}
[Fact]
public async Task ReadAsync_anUnresolvableRef_isNotAConnectivityFact_soHealthAndHostStand()
{
@@ -242,6 +269,147 @@ public sealed class SqlDriverTests
driver.GetHostStatuses()[0].State.ShouldBe(HostState.Running);
}
// ---- the comms-loss watchdog's evidence clock (Driver.Sql-006) ----
[Fact]
public async Task ReadAsync_whenTheDatabaseAnswersWithNoRow_advancesTheEvidenceClock_withoutChangingHealth()
{
// Driver.Sql-006. LastSuccessfulRead is what DriverInstanceActor's comms-loss watchdog reads: a
// non-Healthy driver whose clock has been frozen for 90 s is force-reconnected (Gitea #527), on the
// premise that the clock is the honest evidence the transport works. BadNoData IS that evidence —
// the query ran and the row is absent — so a driver whose key set was deleted must not be
// reconnected every 90 s against a database answering every poll on time. The health STATE stays
// exactly as it was: an absent row is a data fact, not a connectivity one.
using var fixture = new SqlitePollFixture();
await using var driver = NewDriver(fixture, KvEntry("Missing", SqlitePollFixture.AbsentKey));
await driver.InitializeAsync(ConfigJson, CancellationToken.None);
// Something has degraded the driver — the watchdog's first conjunct. Only then does the frozen clock
// matter, which is why the scenario needs both.
driver.HandlePollError(new InvalidOperationException("Reader contract violation: expected 1 snapshots"));
var before = driver.GetHealth().LastSuccessfulRead.ShouldNotBeNull();
await Task.Delay(20, TestContext.Current.CancellationToken);
var snapshots = await driver.ReadAsync(["Missing"], CancellationToken.None);
snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.BadNoData);
var health = driver.GetHealth();
health.LastSuccessfulRead.ShouldNotBeNull().ShouldBeGreaterThan(before); // RED pre-fix: frozen
health.State.ShouldBe(DriverState.Degraded); // classification untouched
}
[Fact]
public async Task ReadAsync_anUnresolvableRef_isNoEvidenceTheDatabaseAnswered_soTheEvidenceClockStands()
{
// The falsifiability control: "advance on anything that is not Good" would pass the test above and
// break the watchdog outright. BadNodeIdUnknown is minted by the resolver without any database
// contact at all, so it proves nothing about the transport and must leave the clock alone —
// otherwise a driver polling only typoed refs would look forever alive to the watchdog.
using var fixture = new SqlitePollFixture();
await using var driver = NewDriver(fixture, KvEntry("Speed", SqlitePollFixture.PresentKey));
await driver.InitializeAsync(ConfigJson, CancellationToken.None);
var before = driver.GetHealth().LastSuccessfulRead.ShouldNotBeNull();
await Task.Delay(20, TestContext.Current.CancellationToken);
var snapshots = await driver.ReadAsync(["TypoedTagName"], CancellationToken.None);
snapshots[0].StatusCode.ShouldBe(SqlStatusCodes.BadNodeIdUnknown);
driver.GetHealth().LastSuccessfulRead.ShouldBe(before);
}
[Theory]
[InlineData(SqlStatusCodes.Good)] // the row is there and readable
[InlineData(SqlStatusCodes.BadNoData)] // the query ran; the row is absent
[InlineData(SqlStatusCodes.Uncertain)] // a present row with a NULL cell (nullIsBad=false)
[InlineData(SqlStatusCodes.Bad)] // …the same row under nullIsBad=true
[InlineData(SqlStatusCodes.BadTypeMismatch)] // a cell came back and would not coerce
[InlineData(SqlStatusCodes.BadCommunicationError)] // the connection opened; the query was refused
public async Task ObservePollOutcome_everyEvidenceOfAnswerCode_advancesTheEvidenceClock(uint statusCode)
{
using var fixture = new SqlitePollFixture();
await using var driver = NewDriver(fixture, KvEntry("Speed", SqlitePollFixture.PresentKey));
await driver.InitializeAsync(ConfigJson, CancellationToken.None);
var before = driver.GetHealth().LastSuccessfulRead.ShouldNotBeNull();
await Task.Delay(20, TestContext.Current.CancellationToken);
driver.ObservePollOutcome([new DataValueSnapshot(null, statusCode, DateTime.UtcNow, DateTime.UtcNow)]);
driver.GetHealth().LastSuccessfulRead.ShouldNotBeNull().ShouldBeGreaterThan(before);
}
[Theory]
[InlineData(SqlStatusCodes.BadTimeout)] // the database did not answer at all
[InlineData(SqlStatusCodes.BadNodeIdUnknown)] // the resolver never reached the database
[InlineData(SqlStatusCodes.BadConfigurationError)] // the planner rejected the tag; nothing was executed
[InlineData(SqlStatusCodes.BadInternalError)] // a reader defect; no statement about the database
public async Task ObservePollOutcome_aCodeThatIsNoEvidenceOfAnAnswer_leavesTheEvidenceClockAlone(
uint statusCode)
{
using var fixture = new SqlitePollFixture();
await using var driver = NewDriver(fixture, KvEntry("Speed", SqlitePollFixture.PresentKey));
await driver.InitializeAsync(ConfigJson, CancellationToken.None);
var before = driver.GetHealth().LastSuccessfulRead.ShouldNotBeNull();
await Task.Delay(20, TestContext.Current.CancellationToken);
driver.ObservePollOutcome([new DataValueSnapshot(null, statusCode, DateTime.UtcNow, DateTime.UtcNow)]);
driver.GetHealth().LastSuccessfulRead.ShouldBe(before);
}
// ---- Driver.Sql-010: the health field is read-modify-written from several poll loops ----
[Fact]
public async Task HealthTransitions_fromConcurrentPolls_neverRewindTheEvidenceClock()
{
// Driver.Sql-010. Every health transition is a read-modify-write of one field: Degrade reads the
// current snapshot to PRESERVE LastSuccessfulRead, then writes a new record. With a plain
// Volatile read + write, a Degrade that read the clock at t0 can be published after a concurrent
// advance to t1 — so the clock goes BACKWARDS, and the advance is lost. That is not cosmetic: the
// comms-loss watchdog measures how long the clock has been frozen, so a lost advance is a spurious
// 90 s reconnect. Monotonicity is the observable the lost update violates, and it is exactly what a
// compare-and-swap guarantees.
//
// A driver with several subscriptions runs one poll loop each, plus any ad-hoc ReadAsync — so two
// classifications landing at once is the ordinary case, not a contrived one.
using var fixture = new SqlitePollFixture();
await using var driver = NewDriver(fixture, KvEntry("Speed", SqlitePollFixture.PresentKey));
await driver.InitializeAsync(ConfigJson, CancellationToken.None);
const int iterations = 30_000;
var advancing = new DataValueSnapshot(1.0, SqlStatusCodes.Good, DateTime.UtcNow, DateTime.UtcNow);
var degrading = new DataValueSnapshot(null, SqlStatusCodes.BadTimeout, DateTime.UtcNow, DateTime.UtcNow);
using var done = new CancellationTokenSource();
var rewinds = 0;
var token = TestContext.Current.CancellationToken;
var observer = Task.Run(
() =>
{
var highWater = driver.GetHealth().LastSuccessfulRead ?? DateTime.MinValue;
while (!done.IsCancellationRequested)
{
var seen = driver.GetHealth().LastSuccessfulRead ?? DateTime.MinValue;
if (seen < highWater) Interlocked.Increment(ref rewinds);
else highWater = seen;
}
},
token);
var writers = new[]
{
Task.Run(
() => { for (var i = 0; i < iterations; i++) driver.ObservePollOutcome([advancing]); }, token),
Task.Run(
() => { for (var i = 0; i < iterations; i++) driver.ObservePollOutcome([degrading]); }, token),
};
await Task.WhenAll(writers);
await done.CancelAsync();
await observer;
Volatile.Read(ref rewinds).ShouldBe(0);
}
// ---- ISubscribable (poll-engine overlay) ----
[Fact]
@@ -391,23 +559,92 @@ public sealed class SqlDriverTests
{
// On a subscribed-read outage the reader throws a DbException; ReadAsync classifies health with a
// good, endpoint-bearing message + host Stopped and rethrows the SAME exception, which the poll
// engine then routes to HandlePollError. HandlePollError must ignore the DbException class ReadAsync
// owns — pre-fix it re-degraded with the bare ex.Message (the WORSE of the two LastErrors, and a
// credential-leak path) and logged the outage a second time.
// engine then routes to HandlePollError. HandlePollError must leave the classification ReadAsync
// already made alone — pre-fix it re-degraded with the bare ex.Message (the WORSE of the two
// LastErrors, and a credential-leak path) and logged the outage a second time.
// Driven through the REAL round trip rather than by handing HandlePollError a bare exception: the
// deferral is owned by ReadAsync marking the throw, not by the exception's type (Driver.Sql-004).
using var fixture = new SqlitePollFixture();
var logger = new CapturingLogger();
await using var driver = NewDriver(fixture, logger, KvEntry("Speed", SqlitePollFixture.PresentKey));
var factory = new BreakableFactory(fixture.Factory);
await using var driver = NewDriver(
fixture, fixture.ConnectionString, logger, factory, KvEntry("Speed", SqlitePollFixture.PresentKey));
await driver.InitializeAsync(ConfigJson, CancellationToken.None);
driver.GetHealth().State.ShouldBe(DriverState.Healthy);
factory.BreakByThrowingOnOpen("connection failed; Password=" + SecretToken);
var thrown = await Should.ThrowAsync<DbException>(
async () => await driver.ReadAsync(["Speed"], CancellationToken.None));
var classified = driver.GetHealth();
classified.State.ShouldBe(DriverState.Degraded);
classified.LastError.ShouldNotBeNull();
classified.LastError.ShouldContain("could not reach");
var warningsAfterRead = logger.Entries.Count(e => e.Level == LogLevel.Warning);
driver.HandlePollError(thrown);
var health = driver.GetHealth();
// Untouched: HandlePollError left the exception to ReadAsync, so it neither re-degraded nor leaked…
health.LastError.ShouldNotBeNull().ShouldBe(classified.LastError);
health.LastError.ShouldNotContain(SecretToken);
// …and it logged nothing (ReadAsync owns the single outage warning).
logger.Entries.Count(e => e.Level == LogLevel.Warning).ShouldBe(warningsAfterRead);
}
[Fact]
public async Task HandlePollError_forANonDbExceptionReadAsyncClassified_stillDefers_soTheGoodLastErrorSurvives()
{
// Driver.Sql-004. Not every unreachable-database throw is a DbException: the reader raises a plain
// InvalidOperationException when the provider factory hands back no connection, and a provider can
// raise ArgumentException/FormatException off a malformed connection string on the same open path.
// ReadAsync classifies those exactly as it classifies a DbException — endpoint-bearing LastError,
// host Stopped — and rethrows, so HandlePollError must defer to it here too. RED against the
// `ex is DbException` type test, which re-degraded with the guidance-free "poll failed: TypeName"
// and logged the outage twice.
using var fixture = new SqlitePollFixture();
var logger = new CapturingLogger();
var factory = new BreakableFactory(fixture.Factory);
await using var driver = NewDriver(
fixture, fixture.ConnectionString, logger, factory, KvEntry("Speed", SqlitePollFixture.PresentKey));
await driver.InitializeAsync(ConfigJson, CancellationToken.None);
factory.BreakByReturningNoConnection();
var thrown = await Should.ThrowAsync<InvalidOperationException>(
async () => await driver.ReadAsync(["Speed"], CancellationToken.None));
var classified = driver.GetHealth();
classified.State.ShouldBe(DriverState.Degraded);
classified.LastError.ShouldNotBeNull();
classified.LastError.ShouldContain("could not reach");
driver.GetHostStatuses()[0].State.ShouldBe(HostState.Stopped);
var warningsAfterRead = logger.Entries.Count(e => e.Level == LogLevel.Warning);
driver.HandlePollError(thrown);
var health = driver.GetHealth();
health.LastError.ShouldNotBeNull()
.ShouldBe(classified.LastError); // the richer message survived…
health.LastError.ShouldNotContain("poll failed"); // …and was not overwritten by the poor one
logger.Entries.Count(e => e.Level == LogLevel.Warning).ShouldBe(warningsAfterRead);
}
[Fact]
public async Task HandlePollError_forAnExceptionReadAsyncNeverSaw_degrades_evenWhenItIsADbException()
{
// The falsifiability control for the two deferrals above: the marker is per-exception, not per-type,
// so a DbException that never went through ReadAsync's classification is NOT owned by it and must
// still reach the health surface. A type test would silently swallow this one.
using var fixture = new SqlitePollFixture();
var driver = NewDriver(fixture, KvEntry("Speed", SqlitePollFixture.PresentKey));
driver.HandlePollError(new FakeDbException("connection failed; Password=" + SecretToken));
var health = driver.GetHealth();
// Untouched: HandlePollError left the DbException to ReadAsync, so it neither degraded nor leaked…
health.State.ShouldBe(DriverState.Healthy);
health.LastError.ShouldBeNull();
// …and it logged nothing (ReadAsync owns the single outage warning).
logger.Entries.Count(e => e.Level == LogLevel.Warning).ShouldBe(0);
health.State.ShouldBe(DriverState.Degraded);
health.LastError.ShouldNotBeNull();
health.LastError.ShouldNotContain(SecretToken); // type only, never the message text
health.LastError.ShouldContain(nameof(FakeDbException));
}
[Fact]
@@ -518,6 +755,26 @@ public sealed class SqlDriverTests
operationTimeout: TimeSpan.FromSeconds(15), commandTimeout: TimeSpan.FromSeconds(10),
rawTags: rawTags);
/// <summary>The same driver, over a caller-supplied provider factory — the outage-injection seam.</summary>
private static SqlDriver NewDriver(
SqlitePollFixture fixture,
string connectionString,
CapturingLogger logger,
DbProviderFactory factory,
params RawTagEntry[] rawTags)
=> new(
new SqlDriverOptions
{
RawTags = rawTags,
OperationTimeout = TimeSpan.FromSeconds(15),
CommandTimeout = TimeSpan.FromSeconds(10),
},
DriverInstanceId,
new SqliteDialect(),
connectionString,
factory: factory,
logger: logger);
/// <summary>
/// Builds the driver through its production constructor. That constructor <b>is</b> the injection
/// seam — dialect, provider factory and already-resolved connection string are all parameters — so no
@@ -668,6 +925,28 @@ public sealed class SqlDriverTests
public override DbConnection CreateConnection() => new ThrowingConnection(openFailureMessage);
}
/// <summary>
/// A provider factory that behaves exactly like the fixture's until it is broken, then reproduces one
/// of the two open-path outages the driver has to classify: a factory that returns <b>no</b>
/// connection (a plain <see cref="InvalidOperationException"/> from the reader — NOT a
/// <see cref="DbException"/>), or one whose connections throw on open.
/// </summary>
private sealed class BreakableFactory(DbProviderFactory inner) : DbProviderFactory
{
private Func<DbConnection?>? _broken;
/// <summary>Hands back no connection, the shape that makes the reader raise InvalidOperationException.</summary>
public void BreakByReturningNoConnection() => Volatile.Write(ref _broken, static () => null);
/// <summary>Hands back a connection that fails to open with <paramref name="message"/>.</summary>
/// <param name="message">The provider message, which may carry credentials.</param>
public void BreakByThrowingOnOpen(string message)
=> Volatile.Write(ref _broken, () => new ThrowingConnection(message));
public override DbConnection? CreateConnection()
=> Volatile.Read(ref _broken) is { } broken ? broken() : inner.CreateConnection();
}
/// <summary>A connection that throws <see cref="FakeDbException"/> on open; every other member is inert.</summary>
private sealed class ThrowingConnection(string openFailureMessage) : DbConnection
{