Joseph Doherty
0b5e9b44f6
Task 4 of the LocalDb Phase 1 adoption plan. OperationTrackingStore took IOptions<OperationTrackingOptions> and opened its own SqliteConnection from ConnectionString. It now takes ILocalDb and gets every connection - writer and the two ad-hoc reader paths - from CreateConnection(). This is not cosmetic. OperationTracking is a RegisterReplicated table, so its capture triggers call zb_hlc_next(). That UDF is registered per connection by ILocalDb and by nothing else, so a raw SqliteConnection would fail closed on every write. Connections from CreateConnection() also arrive already open - calling Open()/OpenAsync() on one throws - hence the removed OpenAsync calls on the reader paths. OperationTrackingOptions.ConnectionString is now vestigial for this store; the database location is LocalDb:Path. The options class stays (retention settings) and the config key stays bound for the site config DB. InitializeSchema is kept but is now always a no-op in the host: onReady runs while ILocalDb is being constructed, strictly before this constructor can receive it. It remains so a directly-constructed store (tests, tooling) is self-sufficient. Tests: the store fixture moves off mode=memory&cache=shared onto a real ILocalDb over a temp file. There is no in-memory mode - LocalDbOptions.Path is a filesystem path - and testing through a raw in-memory connection would no longer resemble the host. Verifier connections stay raw on purpose: this fixture never registers the table, so no trigger fires and the UDF is never reached. Three DI fixtures needed LocalDb:Path added, because resolving IOperationTrackingStore now forces ILocalDb construction: SiteCompositionRootTests, SiteAuditWiringTests, and the AuditLog CombinedTelemetryHarness. Verified: build 0 warnings; Host 294/294, SiteRuntime 530/530, AuditLog 354/354. Claude-Session: https://claude.ai/code/session_01BL2Vu1ESDQ9SCN4gVKkdts
479 lines
21 KiB
C#
479 lines
21 KiB
C#
using System.Globalization;
|
|
using Microsoft.Data.Sqlite;
|
|
using Microsoft.Extensions.Logging;
|
|
using ZB.MOM.WW.LocalDb;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types;
|
|
|
|
namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tracking;
|
|
|
|
/// <summary>
|
|
/// Site-local SQLite source-of-truth for cached-operation tracking — the row
|
|
/// that <c>Tracking.Status(TrackedOperationId)</c> reads.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// <para>
|
|
/// One row per <see cref="TrackedOperationId"/>; lifecycle is
|
|
/// <c>Submitted → Retrying → Delivered / Parked / Failed / Discarded</c>; terminal
|
|
/// rows are purged after the configured retention window
|
|
/// (<see cref="OperationTrackingOptions.RetentionDays"/>). Volume is bounded —
|
|
/// only cached calls produce rows, and only a handful of lifecycle events per
|
|
/// call — so we keep the implementation deliberately simple: a single owned
|
|
/// <see cref="SqliteConnection"/> serialised behind a <see cref="SemaphoreSlim"/>
|
|
/// (one async writer at a time) — simpler than a batched-channel pipeline given
|
|
/// the volume; the batched-channel audit-writer design is reserved for the
|
|
/// high-volume audit hot-path.
|
|
/// </para>
|
|
/// <para>
|
|
/// All mutations are idempotent / monotonic: <see cref="RecordEnqueueAsync"/> is
|
|
/// <c>INSERT OR IGNORE</c>, <see cref="RecordAttemptAsync"/> filters out terminal
|
|
/// rows in the <c>WHERE</c> clause, and <see cref="RecordTerminalAsync"/> only
|
|
/// fires on rows that haven't terminated yet (first-write-wins). This makes the
|
|
/// store safe under the at-least-once semantics of the site→central telemetry
|
|
/// path.
|
|
/// </para>
|
|
/// </remarks>
|
|
public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable, IDisposable
|
|
{
|
|
// Writer state — one owned SqliteConnection serialised behind
|
|
// _writeGate. Readers do NOT share this connection or gate; see GetStatusAsync.
|
|
private readonly SqliteConnection _writeConnection;
|
|
private readonly SemaphoreSlim _writeGate = new(1, 1);
|
|
private readonly ILocalDb _localDb;
|
|
private readonly ILogger<OperationTrackingStore> _logger;
|
|
|
|
// Dispose-once state shared by the sync Dispose and async
|
|
// DisposeAsync paths. Interlocked.Exchange is the race-safe primitive here —
|
|
// a plain bool can be flipped twice if Dispose() and DisposeAsync() are
|
|
// invoked concurrently (e.g. host shutdown bridging both). 0 = live,
|
|
// 1 = disposed. Read by other methods via Volatile.Read after the gate is
|
|
// taken; they raise ObjectDisposedException when set.
|
|
private int _disposeState;
|
|
|
|
/// <summary>
|
|
/// Initializes the tracking store over the consolidated site database and applies the schema.
|
|
/// </summary>
|
|
/// <param name="localDb">
|
|
/// The consolidated site database. Every connection it hands out is already open and carries
|
|
/// the per-connection pragmas plus the <c>zb_hlc_next()</c> UDF that
|
|
/// <c>OperationTracking</c>'s capture triggers call — which is exactly why the store no longer
|
|
/// opens its own <see cref="SqliteConnection"/> from a connection string. A raw connection
|
|
/// would lack the UDF and every write to the replicated table would fail closed.
|
|
/// </param>
|
|
/// <param name="logger">Logger for diagnostics.</param>
|
|
/// <remarks>
|
|
/// <see cref="OperationTrackingOptions.ConnectionString"/> no longer feeds this store — the
|
|
/// database location is <c>LocalDb:Path</c>. The option remains bound for the purge/retention
|
|
/// settings that share the class.
|
|
/// </remarks>
|
|
public OperationTrackingStore(
|
|
ILocalDb localDb,
|
|
ILogger<OperationTrackingStore> logger)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(localDb);
|
|
ArgumentNullException.ThrowIfNull(logger);
|
|
|
|
_logger = logger;
|
|
_localDb = localDb;
|
|
// Already open — CreateConnection returns a live, pragma-configured,
|
|
// UDF-registered connection. Calling Open() on it again would throw.
|
|
_writeConnection = localDb.CreateConnection();
|
|
InitializeSchema();
|
|
}
|
|
|
|
// Schema lives in OperationTrackingSchema so the Host's AddZbLocalDb onReady
|
|
// callback can create this table in the consolidated site database before
|
|
// RegisterReplicated installs its capture triggers. In the host this call is
|
|
// therefore always a no-op — onReady runs while ILocalDb is being constructed,
|
|
// which is strictly before this constructor can receive it. It stays because it
|
|
// keeps a directly-constructed store (tests, tooling) self-sufficient, and the
|
|
// DDL is idempotent.
|
|
private void InitializeSchema() => OperationTrackingSchema.Apply(_writeConnection);
|
|
|
|
/// <inheritdoc/>
|
|
public async Task RecordEnqueueAsync(
|
|
TrackedOperationId id,
|
|
string kind,
|
|
string? targetSummary,
|
|
string? sourceInstanceId,
|
|
string? sourceScript,
|
|
string? sourceNode,
|
|
CancellationToken ct = default)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(kind);
|
|
|
|
await _writeGate.WaitAsync(ct).ConfigureAwait(false);
|
|
try
|
|
{
|
|
ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposeState) != 0, this);
|
|
|
|
var now = DateTime.UtcNow.ToString("o", CultureInfo.InvariantCulture);
|
|
|
|
using var cmd = _writeConnection.CreateCommand();
|
|
// INSERT OR IGNORE: duplicate ids are no-ops (first-write-wins) —
|
|
// matches the at-least-once semantics the site emits under.
|
|
cmd.CommandText = """
|
|
INSERT OR IGNORE INTO OperationTracking (
|
|
TrackedOperationId, Kind, TargetSummary, Status,
|
|
RetryCount, LastError, HttpStatus,
|
|
CreatedAtUtc, UpdatedAtUtc, TerminalAtUtc,
|
|
SourceInstanceId, SourceScript, SourceNode
|
|
) VALUES (
|
|
$id, $kind, $targetSummary, $status,
|
|
0, NULL, NULL,
|
|
$now, $now, NULL,
|
|
$sourceInstanceId, $sourceScript, $sourceNode
|
|
);
|
|
""";
|
|
cmd.Parameters.AddWithValue("$id", id.ToString());
|
|
cmd.Parameters.AddWithValue("$kind", kind);
|
|
cmd.Parameters.AddWithValue("$targetSummary", (object?)targetSummary ?? DBNull.Value);
|
|
cmd.Parameters.AddWithValue("$status", "Submitted");
|
|
cmd.Parameters.AddWithValue("$now", now);
|
|
cmd.Parameters.AddWithValue("$sourceInstanceId", (object?)sourceInstanceId ?? DBNull.Value);
|
|
cmd.Parameters.AddWithValue("$sourceScript", (object?)sourceScript ?? DBNull.Value);
|
|
cmd.Parameters.AddWithValue("$sourceNode", (object?)sourceNode ?? DBNull.Value);
|
|
|
|
cmd.ExecuteNonQuery();
|
|
}
|
|
finally
|
|
{
|
|
_writeGate.Release();
|
|
}
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public async Task RecordAttemptAsync(
|
|
TrackedOperationId id,
|
|
string status,
|
|
int retryCount,
|
|
string? lastError,
|
|
int? httpStatus,
|
|
CancellationToken ct = default)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(status);
|
|
|
|
await _writeGate.WaitAsync(ct).ConfigureAwait(false);
|
|
try
|
|
{
|
|
ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposeState) != 0, this);
|
|
|
|
var now = DateTime.UtcNow.ToString("o", CultureInfo.InvariantCulture);
|
|
|
|
using var cmd = _writeConnection.CreateCommand();
|
|
// Terminal rows are immutable — the WHERE clause filters them out so
|
|
// late-arriving attempt telemetry never overwrites a resolved row.
|
|
cmd.CommandText = """
|
|
UPDATE OperationTracking
|
|
SET Status = $status,
|
|
RetryCount = $retryCount,
|
|
LastError = $lastError,
|
|
HttpStatus = $httpStatus,
|
|
UpdatedAtUtc = $now
|
|
WHERE TrackedOperationId = $id
|
|
AND TerminalAtUtc IS NULL;
|
|
""";
|
|
cmd.Parameters.AddWithValue("$id", id.ToString());
|
|
cmd.Parameters.AddWithValue("$status", status);
|
|
cmd.Parameters.AddWithValue("$retryCount", retryCount);
|
|
cmd.Parameters.AddWithValue("$lastError", (object?)lastError ?? DBNull.Value);
|
|
cmd.Parameters.AddWithValue("$httpStatus", (object?)httpStatus ?? DBNull.Value);
|
|
cmd.Parameters.AddWithValue("$now", now);
|
|
|
|
cmd.ExecuteNonQuery();
|
|
}
|
|
finally
|
|
{
|
|
_writeGate.Release();
|
|
}
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public async Task RecordTerminalAsync(
|
|
TrackedOperationId id,
|
|
string status,
|
|
string? lastError,
|
|
int? httpStatus,
|
|
CancellationToken ct = default)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(status);
|
|
|
|
await _writeGate.WaitAsync(ct).ConfigureAwait(false);
|
|
try
|
|
{
|
|
ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposeState) != 0, this);
|
|
|
|
var now = DateTime.UtcNow.ToString("o", CultureInfo.InvariantCulture);
|
|
|
|
using var cmd = _writeConnection.CreateCommand();
|
|
// First-write-wins on the terminal flip: only update rows that
|
|
// haven't already terminated.
|
|
cmd.CommandText = """
|
|
UPDATE OperationTracking
|
|
SET Status = $status,
|
|
LastError = $lastError,
|
|
HttpStatus = $httpStatus,
|
|
UpdatedAtUtc = $now,
|
|
TerminalAtUtc = $now
|
|
WHERE TrackedOperationId = $id
|
|
AND TerminalAtUtc IS NULL;
|
|
""";
|
|
cmd.Parameters.AddWithValue("$id", id.ToString());
|
|
cmd.Parameters.AddWithValue("$status", status);
|
|
cmd.Parameters.AddWithValue("$lastError", (object?)lastError ?? DBNull.Value);
|
|
cmd.Parameters.AddWithValue("$httpStatus", (object?)httpStatus ?? DBNull.Value);
|
|
cmd.Parameters.AddWithValue("$now", now);
|
|
|
|
cmd.ExecuteNonQuery();
|
|
}
|
|
finally
|
|
{
|
|
_writeGate.Release();
|
|
}
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public async Task<TrackingStatusSnapshot?> GetStatusAsync(
|
|
TrackedOperationId id,
|
|
CancellationToken ct = default)
|
|
{
|
|
ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposeState) != 0, this);
|
|
|
|
// Reads open a fresh, ungated connection so a long-running write doesn't
|
|
// block status queries. It comes from the same ILocalDb as the writer;
|
|
// SQLite handles cross-connection isolation natively (readers see a WAL
|
|
// snapshot). Mirrors the SiteStorageService precedent. Already open — do
|
|
// not call OpenAsync on it.
|
|
await using var readConnection = _localDb.CreateConnection();
|
|
|
|
await using var cmd = readConnection.CreateCommand();
|
|
cmd.CommandText = """
|
|
SELECT TrackedOperationId, Kind, TargetSummary, Status,
|
|
RetryCount, LastError, HttpStatus,
|
|
CreatedAtUtc, UpdatedAtUtc, TerminalAtUtc,
|
|
SourceInstanceId, SourceScript, SourceNode
|
|
FROM OperationTracking
|
|
WHERE TrackedOperationId = $id;
|
|
""";
|
|
cmd.Parameters.AddWithValue("$id", id.ToString());
|
|
|
|
await using var reader = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
|
|
if (!await reader.ReadAsync(ct).ConfigureAwait(false))
|
|
{
|
|
return null;
|
|
}
|
|
|
|
return new TrackingStatusSnapshot(
|
|
Id: TrackedOperationId.Parse(reader.GetString(0)),
|
|
Kind: reader.GetString(1),
|
|
TargetSummary: reader.IsDBNull(2) ? null : reader.GetString(2),
|
|
Status: reader.GetString(3),
|
|
RetryCount: reader.GetInt32(4),
|
|
LastError: reader.IsDBNull(5) ? null : reader.GetString(5),
|
|
HttpStatus: reader.IsDBNull(6) ? null : reader.GetInt32(6),
|
|
CreatedAtUtc: ParseUtc(reader.GetString(7)),
|
|
UpdatedAtUtc: ParseUtc(reader.GetString(8)),
|
|
TerminalAtUtc: reader.IsDBNull(9) ? null : ParseUtc(reader.GetString(9)),
|
|
SourceInstanceId: reader.IsDBNull(10) ? null : reader.GetString(10),
|
|
SourceScript: reader.IsDBNull(11) ? null : reader.GetString(11),
|
|
SourceNode: reader.IsDBNull(12) ? null : reader.GetString(12));
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public async Task PurgeTerminalAsync(
|
|
DateTime olderThanUtc,
|
|
CancellationToken ct = default)
|
|
{
|
|
await _writeGate.WaitAsync(ct).ConfigureAwait(false);
|
|
try
|
|
{
|
|
ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposeState) != 0, this);
|
|
|
|
using var cmd = _writeConnection.CreateCommand();
|
|
// Non-terminal rows (TerminalAtUtc IS NULL) are kept regardless of
|
|
// age — the operation is still in flight.
|
|
cmd.CommandText = """
|
|
DELETE FROM OperationTracking
|
|
WHERE TerminalAtUtc IS NOT NULL
|
|
AND TerminalAtUtc < $threshold;
|
|
""";
|
|
cmd.Parameters.AddWithValue(
|
|
"$threshold",
|
|
olderThanUtc.ToString("o", CultureInfo.InvariantCulture));
|
|
|
|
cmd.ExecuteNonQuery();
|
|
}
|
|
finally
|
|
{
|
|
_writeGate.Release();
|
|
}
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public async Task<IReadOnlyList<SiteCallOperational>> ReadChangedSinceAsync(
|
|
DateTime sinceUtc,
|
|
int batchSize,
|
|
string? afterId = null,
|
|
CancellationToken ct = default)
|
|
{
|
|
ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposeState) != 0, this);
|
|
|
|
// Like GetStatusAsync, the reconciliation pull opens a
|
|
// fresh, ungated read connection so a long-running write never blocks
|
|
// central's PullSiteCalls. The query is a bounded, ordered scan served by
|
|
// the standalone IX_OperationTracking_UpdatedAt index — UpdatedAtUtc is
|
|
// the cursor. (The composite (Status, UpdatedAtUtc) index cannot satisfy a
|
|
// status-less UpdatedAtUtc range scan; this dedicated index does.)
|
|
// Already open — do not call OpenAsync on it.
|
|
await using var readConnection = _localDb.CreateConnection();
|
|
|
|
await using var cmd = readConnection.CreateCommand();
|
|
|
|
// Composite (UpdatedAtUtc, TrackedOperationId) keyset. Ordering
|
|
// is ALWAYS deterministic (UpdatedAtUtc ASC, TrackedOperationId ASC) so the
|
|
// cursor is well-defined even when many rows share one UpdatedAtUtc.
|
|
// • afterId present → strict keyset: skip everything up to and including
|
|
// (sinceUtc, afterId). Without this a page fully inside one UpdatedAtUtc
|
|
// instant re-reads forever (the plain >= resume never advances past the
|
|
// tie). TrackedOperationId is the "D" GUID form on both sides, so the
|
|
// TEXT comparison is a stable total order.
|
|
// • afterId null/empty → legacy inclusive >= sinceUtc (an older central
|
|
// that does not send a cursor is byte-for-byte unaffected).
|
|
var useKeyset = !string.IsNullOrEmpty(afterId);
|
|
var predicate = useKeyset
|
|
? "UpdatedAtUtc > $since OR (UpdatedAtUtc = $since AND TrackedOperationId > $afterId)"
|
|
: "UpdatedAtUtc >= $since";
|
|
cmd.CommandText = $"""
|
|
SELECT TrackedOperationId, Kind, TargetSummary, Status,
|
|
RetryCount, LastError, HttpStatus,
|
|
CreatedAtUtc, UpdatedAtUtc, TerminalAtUtc, SourceNode
|
|
FROM OperationTracking
|
|
WHERE {predicate}
|
|
ORDER BY UpdatedAtUtc ASC, TrackedOperationId ASC
|
|
LIMIT $batchSize;
|
|
""";
|
|
// Force UTC kind before formatting so the cursor's "o" text matches the
|
|
// 'Z'-suffixed round-trip form the write path persists (DateTime.UtcNow
|
|
// .ToString("o")). A first-cycle DateTime.MinValue arrives Unspecified —
|
|
// without this its "o" rendering would lack the 'Z', and the SQLite text
|
|
// compare against 'Z'-suffixed stored values would be subtly inconsistent.
|
|
var sinceText = DateTime
|
|
.SpecifyKind(sinceUtc, DateTimeKind.Utc)
|
|
.ToString("o", CultureInfo.InvariantCulture);
|
|
cmd.Parameters.AddWithValue("$since", sinceText);
|
|
cmd.Parameters.AddWithValue("$batchSize", batchSize);
|
|
if (useKeyset)
|
|
{
|
|
cmd.Parameters.AddWithValue("$afterId", afterId!);
|
|
}
|
|
|
|
var rows = new List<SiteCallOperational>();
|
|
await using var reader = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
|
|
while (await reader.ReadAsync(ct).ConfigureAwait(false))
|
|
{
|
|
var kind = reader.GetString(1);
|
|
rows.Add(new SiteCallOperational(
|
|
TrackedOperationId: TrackedOperationId.Parse(reader.GetString(0)),
|
|
Channel: KindToChannel(kind),
|
|
Target: reader.IsDBNull(2) ? string.Empty : reader.GetString(2),
|
|
// The site id is not a tracking-store column; the central client
|
|
// re-stamps SourceSite from the siteId it dialed.
|
|
SourceSite: string.Empty,
|
|
SourceNode: reader.IsDBNull(10) ? null : reader.GetString(10),
|
|
Status: reader.GetString(3),
|
|
RetryCount: reader.GetInt32(4),
|
|
LastError: reader.IsDBNull(5) ? null : reader.GetString(5),
|
|
HttpStatus: reader.IsDBNull(6) ? null : reader.GetInt32(6),
|
|
CreatedAtUtc: ParseUtc(reader.GetString(7)),
|
|
UpdatedAtUtc: ParseUtc(reader.GetString(8)),
|
|
TerminalAtUtc: reader.IsDBNull(9) ? null : ParseUtc(reader.GetString(9))));
|
|
}
|
|
|
|
return rows;
|
|
}
|
|
|
|
// Cached-call Kind → SiteCalls Channel. Only ApiCallCached / DbWriteCached
|
|
// ever reach the tracking store (RecordEnqueueAsync is the cached-call
|
|
// entry point); DbWriteCached maps to DbOutbound, everything else to the
|
|
// ApiOutbound default. Mirrors CachedCallLifecycleBridge's channel handling.
|
|
private static string KindToChannel(string kind) => kind switch
|
|
{
|
|
nameof(Commons.Types.Enums.AuditKind.DbWriteCached) => nameof(Commons.Types.Enums.AuditChannel.DbOutbound),
|
|
_ => nameof(Commons.Types.Enums.AuditChannel.ApiOutbound),
|
|
};
|
|
|
|
private static DateTime ParseUtc(string raw)
|
|
{
|
|
return DateTime.Parse(
|
|
raw,
|
|
CultureInfo.InvariantCulture,
|
|
DateTimeStyles.RoundtripKind);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Synchronously disposes the tracking store and its SQLite connection.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// This path does NOT bridge to async via
|
|
/// <c>.AsTask().GetAwaiter().GetResult()</c>. Sync-over-async on a SemaphoreSlim
|
|
/// can deadlock when invoked from a non-reentrant SyncContext (e.g. host
|
|
/// shutdown continuations observed on the host sync context). In-flight writes
|
|
/// at the moment of <see cref="Dispose"/> will fail their next operation
|
|
/// against the disposed connection with <see cref="ObjectDisposedException"/> —
|
|
/// the caller's responsibility is to ensure no concurrent operations during
|
|
/// the synchronous dispose. Use <see cref="DisposeAsync"/> if you need to
|
|
/// drain in-flight writes before close.
|
|
/// </remarks>
|
|
public void Dispose()
|
|
{
|
|
if (Interlocked.Exchange(ref _disposeState, 1) != 0)
|
|
{
|
|
return;
|
|
}
|
|
|
|
_writeConnection.Dispose();
|
|
_writeGate.Dispose();
|
|
GC.SuppressFinalize(this);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Asynchronously disposes the tracking store and its SQLite connection.
|
|
/// Drains in-flight writes by acquiring the write gate before closing the
|
|
/// connection, so a write currently executing a SqliteCommand completes
|
|
/// before the connection is freed.
|
|
/// </summary>
|
|
/// <returns>A <see cref="ValueTask"/> that completes when all resources have been released.</returns>
|
|
public async ValueTask DisposeAsync()
|
|
{
|
|
if (Interlocked.Exchange(ref _disposeState, 1) != 0)
|
|
{
|
|
return;
|
|
}
|
|
|
|
// Drain any in-flight write by taking the write gate. Past this point
|
|
// no new write can acquire the gate because _disposeState is set, so
|
|
// the next ThrowIf check in each writer raises ObjectDisposedException.
|
|
try
|
|
{
|
|
await _writeGate.WaitAsync().ConfigureAwait(false);
|
|
}
|
|
catch (ObjectDisposedException)
|
|
{
|
|
// Race with another disposer that already disposed the gate — the
|
|
// _disposeState exchange above should prevent this, but be defensive.
|
|
}
|
|
|
|
try
|
|
{
|
|
_writeConnection.Dispose();
|
|
}
|
|
finally
|
|
{
|
|
try { _writeGate.Release(); } catch (ObjectDisposedException) { }
|
|
_writeGate.Dispose();
|
|
}
|
|
|
|
GC.SuppressFinalize(this);
|
|
}
|
|
}
|