diff --git a/docs/requirements/Component-AuditLog.md b/docs/requirements/Component-AuditLog.md index 859cbb55..c2d5c6aa 100644 --- a/docs/requirements/Component-AuditLog.md +++ b/docs/requirements/Component-AuditLog.md @@ -390,9 +390,25 @@ MS SQL for direct-write events). Unredacted secrets never persist. - **Purge actor:** `AuditLogPurgeActor` singleton on the active central node runs daily, switches out any partition whose latest `OccurredAtUtc` is older than the retention window, then applies any per-channel overrides (see below), - and emits an `AuditLog:Purged` event (partition range, rowcount, duration) per + and emits an `AuditLogPurgedEvent` (partition range, rowcount, duration) per switched partition. A partition-maintenance step rolls forward each month, creating the next month's partition ahead of time. +- **Purge failure is a health metric, not just a log line:** every purge step that + throws — a partition switch-out, a per-channel override DELETE, or the pre-purge + boundary enumeration — publishes an `AuditLogPurgeFailedEvent` (month boundary, + error message, elapsed ms; `DateTime.MinValue` boundary for the enumeration/channel + phases that have no single month in hand) and increments the `PurgeFailures` central + health counter on `AuditCentralHealthSnapshot` (alongside `CentralAuditWriteFailures`). + This is the symmetric failure twin of the success-path `AuditLogPurgedEvent`: a + silently failing retention job would otherwise be invisible until the table grew + unbounded. Per-boundary/per-channel error isolation is unchanged — a single failure + still never abandons the rest of the tick. +- **Maintenance command timeout:** the switch-out drop-and-rebuild dance and each + per-channel `DELETE TOP` batch run with an explicit command timeout + (`AuditLog:Purge:MaintenanceCommandTimeoutMinutes`, default 30, floor 1 min) rather + than the ~30 s ADO.NET default, which could abort the metadata-only SWITCH mid-dance + on a large or contended partition and leave the live table without + `UX_AuditLog_EventId` until a later tick's CATCH branch rebuilt it. - **Per-channel retention overrides (M5.5 T3):** `AuditLog:PerChannelRetentionDays` is a dictionary keyed by canonical channel name (`ApiOutbound`, `DbOutbound`, `Notification`, `ApiInbound`) whose value is a retention window in days that diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditCentralHealthSnapshot.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditCentralHealthSnapshot.cs index e5ab7fb8..e7ecc5f7 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditCentralHealthSnapshot.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditCentralHealthSnapshot.cs @@ -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 _stalled = new(); /// @@ -59,6 +61,10 @@ public sealed class AuditCentralHealthSnapshot public int AuditInboundCeilingHits => Interlocked.CompareExchange(ref _auditInboundCeilingHits, 0, 0); + /// + public int PurgeFailures => + Interlocked.CompareExchange(ref _auditPurgeFailures, 0, 0); + /// public IReadOnlyDictionary SiteAuditTelemetryStalled => new Dictionary(_stalled); @@ -88,4 +94,8 @@ public sealed class AuditCentralHealthSnapshot /// void IAuditInboundCeilingHitsCounter.Increment() => Interlocked.Increment(ref _auditInboundCeilingHits); + + /// + void IAuditPurgeFailureCounter.Increment() => + Interlocked.Increment(ref _auditPurgeFailures); } diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogPurgeActor.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogPurgeActor.cs index d973f8a5..ed1ea3d2 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogPurgeActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogPurgeActor.cs @@ -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(); + IReadOnlyList 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); + } + + /// + /// Publishes an on the EventStream and increments the + /// central purge-failure health counter (when wired). Best-effort — the symmetric failure twin + /// of the success-path publish, so a silently failing + /// retention job is a health-surface condition, not just a log line (arch-review 04, S2). + /// + 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(); } /// @@ -224,7 +253,12 @@ public class AuditLogPurgeActor : ReceiveActor /// per-boundary error isolation of the partition switch-out loop. /// /// The repository resolved for this tick's DI scope. - private async Task RunPerChannelOverridesAsync(IAuditLogRepository repository) + /// The actor-system EventStream captured before the first await. + /// Optional central purge-failure health counter (NoOp on site/test roots). + 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); } } } diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogPurgeFailedEvent.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogPurgeFailedEvent.cs new file mode 100644 index 00000000..0578ff3d --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogPurgeFailedEvent.cs @@ -0,0 +1,29 @@ +namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central; + +/// +/// Symmetric failure twin of . Published on the +/// actor-system EventStream by 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). +/// +/// +/// 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 as a +/// sentinel for "enumeration phase, no boundary in hand". +/// +/// +/// The failure's 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. +/// +/// +/// Wall-clock time spent on the failed step before it threw, in milliseconds. +/// +public sealed record AuditLogPurgeFailedEvent( + DateTime MonthBoundary, + string Error, + long ElapsedMilliseconds); diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/IAuditCentralHealthSnapshot.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/IAuditCentralHealthSnapshot.cs index 64f74449..57042380 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/IAuditCentralHealthSnapshot.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/IAuditCentralHealthSnapshot.cs @@ -61,6 +61,16 @@ public interface IAuditCentralHealthSnapshot /// int AuditInboundCeilingHits { get; } + /// + /// Count of AuditLog partition-purge failures since process start. Incremented by + /// 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 on the + /// EventStream (arch-review 04, S2). + /// + int PurgeFailures { get; } + /// /// Per-site latched stalled state: true when the /// has observed two diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/IAuditPurgeFailureCounter.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/IAuditPurgeFailureCounter.cs new file mode 100644 index 00000000..084a5307 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/IAuditPurgeFailureCounter.cs @@ -0,0 +1,21 @@ +namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central; + +/// +/// Counter sink invoked by every time a purge +/// step throws (partition switch-out, per-channel override DELETE, or the pre-purge +/// boundary enumeration). Mirrors +/// 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). +/// +/// +/// A NoOp default is the safe fallback for site/test composition roots that do not +/// wire the central health surface; is the +/// production binding that routes increments into the aggregated central health +/// snapshot's read surface. +/// +public interface IAuditPurgeFailureCounter +{ + /// Increment the central AuditLog purge-failure counter by one. + void Increment(); +} diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/NoOpAuditPurgeFailureCounter.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/NoOpAuditPurgeFailureCounter.cs new file mode 100644 index 00000000..eac6ff11 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/NoOpAuditPurgeFailureCounter.cs @@ -0,0 +1,16 @@ +namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central; + +/// +/// Default binding used when the central +/// health surface () has not been wired +/// (test composition roots, site-only hosts). Drops every increment on the floor. +/// Mirrors . +/// +public sealed class NoOpAuditPurgeFailureCounter : IAuditPurgeFailureCounter +{ + /// + public void Increment() + { + // intentional no-op + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/ServiceCollectionExtensions.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/ServiceCollectionExtensions.cs index 6a72a30f..ef000387 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/ServiceCollectionExtensions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/ServiceCollectionExtensions.cs @@ -212,6 +212,12 @@ public static class ServiceCollectionExtensions // redaction-failure counters. services.TryAddSingleton(); + // 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(); + // 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( sp => sp.GetRequiredService())); + // 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( + sp => sp.GetRequiredService())); return services; } diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/AuditLogPurgeActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/AuditLogPurgeActorTests.cs index 67411f4f..6ac9dabc 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/AuditLogPurgeActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/AuditLogPurgeActorTests.cs @@ -543,4 +543,50 @@ public class AuditLogPurgeActorTests : TestKit, IClassFixture { boundary }, + ThrowOnBoundary = boundary, + BoundaryException = new InvalidOperationException("simulated switch timeout"), + }; + + var snapshot = new AuditCentralHealthSnapshot(); + var services = new ServiceCollection(); + services.AddScoped(_ => repo); + services.AddSingleton(snapshot); + services.AddSingleton(snapshot); + var sp = services.BuildServiceProvider(); + + var probe = CreateTestProbe(); + Sys.EventStream.Subscribe(probe.Ref, typeof(AuditLogPurgeFailedEvent)); + + Sys.ActorOf(Props.Create(() => new AuditLogPurgeActor( + sp, + Options.Create(FastTickOptions()), + Options.Create(new AuditLogOptions()), + NullLogger.Instance))); + + var evt = probe.ExpectMsg(TimeSpan.FromSeconds(5)); + Assert.Equal(boundary, evt.MonthBoundary); + Assert.Contains("simulated switch timeout", evt.Error); + Assert.True(evt.ElapsedMilliseconds >= 0); + + AwaitAssert( + () => Assert.True(snapshot.PurgeFailures >= 1, + $"expected PurgeFailures >= 1, got {snapshot.PurgeFailures}"), + duration: TimeSpan.FromSeconds(3), + interval: TimeSpan.FromMilliseconds(50)); + } }