dohertj2/ScadaBridge
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

perf(site-runtime): operation tracking store writes off the script thread (single-writer queue, batch per transaction)

Browse Source 
A Flow Debugger live run (execution 1130c01f) spent 108 of a 176 ms site script run
between the cached DB write completing and Notify.Send starting: the immediate-
completion telemetry awaited two autocommit SQLite transactions on the consolidated
LocalDb file (RecordAttemptAsync + RecordTerminalAsync), on the script thread, each a
WAL fsync on a Docker volume; the submit row was a third. The audit half of the same
forwarder was already a channel write, so the tracking half was the whole cost.

OperationTrackingStore is now a single-writer queue: Record*Async enqueue an ordered
work item and return; one background pump drains the bounded channel (8192, wait
backpressure - tracking never drops) and applies each batch (<=256) in ONE transaction,
falling back to per-row apply when a batch fails so a poison row cannot take its
neighbours. Preserved invariants: FIFO across enqueue/attempt/terminal; read-your-
writes (GetStatusAsync and ReadChangedSinceAsync barrier on every write queued before
them, skipped when idle) so Tracking.Status, the telemetry actor's status join and
central's PullSiteCalls never see a stale row; timestamps stamped when recorded;
PurgeTerminalAsync returns after the purge lands; DisposeAsync drains the queue before
closing; sync Dispose keeps its documented in-flight-writes-fail semantics. New public
FlushAsync() is the explicit barrier. Accepted: a queued-but-unapplied row is lost on a
hard crash (ms window); S&F buffer and audit rows are unaffected.

No interface, forwarder or ScriptRuntimeContext change. Tests: write returns while
another connection holds the DB write lock and lands once released (fails on the old
store: 31 s); reads see 100 ordered attempts + terminal; DisposeAsync drains 200
queued rows; write after dispose throws ObjectDisposed. Existing tests that poke the
SQLite file right after a store call now FlushAsync first.

Claude-Session: https://claude.ai/code/session_01NcMG6xXdsJbmD2wYAgkK29
This commit is contained in:
Joseph Doherty
2026-09-06 12:25:58 -04:00
parent 7caf56f332
commit d91135f960
6 changed files with 461 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Tracking/OperationTrackingStore.cs
+275 -64
View File
@@ -1,4 +1,5 @@
using System.Globalization;
using System.Threading.Channels;
using Microsoft.Data.Sqlite;
using Microsoft.Extensions.Logging;
using ZB.MOM.WW.LocalDb;
@@ -18,11 +19,28 @@ namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tracking;
/// 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.
/// call — but every one of those rows used to be its own autocommit transaction
/// <b>awaited on the script thread</b> (through <c>CachedCallTelemetryForwarder</c>),
/// and a WAL commit on the consolidated LocalDb file costs a fsync (~30–50 ms on a
/// Docker volume): a single <c>Database.CachedWrite</c> paid three of them, ~108 ms
/// of a 176 ms script run measured on the rig. Since 2026-09-06 the store is a
/// <b>single-writer queue</b>: <see cref="RecordEnqueueAsync"/> /
/// <see cref="RecordAttemptAsync"/> / <see cref="RecordTerminalAsync"/> enqueue an
/// ordered work item and return, and one background pump applies each drained batch
/// in one transaction on the single owned <see cref="SqliteConnection"/>. Reads
/// (<see cref="GetStatusAsync"/>, <see cref="ReadChangedSinceAsync"/>) wait for every
/// write queued before them — a barrier through the same queue — so
/// <c>Tracking.Status(id)</c> keeps read-your-writes and the telemetry actor's status
/// join never sees a stale row. <see cref="PurgeTerminalAsync"/> rides the queue too
/// and returns after the purge is applied. Design memo:
/// <c>docs/plans/2026-09-06-tracking-store-async-writes-design.md</c>.
/// </para>
/// <para>
/// Accepted trade: a row queued but not yet applied is lost on a hard crash (the
/// pump's cadence, milliseconds). The tracking row then stays at its previous state
/// until reconciled; the S&amp;F buffer and audit rows are unaffected. The queue is
/// bounded (<see cref="MaxQueuedWrites"/>) with wait-mode backpressure — tracking is
/// the status source of truth, so it never drops.
/// </para>
/// <para>
/// All mutations are idempotent / monotonic: <see cref="RecordEnqueueAsync"/> is
@@ -35,13 +53,47 @@ namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tracking;
/// </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.
/// <summary>Queue bound; past it <c>Record*Async</c> waits (never drops).</summary>
internal const int MaxQueuedWrites = 8192;
/// <summary>Most work items applied in one transaction.</summary>
internal const int MaxBatchSize = 256;
// Writer state — one owned SqliteConnection, touched ONLY by the pump (and by
// DisposeAsync once the pump has finished), 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;
// The single-writer queue. One reader (the pump), many writers (script threads,
// the S&F lifecycle bridge, the purge loop). FIFO is what keeps
// enqueue → attempt → terminal applying in the order they were recorded.
private readonly Channel<WorkItem> _queue = Channel.CreateBounded<WorkItem>(
new BoundedChannelOptions(MaxQueuedWrites)
{
SingleReader = true,
SingleWriter = false,
FullMode = BoundedChannelFullMode.Wait,
});
private readonly Task _pump;
// Items queued and not yet applied. A read that observes 0 has nothing to wait
// for and skips the barrier; otherwise it queues one and waits its turn.
private int _pending;
/// <summary>
/// One queued mutation. <see cref="Apply"/> runs on the pump inside the batch
/// transaction; <see cref="Completion"/> (when present) is completed after the
/// item is durably applied, or faulted when it failed — barriers and the purge
/// use it, plain lifecycle writes do not wait.
/// </summary>
private sealed record WorkItem(
string Description,
Action<SqliteConnection, SqliteTransaction>? Apply,
TaskCompletionSource? Completion);
// 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
@@ -79,6 +131,7 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable,
// UDF-registered connection. Calling Open() on it again would throw.
_writeConnection = localDb.CreateConnection();
InitializeSchema();
_pump = Task.Run(PumpAsync);
}
// Schema lives in OperationTrackingSchema so the Host's AddZbLocalDb onReady
@@ -102,14 +155,14 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable,
{
ArgumentNullException.ThrowIfNull(kind);
await _writeGate.WaitAsync(ct).ConfigureAwait(false);
try
// Stamped when RECORDED, not when the batch lands.
var now = DateTime.UtcNow.ToString("o", CultureInfo.InvariantCulture);
var idText = id.ToString();
await EnqueueAsync($"enqueue {idText}", (conn, tx) =>
{
ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposeState) != 0, this);
var now = DateTime.UtcNow.ToString("o", CultureInfo.InvariantCulture);
using var cmd = _writeConnection.CreateCommand();
using var cmd = conn.CreateCommand();
cmd.Transaction = tx;
// INSERT OR IGNORE: duplicate ids are no-ops (first-write-wins) —
// matches the at-least-once semantics the site emits under.
cmd.CommandText = """
@@ -125,7 +178,7 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable,
$sourceInstanceId, $sourceScript, $sourceNode
);
""";
cmd.Parameters.AddWithValue("$id", id.ToString());
cmd.Parameters.AddWithValue("$id", idText);
cmd.Parameters.AddWithValue("$kind", kind);
cmd.Parameters.AddWithValue("$targetSummary", (object?)targetSummary ?? DBNull.Value);
cmd.Parameters.AddWithValue("$status", "Submitted");
@@ -135,11 +188,7 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable,
cmd.Parameters.AddWithValue("$sourceNode", (object?)sourceNode ?? DBNull.Value);
cmd.ExecuteNonQuery();
}
finally
{
_writeGate.Release();
}
}, ct).ConfigureAwait(false);
}
/// <inheritdoc/>
@@ -153,14 +202,13 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable,
{
ArgumentNullException.ThrowIfNull(status);
await _writeGate.WaitAsync(ct).ConfigureAwait(false);
try
var now = DateTime.UtcNow.ToString("o", CultureInfo.InvariantCulture);
var idText = id.ToString();
await EnqueueAsync($"attempt {idText}", (conn, tx) =>
{
ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposeState) != 0, this);
var now = DateTime.UtcNow.ToString("o", CultureInfo.InvariantCulture);
using var cmd = _writeConnection.CreateCommand();
using var cmd = conn.CreateCommand();
cmd.Transaction = tx;
// Terminal rows are immutable — the WHERE clause filters them out so
// late-arriving attempt telemetry never overwrites a resolved row.
cmd.CommandText = """
@@ -173,7 +221,7 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable,
WHERE TrackedOperationId = $id
AND TerminalAtUtc IS NULL;
""";
cmd.Parameters.AddWithValue("$id", id.ToString());
cmd.Parameters.AddWithValue("$id", idText);
cmd.Parameters.AddWithValue("$status", status);
cmd.Parameters.AddWithValue("$retryCount", retryCount);
cmd.Parameters.AddWithValue("$lastError", (object?)lastError ?? DBNull.Value);
@@ -181,11 +229,7 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable,
cmd.Parameters.AddWithValue("$now", now);
cmd.ExecuteNonQuery();
}
finally
{
_writeGate.Release();
}
}, ct).ConfigureAwait(false);
}
/// <inheritdoc/>
@@ -198,14 +242,13 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable,
{
ArgumentNullException.ThrowIfNull(status);
await _writeGate.WaitAsync(ct).ConfigureAwait(false);
try
var now = DateTime.UtcNow.ToString("o", CultureInfo.InvariantCulture);
var idText = id.ToString();
await EnqueueAsync($"terminal {idText}", (conn, tx) =>
{
ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposeState) != 0, this);
var now = DateTime.UtcNow.ToString("o", CultureInfo.InvariantCulture);
using var cmd = _writeConnection.CreateCommand();
using var cmd = conn.CreateCommand();
cmd.Transaction = tx;
// First-write-wins on the terminal flip: only update rows that
// haven't already terminated.
cmd.CommandText = """
@@ -218,18 +261,172 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable,
WHERE TrackedOperationId = $id
AND TerminalAtUtc IS NULL;
""";
cmd.Parameters.AddWithValue("$id", id.ToString());
cmd.Parameters.AddWithValue("$id", idText);
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
}, ct).ConfigureAwait(false);
}
// ── Queue plumbing ─────────────────────────────────────────────────────
/// <summary>
/// Completes once every write queued before this call has been applied. Reads
/// take this barrier implicitly; it is public for callers that go around the
/// store to the file (tests, tooling) and for anyone who wants "durable now".
/// </summary>
public Task FlushAsync(CancellationToken ct = default) => WaitForQueuedWritesAsync(ct);
/// <summary>Queues a lifecycle write and returns once it is queued (not applied).</summary>
private Task EnqueueAsync(string description, Action<SqliteConnection, SqliteTransaction> apply, CancellationToken ct)
=> EnqueueAsync(new WorkItem(description, apply, Completion: null), ct);
/// <summary>Queues a work item; a completion-bearing item is awaited by its owner separately.</summary>
private async Task EnqueueAsync(WorkItem item, CancellationToken ct)
{
ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposeState) != 0, this);
Interlocked.Increment(ref _pending);
try
{
_writeGate.Release();
await _queue.Writer.WriteAsync(item, ct).ConfigureAwait(false);
}
catch (ChannelClosedException)
{
Interlocked.Decrement(ref _pending);
throw new ObjectDisposedException(GetType().FullName);
}
catch
{
Interlocked.Decrement(ref _pending);
throw;
}
}
/// <summary>
/// Read barrier: completes once every write queued before this call has been
/// applied. Skipped when nothing is pending. A disposed store lets the read
/// proceed to its own ObjectDisposed check.
/// </summary>
private async Task WaitForQueuedWritesAsync(CancellationToken ct)
{
if (Volatile.Read(ref _pending) == 0)
return;
var barrier = new WorkItem("barrier", Apply: null,
new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously));
try
{
await EnqueueAsync(barrier, ct).ConfigureAwait(false);
}
catch (ObjectDisposedException)
{
return;
}
await barrier.Completion!.Task.WaitAsync(ct).ConfigureAwait(false);
}
/// <summary>
/// The single writer. Drains the queue in batches, applies each batch in one
/// transaction (one fsync for the whole batch), and falls back to one row at a
/// time when a batch fails so a poison row never takes its neighbours down.
/// Never throws — it runs for the store's lifetime and ends when the queue is
/// completed by disposal.
/// </summary>
private async Task PumpAsync()
{
var reader = _queue.Reader;
var batch = new List<WorkItem>(MaxBatchSize);
try
{
while (await reader.WaitToReadAsync().ConfigureAwait(false))
{
batch.Clear();
while (batch.Count < MaxBatchSize && reader.TryRead(out var item))
batch.Add(item);
if (batch.Count == 0)
continue;
await _writeGate.WaitAsync().ConfigureAwait(false);
try
{
ApplyBatch(batch);
}
finally
{
_writeGate.Release();
}
}
}
catch (Exception ex)
{
// Only the gate itself can throw here (disposed under us on the sync
// Dispose path). Fail whatever is still queued so no awaiter hangs.
_logger.LogWarning(ex, "OperationTrackingStore: write pump stopped; failing queued tracking writes");
foreach (var item in batch)
Fail(item, ex);
while (reader.TryRead(out var left))
Fail(left, ex);
}
}
private void ApplyBatch(List<WorkItem> batch)
{
if (Volatile.Read(ref _disposeState) != 0 && _writeConnection.State != System.Data.ConnectionState.Open)
{
// Sync Dispose() closed the connection while items were queued — the
// documented "in-flight writes fail" semantics of that path.
var disposed = new ObjectDisposedException(GetType().FullName);
foreach (var item in batch)
Fail(item, disposed);
return;
}
try
{
using var tx = _writeConnection.BeginTransaction();
foreach (var item in batch)
item.Apply?.Invoke(_writeConnection, tx);
tx.Commit();
foreach (var item in batch)
Complete(item);
return;
}
catch (Exception ex)
{
_logger.LogWarning(ex,
"OperationTrackingStore: batch of {Count} tracking writes failed; re-applying one row at a time", batch.Count);
}
// Per-row fallback: keep the good rows, log and fail only the bad one.
foreach (var item in batch)
{
try
{
using var tx = _writeConnection.BeginTransaction();
item.Apply?.Invoke(_writeConnection, tx);
tx.Commit();
Complete(item);
}
catch (Exception ex)
{
_logger.LogWarning(ex, "OperationTrackingStore: tracking write failed ({Description})", item.Description);
Fail(item, ex);
}
}
}
private void Complete(WorkItem item)
{
Interlocked.Decrement(ref _pending);
item.Completion?.TrySetResult();
}
private void Fail(WorkItem item, Exception ex)
{
Interlocked.Decrement(ref _pending);
item.Completion?.TrySetException(ex);
}
/// <inheritdoc/>
@@ -237,6 +434,8 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable,
TrackedOperationId id,
CancellationToken ct = default)
{
// Read-your-writes: drain what was queued before this call (no-op when idle).
await WaitForQueuedWritesAsync(ct).ConfigureAwait(false);
ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposeState) != 0, this);
// Reads open a fresh, ungated connection so a long-running write doesn't
@@ -284,12 +483,13 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable,
DateTime olderThanUtc,
CancellationToken ct = default)
{
await _writeGate.WaitAsync(ct).ConfigureAwait(false);
try
var threshold = olderThanUtc.ToString("o", CultureInfo.InvariantCulture);
// The purge rides the queue (FIFO with the lifecycle writes) and, unlike
// them, is awaited to completion so the caller's "purged" means purged.
var purge = new WorkItem("purge", (conn, tx) =>
{
ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposeState) != 0, this);
using var cmd = _writeConnection.CreateCommand();
using var cmd = conn.CreateCommand();
cmd.Transaction = tx;
// Non-terminal rows (TerminalAtUtc IS NULL) are kept regardless of
// age — the operation is still in flight.
cmd.CommandText = """
@@ -297,16 +497,12 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable,
WHERE TerminalAtUtc IS NOT NULL
AND TerminalAtUtc < $threshold;
""";
cmd.Parameters.AddWithValue(
"$threshold",
olderThanUtc.ToString("o", CultureInfo.InvariantCulture));
cmd.Parameters.AddWithValue("$threshold", threshold);
cmd.ExecuteNonQuery();
}
finally
{
_writeGate.Release();
}
}, new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously));
await EnqueueAsync(purge, ct).ConfigureAwait(false);
await purge.Completion!.Task.WaitAsync(ct).ConfigureAwait(false);
}
/// <inheritdoc/>
@@ -316,6 +512,8 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable,
string? afterId = null,
CancellationToken ct = default)
{
// Read-your-writes for central's reconciliation pull, same as GetStatusAsync.
await WaitForQueuedWritesAsync(ct).ConfigureAwait(false);
ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposeState) != 0, this);
// Like GetStatusAsync, the reconciliation pull opens a
@@ -431,6 +629,9 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable,
return;
}
// Stop accepting work; whatever is still queued fails on the pump with
// ObjectDisposedException (see ApplyBatch) — this path does not drain.
_queue.Writer.TryComplete();
_writeConnection.Dispose();
_writeGate.Dispose();
GC.SuppressFinalize(this);
@@ -438,9 +639,9 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable,
/// <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.
/// Drains the write queue first (completes it and awaits the pump), so every
/// write that was queued before disposal is applied before the connection is
/// freed. This is the path host shutdown must use.
/// </summary>
/// <returns>A <see cref="ValueTask"/> that completes when all resources have been released.</returns>
public async ValueTask DisposeAsync()
@@ -450,9 +651,19 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable,
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.
// Drain: no new item can be queued past this point (_disposeState is set
// and the writer is completed), so awaiting the pump means every queued
// write has been applied. Then take the gate so nothing else holds the
// connection while it closes.
_queue.Writer.TryComplete();
try
{
await _pump.ConfigureAwait(false);
}
catch (Exception ex)
{
_logger.LogWarning(ex, "OperationTrackingStore: write pump faulted during DisposeAsync");
}
try
{
await _writeGate.WaitAsync().ConfigureAwait(false);