perf(audit-log): cached-call forwarder enqueues the audit row instead of awaiting its commit (194 ms -> 25 ms)

Measuring the tracking-store queue on the rig moved nothing: the DB-done -> Notify.Send
gap stayed at 108 ms. The cost was the audit half of CachedCallTelemetryForwarder:
SqliteAuditWriter.WriteAsync enqueues synchronously but returns the BATCH completion,
and the writer loop coalesces for FlushIntervalMs (50 ms) before it commits, so each
awaited audit row cost ~50 ms + fsync and a cached call awaited three (submit,
attempted, resolve) on the script thread. Every other script-path emitter already
fired-and-observed; only this forwarder awaited.

TryEmitAuditAsync now hands the row to the writer and observes the completion on a
fault-logging continuation (the FallbackAuditWriter ring-buffer fallback still runs
inside that task). Best-effort contract unchanged; per-store FIFO unchanged; nothing
joins the audit and tracking stores by commit order.

Rig, FlowOutboundDemo, 5-run medians: inbound 194 -> 25 ms, site script run 176 -> 17 ms,
CachedWrite submit->done 61 -> 11 ms, DB done -> Notify.Send 108 -> 0.1 ms. The tracking
queue stays (three fewer fsyncs per call, batched). Docs corrected to attribute the cost
to the flush wait, with the trap recorded in CLAUDE.md and Component-SiteRuntime.md.

Tests: ForwardAsync returns while the audit commit is still pending (never-completing
writer task) and still records tracking; a late commit fault is swallowed.

Claude-Session: https://claude.ai/code/session_01NcMG6xXdsJbmD2wYAgkK29
This commit is contained in:
Joseph Doherty
2026-09-06 12:35:41 -04:00
parent d91135f960
commit 0ccbe140ae
6 changed files with 129 additions and 23 deletions
@@ -34,8 +34,22 @@ namespace ZB.MOM.WW.ScadaBridge.AuditLog.Site.Telemetry;
/// returns normally.
/// </para>
/// <para>
/// <b>Neither half is awaited to durability (2026-09-06).</b> The audit row is
/// enqueued on the writer — <c>SqliteAuditWriter.WriteAsync</c> returns the
/// batch's completion task, and that batch coalesces for <c>FlushIntervalMs</c>
/// (50 ms) before it commits — so awaiting it here made every cached call pay
/// the flush wait three times over, on the script thread (measured: ~160 ms of a
/// 176 ms <c>Database.CachedWrite</c> run on the rig; see
/// <c>docs/plans/2026-09-06-tracking-store-async-writes-design.md</c>). The
/// completion is observed on a continuation that logs a fault; the
/// <c>FallbackAuditWriter</c> chain still routes a failed commit to its ring
/// buffer inside that task. The tracking half is a queue write since the same
/// day (<c>OperationTrackingStore</c>). The ordering the two stores need is
/// per-store FIFO, which both keep; nothing joins them by commit order.
/// </para>
/// <para>
/// <b>Local-write only — the wire push is the drain actor's job.</b> This
/// forwarder is deliberately synchronous against the two site-local SQLite
/// forwarder writes only to the two site-local SQLite
/// stores and never pushes to central itself. The site→central transport is
/// now live: <c>SiteCommunicationAuditClient</c> is the production binding of
/// <see cref="ISiteStreamAuditClient"/> on site roles (with
@@ -98,25 +112,48 @@ public sealed class CachedCallTelemetryForwarder : ICachedCallTelemetryForwarder
await TryEmitTrackingAsync(telemetry, ct).ConfigureAwait(false);
}
private async Task TryEmitAuditAsync(CachedCallTelemetry telemetry, CancellationToken ct)
private Task TryEmitAuditAsync(CachedCallTelemetry telemetry, CancellationToken ct)
{
// Enqueue-and-observe, NOT await-to-commit: WriteAsync hands the row to
// the writer synchronously and returns the batch's completion, which
// lands only after the writer's FlushIntervalMs coalescing wait + fsync.
// The caller is a script thread; it must not pay for that. Same pattern as
// AuditingDbCommand / ScriptRunLauncher's own audit rows.
try
{
await _auditWriter.WriteAsync(telemetry.Audit, ct).ConfigureAwait(false);
var write = _auditWriter.WriteAsync(telemetry.Audit, ct);
if (write.IsCompletedSuccessfully)
return Task.CompletedTask;
if (write.IsFaulted)
{
LogAuditFailure(telemetry, write.Exception!.GetBaseException());
return Task.CompletedTask;
}
_ = write.ContinueWith(
t => LogAuditFailure(telemetry, t.Exception!.GetBaseException()),
CancellationToken.None,
TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously,
TaskScheduler.Default);
}
catch (Exception ex)
{
// alog.md §7 best-effort contract — log and swallow. The audit
// pipeline's own retry/recovery (RingBufferFallback in the
// FallbackAuditWriter) handles transient writer failures upstream;
// a throw bubbling up here means the writer's own swallow contract
// failed, which is itself best-effort-handled.
// Kind/Status are domain fields carried in DetailsJson — decompose to log them.
var d = AuditRowProjection.Decompose(telemetry.Audit);
_logger.LogWarning(ex,
"CachedCallTelemetryForwarder: audit emission threw for EventId {EventId} (Kind {Kind}, Status {Status}, sqlite {SqliteError})",
d.EventId, d.Kind, d.Status, SqliteErrorCodes.Describe(ex));
LogAuditFailure(telemetry, ex);
}
return Task.CompletedTask;
}
private void LogAuditFailure(CachedCallTelemetry telemetry, Exception ex)
{
// alog.md §7 best-effort contract — log and swallow. The audit
// pipeline's own retry/recovery (RingBufferFallback in the
// FallbackAuditWriter) handles transient writer failures upstream;
// a throw reaching here means the writer's own swallow contract
// failed, which is itself best-effort-handled.
// Kind/Status are domain fields carried in DetailsJson — decompose to log them.
var d = AuditRowProjection.Decompose(telemetry.Audit);
_logger.LogWarning(ex,
"CachedCallTelemetryForwarder: audit emission failed for EventId {EventId} (Kind {Kind}, Status {Status}, sqlite {SqliteError})",
d.EventId, d.Kind, d.Status, SqliteErrorCodes.Describe(ex));
}
private async Task TryEmitTrackingAsync(CachedCallTelemetry telemetry, CancellationToken ct)