diff --git a/docs/requirements/Component-AuditLog.md b/docs/requirements/Component-AuditLog.md index e12e0d89..ad0858c2 100644 --- a/docs/requirements/Component-AuditLog.md +++ b/docs/requirements/Component-AuditLog.md @@ -255,13 +255,21 @@ has no anchor and falls back to the unbounded scan (correctness over speed). A SQLite database file on each site node, alongside the Store-and-Forward buffer. Same schema as central minus `IngestedAtUtc` (irrelevant at the source), plus a `ForwardState` column with values `Pending | Forwarded | Reconciled` that -drives the telemetry loop and reconciliation pull. `SourceNode` is stamped by the +drives the telemetry loop and reconciliation pull — and, since review 2026-10-21 +(06/PERF-1, 06/IMPL-1), two site-only **push-exit** states, `PushAbandoned` and +`Rejected`, for rows that left the telemetry push queue *without* a central ack. +`Forwarded` means "central acked it"; `Pending`, `PushAbandoned` and `Rejected` +together are **owed** to central: the reconciliation pull serves exactly those, +a covering cursor retires them to `Reconciled`, and the retention purge never +drops them. The extra states are strings in the site sidecar only (no schema +change, never on the wire); a downgraded node ignores rows in a state it does not +know. `SourceNode` is stamped by the writing node itself (`node-a` / `node-b`) at append time and travels with the row through telemetry and reconciliation unchanged. **Site SQLite retention rule (hard invariant):** -> A row is eligible for purge only when both `OccurredAtUtc < retention threshold` AND `ForwardState IN ('Forwarded', 'Reconciled')`. Pending rows are never purged. +> A row is eligible for purge only when both `OccurredAtUtc < retention threshold` AND `ForwardState IN ('Forwarded', 'Reconciled')`. Pending rows — and the owed push-exit states `PushAbandoned` / `Rejected` — are never purged. A prolonged central outage will grow the site audit table indefinitely until central is reachable again. This is intentional — losing audit rows to make @@ -302,8 +310,21 @@ round-trip on the hot path. A `SiteAuditTelemetryActor` singleton drives the forwarding loop: select up to N `Pending` rows ordered by `OccurredAtUtc`, batch-send them to central via the existing `SiteStream` gRPC channel as `IngestAuditEvents(events)`, and on -central-ack flip `ForwardState = 'Forwarded'` for accepted IDs. Rejected IDs -stay `Pending` for the next sweep. Cadence is short (default 5 s) when +central-ack flip `ForwardState = 'Forwarded'` for accepted IDs. IDs central +answered for but did not accept stay `Pending` for the next sweep, with +`AttemptCount` / `LastAttemptUtc` bumped on every such refusal; after +`AuditLog:SiteTelemetry:MaxRejectedPushAttempts` (default 10) refusals a row +leaves the push queue as `Rejected` and bumps `SiteAuditWriteFailures` +(review 2026-10-21 06/IMPL-1). The drain reads oldest-first, so before this one +row central can never insert (e.g. an over-width column) sat at the head of every +batch forever, failing each set-based insert at central into the per-row fallback. +`Rejected` is not a drop: the reconciliation pull still serves the row, so a +transient refusal still lands, and a genuinely poison row is recorded centrally as +a `ReconciliationAbandoned` row before its cursor retires it. A transport fault (no +answer) is never counted as a refusal. A cached-lifecycle row whose tracking +snapshot is gone past `CachedTrackingGraceSeconds` leaves the cached drain as +`PushAbandoned` (it used to be marked `Forwarded`, which only worked because the +pull re-served every `Forwarded` row). Cadence is short (default 5 s) when non-empty, longer (default 30 s) when idle; telemetry runs on a dedicated dispatcher. @@ -349,10 +370,27 @@ still replied while the outer Asks are still waiting. ### Reconciliation pull (self-healing for missed telemetry) A central `SiteAuditReconciliationActor` periodically (default 5 min per site) -asks each site for its oldest `Pending` row and pending count; if backlog is -non-draining (e.g., telemetry actor wedged), central issues a -`PullAuditEvents(sinceUtc, batchSize[, afterId])` and inserts-if-not-exists. -Same self-healing pattern as Site Call Audit's reconciliation of `SiteCalls`. +issues `PullAuditEvents(sinceUtc, batchSize[, afterId])` against each site and +inserts-if-not-exists. Same self-healing pattern as Site Call Audit's +reconciliation of `SiteCalls`. + +**The pull serves only rows central has not acknowledged (review 2026-10-21 +06/PERF-1).** `ReadPendingSinceAsync` returns `Pending`, `PushAbandoned` and +`Rejected` rows — never `Forwarded`. It used to serve `Forwarded` rows too, so +every row the push had already delivered was re-shipped and re-inserted +(row-at-a-time) one tick later. That capped the pull at 256 rows per 5-minute tick +(≈0.85 rows/s per site, 60× below the push), so any site above that rate fell +permanently behind: a genuinely lost row waited behind days of already-forwarded +rows until the 7-day site purge, the stalled badge latched true on a healthy busy +site, and every central failover (cursor reset to `MinValue`) restarted a +multi-day replay (06/COR-3, resolved by this change — the re-pull after a reset is +now only the owed set). Central ingests each page with ONE set-based +`InsertManyIfNotExistsAsync` (per-row fallback only when the batch throws, so the +poison-row retry/abandon semantics are unchanged) and **pages within a tick** — up +to `AuditLog:Reconciliation:MaxPagesPerTick` (default 50) pulls while the site +reports `MoreAvailable` and the cursor advances, mirroring `SiteCallAuditActor`. +**Mixed versions:** a pre-PERF-1 site still serves `Forwarded` rows; central +tolerates that unchanged (idempotent on `EventId`, merely more pages). **`PullTraceEvents` is a point read, not a reconciliation path (2026-09-05).** The Flow Debugger's value-change run cannot get its steps inline (the device echo @@ -407,10 +445,12 @@ the legacy inclusive `>=` read as the compatible fallback. The cursor stays in-m and resets on singleton restart; idempotent `InsertIfNotExistsAsync` still absorbs any re-pulled duplicates, so exactness is an optimization, not a correctness dependency. -Unlike Site Call Audit, this actor issues ONE pull per tick rather than paging within -it — a lagging drain is meant to surface as the stalled signal. The id tiebreak is -what makes that safe against a same-instant burst larger than one batch: the cursor -advances on every tick even when the timestamp cannot. +Like Site Call Audit, the actor pages within a tick (bounded by `MaxPagesPerTick`), so +the stalled signal fires only when a site still reports `MoreAvailable` after the page +cap — or its cursor is held back by a row that keeps failing — on +`StalledAfterNonDrainingCycles` consecutive ticks: a genuine non-drain, not a busy +site. The id tiebreak keeps paging safe against a same-instant burst larger than one +batch: the cursor advances on every page even when the timestamp cannot. **Only rows that were actually served may retire.** `OccurredAtUtc` is stamped by the caller, so a row can be *inserted* after a batch was served yet carry a timestamp @@ -421,13 +461,14 @@ and, being `Reconciled`, purged on age. That is silent audit loss, and it is a f mode the pre-WP2.3 explicit id-set flip could not produce, so `MarkReconciledUpToAsync` carries a second, insertion-order bound: -> A `Pending` row retires only if its insertion order is at or below the high-water -> mark of rows this site node has actually served from `ReadPendingSinceAsync` -> (SQLite `rowid`). A `Forwarded` row is exempt — central ACKED it over the telemetry -> push path, which is proof independent of the pull. +> An owed row (`Pending` / `PushAbandoned` / `Rejected`) retires only if its insertion +> order is at or below the high-water mark of rows this site node has actually served +> from `ReadPendingSinceAsync` (SQLite `rowid`). `Forwarded` rows are not touched by the +> cursor at all — central ACKED them over the telemetry push path, they are already +> purge-eligible, and the pull never served them. -The bound is per-process, so after a site-node restart the first pull retires only -`Forwarded` rows and the pull after it resumes normal retirement — conservative in the +The bound is per-process, so after a site-node restart the first pull retires nothing +and the pull after it resumes normal retirement — conservative in the safe direction (retirement is delayed, never a row lost). The gRPC handler's ordering (retire, *then* read) is what keeps the bound from ever vouching for the batch it is about to serve. @@ -586,8 +627,8 @@ MS SQL for direct-write events). Unredacted secrets never persist. best-effort drop with a counter, never unbounded memory (review 2026-10-21 06/MEM-1). Audit failure still never aborts the action. - **Reconciliation as fallback.** If two consecutive reconciliation cycles - report a non-draining backlog, the supervisor restarts the telemetry actor - and a `SiteAuditTelemetryStalled` event fires. + still report a non-draining backlog after paging up to `MaxPagesPerTick`, a + `SiteAuditTelemetryStalled` event fires. - **Permanent-abandonment record.** A reconciliation pull row that fails to insert on every retry up to the permanent-abandon threshold is dropped and the cursor advances past it (so one broken row can't block the site forever). The diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs index a450be32..55065999 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs @@ -5,6 +5,7 @@ using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Integration; using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit; using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; @@ -24,8 +25,12 @@ namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central; /// ( + /// IngestAuditEvents) is the primary mechanism. This actor exists so a /// missed push (gRPC blip, central restart, site offline) is eventually -/// repaired by central re-pulling whatever the site still has in -/// Pending/Forwarded state. Idempotency on +/// repaired by central re-pulling whatever the site has NOT yet had acknowledged — +/// Pending rows plus rows that left the push queue without an ack +/// (PushAbandoned, Rejected). Forwarded rows are no longer +/// re-served: central already acked them, and re-shipping every pushed row capped the +/// pull at a throughput no busy site could keep up with (review 2026-10-21 06/PERF-1). +/// A pre-PERF-1 site still serves them; idempotency on /// makes duplicate /// arrivals from both paths a silent no-op. /// @@ -34,15 +39,18 @@ namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central; /// kept in-memory for the actor's lifetime. The cluster singleton normally /// survives the host process; on a deliberate failover OR a singleton restart /// the cursors reset to . That is conservative -/// but correct — the next tick simply asks for everything the site still has, -/// and idempotent ingest swallows the dupes. Persisting cursors to MS SQL was +/// but correct — the next tick simply asks for every row the site still owes, +/// and idempotent ingest swallows the dupes. Because the site serves only +/// unacknowledged rows, that re-pull is the small owed set, not a multi-day +/// replay of every forwarded row (review 2026-10-21 06/COR-3). Persisting cursors to MS SQL was /// considered and rejected: the cost of a write per tick outweighs the /// rare benefit of avoiding one over-broad pull after a restart. /// /// -/// Stalled detection. The brief calls a site "stalled" when two -/// consecutive pull cycles BOTH return non-empty AND MoreAvailable=true -/// — i.e. the backlog isn't draining. The actor publishes +/// Stalled detection. Each tick pages through a site (up to +/// pulls, each ingested with +/// one set-based insert). A site is "stalled" when two consecutive ticks BOTH still end +/// non-empty AND MoreAvailable=true after that — i.e. the backlog isn't draining. The actor publishes /// on the actor system's /// EventStream so a future ICentralHealthCollector bridge /// can flip the health metric without coupling this actor to the health @@ -173,6 +181,8 @@ public class SiteAuditReconciliationActor : ReceiveActor private ICancelable? _timer; + private void RememberAbandoned(Guid eventId) => _abandonedEventIds.Add(eventId); + /// /// Initializes the reconciliation actor with its dependencies and registers the tick handler. /// @@ -288,23 +298,59 @@ public class SiteAuditReconciliationActor : ReceiveActor } } + /// + /// Drains one site for this tick: issues up to + /// PullAuditEvents + /// RPCs back to back while the site reports MoreAvailable and the cursor keeps + /// advancing, then updates the stalled signal from the LAST page. + /// + /// + /// Paging within a tick (review 2026-10-21 06/PERF-1, mirroring SiteCallAuditActor's + /// ≤50-page loop) is what makes the stalled latch mean something: one pull per tick capped + /// the pull at BatchSize rows per ReconciliationInterval (0.85 rows/s at the + /// defaults), so any busy-but-healthy site reported MoreAvailable on every tick and + /// latched SiteAuditTelemetryStalled permanently. Now a site is non-draining only when + /// it still has more rows after the page cap, or its cursor is held back by a row that keeps + /// failing — a genuine non-drain. + /// + private async Task PullSiteAsync(SiteEntry site, IAuditLogRepository repository, Akka.Event.EventStream eventStream) + { + var maxPages = Math.Max(1, _options.MaxPagesPerTick); + var nonDraining = false; + for (var page = 0; page < maxPages; page++) + { + var (response, cursorAdvanced) = await PullPageAsync(site, repository).ConfigureAwait(false); + nonDraining = response.MoreAvailable && response.Events.Count > 0; + if (!nonDraining || !cursorAdvanced) + { + break; + } + } + + UpdateStalledState(site.SiteId, draining: !nonDraining, eventStream); + } + /// /// Issues one PullAuditEvents RPC against the site, ingests the /// returned rows idempotently into the central repository, and advances /// the composite (, ) - /// cursor to the maximum row observed. The brief's "saturate until backlog clears" intent is - /// met by the natural cadence — each tick issues one pull, and a backed-up site - /// drains across consecutive ticks. The stalled signal (two non-draining - /// ticks in a row) surfaces when that drain isn't keeping up. + /// cursor to the maximum row observed. /// /// - /// Unlike SiteCallAuditActor this does NOT page within a tick; one pull per tick is - /// deliberate (the stalled signal is how a lagging drain surfaces). The composite cursor is - /// what makes that safe against a same-instant burst larger than one batch: the id tiebreak - /// advances even when the timestamp cannot, so consecutive ticks make real progress instead - /// of re-serving the same window forever. + /// The page is ingested with ONE set-based + /// call (review 2026-10-21 + /// 06/PERF-1 — it used to cost one InsertIfNotExistsAsync round trip per row). Only + /// when the batch throws does the page fall back to the per-row path, which keeps the + /// documented poison-row isolation: the failing row is retried up to + /// times with the cursor held back, then + /// permanently abandoned. Re-running rows the failed batch may already have committed is + /// safe — every insert is idempotent on , which is also + /// what lets a pre-PERF-1 site that still re-serves Forwarded rows be pulled + /// unchanged. /// - private async Task PullSiteAsync(SiteEntry site, IAuditLogRepository repository, Akka.Event.EventStream eventStream) + /// The site's response and whether the cursor moved. + private async Task<(PullAuditEventsResponse Response, bool CursorAdvanced)> PullPageAsync( + SiteEntry site, IAuditLogRepository repository) { var cursor = _cursors.TryGetValue(site.SiteId, out var c) ? c @@ -320,6 +366,41 @@ public class SiteAuditReconciliationActor : ReceiveActor var maxAfterId = afterId; var hasUnresolvedFailure = false; var nowUtc = DateTime.UtcNow; + + // Set-based fast path for every row not already known-abandoned. + var batchCommitted = false; + var toInsert = new List(response.Events.Count); + foreach (var evt in response.Events) + { + if (!_abandonedEventIds.Contains(evt.EventId)) + { + toInsert.Add(AuditRowProjection.WithIngestedAtUtc(evt, nowUtc)); + } + } + + if (toInsert.Count == 0) + { + batchCommitted = true; + } + else + { + try + { + await repository.InsertManyIfNotExistsAsync(toInsert).ConfigureAwait(false); + batchCommitted = true; + foreach (var evt in toInsert) + { + _failedInsertAttempts.Remove(evt.EventId); + } + } + catch (Exception ex) + { + _logger.LogWarning(ex, + "Set-based reconciliation ingest of {Count} audit event(s) from site {SiteId} failed; falling back to per-row inserts so one bad row does not sink the page.", + toInsert.Count, site.SiteId); + } + } + foreach (var evt in response.Events) { var advanceForThisRow = false; @@ -335,6 +416,10 @@ public class SiteAuditReconciliationActor : ReceiveActor // row collapse into the existing one anyway. advanceForThisRow = true; } + else if (batchCommitted) + { + advanceForThisRow = true; + } else { try @@ -370,7 +455,7 @@ public class SiteAuditReconciliationActor : ReceiveActor site.SiteId, attempts); _failedInsertAttempts.Remove(evt.EventId); - _abandonedEventIds.Add(evt.EventId); + RememberAbandoned(evt.EventId); advanceForThisRow = true; await EmitAbandonmentRecordAsync(repository, evt, site.SiteId, ex, nowUtc) @@ -413,12 +498,13 @@ public class SiteAuditReconciliationActor : ReceiveActor // another attempt. Once it succeeds (or hits the permanent-abandon // threshold) the cursor unblocks naturally. Both halves move together: holding the // timestamp back while advancing the id would skip the very rows being retried. - _cursors[site.SiteId] = hasUnresolvedFailure + var next = hasUnresolvedFailure ? (since, afterId) : (maxOccurred, maxAfterId); + _cursors[site.SiteId] = next; - var nonDraining = response.MoreAvailable && response.Events.Count > 0; - UpdateStalledState(site.SiteId, draining: !nonDraining, eventStream); + var advanced = next.Item1 != since || !string.Equals(next.Item2, afterId, StringComparison.Ordinal); + return (response, advanced); } /// diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationOptions.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationOptions.cs index 33ae7c06..0e003b1d 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationOptions.cs @@ -3,8 +3,9 @@ namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central; /// /// Tuning knobs for the central singleton. /// Defaults: pull every 5 minutes per site, 256 rows per -/// batch, declare a site "stalled" after two consecutive pull cycles return non-empty -/// AND MoreAvailable=true (the backlog is not draining). +/// page and up to 50 pages per site per tick, declare a site "stalled" after two +/// consecutive cycles still end non-empty AND MoreAvailable=true after the page cap +/// (the backlog is not draining). /// /// /// @@ -77,6 +78,15 @@ public sealed class SiteAuditReconciliationOptions /// public int BatchSize { get; set; } = 256; + /// + /// Maximum PullAuditEvents pages issued against one site in one tick while the + /// site keeps reporting MoreAvailable (review 2026-10-21 06/PERF-1, mirroring + /// SiteCallAuditActor's 50-page bound). Bounds a misbehaving site's per-tick cost + /// while letting a busy-but-healthy site drain inside one tick, so the stalled signal + /// only fires on a genuine non-drain. + /// + public int MaxPagesPerTick { get; set; } = 50; + /// /// Number of consecutive non-draining cycles (events returned AND /// MoreAvailable=true) that must accumulate for a site before the actor diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationOptionsValidator.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationOptionsValidator.cs index 0d7a8a11..3b29638a 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationOptionsValidator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationOptionsValidator.cs @@ -24,6 +24,10 @@ public sealed class SiteAuditReconciliationOptionsValidator $"AuditLog:Reconciliation:{nameof(SiteAuditReconciliationOptions.BatchSize)} " + $"({options.BatchSize}) must be 1..2000."); + builder.RequireThat(options.MaxPagesPerTick is > 0 and <= 1000, + $"AuditLog:Reconciliation:{nameof(SiteAuditReconciliationOptions.MaxPagesPerTick)} " + + $"({options.MaxPagesPerTick}) must be 1..1000."); + builder.RequireThat(options.StalledAfterNonDrainingCycles > 0, $"AuditLog:Reconciliation:{nameof(SiteAuditReconciliationOptions.StalledAfterNonDrainingCycles)} " + $"({options.StalledAfterNonDrainingCycles}) must be > 0."); diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/ISiteAuditPushLifecycle.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/ISiteAuditPushLifecycle.cs new file mode 100644 index 00000000..c418ba21 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/ISiteAuditPushLifecycle.cs @@ -0,0 +1,66 @@ +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; + +namespace ZB.MOM.WW.ScadaBridge.AuditLog.Site; + +/// +/// Site-only forwarding states that extend inside the site +/// SQLite sidecar (audit_forward_state.ForwardState is TEXT, so these need no schema +/// change). Both mean "left the telemetry PUSH queue without a central ack, still owed to +/// central through the reconciliation PULL": ReadPendingSinceAsync serves them, +/// MarkReconciledUpToAsync retires them, and the retention purge never drops them +/// (review 2026-10-21 06/PERF-1, 06/IMPL-1). +/// +/// +/// They are deliberately NOT members of the Commons enum: they +/// never leave the site store (the wire and the central table carry no forward state), and a +/// downgraded node simply ignores rows in a state it does not know. +/// +public static class SiteAuditForwardStates +{ + /// + /// A cached-lifecycle row whose operational (SiteCalls) half can never be built — its + /// tracking snapshot is gone — so the combined-telemetry drain gave up pushing it. Its audit + /// half still reaches central through the reconciliation pull. + /// + public const string PushAbandoned = "PushAbandoned"; + + /// + /// A row central answered but refused (not in AcceptedEventIds) on + /// + /// consecutive drains. Moved out of the push queue so one poison row cannot wedge every push + /// batch behind it; the pull then delivers it, or central's permanent-abandon path records the + /// loss as a ReconciliationAbandoned row and retires it. + /// + public const string Rejected = "Rejected"; +} + +/// +/// Push-side lifecycle transitions the SiteAuditTelemetryActor needs beyond the Commons +/// surface. +/// Implemented by ; the actor discovers it by type-testing the +/// injected queue, so composition roots and test doubles that only implement +/// ISiteAuditQueue keep working (the actor falls back to the pre-review behaviour). +/// +public interface ISiteAuditPushLifecycle +{ + /// + /// Flips the supplied EventIds from to + /// . Rows in any other state are untouched. + /// + /// Event IDs whose push is abandoned. + /// Cancellation token. + /// A task that represents the asynchronous operation. + Task MarkPushAbandonedAsync(IReadOnlyList eventIds, CancellationToken ct = default); + + /// + /// Records one refused push attempt for each supplied + /// row (bumps AttemptCount, stamps LastAttemptUtc) and flips every row that has + /// now reached to . + /// + /// Event IDs central answered for but did not accept. + /// Refused attempts after which a row leaves the push queue. + /// Cancellation token. + /// A task that resolves to the number of rows flipped to Rejected by this call. + Task RecordRejectedPushAttemptsAsync( + IReadOnlyList eventIds, int maxAttempts, CancellationToken ct = default); +} diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs index e1ad577f..d446483c 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs @@ -43,7 +43,7 @@ namespace ZB.MOM.WW.ScadaBridge.AuditLog.Site; /// stamped by central on ingest, not a site column. /// /// -public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable, IDisposable +public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, ISiteAuditPushLifecycle, 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) @@ -78,6 +78,7 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable private readonly SqliteAuditWriterOptions _options; private readonly ILogger _logger; private readonly INodeIdentityProvider _nodeIdentity; + private readonly IAuditWriteFailureCounter? _failureCounter; private readonly object _writeLock = new(); private readonly Channel _writeQueue; private readonly Task _writerLoop; @@ -88,11 +89,17 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable /// Logger instance. /// Node identity provider. /// Optional connection string override. + /// + /// Optional SiteAuditWriteFailures sink, bumped once per row the push gives up on as + /// (review 2026-10-21 06/IMPL-1). Null on + /// composition roots without Site Health Monitoring. + /// public SqliteAuditWriter( IOptions options, ILogger logger, INodeIdentityProvider nodeIdentity, - string? connectionStringOverride = null) + string? connectionStringOverride = null, + IAuditWriteFailureCounter? failureCounter = null) { ArgumentNullException.ThrowIfNull(options); ArgumentNullException.ThrowIfNull(logger); @@ -101,6 +108,7 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable _options = options.Value; _logger = logger; _nodeIdentity = nodeIdentity; + _failureCounter = failureCounter; var connectionString = connectionStringOverride ?? $"Data Source={_options.DatabasePath};Cache=Shared"; @@ -801,9 +809,11 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable } /// - /// Returns up to pending or forwarded audit events - /// with >= , oldest first. - /// Used by the reconciliation-pull handler. + /// Returns up to audit events central has not yet acknowledged + /// (, , + /// ) with + /// >= , oldest first. rows + /// are NOT served (review 2026-10-21 06/PERF-1). Used by the reconciliation-pull handler. /// /// Lower bound timestamp (UTC) for event occurrence. /// Maximum number of rows to return. @@ -827,9 +837,14 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable // 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). + // + // ONLY rows central has NOT acknowledged are served: Pending, plus the + // push-exit states PushAbandoned / Rejected (review 2026-10-21 06/PERF-1). + // Forwarded rows used to be served too, which made the pull re-ship every + // row the push had already delivered — central re-inserted each one, and the + // pull's 256-rows-per-tick budget was spent on acked rows, so any site above + // ~0.85 audit rows/s fell permanently behind and latched the stalled signal. + // A Forwarded row is acked by central; the pull has nothing to repair there. lock (_readLock) { ObjectDisposedException.ThrowIf(_disposed, this); @@ -851,13 +866,12 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable fs.rowid FROM audit_event ae JOIN audit_forward_state fs ON fs.EventId = ae.EventId - WHERE fs.ForwardState IN ($pending, $forwarded) + WHERE fs.ForwardState IN ($pending, $pushAbandoned, $rejected) 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()); + AddOwedStateParameters(cmd); // 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. @@ -874,6 +888,165 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable } } + /// + /// Binds the three "owed to central" states — $pending, $pushAbandoned, + /// $rejected — that the reconciliation pull serves and retires. + /// + private static void AddOwedStateParameters(SqliteCommand cmd) + { + cmd.Parameters.AddWithValue("$pending", AuditForwardState.Pending.ToString()); + cmd.Parameters.AddWithValue("$pushAbandoned", SiteAuditForwardStates.PushAbandoned); + cmd.Parameters.AddWithValue("$rejected", SiteAuditForwardStates.Rejected); + } + + /// + public Task MarkPushAbandonedAsync(IReadOnlyList eventIds, CancellationToken ct = default) + { + ArgumentNullException.ThrowIfNull(eventIds); + if (eventIds.Count == 0) + { + return Task.CompletedTask; + } + + lock (_writeLock) + { + ObjectDisposedException.ThrowIf(_disposed, this); + + // Pending → PushAbandoned only. Unlike the old "mark Forwarded" this is honest: + // central never acked these rows, so they stay owed to the reconciliation pull + // (served, retired by a covering cursor, never age-purged) — review 2026-10-21 + // 06/PERF-1. Attempt bookkeeping is bumped like any drain pass that touched them. + var now = DateTime.UtcNow.ToString("o", System.Globalization.CultureInfo.InvariantCulture); + 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 = $abandoned, ") + .Append("AttemptCount = AttemptCount + 1, LastAttemptUtc = $now ") + .Append("WHERE ForwardState = $pending AND EventId IN ("); + AppendIdParameters(cmd, sb, eventIds, start, chunk); + sb.Append(");"); + cmd.CommandText = sb.ToString(); + cmd.Parameters.AddWithValue("$abandoned", SiteAuditForwardStates.PushAbandoned); + cmd.Parameters.AddWithValue("$pending", AuditForwardState.Pending.ToString()); + cmd.Parameters.AddWithValue("$now", now); + cmd.ExecuteNonQuery(); + } + transaction.Commit(); + return Task.CompletedTask; + } + } + + /// + public Task RecordRejectedPushAttemptsAsync( + IReadOnlyList eventIds, int maxAttempts, CancellationToken ct = default) + { + ArgumentNullException.ThrowIfNull(eventIds); + if (eventIds.Count == 0) + { + return Task.FromResult(0); + } + + var threshold = Math.Max(1, maxAttempts); + lock (_writeLock) + { + ObjectDisposedException.ThrowIf(_disposed, this); + + // Review 2026-10-21 06/IMPL-1: AttemptCount/LastAttemptUtc used to move only on + // the Pending→Forwarded flip, so a row central kept refusing had no history and + // no give-up — and since the push reads oldest-first, it sat at the head of every + // batch forever. Bump on every refused attempt, then move rows that reached the + // threshold out of the push queue into Rejected (still owed to the pull). + var now = DateTime.UtcNow.ToString("o", System.Globalization.CultureInfo.InvariantCulture); + var flipped = 0; + using var transaction = _connection.BeginTransaction(); + for (int start = 0; start < eventIds.Count; start += MarkForwardStateChunkSize) + { + var chunk = Math.Min(MarkForwardStateChunkSize, eventIds.Count - start); + + using (var bump = _connection.CreateCommand()) + { + bump.Transaction = transaction; + var sb = new System.Text.StringBuilder(); + sb.Append("UPDATE audit_forward_state SET AttemptCount = AttemptCount + 1, LastAttemptUtc = $now ") + .Append("WHERE ForwardState = $pending AND EventId IN ("); + AppendIdParameters(bump, sb, eventIds, start, chunk); + sb.Append(");"); + bump.CommandText = sb.ToString(); + bump.Parameters.AddWithValue("$pending", AuditForwardState.Pending.ToString()); + bump.Parameters.AddWithValue("$now", now); + bump.ExecuteNonQuery(); + } + + using (var reject = _connection.CreateCommand()) + { + reject.Transaction = transaction; + var sb = new System.Text.StringBuilder(); + sb.Append("UPDATE audit_forward_state SET ForwardState = $rejected ") + .Append("WHERE ForwardState = $pending AND AttemptCount >= $max AND EventId IN ("); + AppendIdParameters(reject, sb, eventIds, start, chunk); + sb.Append(");"); + reject.CommandText = sb.ToString(); + reject.Parameters.AddWithValue("$pending", AuditForwardState.Pending.ToString()); + reject.Parameters.AddWithValue("$rejected", SiteAuditForwardStates.Rejected); + reject.Parameters.AddWithValue("$max", threshold); + flipped += reject.ExecuteNonQuery(); + } + } + transaction.Commit(); + + if (flipped > 0) + { + _logger.LogWarning( + "{Count} site audit row(s) were refused by central on {Max} consecutive pushes and moved out " + + "of the push queue as Rejected; the reconciliation pull still serves them (central records " + + "a ReconciliationAbandoned row if they can never be inserted).", + flipped, threshold); + for (var i = 0; i < flipped; i++) + { + try { _failureCounter?.Increment(); } + catch { /* counter must never throw — defence in depth */ } + } + } + + return Task.FromResult(flipped); + } + } + + private static void AppendIdParameters( + SqliteCommand cmd, System.Text.StringBuilder sb, IReadOnlyList eventIds, int start, int chunk) + { + 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()); + } + } + + /// + /// Test/diagnostic read of one row's sidecar state: (ForwardState, AttemptCount), or + /// null when the row does not exist. + /// + /// The event to look up. + /// The row's forward state and attempt count, or null. + internal (string State, int AttemptCount)? GetForwardStateForTests(Guid eventId) + { + lock (_readLock) + { + ObjectDisposedException.ThrowIf(_disposed, this); + using var cmd = _readConnection.CreateCommand(); + cmd.CommandText = "SELECT ForwardState, AttemptCount FROM audit_forward_state WHERE EventId = $id;"; + cmd.Parameters.AddWithValue("$id", eventId.ToString()); + using var reader = cmd.ExecuteReader(); + return reader.Read() ? (reader.GetString(0), reader.GetInt32(1)) : null; + } + } + /// Flow Debugger PullTraceEvents: hard cap on ids per read (design §5.4). public const int MaxExecutionIdsPerRead = 64; @@ -1042,12 +1215,15 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable // // 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. + // that point are out of the flip's reach no matter where the cursor sits. + // + // Only the OWED states flip (Pending / PushAbandoned / Rejected — exactly what + // ReadPendingSinceAsync serves). Forwarded rows are no longer touched: central + // acked them over the push path, they are already purge-eligible, and flipping + // them was one UPDATE per pushed row for no benefit (review 2026-10-21 06/PERF-1). // // 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 + // a batch, so the first pull retires nothing 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 @@ -1056,13 +1232,12 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable 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) + WHERE fs.ForwardState IN ($pending, $pushAbandoned, $rejected) + AND 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()); + AddOwedStateParameters(cmd); cmd.Parameters.AddWithValue("$maxServedRowId", Volatile.Read(ref _maxServedRowId)); cmd.Parameters.AddWithValue("$since", EnsureUtc(sinceUtc).ToString( "o", System.Globalization.CultureInfo.InvariantCulture)); @@ -1088,8 +1263,8 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable { ObjectDisposedException.ThrowIf(_disposed, this); - // Flip the sidecar from Pending/Forwarded → Reconciled. Rows - // already Reconciled are left untouched (idempotent re-call), and the + // Flip the sidecar from Pending/Forwarded (or a push-exit state) → 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(); @@ -1105,7 +1280,7 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable 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 ("); + .Append("WHERE ForwardState IN ($pending, $forwarded, $pushAbandoned, $rejected) AND EventId IN ("); for (int i = 0; i < chunk; i++) { if (i > 0) sb.Append(','); @@ -1118,6 +1293,8 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable cmd.Parameters.AddWithValue("$reconciled", reconciled); cmd.Parameters.AddWithValue("$pending", pending); cmd.Parameters.AddWithValue("$forwarded", forwarded); + cmd.Parameters.AddWithValue("$pushAbandoned", SiteAuditForwardStates.PushAbandoned); + cmd.Parameters.AddWithValue("$rejected", SiteAuditForwardStates.Rejected); cmd.ExecuteNonQuery(); } transaction.Commit(); @@ -1145,12 +1322,15 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable ObjectDisposedException.ThrowIf(_disposed, this); using var cmd = _readConnection.CreateCommand(); + // "Pending" on the health report means OWED to central: the push-exit states + // (PushAbandoned / Rejected) still await the reconciliation pull and are never + // purged, so they count toward the backlog and its oldest-row age. cmd.CommandText = """ SELECT COUNT(*), MIN(OccurredAtUtc) FROM audit_forward_state - WHERE ForwardState = $pending; + WHERE ForwardState IN ($pending, $pushAbandoned, $rejected); """; - cmd.Parameters.AddWithValue("$pending", AuditForwardState.Pending.ToString()); + AddOwedStateParameters(cmd); using var reader = cmd.ExecuteReader(); reader.Read(); diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryActor.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryActor.cs index 31a2c852..d5d729d4 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryActor.cs @@ -179,6 +179,7 @@ public class SiteAuditTelemetryActor : ReceiveActor } var batch = BuildBatch(pending); + var sentIds = batch.Events.Select(e => e.EventId).ToList(); IngestAck ack; try @@ -203,6 +204,8 @@ public class SiteAuditTelemetryActor : ReceiveActor await _queue.MarkForwardedAsync(acceptedIds, ct) .ConfigureAwait(false); } + + await RecordRefusalsAsync(sentIds, ack, ct).ConfigureAwait(false); } catch (OperationCanceledException) when (ct.IsCancellationRequested) { @@ -269,7 +272,7 @@ public class SiteAuditTelemetryActor : ReceiveActor var wireBytes = AuditWireBudget.InitialBytes; var emittedEventIds = new List(pending.Count); // Rows whose operational half can never be built. They are marked - // Forwarded so they LEAVE this queue — see AbandonUnresolvableAsync + // PushAbandoned so they LEAVE this queue — see AbandonUnresolvableAsync // for why that is safe and why leaving them Pending is not. var abandonedEventIds = new List(); // Counted, not logged per row: a whole batch can fail identically @@ -362,7 +365,7 @@ public class SiteAuditTelemetryActor : ReceiveActor if (batch.Packets.Count == 0) { // Nothing resolvable in this read. Any permanently-unresolvable - // rows have just been marked Forwarded above, so the next drain + // rows have just been marked PushAbandoned above, so the next drain // sees past them rather than re-reading the same head of queue. return; } @@ -387,6 +390,9 @@ public class SiteAuditTelemetryActor : ReceiveActor await _queue.MarkForwardedAsync(acceptedIds, ct) .ConfigureAwait(false); } + + await RecordRefusalsAsync( + batch.Packets.Select(p => p.AuditEvent.EventId).ToList(), ack, ct).ConfigureAwait(false); } catch (OperationCanceledException) when (ct.IsCancellationRequested) { @@ -408,33 +414,35 @@ public class SiteAuditTelemetryActor : ReceiveActor } /// - /// Marks cached rows whose operational half can never be built as - /// Forwarded, so they leave the cached-drain queue. + /// Moves cached rows whose operational half can never be built out of the + /// cached-drain queue, as . /// /// /// - /// Why they must leave. Leaving an unresolvable row Pending — the - /// previous behaviour — is not a harmless skip. The queue is read - /// oldest-first with a fixed BatchSize, so once a batch's worth of - /// unresolvable rows collects at the head, every drain re-reads exactly - /// those rows, fails identically, and never sees the newer rows behind - /// them. The cached-telemetry path stalls permanently, and each pass logs - /// once per row (measured at ~2 800 warnings/minute on a rig). + /// Why they must leave. Leaving an unresolvable row Pending is not a + /// harmless skip. The queue is read oldest-first with a fixed + /// BatchSize, so once a batch's worth of unresolvable rows collects at + /// the head, every drain re-reads exactly those rows, fails identically, and + /// never sees the newer rows behind them. The cached-telemetry path stalls + /// permanently, and each pass logs once per row (measured at ~2 800 + /// warnings/minute on a rig). /// /// - /// Why Forwarded is the honest state. Its role in this state machine - /// is "no longer owed by the drain, still eligible for reconciliation" — - /// which is exactly the situation here. The audit half is NOT dropped: - /// ReadPendingSinceAsync covers Forwarded rows as well as Pending - /// ones and central dedups on EventId, so the reconciliation pull still - /// delivers them. What is genuinely lost is the operational - /// (SiteCalls) half — and that is unrecoverable regardless, because - /// the tracking row it would have been built from no longer exists. + /// Why PushAbandoned, not Forwarded. These rows were marked Forwarded + /// until review 2026-10-21 (06/PERF-1), which only worked because the + /// reconciliation pull used to re-serve every Forwarded row — the coupling + /// that capped the pull's throughput. Forwarded now means "central acked it" + /// and the pull no longer serves it, so an un-acked row needs a state of its + /// own: PushAbandoned is served by the pull, retired by a covering + /// cursor and never age-purged. The audit half therefore still reaches central; + /// what is genuinely lost is the operational (SiteCalls) half, which is + /// unrecoverable regardless because the tracking row no longer exists. /// /// - /// A failure to mark is swallowed: the rows simply stay Pending and the - /// next drain retries. Escalating here would take down the audit drain over - /// a best-effort cleanup. + /// A queue that does not implement (a + /// test double) falls back to the old Forwarded flip. A failure to mark is + /// swallowed: the rows simply stay Pending and the next drain retries. + /// Escalating here would take down the audit drain over a best-effort cleanup. /// /// private async Task AbandonUnresolvableAsync( @@ -442,24 +450,87 @@ public class SiteAuditTelemetryActor : ReceiveActor { try { - await _queue.MarkForwardedAsync(eventIds, ct).ConfigureAwait(false); + if (_queue is ISiteAuditPushLifecycle lifecycle) + { + await lifecycle.MarkPushAbandonedAsync(eventIds, ct).ConfigureAwait(false); + } + else + { + await _queue.MarkForwardedAsync(eventIds, ct).ConfigureAwait(false); + } + _logger.LogWarning( "Cached-telemetry drain: abandoned the operational half of {Count} row(s) with no " + - "resolvable tracking snapshot after {Grace}s; marked Forwarded so they no longer block " + + "resolvable tracking snapshot after {Grace}s; marked PushAbandoned so they no longer block " + "the queue. The audit half still reaches central via the reconciliation pull. This is " + "expected after tracking retention elapses or the tracking store is reset independently " + "of the audit store.", eventIds.Count, graceSeconds); } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + throw; + } catch (Exception ex) { _logger.LogWarning(ex, - "Cached-telemetry drain: could not mark {Count} unresolvable row(s) Forwarded " + + "Cached-telemetry drain: could not mark {Count} unresolvable row(s) PushAbandoned " + "(sqlite {SqliteError}); they stay Pending and will be retried.", eventIds.Count, SqliteErrorCodes.Describe(ex)); } } + /// + /// Records a refused push attempt for every row that was SENT in an answered batch but is + /// missing from the ack's AcceptedEventIds; rows refused + /// times leave the push + /// queue as (review 2026-10-21 06/IMPL-1). + /// + /// + /// Called only after the push RETURNED — a transport fault is not a verdict about any row + /// and never counts. Best-effort: a bookkeeping failure leaves the rows Pending, exactly as + /// before this existed. + /// + private async Task RecordRefusalsAsync(IReadOnlyList sentIds, IngestAck ack, CancellationToken ct) + { + if (_queue is not ISiteAuditPushLifecycle lifecycle || sentIds.Count == 0) + { + return; + } + + var accepted = new HashSet(ack.AcceptedEventIds, StringComparer.OrdinalIgnoreCase); + var refused = new List(); + foreach (var raw in sentIds) + { + if (!accepted.Contains(raw) && Guid.TryParse(raw, out var id)) + { + refused.Add(id); + } + } + + if (refused.Count == 0) + { + return; + } + + try + { + await lifecycle + .RecordRejectedPushAttemptsAsync(refused, Math.Max(1, _options.MaxRejectedPushAttempts), ct) + .ConfigureAwait(false); + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + throw; + } + catch (Exception ex) + { + _logger.LogWarning(ex, + "Could not record {Count} refused push attempt(s) (sqlite {SqliteError}); the rows stay Pending.", + refused.Count, SqliteErrorCodes.Describe(ex)); + } + } + private static AuditEventBatch BuildBatch(IReadOnlyList events) { var batch = new AuditEventBatch(); diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryOptions.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryOptions.cs index 72318e2d..ae6e95c4 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryOptions.cs @@ -29,7 +29,7 @@ public sealed class SiteAuditTelemetryOptions /// /// How long a cached-telemetry audit row may go without a resolvable /// OperationTracking snapshot before the drain ABANDONS its - /// operational half and marks the row Forwarded. Default: 300 s. + /// operational half and marks the row PushAbandoned. Default: 300 s. /// /// /// @@ -45,8 +45,9 @@ public sealed class SiteAuditTelemetryOptions /// /// Abandoning costs only the operational (SiteCalls) half, which is /// unrecoverable anyway once the tracking row is gone. The audit half is - /// NOT lost: the reconciliation pull covers Forwarded rows as well as - /// Pending ones, and central dedups on EventId. + /// NOT lost: the reconciliation pull serves PushAbandoned rows (central + /// never acked them) exactly as it serves Pending ones, and central dedups on + /// EventId (review 2026-10-21 06/PERF-1). /// /// /// Set to 0 to abandon on the first failed lookup. Negative values are @@ -54,4 +55,20 @@ public sealed class SiteAuditTelemetryOptions /// /// public int CachedTrackingGraceSeconds { get; set; } = 300; + + /// + /// Number of drains in which central ANSWERED a push but refused a row (the row is + /// missing from AcceptedEventIds) after which the row leaves the push queue as + /// Rejected. Default: 10 (≈ 50 s at the busy cadence). + /// + /// + /// The push reads oldest-first, so a row central can never insert (e.g. an over-width + /// column) used to sit at the head of every batch forever, failing each set-based insert + /// at central and dragging the batch into the per-row fallback (review 2026-10-21 + /// 06/IMPL-1). A transport fault (no answer) never counts. Rejected is NOT a drop: + /// the reconciliation pull still serves the row, so a transient central refusal still + /// lands, and a genuinely poison row is recorded centrally as ReconciliationAbandoned + /// before it retires. Each flip bumps SiteAuditWriteFailures on the site health report. + /// + public int MaxRejectedPushAttempts { get; set; } = 10; } diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryOptionsValidator.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryOptionsValidator.cs index 30ebee9d..e5daa223 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryOptionsValidator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryOptionsValidator.cs @@ -27,6 +27,10 @@ public sealed class SiteAuditTelemetryOptionsValidator : OptionsValidatorBase 0."); + builder.RequireThat(options.MaxRejectedPushAttempts > 0, + $"AuditLog:SiteTelemetry:{nameof(SiteAuditTelemetryOptions.MaxRejectedPushAttempts)} " + + $"({options.MaxRejectedPushAttempts}) must be > 0."); + builder.RequireThat(options.IdleIntervalSeconds >= options.BusyIntervalSeconds, $"AuditLog:SiteTelemetry:{nameof(SiteAuditTelemetryOptions.IdleIntervalSeconds)} " + $"({options.IdleIntervalSeconds}) must be >= " + diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/SiteAuditReconciliationActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/SiteAuditReconciliationActorTests.cs index deb2a259..5e673b91 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/SiteAuditReconciliationActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/SiteAuditReconciliationActorTests.cs @@ -51,9 +51,11 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture + int stalledAfter = 2, + int maxPagesPerTick = 50) => new() { + MaxPagesPerTick = maxPagesPerTick, // 100 ms tick keeps each test under a second. AwaitAssert covers // schedule jitter so a 100 ms tick has up to ~3 s to fire. ReconciliationIntervalSeconds = 300, @@ -79,6 +81,25 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture _seen = new(); /// Gets the number of times InsertIfNotExistsAsync was called. public int InsertCallCount { get; private set; } + /// Gets the number of set-based InsertManyIfNotExistsAsync calls. + public int BatchCallCount { get; private set; } + + /// + public Task InsertManyIfNotExistsAsync( + IReadOnlyList events, TimeSpan? commandTimeout = null, CancellationToken ct = default) + { + BatchCallCount++; + var inserted = 0; + foreach (var evt in events) + { + if (_seen.Add(evt.EventId)) + { + Inserted.Add(evt); + inserted++; + } + } + return Task.FromResult(inserted); + } /// public Task InsertIfNotExistsAsync(AuditEvent evt, CancellationToken ct = default) @@ -480,9 +501,12 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixtureVerifies that pulled events are ingested through the repository insert-if-not-exists call. + /// + /// Review 2026-10-21 06/PERF-1: a pulled page is ingested with ONE set-based + /// InsertManyIfNotExistsAsync call, not one round trip per row. + /// [Fact] - public void Tick_IngestEvents_ViaInsertIfNotExistsAsync() + public void Tick_IngestEvents_ViaOneSetBasedInsert() { var sites = new StaticEnumerator(new SiteEntry("siteA", "http://siteA:8083")); var e1 = NewEvent("siteA"); @@ -493,16 +517,50 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture Assert.Equal(2, repo.InsertCallCount), + AwaitAssert(() => Assert.Equal(2, repo.Inserted.Count), duration: TimeSpan.FromSeconds(3), interval: TimeSpan.FromMilliseconds(50)); StopReconciliationActor(actor); + Assert.Equal(1, repo.BatchCallCount); + Assert.Equal(0, repo.InsertCallCount); Assert.Contains(repo.Inserted, e => e.EventId == e1.EventId); Assert.Contains(repo.Inserted, e => e.EventId == e2.EventId); } + /// + /// Review 2026-10-21 06/PERF-1: a busy but healthy site whose backlog spans several pages + /// is drained by paging WITHIN one tick, so the stalled latch does not fire. + /// + [Fact] + public void BusyHealthySite_PagesWithinOneTick_AndDoesNotLatchStalled() + { + var sites = new StaticEnumerator(new SiteEntry("siteA", "http://siteA:8083")); + var t = new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc); + var pages = Enumerable.Range(0, 3) + .Select(p => Enumerable.Range(0, 3).Select(i => NewEvent("siteA", t.AddSeconds(p * 10 + i))).ToArray()) + .ToArray(); + var client = new ScriptedPullClient().Script("siteA", + new PullAuditEventsResponse(pages[0], MoreAvailable: true), + new PullAuditEventsResponse(pages[1], MoreAvailable: true), + new PullAuditEventsResponse(pages[2], MoreAvailable: false)); + var repo = new RecordingRepo(); + var (probe, _) = SubscribeStalled(); + + var actor = CreateActor(sites, client, repo, FastTickOptions(stalledAfter: 1)); + + AwaitAssert(() => Assert.Equal(9, repo.Inserted.Count), + duration: TimeSpan.FromSeconds(3), + interval: TimeSpan.FromMilliseconds(50)); + probe.ExpectNoMsg(TimeSpan.FromMilliseconds(500)); + StopReconciliationActor(actor); + + // The three pages were issued back to back with an advancing cursor. + Assert.Equal(pages[0][^1].EventId.ToString(), client.Calls[1].AfterId); + Assert.Equal(pages[1][^1].EventId.ToString(), client.Calls[2].AfterId); + } + // --------------------------------------------------------------------- // 4. Tick_Duplicates_NotDoubleInserted (real MSSQL idempotency) // --------------------------------------------------------------------- @@ -774,7 +832,9 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture +/// Forward-state lifecycle regressions for review 2026-10-21 06/PERF-1 and 06/IMPL-1: the +/// reconciliation pull serves only rows central has NOT acknowledged (never Forwarded), the +/// push-exit states and +/// stay owed to the pull until a covering cursor +/// retires them, and a row central keeps refusing leaves the push queue after N attempts. +/// +public class SqliteAuditWriterForwardStateTests : IDisposable +{ + private readonly string _dbPath = Path.Combine(Path.GetTempPath(), $"audit-fwdstate-{Guid.NewGuid():N}.db"); + private readonly CountingFailureCounter _counter = new(); + + /// Deletes the temporary database. + public void Dispose() + { + foreach (var suffix in new[] { "", "-wal", "-shm" }) + { + try { File.Delete(_dbPath + suffix); } catch { /* best-effort */ } + } + } + + private SqliteAuditWriter CreateWriter() => new( + Options.Create(new SqliteAuditWriterOptions { DatabasePath = _dbPath }), + NullLogger.Instance, + new FakeNodeIdentityProvider(), + failureCounter: _counter); + + private static async Task WriteAsync(SqliteAuditWriter writer, DateTime occurredAtUtc) + { + var id = Guid.NewGuid(); + await writer.WriteAsync(ScadaBridgeAuditEventFactory.Create( + eventId: id, + occurredAtUtc: occurredAtUtc, + channel: AuditChannel.ApiOutbound, + kind: AuditKind.ApiCall, + status: AuditStatus.Delivered)); + return id; + } + + /// The pull no longer re-ships rows the push already delivered. + /// A task that represents the asynchronous operation. + [Fact] + public async Task ReadPendingSince_ExcludesForwardedRows() + { + await using var writer = CreateWriter(); + var t = DateTime.UtcNow.AddMinutes(-5); + var forwarded = await WriteAsync(writer, t); + var pending = await WriteAsync(writer, t.AddSeconds(1)); + await writer.MarkForwardedAsync(new[] { forwarded }); + + var served = await writer.ReadPendingSinceAsync(DateTime.MinValue, 100); + + Assert.Equal(pending, Assert.Single(served).EventId); + } + + /// A covering cursor retires only owed rows and leaves Forwarded rows alone. + /// A task that represents the asynchronous operation. + [Fact] + public async Task MarkReconciledUpTo_DoesNotTouchForwardedRows() + { + await using var writer = CreateWriter(); + var t = DateTime.UtcNow.AddMinutes(-5); + var forwarded = await WriteAsync(writer, t); + var pending = await WriteAsync(writer, t.AddSeconds(1)); + await writer.MarkForwardedAsync(new[] { forwarded }); + var served = await writer.ReadPendingSinceAsync(DateTime.MinValue, 100); + + var flipped = await writer.MarkReconciledUpToAsync( + served[^1].OccurredAtUtc.UtcDateTime, served[^1].EventId.ToString()); + + Assert.Equal(1, flipped); + Assert.Equal("Forwarded", writer.GetForwardStateForTests(forwarded)!.Value.State); + Assert.Equal("Reconciled", writer.GetForwardStateForTests(pending)!.Value.State); + } + + /// A push-abandoned row leaves the push queue, is served by the pull, then retires. + /// A task that represents the asynchronous operation. + [Fact] + public async Task PushAbandoned_IsServedByThePull_ThenMarkedReconciled_AndNeverPurgedBeforeThat() + { + await using var writer = CreateWriter(); + var old = DateTime.UtcNow.AddDays(-10); + var abandoned = await WriteAsync(writer, old); + + await writer.MarkPushAbandonedAsync(new[] { abandoned }); + + Assert.Empty(await writer.ReadPendingAsync(100)); + // Still owed: never purged on age, counted on the backlog. + Assert.Equal(0, await writer.PurgeExpiredAsync(DateTime.UtcNow)); + Assert.Equal(1, (await writer.GetBacklogStatsAsync()).PendingCount); + + var served = await writer.ReadPendingSinceAsync(DateTime.MinValue, 100); + Assert.Equal(abandoned, Assert.Single(served).EventId); + + await writer.MarkReconciledUpToAsync(old, abandoned.ToString()); + + Assert.Equal("Reconciled", writer.GetForwardStateForTests(abandoned)!.Value.State); + Assert.Equal(1, await writer.PurgeExpiredAsync(DateTime.UtcNow)); + } + + /// + /// A row central keeps refusing bumps its attempt history every time and, at the threshold, + /// leaves the push queue as Rejected — unwedging the push head — while staying owed to the pull. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task RefusedRow_BumpsAttempts_ThenFlipsToRejected_UnwedgingThePushHead() + { + await using var writer = CreateWriter(); + var t = DateTime.UtcNow.AddMinutes(-5); + var poison = await WriteAsync(writer, t); + var behind = await WriteAsync(writer, t.AddSeconds(1)); + + Assert.Equal(0, await writer.RecordRejectedPushAttemptsAsync(new[] { poison }, maxAttempts: 3)); + Assert.Equal(0, await writer.RecordRejectedPushAttemptsAsync(new[] { poison }, maxAttempts: 3)); + Assert.Equal(("Pending", 2), writer.GetForwardStateForTests(poison)!.Value); + Assert.Equal(poison, (await writer.ReadPendingAsync(1))[0].EventId); // still the head + + Assert.Equal(1, await writer.RecordRejectedPushAttemptsAsync(new[] { poison }, maxAttempts: 3)); + + Assert.Equal(("Rejected", 3), writer.GetForwardStateForTests(poison)!.Value); + Assert.Equal(behind, Assert.Single(await writer.ReadPendingAsync(1)).EventId); + Assert.Contains(await writer.ReadPendingSinceAsync(DateTime.MinValue, 100), e => e.EventId == poison); + Assert.Equal(0, await writer.PurgeExpiredAsync(DateTime.UtcNow.AddDays(1))); + Assert.Equal(1, _counter.Count); + } + + private sealed class CountingFailureCounter : IAuditWriteFailureCounter + { + private int _count; + public int Count => Volatile.Read(ref _count); + public void Increment() => Interlocked.Increment(ref _count); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterWriteTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterWriteTests.cs index 07aee1cc..632f8815 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterWriteTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterWriteTests.cs @@ -488,12 +488,16 @@ public class SqliteAuditWriterWriteTests // ----- Reconciliation pull surface ----- // - /// Verifies that the since-read returns pending and forwarded rows oldest first, limited to the requested count. + /// + /// Verifies that the since-read returns only rows central has NOT acknowledged, oldest first, + /// limited to the requested count — Forwarded rows are no longer re-shipped + /// (review 2026-10-21 06/PERF-1). + /// /// A task that represents the asynchronous operation. [Fact] - public async Task ReadPendingSinceAsync_Returns_PendingAndForwarded_OldestFirst_LimitedToN() + public async Task ReadPendingSinceAsync_Returns_OnlyUnackedRows_OldestFirst_LimitedToN() { - var (writer, dataSource) = CreateWriter(nameof(ReadPendingSinceAsync_Returns_PendingAndForwarded_OldestFirst_LimitedToN)); + var (writer, dataSource) = CreateWriter(nameof(ReadPendingSinceAsync_Returns_OnlyUnackedRows_OldestFirst_LimitedToN)); await using var _ = writer; var baseTime = new DateTime(2026, 5, 20, 12, 0, 0, DateTimeKind.Utc); @@ -504,11 +508,11 @@ public class SqliteAuditWriterWriteTests NewEvent(occurredAtUtc: baseTime.AddSeconds(3)), NewEvent(occurredAtUtc: baseTime.AddSeconds(2)), NewEvent(occurredAtUtc: baseTime.AddSeconds(4)), + NewEvent(occurredAtUtc: baseTime.AddSeconds(6)), }; foreach (var e in evts) await writer.WriteAsync(e); - // Flip half to Forwarded — they must still surface in the reconciliation pull - // because central hasn't confirmed they were ingested yet. + // Central acked these over the push: the pull has nothing to repair for them. await writer.MarkForwardedAsync(new[] { evts[0].EventId, evts[2].EventId }); var rows = await writer.ReadPendingSinceAsync(sinceUtc: DateTime.MinValue, batchSize: 3); @@ -516,7 +520,7 @@ public class SqliteAuditWriterWriteTests Assert.Equal(3, rows.Count); Assert.Equal(baseTime.AddSeconds(1), rows[0].OccurredAtUtc); Assert.Equal(baseTime.AddSeconds(2), rows[1].OccurredAtUtc); - Assert.Equal(baseTime.AddSeconds(3), rows[2].OccurredAtUtc); + Assert.Equal(baseTime.AddSeconds(4), rows[2].OccurredAtUtc); } /// Verifies that the since-read excludes rows older than the supplied cutoff. @@ -740,17 +744,17 @@ public class SqliteAuditWriterWriteTests Assert.Equal(lateStamped.EventId, Assert.Single(reread).EventId); } - /// Verifies that forwarded rows are retired by reconciling up to a cursor even if they were never pulled. + /// + /// Verifies that a reconciliation cursor leaves Forwarded rows alone: central acked them over + /// the push, they are already purge-eligible, and the pull never served them + /// (review 2026-10-21 06/PERF-1). + /// /// A task that represents the asynchronous operation. [Fact] - public async Task MarkReconciledUpToAsync_ForwardedRowsRetire_EvenWithoutHavingBeenPulled() + public async Task MarkReconciledUpToAsync_LeavesForwardedRowsAlone() { - // The bound applies to PENDING rows (nothing proves central saw them but the pull). - // A FORWARDED row was ACKED by central through the telemetry push path, so the cursor - // may retire it whether or not this node has ever served it in a pull — otherwise a - // site node that never serves a pull would accumulate acked rows forever. var (writer, dataSource) = CreateWriter( - nameof(MarkReconciledUpToAsync_ForwardedRowsRetire_EvenWithoutHavingBeenPulled)); + nameof(MarkReconciledUpToAsync_LeavesForwardedRowsAlone)); await using var _w = writer; var t0 = new DateTime(2026, 5, 20, 12, 0, 0, DateTimeKind.Utc); @@ -764,8 +768,8 @@ public class SqliteAuditWriterWriteTests var flipped = await writer.MarkReconciledUpToAsync(t0.AddSeconds(30), afterId: null); - Assert.Equal(1, flipped); - Assert.Equal(AuditForwardState.Reconciled.ToString(), ReadForwardState(dataSource, pushed.EventId)); + Assert.Equal(0, flipped); + Assert.Equal(AuditForwardState.Forwarded.ToString(), ReadForwardState(dataSource, pushed.EventId)); Assert.Equal(AuditForwardState.Pending.ToString(), ReadForwardState(dataSource, neverShipped.EventId)); } diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/Telemetry/SiteAuditTelemetryActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/Telemetry/SiteAuditTelemetryActorTests.cs index b7f33023..76aaf0e3 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/Telemetry/SiteAuditTelemetryActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/Telemetry/SiteAuditTelemetryActorTests.cs @@ -752,6 +752,76 @@ public class SiteAuditTelemetryActorTests : TestKit Assert.DoesNotContain(logger.Levels, l => l >= Microsoft.Extensions.Logging.LogLevel.Error); } + /// + /// Review 2026-10-21 06/PERF-1: against a queue that supports the push lifecycle, an orphan + /// cached row past grace is marked PushAbandoned (still owed to the pull), NOT Forwarded. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task CachedDrain_OrphanPastGrace_IsMarkedPushAbandoned_NotForwarded() + { + var queue = Substitute.For(); + var lifecycle = (ZB.MOM.WW.ScadaBridge.AuditLog.Site.ISiteAuditPushLifecycle)queue; + var orphan = NewCachedEvent(AuditKind.CachedSubmit); + queue.ReadPendingAsync(Arg.Any(), Arg.Any()) + .Returns(Task.FromResult>(Array.Empty())); + queue.ReadPendingCachedTelemetryAsync(Arg.Any(), Arg.Any()) + .Returns( + Task.FromResult>(new[] { orphan }), + Task.FromResult>(Array.Empty())); + _trackingStore.GetStatusAsync(Arg.Any(), Arg.Any()) + .Returns(Task.FromResult(null)); + + Sys.ActorOf(Props.Create(() => new SiteAuditTelemetryActor( + queue, _client, Opts(), NullLogger.Instance, + (IOperationTrackingStore?)_trackingStore))); + + await AwaitAssertAsync(async () => + await lifecycle.Received(1).MarkPushAbandonedAsync( + Arg.Is>(g => g.Count == 1 && g[0] == orphan.EventId), + Arg.Any()), + TimeSpan.FromSeconds(10)); + await queue.DidNotReceive().MarkForwardedAsync(Arg.Any>(), Arg.Any()); + } + + /// + /// Review 2026-10-21 06/IMPL-1: rows central ANSWERED for but did not accept get a refused + /// attempt recorded (so a poison head row eventually leaves the push queue); accepted rows + /// do not, and a transport fault records nothing. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task Drain_RefusedRows_RecordRejectedAttempts_TransportFaultDoesNot() + { + var queue = Substitute.For(); + var lifecycle = (ZB.MOM.WW.ScadaBridge.AuditLog.Site.ISiteAuditPushLifecycle)queue; + var accepted = NewEvent(); + var refused = NewEvent(); + queue.ReadPendingAsync(Arg.Any(), Arg.Any()) + .Returns( + Task.FromResult>(new[] { accepted, refused }), + Task.FromResult>(new[] { accepted, refused }), + Task.FromResult>(Array.Empty())); + _client.IngestAuditEventsAsync(Arg.Any(), Arg.Any()) + .Returns( + _ => Task.FromException(new InvalidOperationException("transport down")), + _ => Task.FromResult(AckAll(new[] { accepted }))); + + Sys.ActorOf(Props.Create(() => new SiteAuditTelemetryActor( + queue, _client, Opts(), NullLogger.Instance, + (IOperationTrackingStore?)null))); + + await AwaitAssertAsync(async () => + await lifecycle.Received(1).RecordRejectedPushAttemptsAsync( + Arg.Is>(g => g.Count == 1 && g[0] == refused.EventId), + 10, + Arg.Any()), + TimeSpan.FromSeconds(10)); + await queue.Received(1).MarkForwardedAsync( + Arg.Is>(g => g.Count == 1 && g[0] == accepted.EventId), + Arg.Any()); + } + private sealed class CapturingLogger : Microsoft.Extensions.Logging.ILogger { public System.Collections.Concurrent.ConcurrentQueue Levels { get; } = new();