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

feat(sql): IBatchSubscribable and lifecycle-generation-stamped health (R02:P1, R06:S-9)

Browse Source 
R02:P1: SqlDriver runs PollGroupEngine, so it now implements IBatchSubscribable and
raises one OnDataChangeBatch per poll burst (DriverInstanceActor forwards one message
per cycle instead of one per tag).
S-9: InitializeAsync/ShutdownAsync take a lifecycle generation; their health writes
apply only while current, and a poll that started under an older generation (one that
outlived the engine's teardown wait) no longer writes its verdict onto the new
configuration. Poll-driven updates can no longer end Initializing - only the
initialize's own final write does. Lifecycle writes now preserve LastSuccessfulRead.

Refs #744
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 16:03:23 -04:00
parent ab29b76be0
commit a7d21d5a65
2 changed files with 179 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriver.cs
+75 -21
View File
@@ -28,7 +28,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Sql;
/// manufacture an exception to force backoff.</para>
/// </summary>
public sealed class SqlDriver
: IDriver, ITagDiscovery, IReadable, ISubscribable, IHostConnectivityProbe, ISkippedTagReporter, IAsyncDisposable
: IDriver, ITagDiscovery, IReadable, IBatchSubscribable, IHostConnectivityProbe, ISkippedTagReporter, IAsyncDisposable
{
/// <summary>Authored tags this driver could not build into its poll table.</summary>
private readonly SkippedTagLedger _skippedTags = new();
@@ -156,6 +156,16 @@ public sealed class SqlDriver
private DriverHealth _health = new(DriverState.Unknown, null, null);
/// <summary>Guards the lifecycle-generation bump against the generation-checked lifecycle health write.</summary>
private readonly object _healthLock = new();
/// <summary>
/// Bumped by every <see cref="InitializeAsync"/> and <see cref="ShutdownAsync"/>. A lifecycle write from
/// a superseded call, and a poll that STARTED under an older generation (a stale poll outliving the
/// engine's 5 s teardown wait, R06:S-9), can no longer overwrite the newer state.
/// </summary>
private long _lifecycleGeneration;
/// <summary>
/// The exception instance <see cref="ReadAsync"/> has already turned into a health verdict, held only
/// until <see cref="HandlePollError"/> — which the poll engine calls with that same object — has
@@ -173,6 +183,9 @@ public sealed class SqlDriver
/// <summary>Occurs when a subscribed tag's value or quality changes.</summary>
public event EventHandler<DataChangeEventArgs>? OnDataChange;
/// <summary>Occurs once per poll cycle that produced at least one change, carrying every change of that cycle.</summary>
public event EventHandler<DataChangeBatchEventArgs>? OnDataChangeBatch;
/// <summary>Occurs when the database endpoint transitions between reachable and unreachable.</summary>
public event EventHandler<HostStatusChangedEventArgs>? OnHostStatusChanged;
@@ -230,6 +243,9 @@ public sealed class SqlDriver
OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, tagRef, snapshot)),
onError: HandlePollError,
backoffCap: PollBackoffCap,
// R02:P1 — one batch per poll burst, so DriverInstanceActor forwards one message per cycle instead
// of one per tag (it subscribes exactly one of the two events).
onBatch: (handle, items) => OnDataChangeBatch?.Invoke(this, new DataChangeBatchEventArgs(handle, items)),
// The engine owns no Meter (Core.Abstractions takes no metrics dependency), so this
// closure IS the production consumer of its overrun signal.
onOverrun: _ => OtOpcUaTelemetry.PollOverruns.Add(1,
@@ -352,7 +368,9 @@ public sealed class SqlDriver
// identifier.
public async Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
WriteHealth(new DriverHealth(DriverState.Initializing, null, null));
var generation = BeginLifecycle();
// LastSuccessfulRead is PRESERVED across the lifecycle states — it is the watchdog's evidence clock.
WriteHealthIfCurrent(generation, new DriverHealth(DriverState.Initializing, ReadHealth().LastSuccessfulRead, null));
// adopt a changed config BEFORE the tag table is built, so the table and the connection it
// will be polled over come from the same revision. ApplyBinding takes the options, dialect and
@@ -374,7 +392,7 @@ public sealed class SqlDriver
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
// The caller tore the initialization down; not a database verdict.
WriteHealth(new DriverHealth(DriverState.Unknown, null, null));
WriteHealthIfCurrent(generation, new DriverHealth(DriverState.Unknown, ReadHealth().LastSuccessfulRead, null));
throw;
}
catch (Exception ex)
@@ -390,8 +408,8 @@ public sealed class SqlDriver
$"Check the database is reachable and the connection string is valid.";
_logger.LogError(ex, "Sql driver {DriverInstanceId} liveness check against {Endpoint} failed.",
_driverInstanceId, Endpoint);
WriteHealth(new DriverHealth(DriverState.Faulted, null, message));
TransitionHostTo(HostState.Stopped);
if (WriteHealthIfCurrent(generation, new DriverHealth(DriverState.Faulted, ReadHealth().LastSuccessfulRead, message)))
TransitionHostTo(HostState.Stopped);
throw new InvalidOperationException(message, ex);
}
@@ -404,7 +422,7 @@ public sealed class SqlDriver
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
WriteHealth(new DriverHealth(DriverState.Unknown, null, null));
WriteHealthIfCurrent(generation, new DriverHealth(DriverState.Unknown, ReadHealth().LastSuccessfulRead, null));
throw;
}
catch (Exception ex)
@@ -420,13 +438,13 @@ public sealed class SqlDriver
_logger.LogError(ex,
"Sql driver {DriverInstanceId} catalog validation against {Endpoint} failed.",
_driverInstanceId, Endpoint);
WriteHealth(new DriverHealth(DriverState.Faulted, null, message));
TransitionHostTo(HostState.Stopped);
if (WriteHealthIfCurrent(generation, new DriverHealth(DriverState.Faulted, ReadHealth().LastSuccessfulRead, message)))
TransitionHostTo(HostState.Stopped);
throw new InvalidOperationException(message, ex);
}
WriteHealth(new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null));
TransitionHostTo(HostState.Running);
if (WriteHealthIfCurrent(generation, new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null)))
TransitionHostTo(HostState.Running);
_logger.LogInformation(
"Sql driver {DriverInstanceId} initialized against {Endpoint} with {TagCount} authored tag(s).",
_driverInstanceId, Endpoint, Tags.Count);
@@ -442,9 +460,10 @@ public sealed class SqlDriver
/// <inheritdoc />
public async Task ShutdownAsync(CancellationToken cancellationToken)
{
var generation = BeginLifecycle();
var lastRead = ReadHealth().LastSuccessfulRead;
await TeardownAsync().ConfigureAwait(false);
WriteHealth(new DriverHealth(DriverState.Unknown, lastRead, null));
WriteHealthIfCurrent(generation, new DriverHealth(DriverState.Unknown, lastRead, null));
}
/// <inheritdoc />
@@ -517,10 +536,13 @@ public sealed class SqlDriver
// unreachable" and degrade a perfectly healthy driver.
ArgumentNullException.ThrowIfNull(fullReferences);
// The generation this read started under: if a Shutdown/Initialize lands while it is in flight, its
// verdict describes a superseded configuration and must not touch the newer health (R06:S-9).
var generation = Interlocked.Read(ref _lifecycleGeneration);
try
{
var result = await _reader.ReadDetailedAsync(fullReferences, cancellationToken).ConfigureAwait(false);
ObservePollOutcome(result.Snapshots, result.RowCapExceededGroups);
ObservePollOutcome(result.Snapshots, result.RowCapExceededGroups, generation);
return result.Snapshots;
}
catch (OperationCanceledException)
@@ -535,8 +557,12 @@ public sealed class SqlDriver
_driverInstanceId, Endpoint);
// Type name only — never ex.Message (see InitializeAsync for why); the full exception is already
// on the log sink via the LogWarning exception parameter above.
Degrade($"Sql driver '{_driverInstanceId}' could not reach {Endpoint}: {ex.GetType().Name}.");
TransitionHostTo(HostState.Stopped);
if (IsCurrentLifecycle(generation))
{
Degrade($"Sql driver '{_driverInstanceId}' could not reach {Endpoint}: {ex.GetType().Name}.");
TransitionHostTo(HostState.Stopped);
}
// Mark THIS exception object as already classified before it goes back to the poll engine, which
// routes it to HandlePollError next. The mark is what makes the deferral there exact — see that
// method for why inferring it from the exception's type was not.
@@ -614,9 +640,15 @@ public sealed class SqlDriver
/// (the planner mints it before any I/O), so the count is what makes them count as the database
/// answering.
/// </param>
internal void ObservePollOutcome(IReadOnlyList<DataValueSnapshot> snapshots, int answeredRefusals = 0)
/// <param name="generation">
/// The lifecycle generation the poll started under; a superseded one is ignored outright. Null (tests)
/// means "current".
/// </param>
internal void ObservePollOutcome(
IReadOnlyList<DataValueSnapshot> snapshots, int answeredRefusals = 0, long? generation = null)
{
if (snapshots.Count == 0) return;
if (generation is { } started && !IsCurrentLifecycle(started)) return;
var good = 0;
var unreachable = 0;
@@ -800,7 +832,11 @@ public sealed class SqlDriver
/// </summary>
/// <param name="project">Builds the snapshot to publish from the current one.</param>
private void UpdateHealthUnlessFaulted(Func<DriverHealth, DriverHealth> project)
=> UpdateHealth(current => current.State == DriverState.Faulted ? null : project(current));
// Initializing too: only the initialize's own final write may end it — a poll landing mid-init
// (a re-init without shutdown keeps the engine running) used to flip Initializing → Healthy.
=> UpdateHealth(current => current.State is DriverState.Faulted or DriverState.Initializing
? null
: project(current));
/// <summary>
/// The one read-modify-write of the health field, as a compare-and-swap retry loop.
@@ -831,13 +867,31 @@ public sealed class SqlDriver
/// <summary>Barrier-protected read of the multi-threaded health field.</summary>
private DriverHealth ReadHealth() => Volatile.Read(ref _health);
/// <summary>Starts a lifecycle operation and returns its generation; every later one supersedes it.</summary>
private long BeginLifecycle()
{
lock (_healthLock) return ++_lifecycleGeneration;
}
/// <summary>Whether <paramref name="generation"/> is still the newest lifecycle operation.</summary>
private bool IsCurrentLifecycle(long generation) => Interlocked.Read(ref _lifecycleGeneration) == generation;
/// <summary>The current lifecycle generation. Test observation of the S-9 guard.</summary>
internal long LifecycleGenerationForTests => Interlocked.Read(ref _lifecycleGeneration);
/// <summary>
/// Unconditional publish of a new health snapshot, for the lifecycle transitions that are
/// last-writer-wins by definition (Initialize / Shutdown). Everything driven by a poll goes through
/// <see cref="UpdateHealth"/> instead, because it has to read the current snapshot first.
/// A lifecycle write that applies only while <paramref name="generation"/> is current — atomic against
/// <see cref="BeginLifecycle"/>, so a superseded initialize/shutdown never overwrites a newer one.
/// </summary>
/// <param name="value">The health snapshot to publish.</param>
private void WriteHealth(DriverHealth value) => Volatile.Write(ref _health, value);
private bool WriteHealthIfCurrent(long generation, DriverHealth value)
{
lock (_healthLock)
{
if (_lifecycleGeneration != generation) return false;
Volatile.Write(ref _health, value);
return true;
}
}
/// <summary>Records a host-state change and raises <see cref="OnHostStatusChanged"/> — on transitions only.</summary>
private void TransitionHostTo(HostState newState)
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlDriverTests.cs
+104
View File
@@ -385,6 +385,85 @@ public sealed class SqlDriverTests
driver.GetHealth().LastSuccessfulRead.ShouldNotBeNull().ShouldBeGreaterThan(before);
}
/// <summary>
/// <b>R06:S-9 / lifecycle generations.</b> A poll that started before a Shutdown → Initialize cycle
/// (it can outlive the engine's 5 s teardown wait) must not write its verdict onto the NEW
/// configuration's health. Control: the same BadTimeout verdict under the current generation degrades.
/// </summary>
[Fact]
public async Task ObservePollOutcome_fromASupersededGeneration_isIgnored()
{
using var fixture = new SqlitePollFixture();
await using var driver = NewDriver(fixture, KvEntry("Speed", SqlitePollFixture.PresentKey));
await driver.InitializeAsync(ConfigJson, CancellationToken.None);
var stale = driver.LifecycleGenerationForTests;
await driver.ShutdownAsync(CancellationToken.None);
await driver.InitializeAsync(ConfigJson, CancellationToken.None);
driver.GetHealth().State.ShouldBe(DriverState.Healthy);
var timedOut = new DataValueSnapshot(null, SqlStatusCodes.BadTimeout, null, DateTime.UtcNow);
driver.ObservePollOutcome([timedOut], generation: stale);
driver.GetHealth().State.ShouldBe(DriverState.Healthy, "a superseded poll must not degrade the new init");
driver.ObservePollOutcome([timedOut], generation: driver.LifecycleGenerationForTests);
driver.GetHealth().State.ShouldBe(DriverState.Degraded);
}
/// <summary>
/// Only the initialize's own final write ends <c>Initializing</c>: a Good poll landing mid-init used to
/// flip it to Healthy before liveness had even answered. Control: once the init completes it is Healthy.
/// </summary>
[Fact]
public async Task APollLandingMidInitialize_cannotEndInitializing()
{
using var fixture = new SqlitePollFixture();
var factory = new GatedFactory();
await using var driver = NewDriver(fixture, fixture.ConnectionString, CapturingLogger.Null, factory,
KvEntry("Speed", SqlitePollFixture.PresentKey));
factory.Arm();
var init = driver.InitializeAsync(ConfigJson, CancellationToken.None);
(await factory.Entered.Task.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken)).ShouldBeTrue();
driver.GetHealth().State.ShouldBe(DriverState.Initializing);
driver.ObservePollOutcome([new DataValueSnapshot(42.0, SqlStatusCodes.Good, DateTime.UtcNow, DateTime.UtcNow)]);
driver.GetHealth().State.ShouldBe(DriverState.Initializing);
factory.Release();
await init.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken);
driver.GetHealth().State.ShouldBe(DriverState.Healthy);
}
/// <summary>A SQLite factory whose next connection creation blocks until released — parks an initialize.</summary>
private sealed class GatedFactory : DbProviderFactory
{
private readonly ManualResetEventSlim _release = new(initialState: true);
private volatile bool _armed;
public TaskCompletionSource<bool> Entered { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
public void Arm()
{
_release.Reset();
_armed = true;
}
public void Release() => _release.Set();
public override DbConnection CreateConnection()
{
if (_armed)
{
_armed = false;
Entered.TrySetResult(true);
_release.Wait(TimeSpan.FromSeconds(30));
}
return Microsoft.Data.Sqlite.SqliteFactory.Instance.CreateConnection()!;
}
}
// ---- Driver.Sql-010: the health field is read-modify-written from several poll loops ----
[Fact]
@@ -474,6 +553,31 @@ public sealed class SqlDriverTests
Volatile.Read(ref changes).ShouldBe(afterUnsubscribe);
}
/// <summary>
/// <b>R02:P1.</b> The driver runs <c>PollGroupEngine</c>, so it raises one batch per poll burst:
/// <c>DriverInstanceActor</c> then forwards one message per cycle instead of one per tag. Both tags of
/// the first (force-raised) poll arrive in the SAME batch, on the subscription's handle.
/// </summary>
[Fact]
public async Task Subscribe_raisesOneBatchPerPollBurst()
{
using var fixture = new SqlitePollFixture();
await using var driver = NewDriver(fixture,
KvEntry("Speed", SqlitePollFixture.PresentKey), KvEntry("Pressure", SqlitePollFixture.SecondPresentKey));
await driver.InitializeAsync(ConfigJson, CancellationToken.None);
var batchable = driver.ShouldBeAssignableTo<IBatchSubscribable>();
var first = new TaskCompletionSource<DataChangeBatchEventArgs>(TaskCreationOptions.RunContinuationsAsynchronously);
batchable!.OnDataChangeBatch += (_, args) => first.TrySetResult(args);
var handle = await driver.SubscribeAsync(["Speed", "Pressure"], TimeSpan.FromMilliseconds(100), CancellationToken.None);
var batch = await first.Task.WaitAsync(TimeSpan.FromSeconds(10), TestContext.Current.CancellationToken);
batch.SubscriptionHandle.ShouldBe(handle);
batch.Items.Select(i => i.FullReference).OrderBy(r => r, StringComparer.Ordinal).ShouldBe(["Pressure", "Speed"]);
batch.Items.Single(i => i.FullReference == "Speed").Snapshot.Value.ShouldBe(SqlitePollFixture.PresentValue);
}
// ---- disposal ----
[Fact]