From 336e79ea9acd99eb0afeb316e9ab9df554eb8cf2 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 10 Aug 2026 08:05:23 -0400 Subject: [PATCH] fix(driver.sql): compare-and-swap the health field so a concurrent transition cannot be lost (Driver.Sql-010) --- .../ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriver.cs | 72 ++++++++++++++----- .../SqlDriverTests.cs | 49 +++++++++++++ 2 files changed, 102 insertions(+), 19 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 935f337f..e0db65c8 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriver.cs @@ -686,7 +686,7 @@ public sealed class SqlDriver // ReinitializeAsync has recorded Faulted (the engine's ~5 s teardown wait is shorter than the 15 s // default operationTimeout, so a stale poll can still complete afterwards) must not flip Faulted back // to Healthy with no real recovery. Only a successful (re)initialize clears Faulted. - SetHealthUnlessFaulted(new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null)); + RecordHealthyRead(); TransitionHostTo(HostState.Running); } @@ -782,36 +782,70 @@ public sealed class SqlDriver /// /// The operator-facing reason, which must never carry the connection string. private void Degrade(string reason) - { - var current = ReadHealth(); - SetHealthUnlessFaulted(new DriverHealth(DriverState.Degraded, current.LastSuccessfulRead, reason)); - } + => UpdateHealthUnlessFaulted( + current => new DriverHealth(DriverState.Degraded, current.LastSuccessfulRead, reason)); /// - /// Publishes unless the driver is already - /// — a liveness/config verdict that only a successful (re)initialize may clear. Every non-Initialize - /// health write routes through here: and the poll's Healthy verdict in - /// , so a late in-flight poll cannot un-fault a driver a concurrent - /// just faulted. + /// Records a poll that returned at least one Good value: Healthy, with the evidence clock advanced. + /// The timestamp is taken inside the transform, so a retried compare-and-swap publishes + /// a fresh one rather than the one this call started with — which a concurrent advance could already + /// have overtaken. /// - /// The health snapshot to publish when the driver is not Faulted. - private void SetHealthUnlessFaulted(DriverHealth value) - { - if (ReadHealth().State == DriverState.Faulted) return; - WriteHealth(value); - } + private void RecordHealthyRead() + => UpdateHealthUnlessFaulted( + static _ => new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null)); /// Advances LastSuccessfulRead without changing the current state or error. private void TouchLastSuccessfulRead() + => UpdateHealth(static current => current with { LastSuccessfulRead = DateTime.UtcNow }); + + /// + /// Applies unless the driver is already + /// — a liveness/config verdict that only a successful (re)initialize may clear. Every non-Initialize + /// state write routes through here: and the poll's Healthy verdict in + /// , so a late in-flight poll cannot un-fault a driver a concurrent + /// just faulted. The guard is evaluated against the same + /// snapshot the write then swaps out, so a Faulted verdict landing mid-update cannot be lost. + /// + /// Builds the snapshot to publish from the current one. + private void UpdateHealthUnlessFaulted(Func project) + => UpdateHealth(current => current.State == DriverState.Faulted ? null : project(current)); + + /// + /// The one read-modify-write of the health field, as a compare-and-swap retry loop. + /// + /// + /// Driver.Sql-010. Health is transitioned from every poll loop (one per subscription) plus any + /// ad-hoc , and every transition needs the current snapshot to build the next + /// one — preserves LastSuccessfulRead, the Faulted guard reads the state. + /// A plain Volatile read followed by a Volatile write lets a concurrent update land in + /// between and be silently overwritten: an advance of the evidence clock the comms-loss watchdog + /// reads is lost (and the clock even moves backwards), or a Faulted verdict is downgraded despite the + /// guard. is re-run on each attempt, so it must be side-effect-free and + /// must derive everything — timestamps included — from the snapshot it is handed or from the clock. + /// Returning null declines the update. + /// + /// Builds the snapshot to publish, or returns null to leave health untouched. + private void UpdateHealth(Func transform) { - var current = ReadHealth(); - WriteHealth(current with { LastSuccessfulRead = DateTime.UtcNow }); + while (true) + { + var current = Volatile.Read(ref _health); + var next = transform(current); + if (next is null) return; + if (ReferenceEquals(Interlocked.CompareExchange(ref _health, next, current), current)) return; + } } /// Barrier-protected read of the multi-threaded health field. private DriverHealth ReadHealth() => Volatile.Read(ref _health); - /// Barrier-protected publish of a new health snapshot. + /// + /// 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. + /// + /// The health snapshot to publish. private void WriteHealth(DriverHealth value) => Volatile.Write(ref _health, value); /// Records a host-state change and raises — on transitions only. 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 2711da6b..c059e476 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 @@ -356,6 +356,55 @@ public sealed class SqlDriverTests 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 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; + } + }); + + var writers = new[] + { + Task.Run(() => { for (var i = 0; i < iterations; i++) driver.ObservePollOutcome([advancing]); }), + Task.Run(() => { for (var i = 0; i < iterations; i++) driver.ObservePollOutcome([degrading]); }), + }; + + await Task.WhenAll(writers); + await done.CancelAsync(); + await observer; + + Volatile.Read(ref rewinds).ShouldBe(0); + } + // ---- ISubscribable (poll-engine overlay) ---- [Fact]