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

fix(sql): dispose the poll reader's concurrency gate once no group holds it (R06:S-3)

Browse Source 
SqlPollReader's SemaphoreSlim was never disposed and a new reader is built per
rebind. The reader is now IDisposable with a reference count (owner + one per
in-flight group): the gate is disposed only when the last group - including a
timed-out one still wedged in the provider - releases it, so nothing ever releases
into a disposed semaphore. ApplyBinding disposes the replaced reader; the final
DisposeAsync disposes the live one (Shutdown does not, since a re-init without a
rebind keeps polling through it).

Refs #744
Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 15:46:21 -04:00
parent c529948745
commit 16f401b9e7
4 changed files with 171 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlDriver.cs
+16 -1
View File
@@ -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();
}
/// <summary>True when the supplied DriverConfig JSON carries a real body. The bootstrapper always passes
@@ -1150,7 +1155,17 @@ public sealed class SqlDriver
/// <c>await using</c> does not leak the poll loops.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
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();
}
/// <summary>The live poll reader. Test observation of the S-3 lifecycle only.</summary>
internal SqlPollReader ReaderForTests => _reader;
/// <summary>
/// Shared teardown: stops every polled subscription and drops the authored table. Idempotent — safe
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql/SqlPollReader.cs
+64 -4
View File
@@ -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
/// <see cref="PollGroupEngine"/> and own those concerns.</para>
/// </summary>
public sealed class SqlPollReader
public sealed class SqlPollReader : IDisposable
{
/// <summary>Minimum spacing between "your source violates the query contract" warnings, per reader.</summary>
private static readonly TimeSpan ContractWarningInterval = TimeSpan.FromMinutes(1);
@@ -52,12 +52,20 @@ public sealed class SqlPollReader
/// <summary>
/// Caps concurrently executing group queries — and therefore concurrently held connections
/// (design §8.4). Never disposed: a <see cref="SemaphoreSlim"/> only needs disposal once its
/// <c>AvailableWaitHandle</c> 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 <see cref="Dispose"/> only once every group that ever acquired it has
/// released it (<see cref="_refs"/>), 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).
/// </summary>
private readonly SemaphoreSlim _gate;
/// <summary>
/// Owner reference (1) plus one per in-flight group. <see cref="Dispose"/> drops the owner's; whoever
/// takes it to zero disposes <see cref="_gate"/>. Zero is terminal — no group can start after it.
/// </summary>
private int _refs = 1;
private int _disposed;
private long _lastContractWarningTicks;
/// <summary>Initializes a new instance of the <see cref="SqlPollReader"/> class.</summary>
@@ -273,6 +281,22 @@ public sealed class SqlPollReader
/// inside the deadline, so a poll behind a wedged group still returns on time — as BadTimeout.</para>
/// </summary>
private async Task<GroupResult> 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<GroupResult> 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
}
}
/// <summary>
/// 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.
/// </summary>
public void Dispose()
{
if (Interlocked.Exchange(ref _disposed, 1) == 0) ReleaseRef();
}
/// <summary>True once <see cref="_gate"/> has actually been disposed. Test observation only.</summary>
internal bool GateDisposedForTests => Volatile.Read(ref _refs) == 0;
/// <summary>Takes a reference unless the reader is already fully released.</summary>
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;
}
}
/// <summary>A further reference for work started under an already-held one (so never from zero).</summary>
private void AddRefForWork() => Interlocked.Increment(ref _refs);
/// <summary>Drops a reference; the one that reaches zero disposes the gate.</summary>
private void ReleaseRef()
{
if (Interlocked.Decrement(ref _refs) == 0) _gate.Dispose();
}
/// <summary>How much of this group's <c>operationTimeout</c> budget is left; never negative.</summary>
private TimeSpan Remaining(Stopwatch clock)
{
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlPollReaderTests.cs
+54
View File
@@ -522,6 +522,60 @@ public sealed class SqlPollReaderTests
dialect.Factory, " ", dialect, CommandTimeout, OperationTimeout, 4, false, _ => null));
}
// ---- R06:S-3: the concurrency gate is released ----
/// <summary>An idle reader releases its gate on Dispose, and a disposed reader starts no new group.</summary>
[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<OperationCanceledException>(() => reader.ReadAsync(["Speed"], CancellationToken.None));
}
/// <summary>
/// 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.
/// </summary>
[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<ArgumentNullException>(async () =>
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Sql.Tests/SqlReinitConfigAdoptionTests.cs
+37
View File
@@ -101,6 +101,43 @@ public sealed class SqlReinitConfigAdoptionTests
driver.Endpoint.ShouldContain("old-server");
}
/// <summary>
/// <b>R06:S-3.</b> Every rebind builds a new <see cref="SqlPollReader"/>; the replaced one (and its
/// <c>SemaphoreSlim</c>) used to be dropped undisposed, one per reinitialize. The control: the reader
/// that replaced it is live.
/// </summary>
[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");
}
/// <summary>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.</summary>
private sealed class ScopedEnvironmentVariable : IDisposable