diff --git a/docs/requirements/Component-AuditLog.md b/docs/requirements/Component-AuditLog.md index 18d6aef8..c08b9494 100644 --- a/docs/requirements/Component-AuditLog.md +++ b/docs/requirements/Component-AuditLog.md @@ -311,17 +311,27 @@ 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. 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 +answered for but did not accept stay `Pending` for the next sweep. Central +additionally lists, in the ack's additive `rejected_event_ids` (sitestream.proto +`IngestAck` field 2), the ids it refused for a **permanent, row-intrinsic** SQL +reason (string truncation 8152/2628, NOT NULL 515, conversion/overflow +245/241/242/8114/8115/220/232, constraint 547). Only those are refusals: the site +bumps `AttemptCount` / `LastAttemptUtc` for them and, after +`AuditLog:SiteTelemetry:MaxRejectedPushAttempts` (default 10), moves the row out of +the push queue as `Rejected`, counted on its own `scadabridge.site.audit.rejected_rows` +counter (not `SiteAuditWriteFailures` — the site's write succeeded). A row merely +absent from the accepted list — a SQL maintenance window, a timeout, an exhausted +ingest budget — is never counted, and an older central that does not send the field +never triggers it (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 +answer) is never counted as a refusal. Symmetrically, the reconciliation puller counts +only non-transient failures toward its permanent-abandon threshold: a timeout, +connection loss, deadlock or throttling error holds the cursor and retries without +ever abandoning the row. 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 @@ -392,9 +402,23 @@ 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). +**Rollback caveat.** Rolling a site back to a pre-PERF-1 binary strands any rows already in +`PushAbandoned` / `Rejected` on disk: the old code neither pushes, serves, retires nor purges +a state it does not know, so they sit in the audit file (harmless, but never delivered and +never purged) until the node is upgraded again — at which point the pull serves them +normally. Nothing else in the file changes shape (the `audit_site_state` table is ignored by +an old binary). + +**Oversize rows (accepted, documented).** A single site audit row larger than the 32 MB gRPC +ceiling (`GrpcMessageLimits.MaxMessageBytes`) cannot be pushed or pulled and would wedge the +head of both paths. It is unreachable under the payload caps (8 KB / 64 KB on error by +default) unless `DefaultCapBytes`/`ErrorCapBytes` are configured into the tens of megabytes; +rows above the 2 MB batch budget but under the ceiling already travel alone. + **Per-site state follows the site list (review 2026-10-21 06/MEM-2).** Each tick drops the cursor, non-draining counter and stalled latch of any site no longer returned by -`ISiteEnumerator` (a latched site publishes `Stalled=false` as it goes), and the +`ISiteEnumerator` (a latched site publishes `Stalled=false` as it goes; an EMPTY enumeration +prunes nothing, since it is as likely a transient lookup glitch), and the `PullAuditEvents` / `PullSiteCalls` / `PullTraceEvents` invokers evict and dispose any cached `(site, endpoint)` gRPC channel that is not a live site's primary or fallback address on every enumeration. The in-memory abandoned-`EventId` short-circuit set is @@ -476,16 +500,18 @@ carries a second, insertion-order bound: > 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 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 +The bound is persisted in a node-local `audit_site_state` key/value table in the audit +file (loaded at start-up, clamped to the live `MAX(rowid)`), so a site-node restart no longer +resets it — before that, a healthy site whose pull serves nothing could leave already-served +`PushAbandoned`/`Rejected` rows owed (and counted on the backlog) indefinitely. It stays +conservative in the safe direction (retirement can be 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. **Liveness note (accepted, documented).** A row that was served but never covered by a -later cursor — central reconciliation stopped for good, or the bound reset over a -restart before the next cursor arrived — stays `Pending` indefinitely, and the site -retention purge never purges `Pending`. That is the hard `ForwardState` invariant +later cursor — central reconciliation stopped for good — stays owed (`Pending` / +`PushAbandoned` / `Rejected`) indefinitely, and the site retention purge never purges +an owed row. That is the hard `ForwardState` invariant working as intended (an unacknowledged row is not droppable), but it means the site store's floor depends on reconciliation actually running, not merely on the retention window elapsing. It is observable rather than silent: `GetBacklogStatsAsync` reports diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditIngestErrorClassifier.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditIngestErrorClassifier.cs new file mode 100644 index 00000000..c01f2a4e --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditIngestErrorClassifier.cs @@ -0,0 +1,125 @@ +using System.Data.Common; +using System.Net.Sockets; +using Microsoft.Data.SqlClient; + +namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central; + +/// +/// Classifies a central audit-insert failure (review 2026-10-21 06/IMPL-1 follow-up) so that a +/// DATABASE problem — a maintenance window, a timeout, a dropped connection, an exhausted budget — +/// is never mistaken for a verdict about the ROW. Two questions, two predicates: +/// +/// — narrow: the row itself can never be inserted +/// (string truncation, NULL into NOT NULL, a conversion / overflow, a constraint). Only these ids +/// go back to the site as rejected_event_ids, which is what lets the site give a row up as +/// Rejected. +/// — broad: anything that may clear on its own. The reconciliation +/// puller never counts these toward permanently abandoning a row. +/// +/// The exception's whole chain is inspected, because EF Core +/// and the execution strategy wrap the provider exception. +/// +internal static class AuditIngestErrorClassifier +{ + /// + /// SQL Server errors that describe the ROW, not the server: 8152 / 2628 string or binary data + /// would be truncated, 515 NULL into a NOT NULL column, 245 / 241 / 242 / 8114 / 8115 / 220 / 232 + /// conversion and arithmetic overflow, 547 CHECK / FOREIGN KEY constraint. + /// + private static readonly HashSet PermanentRowErrorNumbers = new() + { + 8152, 2628, 515, 245, 241, 242, 8114, 8115, 220, 232, 547, + }; + + /// + /// SQL Server errors that may clear on their own (timeout, connection loss, deadlock, lock + /// timeout, resource / Azure throttling, database unavailable or in a restricted state). + /// + private static readonly HashSet TransientErrorNumbers = new() + { + -2, -1, 0, 2, 20, 53, 64, 121, 233, 1205, 1222, 3960, 4060, 4221, 8645, 8651, + 10053, 10054, 10060, 10928, 10929, 40143, 40197, 40501, 40540, 40613, + 49918, 49919, 49920, 942, 952, 983, 976, 978, 3906, 9001, 9002, + }; + + /// + /// True when the failure is a SQL Server error that says the row itself can never be inserted. + /// Everything else — including every transient error and every non-SQL exception — is false, + /// so the site keeps such rows Pending and simply retries them. + /// + /// The insert failure. + /// Whether the row should be reported back to the site as permanently rejected. + public static bool IsPermanentRowRejection(Exception ex) + { + if (IsTransient(ex)) + { + return false; + } + + for (var e = ex; e is not null; e = e.InnerException) + { + if (e is SqlException sql) + { + foreach (SqlError error in sql.Errors) + { + if (PermanentRowErrorNumbers.Contains(error.Number)) + { + return true; + } + } + return PermanentRowErrorNumbers.Contains(sql.Number); + } + } + return false; + } + + /// + /// True when any exception in the chain is a timeout, cancellation, socket / IO / connection + /// failure, an EF retry-limit exhaustion, a provider error flagged transient, or a SQL Server + /// error number in the transient set. + /// + /// The insert failure. + /// Whether the failure may clear on its own. + public static bool IsTransient(Exception ex) + { + for (var e = ex; e is not null; e = e.InnerException) + { + switch (e) + { + case TimeoutException: + case OperationCanceledException: + case SocketException: + case IOException: + case HttpRequestException: + return true; + case SqlException sql: + if (TransientErrorNumbers.Contains(sql.Number)) + { + return true; + } + foreach (SqlError error in sql.Errors) + { + if (TransientErrorNumbers.Contains(error.Number)) + { + return true; + } + } + if (sql.IsTransient) + { + return true; + } + break; + case DbException db when db.IsTransient: + return true; + } + + // EF Core's RetryLimitExceededException (an InvalidOperationException) wraps a + // transient provider error once the execution strategy gives up. + if (e.GetType().Name == "RetryLimitExceededException") + { + return true; + } + } + return false; + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogIngestActor.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogIngestActor.cs index 7be9bb4a..55e1404a 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogIngestActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/AuditLogIngestActor.cs @@ -152,6 +152,7 @@ public class AuditLogIngestActor : ReceiveActor var replyTo = Sender; var nowUtc = DateTime.UtcNow; var accepted = new List(cmd.Events.Count); + var rejected = new List(); // Resolve the repository for the whole batch — one DbContext per // message, mirroring NotificationOutboxActor. @@ -184,7 +185,7 @@ public class AuditLogIngestActor : ReceiveActor // unregistered (test composition roots) means the per-row catch // simply logs without surfacing on the health dashboard. var failureCounter = scope.ServiceProvider.GetService(); - await IngestWithRepositoryAsync(repository, redactor, failureCounter, cmd, nowUtc, accepted) + await IngestWithRepositoryAsync(repository, redactor, failureCounter, cmd, nowUtc, accepted, rejected) .ConfigureAwait(false); } catch (Exception ex) @@ -201,7 +202,7 @@ public class AuditLogIngestActor : ReceiveActor accepted.Count); } - replyTo.Tell(new IngestAuditEventsReply(accepted)); + replyTo.Tell(new IngestAuditEventsReply(accepted) { RejectedEventIds = rejected }); } private async Task IngestWithRepositoryAsync( @@ -210,7 +211,8 @@ public class AuditLogIngestActor : ReceiveActor ICentralAuditWriteFailureCounter? failureCounter, IngestAuditEventsCommand cmd, DateTime nowUtc, - List accepted) + List accepted, + List rejected) { // Stamp the ingest timestamp here, not at the site. Redact BEFORE the // IngestedAtUtc stamp so the redacted copy carries the central-side @@ -303,6 +305,15 @@ public class AuditLogIngestActor : ReceiveActor catch { /* counter must never throw — defence in depth */ } } + // Review 2026-10-21 06/IMPL-1: tell the site ONLY about rows the server + // refused for a reason intrinsic to the row (truncation, NOT NULL, conversion, + // constraint). A timeout / connection loss / maintenance window leaves the id + // in neither list, so the site retries it without counting a refusal. + if (AuditIngestErrorClassifier.IsPermanentRowRejection(rowEx)) + { + rejected.Add(cmd.Events[i].EventId); + } + _logger.LogError(rowEx, "Failed to persist audit event {EventId} during batch ingest; row will be retried by the site.", cmd.Events[i].EventId); @@ -336,6 +347,7 @@ public class AuditLogIngestActor : ReceiveActor { var replyTo = Sender; var accepted = new List(cmd.Entries.Count); + var rejected = new List(); try { @@ -370,7 +382,7 @@ public class AuditLogIngestActor : ReceiveActor strategy, dbContext, auditRepo, siteCallRepo, redactor, cmd, accepted, budget.Token) .ConfigureAwait(false)) { - replyTo.Tell(new IngestCachedTelemetryReply(accepted)); + replyTo.Tell(new IngestCachedTelemetryReply(accepted) { RejectedEventIds = rejected }); return; } @@ -445,6 +457,11 @@ public class AuditLogIngestActor : ReceiveActor // sustained dual-write failure surfaces on the dashboard. try { failureCounter?.Increment(); } catch { /* counter must never throw — defence in depth */ } + if (AuditIngestErrorClassifier.IsPermanentRowRejection(ex)) + { + rejected.Add(entry.Audit.EventId); + } + _logger.LogError( ex, "Combined telemetry dual-write failed for AuditEvent {EventId} / TrackedOperationId {TrackedOpId}; rolled back.", @@ -463,7 +480,7 @@ public class AuditLogIngestActor : ReceiveActor "Combined telemetry batch ingest failed before per-entry processing."); } - replyTo.Tell(new IngestCachedTelemetryReply(accepted)); + replyTo.Tell(new IngestCachedTelemetryReply(accepted) { RejectedEventIds = rejected }); } /// diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs index ef60235d..4dd3e3e2 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs @@ -302,13 +302,15 @@ public class SiteAuditReconciliationActor : ReceiveActor return; } - PruneUnknownSites(sites, eventStream); - if (sites.Count == 0) { return; } + // After the empty guard: an empty enumeration is as likely a transient lookup glitch + // as a fleet with no sites, and pruning on it would throw away every cursor. + PruneUnknownSites(sites, eventStream); + // Use CreateAsyncScope + await using so scoped EF Core // services (IAsyncDisposable DbContexts) dispose asynchronously // without blocking on sync Dispose() of pending connection cleanup. @@ -489,34 +491,51 @@ public class SiteAuditReconciliationActor : ReceiveActor // is permanently abandoned (LogCritical + cursor advances past) // to keep a truly broken row from blocking all subsequent // progress for the site. - var attempts = _failedInsertAttempts.GetValueOrDefault(evt.EventId) + 1; - _failedInsertAttempts[evt.EventId] = attempts; - - if (attempts >= MaxPermanentInsertAttempts) + if (AuditIngestErrorClassifier.IsTransient(ex)) { - _logger.LogCritical( + // A database problem (timeout, connection loss, deadlock, throttling, a + // maintenance window) is not a verdict about this row: hold the cursor + // and retry, but never count it toward permanent abandonment — a long + // outage must not turn healthy rows into ReconciliationAbandoned losses + // (review 2026-10-21 06/PERF-1 adversarial follow-up). + _logger.LogWarning( ex, - "Permanently abandoning AuditEvent {EventId} from site {SiteId} after {Attempts} consecutive insert failures; cursor will advance past it.", + "Reconciliation ingest of AuditEvent {EventId} from site {SiteId} hit a transient database fault; cursor held back, not counted toward abandonment.", evt.EventId, - site.SiteId, - attempts); - _failedInsertAttempts.Remove(evt.EventId); - RememberAbandoned(evt.EventId); - advanceForThisRow = true; - - await EmitAbandonmentRecordAsync(repository, evt, site.SiteId, ex, nowUtc) - .ConfigureAwait(false); + site.SiteId); + hasUnresolvedFailure = true; } else { - _logger.LogError( - ex, - "Reconciliation ingest failed for AuditEvent {EventId} from site {SiteId} (attempt {Attempts}/{Max}); cursor held back for retry.", - evt.EventId, - site.SiteId, - attempts, - MaxPermanentInsertAttempts); - hasUnresolvedFailure = true; + var attempts = _failedInsertAttempts.GetValueOrDefault(evt.EventId) + 1; + _failedInsertAttempts[evt.EventId] = attempts; + + if (attempts >= MaxPermanentInsertAttempts) + { + _logger.LogCritical( + ex, + "Permanently abandoning AuditEvent {EventId} from site {SiteId} after {Attempts} consecutive insert failures; cursor will advance past it.", + evt.EventId, + site.SiteId, + attempts); + _failedInsertAttempts.Remove(evt.EventId); + RememberAbandoned(evt.EventId); + advanceForThisRow = true; + + await EmitAbandonmentRecordAsync(repository, evt, site.SiteId, ex, nowUtc) + .ConfigureAwait(false); + } + else + { + _logger.LogError( + ex, + "Reconciliation ingest failed for AuditEvent {EventId} from site {SiteId} (attempt {Attempts}/{Max}); cursor held back for retry.", + evt.EventId, + site.SiteId, + attempts, + MaxPermanentInsertAttempts); + hasUnresolvedFailure = true; + } } } } diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/ISiteAuditPushLifecycle.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/ISiteAuditPushLifecycle.cs index c418ba21..2b3d25df 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/ISiteAuditPushLifecycle.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/ISiteAuditPushLifecycle.cs @@ -25,7 +25,7 @@ public static class SiteAuditForwardStates public const string PushAbandoned = "PushAbandoned"; /// - /// A row central answered but refused (not in AcceptedEventIds) on + /// A row central explicitly and permanently refused (listed in the ack's rejected_event_ids) 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 @@ -53,11 +53,11 @@ public interface ISiteAuditPushLifecycle Task MarkPushAbandonedAsync(IReadOnlyList eventIds, CancellationToken ct = default); /// - /// Records one refused push attempt for each supplied + /// Records one explicitly-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. + /// Event IDs central listed as permanently rejected. /// 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. diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SiteAuditTelemetryMetrics.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SiteAuditTelemetryMetrics.cs new file mode 100644 index 00000000..ac4c5946 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SiteAuditTelemetryMetrics.cs @@ -0,0 +1,28 @@ +using System.Diagnostics.Metrics; +using ZB.MOM.WW.ScadaBridge.Commons.Observability; + +namespace ZB.MOM.WW.ScadaBridge.AuditLog.Site; + +/// +/// Site audit forwarding instruments. Published on a meter named +/// — OpenTelemetry matches meters by NAME, so these +/// export through the existing ObservedMeters allowlist entry without a Host change. +/// +internal static class SiteAuditTelemetryMetrics +{ + private static readonly Meter Meter = new(ScadaBridgeTelemetry.MeterName); + + private static readonly Counter RejectedRows = + Meter.CreateCounter("scadabridge.site.audit.rejected_rows", unit: "1", + description: "Site audit rows central permanently refused MaxRejectedPushAttempts times, moved out of the push queue as Rejected (still served by the reconciliation pull)."); + + /// Records rows flipped to Rejected. + /// Rows flipped. + public static void RecordRejectedRows(long count) + { + if (count > 0) + { + RejectedRows.Add(count); + } + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs index d446483c..9cdac6d8 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs @@ -78,7 +78,6 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, ISiteAuditPushLi 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; @@ -89,17 +88,11 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, ISiteAuditPushLi /// 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, - IAuditWriteFailureCounter? failureCounter = null) + string? connectionStringOverride = null) { ArgumentNullException.ThrowIfNull(options); ArgumentNullException.ThrowIfNull(logger); @@ -108,7 +101,6 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, ISiteAuditPushLi _options = options.Value; _logger = logger; _nodeIdentity = nodeIdentity; - _failureCounter = failureCounter; var connectionString = connectionStringOverride ?? $"Data Source={_options.DatabasePath};Cache=Shared"; @@ -116,6 +108,7 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, ISiteAuditPushLi _connection.Open(); InitializeSchema(); + LoadMaxServedRowId(); // Open a second connection for read-only callers // (GetBacklogStatsAsync, ReadPendingAsync, ReadPendingSinceAsync, @@ -275,6 +268,16 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, ISiteAuditPushLi CREATE INDEX IF NOT EXISTS IX_fwd ON audit_forward_state (ForwardState, IsCachedKind, OccurredAtUtc); + -- Node-local key/value state that must survive a process restart. Holds the + -- reconciliation pull's served-rows high-water mark ('max_served_rowid'), so a + -- restart no longer resets the bound that lets a covering cursor retire rows the + -- pull already delivered (review 2026-10-21 06/PERF-1 adversarial follow-up). + -- Additive and idempotent: an older binary simply ignores the table. + CREATE TABLE IF NOT EXISTS audit_site_state ( + Key TEXT NOT NULL PRIMARY KEY, + Value INTEGER NOT NULL + ); + -- Flow Debugger (design 2026-09-05 §5.4): PullTraceEvents reads audit_event by -- time window + json_extract on ExecutionId/ParentExecutionId; this bounds the scan -- to the window. Additive and idempotent. @@ -345,6 +348,13 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, ISiteAuditPushLi // TryWrite can tell "disposed" from "full" without racing the channel's state. private bool _writerCompleted; + /// + /// Rows moved to since construction (also exported + /// as the scadabridge.site.audit.rejected_rows counter). + /// + public long RejectedPushRowCount => Interlocked.Read(ref _rejectedPushRows); + private long _rejectedPushRows; + private async Task ProcessWriteQueueAsync() { var batch = new List(_options.BatchSize); @@ -845,6 +855,8 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, ISiteAuditPushLi // 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. + IReadOnlyList rows; + var before = Volatile.Read(ref _maxServedRowId); lock (_readLock) { ObjectDisposedException.ThrowIf(_disposed, this); @@ -884,8 +896,33 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, ISiteAuditPushLi } cmd.Parameters.AddWithValue("$limit", batchSize); - return Task.FromResult(ReadServedRows(cmd, batchSize)); + rows = ReadServedRows(cmd, batchSize); } + + // Persist the (possibly advanced) mark OUTSIDE _readLock so the two locks are never + // nested in read→write order. Only when it moved — at most one tiny upsert per pull. + var mark = Volatile.Read(ref _maxServedRowId); + if (mark > before) + { + try + { + lock (_writeLock) + { + if (!_disposed) + { + PersistMaxServedRowId(mark); + } + } + } + catch (Exception ex) + { + _logger.LogWarning(ex, + "SqliteAuditWriter could not persist the served high-water mark {Mark}; a restart re-learns it from the next pull.", + mark); + } + } + + return Task.FromResult(rows); } /// @@ -1005,11 +1042,11 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, ISiteAuditPushLi "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 */ } - } + // Its OWN counter, not SiteAuditWriteFailures: the local write succeeded, and + // the refusal is central's verdict on the row — blaming the site's writer for it + // would be wrong (review 2026-10-21 adversarial follow-up). + Interlocked.Add(ref _rejectedPushRows, flipped); + SiteAuditTelemetryMetrics.RecordRejectedRows(flipped); } return Task.FromResult(flipped); @@ -1154,6 +1191,49 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, ISiteAuditPushLi /// private long _maxServedRowId; + private const string MaxServedRowIdKey = "max_served_rowid"; + + /// + /// Restores from audit_site_state, clamped to the live + /// MAX(rowid) so a stale value can never vouch for a reused rowid. Before this the + /// bound was per-process: after a site-node restart it was 0, and because a healthy site's + /// pull now often serves nothing, owed rows central had already ingested (e.g. PushAbandoned) + /// could stay unretired — and counted on the backlog — indefinitely. + /// + private void LoadMaxServedRowId() + { + using var cmd = _connection.CreateCommand(); + cmd.CommandText = """ + SELECT MIN( + IFNULL((SELECT Value FROM audit_site_state WHERE Key = $key), 0), + IFNULL((SELECT MAX(rowid) FROM audit_forward_state), 0)); + """; + cmd.Parameters.AddWithValue("$key", MaxServedRowIdKey); + var value = Convert.ToInt64(cmd.ExecuteScalar(), System.Globalization.CultureInfo.InvariantCulture); + Volatile.Write(ref _maxServedRowId, value); + } + + /// + /// Durably records the served high-water mark. Runs on the WRITE connection under + /// _writeLock (never nested inside _readLock). Best-effort: a failure leaves + /// the in-memory bound intact and only means a restart re-learns it from the next pull. + /// + private void PersistMaxServedRowId(long value, SqliteTransaction? transaction = null) + { + using var cmd = _connection.CreateCommand(); + cmd.Transaction = transaction; + cmd.CommandText = """ + INSERT INTO audit_site_state (Key, Value) VALUES ($key, $value) + ON CONFLICT(Key) DO UPDATE SET Value = excluded.Value; + """; + cmd.Parameters.AddWithValue("$key", MaxServedRowIdKey); + cmd.Parameters.AddWithValue("$value", value); + cmd.ExecuteNonQuery(); + } + + /// Test seam: the current in-memory served high-water mark. + internal long MaxServedRowIdForTests => Volatile.Read(ref _maxServedRowId); + /// /// Executes a reconciliation-pull read whose projection carries fs.rowid as an /// 11th column, materialising the canonical rows while advancing @@ -1222,12 +1302,12 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, ISiteAuditPushLi // 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 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 - // reconciliation means for such rows. + // The bound is persisted in audit_site_state (loaded at construction, clamped to + // the live MAX(rowid)), so a site-node restart no longer resets it to 0 and strands + // rows the pull had already served. Conservative in the safe direction (it can + // only delay retirement, never lose a row); the liveness note in + // MarkReconciledUpToAsync's interface doc + Component-AuditLog.md covers what a + // permanently halted reconciliation means for such rows. using var cmd = _connection.CreateCommand(); cmd.CommandText = $""" UPDATE audit_forward_state AS fs @@ -1506,7 +1586,10 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, ISiteAuditPushLi var liveMax = Convert.ToInt64(maxCmd.ExecuteScalar(), System.Globalization.CultureInfo.InvariantCulture); if (liveMax < Volatile.Read(ref _maxServedRowId)) + { Volatile.Write(ref _maxServedRowId, liveMax); + PersistMaxServedRowId(liveMax, transaction); + } } transaction.Commit(); 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 d5d729d4..6526bab1 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryActor.cs @@ -481,28 +481,30 @@ public class SiteAuditTelemetryActor : ReceiveActor } /// - /// 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). + /// Records a refused push attempt for every SENT row central explicitly listed in the ack's + /// RejectedEventIds — a permanent, row-intrinsic insert failure. 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. + /// "Not accepted" is NOT a refusal: central answers with an empty or partial ack during a SQL + /// maintenance window, a timeout, or an exhausted ingest budget, and counting those would flip + /// healthy rows to Rejected. Only the explicit list counts; an older central never sends it, so + /// against it this degrades to the pre-IMPL-1 behaviour (rows stay Pending). Best-effort: a + /// bookkeeping failure leaves the rows Pending. /// private async Task RecordRefusalsAsync(IReadOnlyList sentIds, IngestAck ack, CancellationToken ct) { - if (_queue is not ISiteAuditPushLifecycle lifecycle || sentIds.Count == 0) + if (_queue is not ISiteAuditPushLifecycle lifecycle || sentIds.Count == 0 || ack.RejectedEventIds.Count == 0) { return; } - var accepted = new HashSet(ack.AcceptedEventIds, StringComparer.OrdinalIgnoreCase); + var sent = new HashSet(sentIds, StringComparer.OrdinalIgnoreCase); var refused = new List(); - foreach (var raw in sentIds) + foreach (var raw in ack.RejectedEventIds) { - if (!accepted.Contains(raw) && Guid.TryParse(raw, out var id)) + if (sent.Contains(raw) && Guid.TryParse(raw, out var id)) { refused.Add(id); } 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 ae6e95c4..c1e25e64 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryOptions.cs @@ -57,18 +57,22 @@ 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). + /// Number of drains in which central EXPLICITLY refused a row (listed it in the ack's + /// additive rejected_event_ids — a permanent, row-intrinsic insert failure such as an + /// over-width value) 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: + /// 06/IMPL-1). A transport fault, and a row merely ABSENT from the accepted list (a database + /// outage, a timeout, an exhausted ingest budget), never counts; an older central that does not + /// send the field therefore never triggers it. 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. + /// before it retires. Each flip is counted on scadabridge.site.audit.rejected_rows (NOT + /// SiteAuditWriteFailures — the site's own write succeeded). /// public int MaxRejectedPushAttempts { get; set; } = 10; } diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteCommunicationAuditClient.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteCommunicationAuditClient.cs index a56b5e90..66fcd94a 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteCommunicationAuditClient.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteCommunicationAuditClient.cs @@ -82,7 +82,7 @@ public sealed class SiteCommunicationAuditClient : ISiteStreamAuditClient .Ask(new IngestAuditEventsCommand(events), _askTimeout, ct) .ConfigureAwait(false); - return ToAck(reply.AcceptedEventIds); + return ToAck(reply.AcceptedEventIds, reply.RejectedEventIds); } /// @@ -105,16 +105,21 @@ public sealed class SiteCommunicationAuditClient : ISiteStreamAuditClient .Ask(new IngestCachedTelemetryCommand(entries), _askTimeout, ct) .ConfigureAwait(false); - return ToAck(reply.AcceptedEventIds); + return ToAck(reply.AcceptedEventIds, reply.RejectedEventIds); } - private static IngestAck ToAck(IReadOnlyList acceptedEventIds) + private static IngestAck ToAck(IReadOnlyList acceptedEventIds, IReadOnlyList rejectedEventIds) { var ack = new IngestAck(); foreach (var id in acceptedEventIds) { ack.AcceptedEventIds.Add(id.ToString()); } + // Explicit permanent refusals only (review 2026-10-21 06/IMPL-1); empty from an older central. + foreach (var id in rejectedEventIds) + { + ack.RejectedEventIds.Add(id.ToString()); + } return ack; } } diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Audit/IngestAuditEventsReply.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Audit/IngestAuditEventsReply.cs index dbca0346..5e7ba190 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Audit/IngestAuditEventsReply.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Audit/IngestAuditEventsReply.cs @@ -8,4 +8,13 @@ namespace ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit; /// The gRPC handler echoes these ids back over the wire as the IngestAck /// the site uses to flip rows to Forwarded. /// -public sealed record IngestAuditEventsReply(IReadOnlyList AcceptedEventIds); +public sealed record IngestAuditEventsReply(IReadOnlyList AcceptedEventIds) +{ + /// + /// Ids central PERMANENTLY refused — a non-transient per-row insert failure (e.g. an + /// over-width value), never a database outage or an exhausted budget. Additive (review + /// 2026-10-21 06/IMPL-1): the site counts only these toward giving a row up as + /// Rejected; empty from an older central, which therefore never triggers it. + /// + public IReadOnlyList RejectedEventIds { get; init; } = Array.Empty(); +} diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Audit/IngestCachedTelemetryReply.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Audit/IngestCachedTelemetryReply.cs index e05340c9..2f39da3b 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Audit/IngestCachedTelemetryReply.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Audit/IngestCachedTelemetryReply.cs @@ -7,4 +7,13 @@ namespace ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit; /// UPSERT) committed; entries whose transaction rolled back are absent so the /// site can leave the row Pending and retry on the next drain. /// -public sealed record IngestCachedTelemetryReply(IReadOnlyList AcceptedEventIds); +public sealed record IngestCachedTelemetryReply(IReadOnlyList AcceptedEventIds) +{ + /// + /// Ids central PERMANENTLY refused — a non-transient per-row insert failure (e.g. an + /// over-width value), never a database outage or an exhausted budget. Additive (review + /// 2026-10-21 06/IMPL-1): the site counts only these toward giving a row up as + /// Rejected; empty from an older central, which therefore never triggers it. + /// + public IReadOnlyList RejectedEventIds { get; init; } = Array.Empty(); +} diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralControlDtoMapper.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralControlDtoMapper.cs index a981aab3..2f555e3c 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralControlDtoMapper.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralControlDtoMapper.cs @@ -314,7 +314,9 @@ public static class CentralControlDtoMapper /// /// Ids central considers durably persisted. /// The wire ack carrying the ids in "D" string form, in order. - public static IngestAck ToIngestAck(IReadOnlyList acceptedEventIds) + /// Ids central permanently refused (additive field 2); null = none. + public static IngestAck ToIngestAck( + IReadOnlyList acceptedEventIds, IReadOnlyList? rejectedEventIds = null) { ArgumentNullException.ThrowIfNull(acceptedEventIds); @@ -324,9 +326,39 @@ public static class CentralControlDtoMapper ack.AcceptedEventIds.Add(id.ToString()); } + if (rejectedEventIds is not null) + { + foreach (var id in rejectedEventIds) + { + ack.RejectedEventIds.Add(id.ToString()); + } + } + return ack; } + /// + /// Reads the permanently-rejected id list (additive field 2) out of an ; + /// empty from an older central. Unparseable entries are skipped. + /// + /// The wire ack to read. + /// The rejected event ids, in wire order. + public static IReadOnlyList RejectedFromIngestAck(IngestAck ack) + { + ArgumentNullException.ThrowIfNull(ack); + + var ids = new List(ack.RejectedEventIds.Count); + foreach (var id in ack.RejectedEventIds) + { + if (Guid.TryParse(id, out var g)) + { + ids.Add(g); + } + } + + return ids; + } + /// /// Reads the accepted-id list back out of an . /// diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralControlGrpcService.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralControlGrpcService.cs index 75bd4e96..b6915d24 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralControlGrpcService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/CentralControlGrpcService.cs @@ -198,7 +198,7 @@ public sealed class CentralControlGrpcService : CentralControlService.CentralCon SiteStreamGrpcServer.AuditIngestAskTimeout, context).ConfigureAwait(false); - return CentralControlDtoMapper.ToIngestAck(reply.AcceptedEventIds); + return CentralControlDtoMapper.ToIngestAck(reply.AcceptedEventIds, reply.RejectedEventIds); } /// @@ -223,7 +223,7 @@ public sealed class CentralControlGrpcService : CentralControlService.CentralCon SiteStreamGrpcServer.AuditIngestAskTimeout, context).ConfigureAwait(false); - return CentralControlDtoMapper.ToIngestAck(reply.AcceptedEventIds); + return CentralControlDtoMapper.ToIngestAck(reply.AcceptedEventIds, reply.RejectedEventIds); } /// diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcCentralTransport.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcCentralTransport.cs index d72fcf4a..f914d7bc 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcCentralTransport.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/GrpcCentralTransport.cs @@ -88,7 +88,10 @@ public sealed class GrpcCentralTransport : ICentralTransport var batch = CentralControlDtoMapper.ToDto(message); Dispatch(replyTo, SiteStreamGrpcServer.AuditIngestAskTimeout, (c, o) => c.IngestAuditEventsAsync(batch, o), - ack => new IngestAuditEventsReply(CentralControlDtoMapper.FromIngestAck(ack)), retrySafe: true); + ack => new IngestAuditEventsReply(CentralControlDtoMapper.FromIngestAck(ack)) + { + RejectedEventIds = CentralControlDtoMapper.RejectedFromIngestAck(ack), + }, retrySafe: true); } /// @@ -97,7 +100,10 @@ public sealed class GrpcCentralTransport : ICentralTransport var batch = CentralControlDtoMapper.ToDto(message); Dispatch(replyTo, SiteStreamGrpcServer.AuditIngestAskTimeout, (c, o) => c.IngestCachedTelemetryAsync(batch, o), - ack => new IngestCachedTelemetryReply(CentralControlDtoMapper.FromIngestAck(ack)), retrySafe: true); + ack => new IngestCachedTelemetryReply(CentralControlDtoMapper.FromIngestAck(ack)) + { + RejectedEventIds = CentralControlDtoMapper.RejectedFromIngestAck(ack), + }, retrySafe: true); } /// diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Protos/sitestream.proto b/src/ZB.MOM.WW.ScadaBridge.Communication/Protos/sitestream.proto index 864bd17d..535a8afd 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Protos/sitestream.proto +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Protos/sitestream.proto @@ -172,7 +172,15 @@ message AuditEventDto { } message AuditEventBatch { repeated AuditEventDto events = 1; } -message IngestAck { repeated string accepted_event_ids = 1; } +// rejected_event_ids (additive, field 2 — review 2026-10-21 06/IMPL-1): ids central +// PERMANENTLY refused (a non-transient per-row SQL error such as an over-width +// value). Only these count toward the site's MaxRejectedPushAttempts; ids that are +// neither accepted nor rejected (DB outage, budget exhaustion) are simply retried. +// An older central never sets it, so the site never counts a refusal. +message IngestAck { + repeated string accepted_event_ids = 1; + repeated string rejected_event_ids = 2; +} // Audit Log (#23) M3 cached-call combined telemetry: a single packet carries // both the AuditEvent row to insert and the SiteCalls operational-state upsert diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/SiteStreamGrpc/Sitestream.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/SiteStreamGrpc/Sitestream.cs index 159a9b05..db029242 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/SiteStreamGrpc/Sitestream.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/SiteStreamGrpc/Sitestream.cs @@ -71,57 +71,58 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { "b2FkX3RydW5jYXRlZBgSIAEoCBINCgVleHRyYRgTIAEoCRIUCgxleGVjdXRp", "b25faWQYFCABKAkSGwoTcGFyZW50X2V4ZWN1dGlvbl9pZBgVIAEoCRITCgtz", "b3VyY2Vfbm9kZRgWIAEoCSI8Cg9BdWRpdEV2ZW50QmF0Y2gSKQoGZXZlbnRz", - "GAEgAygLMhkuc2l0ZXN0cmVhbS5BdWRpdEV2ZW50RHRvIicKCUluZ2VzdEFj", - "axIaChJhY2NlcHRlZF9ldmVudF9pZHMYASADKAkiiQMKFlNpdGVDYWxsT3Bl", - "cmF0aW9uYWxEdG8SHAoUdHJhY2tlZF9vcGVyYXRpb25faWQYASABKAkSDwoH", - "Y2hhbm5lbBgCIAEoCRIOCgZ0YXJnZXQYAyABKAkSEwoLc291cmNlX3NpdGUY", - "BCABKAkSDgoGc3RhdHVzGAUgASgJEhMKC3JldHJ5X2NvdW50GAYgASgFEhIK", - "Cmxhc3RfZXJyb3IYByABKAkSMAoLaHR0cF9zdGF0dXMYCCABKAsyGy5nb29n", - "bGUucHJvdG9idWYuSW50MzJWYWx1ZRIyCg5jcmVhdGVkX2F0X3V0YxgJIAEo", - "CzIaLmdvb2dsZS5wcm90b2J1Zi5UaW1lc3RhbXASMgoOdXBkYXRlZF9hdF91", - "dGMYCiABKAsyGi5nb29nbGUucHJvdG9idWYuVGltZXN0YW1wEjMKD3Rlcm1p", - "bmFsX2F0X3V0YxgLIAEoCzIaLmdvb2dsZS5wcm90b2J1Zi5UaW1lc3RhbXAS", - "EwoLc291cmNlX25vZGUYDCABKAkigAEKFUNhY2hlZFRlbGVtZXRyeVBhY2tl", - "dBIuCgthdWRpdF9ldmVudBgBIAEoCzIZLnNpdGVzdHJlYW0uQXVkaXRFdmVu", - "dER0bxI3CgtvcGVyYXRpb25hbBgCIAEoCzIiLnNpdGVzdHJlYW0uU2l0ZUNh", - "bGxPcGVyYXRpb25hbER0byJKChRDYWNoZWRUZWxlbWV0cnlCYXRjaBIyCgdw", - "YWNrZXRzGAEgAygLMiEuc2l0ZXN0cmVhbS5DYWNoZWRUZWxlbWV0cnlQYWNr", - "ZXQibQoWUHVsbEF1ZGl0RXZlbnRzUmVxdWVzdBItCglzaW5jZV91dGMYASAB", - "KAsyGi5nb29nbGUucHJvdG9idWYuVGltZXN0YW1wEhIKCmJhdGNoX3NpemUY", - "AiABKAUSEAoIYWZ0ZXJfaWQYAyABKAkiXAoXUHVsbEF1ZGl0RXZlbnRzUmVz", - "cG9uc2USKQoGZXZlbnRzGAEgAygLMhkuc2l0ZXN0cmVhbS5BdWRpdEV2ZW50", - "RHRvEhYKDm1vcmVfYXZhaWxhYmxlGAIgASgIIoQBChZQdWxsVHJhY2VFdmVu", - "dHNSZXF1ZXN0EhAKCGFmdGVyX2lkGAQgASgJEhUKDWV4ZWN1dGlvbl9pZHMY", - "ASADKAkSLQoJc2luY2VfdXRjGAIgASgLMhouZ29vZ2xlLnByb3RvYnVmLlRp", - "bWVzdGFtcBISCgptYXhfZXZlbnRzGAMgASgFIkQKF1B1bGxUcmFjZUV2ZW50", - "c1Jlc3BvbnNlEikKBmV2ZW50cxgBIAMoCzIZLnNpdGVzdHJlYW0uQXVkaXRF", - "dmVudER0byJrChRQdWxsU2l0ZUNhbGxzUmVxdWVzdBItCglzaW5jZV91dGMY", - "ASABKAsyGi5nb29nbGUucHJvdG9idWYuVGltZXN0YW1wEhIKCmJhdGNoX3Np", - "emUYAiABKAUSEAoIYWZ0ZXJfaWQYAyABKAkiaQoVUHVsbFNpdGVDYWxsc1Jl", - "c3BvbnNlEjgKDG9wZXJhdGlvbmFscxgBIAMoCzIiLnNpdGVzdHJlYW0uU2l0", - "ZUNhbGxPcGVyYXRpb25hbER0bxIWCg5tb3JlX2F2YWlsYWJsZRgCIAEoCCpc", - "CgdRdWFsaXR5EhcKE1FVQUxJVFlfVU5TUEVDSUZJRUQQABIQCgxRVUFMSVRZ", - "X0dPT0QQARIVChFRVUFMSVRZX1VOQ0VSVEFJThACEg8KC1FVQUxJVFlfQkFE", - "EAMqXQoOQWxhcm1TdGF0ZUVudW0SGwoXQUxBUk1fU1RBVEVfVU5TUEVDSUZJ", - "RUQQABIWChJBTEFSTV9TVEFURV9OT1JNQUwQARIWChJBTEFSTV9TVEFURV9B", - "Q1RJVkUQAiqFAQoOQWxhcm1MZXZlbEVudW0SFAoQQUxBUk1fTEVWRUxfTk9O", - "RRAAEhMKD0FMQVJNX0xFVkVMX0xPVxABEhcKE0FMQVJNX0xFVkVMX0xPV19M", - "T1cQAhIUChBBTEFSTV9MRVZFTF9ISUdIEAMSGQoVQUxBUk1fTEVWRUxfSElH", - "SF9ISUdIEAQy4gQKEVNpdGVTdHJlYW1TZXJ2aWNlElUKEVN1YnNjcmliZUlu", - "c3RhbmNlEiEuc2l0ZXN0cmVhbS5JbnN0YW5jZVN0cmVhbVJlcXVlc3QaGy5z", - "aXRlc3RyZWFtLlNpdGVTdHJlYW1FdmVudDABEk0KDVN1YnNjcmliZVNpdGUS", - "HS5zaXRlc3RyZWFtLlNpdGVTdHJlYW1SZXF1ZXN0Ghsuc2l0ZXN0cmVhbS5T", - "aXRlU3RyZWFtRXZlbnQwARJHChFJbmdlc3RBdWRpdEV2ZW50cxIbLnNpdGVz", - "dHJlYW0uQXVkaXRFdmVudEJhdGNoGhUuc2l0ZXN0cmVhbS5Jbmdlc3RBY2sS", - "UAoVSW5nZXN0Q2FjaGVkVGVsZW1ldHJ5EiAuc2l0ZXN0cmVhbS5DYWNoZWRU", - "ZWxlbWV0cnlCYXRjaBoVLnNpdGVzdHJlYW0uSW5nZXN0QWNrEloKD1B1bGxB", - "dWRpdEV2ZW50cxIiLnNpdGVzdHJlYW0uUHVsbEF1ZGl0RXZlbnRzUmVxdWVz", - "dBojLnNpdGVzdHJlYW0uUHVsbEF1ZGl0RXZlbnRzUmVzcG9uc2USWgoPUHVs", - "bFRyYWNlRXZlbnRzEiIuc2l0ZXN0cmVhbS5QdWxsVHJhY2VFdmVudHNSZXF1", - "ZXN0GiMuc2l0ZXN0cmVhbS5QdWxsVHJhY2VFdmVudHNSZXNwb25zZRJUCg1Q", - "dWxsU2l0ZUNhbGxzEiAuc2l0ZXN0cmVhbS5QdWxsU2l0ZUNhbGxzUmVxdWVz", - "dBohLnNpdGVzdHJlYW0uUHVsbFNpdGVDYWxsc1Jlc3BvbnNlQiuqAihaQi5N", - "T00uV1cuU2NhZGFCcmlkZ2UuQ29tbXVuaWNhdGlvbi5HcnBjYgZwcm90bzM=")); + "GAEgAygLMhkuc2l0ZXN0cmVhbS5BdWRpdEV2ZW50RHRvIkMKCUluZ2VzdEFj", + "axIaChJhY2NlcHRlZF9ldmVudF9pZHMYASADKAkSGgoScmVqZWN0ZWRfZXZl", + "bnRfaWRzGAIgAygJIokDChZTaXRlQ2FsbE9wZXJhdGlvbmFsRHRvEhwKFHRy", + "YWNrZWRfb3BlcmF0aW9uX2lkGAEgASgJEg8KB2NoYW5uZWwYAiABKAkSDgoG", + "dGFyZ2V0GAMgASgJEhMKC3NvdXJjZV9zaXRlGAQgASgJEg4KBnN0YXR1cxgF", + "IAEoCRITCgtyZXRyeV9jb3VudBgGIAEoBRISCgpsYXN0X2Vycm9yGAcgASgJ", + "EjAKC2h0dHBfc3RhdHVzGAggASgLMhsuZ29vZ2xlLnByb3RvYnVmLkludDMy", + "VmFsdWUSMgoOY3JlYXRlZF9hdF91dGMYCSABKAsyGi5nb29nbGUucHJvdG9i", + "dWYuVGltZXN0YW1wEjIKDnVwZGF0ZWRfYXRfdXRjGAogASgLMhouZ29vZ2xl", + "LnByb3RvYnVmLlRpbWVzdGFtcBIzCg90ZXJtaW5hbF9hdF91dGMYCyABKAsy", + "Gi5nb29nbGUucHJvdG9idWYuVGltZXN0YW1wEhMKC3NvdXJjZV9ub2RlGAwg", + "ASgJIoABChVDYWNoZWRUZWxlbWV0cnlQYWNrZXQSLgoLYXVkaXRfZXZlbnQY", + "ASABKAsyGS5zaXRlc3RyZWFtLkF1ZGl0RXZlbnREdG8SNwoLb3BlcmF0aW9u", + "YWwYAiABKAsyIi5zaXRlc3RyZWFtLlNpdGVDYWxsT3BlcmF0aW9uYWxEdG8i", + "SgoUQ2FjaGVkVGVsZW1ldHJ5QmF0Y2gSMgoHcGFja2V0cxgBIAMoCzIhLnNp", + "dGVzdHJlYW0uQ2FjaGVkVGVsZW1ldHJ5UGFja2V0Im0KFlB1bGxBdWRpdEV2", + "ZW50c1JlcXVlc3QSLQoJc2luY2VfdXRjGAEgASgLMhouZ29vZ2xlLnByb3Rv", + "YnVmLlRpbWVzdGFtcBISCgpiYXRjaF9zaXplGAIgASgFEhAKCGFmdGVyX2lk", + "GAMgASgJIlwKF1B1bGxBdWRpdEV2ZW50c1Jlc3BvbnNlEikKBmV2ZW50cxgB", + "IAMoCzIZLnNpdGVzdHJlYW0uQXVkaXRFdmVudER0bxIWCg5tb3JlX2F2YWls", + "YWJsZRgCIAEoCCKEAQoWUHVsbFRyYWNlRXZlbnRzUmVxdWVzdBIQCghhZnRl", + "cl9pZBgEIAEoCRIVCg1leGVjdXRpb25faWRzGAEgAygJEi0KCXNpbmNlX3V0", + "YxgCIAEoCzIaLmdvb2dsZS5wcm90b2J1Zi5UaW1lc3RhbXASEgoKbWF4X2V2", + "ZW50cxgDIAEoBSJEChdQdWxsVHJhY2VFdmVudHNSZXNwb25zZRIpCgZldmVu", + "dHMYASADKAsyGS5zaXRlc3RyZWFtLkF1ZGl0RXZlbnREdG8iawoUUHVsbFNp", + "dGVDYWxsc1JlcXVlc3QSLQoJc2luY2VfdXRjGAEgASgLMhouZ29vZ2xlLnBy", + "b3RvYnVmLlRpbWVzdGFtcBISCgpiYXRjaF9zaXplGAIgASgFEhAKCGFmdGVy", + "X2lkGAMgASgJImkKFVB1bGxTaXRlQ2FsbHNSZXNwb25zZRI4CgxvcGVyYXRp", + "b25hbHMYASADKAsyIi5zaXRlc3RyZWFtLlNpdGVDYWxsT3BlcmF0aW9uYWxE", + "dG8SFgoObW9yZV9hdmFpbGFibGUYAiABKAgqXAoHUXVhbGl0eRIXChNRVUFM", + "SVRZX1VOU1BFQ0lGSUVEEAASEAoMUVVBTElUWV9HT09EEAESFQoRUVVBTElU", + "WV9VTkNFUlRBSU4QAhIPCgtRVUFMSVRZX0JBRBADKl0KDkFsYXJtU3RhdGVF", + "bnVtEhsKF0FMQVJNX1NUQVRFX1VOU1BFQ0lGSUVEEAASFgoSQUxBUk1fU1RB", + "VEVfTk9STUFMEAESFgoSQUxBUk1fU1RBVEVfQUNUSVZFEAIqhQEKDkFsYXJt", + "TGV2ZWxFbnVtEhQKEEFMQVJNX0xFVkVMX05PTkUQABITCg9BTEFSTV9MRVZF", + "TF9MT1cQARIXChNBTEFSTV9MRVZFTF9MT1dfTE9XEAISFAoQQUxBUk1fTEVW", + "RUxfSElHSBADEhkKFUFMQVJNX0xFVkVMX0hJR0hfSElHSBAEMuIEChFTaXRl", + "U3RyZWFtU2VydmljZRJVChFTdWJzY3JpYmVJbnN0YW5jZRIhLnNpdGVzdHJl", + "YW0uSW5zdGFuY2VTdHJlYW1SZXF1ZXN0Ghsuc2l0ZXN0cmVhbS5TaXRlU3Ry", + "ZWFtRXZlbnQwARJNCg1TdWJzY3JpYmVTaXRlEh0uc2l0ZXN0cmVhbS5TaXRl", + "U3RyZWFtUmVxdWVzdBobLnNpdGVzdHJlYW0uU2l0ZVN0cmVhbUV2ZW50MAES", + "RwoRSW5nZXN0QXVkaXRFdmVudHMSGy5zaXRlc3RyZWFtLkF1ZGl0RXZlbnRC", + "YXRjaBoVLnNpdGVzdHJlYW0uSW5nZXN0QWNrElAKFUluZ2VzdENhY2hlZFRl", + "bGVtZXRyeRIgLnNpdGVzdHJlYW0uQ2FjaGVkVGVsZW1ldHJ5QmF0Y2gaFS5z", + "aXRlc3RyZWFtLkluZ2VzdEFjaxJaCg9QdWxsQXVkaXRFdmVudHMSIi5zaXRl", + "c3RyZWFtLlB1bGxBdWRpdEV2ZW50c1JlcXVlc3QaIy5zaXRlc3RyZWFtLlB1", + "bGxBdWRpdEV2ZW50c1Jlc3BvbnNlEloKD1B1bGxUcmFjZUV2ZW50cxIiLnNp", + "dGVzdHJlYW0uUHVsbFRyYWNlRXZlbnRzUmVxdWVzdBojLnNpdGVzdHJlYW0u", + "UHVsbFRyYWNlRXZlbnRzUmVzcG9uc2USVAoNUHVsbFNpdGVDYWxscxIgLnNp", + "dGVzdHJlYW0uUHVsbFNpdGVDYWxsc1JlcXVlc3QaIS5zaXRlc3RyZWFtLlB1", + "bGxTaXRlQ2FsbHNSZXNwb25zZUIrqgIoWkIuTU9NLldXLlNjYWRhQnJpZGdl", + "LkNvbW11bmljYXRpb24uR3JwY2IGcHJvdG8z")); descriptor = pbr::FileDescriptor.FromGeneratedCode(descriptorData, new pbr::FileDescriptor[] { global::Google.Protobuf.WellKnownTypes.TimestampReflection.Descriptor, global::Google.Protobuf.WellKnownTypes.WrappersReflection.Descriptor, }, new pbr::GeneratedClrTypeInfo(new[] {typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.Quality), typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AlarmStateEnum), typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AlarmLevelEnum), }, null, new pbr::GeneratedClrTypeInfo[] { @@ -133,7 +134,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AlarmStateUpdate), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AlarmStateUpdate.Parser, new[]{ "InstanceUniqueName", "AlarmName", "State", "Priority", "Timestamp", "Level", "Message", "Kind", "Active", "Acknowledged", "Confirmed", "ShelveState", "Suppressed", "SourceReference", "AlarmTypeName", "Category", "OperatorUser", "OperatorComment", "OriginalRaiseTime", "CurrentValue", "LimitValue", "NativeSourceCanonicalName", "IsConfiguredPlaceholder", "AckTime" }, null, null, null, null), new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AuditEventDto), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AuditEventDto.Parser, new[]{ "EventId", "OccurredAtUtc", "Channel", "Kind", "CorrelationId", "SourceSiteId", "SourceInstanceId", "SourceScript", "Actor", "Target", "Status", "HttpStatus", "DurationMs", "ErrorMessage", "ErrorDetail", "RequestSummary", "ResponseSummary", "PayloadTruncated", "Extra", "ExecutionId", "ParentExecutionId", "SourceNode" }, null, null, null, null), new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AuditEventBatch), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.AuditEventBatch.Parser, new[]{ "Events" }, null, null, null, null), - new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.IngestAck), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.IngestAck.Parser, new[]{ "AcceptedEventIds" }, null, null, null, null), + new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.IngestAck), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.IngestAck.Parser, new[]{ "AcceptedEventIds", "RejectedEventIds" }, null, null, null, null), new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteCallOperationalDto), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteCallOperationalDto.Parser, new[]{ "TrackedOperationId", "Channel", "Target", "SourceSite", "Status", "RetryCount", "LastError", "HttpStatus", "CreatedAtUtc", "UpdatedAtUtc", "TerminalAtUtc", "SourceNode" }, null, null, null, null), new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.CachedTelemetryPacket), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.CachedTelemetryPacket.Parser, new[]{ "AuditEvent", "Operational" }, null, null, null, null), new pbr::GeneratedClrTypeInfo(typeof(global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.CachedTelemetryBatch), global::ZB.MOM.WW.ScadaBridge.Communication.Grpc.CachedTelemetryBatch.Parser, new[]{ "Packets" }, null, null, null, null), @@ -4085,6 +4086,13 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { } + /// + /// rejected_event_ids (additive, field 2 — review 2026-10-21 06/IMPL-1): ids central + /// PERMANENTLY refused (a non-transient per-row SQL error such as an over-width + /// value). Only these count toward the site's MaxRejectedPushAttempts; ids that are + /// neither accepted nor rejected (DB outage, budget exhaustion) are simply retried. + /// An older central never sets it, so the site never counts a refusal. + /// [global::System.Diagnostics.DebuggerDisplayAttribute("{ToString(),nq}")] public sealed partial class IngestAck : pb::IMessage #if !GOOGLE_PROTOBUF_REFSTRUCT_COMPATIBILITY_MODE @@ -4121,6 +4129,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] public IngestAck(IngestAck other) : this() { acceptedEventIds_ = other.acceptedEventIds_.Clone(); + rejectedEventIds_ = other.rejectedEventIds_.Clone(); _unknownFields = pb::UnknownFieldSet.Clone(other._unknownFields); } @@ -4141,6 +4150,17 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { get { return acceptedEventIds_; } } + /// Field number for the "rejected_event_ids" field. + public const int RejectedEventIdsFieldNumber = 2; + private static readonly pb::FieldCodec _repeated_rejectedEventIds_codec + = pb::FieldCodec.ForString(18); + private readonly pbc::RepeatedField rejectedEventIds_ = new pbc::RepeatedField(); + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] + [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] + public pbc::RepeatedField RejectedEventIds { + get { return rejectedEventIds_; } + } + [global::System.Diagnostics.DebuggerNonUserCodeAttribute] [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] public override bool Equals(object other) { @@ -4157,6 +4177,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { return true; } if(!acceptedEventIds_.Equals(other.acceptedEventIds_)) return false; + if(!rejectedEventIds_.Equals(other.rejectedEventIds_)) return false; return Equals(_unknownFields, other._unknownFields); } @@ -4165,6 +4186,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { public override int GetHashCode() { int hash = 1; hash ^= acceptedEventIds_.GetHashCode(); + hash ^= rejectedEventIds_.GetHashCode(); if (_unknownFields != null) { hash ^= _unknownFields.GetHashCode(); } @@ -4184,6 +4206,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { output.WriteRawMessage(this); #else acceptedEventIds_.WriteTo(output, _repeated_acceptedEventIds_codec); + rejectedEventIds_.WriteTo(output, _repeated_rejectedEventIds_codec); if (_unknownFields != null) { _unknownFields.WriteTo(output); } @@ -4195,6 +4218,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { [global::System.CodeDom.Compiler.GeneratedCode("protoc", null)] void pb::IBufferMessage.InternalWriteTo(ref pb::WriteContext output) { acceptedEventIds_.WriteTo(ref output, _repeated_acceptedEventIds_codec); + rejectedEventIds_.WriteTo(ref output, _repeated_rejectedEventIds_codec); if (_unknownFields != null) { _unknownFields.WriteTo(ref output); } @@ -4206,6 +4230,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { public int CalculateSize() { int size = 0; size += acceptedEventIds_.CalculateSize(_repeated_acceptedEventIds_codec); + size += rejectedEventIds_.CalculateSize(_repeated_rejectedEventIds_codec); if (_unknownFields != null) { size += _unknownFields.CalculateSize(); } @@ -4219,6 +4244,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { return; } acceptedEventIds_.Add(other.acceptedEventIds_); + rejectedEventIds_.Add(other.rejectedEventIds_); _unknownFields = pb::UnknownFieldSet.MergeFrom(_unknownFields, other._unknownFields); } @@ -4242,6 +4268,10 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { acceptedEventIds_.AddEntriesFrom(input, _repeated_acceptedEventIds_codec); break; } + case 18: { + rejectedEventIds_.AddEntriesFrom(input, _repeated_rejectedEventIds_codec); + break; + } } } #endif @@ -4265,6 +4295,10 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc { acceptedEventIds_.AddEntriesFrom(ref input, _repeated_acceptedEventIds_codec); break; } + case 18: { + rejectedEventIds_.AddEntriesFrom(ref input, _repeated_rejectedEventIds_codec); + break; + } } } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/AuditIngestRejectionTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/AuditIngestRejectionTests.cs new file mode 100644 index 00000000..1b33c01d --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/AuditIngestRejectionTests.cs @@ -0,0 +1,132 @@ +using Akka.Actor; +using Akka.TestKit.Xunit2; +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; +using NSubstitute.ExceptionExtensions; +using ZB.MOM.WW.Audit; +using ZB.MOM.WW.ScadaBridge.AuditLog.Central; +using ZB.MOM.WW.ScadaBridge.AuditLog.Tests.TestSupport; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; + +namespace ZB.MOM.WW.ScadaBridge.AuditLog.Tests.Central; + +/// +/// Review 2026-10-21 06/IMPL-1 adversarial follow-up: central reports a row as REJECTED only for a +/// permanent, row-intrinsic SQL failure; a database outage / timeout leaves the id in neither list, +/// so the site never mistakes a maintenance window for a poison row. Also pins the classifier the +/// reconciliation puller uses to decide what counts toward permanent abandonment (06/COR follow-up 3). +/// +public class AuditIngestRejectionTests : TestKit +{ + private static AuditEvent NewEvent() => ScadaBridgeAuditEventFactory.Create( + eventId: Guid.NewGuid(), + occurredAtUtc: new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc), + channel: AuditChannel.ApiOutbound, + kind: AuditKind.ApiCall, + status: AuditStatus.Delivered, + sourceSiteId: "site-a"); + + /// The batch reply lists only the over-width row as rejected; the timed-out row is in neither list. + [Fact] + public void Ingest_ReportsOnlyPermanentRowFailuresAsRejected() + { + var ok = NewEvent(); + var overWidth = NewEvent(); + var timedOut = NewEvent(); + var repo = Substitute.For(); + repo.InsertManyIfNotExistsAsync(Arg.Any>(), Arg.Any(), Arg.Any()) + .ThrowsAsync(SqlExceptionFactory.Create(2628)); + repo.InsertIfNotExistsAsync(Arg.Is(e => e.EventId == overWidth.EventId), Arg.Any()) + .ThrowsAsync(SqlExceptionFactory.Create(2628)); + repo.InsertIfNotExistsAsync(Arg.Is(e => e.EventId == timedOut.EventId), Arg.Any()) + .ThrowsAsync(SqlExceptionFactory.Create(-2)); + + var actor = Sys.ActorOf(Props.Create(() => new AuditLogIngestActor( + StubAuditRepositoryProvider.Build(repo), NullLogger.Instance))); + actor.Tell(new IngestAuditEventsCommand(new[] { ok, overWidth, timedOut }), TestActor); + + var reply = ExpectMsg(TimeSpan.FromSeconds(10)); + Assert.Equal(new[] { ok.EventId }, reply.AcceptedEventIds); + Assert.Equal(new[] { overWidth.EventId }, reply.RejectedEventIds); + } + + /// Classifier: row-intrinsic SQL errors are permanent; timeouts, connection loss and non-SQL faults are not. + [Fact] + public void Classifier_SeparatesRowVerdictsFromDatabaseProblems() + { + Assert.True(AuditIngestErrorClassifier.IsPermanentRowRejection(SqlExceptionFactory.Create(2628))); + Assert.True(AuditIngestErrorClassifier.IsPermanentRowRejection(SqlExceptionFactory.Create(515))); + Assert.True(AuditIngestErrorClassifier.IsPermanentRowRejection( + new InvalidOperationException("wrapped", SqlExceptionFactory.Create(8152)))); + Assert.False(AuditIngestErrorClassifier.IsPermanentRowRejection(SqlExceptionFactory.Create(-2))); + Assert.False(AuditIngestErrorClassifier.IsPermanentRowRejection(new TimeoutException())); + Assert.False(AuditIngestErrorClassifier.IsPermanentRowRejection(new InvalidOperationException("bug"))); + + Assert.True(AuditIngestErrorClassifier.IsTransient(SqlExceptionFactory.Create(-2))); + Assert.True(AuditIngestErrorClassifier.IsTransient(SqlExceptionFactory.Create(1205))); + Assert.True(AuditIngestErrorClassifier.IsTransient(SqlExceptionFactory.Create(40613))); + Assert.True(AuditIngestErrorClassifier.IsTransient(new TimeoutException())); + Assert.True(AuditIngestErrorClassifier.IsTransient(new TaskCanceledException())); + Assert.True(AuditIngestErrorClassifier.IsTransient( + new InvalidOperationException("wrapped", new System.Net.Sockets.SocketException()))); + Assert.False(AuditIngestErrorClassifier.IsTransient(SqlExceptionFactory.Create(2628))); + Assert.False(AuditIngestErrorClassifier.IsTransient(new InvalidOperationException("bug"))); + } + + private sealed class OneRowClient : IPullAuditEventsClient + { + private readonly AuditEvent _row; + public OneRowClient(AuditEvent row) => _row = row; + public System.Collections.Concurrent.ConcurrentQueue Since { get; } = new(); + public Task PullAsync( + string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct) + { + Since.Enqueue(sinceUtc); + return Task.FromResult(new ZB.MOM.WW.ScadaBridge.Commons.Messages.Integration.PullAuditEventsResponse( + new[] { _row }, MoreAvailable: false)); + } + } + + private sealed class OneSite : ISiteEnumerator + { + public Task> EnumerateAsync(CancellationToken ct = default) => + Task.FromResult>(new[] { new SiteEntry("site-a", "http://a:8083") }); + } + + /// + /// A row that keeps hitting a TRANSIENT database fault is retried with the cursor held back for + /// far longer than the permanent-abandon threshold, and is never abandoned. + /// + [Fact] + public void Reconciliation_TransientFaults_NeverCountTowardAbandonment() + { + var row = NewEvent(); + var repo = Substitute.For(); + repo.InsertManyIfNotExistsAsync(Arg.Any>(), Arg.Any(), Arg.Any()) + .ThrowsAsync(SqlExceptionFactory.Create(-2)); + repo.InsertIfNotExistsAsync(Arg.Any(), Arg.Any()) + .ThrowsAsync(SqlExceptionFactory.Create(-2)); + var client = new OneRowClient(row); + + var actor = Sys.ActorOf(Props.Create(() => new SiteAuditReconciliationActor( + new OneSite(), client, StubAuditRepositoryProvider.Build(repo), + Microsoft.Extensions.Options.Options.Create(new SiteAuditReconciliationOptions + { + ReconciliationIntervalOverride = TimeSpan.FromMilliseconds(30), + }), + NullLogger.Instance))); + + AwaitCondition(() => client.Since.Count >= 12, TimeSpan.FromSeconds(10)); + Watch(actor); + Sys.Stop(actor); + ExpectTerminated(actor); + + Assert.All(client.Since, s => Assert.Equal(DateTime.MinValue, s)); // cursor held every tick + repo.DidNotReceive().InsertIfNotExistsAsync( + Arg.Is(e => e.Action.EndsWith(".ReconciliationAbandoned", StringComparison.Ordinal)), + Arg.Any()); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/ReconciliationStatePruningTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/ReconciliationStatePruningTests.cs index b57f93ad..659c3802 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/ReconciliationStatePruningTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/ReconciliationStatePruningTests.cs @@ -116,9 +116,11 @@ public class ReconciliationStatePruningTests : TestKit Assert.True(probe.ExpectMsg(TimeSpan.FromSeconds(5)).Stalled); - sites.Sites = Array.Empty(); + // Not an empty list: an empty enumeration deliberately prunes nothing. + sites.Sites = new[] { new SiteEntry("survivor", "http://s:8083") }; - var cleared = probe.ExpectMsg(TimeSpan.FromSeconds(5)); + var cleared = probe.FishForMessage( + m => m.SiteId == "doomed", TimeSpan.FromSeconds(5)); Assert.Equal("doomed", cleared.SiteId); Assert.False(cleared.Stalled); diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterForwardStateTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterForwardStateTests.cs index 0f6c6ec6..bbeaa2c1 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterForwardStateTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterForwardStateTests.cs @@ -18,7 +18,6 @@ namespace ZB.MOM.WW.ScadaBridge.AuditLog.Tests.Site; 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() @@ -32,8 +31,7 @@ public class SqliteAuditWriterForwardStateTests : IDisposable private SqliteAuditWriter CreateWriter() => new( Options.Create(new SqliteAuditWriterOptions { DatabasePath = _dbPath }), NullLogger.Instance, - new FakeNodeIdentityProvider(), - failureCounter: _counter); + new FakeNodeIdentityProvider()); private static async Task WriteAsync(SqliteAuditWriter writer, DateTime occurredAtUtc) { @@ -132,13 +130,39 @@ public class SqliteAuditWriterForwardStateTests : IDisposable 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); + Assert.Equal(1, writer.RejectedPushRowCount); } - private sealed class CountingFailureCounter : IAuditWriteFailureCounter + /// + /// The served high-water mark survives a writer restart: rows served before the restart are + /// retired by the first cursor that covers them after it, instead of being stranded owed + /// (and counted on the backlog) until something happens to re-serve them. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task ServedHighWaterMark_SurvivesRestart_SoTheNextCursorStillRetires() { - private int _count; - public int Count => Volatile.Read(ref _count); - public void Increment() => Interlocked.Increment(ref _count); + var t = DateTime.UtcNow.AddMinutes(-5); + Guid abandoned; + IReadOnlyList served; + await using (var writer = CreateWriter()) + { + abandoned = await WriteAsync(writer, t); + await writer.MarkPushAbandonedAsync(new[] { abandoned }); + served = await writer.ReadPendingSinceAsync(DateTime.MinValue, 100); + Assert.Single(served); + } + + await using (var restarted = CreateWriter()) + { + Assert.True(restarted.MaxServedRowIdForTests > 0); + + var flipped = await restarted.MarkReconciledUpToAsync( + served[0].OccurredAtUtc.UtcDateTime, served[0].EventId.ToString()); + + Assert.Equal(1, flipped); + Assert.Equal("Reconciled", restarted.GetForwardStateForTests(abandoned)!.Value.State); + Assert.Equal(0, (await restarted.GetBacklogStatsAsync()).PendingCount); + } } } 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 76aaf0e3..ed737846 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 @@ -785,27 +785,33 @@ public class SiteAuditTelemetryActorTests : TestKit } /// - /// 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. + /// Review 2026-10-21 06/IMPL-1 (adversarial follow-up): only rows central EXPLICITLY lists in + /// RejectedEventIds get a refused attempt recorded. A row merely missing from the accepted + /// list (central DB outage / budget exhaustion) and a transport fault record nothing. /// /// A task that represents the asynchronous operation. [Fact] - public async Task Drain_RefusedRows_RecordRejectedAttempts_TransportFaultDoesNot() + public async Task Drain_OnlyExplicitlyRejectedRows_RecordRefusals() { var queue = Substitute.For(); var lifecycle = (ZB.MOM.WW.ScadaBridge.AuditLog.Site.ISiteAuditPushLifecycle)queue; var accepted = NewEvent(); - var refused = NewEvent(); + var poison = NewEvent(); + var unanswered = NewEvent(); + var rows = new[] { accepted, poison, unanswered }; queue.ReadPendingAsync(Arg.Any(), Arg.Any()) .Returns( - Task.FromResult>(new[] { accepted, refused }), - Task.FromResult>(new[] { accepted, refused }), + Task.FromResult>(rows), + Task.FromResult>(rows), + Task.FromResult>(rows), Task.FromResult>(Array.Empty())); + var explicitAck = AckAll(new[] { accepted }); + explicitAck.RejectedEventIds.Add(poison.EventId.ToString()); _client.IngestAuditEventsAsync(Arg.Any(), Arg.Any()) .Returns( _ => Task.FromException(new InvalidOperationException("transport down")), - _ => Task.FromResult(AckAll(new[] { accepted }))); + _ => Task.FromResult(AckAll(new[] { accepted })), // partial ack, no verdict (old central / DB outage) + _ => Task.FromResult(explicitAck)); Sys.ActorOf(Props.Create(() => new SiteAuditTelemetryActor( queue, _client, Opts(), NullLogger.Instance, @@ -813,13 +819,12 @@ public class SiteAuditTelemetryActorTests : TestKit await AwaitAssertAsync(async () => await lifecycle.Received(1).RecordRejectedPushAttemptsAsync( - Arg.Is>(g => g.Count == 1 && g[0] == refused.EventId), + Arg.Is>(g => g.Count == 1 && g[0] == poison.EventId), 10, Arg.Any()), - TimeSpan.FromSeconds(10)); - await queue.Received(1).MarkForwardedAsync( - Arg.Is>(g => g.Count == 1 && g[0] == accepted.EventId), - Arg.Any()); + TimeSpan.FromSeconds(15)); + await lifecycle.Received(1).RecordRejectedPushAttemptsAsync( + Arg.Any>(), Arg.Any(), Arg.Any()); } private sealed class CapturingLogger : Microsoft.Extensions.Logging.ILogger diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/TestSupport/SqlExceptionFactory.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/TestSupport/SqlExceptionFactory.cs new file mode 100644 index 00000000..3ceb8217 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/TestSupport/SqlExceptionFactory.cs @@ -0,0 +1,35 @@ +using System.Reflection; +using Microsoft.Data.SqlClient; + +namespace ZB.MOM.WW.ScadaBridge.AuditLog.Tests.TestSupport; + +/// +/// Fabricates a with a given error number via the driver's internal +/// CreateException factory (the type has no public constructor) — same technique as +/// ExternalSystemGateway.Tests' FabricateSqlException. +/// +internal static class SqlExceptionFactory +{ + /// Creates a carrying one error with . + /// The SQL Server error number. + /// The error message. + /// The fabricated exception. + public static SqlException Create(int number, string message = "simulated") + { + var errorCtor = typeof(SqlError).GetConstructors(BindingFlags.Instance | BindingFlags.NonPublic) + .First(c => c.GetParameters().Length == 8 && c.GetParameters()[7].ParameterType == typeof(Exception)); + var sqlError = (SqlError)errorCtor.Invoke( + new object?[] { number, (byte)0, (byte)0, "server", message, "procedure", 0, null }); + + var collection = (SqlErrorCollection)typeof(SqlErrorCollection) + .GetConstructors(BindingFlags.Instance | BindingFlags.NonPublic).First() + .Invoke(Array.Empty()); + typeof(SqlErrorCollection).GetMethod("Add", BindingFlags.Instance | BindingFlags.NonPublic)! + .Invoke(collection, new object?[] { sqlError }); + + var create = typeof(SqlException).GetMethod("CreateException", + BindingFlags.Static | BindingFlags.NonPublic, + new[] { typeof(SqlErrorCollection), typeof(string) })!; + return (SqlException)create.Invoke(null, new object?[] { collection, "6.0.0" })!; + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/CentralControlDtoMapperTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/CentralControlDtoMapperTests.cs index 93025d32..9fb89fde 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/CentralControlDtoMapperTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/CentralControlDtoMapperTests.cs @@ -259,6 +259,23 @@ public class CentralControlDtoMapperTests Assert.Equal(ids, roundTripped); } + /// Verifies the additive rejected-id list round-trips and reads empty when absent. + [Fact] + public void IngestAck_RejectedIds_RoundTrip_AndAbsentFieldReadsEmpty() + { + // Review 2026-10-21 06/IMPL-1: additive field 2. An older central never sets it, so the + // site must read an EMPTY list (never a refusal) from a pre-field ack. + var accepted = new[] { Guid.NewGuid() }; + var rejected = new[] { Guid.NewGuid(), Guid.NewGuid() }; + + var wire = Wire(CentralControlDtoMapper.ToIngestAck(accepted, rejected)); + + Assert.Equal(accepted, CentralControlDtoMapper.FromIngestAck(wire)); + Assert.Equal(rejected, CentralControlDtoMapper.RejectedFromIngestAck(wire)); + Assert.Empty(CentralControlDtoMapper.RejectedFromIngestAck( + Wire(CentralControlDtoMapper.ToIngestAck(accepted)))); + } + /// Verifies that an ingest acknowledgement with nothing accepted stays empty. [Fact] public void IngestAck_RoundTrip_NothingAccepted_StaysEmpty()