From d91135f960105b88bed3601b89968612ad28917e Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sun, 6 Sep 2026 12:25:58 -0400 Subject: [PATCH] perf(site-runtime): operation tracking store writes off the script thread (single-writer queue, batch per transaction) 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 --- CLAUDE.md | 2 +- ...9-06-tracking-store-async-writes-design.md | 81 +++++ docs/requirements/Component-SiteRuntime.md | 1 + .../requirements/Component-StoreAndForward.md | 2 +- .../Tracking/OperationTrackingStore.cs | 339 ++++++++++++++---- .../Tracking/OperationTrackingStoreTests.cs | 104 +++++- 6 files changed, 461 insertions(+), 68 deletions(-) create mode 100644 docs/plans/2026-09-06-tracking-store-async-writes-design.md diff --git a/CLAUDE.md b/CLAUDE.md index 1d386455..09a6edde 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -152,7 +152,7 @@ spec for each is `docs/requirements/Component-.md`, and `README.md` carrie - No Akka replication — MS SQL is the HA store; daily purge of terminal rows after a configurable window (default 365 days). - Notification Outbox retry reuses central SMTP max-retry-count and fixed interval for Email; Sms uses SmsConfiguration.MaxRetries/RetryDelay, falling back to the SMTP policy when unset. - Cached calls (`ExternalSystem.CachedCall`, `Database.CachedWrite`) return a `TrackedOperationId` tracking handle, unified with `Notify.Send`'s existing tracking model (`Notify.Status` retained as a thin alias). -- A site-local operation tracking table (SQLite, alongside the S&F buffer) is the source of truth for cached-call status; `Tracking.Status(id)` reads it site-locally and authoritatively; terminal rows purged after a configurable window (default 7 days). +- A site-local operation tracking table (SQLite, alongside the S&F buffer) is the source of truth for cached-call status; `Tracking.Status(id)` reads it site-locally and authoritatively; terminal rows purged after a configurable window (default 7 days). **Its writes are OFF the script thread as of 2026-09-06:** `OperationTrackingStore` is a single-writer queue (FIFO, batch-per-transaction pump, reads barrier on earlier queued writes so `Tracking.Status` keeps read-your-writes, `DisposeAsync` drains, sync `Dispose()` doesn't). Motivation: a Flow Debugger run showed one `Database.CachedWrite` paying three awaited SQLite commits, ~108 ms of a 176 ms site run. Trade: a queued-but-unapplied row is lost on a hard crash (ms window). Design: `docs/plans/2026-09-06-tracking-store-async-writes-design.md`. - Unified tracking status lifecycle `Pending → Retrying → Delivered / Parked / Failed / Discarded`; `Failed` = permanent failure (also returned synchronously to the calling script). No `Forwarding` state for cached calls. - Site Call Audit (#22): central `SiteCallAuditActor` singleton with a `SiteCalls` audit table (central MS SQL) fed by best-effort site telemetry plus periodic reconciliation pulls — an eventually-consistent mirror, NOT a dispatcher; cached-call delivery stays site-local. Ingest is insert-if-not-exists then upsert-on-newer-status. - Central UI Site Calls page + central→site `RetryParkedOperation`/`DiscardParkedOperation` relay for parked cached calls; central never mutates the `SiteCalls` row directly. diff --git a/docs/plans/2026-09-06-tracking-store-async-writes-design.md b/docs/plans/2026-09-06-tracking-store-async-writes-design.md new file mode 100644 index 00000000..28571d30 --- /dev/null +++ b/docs/plans/2026-09-06-tracking-store-async-writes-design.md @@ -0,0 +1,81 @@ +# Operation tracking store: writes off the script thread + +**Date:** 2026-09-06 · **Status:** implemented the same day (branch `feat/tracking-store-async-writes`) + +## Problem + +A Flow Debugger live run of `FlowOutboundDemo` (execution `1130c01f-…`) spent 194 ms end to end, of +which the site script run was 176 ms. Reading the audit rows against the code path: + +| Segment | Time | What it is | +|---|---|---| +| `ExternalSystem.Call` RigRest.Echo | 4 ms | the REST round trip | +| `Database.CachedWrite` submit → write completed | 61 ms | one tracking-store commit (`RecordEnqueueAsync`) + the MS SQL write | +| write completed → `Notify.Send` starts | **108 ms** | no script statement: the immediate-completion telemetry — two more tracking-store commits (`RecordAttemptAsync`, `RecordTerminalAsync`) | +| `Notify.Send` | ~2 ms | S&F enqueue | + +`OperationTrackingStore` executed every `Record*Async` as its own autocommit SQLite transaction on the +consolidated LocalDb file, **awaited inline on the script thread** through +`CachedCallTelemetryForwarder.ForwardAsync`. Each commit is a WAL append + fsync +(`synchronous=NORMAL`, Docker volume): ~30–50 ms apiece here. The audit half of the same forwarder was +already cheap — `SqliteAuditWriter.WriteAsync` is a `Channel.TryWrite` — so the tracking half was the +whole cost. The API cached-call path (`ExternalSystem.CachedCall`) pays the same three commits. + +## Decision + +Make `OperationTrackingStore` a **single-writer queue**: `Record*Async` enqueue an ordered work item +and return once it is queued; one background pump drains the queue and applies each drained batch in +**one transaction** (one fsync per batch instead of per row). Nothing above the store changes — +`IOperationTrackingStore`, the forwarder, `ScriptRuntimeContext` and the S&F lifecycle bridge are +untouched, and their existing `await`s now cost a channel write. + +### Invariants kept + +- **FIFO per store.** One bounded channel, one reader, so enqueue → attempt → terminal apply in the + order they were recorded. The idempotent/first-write-wins SQL is unchanged. +- **Read-your-writes.** `Tracking.Status(id)` (script), the telemetry actor's status join and central's + `PullSiteCalls` reconciliation all go through `GetStatusAsync`/`ReadChangedSinceAsync`, which now + **wait for every write queued before the call to be applied** (a barrier through the same queue, + skipped when nothing is pending). A script that calls `CachedWrite` and then `Tracking.Status` in + the next statement sees the row exactly as before. +- **Timestamps.** `CreatedAtUtc`/`UpdatedAtUtc`/`TerminalAtUtc` are stamped when the call is + *recorded*, not when the batch is applied. +- **`PurgeTerminalAsync` returns after the purge is applied** (it rides the queue with a completion). +- **Shutdown drains.** `DisposeAsync` completes the queue and awaits the pump, so queued writes land + before the connection closes. The synchronous `Dispose()` keeps its documented "in-flight writes + fail" semantics and must not be used where a drain matters (host shutdown uses the async path). +- **Best-effort failure handling, per row.** A batch that fails is rolled back and re-applied one + row at a time so a single poison row cannot take the others with it; each failed row logs a + warning, as the forwarder already did. + +### Costs accepted + +- **Crash window.** A row queued but not yet applied is lost on a hard crash (SIGKILL, power loss); + the tracking row then stays at its previous state — e.g. `Submitted` for a call that actually + delivered — until central's mirror is reconciled from the audit rows, or forever if it was the only + record. The window is the pump's cadence (sub-millisecond to a few milliseconds under load), and + the S&F buffer and audit rows, which are the delivery and evidence records, are unaffected. This is + the same trade the site audit writer made (`FlushIntervalMs`, `synchronous=NORMAL`). +- **Bounded backpressure.** The queue holds `MaxQueuedWrites` (8 192) items; beyond that + `Record*Async` waits rather than drops — tracking is the status source of truth, so dropping is + not acceptable. At the volumes this table sees (a handful of rows per cached call) the bound is + never reached; if it is, the SQLite file is the bottleneck and the wait is the honest signal. +- **A barrier read pays one pump round trip** when writes are pending. It is queued behind at most + one batch, so worst case it waits for one transaction (~one fsync) — the same cost a read used to + pay contending for the old write gate. + +### Rejected + +- *Fire-and-forget from the forwarder* (`_ = _trackingStore.Record…`): loses ordering across the + three lifecycle rows and read-your-writes, and moves the failure handling to an unobserved task. +- *Batching in the forwarder*: the forwarder is per-packet; only the store sees consecutive rows from + different runs, which is where batching pays. +- *A bigger LocalDb `synchronous` relaxation*: `OFF` would remove the fsync but risk corruption on + power loss for every replicated table, not just this one. + +## Verification + +- Store tests: a write returns while another connection holds the database write lock and lands + once it is released; reads observe every earlier write in order; `DisposeAsync` drains the queue. +- Rig: rerun `FlowOutboundDemo` through the Flow Debugger and compare the `DbCall`-to-`NotifySend` + gap and the run total against the 108 ms / 176 ms baseline above. diff --git a/docs/requirements/Component-SiteRuntime.md b/docs/requirements/Component-SiteRuntime.md index a1326963..7eb2e8e8 100644 --- a/docs/requirements/Component-SiteRuntime.md +++ b/docs/requirements/Component-SiteRuntime.md @@ -504,6 +504,7 @@ Available to all script runs and alarm on-trigger runs: ### Operation Tracking - `Tracking.Status("trackedOperationId")` — Returns a status record (status, retry count, last error, key timestamps) for any tracked operation: a cached external system call, a cached database write, or a notification. For cached calls and writes the answer is always site-local and authoritative — the site owns the operation tracking table. (`Notify.Status(...)` is a thin alias scoped to the notification domain.) +- **Tracking writes are off the script thread (2026-09-06).** `OperationTrackingStore` is a single-writer queue: `RecordEnqueue/Attempt/TerminalAsync` (reached from the script through `CachedCallTelemetryForwarder`) return once the row is *queued*, and one background pump applies each drained batch in one SQLite transaction — one fsync per batch instead of one per lifecycle row. Before this, a `Database.CachedWrite` that delivered immediately paid three awaited autocommit transactions on the consolidated LocalDb file, ~108 ms of a 176 ms script run measured on the rig. Preserved: FIFO order across the three lifecycle rows; **read-your-writes** — `GetStatusAsync` (behind `Tracking.Status`, and the telemetry actor's status join) and `ReadChangedSinceAsync` (central's `PullSiteCalls`) wait for every write queued before them; timestamps are stamped when recorded, not when applied; `PurgeTerminalAsync` returns after the purge lands; `DisposeAsync` drains the queue (the sync `Dispose()` does not — host shutdown uses the async path). Accepted: a row queued but unapplied is lost on a hard crash (milliseconds), leaving the tracking row at its previous state until reconciled; the S&F buffer and audit rows are unaffected. The queue is bounded (8 192) with wait backpressure — it never drops. `FlushAsync()` is the explicit barrier for callers that go to the file directly. Design: `docs/plans/2026-09-06-tracking-store-async-writes-design.md`. ### Alarm State (read-only) - `Alarms.CurrentAsync()` — Returns a snapshot of the instance's **current** alarm conditions as `IReadOnlyList` (`Commons.Types.Scripts`), covering computed alarms and mirrored native (OPC UA A&C / MxAccess Gateway) conditions alike. Served by a **local Ask** (`GetAlarmSnapshotRequest`/`GetAlarmSnapshotResponse`) to the script's own Instance Actor — the same mechanism attribute reads use, never a cross-cluster hop. diff --git a/docs/requirements/Component-StoreAndForward.md b/docs/requirements/Component-StoreAndForward.md index 589939ec..a8ff7ae3 100644 --- a/docs/requirements/Component-StoreAndForward.md +++ b/docs/requirements/Component-StoreAndForward.md @@ -53,7 +53,7 @@ For notifications, "delivery" means forwarding the message to the central cluste There are now exactly two ways a notification parks. First, a **corrupt buffered payload** (a row whose stored JSON no longer deserialises to a `NotificationSubmit`, or deserialises to null): the forwarder returns the delivery-handler contract's permanent-failure signal (`false`) so the engine parks the row, preserving the payload for operator forensics. This **supersedes `StoreAndForward-018`**, which discarded such rows by reporting them as delivered — a silent data loss with no parked row, no central `Notifications` row, and no audit trail. Retrying a corrupt payload is pointless (it can never deserialise), so parking, not retrying, is the correct home for it. Second, retry exhaustion under the `DefaultMaxRetries` cap — but because `Notify.Send` enqueues with `maxRetries: 0` (its unbounded-retry contract, above), that second cause is **off for script-produced notifications**, leaving corrupt payload as their only parking cause. (Retry-exhaustion parking remains reachable only for a hypothetical notification enqueued with a positive `maxRetries`.) -For the cached-call categories (`ExternalCall` and `DatabaseWrite`), the operation tracking table is the status record and the S&F buffer is purely the retry mechanism. A cached call that succeeds on its first immediate attempt is written directly as a terminal `Delivered` tracking row and never enters the S&F buffer. When immediate delivery fails transiently, the message is buffered and its tracking row moves to `Pending`/`Retrying`; the buffered message carries its `TrackedOperationId` so the tracking row and the retry record stay linked. When immediate delivery fails **permanently** (e.g. HTTP 4xx), the message is not buffered — the error is returned synchronously to the calling script as before — but the tracking row is written directly as a terminal `Failed` row capturing the error. On every tracking-table status transition the site emits `CachedCallTelemetry` to central. +For the cached-call categories (`ExternalCall` and `DatabaseWrite`), the operation tracking table is the status record and the S&F buffer is purely the retry mechanism. A cached call that succeeds on its first immediate attempt is written directly as a terminal `Delivered` tracking row and never enters the S&F buffer. When immediate delivery fails transiently, the message is buffered and its tracking row moves to `Pending`/`Retrying`; the buffered message carries its `TrackedOperationId` so the tracking row and the retry record stay linked. When immediate delivery fails **permanently** (e.g. HTTP 4xx), the message is not buffered — the error is returned synchronously to the calling script as before — but the tracking row is written directly as a terminal `Failed` row capturing the error. On every tracking-table status transition the site emits `CachedCallTelemetry` to central. The tracking row itself is written through a single-writer queue (Site Runtime, 2026-09-06): the script's `await` on each lifecycle emission returns once the row is queued, not committed, and reads of the table wait for earlier queued writes so status stays read-your-writes — see `Component-SiteRuntime.md` → *Operation Tracking*. Every cached-call outcome maps to a tracking-table state: immediate success → `Delivered`; transient failure → `Pending`/`Retrying`, eventually `Delivered` or `Parked`; permanent failure → terminal `Failed`; operator discard of a parked row → terminal `Discarded`. diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Tracking/OperationTrackingStore.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Tracking/OperationTrackingStore.cs index feb1464a..70a3aedd 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Tracking/OperationTrackingStore.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Tracking/OperationTrackingStore.cs @@ -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 /// (). 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 -/// serialised behind a -/// (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 +/// awaited on the script thread (through CachedCallTelemetryForwarder), +/// and a WAL commit on the consolidated LocalDb file costs a fsync (~30–50 ms on a +/// Docker volume): a single Database.CachedWrite paid three of them, ~108 ms +/// of a 176 ms script run measured on the rig. Since 2026-09-06 the store is a +/// single-writer queue: / +/// / enqueue an +/// ordered work item and return, and one background pump applies each drained batch +/// in one transaction on the single owned . Reads +/// (, ) wait for every +/// write queued before them — a barrier through the same queue — so +/// Tracking.Status(id) keeps read-your-writes and the telemetry actor's status +/// join never sees a stale row. rides the queue too +/// and returns after the purge is applied. Design memo: +/// docs/plans/2026-09-06-tracking-store-async-writes-design.md. +/// +/// +/// 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&F buffer and audit rows are unaffected. The queue is +/// bounded () with wait-mode backpressure — tracking is +/// the status source of truth, so it never drops. /// /// /// All mutations are idempotent / monotonic: is @@ -35,13 +53,47 @@ namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tracking; /// public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable, IDisposable { - // Writer state — one owned SqliteConnection serialised behind - // _writeGate. Readers do NOT share this connection or gate; see GetStatusAsync. + /// Queue bound; past it Record*Async waits (never drops). + internal const int MaxQueuedWrites = 8192; + + /// Most work items applied in one transaction. + 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 _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 _queue = Channel.CreateBounded( + 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; + + /// + /// One queued mutation. runs on the pump inside the batch + /// transaction; (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. + /// + private sealed record WorkItem( + string Description, + Action? 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); } /// @@ -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); } /// @@ -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 ───────────────────────────────────────────────────── + + /// + /// 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". + /// + public Task FlushAsync(CancellationToken ct = default) => WaitForQueuedWritesAsync(ct); + + /// Queues a lifecycle write and returns once it is queued (not applied). + private Task EnqueueAsync(string description, Action apply, CancellationToken ct) + => EnqueueAsync(new WorkItem(description, apply, Completion: null), ct); + + /// Queues a work item; a completion-bearing item is awaited by its owner separately. + 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; + } + } + + /// + /// 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. + /// + 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); + } + + /// + /// 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. + /// + private async Task PumpAsync() + { + var reader = _queue.Reader; + var batch = new List(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 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); } /// @@ -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); } /// @@ -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, /// /// 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. /// /// A that completes when all resources have been released. 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); diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Tracking/OperationTrackingStoreTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Tracking/OperationTrackingStoreTests.cs index c327324b..977d6b65 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Tracking/OperationTrackingStoreTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Tracking/OperationTrackingStoreTests.cs @@ -489,8 +489,8 @@ public class OperationTrackingStoreTests : IDisposable // Backdate the (a) row's TerminalAtUtc to 30 days ago via a direct UPDATE // — RecordTerminalAsync stamps DateTime.UtcNow which we cannot inject. - // The verifier connection shares the same in-memory store thanks to - // mode=memory&cache=shared. + // Flush first: the writes above are queued, and the verifier goes to the file. + await store.FlushAsync(); using (var connection = OpenVerifierConnection(dataSource)) using (var cmd = connection.CreateCommand()) { @@ -531,6 +531,7 @@ public class OperationTrackingStoreTests : IDisposable // can be placed cleanly between rows. (Enqueue stamps DateTime.UtcNow; // we cannot inject the clock, so set the timestamps directly.) var t0 = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc); + await store.FlushAsync(); // writes are queued; the verifier connection needs them applied SetUpdatedAt(dataSource, older, t0); SetUpdatedAt(dataSource, middle, t0.AddMinutes(10)); SetUpdatedAt(dataSource, newer, t0.AddMinutes(20)); @@ -572,6 +573,7 @@ public class OperationTrackingStoreTests : IDisposable var id = TrackedOperationId.New(); ids.Add(id); await store.RecordEnqueueAsync(id, nameof(AuditKind.ApiCallCached), $"T{i}", null, null, null); + await store.FlushAsync(); SetUpdatedAt(dataSource, id, t0.AddMinutes(i)); } @@ -642,6 +644,102 @@ public class OperationTrackingStoreTests : IDisposable cmd.ExecuteNonQuery(); } + // ── 2026-09-06: writes off the script thread (single-writer queue) ───── + // Design: docs/plans/2026-09-06-tracking-store-async-writes-design.md + + [Fact] + public async Task Write_ReturnsWhileAnotherConnectionHoldsTheWriteLock_AndLandsOnceReleased() + { + // The caller (a script thread, via CachedCallTelemetryForwarder) must not pay for the + // SQLite commit. Proof: hold the database's write lock from a second connection so the + // commit CANNOT proceed, and require RecordEnqueueAsync to return anyway. LocalDb's + // busy_timeout is 5 s, so the pump's commit blocks and then succeeds once we release. + var (store, path) = CreateStore(nameof(Write_ReturnsWhileAnotherConnectionHoldsTheWriteLock_AndLandsOnceReleased)); + await using (store) + { + using var locker = OpenVerifierConnection(path); + using (var lockCmd = locker.CreateCommand()) + { + lockCmd.CommandText = "BEGIN IMMEDIATE;"; + lockCmd.ExecuteNonQuery(); + } + + var id = TrackedOperationId.New(); + var sw = System.Diagnostics.Stopwatch.StartNew(); + var write = store.RecordEnqueueAsync(id, "DbWriteCached", "RigDb", "Sim01", "OutboundDemo", "node-a"); + var finished = await Task.WhenAny(write, Task.Delay(TimeSpan.FromSeconds(2))); + sw.Stop(); + Assert.Same(write, finished); // returned although the commit is blocked + Assert.True(sw.Elapsed < TimeSpan.FromSeconds(1), $"write took {sw.Elapsed}"); + + using (var unlock = locker.CreateCommand()) + { + unlock.CommandText = "COMMIT;"; + unlock.ExecuteNonQuery(); + } + + var snapshot = await store.GetStatusAsync(id); + Assert.NotNull(snapshot); + Assert.Equal("Submitted", snapshot!.Status); + } + } + + [Fact] + public async Task Reads_SeeEveryWriteQueuedBeforeThem_InOrder() + { + // Read-your-writes + FIFO: Tracking.Status right after a burst of lifecycle writes must + // observe the LAST one, never a stale intermediate state. + var (store, _) = CreateStore(nameof(Reads_SeeEveryWriteQueuedBeforeThem_InOrder)); + await using (store) + { + var id = TrackedOperationId.New(); + var writes = new List + { + store.RecordEnqueueAsync(id, "ApiCallCached", "ERP.GetOrder", null, null, "node-a"), + }; + for (var i = 1; i <= 100; i++) + writes.Add(store.RecordAttemptAsync(id, "Retrying", retryCount: i, lastError: $"attempt {i}", httpStatus: 503)); + writes.Add(store.RecordTerminalAsync(id, "Delivered", null, 200)); + await Task.WhenAll(writes); + + var snapshot = await store.GetStatusAsync(id); + Assert.NotNull(snapshot); + Assert.Equal("Delivered", snapshot!.Status); + Assert.Equal(100, snapshot.RetryCount); + Assert.NotNull(snapshot.TerminalAtUtc); + + var changed = await store.ReadChangedSinceAsync(DateTime.MinValue, batchSize: 10); + Assert.Single(changed); + Assert.Equal("Delivered", changed[0].Status); + } + } + + [Fact] + public async Task DisposeAsync_DrainsQueuedWrites_BeforeClosingTheConnection() + { + var (store, path) = CreateStore(nameof(DisposeAsync_DrainsQueuedWrites_BeforeClosingTheConnection)); + var writes = Enumerable.Range(0, 200) + .Select(i => store.RecordEnqueueAsync(TrackedOperationId.New(), "ApiCallCached", $"ERP.Op{i}", null, null, "node-a")) + .ToArray(); + await Task.WhenAll(writes); // queued, not necessarily applied + + await store.DisposeAsync(); + + using var verifier = OpenVerifierConnection(path); + using var count = verifier.CreateCommand(); + count.CommandText = "SELECT COUNT(*) FROM OperationTracking;"; + Assert.Equal(200L, (long)count.ExecuteScalar()!); + } + + [Fact] + public async Task Write_AfterDispose_ThrowsObjectDisposed() + { + var (store, _) = CreateStore(nameof(Write_AfterDispose_ThrowsObjectDisposed)); + await store.DisposeAsync(); + await Assert.ThrowsAsync(() => + store.RecordEnqueueAsync(TrackedOperationId.New(), "ApiCallCached", null, null, null, null)); + } + // ── SiteRuntime-024: read/write split + sync-safe Dispose ────────────── [Fact] @@ -815,6 +913,7 @@ public class OperationTrackingStoreTests : IDisposable sourceInstanceId: "I", sourceScript: "S", sourceNode: "node-a"); } + await store.FlushAsync(); var boundary = ForceSharedUpdatedAt(dataSource); // Page 1: no cursor → the first batchSize rows, ordered by TrackedOperationId. @@ -855,6 +954,7 @@ public class OperationTrackingStoreTests : IDisposable sourceInstanceId: "I", sourceScript: "S", sourceNode: "node-a"); } + await store.FlushAsync(); var boundary = ForceSharedUpdatedAt(dataSource); var page1 = await store.ReadChangedSinceAsync(boundary, batchSize);