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 59ae4956..945fc32b 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriver.cs @@ -261,6 +261,10 @@ public sealed class SqlDriver _factory = _explicitFactory ?? dialect.Factory; _connectionString = connectionString; Endpoint = DescribeEndpoint(connectionString); + + // The reader being replaced owns a SemaphoreSlim (R06:S-3). Disposing it is safe even with a stale + // poll still inside it: the gate goes only when the last in-flight group releases it. + var previous = _reader; _reader = new SqlPollReader( _factory, connectionString, @@ -271,6 +275,7 @@ public sealed class SqlDriver nullIsBad: options.NullIsBad, resolve: rawPath => _resolver.TryResolve(rawPath, out var definition) ? definition : null, logger: _logger); + previous?.Dispose(); } /// True when the supplied DriverConfig JSON carries a real body. The bootstrapper always passes @@ -1150,7 +1155,17 @@ public sealed class SqlDriver /// await using does not leak the poll loops. /// /// A task that represents the asynchronous operation. - public async ValueTask DisposeAsync() => await TeardownAsync().ConfigureAwait(false); + public async ValueTask DisposeAsync() + { + await TeardownAsync().ConfigureAwait(false); + + // Only on the FINAL dispose: a Shutdown → Initialize cycle on the same instance keeps polling + // through this reader when the init does not rebind (no rebinder / placeholder config). + _reader.Dispose(); + } + + /// The live poll reader. Test observation of the S-3 lifecycle only. + internal SqlPollReader ReaderForTests => _reader; /// /// Shared teardown: stops every polled subscription and drops the authored table. Idempotent — safe diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs index 04789c2d..da6ecfb5 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs @@ -36,7 +36,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Sql; /// no subscription state and no connection between polls, so the driver shell can wrap it in /// and own those concerns. /// -public sealed class SqlPollReader +public sealed class SqlPollReader : IDisposable { /// Minimum spacing between "your source violates the query contract" warnings, per reader. private static readonly TimeSpan ContractWarningInterval = TimeSpan.FromMinutes(1); @@ -52,12 +52,20 @@ public sealed class SqlPollReader /// /// Caps concurrently executing group queries — and therefore concurrently held connections - /// (design §8.4). Never disposed: a only needs disposal once its - /// AvailableWaitHandle has been touched, and not disposing it also removes the hazard of a - /// timed-out-but-still-running group releasing into a disposed semaphore. + /// (design §8.4). Disposed by only once every group that ever acquired it has + /// released it (), so a timed-out-but-still-running group can never release into a + /// disposed semaphore (R06:S-3 — it used to be never disposed, one per reinitialize). /// private readonly SemaphoreSlim _gate; + /// + /// Owner reference (1) plus one per in-flight group. drops the owner's; whoever + /// takes it to zero disposes . Zero is terminal — no group can start after it. + /// + private int _refs = 1; + + private int _disposed; + private long _lastContractWarningTicks; /// Initializes a new instance of the class. @@ -273,6 +281,22 @@ public sealed class SqlPollReader /// inside the deadline, so a poll behind a wedged group still returns on time — as BadTimeout. /// private async Task RunGroupAsync(SqlQueryPlan plan, CancellationToken cancellationToken) + { + // A reader replaced by a reinitialize (and disposed) can still be holding a stale poll: it runs no + // new group. Cancellation is the honest answer — the poll that asked has been superseded. + if (!TryAddRef()) throw new OperationCanceledException("The Sql poll reader was disposed."); + + try + { + return await RunGroupCoreAsync(plan, cancellationToken).ConfigureAwait(false); + } + finally + { + ReleaseRef(); + } + } + + private async Task RunGroupCoreAsync(SqlQueryPlan plan, CancellationToken cancellationToken) { var clock = Stopwatch.StartNew(); @@ -288,6 +312,9 @@ public sealed class SqlPollReader if (budget <= TimeSpan.Zero) return GroupResult.TimedOut; deadline.CancelAfter(budget); + // The work holds its OWN reference: it can outlive this method (a timed-out group is abandoned + // still running) and its finally releases the slot into _gate, which must therefore still exist. + AddRefForWork(); work = Task.Run( async () => { @@ -299,6 +326,7 @@ public sealed class SqlPollReader { _gate.Release(); deadline.Dispose(); + ReleaseRef(); } }, CancellationToken.None); @@ -334,6 +362,38 @@ public sealed class SqlPollReader } } + /// + /// Releases the concurrency gate once no group still holds it. Idempotent; safe while groups are in + /// flight (the last one out disposes it). A disposed reader starts no new group. + /// + public void Dispose() + { + if (Interlocked.Exchange(ref _disposed, 1) == 0) ReleaseRef(); + } + + /// True once has actually been disposed. Test observation only. + internal bool GateDisposedForTests => Volatile.Read(ref _refs) == 0; + + /// Takes a reference unless the reader is already fully released. + private bool TryAddRef() + { + while (true) + { + var current = Volatile.Read(ref _refs); + if (current == 0) return false; + if (Interlocked.CompareExchange(ref _refs, current + 1, current) == current) return true; + } + } + + /// A further reference for work started under an already-held one (so never from zero). + private void AddRefForWork() => Interlocked.Increment(ref _refs); + + /// Drops a reference; the one that reaches zero disposes the gate. + private void ReleaseRef() + { + if (Interlocked.Decrement(ref _refs) == 0) _gate.Dispose(); + } + /// How much of this group's operationTimeout budget is left; never negative. private TimeSpan Remaining(Stopwatch clock) { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlPollReaderTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlPollReaderTests.cs index da658517..87a0d02a 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlPollReaderTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlPollReaderTests.cs @@ -522,6 +522,60 @@ public sealed class SqlPollReaderTests dialect.Factory, " ", dialect, CommandTimeout, OperationTimeout, 4, false, _ => null)); } + // ---- R06:S-3: the concurrency gate is released ---- + + /// An idle reader releases its gate on Dispose, and a disposed reader starts no new group. + [Fact] + public async Task Dispose_releasesTheGate_andADisposedReaderRunsNoNewGroup() + { + using var fixture = new SqlitePollFixture(); + var reader = Reader(fixture, KvTag("Speed", SqlitePollFixture.PresentKey)); + (await reader.ReadAsync(["Speed"], CancellationToken.None))[0].StatusCode.ShouldBe(SqlStatusCodes.Good); + + reader.Dispose(); + reader.Dispose(); // idempotent + + reader.GateDisposedForTests.ShouldBeTrue(); + await Should.ThrowAsync(() => reader.ReadAsync(["Speed"], CancellationToken.None)); + } + + /// + /// Disposing while a timed-out group is still wedged inside the provider must NOT dispose the gate + /// under it — the zombie releases its slot into the gate when it finally finishes, which is exactly + /// the hazard that kept the gate undisposed before. The gate goes once the zombie is out. + /// + [Fact] + public async Task Dispose_withAWedgedGroupInFlight_waitsForItBeforeReleasingTheGate() + { + using var fixture = new SqlitePollFixture(); + var factory = new TrackingFactory(); + var reader = new SqlPollReader( + factory, fixture.ConnectionString, new SqliteDialect(), + 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"); + try + { + (await reader.ReadAsync(["Speed"], CancellationToken.None))[0].StatusCode.ShouldBe(SqlStatusCodes.BadTimeout); + factory.Live.ShouldBe(1); // the zombie is still inside the provider — the control + + reader.Dispose(); + reader.GateDisposedForTests.ShouldBeFalse("a group still holds the gate"); + } + finally + { + Execute(locker, "ROLLBACK"); + locker.Dispose(); + } + + (await WaitUntilAsync(() => reader.GateDisposedForTests, TimeSpan.FromSeconds(30))) + .ShouldBeTrue("the last group out must release the gate"); + factory.Live.ShouldBe(0); + } + [Fact] public async Task ReadAsync_rejectsANullReferenceList() => await Should.ThrowAsync(async () => diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlReinitConfigAdoptionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlReinitConfigAdoptionTests.cs index 1181de1a..f9223f94 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlReinitConfigAdoptionTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlReinitConfigAdoptionTests.cs @@ -101,6 +101,43 @@ public sealed class SqlReinitConfigAdoptionTests driver.Endpoint.ShouldContain("old-server"); } + /// + /// R06:S-3. Every rebind builds a new ; the replaced one (and its + /// SemaphoreSlim) used to be dropped undisposed, one per reinitialize. The control: the reader + /// that replaced it is live. + /// + [Fact] + public async Task A_rebind_disposes_the_reader_it_replaces() + { + using var _ = new ScopedEnvironmentVariable( + SqlConnectionStringResolver.EnvironmentVariablePrefix + RefA, + "Server=old-server;Database=OldDb;Integrated Security=true"); + using var __ = new ScopedEnvironmentVariable( + SqlConnectionStringResolver.EnvironmentVariablePrefix + RefB, + "Server=new-server;Database=NewDb;Integrated Security=true"); + + var driver = SqlDriverFactoryExtensions.CreateInstance( + "sql-1", $$"""{"provider":"SqlServer","connectionStringRef":"{{RefA}}"}""", loggerFactory: null); + var original = driver.ReaderForTests; + + try + { + await driver.ReinitializeAsync( + $$"""{"provider":"SqlServer","connectionStringRef":"{{RefB}}"}""", CancellationToken.None); + } + catch + { + // Liveness failure against a non-existent server is expected; the rebind ran first. + } + + driver.ReaderForTests.ShouldNotBeSameAs(original); + original.GateDisposedForTests.ShouldBeTrue(); + driver.ReaderForTests.GateDisposedForTests.ShouldBeFalse(); + + await driver.DisposeAsync(); + driver.ReaderForTests.GateDisposedForTests.ShouldBeTrue("the final dispose releases the live reader too"); + } + /// Sets an environment variable for the duration of a test and restores it after. The resolver /// reads process-global state, so a leaked variable would race every sibling test. private sealed class ScopedEnvironmentVariable : IDisposable