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

@@ -152,7 +152,7 @@ spec for each is `docs/requirements/Component-<Name>.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.

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.

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<ScriptAlarm>` (`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.

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`.
 

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
 /// (<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
+/// call — but every one of those rows used to be its own autocommit transaction
-/// <see cref="SqliteConnection"/> serialised behind a <see cref="SemaphoreSlim"/>
+/// <b>awaited on the script thread</b> (through <c>CachedCallTelemetryForwarder</c>),
-/// (one async writer at a time) — simpler than a batched-channel pipeline given
+/// and a WAL commit on the consolidated LocalDb file costs a fsync (~30–50 ms on a
-/// the volume; the batched-channel audit-writer design is reserved for the
+/// Docker volume): a single <c>Database.CachedWrite</c> paid three of them, ~108 ms
-/// high-volume audit hot-path.
+/// 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
+    /// <summary>Queue bound; past it <c>Record*Async</c> waits (never drops).</summary>
-    // _writeGate. Readers do NOT share this connection or gate; see GetStatusAsync.
+    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);
+        // Stamped when RECORDED, not when the batch lands.
-        try
-        {
-            ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposeState) != 0, this);
-
         var now = DateTime.UtcNow.ToString("o", CultureInfo.InvariantCulture);
+        var idText = id.ToString();
 
-            using var cmd = _writeConnection.CreateCommand();
+        await EnqueueAsync($"enqueue {idText}", (conn, tx) =>
+        {
+            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();
-        }
+        }, ct).ConfigureAwait(false);
-        finally
-        {
-            _writeGate.Release();
-        }
     }
 
     /// <inheritdoc/>
@@ -153,14 +202,13 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable,
     {
         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);
+        var idText = id.ToString();
 
-            using var cmd = _writeConnection.CreateCommand();
+        await EnqueueAsync($"attempt {idText}", (conn, tx) =>
+        {
+            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();
-        }
+        }, ct).ConfigureAwait(false);
-        finally
-        {
-            _writeGate.Release();
-        }
     }
 
     /// <inheritdoc/>
@@ -198,14 +242,13 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable,
     {
         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);
+        var idText = id.ToString();
 
-            using var cmd = _writeConnection.CreateCommand();
+        await EnqueueAsync($"terminal {idText}", (conn, tx) =>
+        {
+            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,25 +261,181 @@ 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();
+        }, 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
+        {
+            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/>
     public async Task<TrackingStatusSnapshot?> GetStatusAsync(
         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);
+        var threshold = olderThanUtc.ToString("o", CultureInfo.InvariantCulture);
-        try
+        // 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 = conn.CreateCommand();
-
+            cmd.Transaction = tx;
-            using var cmd = _writeConnection.CreateCommand();
             // 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(
+            cmd.Parameters.AddWithValue("$threshold", threshold);
-                "$threshold",
-                olderThanUtc.ToString("o", CultureInfo.InvariantCulture));
-
             cmd.ExecuteNonQuery();
-        }
+        }, new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously));
-        finally
+
-        {
+        await EnqueueAsync(purge, ct).ConfigureAwait(false);
-            _writeGate.Release();
+        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
+    /// Drains the write queue first (completes it and awaits the pump), so every
-    /// connection, so a write currently executing a SqliteCommand completes
+    /// write that was queued before disposal is applied before the connection is
-    /// before the connection is freed.
+    /// 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
+        // Drain: no new item can be queued past this point (_disposeState is set
-        // no new write can acquire the gate because _disposeState is set, so
+        // and the writer is completed), so awaiting the pump means every queued
-        // the next ThrowIf check in each writer raises ObjectDisposedException.
+        // 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);

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
+        // Flush first: the writes above are queued, and the verifier goes to the file.
-        // mode=memory&cache=shared.
+        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<Task>
+            {
+                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<ObjectDisposedException>(() =>
+            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);