From a7d21d5a659fc905c612086192a1e27f4513cc5c Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 16:03:23 -0400 Subject: [PATCH] feat(sql): IBatchSubscribable and lifecycle-generation-stamped health (R02:P1, R06:S-9) 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 --- .../ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriver.cs | 96 ++++++++++++---- .../SqlDriverTests.cs | 104 ++++++++++++++++++ 2 files changed, 179 insertions(+), 21 deletions(-) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriver.cs index 4a8936c9..5492c293 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriver.cs @@ -28,7 +28,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Sql; /// manufacture an exception to force backoff. /// public sealed class SqlDriver - : IDriver, ITagDiscovery, IReadable, ISubscribable, IHostConnectivityProbe, ISkippedTagReporter, IAsyncDisposable + : IDriver, ITagDiscovery, IReadable, IBatchSubscribable, IHostConnectivityProbe, ISkippedTagReporter, IAsyncDisposable { /// Authored tags this driver could not build into its poll table. private readonly SkippedTagLedger _skippedTags = new(); @@ -156,6 +156,16 @@ public sealed class SqlDriver private DriverHealth _health = new(DriverState.Unknown, null, null); + /// Guards the lifecycle-generation bump against the generation-checked lifecycle health write. + private readonly object _healthLock = new(); + + /// + /// Bumped by every and . 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. + /// + private long _lifecycleGeneration; + /// /// The exception instance has already turned into a health verdict, held only /// until — which the poll engine calls with that same object — has @@ -173,6 +183,9 @@ public sealed class SqlDriver /// Occurs when a subscribed tag's value or quality changes. public event EventHandler? OnDataChange; + /// Occurs once per poll cycle that produced at least one change, carrying every change of that cycle. + public event EventHandler? OnDataChangeBatch; + /// Occurs when the database endpoint transitions between reachable and unreachable. public event EventHandler? 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 /// 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)); } /// @@ -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. /// - internal void ObservePollOutcome(IReadOnlyList snapshots, int answeredRefusals = 0) + /// + /// The lifecycle generation the poll started under; a superseded one is ignored outright. Null (tests) + /// means "current". + /// + internal void ObservePollOutcome( + IReadOnlyList 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 /// /// Builds the snapshot to publish from the current one. private void UpdateHealthUnlessFaulted(Func 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)); /// /// The one read-modify-write of the health field, as a compare-and-swap retry loop. @@ -831,13 +867,31 @@ public sealed class SqlDriver /// Barrier-protected read of the multi-threaded health field. private DriverHealth ReadHealth() => Volatile.Read(ref _health); + /// Starts a lifecycle operation and returns its generation; every later one supersedes it. + private long BeginLifecycle() + { + lock (_healthLock) return ++_lifecycleGeneration; + } + + /// Whether is still the newest lifecycle operation. + private bool IsCurrentLifecycle(long generation) => Interlocked.Read(ref _lifecycleGeneration) == generation; + + /// The current lifecycle generation. Test observation of the S-9 guard. + internal long LifecycleGenerationForTests => Interlocked.Read(ref _lifecycleGeneration); + /// - /// 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 - /// instead, because it has to read the current snapshot first. + /// A lifecycle write that applies only while is current — atomic against + /// , so a superseded initialize/shutdown never overwrites a newer one. /// - /// The health snapshot to publish. - 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; + } + } /// Records a host-state change and raises — on transitions only. private void TransitionHostTo(HostState newState) diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlDriverTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlDriverTests.cs index 6dd21ce9..320a0a0f 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlDriverTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlDriverTests.cs @@ -385,6 +385,85 @@ public sealed class SqlDriverTests driver.GetHealth().LastSuccessfulRead.ShouldNotBeNull().ShouldBeGreaterThan(before); } + /// + /// R06:S-9 / lifecycle generations. 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. + /// + [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); + } + + /// + /// Only the initialize's own final write ends Initializing: 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. + /// + [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); + } + + /// A SQLite factory whose next connection creation blocks until released — parks an initialize. + private sealed class GatedFactory : DbProviderFactory + { + private readonly ManualResetEventSlim _release = new(initialState: true); + private volatile bool _armed; + + public TaskCompletionSource 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); } + /// + /// R02:P1. The driver runs PollGroupEngine, so it raises one batch per poll burst: + /// DriverInstanceActor 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. + /// + [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(); + var first = new TaskCompletionSource(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]