Joseph Doherty
763 lines
34 KiB
C#
763 lines
34 KiB
C#
using System.Globalization;
|
||
using System.Threading.Channels;
|
||
using Microsoft.Data.Sqlite;
|
||
using Microsoft.Extensions.Logging;
|
||
using ZB.MOM.WW.LocalDb;
|
||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces;
|
||
using ZB.MOM.WW.ScadaBridge.Commons.Types;
|
||
|
||
namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tracking;
|
||
|
||
/// <summary>
|
||
/// Site-local SQLite source-of-truth for cached-operation tracking — the row
|
||
/// that <c>Tracking.Status(TrackedOperationId)</c> reads.
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// <para>
|
||
/// One row per <see cref="TrackedOperationId"/>; lifecycle is
|
||
/// <c>Submitted → Retrying → Delivered / Parked / Failed / Discarded</c>; terminal
|
||
/// rows are purged after the configured retention window
|
||
/// (<see cref="OperationTrackingOptions.RetentionDays"/>). Volume is bounded —
|
||
/// only cached calls produce rows, and only a handful of lifecycle events per
|
||
/// call — 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&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
|
||
/// <c>INSERT OR IGNORE</c>, <see cref="RecordAttemptAsync"/> filters out terminal
|
||
/// rows in the <c>WHERE</c> clause, and <see cref="RecordTerminalAsync"/> only
|
||
/// fires on rows that haven't terminated yet (first-write-wins). This makes the
|
||
/// store safe under the at-least-once semantics of the site→central telemetry
|
||
/// path.
|
||
/// </para>
|
||
/// </remarks>
|
||
public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable, IDisposable
|
||
{
|
||
/// <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
|
||
// invoked concurrently (e.g. host shutdown bridging both). 0 = live,
|
||
// 1 = disposed. Read by other methods via Volatile.Read after the gate is
|
||
// taken; they raise ObjectDisposedException when set.
|
||
private int _disposeState;
|
||
|
||
/// <summary>
|
||
/// Initializes the tracking store over the consolidated site database and applies the schema.
|
||
/// </summary>
|
||
/// <param name="localDb">
|
||
/// The consolidated site database. Every connection it hands out is already open and carries
|
||
/// the per-connection pragmas plus the <c>zb_hlc_next()</c> UDF that
|
||
/// <c>OperationTracking</c>'s capture triggers call — which is exactly why the store no longer
|
||
/// opens its own <see cref="SqliteConnection"/> from a connection string. A raw connection
|
||
/// would lack the UDF and every write to the replicated table would fail closed.
|
||
/// </param>
|
||
/// <param name="logger">Logger for diagnostics.</param>
|
||
/// <remarks>
|
||
/// <see cref="OperationTrackingOptions.ConnectionString"/> no longer feeds this store — the
|
||
/// database location is <c>LocalDb:Path</c>. The option remains bound for the purge/retention
|
||
/// settings that share the class.
|
||
/// </remarks>
|
||
public OperationTrackingStore(
|
||
ILocalDb localDb,
|
||
ILogger<OperationTrackingStore> logger)
|
||
{
|
||
ArgumentNullException.ThrowIfNull(localDb);
|
||
ArgumentNullException.ThrowIfNull(logger);
|
||
|
||
_logger = logger;
|
||
_localDb = localDb;
|
||
// Already open — CreateConnection returns a live, pragma-configured,
|
||
// UDF-registered connection. Calling Open() on it again would throw.
|
||
_writeConnection = localDb.CreateConnection();
|
||
InitializeSchema();
|
||
_pump = Task.Run(PumpAsync);
|
||
}
|
||
|
||
// Schema lives in OperationTrackingSchema so the Host's AddZbLocalDb onReady
|
||
// callback can create this table in the consolidated site database before
|
||
// RegisterReplicated installs its capture triggers. In the host this call is
|
||
// therefore always a no-op — onReady runs while ILocalDb is being constructed,
|
||
// which is strictly before this constructor can receive it. It stays because it
|
||
// keeps a directly-constructed store (tests, tooling) self-sufficient, and the
|
||
// DDL is idempotent.
|
||
private void InitializeSchema()
|
||
{
|
||
OperationTrackingSchema.Apply(_writeConnection);
|
||
using var cmd = _writeConnection.CreateCommand();
|
||
cmd.CommandText = CentralAckSchemaSql;
|
||
cmd.ExecuteNonQuery();
|
||
}
|
||
|
||
/// <summary>
|
||
/// Node-local ledger of the tracking-row version central last acknowledged per
|
||
/// operation (review 2026-10-21 06/PERF-2). Deliberately a SEPARATE table, never a
|
||
/// column on <c>OperationTracking</c>: that table is LocalDb-replicated and its
|
||
/// capture triggers are built from its column list, while acks are per-node fact —
|
||
/// only the node that pushed the telemetry saw central accept it. Like
|
||
/// <c>mirrored_secrets</c>, it is not in <c>SiteLocalDbSetup.ReplicatedTables</c>.
|
||
/// <para>
|
||
/// Failover: the standby holds no ack rows, so after it takes over its
|
||
/// reconciliation read serves every row once (bounded by central's cursor), and
|
||
/// central's idempotent monotonic upsert absorbs the duplicates. Pushes from the new
|
||
/// active node rebuild the ledger from then on.
|
||
/// </para>
|
||
/// </summary>
|
||
private const string CentralAckSchemaSql = """
|
||
CREATE TABLE IF NOT EXISTS OperationTrackingCentralAck (
|
||
TrackedOperationId TEXT NOT NULL PRIMARY KEY,
|
||
AckedUpdatedAtUtc TEXT NOT NULL
|
||
);
|
||
""";
|
||
|
||
/// <inheritdoc/>
|
||
public async Task RecordEnqueueAsync(
|
||
TrackedOperationId id,
|
||
string kind,
|
||
string? targetSummary,
|
||
string? sourceInstanceId,
|
||
string? sourceScript,
|
||
string? sourceNode,
|
||
CancellationToken ct = default)
|
||
{
|
||
ArgumentNullException.ThrowIfNull(kind);
|
||
|
||
// 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) =>
|
||
{
|
||
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 = """
|
||
INSERT OR IGNORE INTO OperationTracking (
|
||
TrackedOperationId, Kind, TargetSummary, Status,
|
||
RetryCount, LastError, HttpStatus,
|
||
CreatedAtUtc, UpdatedAtUtc, TerminalAtUtc,
|
||
SourceInstanceId, SourceScript, SourceNode
|
||
) VALUES (
|
||
$id, $kind, $targetSummary, $status,
|
||
0, NULL, NULL,
|
||
$now, $now, NULL,
|
||
$sourceInstanceId, $sourceScript, $sourceNode
|
||
);
|
||
""";
|
||
cmd.Parameters.AddWithValue("$id", idText);
|
||
cmd.Parameters.AddWithValue("$kind", kind);
|
||
cmd.Parameters.AddWithValue("$targetSummary", (object?)targetSummary ?? DBNull.Value);
|
||
cmd.Parameters.AddWithValue("$status", "Submitted");
|
||
cmd.Parameters.AddWithValue("$now", now);
|
||
cmd.Parameters.AddWithValue("$sourceInstanceId", (object?)sourceInstanceId ?? DBNull.Value);
|
||
cmd.Parameters.AddWithValue("$sourceScript", (object?)sourceScript ?? DBNull.Value);
|
||
cmd.Parameters.AddWithValue("$sourceNode", (object?)sourceNode ?? DBNull.Value);
|
||
|
||
cmd.ExecuteNonQuery();
|
||
}, ct).ConfigureAwait(false);
|
||
}
|
||
|
||
/// <inheritdoc/>
|
||
public async Task RecordAttemptAsync(
|
||
TrackedOperationId id,
|
||
string status,
|
||
int retryCount,
|
||
string? lastError,
|
||
int? httpStatus,
|
||
CancellationToken ct = default)
|
||
{
|
||
ArgumentNullException.ThrowIfNull(status);
|
||
|
||
var now = DateTime.UtcNow.ToString("o", CultureInfo.InvariantCulture);
|
||
var idText = id.ToString();
|
||
|
||
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 = """
|
||
UPDATE OperationTracking
|
||
SET Status = $status,
|
||
RetryCount = $retryCount,
|
||
LastError = $lastError,
|
||
HttpStatus = $httpStatus,
|
||
UpdatedAtUtc = $now
|
||
WHERE TrackedOperationId = $id
|
||
AND TerminalAtUtc IS NULL;
|
||
""";
|
||
cmd.Parameters.AddWithValue("$id", idText);
|
||
cmd.Parameters.AddWithValue("$status", status);
|
||
cmd.Parameters.AddWithValue("$retryCount", retryCount);
|
||
cmd.Parameters.AddWithValue("$lastError", (object?)lastError ?? DBNull.Value);
|
||
cmd.Parameters.AddWithValue("$httpStatus", (object?)httpStatus ?? DBNull.Value);
|
||
cmd.Parameters.AddWithValue("$now", now);
|
||
|
||
cmd.ExecuteNonQuery();
|
||
}, ct).ConfigureAwait(false);
|
||
}
|
||
|
||
/// <inheritdoc/>
|
||
public async Task RecordTerminalAsync(
|
||
TrackedOperationId id,
|
||
string status,
|
||
string? lastError,
|
||
int? httpStatus,
|
||
CancellationToken ct = default)
|
||
{
|
||
ArgumentNullException.ThrowIfNull(status);
|
||
|
||
var now = DateTime.UtcNow.ToString("o", CultureInfo.InvariantCulture);
|
||
var idText = id.ToString();
|
||
|
||
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 = """
|
||
UPDATE OperationTracking
|
||
SET Status = $status,
|
||
LastError = $lastError,
|
||
HttpStatus = $httpStatus,
|
||
UpdatedAtUtc = $now,
|
||
TerminalAtUtc = $now
|
||
WHERE TrackedOperationId = $id
|
||
AND TerminalAtUtc IS NULL;
|
||
""";
|
||
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);
|
||
}
|
||
|
||
/// <inheritdoc/>
|
||
public async Task RecordCentralAcksAsync(
|
||
IReadOnlyDictionary<TrackedOperationId, DateTime> ackedVersions,
|
||
CancellationToken ct = default)
|
||
{
|
||
ArgumentNullException.ThrowIfNull(ackedVersions);
|
||
if (ackedVersions.Count == 0)
|
||
{
|
||
return;
|
||
}
|
||
|
||
// Rendered exactly as the write path stamps UpdatedAtUtc ("o", UTC 'Z'), so the
|
||
// read filter's text equality compares like with like.
|
||
var acks = ackedVersions
|
||
.Select(a => (Id: a.Key.ToString(), Version: DateTime
|
||
.SpecifyKind(a.Value, DateTimeKind.Utc)
|
||
.ToString("o", CultureInfo.InvariantCulture)))
|
||
.ToList();
|
||
|
||
// Rides the single-writer queue like a lifecycle write (enqueue and return) —
|
||
// the ack is best-effort, and FIFO keeps it behind the writes it describes.
|
||
await EnqueueAsync($"central-ack x{acks.Count}", (conn, tx) =>
|
||
{
|
||
using var cmd = conn.CreateCommand();
|
||
cmd.Transaction = tx;
|
||
// Monotonic: an older ack never rolls a newer one back. ISO-8601 "o" text
|
||
// orders lexicographically.
|
||
cmd.CommandText = """
|
||
INSERT INTO OperationTrackingCentralAck (TrackedOperationId, AckedUpdatedAtUtc)
|
||
VALUES ($id, $version)
|
||
ON CONFLICT (TrackedOperationId) DO UPDATE
|
||
SET AckedUpdatedAtUtc = excluded.AckedUpdatedAtUtc
|
||
WHERE excluded.AckedUpdatedAtUtc > AckedUpdatedAtUtc;
|
||
""";
|
||
var idParam = cmd.Parameters.Add("$id", SqliteType.Text);
|
||
var versionParam = cmd.Parameters.Add("$version", SqliteType.Text);
|
||
foreach (var (id, version) in acks)
|
||
{
|
||
idParam.Value = id;
|
||
versionParam.Value = version;
|
||
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>
|
||
/// <param name="ct">Token to cancel the wait.</param>
|
||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||
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
|
||
// block status queries. It comes from the same ILocalDb as the writer;
|
||
// SQLite handles cross-connection isolation natively (readers see a WAL
|
||
// snapshot). Mirrors the SiteStorageService precedent. Already open — do
|
||
// not call OpenAsync on it.
|
||
await using var readConnection = _localDb.CreateConnection();
|
||
|
||
await using var cmd = readConnection.CreateCommand();
|
||
cmd.CommandText = """
|
||
SELECT TrackedOperationId, Kind, TargetSummary, Status,
|
||
RetryCount, LastError, HttpStatus,
|
||
CreatedAtUtc, UpdatedAtUtc, TerminalAtUtc,
|
||
SourceInstanceId, SourceScript, SourceNode
|
||
FROM OperationTracking
|
||
WHERE TrackedOperationId = $id;
|
||
""";
|
||
cmd.Parameters.AddWithValue("$id", id.ToString());
|
||
|
||
await using var reader = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
|
||
if (!await reader.ReadAsync(ct).ConfigureAwait(false))
|
||
{
|
||
return null;
|
||
}
|
||
|
||
return new TrackingStatusSnapshot(
|
||
Id: TrackedOperationId.Parse(reader.GetString(0)),
|
||
Kind: reader.GetString(1),
|
||
TargetSummary: reader.IsDBNull(2) ? null : reader.GetString(2),
|
||
Status: reader.GetString(3),
|
||
RetryCount: reader.GetInt32(4),
|
||
LastError: reader.IsDBNull(5) ? null : reader.GetString(5),
|
||
HttpStatus: reader.IsDBNull(6) ? null : reader.GetInt32(6),
|
||
CreatedAtUtc: ParseUtc(reader.GetString(7)),
|
||
UpdatedAtUtc: ParseUtc(reader.GetString(8)),
|
||
TerminalAtUtc: reader.IsDBNull(9) ? null : ParseUtc(reader.GetString(9)),
|
||
SourceInstanceId: reader.IsDBNull(10) ? null : reader.GetString(10),
|
||
SourceScript: reader.IsDBNull(11) ? null : reader.GetString(11),
|
||
SourceNode: reader.IsDBNull(12) ? null : reader.GetString(12));
|
||
}
|
||
|
||
/// <inheritdoc/>
|
||
public async Task PurgeTerminalAsync(
|
||
DateTime olderThanUtc,
|
||
CancellationToken ct = default)
|
||
{
|
||
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) =>
|
||
{
|
||
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.
|
||
// The node-local ack ledger is pruned in the same transaction so it never
|
||
// outlives the rows it describes (it also drops acks for rows the peer's
|
||
// purge removed through replication).
|
||
cmd.CommandText = """
|
||
DELETE FROM OperationTracking
|
||
WHERE TerminalAtUtc IS NOT NULL
|
||
AND TerminalAtUtc < $threshold;
|
||
DELETE FROM OperationTrackingCentralAck
|
||
WHERE NOT EXISTS (
|
||
SELECT 1 FROM OperationTracking t
|
||
WHERE t.TrackedOperationId = OperationTrackingCentralAck.TrackedOperationId);
|
||
""";
|
||
cmd.Parameters.AddWithValue("$threshold", threshold);
|
||
cmd.ExecuteNonQuery();
|
||
}, new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously));
|
||
|
||
await EnqueueAsync(purge, ct).ConfigureAwait(false);
|
||
await purge.Completion!.Task.WaitAsync(ct).ConfigureAwait(false);
|
||
}
|
||
|
||
/// <inheritdoc/>
|
||
public async Task<IReadOnlyList<SiteCallOperational>> ReadChangedSinceAsync(
|
||
DateTime sinceUtc,
|
||
int batchSize,
|
||
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
|
||
// fresh, ungated read connection so a long-running write never blocks
|
||
// central's PullSiteCalls. The query is a bounded, ordered scan served by
|
||
// the standalone IX_OperationTracking_UpdatedAt index — UpdatedAtUtc is
|
||
// the cursor. (The composite (Status, UpdatedAtUtc) index cannot satisfy a
|
||
// status-less UpdatedAtUtc range scan; this dedicated index does.)
|
||
// Already open — do not call OpenAsync on it.
|
||
await using var readConnection = _localDb.CreateConnection();
|
||
|
||
await using var cmd = readConnection.CreateCommand();
|
||
|
||
// Composite (UpdatedAtUtc, TrackedOperationId) keyset. Ordering
|
||
// is ALWAYS deterministic (UpdatedAtUtc ASC, TrackedOperationId ASC) so the
|
||
// cursor is well-defined even when many rows share one UpdatedAtUtc.
|
||
// • afterId present → strict keyset: skip everything up to and including
|
||
// (sinceUtc, afterId). Without this a page fully inside one UpdatedAtUtc
|
||
// instant re-reads forever (the plain >= resume never advances past the
|
||
// tie). TrackedOperationId is the "D" GUID form on both sides, so the
|
||
// TEXT comparison is a stable total order.
|
||
// • afterId null/empty → legacy inclusive >= sinceUtc (an older central
|
||
// that does not send a cursor is byte-for-byte unaffected).
|
||
var useKeyset = !string.IsNullOrEmpty(afterId);
|
||
var predicate = useKeyset
|
||
? "UpdatedAtUtc > $since OR (UpdatedAtUtc = $since AND TrackedOperationId > $afterId)"
|
||
: "UpdatedAtUtc >= $since";
|
||
// Rows whose CURRENT version central already acknowledged through the
|
||
// cached-telemetry push are skipped (review 2026-10-21 06/PERF-2): central holds
|
||
// exactly what this read would send, so serving them only replays every
|
||
// lifecycle step a second time — and, after a central failover reset the
|
||
// cursor, the whole retention window. Equality (not >=) on the version is the
|
||
// safe direction: any later change, even one stamped by a clock that stepped
|
||
// backwards, no longer matches and is served.
|
||
cmd.CommandText = $"""
|
||
SELECT TrackedOperationId, Kind, TargetSummary, Status,
|
||
RetryCount, LastError, HttpStatus,
|
||
CreatedAtUtc, UpdatedAtUtc, TerminalAtUtc, SourceNode
|
||
FROM OperationTracking
|
||
WHERE ({predicate})
|
||
AND NOT EXISTS (
|
||
SELECT 1 FROM OperationTrackingCentralAck a
|
||
WHERE a.TrackedOperationId = OperationTracking.TrackedOperationId
|
||
AND a.AckedUpdatedAtUtc = OperationTracking.UpdatedAtUtc)
|
||
ORDER BY UpdatedAtUtc ASC, TrackedOperationId ASC
|
||
LIMIT $batchSize;
|
||
""";
|
||
// Force UTC kind before formatting so the cursor's "o" text matches the
|
||
// 'Z'-suffixed round-trip form the write path persists (DateTime.UtcNow
|
||
// .ToString("o")). A first-cycle DateTime.MinValue arrives Unspecified —
|
||
// without this its "o" rendering would lack the 'Z', and the SQLite text
|
||
// compare against 'Z'-suffixed stored values would be subtly inconsistent.
|
||
var sinceText = DateTime
|
||
.SpecifyKind(sinceUtc, DateTimeKind.Utc)
|
||
.ToString("o", CultureInfo.InvariantCulture);
|
||
cmd.Parameters.AddWithValue("$since", sinceText);
|
||
cmd.Parameters.AddWithValue("$batchSize", batchSize);
|
||
if (useKeyset)
|
||
{
|
||
cmd.Parameters.AddWithValue("$afterId", afterId!);
|
||
}
|
||
|
||
var rows = new List<SiteCallOperational>();
|
||
await using var reader = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false);
|
||
while (await reader.ReadAsync(ct).ConfigureAwait(false))
|
||
{
|
||
var kind = reader.GetString(1);
|
||
rows.Add(new SiteCallOperational(
|
||
TrackedOperationId: TrackedOperationId.Parse(reader.GetString(0)),
|
||
Channel: KindToChannel(kind),
|
||
Target: reader.IsDBNull(2) ? string.Empty : reader.GetString(2),
|
||
// The site id is not a tracking-store column; the central client
|
||
// re-stamps SourceSite from the siteId it dialed.
|
||
SourceSite: string.Empty,
|
||
SourceNode: reader.IsDBNull(10) ? null : reader.GetString(10),
|
||
Status: reader.GetString(3),
|
||
RetryCount: reader.GetInt32(4),
|
||
LastError: reader.IsDBNull(5) ? null : reader.GetString(5),
|
||
HttpStatus: reader.IsDBNull(6) ? null : reader.GetInt32(6),
|
||
CreatedAtUtc: ParseUtc(reader.GetString(7)),
|
||
UpdatedAtUtc: ParseUtc(reader.GetString(8)),
|
||
TerminalAtUtc: reader.IsDBNull(9) ? null : ParseUtc(reader.GetString(9))));
|
||
}
|
||
|
||
return rows;
|
||
}
|
||
|
||
// Cached-call Kind → SiteCalls Channel. Only ApiCallCached / DbWriteCached
|
||
// ever reach the tracking store (RecordEnqueueAsync is the cached-call
|
||
// entry point); DbWriteCached maps to DbOutbound, everything else to the
|
||
// ApiOutbound default. Mirrors CachedCallLifecycleBridge's channel handling.
|
||
private static string KindToChannel(string kind) => kind switch
|
||
{
|
||
nameof(Commons.Types.Enums.AuditKind.DbWriteCached) => nameof(Commons.Types.Enums.AuditChannel.DbOutbound),
|
||
_ => nameof(Commons.Types.Enums.AuditChannel.ApiOutbound),
|
||
};
|
||
|
||
private static DateTime ParseUtc(string raw)
|
||
{
|
||
return DateTime.Parse(
|
||
raw,
|
||
CultureInfo.InvariantCulture,
|
||
DateTimeStyles.RoundtripKind);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Synchronously disposes the tracking store and its SQLite connection.
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// The pump runs on the thread pool and its drain does not capture a caller's
|
||
/// synchronization context. Join it before releasing SQLite resources, exactly
|
||
/// as the asynchronous path does; never close a live writer transaction.
|
||
/// </remarks>
|
||
public void Dispose() => DisposeAsync().AsTask().GetAwaiter().GetResult();
|
||
|
||
/// <summary>
|
||
/// Asynchronously disposes the tracking store and its SQLite connection.
|
||
/// 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()
|
||
{
|
||
if (Interlocked.Exchange(ref _disposeState, 1) != 0)
|
||
{
|
||
return;
|
||
}
|
||
|
||
// 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);
|
||
}
|
||
catch (ObjectDisposedException)
|
||
{
|
||
// Race with another disposer that already disposed the gate — the
|
||
// _disposeState exchange above should prevent this, but be defensive.
|
||
}
|
||
|
||
try
|
||
{
|
||
_writeConnection.Dispose();
|
||
}
|
||
finally
|
||
{
|
||
try { _writeGate.Release(); } catch (ObjectDisposedException) { }
|
||
_writeGate.Dispose();
|
||
}
|
||
|
||
GC.SuppressFinalize(this);
|
||
}
|
||
}
|