feat(audit-log): surface partition-purge failure as health event + counter

This commit is contained in:
Joseph Doherty
2026-07-09 06:49:28 -04:00
parent b59aa2d717
commit 14c4df54f9
9 changed files with 202 additions and 5 deletions
@@ -40,11 +40,13 @@ public sealed class AuditCentralHealthSnapshot
: IAuditCentralHealthSnapshot,
ICentralAuditWriteFailureCounter,
IAuditRedactionFailureCounter,
IAuditInboundCeilingHitsCounter
IAuditInboundCeilingHitsCounter,
IAuditPurgeFailureCounter
{
private int _centralAuditWriteFailures;
private int _auditRedactionFailure;
private int _auditInboundCeilingHits;
private int _auditPurgeFailures;
private readonly ConcurrentDictionary<string, bool> _stalled = new();
/// <inheritdoc/>
@@ -59,6 +61,10 @@ public sealed class AuditCentralHealthSnapshot
public int AuditInboundCeilingHits =>
Interlocked.CompareExchange(ref _auditInboundCeilingHits, 0, 0);
/// <inheritdoc/>
public int PurgeFailures =>
Interlocked.CompareExchange(ref _auditPurgeFailures, 0, 0);
/// <inheritdoc/>
public IReadOnlyDictionary<string, bool> SiteAuditTelemetryStalled =>
new Dictionary<string, bool>(_stalled);
@@ -88,4 +94,8 @@ public sealed class AuditCentralHealthSnapshot
/// <inheritdoc/>
void IAuditInboundCeilingHitsCounter.Increment() =>
Interlocked.Increment(ref _auditInboundCeilingHits);
/// <inheritdoc/>
void IAuditPurgeFailureCounter.Increment() =>
Interlocked.Increment(ref _auditPurgeFailures);
}
@@ -149,7 +149,14 @@ public class AuditLogPurgeActor : ReceiveActor
return;
}
// Optional purge-failure counter — resolved from the same per-tick scope the repository
// comes from. NoOp default on site/test roots; the central composition root binds the
// AuditCentralHealthSnapshot so failures surface on the central health dashboard. Mirrors
// AuditLogIngestActor's ICentralAuditWriteFailureCounter resolution shape.
var purgeFailureCounter = scope.ServiceProvider.GetService<IAuditPurgeFailureCounter>();
IReadOnlyList<DateTime> boundaries;
var enumSw = Stopwatch.StartNew();
try
{
boundaries = await repository
@@ -158,10 +165,14 @@ public class AuditLogPurgeActor : ReceiveActor
}
catch (Exception ex)
{
enumSw.Stop();
_logger.LogError(
ex,
"Failed to enumerate eligible AuditLog partition boundaries (threshold {ThresholdUtc:o}); skipping purge tick.",
threshold);
// Enumeration threw before any boundary was selected — publish with the
// DateTime.MinValue sentinel so the failure is still observable on the health surface.
PublishPurgeFailure(eventStream, purgeFailureCounter, DateTime.MinValue, ex, enumSw.ElapsedMilliseconds);
return;
}
@@ -169,7 +180,7 @@ public class AuditLogPurgeActor : ReceiveActor
{
// No whole-month partitions are eligible, but per-channel overrides may
// still expire rows earlier than the global window — run them below.
await RunPerChannelOverridesAsync(repository).ConfigureAwait(false);
await RunPerChannelOverridesAsync(repository, eventStream, purgeFailureCounter).ConfigureAwait(false);
return;
}
@@ -203,6 +214,7 @@ public class AuditLogPurgeActor : ReceiveActor
"Failed to purge AuditLog partition {MonthBoundary:yyyy-MM-dd}; other partitions continue. Elapsed {DurationMs} ms.",
boundary,
sw.ElapsedMilliseconds);
PublishPurgeFailure(eventStream, purgeFailureCounter, boundary, ex, sw.ElapsedMilliseconds);
}
}
@@ -212,7 +224,24 @@ public class AuditLogPurgeActor : ReceiveActor
// switch-out has already dropped whole months older than RetentionDays; these
// deletes only ever expire rows EARLIER than that, so they run last and are a
// strict tightening.
await RunPerChannelOverridesAsync(repository).ConfigureAwait(false);
await RunPerChannelOverridesAsync(repository, eventStream, purgeFailureCounter).ConfigureAwait(false);
}
/// <summary>
/// Publishes an <see cref="AuditLogPurgeFailedEvent"/> on the EventStream and increments the
/// central purge-failure health counter (when wired). Best-effort — the symmetric failure twin
/// of the success-path <see cref="AuditLogPurgedEvent"/> publish, so a silently failing
/// retention job is a health-surface condition, not just a log line (arch-review 04, S2).
/// </summary>
private static void PublishPurgeFailure(
Akka.Event.EventStream eventStream,
IAuditPurgeFailureCounter? counter,
DateTime monthBoundary,
Exception ex,
long elapsedMilliseconds)
{
eventStream.Publish(new AuditLogPurgeFailedEvent(monthBoundary, ex.Message, elapsedMilliseconds));
counter?.Increment();
}
/// <summary>
@@ -224,7 +253,12 @@ public class AuditLogPurgeActor : ReceiveActor
/// per-boundary error isolation of the partition switch-out loop.
/// </summary>
/// <param name="repository">The repository resolved for this tick's DI scope.</param>
private async Task RunPerChannelOverridesAsync(IAuditLogRepository repository)
/// <param name="eventStream">The actor-system EventStream captured before the first await.</param>
/// <param name="purgeFailureCounter">Optional central purge-failure health counter (NoOp on site/test roots).</param>
private async Task RunPerChannelOverridesAsync(
IAuditLogRepository repository,
Akka.Event.EventStream eventStream,
IAuditPurgeFailureCounter? purgeFailureCounter)
{
var overrides = _auditOptions.PerChannelRetentionDays;
if (overrides is null || overrides.Count == 0)
@@ -281,6 +315,10 @@ public class AuditLogPurgeActor : ReceiveActor
channel,
days,
sw.ElapsedMilliseconds);
// A per-channel DELETE failure is a purge failure too — surface it alongside the
// switch-out failures. MonthBoundary is DateTime.MinValue (this is not a
// whole-month switch), matching the enumeration-phase sentinel.
PublishPurgeFailure(eventStream, purgeFailureCounter, DateTime.MinValue, ex, sw.ElapsedMilliseconds);
}
}
}
@@ -0,0 +1,29 @@
namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central;
/// <summary>
/// Symmetric failure twin of <see cref="AuditLogPurgedEvent"/>. Published on the
/// actor-system EventStream by <see cref="AuditLogPurgeActor"/> whenever a purge
/// step throws — a partition switch-out, a per-channel override DELETE, or the
/// boundary enumeration that precedes them. A silently failing retention job is a
/// health-surface condition, not just a log line, so this event lets the central
/// health collector + ops dashboards observe purge failures without coupling to
/// the actor (arch-review 04, S2).
/// </summary>
/// <param name="MonthBoundary">
/// The pf_AuditLog_Month lower-bound boundary whose switch-out failed. When the
/// failure occurs BEFORE any specific boundary is selected — i.e. the pre-purge
/// boundary enumeration threw — this is <see cref="System.DateTime.MinValue"/> as a
/// sentinel for "enumeration phase, no boundary in hand".
/// </param>
/// <param name="Error">
/// The failure's message (<see cref="System.Exception.Message"/>). The full
/// exception is logged at Error on the actor's logger; this carries the short form
/// for surfacing on health/ops without leaking a full stack trace onto the stream.
/// </param>
/// <param name="ElapsedMilliseconds">
/// Wall-clock time spent on the failed step before it threw, in milliseconds.
/// </param>
public sealed record AuditLogPurgeFailedEvent(
DateTime MonthBoundary,
string Error,
long ElapsedMilliseconds);
@@ -61,6 +61,16 @@ public interface IAuditCentralHealthSnapshot
/// </summary>
int AuditInboundCeilingHits { get; }
/// <summary>
/// Count of AuditLog partition-purge failures since process start. Incremented by
/// <see cref="AuditLogPurgeActor"/> every time a purge step throws — a partition
/// switch-out, a per-channel override DELETE, or the pre-purge boundary enumeration.
/// A sustained non-zero count means the retention job is silently not draining the
/// table; each failure also publishes an <see cref="AuditLogPurgeFailedEvent"/> on the
/// EventStream (arch-review 04, S2).
/// </summary>
int PurgeFailures { get; }
/// <summary>
/// Per-site latched stalled state: <c>true</c> when the
/// <see cref="SiteAuditReconciliationActor"/> has observed two
@@ -0,0 +1,21 @@
namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central;
/// <summary>
/// Counter sink invoked by <see cref="AuditLogPurgeActor"/> every time a purge
/// step throws (partition switch-out, per-channel override DELETE, or the pre-purge
/// boundary enumeration). Mirrors <see cref="ICentralAuditWriteFailureCounter"/>
/// one-for-one — same one-method contract, same NoOp default, same
/// never-abort-the-tick invariant (the purge actor swallows the failure per
/// boundary and continues).
/// </summary>
/// <remarks>
/// A NoOp default is the safe fallback for site/test composition roots that do not
/// wire the central health surface; <see cref="AuditCentralHealthSnapshot"/> is the
/// production binding that routes increments into the aggregated central health
/// snapshot's <see cref="IAuditCentralHealthSnapshot.PurgeFailures"/> read surface.
/// </remarks>
public interface IAuditPurgeFailureCounter
{
/// <summary>Increment the central AuditLog purge-failure counter by one.</summary>
void Increment();
}
@@ -0,0 +1,16 @@
namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central;
/// <summary>
/// Default <see cref="IAuditPurgeFailureCounter"/> binding used when the central
/// health surface (<see cref="AuditCentralHealthSnapshot"/>) has not been wired
/// (test composition roots, site-only hosts). Drops every increment on the floor.
/// Mirrors <see cref="NoOpCentralAuditWriteFailureCounter"/>.
/// </summary>
public sealed class NoOpAuditPurgeFailureCounter : IAuditPurgeFailureCounter
{
/// <inheritdoc/>
public void Increment()
{
// intentional no-op
}
}
@@ -212,6 +212,12 @@ public static class ServiceCollectionExtensions
// redaction-failure counters.
services.TryAddSingleton<IAuditInboundCeilingHitsCounter, NoOpAuditInboundCeilingHitsCounter>();
// AuditLog purge-failure counter — NoOp default for site/test
// composition roots. AddAuditLogCentralMaintenance below replaces this binding
// with the AuditCentralHealthSnapshot implementation so partition/channel purge
// failures surface on the central dashboard alongside the write-failure counter.
services.TryAddSingleton<IAuditPurgeFailureCounter, NoOpAuditPurgeFailureCounter>();
// Central direct-write audit writer used by
// NotificationOutboxActor and Inbound API to
// emit AuditLog rows that originate ON central, not via site telemetry.
@@ -419,6 +425,11 @@ public static class ServiceCollectionExtensions
// ICentralAuditWriteFailureCounter above.
services.Replace(ServiceDescriptor.Singleton<IAuditInboundCeilingHitsCounter>(
sp => sp.GetRequiredService<AuditCentralHealthSnapshot>()));
// Replace the NoOp IAuditPurgeFailureCounter with the AuditCentralHealthSnapshot so
// partition/channel purge failures surface on the central dashboard. Same
// singleton-forward pattern as ICentralAuditWriteFailureCounter above (arch-review 04, S2).
services.Replace(ServiceDescriptor.Singleton<IAuditPurgeFailureCounter>(
sp => sp.GetRequiredService<AuditCentralHealthSnapshot>()));
return services;
}