Joseph Doherty
fd5e023d08
F1 (HIGH) DebugStreamBridgeActor: the 5-minute orphan net measured the MAILBOX (SetReceiveTimeout), and once stream events were correctly marked INotInfluenceReceiveTimeout nothing recurring reset it — the snapshot lands once and GrpcStreamStable once — so every healthy session self-terminated at ~6 min with a false "Site disconnected". Replaced with a periodic self-tick (ConsumerLivenessCheckInterval, 30s) over a consumer-last-seen stamp renewed only by DebugStreamConsumerAlive, which DebugStreamService Tells on a shared timer to every session still in its registry (holding a session there IS "a consumer is attached" — both the Blazor view and the SignalR hub release it on dispose/disconnect, and it works headless). Reverting the wrapper was rejected: it would restore the quiet-instance orphan bug. F2 (MED) SiteStreamGrpcClient: the RpcException(Cancelled) filter now requires cts.IsCancellationRequested. A peer-originated / channel-dispose Cancelled fired none of onError/onCompleted/onConnected, leaving SiteAlarmAggregatorActor with _streamDown=false forever (IsLive stuck true, reconcile reopen guard never fired). F3 (MED) SiteStreamGrpcClient: a header TIMEOUT is no longer reported as connected — that shape is exactly what an unreachable site produces, and it cleared _streamDown, consumed _seedOnConnect and launched a full snapshot fan-out at a dead site. AwaitHeadersAsync returns bool; the first received event is the fallback connected signal, fired at most once from headers OR first event. F4 (LOW-MED) SqliteAuditWriter.MarkReconciledUpToAsync: the blanket below-cursor UPDATE retired late-stamped inserts that were never served (then age-purged — silent loss). The flip is now bounded by insertion order: a Pending row retires only if its rowid is at or below the high-water mark of rows this instance has served from ReadPendingSinceAsync (clamped on purge, since SQLite reuses rowids); Forwarded rows are exempt (central ACKed them over the push path). At-least-once is unchanged. F5 (LOW) Documented the liveness dependency (a served row never covered by a later cursor stays Pending forever; PurgeExpiredAsync never purges Pending) in ISiteAuditQueue + Component-AuditLog.md, and added a cheap site-health signal: SiteAuditBacklogReporter logs a rate-limited warning when the existing oldest-pending metric exceeds 24h. F6 (MED) SiteAlarmAggregatorActor: _fanoutSinceLastTick was armed by the reconcile's OWN fan-out, so steady state ran fan-out→skip→fan-out→skip — one reconcile per 2x interval (120s), halving the not-reporting refresh and the alarm reconcile backstop. The skip is now armed only by connect/failover-driven seeds (initial, _seedOnConnect, and a re-seed queued behind one). Tests: Communication.Tests 691 passed (+13), AuditLog.Tests 382 passed (+5).
1328 lines
66 KiB
C#
1328 lines
66 KiB
C#
using System.Threading.Channels;
|
|
using Microsoft.Data.Sqlite;
|
|
using Microsoft.Extensions.Logging;
|
|
using Microsoft.Extensions.Options;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
|
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
|
using AuditEvent = ZB.MOM.WW.Audit.AuditEvent;
|
|
using AuditOutcome = ZB.MOM.WW.Audit.AuditOutcome;
|
|
|
|
namespace ZB.MOM.WW.ScadaBridge.AuditLog.Site;
|
|
|
|
/// <summary>
|
|
/// Site-side SQLite hot-path writer for Audit Log events. Mirrors the
|
|
/// <see cref="ZB.MOM.WW.ScadaBridge.SiteEventLogging.SiteEventLogger"/> design — a single
|
|
/// owned <see cref="SqliteConnection"/> serialised behind a write lock, fed by a
|
|
/// bounded <see cref="Channel{T}"/> drained on a dedicated background writer
|
|
/// task — so script-thread callers never block on disk I/O.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// <para>
|
|
/// <b>Two-table schema.</b> The site store is now two tables:
|
|
/// the append-only canonical <c>audit_event</c> (the 10 canonical
|
|
/// <see cref="AuditEvent"/> fields stored directly — NO 24-column decompose) and
|
|
/// the mutable operational <c>audit_forward_state</c> sidecar that carries the
|
|
/// forwarding lifecycle (<see cref="AuditForwardState"/>), a duplicated
|
|
/// <c>OccurredAtUtc</c> for the drain index range-scan, a precomputed
|
|
/// <c>IsCachedKind</c> flag that drives the cached/non-cached drain split without
|
|
/// re-parsing <c>DetailsJson</c> on the read hot-path, plus attempt bookkeeping.
|
|
/// </para>
|
|
/// <para>
|
|
/// <b>Ephemeral reset.</b> The site SQLite store is ephemeral (≈7-day retention,
|
|
/// recreated per deployment), so this schema change is an in-place RESET: the new
|
|
/// tables are created and the old single 24-column <c>AuditLog</c> table is
|
|
/// DROP-ped if present. No SQLite data migration is performed (and none is
|
|
/// needed) — any rows in the old <c>AuditLog</c> table are within the retention
|
|
/// window and are discarded by the drop.
|
|
/// </para>
|
|
/// <para>
|
|
/// Site rows always carry <see cref="AuditForwardState.Pending"/> on first
|
|
/// insert; the central row-shape's <c>IngestedAtUtc</c> is a DetailsJson field
|
|
/// stamped by central on ingest, not a site column.
|
|
/// </para>
|
|
/// </remarks>
|
|
public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable, IDisposable
|
|
{
|
|
// Microsoft.Data.Sqlite reports a generic SQLITE_CONSTRAINT (error code 19)
|
|
// on a PRIMARY KEY violation; the extended subcode 1555 (SQLITE_CONSTRAINT_PRIMARYKEY)
|
|
// is exposed via SqliteException.SqliteExtendedErrorCode but isn't reliably
|
|
// surfaced across all SQLite builds. We treat any constraint error on insert
|
|
// as a duplicate-eventid race and swallow it (first-write-wins) — the PRIMARY
|
|
// KEY on audit_event.EventId is the constraint that fires first, so this scope
|
|
// is precise (the sidecar insert for the same EventId is in the same
|
|
// transaction and never reached once audit_event's insert throws).
|
|
private const int SqliteErrorConstraint = 19;
|
|
|
|
// Max EventIds bound into a single IN (...) UPDATE by MarkForwardedAsync /
|
|
// MarkReconciledAsync. SQLite caps bound parameters per statement (999 on older
|
|
// builds, SQLITE_MAX_VARIABLE_NUMBER); a large drain batch would otherwise blow the
|
|
// ceiling as batch sizes grow. Chunk the id list into ≤500-id slices, all inside one
|
|
// transaction so the batch stays atomic (arch-review 04, P8).
|
|
private const int MarkForwardStateChunkSize = 500;
|
|
|
|
private readonly SqliteConnection _connection;
|
|
// Dedicated read-only connection used by GetBacklogStatsAsync,
|
|
// ReadPendingAsync, ReadPendingSinceAsync, and ReadForwardedAsync so a slow
|
|
// backlog scan (COUNT(*) over hundreds of thousands of Pending rows under a
|
|
// central outage) never parks the hot-path writer behind _writeLock.
|
|
// SQLite-with-WAL allows a second connection on the same file to read
|
|
// concurrently with the writer; the writer's WAL pragma is set in
|
|
// InitializeSchema before this connection is opened. The reader connection
|
|
// has its own _readLock because SqliteConnection itself is not thread-safe
|
|
// even in read-only mode — multiple read callers can otherwise interleave
|
|
// commands on the shared connection.
|
|
private readonly SqliteConnection _readConnection;
|
|
private readonly object _readLock = new();
|
|
private readonly SqliteAuditWriterOptions _options;
|
|
private readonly ILogger<SqliteAuditWriter> _logger;
|
|
private readonly INodeIdentityProvider _nodeIdentity;
|
|
private readonly object _writeLock = new();
|
|
private readonly Channel<PendingAuditEvent> _writeQueue;
|
|
private readonly Task _writerLoop;
|
|
private bool _disposed;
|
|
|
|
/// <summary>Initializes a new instance of the SqliteAuditWriter class.</summary>
|
|
/// <param name="options">Configuration options for the audit writer.</param>
|
|
/// <param name="logger">Logger instance.</param>
|
|
/// <param name="nodeIdentity">Node identity provider.</param>
|
|
/// <param name="connectionStringOverride">Optional connection string override.</param>
|
|
public SqliteAuditWriter(
|
|
IOptions<SqliteAuditWriterOptions> options,
|
|
ILogger<SqliteAuditWriter> logger,
|
|
INodeIdentityProvider nodeIdentity,
|
|
string? connectionStringOverride = null)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(options);
|
|
ArgumentNullException.ThrowIfNull(logger);
|
|
ArgumentNullException.ThrowIfNull(nodeIdentity);
|
|
|
|
_options = options.Value;
|
|
_logger = logger;
|
|
_nodeIdentity = nodeIdentity;
|
|
|
|
var connectionString = connectionStringOverride
|
|
?? $"Data Source={_options.DatabasePath};Cache=Shared";
|
|
_connection = new SqliteConnection(connectionString);
|
|
_connection.Open();
|
|
|
|
InitializeSchema();
|
|
|
|
// Open a second connection for read-only callers
|
|
// (GetBacklogStatsAsync, ReadPendingAsync, ReadPendingSinceAsync,
|
|
// ReadForwardedAsync). InitializeSchema set journal_mode=WAL on the
|
|
// writer connection, which is a database-level setting that persists
|
|
// for the file — subsequent connections to the same file see WAL and
|
|
// can read concurrently with the writer without taking _writeLock.
|
|
// Reuse the same connection string so the read connection sees the
|
|
// same Data Source / Cache settings as the writer.
|
|
_readConnection = new SqliteConnection(connectionString);
|
|
_readConnection.Open();
|
|
|
|
// PRAGMA foreign_keys is a per-connection setting. Set it on the read
|
|
// connection as well so that any future read-path change (e.g. a
|
|
// DELETE that may be added later) also benefits from FK enforcement.
|
|
// Pure SELECT queries are unaffected — this is defensive belt-and-
|
|
// suspenders for the read connection.
|
|
using (var pragmaCmd = _readConnection.CreateCommand())
|
|
{
|
|
pragmaCmd.CommandText = "PRAGMA foreign_keys = ON";
|
|
pragmaCmd.ExecuteNonQuery();
|
|
}
|
|
|
|
_writeQueue = Channel.CreateBounded<PendingAuditEvent>(
|
|
new BoundedChannelOptions(_options.ChannelCapacity)
|
|
{
|
|
// The hot-path enqueue must back-pressure if the background
|
|
// writer falls behind; a higher-level fallback
|
|
// handles truly catastrophic primary failure with a drop-oldest
|
|
// ring buffer.
|
|
FullMode = BoundedChannelFullMode.Wait,
|
|
SingleReader = true,
|
|
SingleWriter = false,
|
|
});
|
|
_writerLoop = Task.Run(ProcessWriteQueueAsync);
|
|
}
|
|
|
|
private void InitializeSchema()
|
|
{
|
|
// auto_vacuum must be set before any table is created for it to take
|
|
// effect on a fresh database. INCREMENTAL lets the `PRAGMA
|
|
// incremental_vacuum` run at the end of PurgeExpiredAsync shrink the file
|
|
// after the ~7-day retention purge — see alog.md §10.
|
|
using (var pragmaCmd = _connection.CreateCommand())
|
|
{
|
|
pragmaCmd.CommandText = "PRAGMA auto_vacuum = INCREMENTAL";
|
|
pragmaCmd.ExecuteNonQuery();
|
|
}
|
|
|
|
// Enable WAL so a second connection on the same file can
|
|
// serve read-only callers (GetBacklogStatsAsync, ReadPendingAsync,
|
|
// ReadPendingSinceAsync, ReadForwardedAsync) concurrently with the
|
|
// batched writer, decoupling those reads from _writeLock. WAL is a
|
|
// database-level setting persisted in the file header; setting it on
|
|
// the writer connection means every connection opened to the file
|
|
// afterwards inherits WAL behaviour. PRAGMA journal_mode returns the
|
|
// mode actually adopted ("memory" for ":memory:" / shared-cache memory
|
|
// mode, "wal" for file-backed) — we don't error if WAL was rejected
|
|
// because the read connection's correctness does not depend on WAL
|
|
// itself, only its concurrency advantage does.
|
|
using (var pragmaCmd = _connection.CreateCommand())
|
|
{
|
|
pragmaCmd.CommandText = "PRAGMA journal_mode = WAL";
|
|
pragmaCmd.ExecuteNonQuery();
|
|
}
|
|
|
|
// synchronous=NORMAL (arch-review remediation WP1.2): SQLite's FULL default
|
|
// fsyncs on every transaction commit, which is the dominant cost of the
|
|
// per-event write path this same change coalesces via FlushIntervalMs — FULL
|
|
// would still fsync once per FLUSH even after that fix. NORMAL fsyncs at
|
|
// fewer, WAL-checkpoint-aligned points instead of every commit; the window
|
|
// this opens is a handful of the most recent commits lost on OS crash / power
|
|
// loss (WAL mode itself still guarantees no *corruption*, only a possible
|
|
// short rollback on unclean restart). That trade is acceptable here because
|
|
// audit writes are explicitly best-effort by design (CLAUDE.md: "Audit-write
|
|
// failure NEVER aborts the user-facing action — audit is best-effort, the
|
|
// action's own success/failure path is authoritative") — this is the same
|
|
// durability class as a dropped audit row, not a new risk class.
|
|
using (var pragmaCmd = _connection.CreateCommand())
|
|
{
|
|
pragmaCmd.CommandText = "PRAGMA synchronous = NORMAL";
|
|
pragmaCmd.ExecuteNonQuery();
|
|
}
|
|
|
|
// Enable FK enforcement on the WRITE connection. PRAGMA foreign_keys is
|
|
// a per-connection, per-session setting in SQLite — it is NOT persisted
|
|
// in the database file, so every new connection that may INSERT into
|
|
// audit_forward_state must set it for the FK
|
|
// audit_forward_state.EventId → audit_event.EventId
|
|
// to be a real runtime guard rather than decorative DDL. The write
|
|
// connection owns all INSERTs (and the MarkForwardedAsync /
|
|
// MarkReconciledAsync UPDATEs), so setting it here — after WAL is
|
|
// established, before the CREATE TABLEs — ensures the FK is live for
|
|
// every insert that follows. The existing insert order (audit_event
|
|
// first, then audit_forward_state, inside the same transaction) already
|
|
// satisfies the FK, so no pre-existing rows can violate the constraint.
|
|
using (var pragmaCmd = _connection.CreateCommand())
|
|
{
|
|
pragmaCmd.CommandText = "PRAGMA foreign_keys = ON";
|
|
pragmaCmd.ExecuteNonQuery();
|
|
}
|
|
|
|
// In-place reset. The site store is EPHEMERAL (≈7-day
|
|
// retention, recreated per deployment), so we do NOT migrate the old
|
|
// single 24-column AuditLog table to the new two-table shape: any rows
|
|
// it holds are within the retention window and discarded. DROP it if an
|
|
// earlier deployment left it behind, then CREATE the two new tables. This
|
|
// is safe precisely BECAUSE the site store is ephemeral — never do this
|
|
// on a durable store (the central SQL Server side keeps its shim until
|
|
// it is migrated, not reset).
|
|
using (var dropCmd = _connection.CreateCommand())
|
|
{
|
|
dropCmd.CommandText = "DROP TABLE IF EXISTS AuditLog;";
|
|
dropCmd.ExecuteNonQuery();
|
|
}
|
|
|
|
using var cmd = _connection.CreateCommand();
|
|
cmd.CommandText = """
|
|
-- Canonical, append-only / write-once: the 10 fields of the canonical
|
|
-- ZB.MOM.WW.Audit.AuditEvent stored directly (DetailsJson carries the
|
|
-- ScadaBridge domain fields). No forwarding state lives here — that is
|
|
-- the audit_forward_state sidecar's concern.
|
|
CREATE TABLE IF NOT EXISTS audit_event (
|
|
EventId TEXT NOT NULL,
|
|
OccurredAtUtc TEXT NOT NULL,
|
|
Actor TEXT NOT NULL,
|
|
Action TEXT NOT NULL,
|
|
Outcome TEXT NOT NULL,
|
|
Category TEXT NULL,
|
|
Target TEXT NULL,
|
|
SourceNode TEXT NULL,
|
|
CorrelationId TEXT NULL,
|
|
DetailsJson TEXT NULL,
|
|
PRIMARY KEY (EventId)
|
|
);
|
|
|
|
-- Operational, mutable: the forwarding lifecycle for each canonical
|
|
-- row. OccurredAtUtc is duplicated here so the drain range-scan stays
|
|
-- on this one table's index; IsCachedKind is precomputed at insert so
|
|
-- the cached/non-cached drain split never re-parses DetailsJson on the
|
|
-- read hot-path.
|
|
CREATE TABLE IF NOT EXISTS audit_forward_state (
|
|
EventId TEXT NOT NULL,
|
|
ForwardState TEXT NOT NULL,
|
|
OccurredAtUtc TEXT NOT NULL,
|
|
IsCachedKind INTEGER NOT NULL,
|
|
AttemptCount INTEGER NOT NULL DEFAULT 0,
|
|
LastAttemptUtc TEXT NULL,
|
|
PRIMARY KEY (EventId),
|
|
FOREIGN KEY (EventId) REFERENCES audit_event(EventId)
|
|
);
|
|
|
|
-- Drain index: every read filters on (ForwardState, IsCachedKind) and
|
|
-- range-scans/orders by OccurredAtUtc, so this composite covers the
|
|
-- four reads + the backlog COUNT/MIN.
|
|
CREATE INDEX IF NOT EXISTS IX_fwd
|
|
ON audit_forward_state (ForwardState, IsCachedKind, OccurredAtUtc);
|
|
""";
|
|
cmd.ExecuteNonQuery();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Enqueues an audit event for asynchronous batched persistence to SQLite.
|
|
/// Back-pressure is applied when the write channel is full.
|
|
/// </summary>
|
|
/// <param name="evt">The audit event to persist.</param>
|
|
/// <param name="ct">Cancellation token.</param>
|
|
/// <returns>A task that completes when the event has been persisted.</returns>
|
|
public Task WriteAsync(AuditEvent evt, CancellationToken ct = default)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(evt);
|
|
|
|
// The canonical record carries no ForwardState (a site-storage-only
|
|
// concern). Site rows always start Pending; the sidecar row is written
|
|
// alongside the canonical row in the same transaction.
|
|
var pending = new PendingAuditEvent(evt, AuditForwardState.Pending);
|
|
|
|
// CreateBounded(FullMode=Wait) means WriteAsync will await room rather
|
|
// than throw when full — exactly the hot-path back-pressure semantics
|
|
// we want.
|
|
if (!_writeQueue.Writer.TryWrite(pending))
|
|
{
|
|
// The writer is either completed (logger disposed) or the channel
|
|
// is at capacity. Fall back to the async path which honours the
|
|
// FullMode=Wait policy.
|
|
return WriteSlowPathAsync(pending, ct);
|
|
}
|
|
|
|
return pending.Completion.Task;
|
|
}
|
|
|
|
private async Task WriteSlowPathAsync(PendingAuditEvent pending, CancellationToken ct)
|
|
{
|
|
try
|
|
{
|
|
await _writeQueue.Writer.WriteAsync(pending, ct).ConfigureAwait(false);
|
|
}
|
|
catch (ChannelClosedException)
|
|
{
|
|
pending.Completion.TrySetException(
|
|
new ObjectDisposedException(nameof(SqliteAuditWriter),
|
|
"Event could not be recorded: the audit writer has been disposed."));
|
|
}
|
|
|
|
await pending.Completion.Task.ConfigureAwait(false);
|
|
}
|
|
|
|
private async Task ProcessWriteQueueAsync()
|
|
{
|
|
var batch = new List<PendingAuditEvent>(_options.BatchSize);
|
|
|
|
// ReadAllAsync completes when the channel is marked complete (Dispose).
|
|
await foreach (var first in _writeQueue.Reader.ReadAllAsync().ConfigureAwait(false))
|
|
{
|
|
batch.Clear();
|
|
batch.Add(first);
|
|
var deadline = DateTime.UtcNow.AddMilliseconds(_options.FlushIntervalMs);
|
|
|
|
// Pull additional ready events up to BatchSize. TryRead is non-
|
|
// blocking and lets us amortise the transaction overhead across a
|
|
// burst of concurrent enqueues. Once the immediately-available
|
|
// burst is drained, keep the transaction open for up to
|
|
// FlushIntervalMs waiting for MORE events to trickle in — this is
|
|
// the soft-flush coalescing arch-review remediation WP1.2 adds:
|
|
// without it, a low-rate trickle of writes (one script call every
|
|
// few ms, well under BatchSize) fsyncs once PER EVENT, because the
|
|
// writer loop reaches an empty channel and flushes immediately.
|
|
// FlushIntervalMs bounds the worst-case added latency any single
|
|
// event can see from this coalescing.
|
|
while (batch.Count < _options.BatchSize)
|
|
{
|
|
if (_writeQueue.Reader.TryRead(out var next))
|
|
{
|
|
batch.Add(next);
|
|
continue;
|
|
}
|
|
|
|
var remaining = deadline - DateTime.UtcNow;
|
|
if (remaining <= TimeSpan.Zero)
|
|
{
|
|
break;
|
|
}
|
|
|
|
var waitToRead = _writeQueue.Reader.WaitToReadAsync().AsTask();
|
|
var completed = await Task.WhenAny(
|
|
waitToRead, Task.Delay(remaining)).ConfigureAwait(false);
|
|
|
|
if (completed != waitToRead)
|
|
{
|
|
// Deadline elapsed with nothing new — flush what we have.
|
|
break;
|
|
}
|
|
|
|
// WaitToReadAsync resolving false means the channel completed
|
|
// (Dispose) with nothing left to read; resolving true means at
|
|
// least one item is available, so the top of the loop's
|
|
// TryRead will pick it up.
|
|
if (!await waitToRead.ConfigureAwait(false))
|
|
{
|
|
break;
|
|
}
|
|
}
|
|
|
|
FlushBatch(batch);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Total number of <see cref="FlushBatch"/> transactions committed since
|
|
/// construction. Test-only observability seam (arch-review remediation
|
|
/// WP1.2) for asserting that FlushIntervalMs coalesces a trickle of
|
|
/// near-simultaneous writes into a single transaction rather than one
|
|
/// commit per event; production code never reads this.
|
|
/// </summary>
|
|
internal int FlushCountForTests => _flushCountForTests;
|
|
private int _flushCountForTests;
|
|
|
|
private void FlushBatch(IReadOnlyList<PendingAuditEvent> batch)
|
|
{
|
|
lock (_writeLock)
|
|
{
|
|
if (_disposed)
|
|
{
|
|
foreach (var pending in batch)
|
|
{
|
|
pending.Completion.TrySetException(
|
|
new ObjectDisposedException(nameof(SqliteAuditWriter),
|
|
"Event could not be recorded: the audit writer was disposed before the write completed."));
|
|
}
|
|
return;
|
|
}
|
|
|
|
using var transaction = _connection.BeginTransaction();
|
|
try
|
|
{
|
|
// INSERT 1: the canonical row, stored DIRECTLY (the 10 canonical
|
|
// fields straight off the AuditEvent — no Decompose; audit_event
|
|
// holds canonical shape, not the legacy 24-column shape).
|
|
using var eventCmd = _connection.CreateCommand();
|
|
eventCmd.Transaction = transaction;
|
|
eventCmd.CommandText = """
|
|
INSERT INTO audit_event (
|
|
EventId, OccurredAtUtc, Actor, Action, Outcome,
|
|
Category, Target, SourceNode, CorrelationId, DetailsJson
|
|
) VALUES (
|
|
$EventId, $OccurredAtUtc, $Actor, $Action, $Outcome,
|
|
$Category, $Target, $SourceNode, $CorrelationId, $DetailsJson
|
|
);
|
|
""";
|
|
var eEventId = eventCmd.Parameters.Add("$EventId", SqliteType.Text);
|
|
var eOccurredAt = eventCmd.Parameters.Add("$OccurredAtUtc", SqliteType.Text);
|
|
var eActor = eventCmd.Parameters.Add("$Actor", SqliteType.Text);
|
|
var eAction = eventCmd.Parameters.Add("$Action", SqliteType.Text);
|
|
var eOutcome = eventCmd.Parameters.Add("$Outcome", SqliteType.Text);
|
|
var eCategory = eventCmd.Parameters.Add("$Category", SqliteType.Text);
|
|
var eTarget = eventCmd.Parameters.Add("$Target", SqliteType.Text);
|
|
var eSourceNode = eventCmd.Parameters.Add("$SourceNode", SqliteType.Text);
|
|
var eCorrelationId = eventCmd.Parameters.Add("$CorrelationId", SqliteType.Text);
|
|
var eDetailsJson = eventCmd.Parameters.Add("$DetailsJson", SqliteType.Text);
|
|
|
|
// INSERT 2: the operational sidecar row. ForwardState=Pending,
|
|
// OccurredAtUtc duplicated for the drain index, IsCachedKind
|
|
// precomputed (so the read split never parses DetailsJson),
|
|
// AttemptCount=0, LastAttemptUtc=NULL.
|
|
using var fwdCmd = _connection.CreateCommand();
|
|
fwdCmd.Transaction = transaction;
|
|
fwdCmd.CommandText = """
|
|
INSERT INTO audit_forward_state (
|
|
EventId, ForwardState, OccurredAtUtc, IsCachedKind, AttemptCount, LastAttemptUtc
|
|
) VALUES (
|
|
$EventId, $ForwardState, $OccurredAtUtc, $IsCachedKind, 0, NULL
|
|
);
|
|
""";
|
|
var fEventId = fwdCmd.Parameters.Add("$EventId", SqliteType.Text);
|
|
var fForwardState = fwdCmd.Parameters.Add("$ForwardState", SqliteType.Text);
|
|
var fOccurredAt = fwdCmd.Parameters.Add("$OccurredAtUtc", SqliteType.Text);
|
|
var fIsCachedKind = fwdCmd.Parameters.Add("$IsCachedKind", SqliteType.Integer);
|
|
|
|
foreach (var pending in batch)
|
|
{
|
|
var evt = pending.Event;
|
|
// Canonical OccurredAtUtc is UTC by construction; store the
|
|
// round-trip "o" form so string comparison stays monotonic
|
|
// (the drain range-scan and ORDER BY rely on it).
|
|
var occurredText = evt.OccurredAtUtc.UtcDateTime.ToString(
|
|
"o", System.Globalization.CultureInfo.InvariantCulture);
|
|
|
|
eEventId.Value = evt.EventId.ToString();
|
|
eOccurredAt.Value = occurredText;
|
|
// Canonical Actor is a required non-null string.
|
|
eActor.Value = evt.Actor ?? string.Empty;
|
|
eAction.Value = evt.Action;
|
|
eOutcome.Value = evt.Outcome.ToString();
|
|
eCategory.Value = (object?)evt.Category ?? DBNull.Value;
|
|
eTarget.Value = (object?)evt.Target ?? DBNull.Value;
|
|
// SourceNode-stamping: caller-provided value wins (preserves
|
|
// rows reconciled in from other nodes via the same writer);
|
|
// otherwise stamp from the local INodeIdentityProvider. The
|
|
// event record itself is NOT mutated — stamping is at write
|
|
// time only. If the provider also returns null (unconfigured
|
|
// node), the column stays NULL — operators see "needs config"
|
|
// via the schema, not a magic fallback string.
|
|
var sourceNode = evt.SourceNode ?? _nodeIdentity.NodeName;
|
|
eSourceNode.Value = (object?)sourceNode ?? DBNull.Value;
|
|
eCorrelationId.Value = (object?)evt.CorrelationId?.ToString() ?? DBNull.Value;
|
|
eDetailsJson.Value = (object?)evt.DetailsJson ?? DBNull.Value;
|
|
|
|
fEventId.Value = evt.EventId.ToString();
|
|
fForwardState.Value = pending.ForwardState.ToString();
|
|
fOccurredAt.Value = occurredText;
|
|
fIsCachedKind.Value = IsCachedKind(evt.DetailsJson) ? 1 : 0;
|
|
|
|
try
|
|
{
|
|
eventCmd.ExecuteNonQuery();
|
|
fwdCmd.ExecuteNonQuery();
|
|
pending.Completion.TrySetResult();
|
|
}
|
|
catch (SqliteException ex) when (ex.SqliteErrorCode == SqliteErrorConstraint)
|
|
{
|
|
// Duplicate EventId — first-write-wins (alog.md §11). The
|
|
// audit_event PRIMARY KEY throws before the sidecar insert
|
|
// runs, so neither table gains a second row. Treat as
|
|
// success: the lifecycle event is durably recorded under
|
|
// the first writer's payload.
|
|
_logger.LogDebug(ex,
|
|
"Duplicate EventId {EventId} swallowed by SqliteAuditWriter",
|
|
evt.EventId);
|
|
pending.Completion.TrySetResult();
|
|
}
|
|
}
|
|
|
|
transaction.Commit();
|
|
_flushCountForTests++;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
transaction.Rollback();
|
|
_logger.LogError(ex, "SqliteAuditWriter batch insert failed; faulting {Count} pending events", batch.Count);
|
|
foreach (var pending in batch)
|
|
{
|
|
pending.Completion.TrySetException(ex);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Cached-lifecycle audit kinds that ride the combined-telemetry
|
|
// drain (joined with the operational tracking row + pushed via
|
|
// IngestCachedTelemetryAsync into the central dual-write transaction).
|
|
// This is the SAME set the earlier ReadPendingCachedTelemetryAsync query
|
|
// filtered on (Kind IN (...)); it is now precomputed into the sidecar's
|
|
// IsCachedKind flag at INSERT (see IsCachedKind) so the read split is a cheap
|
|
// integer predicate, not a JSON parse. ReadPendingAsync drains everything
|
|
// with IsCachedKind=0; ReadPendingCachedTelemetryAsync drains IsCachedKind=1.
|
|
private static readonly HashSet<AuditKind> CachedTelemetryKinds = new()
|
|
{
|
|
AuditKind.CachedSubmit,
|
|
AuditKind.ApiCallCached,
|
|
AuditKind.DbWriteCached,
|
|
AuditKind.CachedResolve,
|
|
};
|
|
|
|
/// <summary>
|
|
/// Precomputes the sidecar's <c>IsCachedKind</c> flag from a canonical
|
|
/// row's <c>DetailsJson</c>. Parses the <see cref="AuditDetails.Kind"/>
|
|
/// discriminator via <see cref="AuditDetailsCodec"/> and returns <c>true</c>
|
|
/// iff it is one of the cached-lifecycle kinds
|
|
/// (<see cref="AuditKind.CachedSubmit"/>, <see cref="AuditKind.ApiCallCached"/>,
|
|
/// <see cref="AuditKind.DbWriteCached"/>, <see cref="AuditKind.CachedResolve"/>).
|
|
/// Runs once per event at INSERT time so the cached/non-cached drain split is
|
|
/// a cheap integer predicate on read, never a JSON parse on the hot path.
|
|
/// </summary>
|
|
private static bool IsCachedKind(string? detailsJson)
|
|
{
|
|
var details = AuditDetailsCodec.Deserialize(detailsJson);
|
|
var kind = AuditRowProjection.ParseEnum(details.Kind, AuditKind.InboundRequest);
|
|
return CachedTelemetryKinds.Contains(kind);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Returns up to <paramref name="limit"/> non-cached pending audit events, oldest first.
|
|
/// Cached-lifecycle kinds are excluded; use <see cref="ReadPendingCachedTelemetryAsync"/> for those.
|
|
/// </summary>
|
|
/// <param name="limit">Maximum number of rows to return.</param>
|
|
/// <param name="ct">Cancellation token.</param>
|
|
/// <returns>A task that resolves to a read-only list of pending audit events.</returns>
|
|
public Task<IReadOnlyList<AuditEvent>> ReadPendingAsync(int limit, CancellationToken ct = default)
|
|
{
|
|
if (limit <= 0)
|
|
{
|
|
throw new ArgumentOutOfRangeException(nameof(limit), "limit must be > 0.");
|
|
}
|
|
|
|
// Read via the dedicated _readConnection so this scan
|
|
// (which can be expensive when the backlog grows under a central
|
|
// outage) does not block the batched writer on _writeLock. WAL mode
|
|
// gives us a stable snapshot of the tables while writes proceed on the
|
|
// writer connection. _readLock serialises this connection across
|
|
// multiple concurrent read callers since SqliteConnection itself is
|
|
// not thread-safe.
|
|
// JOIN the sidecar and filter on IsCachedKind=0 — the cached-
|
|
// lifecycle kinds (IsCachedKind=1) flow through
|
|
// ReadPendingCachedTelemetryAsync + the combined-telemetry drain. The
|
|
// split is a precomputed integer predicate on the indexed sidecar, not
|
|
// a DetailsJson parse. Ordering is by the sidecar's OccurredAtUtc with
|
|
// EventId as the deterministic tiebreaker.
|
|
lock (_readLock)
|
|
{
|
|
ObjectDisposedException.ThrowIf(_disposed, this);
|
|
|
|
using var cmd = _readConnection.CreateCommand();
|
|
cmd.CommandText = """
|
|
SELECT ae.EventId, ae.OccurredAtUtc, ae.Actor, ae.Action, ae.Outcome,
|
|
ae.Category, ae.Target, ae.SourceNode, ae.CorrelationId, ae.DetailsJson
|
|
FROM audit_event ae
|
|
JOIN audit_forward_state fs ON fs.EventId = ae.EventId
|
|
WHERE fs.ForwardState = $pending
|
|
AND fs.IsCachedKind = 0
|
|
ORDER BY fs.OccurredAtUtc ASC, ae.EventId ASC
|
|
LIMIT $limit;
|
|
""";
|
|
cmd.Parameters.AddWithValue("$pending", AuditForwardState.Pending.ToString());
|
|
cmd.Parameters.AddWithValue("$limit", limit);
|
|
|
|
return Task.FromResult(ReadRows(cmd, limit));
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Returns up to <paramref name="limit"/> pending cached-lifecycle audit events, oldest first.
|
|
/// Only rows with cached-call kinds (CachedSubmit, ApiCallCached, DbWriteCached, CachedResolve) are included.
|
|
/// </summary>
|
|
/// <param name="limit">Maximum number of rows to return.</param>
|
|
/// <param name="ct">Cancellation token.</param>
|
|
/// <returns>A task that resolves to a read-only list of pending cached-telemetry audit events.</returns>
|
|
public Task<IReadOnlyList<AuditEvent>> ReadPendingCachedTelemetryAsync(
|
|
int limit, CancellationToken ct = default)
|
|
{
|
|
if (limit <= 0)
|
|
{
|
|
throw new ArgumentOutOfRangeException(nameof(limit), "limit must be > 0.");
|
|
}
|
|
|
|
// Dedicated read surface for the cached-call lifecycle
|
|
// drain — symmetric to ReadPendingAsync but filtered to IsCachedKind=1.
|
|
// Same _readConnection + _readLock pattern so the hot-path writer is not
|
|
// contended.
|
|
lock (_readLock)
|
|
{
|
|
ObjectDisposedException.ThrowIf(_disposed, this);
|
|
|
|
using var cmd = _readConnection.CreateCommand();
|
|
cmd.CommandText = """
|
|
SELECT ae.EventId, ae.OccurredAtUtc, ae.Actor, ae.Action, ae.Outcome,
|
|
ae.Category, ae.Target, ae.SourceNode, ae.CorrelationId, ae.DetailsJson
|
|
FROM audit_event ae
|
|
JOIN audit_forward_state fs ON fs.EventId = ae.EventId
|
|
WHERE fs.ForwardState = $pending
|
|
AND fs.IsCachedKind = 1
|
|
ORDER BY fs.OccurredAtUtc ASC, ae.EventId ASC
|
|
LIMIT $limit;
|
|
""";
|
|
cmd.Parameters.AddWithValue("$pending", AuditForwardState.Pending.ToString());
|
|
cmd.Parameters.AddWithValue("$limit", limit);
|
|
|
|
return Task.FromResult(ReadRows(cmd, limit));
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Returns up to <paramref name="limit"/> rows in
|
|
/// <see cref="AuditForwardState.Forwarded"/>, oldest
|
|
/// <see cref="AuditEvent.OccurredAtUtc"/> first, with
|
|
/// <see cref="AuditEvent.EventId"/> as the deterministic tiebreaker. The
|
|
/// <see cref="AuditForwardState.Forwarded"/>-specific counterpart of
|
|
/// <see cref="ReadPendingAsync"/>; used by tests to assert a row reached the
|
|
/// <see cref="AuditForwardState.Forwarded"/> state specifically (unlike
|
|
/// <see cref="ReadPendingSinceAsync"/>, which also returns
|
|
/// <see cref="AuditForwardState.Pending"/> rows).
|
|
/// </summary>
|
|
/// <param name="limit">Maximum number of rows to return.</param>
|
|
/// <param name="ct">Cancellation token.</param>
|
|
/// <returns>A task that resolves to a read-only list of forwarded audit events.</returns>
|
|
public Task<IReadOnlyList<AuditEvent>> ReadForwardedAsync(int limit, CancellationToken ct = default)
|
|
{
|
|
if (limit <= 0)
|
|
{
|
|
throw new ArgumentOutOfRangeException(nameof(limit), "limit must be > 0.");
|
|
}
|
|
|
|
// Mirror ReadPendingAsync — read via _readConnection /
|
|
// _readLock so this query never contends with the batched writer on
|
|
// _writeLock. JOIN the sidecar and filter on ForwardState='Forwarded'
|
|
// (no IsCachedKind split — both cached and non-cached Forwarded rows are
|
|
// returned, as before).
|
|
lock (_readLock)
|
|
{
|
|
ObjectDisposedException.ThrowIf(_disposed, this);
|
|
|
|
using var cmd = _readConnection.CreateCommand();
|
|
cmd.CommandText = """
|
|
SELECT ae.EventId, ae.OccurredAtUtc, ae.Actor, ae.Action, ae.Outcome,
|
|
ae.Category, ae.Target, ae.SourceNode, ae.CorrelationId, ae.DetailsJson
|
|
FROM audit_event ae
|
|
JOIN audit_forward_state fs ON fs.EventId = ae.EventId
|
|
WHERE fs.ForwardState = $forwarded
|
|
ORDER BY fs.OccurredAtUtc ASC, ae.EventId ASC
|
|
LIMIT $limit;
|
|
""";
|
|
cmd.Parameters.AddWithValue("$forwarded", AuditForwardState.Forwarded.ToString());
|
|
cmd.Parameters.AddWithValue("$limit", limit);
|
|
|
|
return Task.FromResult(ReadRows(cmd, limit));
|
|
}
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
public Task MarkForwardedAsync(IReadOnlyList<Guid> eventIds, CancellationToken ct = default)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(eventIds);
|
|
if (eventIds.Count == 0)
|
|
{
|
|
return Task.CompletedTask;
|
|
}
|
|
|
|
lock (_writeLock)
|
|
{
|
|
ObjectDisposedException.ThrowIf(_disposed, this);
|
|
|
|
// Flip the sidecar — UPDATE audit_forward_state, not the canonical
|
|
// audit_event (which is append-only / write-once). Bump AttemptCount +
|
|
// stamp LastAttemptUtc so operators can see how many drain passes a row
|
|
// took to forward. Each id is bound as its own parameter, so no string
|
|
// concatenation of user data ever enters the SQL.
|
|
//
|
|
// Defensive state guard: only transition rows that are still Pending or
|
|
// Forwarded (i.e. not yet Reconciled). Without this guard a mis-called
|
|
// batch that includes a Reconciled EventId would silently demote it back
|
|
// to Forwarded — a state regression that would cause duplicate central
|
|
// ingestion. Symmetric with MarkReconciledAsync's
|
|
// WHERE ForwardState IN ($pending, $forwarded)
|
|
// guard. Current callers only pass Pending IDs, so normal-path behaviour
|
|
// is unchanged; the guard is purely defensive.
|
|
//
|
|
// Stamp one $now for the whole call so every chunk records the same
|
|
// LastAttemptUtc (matching the prior single-statement behaviour).
|
|
var now = DateTime.UtcNow.ToString(
|
|
"o", System.Globalization.CultureInfo.InvariantCulture);
|
|
var forwarded = AuditForwardState.Forwarded.ToString();
|
|
var pending = AuditForwardState.Pending.ToString();
|
|
|
|
// Chunk into ≤MarkForwardStateChunkSize-id UPDATEs, all inside ONE
|
|
// transaction so the whole batch commits atomically (arch-review 04, P8).
|
|
using var transaction = _connection.BeginTransaction();
|
|
for (int start = 0; start < eventIds.Count; start += MarkForwardStateChunkSize)
|
|
{
|
|
var chunk = Math.Min(MarkForwardStateChunkSize, eventIds.Count - start);
|
|
using var cmd = _connection.CreateCommand();
|
|
cmd.Transaction = transaction;
|
|
var sb = new System.Text.StringBuilder();
|
|
sb.Append("UPDATE audit_forward_state SET ForwardState = $forwarded, ")
|
|
.Append("AttemptCount = AttemptCount + 1, LastAttemptUtc = $now ")
|
|
.Append("WHERE ForwardState IN ($pending, $forwarded) AND EventId IN (");
|
|
for (int i = 0; i < chunk; i++)
|
|
{
|
|
if (i > 0) sb.Append(',');
|
|
var p = $"$id{i}";
|
|
sb.Append(p);
|
|
cmd.Parameters.AddWithValue(p, eventIds[start + i].ToString());
|
|
}
|
|
sb.Append(");");
|
|
cmd.CommandText = sb.ToString();
|
|
cmd.Parameters.AddWithValue("$forwarded", forwarded);
|
|
cmd.Parameters.AddWithValue("$pending", pending);
|
|
cmd.Parameters.AddWithValue("$now", now);
|
|
cmd.ExecuteNonQuery();
|
|
}
|
|
transaction.Commit();
|
|
return Task.CompletedTask;
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Returns up to <paramref name="batchSize"/> pending or forwarded audit events
|
|
/// with <see cref="AuditEvent.OccurredAtUtc"/> >= <paramref name="sinceUtc"/>, oldest first.
|
|
/// Used by the reconciliation-pull handler.
|
|
/// </summary>
|
|
/// <param name="sinceUtc">Lower bound timestamp (UTC) for event occurrence.</param>
|
|
/// <param name="batchSize">Maximum number of rows to return.</param>
|
|
/// <param name="afterId">
|
|
/// Composite-keyset tiebreak: the EventId of the last row already consumed at
|
|
/// <paramref name="sinceUtc"/>. Non-null switches the predicate from the inclusive
|
|
/// <c>OccurredAtUtc >= $since</c> to the strict composite
|
|
/// <c>(OccurredAtUtc, EventId) > ($since, $afterId)</c>, matching the query's own
|
|
/// ORDER BY so a batch cannot stall on rows sharing one instant.
|
|
/// </param>
|
|
/// <param name="ct">Cancellation token.</param>
|
|
/// <returns>A task that resolves to a read-only list of audit events since the given timestamp.</returns>
|
|
public Task<IReadOnlyList<AuditEvent>> ReadPendingSinceAsync(
|
|
DateTime sinceUtc, int batchSize, string? afterId = null, CancellationToken ct = default)
|
|
{
|
|
if (batchSize <= 0)
|
|
{
|
|
throw new ArgumentOutOfRangeException(nameof(batchSize), "batchSize must be > 0.");
|
|
}
|
|
|
|
// Read via _readConnection / _readLock — same lock-
|
|
// decoupling as ReadPendingAsync. JOIN the sidecar; the range scan
|
|
// is on the sidecar's duplicated OccurredAtUtc so it stays on IX_fwd.
|
|
// Both Pending and Forwarded rows are returned (the central reconciliation
|
|
// puller dedups on EventId; re-shipping a Forwarded-but-not-yet-ingested
|
|
// row is safe).
|
|
lock (_readLock)
|
|
{
|
|
ObjectDisposedException.ThrowIf(_disposed, this);
|
|
|
|
using var cmd = _readConnection.CreateCommand();
|
|
var cursorPredicate = afterId is null
|
|
? "fs.OccurredAtUtc >= $since"
|
|
// Composite keyset, lexicographic on (OccurredAtUtc, EventId) — the same
|
|
// ordering the query applies, so it is a strict "everything after this row".
|
|
: "(fs.OccurredAtUtc > $since OR (fs.OccurredAtUtc = $since AND ae.EventId > $afterId))";
|
|
|
|
// fs.rowid rides along as an 11th column purely to maintain
|
|
// _maxServedRowId — the insertion-order high-water mark that bounds
|
|
// MarkReconciledUpToAsync (see that method). It is never mapped onto
|
|
// the returned AuditEvent; rowid stays a storage-layer concept.
|
|
cmd.CommandText = $"""
|
|
SELECT ae.EventId, ae.OccurredAtUtc, ae.Actor, ae.Action, ae.Outcome,
|
|
ae.Category, ae.Target, ae.SourceNode, ae.CorrelationId, ae.DetailsJson,
|
|
fs.rowid
|
|
FROM audit_event ae
|
|
JOIN audit_forward_state fs ON fs.EventId = ae.EventId
|
|
WHERE fs.ForwardState IN ($pending, $forwarded)
|
|
AND {cursorPredicate}
|
|
ORDER BY fs.OccurredAtUtc ASC, ae.EventId ASC
|
|
LIMIT $limit;
|
|
""";
|
|
cmd.Parameters.AddWithValue("$pending", AuditForwardState.Pending.ToString());
|
|
cmd.Parameters.AddWithValue("$forwarded", AuditForwardState.Forwarded.ToString());
|
|
// Normalise to UTC ISO-8601 round-trip format to match how OccurredAtUtc
|
|
// is stored on insert ("o" format) — string comparison is monotonic for
|
|
// that encoding so we can index-scan against it.
|
|
cmd.Parameters.AddWithValue("$since", EnsureUtc(sinceUtc).ToString(
|
|
"o", System.Globalization.CultureInfo.InvariantCulture));
|
|
if (afterId is not null)
|
|
{
|
|
// EventIds are stored as Guid.ToString() ("D"), so compare in that form.
|
|
cmd.Parameters.AddWithValue("$afterId", NormalizeEventId(afterId));
|
|
}
|
|
cmd.Parameters.AddWithValue("$limit", batchSize);
|
|
|
|
return Task.FromResult(ReadServedRows(cmd, batchSize));
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Highest <c>audit_forward_state.rowid</c> this instance has ever SERVED from
|
|
/// <see cref="ReadPendingSinceAsync"/> — i.e. the insertion-order high-water mark of
|
|
/// rows central could possibly have received through the pull path. Bounds
|
|
/// <see cref="MarkReconciledUpToAsync"/> so a row inserted after that point can never be
|
|
/// retired by a cursor that happens to sit above its timestamp. Monotonic (a later, smaller
|
|
/// batch never lowers it) except for the purge clamp. Written under <c>_readLock</c> (the
|
|
/// read path) and read under <c>_writeLock</c> (the flip/purge paths) — two different
|
|
/// locks, so the accesses go through <see cref="Volatile"/> for visibility. A stale read
|
|
/// can only make the bound smaller, i.e. more conservative, never unsafe.
|
|
/// </summary>
|
|
private long _maxServedRowId;
|
|
|
|
/// <summary>
|
|
/// Executes a reconciliation-pull read whose projection carries <c>fs.rowid</c> as an
|
|
/// 11th column, materialising the canonical rows while advancing
|
|
/// <see cref="_maxServedRowId"/>. Serving a row is what makes it eligible for
|
|
/// cursor-proved retirement, so the two must happen in the same step.
|
|
/// </summary>
|
|
private IReadOnlyList<AuditEvent> ReadServedRows(SqliteCommand cmd, int capacityHint)
|
|
{
|
|
var rows = new List<AuditEvent>(Math.Min(capacityHint, 256));
|
|
var maxRowId = Volatile.Read(ref _maxServedRowId);
|
|
using (var reader = cmd.ExecuteReader())
|
|
{
|
|
while (reader.Read())
|
|
{
|
|
rows.Add(MapRow(reader));
|
|
var rowId = reader.GetInt64(10);
|
|
if (rowId > maxRowId) maxRowId = rowId;
|
|
}
|
|
}
|
|
|
|
Volatile.Write(ref _maxServedRowId, maxRowId);
|
|
return rows;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Normalises a wire-supplied event id to the exact textual form stored in
|
|
/// <c>audit_event.EventId</c> so the keyset comparison is apples-to-apples. An
|
|
/// unparseable value is passed through verbatim rather than throwing — a malformed
|
|
/// cursor must degrade to "serves a bit too much", never to a failed pull.
|
|
/// </summary>
|
|
private static string NormalizeEventId(string afterId) =>
|
|
Guid.TryParse(afterId, out var g) ? g.ToString() : afterId;
|
|
|
|
/// <inheritdoc />
|
|
public Task<int> MarkReconciledUpToAsync(
|
|
DateTime sinceUtc, string? afterId, CancellationToken ct = default)
|
|
{
|
|
lock (_writeLock)
|
|
{
|
|
ObjectDisposedException.ThrowIf(_disposed, this);
|
|
|
|
// Everything at or before central's cursor is proven received. With no
|
|
// afterId the read contract is inclusive (>= $since), so only rows STRICTLY
|
|
// older than the cursor instant are proven — the boundary instant may be
|
|
// half-consumed and must stay servable.
|
|
var predicate = afterId is null
|
|
? "fs.OccurredAtUtc < $since"
|
|
: "(fs.OccurredAtUtc < $since OR (fs.OccurredAtUtc = $since AND fs.EventId <= $afterId))";
|
|
|
|
// INSERTION-ORDER BOUND — "only rows we actually served can retire".
|
|
//
|
|
// The timestamp cursor alone is not enough. OccurredAtUtc is stamped by the
|
|
// caller, so a row can be INSERTED after a batch was served yet carry a
|
|
// timestamp BELOW central's (by then advanced) cursor — a late-stamped insert.
|
|
// The blanket cursor UPDATE retired exactly those rows: never served, never
|
|
// servable again (ReadPendingSinceAsync's keyset has moved past them), and,
|
|
// being Reconciled, purged on age — silent audit loss, a failure mode the old
|
|
// explicit id-set flip could not produce.
|
|
//
|
|
// fs.rowid is insertion order, so bounding the flip at the highest rowid this
|
|
// instance has ever served restores the invariant exactly: rows inserted after
|
|
// that point are out of the flip's reach no matter where the cursor sits. A
|
|
// FORWARDED row is exempt from the bound — central ACKED it through the
|
|
// telemetry push path, which is receipt-proof independent of the pull.
|
|
//
|
|
// Bound state is per-process: after a restart it is 0 until this instance serves
|
|
// a batch, so the first pull retires only Forwarded rows and the pull after it
|
|
// resumes normal retirement. Conservative in the safe direction (delays
|
|
// retirement, never loses a row); the liveness note in MarkReconciledUpToAsync's
|
|
// interface doc + Component-AuditLog.md covers what a permanently halted
|
|
// reconciliation means for such rows.
|
|
using var cmd = _connection.CreateCommand();
|
|
cmd.CommandText = $"""
|
|
UPDATE audit_forward_state AS fs
|
|
SET ForwardState = $reconciled
|
|
WHERE fs.ForwardState IN ($pending, $forwarded)
|
|
AND (fs.ForwardState = $forwarded OR fs.rowid <= $maxServedRowId)
|
|
AND {predicate};
|
|
""";
|
|
cmd.Parameters.AddWithValue("$reconciled", AuditForwardState.Reconciled.ToString());
|
|
cmd.Parameters.AddWithValue("$pending", AuditForwardState.Pending.ToString());
|
|
cmd.Parameters.AddWithValue("$forwarded", AuditForwardState.Forwarded.ToString());
|
|
cmd.Parameters.AddWithValue("$maxServedRowId", Volatile.Read(ref _maxServedRowId));
|
|
cmd.Parameters.AddWithValue("$since", EnsureUtc(sinceUtc).ToString(
|
|
"o", System.Globalization.CultureInfo.InvariantCulture));
|
|
if (afterId is not null)
|
|
{
|
|
cmd.Parameters.AddWithValue("$afterId", NormalizeEventId(afterId));
|
|
}
|
|
|
|
return Task.FromResult(cmd.ExecuteNonQuery());
|
|
}
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
public Task MarkReconciledAsync(IReadOnlyList<Guid> eventIds, CancellationToken ct = default)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(eventIds);
|
|
if (eventIds.Count == 0)
|
|
{
|
|
return Task.CompletedTask;
|
|
}
|
|
|
|
lock (_writeLock)
|
|
{
|
|
ObjectDisposedException.ThrowIf(_disposed, this);
|
|
|
|
// Flip the sidecar from Pending/Forwarded → Reconciled. Rows
|
|
// already Reconciled are left untouched (idempotent re-call), and the
|
|
// canonical audit_event row is never modified.
|
|
var reconciled = AuditForwardState.Reconciled.ToString();
|
|
var pending = AuditForwardState.Pending.ToString();
|
|
var forwarded = AuditForwardState.Forwarded.ToString();
|
|
|
|
// Chunk into ≤MarkForwardStateChunkSize-id UPDATEs, all inside ONE
|
|
// transaction so the whole batch commits atomically (arch-review 04, P8).
|
|
using var transaction = _connection.BeginTransaction();
|
|
for (int start = 0; start < eventIds.Count; start += MarkForwardStateChunkSize)
|
|
{
|
|
var chunk = Math.Min(MarkForwardStateChunkSize, eventIds.Count - start);
|
|
using var cmd = _connection.CreateCommand();
|
|
cmd.Transaction = transaction;
|
|
var sb = new System.Text.StringBuilder();
|
|
sb.Append("UPDATE audit_forward_state SET ForwardState = $reconciled ")
|
|
.Append("WHERE ForwardState IN ($pending, $forwarded) AND EventId IN (");
|
|
for (int i = 0; i < chunk; i++)
|
|
{
|
|
if (i > 0) sb.Append(',');
|
|
var p = $"$id{i}";
|
|
sb.Append(p);
|
|
cmd.Parameters.AddWithValue(p, eventIds[start + i].ToString());
|
|
}
|
|
sb.Append(");");
|
|
cmd.CommandText = sb.ToString();
|
|
cmd.Parameters.AddWithValue("$reconciled", reconciled);
|
|
cmd.Parameters.AddWithValue("$pending", pending);
|
|
cmd.Parameters.AddWithValue("$forwarded", forwarded);
|
|
cmd.ExecuteNonQuery();
|
|
}
|
|
transaction.Commit();
|
|
return Task.CompletedTask;
|
|
}
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
public Task<SiteAuditBacklogSnapshot> GetBacklogStatsAsync(CancellationToken ct = default)
|
|
{
|
|
int pendingCount;
|
|
DateTime? oldestPending;
|
|
|
|
// Read via the dedicated _readConnection (under
|
|
// _readLock) so this probe — polled every 30 s by SiteAuditBacklogReporter
|
|
// — never blocks the batched hot-path writer on _writeLock. Under a
|
|
// central outage the Pending backlog can grow to hundreds of thousands
|
|
// of rows and the COUNT(*) scan correspondingly stretches; that no
|
|
// longer adds tail latency to user-facing audit writes.
|
|
// Count over the sidecar (audit_forward_state) — the canonical
|
|
// audit_event table carries no ForwardState. The IX_fwd index makes both
|
|
// aggregates cheap (count is a covering scan, min is the first key).
|
|
lock (_readLock)
|
|
{
|
|
ObjectDisposedException.ThrowIf(_disposed, this);
|
|
|
|
using var cmd = _readConnection.CreateCommand();
|
|
cmd.CommandText = """
|
|
SELECT COUNT(*), MIN(OccurredAtUtc)
|
|
FROM audit_forward_state
|
|
WHERE ForwardState = $pending;
|
|
""";
|
|
cmd.Parameters.AddWithValue("$pending", AuditForwardState.Pending.ToString());
|
|
|
|
using var reader = cmd.ExecuteReader();
|
|
reader.Read();
|
|
pendingCount = reader.GetInt32(0);
|
|
oldestPending = reader.IsDBNull(1)
|
|
? null
|
|
: DateTime.Parse(reader.GetString(1),
|
|
System.Globalization.CultureInfo.InvariantCulture,
|
|
System.Globalization.DateTimeStyles.RoundtripKind);
|
|
}
|
|
|
|
// File-size lookup outside the lock — the DatabasePath option is the
|
|
// canonical source. The connection-string-override branch (used by
|
|
// some tests) keeps the same DatabasePath value, so this works
|
|
// uniformly. In-memory / mode=memory paths return 0 because the file
|
|
// doesn't exist on disk.
|
|
long onDiskBytes = 0;
|
|
try
|
|
{
|
|
if (!string.IsNullOrEmpty(_options.DatabasePath) &&
|
|
!_options.DatabasePath.StartsWith(":memory:", StringComparison.Ordinal) &&
|
|
!_options.DatabasePath.Contains("mode=memory", StringComparison.OrdinalIgnoreCase) &&
|
|
File.Exists(_options.DatabasePath))
|
|
{
|
|
onDiskBytes = new FileInfo(_options.DatabasePath).Length;
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
// File system probe is a best-effort health-metric — never abort
|
|
// a backlog snapshot because stat() failed. Log and report 0.
|
|
_logger.LogDebug(ex,
|
|
"SqliteAuditWriter could not stat DB path {Path} for backlog snapshot.",
|
|
_options.DatabasePath);
|
|
}
|
|
|
|
return Task.FromResult(new SiteAuditBacklogSnapshot(
|
|
PendingCount: pendingCount,
|
|
OldestPendingUtc: oldestPending,
|
|
OnDiskBytes: onDiskBytes));
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
public Task<int> PurgeExpiredAsync(DateTime olderThanUtc, CancellationToken ct = default)
|
|
{
|
|
// Runs on the WRITE connection under _writeLock (a DELETE mutates the file),
|
|
// mirroring MarkForwardedAsync / MarkReconciledAsync. Serialising with the
|
|
// batched writer means the purge never races an in-flight FlushBatch.
|
|
lock (_writeLock)
|
|
{
|
|
ObjectDisposedException.ThrowIf(_disposed, this);
|
|
|
|
// Cutoff uses the same round-trip "o" encoding the writer stores
|
|
// OccurredAtUtc in, so string comparison stays monotonic (see FlushBatch).
|
|
var cutoff = EnsureUtc(olderThanUtc).ToString(
|
|
"o", System.Globalization.CultureInfo.InvariantCulture);
|
|
|
|
int purged;
|
|
using (var transaction = _connection.BeginTransaction())
|
|
{
|
|
try
|
|
{
|
|
// Capture the eligible EventIds from the sidecar FIRST: the
|
|
// Forwarded/Reconciled + age criteria live on
|
|
// audit_forward_state, and it is the FK child of audit_event, so
|
|
// it must be deleted before the parent. A temp table (rather than
|
|
// an IN (...) list) keeps an arbitrarily large purge set off the
|
|
// SQLite 999-parameter limit. The hard invariant is encoded here:
|
|
// only Forwarded/Reconciled rows qualify — Pending is never purged
|
|
// on age alone.
|
|
using (var createCmd = _connection.CreateCommand())
|
|
{
|
|
createCmd.Transaction = transaction;
|
|
createCmd.CommandText = """
|
|
DROP TABLE IF EXISTS purge_ids;
|
|
CREATE TEMP TABLE purge_ids AS
|
|
SELECT EventId FROM audit_forward_state
|
|
WHERE ForwardState IN ($forwarded, $reconciled)
|
|
AND OccurredAtUtc < $cutoff;
|
|
""";
|
|
createCmd.Parameters.AddWithValue("$forwarded", AuditForwardState.Forwarded.ToString());
|
|
createCmd.Parameters.AddWithValue("$reconciled", AuditForwardState.Reconciled.ToString());
|
|
createCmd.Parameters.AddWithValue("$cutoff", cutoff);
|
|
createCmd.ExecuteNonQuery();
|
|
}
|
|
|
|
// Delete the sidecar (FK child) first, then the canonical row.
|
|
using (var delSidecar = _connection.CreateCommand())
|
|
{
|
|
delSidecar.Transaction = transaction;
|
|
delSidecar.CommandText =
|
|
"DELETE FROM audit_forward_state WHERE EventId IN (SELECT EventId FROM purge_ids);";
|
|
delSidecar.ExecuteNonQuery();
|
|
}
|
|
|
|
using (var delEvent = _connection.CreateCommand())
|
|
{
|
|
delEvent.Transaction = transaction;
|
|
delEvent.CommandText =
|
|
"DELETE FROM audit_event WHERE EventId IN (SELECT EventId FROM purge_ids);";
|
|
delEvent.ExecuteNonQuery();
|
|
}
|
|
|
|
// changes() reports rows affected by the most recent DELETE on
|
|
// this connection (the audit_event delete above); read it before
|
|
// the DROP so no later statement can perturb the count.
|
|
using (var changesCmd = _connection.CreateCommand())
|
|
{
|
|
changesCmd.Transaction = transaction;
|
|
changesCmd.CommandText = "SELECT changes();";
|
|
purged = Convert.ToInt32(changesCmd.ExecuteScalar(),
|
|
System.Globalization.CultureInfo.InvariantCulture);
|
|
}
|
|
|
|
using (var dropCmd = _connection.CreateCommand())
|
|
{
|
|
dropCmd.Transaction = transaction;
|
|
dropCmd.CommandText = "DROP TABLE IF EXISTS purge_ids;";
|
|
dropCmd.ExecuteNonQuery();
|
|
}
|
|
|
|
// Clamp the served-rows high-water mark to what is still in the table.
|
|
// SQLite hands a fresh insert MAX(rowid)+1, so a purge that empties the
|
|
// sidecar (or removes its top rows) makes rowids REUSABLE — a stale-high
|
|
// _maxServedRowId would then vouch for rows nobody has served. Cheap:
|
|
// MAX(rowid) is an index-less O(1) lookup.
|
|
if (purged > 0)
|
|
{
|
|
using var maxCmd = _connection.CreateCommand();
|
|
maxCmd.Transaction = transaction;
|
|
maxCmd.CommandText = "SELECT IFNULL(MAX(rowid), 0) FROM audit_forward_state;";
|
|
var liveMax = Convert.ToInt64(maxCmd.ExecuteScalar(),
|
|
System.Globalization.CultureInfo.InvariantCulture);
|
|
if (liveMax < Volatile.Read(ref _maxServedRowId))
|
|
Volatile.Write(ref _maxServedRowId, liveMax);
|
|
}
|
|
|
|
transaction.Commit();
|
|
}
|
|
catch
|
|
{
|
|
transaction.Rollback();
|
|
throw;
|
|
}
|
|
}
|
|
|
|
// Reclaim the freed pages back to the OS. auto_vacuum=INCREMENTAL (set in
|
|
// InitializeSchema) makes this a bounded incremental step rather than a
|
|
// full VACUUM rewrite. Runs outside the transaction; a vacuum failure must
|
|
// not fault the purge itself, so it is best-effort.
|
|
if (purged > 0)
|
|
{
|
|
try
|
|
{
|
|
using var vacuumCmd = _connection.CreateCommand();
|
|
vacuumCmd.CommandText = "PRAGMA incremental_vacuum;";
|
|
vacuumCmd.ExecuteNonQuery();
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogDebug(ex,
|
|
"SqliteAuditWriter incremental_vacuum after retention purge failed (non-fatal).");
|
|
}
|
|
}
|
|
|
|
return Task.FromResult(purged);
|
|
}
|
|
}
|
|
|
|
private static DateTime EnsureUtc(DateTime value) =>
|
|
value.Kind == DateTimeKind.Utc
|
|
? value
|
|
: DateTime.SpecifyKind(value.ToUniversalTime(), DateTimeKind.Utc);
|
|
|
|
/// <summary>
|
|
/// Executes <paramref name="cmd"/> (one of the four reads, each already
|
|
/// projecting the 10 <c>audit_event</c> columns in canonical order) and
|
|
/// materialises the rows via <see cref="MapRow"/>.
|
|
/// </summary>
|
|
private static IReadOnlyList<AuditEvent> ReadRows(SqliteCommand cmd, int capacityHint)
|
|
{
|
|
var rows = new List<AuditEvent>(Math.Min(capacityHint, 256));
|
|
using var reader = cmd.ExecuteReader();
|
|
while (reader.Read())
|
|
{
|
|
rows.Add(MapRow(reader));
|
|
}
|
|
return rows;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Builds the canonical <see cref="AuditEvent"/> DIRECTLY from the 10
|
|
/// stored <c>audit_event</c> columns — no 24-column <c>Recompose</c>, because
|
|
/// <c>audit_event</c> already holds the canonical fields + <c>DetailsJson</c>.
|
|
/// <c>Outcome</c> is stored as the enum's name; the safe
|
|
/// <see cref="AuditRowProjection.ParseEnum{TEnum}"/> degrades an unknown/renamed
|
|
/// value gracefully rather than throwing.
|
|
/// </summary>
|
|
private static AuditEvent MapRow(SqliteDataReader reader)
|
|
{
|
|
return new AuditEvent
|
|
{
|
|
EventId = Guid.Parse(reader.GetString(0)),
|
|
OccurredAtUtc = new DateTimeOffset(DateTime.SpecifyKind(
|
|
DateTime.Parse(reader.GetString(1),
|
|
System.Globalization.CultureInfo.InvariantCulture,
|
|
System.Globalization.DateTimeStyles.RoundtripKind),
|
|
DateTimeKind.Utc)),
|
|
Actor = reader.GetString(2),
|
|
Action = reader.GetString(3),
|
|
Outcome = AuditRowProjection.ParseEnum(reader.GetString(4), AuditOutcome.Success),
|
|
Category = reader.IsDBNull(5) ? null : reader.GetString(5),
|
|
Target = reader.IsDBNull(6) ? null : reader.GetString(6),
|
|
SourceNode = reader.IsDBNull(7) ? null : reader.GetString(7),
|
|
CorrelationId = reader.IsDBNull(8) ? null : Guid.Parse(reader.GetString(8)),
|
|
DetailsJson = reader.IsDBNull(9) ? null : reader.GetString(9),
|
|
};
|
|
}
|
|
|
|
/// <summary>
|
|
/// Disposes the audit writer and releases resources.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// Prefer <see cref="DisposeAsync"/> when possible (DI honours
|
|
/// <see cref="IAsyncDisposable"/> on singletons). The sync path remains for
|
|
/// callers that only know about <see cref="IDisposable"/> (e.g. legacy
|
|
/// composition roots, <c>using</c> statements without <c>await</c>). To
|
|
/// avoid the classic sync-over-async deadlock on a captured
|
|
/// <see cref="SynchronizationContext"/> (ASP.NET request thread, Akka
|
|
/// dispatcher under some configurations), we hop to the thread pool via
|
|
/// <see cref="Task.Run(Func{Task})"/> before blocking on the result — the
|
|
/// async continuation inside <see cref="DisposeAsync"/> then resumes on a
|
|
/// pool thread with no captured context, so <c>GetResult()</c> never waits
|
|
/// on the very thread the continuation needs.
|
|
/// </remarks>
|
|
public void Dispose()
|
|
{
|
|
Task.Run(async () => await DisposeAsync().ConfigureAwait(false))
|
|
.GetAwaiter().GetResult();
|
|
}
|
|
|
|
/// <summary>Asynchronously disposes the audit writer and releases resources.</summary>
|
|
/// <returns>A <see cref="ValueTask"/> that completes when all resources have been released.</returns>
|
|
public async ValueTask DisposeAsync()
|
|
{
|
|
Task? writerLoop;
|
|
lock (_writeLock)
|
|
{
|
|
if (_disposed) return;
|
|
// Stop accepting new events. Completing the channel writer is the
|
|
// shutdown signal: WriteAsync calls observe the completion and
|
|
// fault, and the writer loop drains any already-buffered items
|
|
// before exiting. _disposed is intentionally NOT set here — it
|
|
// flips only after the loop has fully drained (second lock block
|
|
// below), so FlushBatch's existing _disposed check guards the
|
|
// post-drain window when the connection is about to close.
|
|
_writeQueue.Writer.TryComplete();
|
|
writerLoop = _writerLoop;
|
|
}
|
|
|
|
// Wait outside the lock — the loop reacquires it for each batch.
|
|
try
|
|
{
|
|
if (writerLoop is not null)
|
|
{
|
|
await writerLoop.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
|
}
|
|
}
|
|
catch (TimeoutException)
|
|
{
|
|
_logger.LogWarning("SqliteAuditWriter writer loop did not drain within 5s of dispose.");
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
// The loop's per-batch try/catch already routed individual failures
|
|
// to pending TCSes; a top-level fault here is unexpected.
|
|
_logger.LogError(ex, "SqliteAuditWriter writer loop faulted during dispose.");
|
|
}
|
|
|
|
lock (_writeLock)
|
|
{
|
|
if (_disposed) return;
|
|
_disposed = true;
|
|
_connection.Dispose();
|
|
}
|
|
|
|
// Dispose the dedicated read connection after the writer
|
|
// is fully drained and closed. _readLock is taken to fence out any
|
|
// in-flight read caller that grabbed the lock before _disposed flipped
|
|
// — they observe ObjectDisposedException on the next attempt.
|
|
lock (_readLock)
|
|
{
|
|
_readConnection.Dispose();
|
|
}
|
|
}
|
|
|
|
/// <summary>An audit event awaiting persistence by the background writer.</summary>
|
|
private sealed class PendingAuditEvent
|
|
{
|
|
/// <summary>Initializes a new instance of the PendingAuditEvent class.</summary>
|
|
/// <param name="evt">The canonical audit event to persist.</param>
|
|
/// <param name="forwardState">Initial site-local forwarding state written to the sidecar row (always Pending for fresh events).</param>
|
|
public PendingAuditEvent(AuditEvent evt, AuditForwardState forwardState)
|
|
{
|
|
Event = evt;
|
|
ForwardState = forwardState;
|
|
Completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
}
|
|
|
|
/// <summary>The canonical audit event to persist.</summary>
|
|
public AuditEvent Event { get; }
|
|
/// <summary>Initial forwarding state for this row's sidecar (bound to audit_forward_state.ForwardState).</summary>
|
|
public AuditForwardState ForwardState { get; }
|
|
/// <summary>Task completion source for write completion signaling.</summary>
|
|
public TaskCompletionSource Completion { get; }
|
|
}
|
|
}
|