Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 34227991ea | |||
| 9d925c3347 | |||
| 8c6fc2f886 |
@@ -134,8 +134,13 @@ Other peers in the `scadaproj` family (see `scadaproj/CLAUDE.md` for details): `
|
|||||||
- **Operational constraints (read before upgrading a site pair):** stop and start both nodes TOGETHER — rolling one at a time is no longer supported, since the legacy `SfBufferSnapshot` compatibility handler went with the replicator. And a node offline longer than `LocalDb:Replication:TombstoneRetention` (default 7 days) can resurrect deleted rows on rejoin. See `docs/deployment/topology-guide.md`.
|
- **Operational constraints (read before upgrading a site pair):** stop and start both nodes TOGETHER — rolling one at a time is no longer supported, since the legacy `SfBufferSnapshot` compatibility handler went with the replicator. And a node offline longer than `LocalDb:Replication:TombstoneRetention` (default 7 days) can resurrect deleted rows on rejoin. See `docs/deployment/topology-guide.md`.
|
||||||
- `LocalDb:Replication:MaxBatchSize` batches by ROW COUNT, not bytes, against a 4 MB gRPC cap — the rig pins it to **16** (~70 KB worst-case `config_json` x 16 ~= 1.1 MB). The 500 default would allow ~35 MB.
|
- `LocalDb:Replication:MaxBatchSize` batches by ROW COUNT, not bytes, against a 4 MB gRPC cap — the rig pins it to **16** (~70 KB worst-case `config_json` x 16 ~= 1.1 MB). The 500 default would allow ~35 MB.
|
||||||
- All timestamps are UTC throughout the system.
|
- All timestamps are UTC throughout the system.
|
||||||
- Inter-cluster communication uses two transports: ClusterClient for command/control (deployments, lifecycle, subscribe/unsubscribe handshake, snapshots) and gRPC server-streaming for real-time data (attribute values, alarm states). Both CentralCommunicationActor and SiteCommunicationActor registered with receptionist. Central creates one ClusterClient per site using NodeA/NodeB as contact points. Sites configure multiple central contact points for failover. Addresses cached in CentralCommunicationActor, refreshed periodically (60s) and on admin changes. Heartbeats serve health monitoring only.
|
- Inter-cluster communication uses **three** transports, not two: **ClusterClient** for command/control (deployments, lifecycle, subscribe/unsubscribe handshake, snapshots); **gRPC** server-streaming for real-time data (attribute values, alarm states); and **plain token-gated HTTP** for the deployment config itself — notify-and-fetch, the site pulls the config from `DeploymentConfigEndpoints` (`ManagementService/DeploymentConfigEndpoints.cs`) with an `X-Deployment-Token` header, `AllowAnonymous` with the per-deployment token as the entire security boundary. Both CentralCommunicationActor and SiteCommunicationActor registered with receptionist (**per node, not as a singleton** — contact rotation reaches whichever node answers). Central creates one ClusterClient per site using NodeA/NodeB as contact points. Sites configure multiple central contact points for failover. Addresses cached in CentralCommunicationActor, refreshed periodically (60s) and on admin changes. Heartbeats serve health monitoring only. **Discovery is asymmetric by design:** central discovers sites from the *database* (`Site.NodeAAddress`/`NodeBAddress`, refreshable at runtime), sites discover central from *appsettings* (`ScadaBridge:Communication:CentralContactPoints`, static — restart required). **Central never buffers for an unreachable site** — the send is dropped with a warning and the caller's Ask times out; a `ConnectionStateChanged` mechanism built for this was deleted as dead code.
|
||||||
- gRPC streaming channel: SiteStreamGrpcServer on each site node (Kestrel HTTP/2, port 8083); central creates per-site SiteStreamGrpcClient via SiteStreamGrpcClientFactory. Site entity has GrpcNodeAAddress/GrpcNodeBAddress fields. Proto: sitestream.proto with SiteStreamService, SiteStreamEvent (oneof: AttributeValueUpdate, AlarmStateUpdate). DebugStreamEvent message removed (no longer flows through ClusterClient).
|
- **All clusters share ONE ActorSystem name**, `"scadabridge"` — hardcoded at `AkkaHostedService.cs:191`. Central and each site are separate clusters *only* by seed-node partitioning. This is required, not incidental: Akka.Remote address matching means a ClusterClient could not reach a differently-named system.
|
||||||
|
- **`ActiveNodeEvaluator.SelfIsOldestUp` is THE single definition of "active node"** (`Communication/ClusterState/ActiveNodeEvaluator.cs`) — the **oldest Up member** in a role scope, and explicitly **never `cluster.State.Leader`**: leadership (lowest address) is an Akka-internal concept that diverges from singleton placement permanently once the original first node restarts and rejoins, and both sides claim it during a partition. The equivalence *oldest-Up == where `ClusterSingletonManager` places singletons* **is** the design. `ClusterActivityEvaluator.SelfIsOldest`, the S&F delivery gate, `/health/active` and the heartbeat `IsActive` stamp all delegate here.
|
||||||
|
- Site nodes carry **two Akka roles**: the base `Site` plus a site-specific `site-{SiteId}` (`BuildRoles`, `AkkaHostedService.cs:386`). Singletons scope to the **site-specific** role.
|
||||||
|
- **Neither inter-cluster transport is authenticated or encrypted.** Akka remoting sets no `enable-ssl`, no secure cookie, no `trusted-selection-paths`; the gRPC listener is **h2c** and `LocalDbSyncAuthInterceptor` gates *only* `/localdb_sync.v1.LocalDbSync/`, so the whole `SiteStreamService` surface — including the `PullAuditEvents`/`PullSiteCalls` RPCs that return audit rows — is callable by anyone who can reach :8083. The boundary assumes a trusted network. (The HTTP fetch token and the LocalDb sync interceptor are the only authenticated pieces.)
|
||||||
|
- **Akka frame size is the default 128 KB with `log-frame-size-exceeding` off**, and no custom serializer is configured — so payload carried over ClusterClient is JSON-escaped a second time by the default Newtonsoft serializer, roughly doubling it. Over the limit the transport drops **that one message** without tearing down the association (heartbeats keep flowing, the site still reports healthy) and the central Ask simply times out. See `docs/known-issues/2026-06-26-deploy-config-exceeds-akka-frame-size.md`; `DeployArtifactsCommand` still carries payload and remains exposed.
|
||||||
|
- gRPC streaming channel — **note the direction is inverted from the data flow**: data moves site→central, but each **site node hosts the gRPC server** (`SiteStreamGrpcServer`, Kestrel h2c, port 8083, mapped **only in the Site branch** of `Program.cs`) and **central is the client**, dialling in. There is **no gRPC server on central at all**, which is why the two `Ingest*` unary RPCs — documented as a "central-side ingest surface" — are dead in practice (acknowledged in `AkkaHostedService.cs:510-519`); sites reach central over ClusterClient instead. Central creates per-site `SiteStreamGrpcClient` via `SiteStreamGrpcClientFactory`, keyed **`(siteId, endpoint)`** — the key was widened from site-only to fix an arch-review High where one session's NodeA→NodeB flip disposed a channel another session was still using. Site entity has GrpcNodeAAddress/GrpcNodeBAddress fields. Proto: `sitestream.proto`, **6 RPCs** (2 server-streaming: `SubscribeInstance`, `SubscribeSite` (site-wide, alarm-only); 4 unary: `IngestAuditEvents`, `IngestCachedTelemetry`, `PullAuditEvents`, `PullSiteCalls`), `SiteStreamEvent` (oneof: AttributeValueUpdate, AlarmStateUpdate). Field numbers are never reused; evolution is additive only (`AlarmStateUpdate` grew 7→23 fields for the native-alarm mirror). Generated C# is **vendored** under `Communication/SiteStreamGrpc/` with the `<Protobuf>` include commented out — regeneration is a manual toggle-build-copy-untoggle. DebugStreamEvent message removed (no longer flows through ClusterClient).
|
||||||
- Native alarms: a read-only mirror of native alarms from OPC UA Alarms & Conditions servers and the MxAccess Gateway, unified onto an A&C-style condition model (`AlarmConditionState`: orthogonal Active/Acked/Confirmed/Shelved/Suppressed + 0–1000 severity) plus an `AlarmKind` discriminator (Computed/NativeOpcUa/NativeMxAccess). New DCL capability seam `IAlarmSubscribableConnection` (implemented by the OPC UA and MxGateway adapters); the `DataConnectionActor` opens ONE alarm feed per connection and routes transitions to instances by source-object reference. A `NativeAlarmActor` (peer to the computed `AlarmActor` under `InstanceActor`) mirrors one source binding: snapshot atomic-swap on (re)subscribe, retention (drops once inactive+acked), per-source cap, and site SQLite persistence (`native_alarm_state`, survives failover, cleared on redeploy/undeploy — mirrors static overrides). State streams to central over the additively-enriched gRPC `AlarmStateUpdate` (the existing computed `AlarmStateChanged` was enriched additively) and seeds via the DebugView snapshot. Authoring: `TemplateNativeAlarmSource` / `InstanceNativeAlarmSourceOverride` entities flatten to `ResolvedNativeAlarmSource` (inherit/compose/override); management commands + ManagementActor handlers + CLI (`template/instance native-alarm-source`) + Central UI (template editor tab + instance override panel) + enriched DebugView alarm table. Read-only — no ack-back; no central tables.
|
- Native alarms: a read-only mirror of native alarms from OPC UA Alarms & Conditions servers and the MxAccess Gateway, unified onto an A&C-style condition model (`AlarmConditionState`: orthogonal Active/Acked/Confirmed/Shelved/Suppressed + 0–1000 severity) plus an `AlarmKind` discriminator (Computed/NativeOpcUa/NativeMxAccess). New DCL capability seam `IAlarmSubscribableConnection` (implemented by the OPC UA and MxGateway adapters); the `DataConnectionActor` opens ONE alarm feed per connection and routes transitions to instances by source-object reference. A `NativeAlarmActor` (peer to the computed `AlarmActor` under `InstanceActor`) mirrors one source binding: snapshot atomic-swap on (re)subscribe, retention (drops once inactive+acked), per-source cap, and site SQLite persistence (`native_alarm_state`, survives failover, cleared on redeploy/undeploy — mirrors static overrides). State streams to central over the additively-enriched gRPC `AlarmStateUpdate` (the existing computed `AlarmStateChanged` was enriched additively) and seeds via the DebugView snapshot. Authoring: `TemplateNativeAlarmSource` / `InstanceNativeAlarmSourceOverride` entities flatten to `ResolvedNativeAlarmSource` (inherit/compose/override); management commands + ManagementActor handlers + CLI (`template/instance native-alarm-source`) + Central UI (template editor tab + instance override panel) + enriched DebugView alarm table. Read-only — no ack-back; no central tables.
|
||||||
- OPC UA / MxGateway UX (M7): operator **Alarm Summary** page (`/monitoring/alarms`, RequireDeployment, read-only) fans out the existing per-instance `DebugViewSnapshot` Ask (SemaphoreSlim-capped, partial-results tolerant) and aggregates client-side — no central alarm store; shared `AlarmStateBadges` component. **Aggregated live stream shipped 2026-07-10** (`docs/plans/2026-07-10-aggregated-live-alarm-stream-plan.md`): a transient in-memory per-site central live cache (`ISiteAlarmLiveCache`) fed by a site-wide, alarm-only `SubscribeSite` gRPC stream (seed-then-stream), pushing near-real-time deltas to the page over the Blazor circuit with the 15s poll kept as fallback + NotReporting authority — still no persisted central alarm store. OPC UA node browser gains `BrowseNext` continuation paging ("Load more"), a bounded recursive address-space **search** (`IAddressSpaceSearchable` seam; depth + result caps; substring on DisplayName/path), and **type-info** (DataType/ValueRank/Writable on `BrowseNode` for Variables). Attribute-override **CSV bulk import** (`OverrideCsvParser`, all-or-nothing) via InstanceConfigure `InputFile` + CLI `instance import-overrides --file` (native-alarm-source-override CSV deferred). **Verify-endpoint** probe (temporary `RealOpcUaClient`, short timeout, captures an untrusted server cert but NEVER trusts it) + **site-local cert trust**: per-node `CertStoreActor` (runs on every site node, not a singleton) writing the `.der` into the node's OPC UA trusted-peer PKI store; DeploymentManager broadcasts `TrustServerCertCommand`/`RemoveServerCertCommand` to BOTH site nodes so PKI stores stay consistent across failover; Admin-gated cert-management UI (`/design/connections/{id}/certificates`). No central persistence of cert trust (follow-up).
|
- OPC UA / MxGateway UX (M7): operator **Alarm Summary** page (`/monitoring/alarms`, RequireDeployment, read-only) fans out the existing per-instance `DebugViewSnapshot` Ask (SemaphoreSlim-capped, partial-results tolerant) and aggregates client-side — no central alarm store; shared `AlarmStateBadges` component. **Aggregated live stream shipped 2026-07-10** (`docs/plans/2026-07-10-aggregated-live-alarm-stream-plan.md`): a transient in-memory per-site central live cache (`ISiteAlarmLiveCache`) fed by a site-wide, alarm-only `SubscribeSite` gRPC stream (seed-then-stream), pushing near-real-time deltas to the page over the Blazor circuit with the 15s poll kept as fallback + NotReporting authority — still no persisted central alarm store. OPC UA node browser gains `BrowseNext` continuation paging ("Load more"), a bounded recursive address-space **search** (`IAddressSpaceSearchable` seam; depth + result caps; substring on DisplayName/path), and **type-info** (DataType/ValueRank/Writable on `BrowseNode` for Variables). Attribute-override **CSV bulk import** (`OverrideCsvParser`, all-or-nothing) via InstanceConfigure `InputFile` + CLI `instance import-overrides --file` (native-alarm-source-override CSV deferred). **Verify-endpoint** probe (temporary `RealOpcUaClient`, short timeout, captures an untrusted server cert but NEVER trusts it) + **site-local cert trust**: per-node `CertStoreActor` (runs on every site node, not a singleton) writing the `.der` into the node's OPC UA trusted-peer PKI store; DeploymentManager broadcasts `TrustServerCertCommand`/`RemoveServerCertCommand` to BOTH site nodes so PKI stores stay consistent across failover; Admin-gated cert-management UI (`/design/connections/{id}/certificates`). No central persistence of cert trust (follow-up).
|
||||||
|
|
||||||
@@ -210,9 +215,11 @@ Other peers in the `scadaproj` family (see `scadaproj/CLAUDE.md` for details): `
|
|||||||
### Cluster & Failover
|
### Cluster & Failover
|
||||||
- Keep-oldest split-brain resolver with `down-if-alone = on`, 15s stable-after.
|
- Keep-oldest split-brain resolver with `down-if-alone = on`, 15s stable-after.
|
||||||
- Both nodes are seed nodes. `min-nr-of-members = 1`.
|
- Both nodes are seed nodes. `min-nr-of-members = 1`.
|
||||||
- Failure detection: 2s heartbeat, 10s threshold. Total failover ~25s.
|
- Failure detection: 2s heartbeat, 10s threshold. Total failover ~25s (drill-measured 27s, 0 routing blips — `docker/failover-drill.sh`).
|
||||||
- CoordinatedShutdown for graceful singleton handover.
|
- CoordinatedShutdown for graceful singleton handover.
|
||||||
- Automatic dual-node recovery from persistent storage.
|
- Automatic dual-node recovery from persistent storage.
|
||||||
|
- **Active/standby is decided by `ActiveNodeEvaluator.SelfIsOldestUp`, never by cluster leadership** — see the Architecture note above. `/health/active` is **central-only** (site nodes map no `/health/*` at all) and backs both Traefik's active-node routing and `IActiveNodeGate`, so the proxy and the Inbound API always agree on which node is active. Central never needs to know which *site* node is active: ClusterClient contact rotation reaches either receptionist and the site-internal `ClusterSingletonProxy` lands the work on the active node for free. The **exception is gRPC**, which picks `GrpcNodeAAddress`/`GrpcNodeBAddress` explicitly and flips on error.
|
||||||
|
- **KNOWN GAP — two-node keep-oldest has a total-outage hole.** Only the *first* seed listed in `Cluster:SeedNodes` may self-join to form a new cluster, so a lone restarted non-first-seed node (first seed still down) loops on `InitJoin` forever — never `Up`, never routable. After the oldest/active node crashes, the younger survivor self-downs and **cannot re-bootstrap alone**; recovery is operator-driven. `failover-drill.sh` has `DRILL_MODE=active` specifically *"to make the registered gap observable — not to pretend it is covered."* Closing it is the registered deferred keep-oldest topology/strategy decision. See `docs/requirements/Component-ClusterInfrastructure.md:113`.
|
||||||
|
|
||||||
### UI & Monitoring
|
### UI & Monitoring
|
||||||
- Central UI: Blazor Server (ASP.NET Core + SignalR) with Bootstrap CSS. No third-party component frameworks (no Blazorise, MudBlazor, Radzen, etc.). Build custom Blazor components for tables, grids, forms, etc.
|
- Central UI: Blazor Server (ASP.NET Core + SignalR) with Bootstrap CSS. No third-party component frameworks (no Blazorise, MudBlazor, Radzen, etc.). Build custom Blazor components for tables, grids, forms, etc.
|
||||||
|
|||||||
@@ -108,9 +108,9 @@
|
|||||||
<PackageVersion Include="ZB.MOM.WW.Secrets.Abstractions" Version="0.2.3" />
|
<PackageVersion Include="ZB.MOM.WW.Secrets.Abstractions" Version="0.2.3" />
|
||||||
<PackageVersion Include="ZB.MOM.WW.Secrets.Ui" Version="0.2.3" />
|
<PackageVersion Include="ZB.MOM.WW.Secrets.Ui" Version="0.2.3" />
|
||||||
<PackageVersion Include="ZB.MOM.WW.Secrets.Replicator.SqlServer" Version="0.2.3" />
|
<PackageVersion Include="ZB.MOM.WW.Secrets.Replicator.SqlServer" Version="0.2.3" />
|
||||||
<PackageVersion Include="ZB.MOM.WW.LocalDb" Version="0.1.1" />
|
<PackageVersion Include="ZB.MOM.WW.LocalDb" Version="0.1.3" />
|
||||||
<PackageVersion Include="ZB.MOM.WW.LocalDb.Replication" Version="0.1.1" />
|
<PackageVersion Include="ZB.MOM.WW.LocalDb.Replication" Version="0.1.3" />
|
||||||
<PackageVersion Include="ZB.MOM.WW.LocalDb.Contracts" Version="0.1.1" />
|
<PackageVersion Include="ZB.MOM.WW.LocalDb.Contracts" Version="0.1.3" />
|
||||||
</ItemGroup>
|
</ItemGroup>
|
||||||
|
|
||||||
<!--
|
<!--
|
||||||
|
|||||||
@@ -1,55 +1,8 @@
|
|||||||
# Cached-telemetry drain hot-loops forever on a row whose tracking snapshot is gone
|
# Cached-telemetry drain hot-loops forever on a row whose tracking snapshot is gone
|
||||||
|
|
||||||
**Date:** 2026-07-20 · **Status:** RESOLVED (2026-07-20) · **Severity:** High — revised up from Medium on investigation (permanent stall of the cached-telemetry path, not just a log flood)
|
**Date:** 2026-07-20 · **Status:** OPEN · **Severity:** Medium (log flood + wasted I/O; no data loss)
|
||||||
· **Area:** AuditLog / Site Telemetry
|
· **Area:** AuditLog / Site Telemetry
|
||||||
|
|
||||||
## Resolution (2026-07-20)
|
|
||||||
|
|
||||||
Fixed by letting an unresolvable row LEAVE the queue.
|
|
||||||
`SiteAuditTelemetryActor.OnCachedDrainAsync` now abandons the operational half of a cached row
|
|
||||||
whose tracking snapshot is still unresolvable after a grace period
|
|
||||||
(`SiteAuditTelemetryOptions.CachedTrackingGraceSeconds`, default **300 s**) and marks it
|
|
||||||
`Forwarded`. A row with no `CorrelationId` at all is abandoned immediately — it can never resolve.
|
|
||||||
|
|
||||||
**Marking `Forwarded` does not drop the audit data.** That state's role in this machine is "no
|
|
||||||
longer owed by the drain, still eligible for reconciliation", which is exactly the situation:
|
|
||||||
`ISiteAuditQueue.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 — unrecoverable regardless, since the tracking row it would
|
|
||||||
have been built from no longer exists.
|
|
||||||
|
|
||||||
Three further behaviours, each deliberate:
|
|
||||||
|
|
||||||
- **Inside the grace window the row is still retried.** A missing snapshot is normally a brief
|
|
||||||
write race (the audit row lands microseconds before the tracking row); abandoning on the first
|
|
||||||
failed lookup would discard the operational half of every cached call that lost that race.
|
|
||||||
- **A tracking-store THROW never abandons.** A throw is a store fault (locked, corrupt,
|
|
||||||
mid-restore), not a verdict about the row. Those rows stay `Pending` however old they are.
|
|
||||||
- **Logging is per drain pass, not per row.** One summary line each for abandoned, lookup-failed
|
|
||||||
(with the first exception attached) and deferred rows. The deferred case dropped to Debug — being
|
|
||||||
inside the grace window is normal operation, not a warning.
|
|
||||||
|
|
||||||
### Severity was revised UP during the fix
|
|
||||||
|
|
||||||
The original write-up called this a log flood with "no data loss", which understated it. The queue
|
|
||||||
is read **oldest-first with a fixed `BatchSize` (default 256)**, so once a batch's worth of
|
|
||||||
permanently-unresolvable rows collects at the head, every drain re-reads exactly those rows, fails
|
|
||||||
identically, and **never reaches the newer rows behind them**. The cached-telemetry path stalls
|
|
||||||
permanently. The audit half still reached central by reconciliation, so "no data loss" held — but
|
|
||||||
"Medium" did not.
|
|
||||||
|
|
||||||
**Fixed in:** `SiteAuditTelemetryActor.cs` (`AbandonUnresolvableAsync` + the rewritten skip
|
|
||||||
branches), `SiteAuditTelemetryOptions.cs` (`CachedTrackingGraceSeconds`).
|
|
||||||
**Tests:** `SiteAuditTelemetryActorTests` — `CachedDrain_OrphanRow_PastGrace_IsAbandoned_...`,
|
|
||||||
`..._InsideGrace_IsRetried_NotAbandoned`, `..._FullBatchOfUnresolvableRows_DoesNotStarveTheQueue`,
|
|
||||||
`..._TrackingStoreThrow_DoesNotAbandonTheRow`. The pre-existing
|
|
||||||
`CachedDrain_OrphanRow_NoTrackingSnapshot_IsSkipped_DoesNotCrash` was **retargeted, not deleted** —
|
|
||||||
it had pinned the defect ("skipped and stays Pending"), so its orphan assertion is inverted while
|
|
||||||
the half that still holds is kept verbatim. Verified non-vacuous: with abandonment disabled the two
|
|
||||||
abandonment tests go red; the two must-NOT-abandon guards stay green in both directions.
|
|
||||||
|
|
||||||
The diagnosis below is retained as the record of how it was found.
|
|
||||||
|
|
||||||
## Summary
|
## Summary
|
||||||
|
|
||||||
`SiteAuditTelemetryActor`'s cached-telemetry drain reads Pending audit rows, looks up each row's
|
`SiteAuditTelemetryActor`'s cached-telemetry drain reads Pending audit rows, looks up each row's
|
||||||
|
|||||||
@@ -261,18 +261,6 @@ public class SiteAuditTelemetryActor : ReceiveActor
|
|||||||
|
|
||||||
var batch = new CachedTelemetryBatch();
|
var batch = new CachedTelemetryBatch();
|
||||||
var emittedEventIds = new List<Guid>(pending.Count);
|
var emittedEventIds = new List<Guid>(pending.Count);
|
||||||
// Rows whose operational half can never be built. They are marked
|
|
||||||
// Forwarded so they LEAVE this queue — see AbandonUnresolvableAsync
|
|
||||||
// for why that is safe and why leaving them Pending is not.
|
|
||||||
var abandonedEventIds = new List<Guid>();
|
|
||||||
// Counted, not logged per row: a whole batch can fail identically
|
|
||||||
// (tracking store down, tracking retention elapsed), and one line
|
|
||||||
// per row per drain is what turned this into a log flood.
|
|
||||||
var deferredNoSnapshot = 0;
|
|
||||||
var lookupFailures = 0;
|
|
||||||
Exception? firstLookupFailure = null;
|
|
||||||
var graceSeconds = Math.Max(0, _options.CachedTrackingGraceSeconds);
|
|
||||||
var abandonBefore = DateTime.UtcNow - TimeSpan.FromSeconds(graceSeconds);
|
|
||||||
|
|
||||||
foreach (var auditRow in pending)
|
foreach (var auditRow in pending)
|
||||||
{
|
{
|
||||||
@@ -280,9 +268,13 @@ public class SiteAuditTelemetryActor : ReceiveActor
|
|||||||
{
|
{
|
||||||
// CorrelationId carries the TrackedOperationId for cached
|
// CorrelationId carries the TrackedOperationId for cached
|
||||||
// rows — see CachedCallLifecycleBridge.BuildPacket. Without
|
// rows — see CachedCallLifecycleBridge.BuildPacket. Without
|
||||||
// it there is nothing to look up, ever, so this row is
|
// it we can't look up the tracking row; log + skip so the
|
||||||
// abandoned immediately rather than after the grace period.
|
// bad row doesn't block the rest of the batch. The audit
|
||||||
abandonedEventIds.Add(auditRow.EventId);
|
// row stays Pending (still not in emittedEventIds) and
|
||||||
|
// central reconciliation will pick it up.
|
||||||
|
_logger.LogWarning(
|
||||||
|
"Cached-telemetry drain: audit row {EventId} ({Action}) has no CorrelationId; skipping.",
|
||||||
|
auditRow.EventId, auditRow.Action);
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -296,32 +288,24 @@ public class SiteAuditTelemetryActor : ReceiveActor
|
|||||||
catch (Exception ex)
|
catch (Exception ex)
|
||||||
{
|
{
|
||||||
// A tracking-store throw must NOT abort the rest of the
|
// A tracking-store throw must NOT abort the rest of the
|
||||||
// batch — the audit half is best-effort. A throw is a
|
// batch — the audit half is best-effort. Log and skip
|
||||||
// STORE fault (locked, corrupt) rather than a verdict about
|
// this row; it stays Pending for the next drain.
|
||||||
// this row, so the row is never abandoned on this path: it
|
_logger.LogWarning(ex,
|
||||||
// stays Pending and retries once the store recovers.
|
"Cached-telemetry drain: tracking lookup threw for {EventId} (TrackedOperationId {Tid}, sqlite {SqliteError}); skipping.",
|
||||||
lookupFailures++;
|
auditRow.EventId, auditRow.CorrelationId, SqliteErrorCodes.Describe(ex));
|
||||||
firstLookupFailure ??= ex;
|
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
|
||||||
if (snapshot is null)
|
if (snapshot is null)
|
||||||
{
|
{
|
||||||
// No tracking row. Within the grace window this is the
|
// No tracking row — possible if the audit row is older
|
||||||
// ordinary write race (the audit row landed first), so
|
// than the tracking retention window, or the tracking
|
||||||
// retry. Past it the snapshot is gone for good — tracking
|
// store was reset. The audit half remains valid and will
|
||||||
// retention elapsed, or the two stores were reset
|
// be picked up by central reconciliation; skip the
|
||||||
// independently — and retrying forever would wedge the
|
// combined push for this row.
|
||||||
// queue behind these rows.
|
_logger.LogWarning(
|
||||||
if (auditRow.OccurredAtUtc <= abandonBefore)
|
"Cached-telemetry drain: no tracking snapshot for {EventId} (TrackedOperationId {Tid}); skipping.",
|
||||||
{
|
auditRow.EventId, auditRow.CorrelationId);
|
||||||
abandonedEventIds.Add(auditRow.EventId);
|
|
||||||
}
|
|
||||||
else
|
|
||||||
{
|
|
||||||
deferredNoSnapshot++;
|
|
||||||
}
|
|
||||||
|
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -330,33 +314,11 @@ public class SiteAuditTelemetryActor : ReceiveActor
|
|||||||
emittedEventIds.Add(auditRow.EventId);
|
emittedEventIds.Add(auditRow.EventId);
|
||||||
}
|
}
|
||||||
|
|
||||||
if (lookupFailures > 0)
|
|
||||||
{
|
|
||||||
_logger.LogWarning(firstLookupFailure,
|
|
||||||
"Cached-telemetry drain: tracking lookup failed for {Count} of {Total} rows " +
|
|
||||||
"(sqlite {SqliteError}); they stay Pending and retry next drain. First failure attached.",
|
|
||||||
lookupFailures, pending.Count, SqliteErrorCodes.Describe(firstLookupFailure!));
|
|
||||||
}
|
|
||||||
|
|
||||||
if (deferredNoSnapshot > 0)
|
|
||||||
{
|
|
||||||
_logger.LogDebug(
|
|
||||||
"Cached-telemetry drain: {Count} row(s) have no tracking snapshot yet and are inside the " +
|
|
||||||
"{Grace}s grace window; retrying next drain.",
|
|
||||||
deferredNoSnapshot, graceSeconds);
|
|
||||||
}
|
|
||||||
|
|
||||||
if (abandonedEventIds.Count > 0)
|
|
||||||
{
|
|
||||||
await AbandonUnresolvableAsync(abandonedEventIds, graceSeconds, ct)
|
|
||||||
.ConfigureAwait(false);
|
|
||||||
}
|
|
||||||
|
|
||||||
if (batch.Packets.Count == 0)
|
if (batch.Packets.Count == 0)
|
||||||
{
|
{
|
||||||
// Nothing resolvable in this read. Any permanently-unresolvable
|
// Every row in this read was skipped (no CorrelationId / no
|
||||||
// rows have just been marked Forwarded above, so the next drain
|
// tracking snapshot). Leave them Pending and try again next
|
||||||
// sees past them rather than re-reading the same head of queue.
|
// drain — the underlying race normally resolves on its own.
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -396,59 +358,6 @@ public class SiteAuditTelemetryActor : ReceiveActor
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
|
||||||
/// Marks cached rows whose operational half can never be built as
|
|
||||||
/// Forwarded, so they leave the cached-drain queue.
|
|
||||||
/// </summary>
|
|
||||||
/// <remarks>
|
|
||||||
/// <para>
|
|
||||||
/// <b>Why they must leave.</b> Leaving an unresolvable row Pending — the
|
|
||||||
/// previous behaviour — is not a harmless skip. The queue is read
|
|
||||||
/// oldest-first with a fixed <c>BatchSize</c>, 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).
|
|
||||||
/// </para>
|
|
||||||
/// <para>
|
|
||||||
/// <b>Why Forwarded is the honest state.</b> 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:
|
|
||||||
/// <c>ReadPendingSinceAsync</c> 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
|
|
||||||
/// (<c>SiteCalls</c>) half — and that is unrecoverable regardless, because
|
|
||||||
/// the tracking row it would have been built from no longer exists.
|
|
||||||
/// </para>
|
|
||||||
/// <para>
|
|
||||||
/// 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.
|
|
||||||
/// </para>
|
|
||||||
/// </remarks>
|
|
||||||
private async Task AbandonUnresolvableAsync(
|
|
||||||
IReadOnlyList<Guid> eventIds, int graceSeconds, CancellationToken ct)
|
|
||||||
{
|
|
||||||
try
|
|
||||||
{
|
|
||||||
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 " +
|
|
||||||
"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 (Exception ex)
|
|
||||||
{
|
|
||||||
_logger.LogWarning(ex,
|
|
||||||
"Cached-telemetry drain: could not mark {Count} unresolvable row(s) Forwarded " +
|
|
||||||
"(sqlite {SqliteError}); they stay Pending and will be retried.",
|
|
||||||
eventIds.Count, SqliteErrorCodes.Describe(ex));
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private static AuditEventBatch BuildBatch(IReadOnlyList<AuditEvent> events)
|
private static AuditEventBatch BuildBatch(IReadOnlyList<AuditEvent> events)
|
||||||
{
|
{
|
||||||
var batch = new AuditEventBatch();
|
var batch = new AuditEventBatch();
|
||||||
|
|||||||
@@ -25,33 +25,4 @@ public sealed class SiteAuditTelemetryOptions
|
|||||||
/// Longer interval avoids hammering an idle SQLite + gRPC channel.
|
/// Longer interval avoids hammering an idle SQLite + gRPC channel.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
public int IdleIntervalSeconds { get; set; } = 30;
|
public int IdleIntervalSeconds { get; set; } = 30;
|
||||||
|
|
||||||
/// <summary>
|
|
||||||
/// How long a cached-telemetry audit row may go without a resolvable
|
|
||||||
/// <c>OperationTracking</c> snapshot before the drain ABANDONS its
|
|
||||||
/// operational half and marks the row Forwarded. Default: 300 s.
|
|
||||||
/// </summary>
|
|
||||||
/// <remarks>
|
|
||||||
/// <para>
|
|
||||||
/// A missing tracking snapshot is normally a brief write race — the audit
|
|
||||||
/// row lands microseconds before the tracking row — so the drain retries
|
|
||||||
/// within this grace period. Past it the snapshot is almost certainly gone
|
|
||||||
/// for good (tracking retention elapsed while central was unreachable, or
|
|
||||||
/// the two stores were reset independently), and retrying forever is
|
|
||||||
/// actively harmful: the queue is read oldest-first with a fixed
|
|
||||||
/// <see cref="BatchSize"/>, so a batch's worth of permanently-unresolvable
|
|
||||||
/// rows at the head STARVES every newer cached row behind them.
|
|
||||||
/// </para>
|
|
||||||
/// <para>
|
|
||||||
/// Abandoning costs only the operational (<c>SiteCalls</c>) 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.
|
|
||||||
/// </para>
|
|
||||||
/// <para>
|
|
||||||
/// Set to 0 to abandon on the first failed lookup. Negative values are
|
|
||||||
/// treated as 0.
|
|
||||||
/// </para>
|
|
||||||
/// </remarks>
|
|
||||||
public int CachedTrackingGraceSeconds { get; set; } = 300;
|
|
||||||
}
|
}
|
||||||
|
|||||||
+4
-137
@@ -362,22 +362,11 @@ public class SiteAuditTelemetryActorTests : TestKit
|
|||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public async Task CachedDrain_OrphanRow_PastGrace_IsAbandoned_AndTheValidRowStillFlows()
|
public async Task CachedDrain_OrphanRow_NoTrackingSnapshot_IsSkipped_DoesNotCrash()
|
||||||
{
|
{
|
||||||
// RETARGETED, not rewritten from scratch: this test previously asserted
|
|
||||||
// the orphan "is skipped and stays Pending", which is precisely the
|
|
||||||
// defect fixed here (known-issues/2026-07-20-cached-telemetry-drain-
|
|
||||||
// hot-loop.md) — a Pending row that can never resolve is re-read every
|
|
||||||
// drain forever and starves the queue behind it. The half of the test
|
|
||||||
// that still holds (the valid row flows, the drain does not crash) is
|
|
||||||
// kept verbatim; only the orphan's fate is inverted.
|
|
||||||
//
|
|
||||||
// NewCachedEvent stamps occurredAtUtc in the past, well beyond the
|
|
||||||
// default 300 s grace, so this orphan is past the abandon threshold.
|
|
||||||
//
|
|
||||||
// Arrange — two cached audit rows: one with a tracking snapshot, one
|
// Arrange — two cached audit rows: one with a tracking snapshot, one
|
||||||
// orphaned (the tracking store returns null). The orphaned row must be
|
// orphaned (the tracking store returns null). The orphaned row must be
|
||||||
// abandoned without aborting the batch — the valid row still flows.
|
// skipped without aborting the batch — the valid row still flows.
|
||||||
var orphan = NewCachedEvent(AuditKind.CachedSubmit);
|
var orphan = NewCachedEvent(AuditKind.CachedSubmit);
|
||||||
var valid = NewCachedEvent(AuditKind.CachedResolve);
|
var valid = NewCachedEvent(AuditKind.CachedResolve);
|
||||||
|
|
||||||
@@ -414,8 +403,8 @@ public class SiteAuditTelemetryActorTests : TestKit
|
|||||||
// Act
|
// Act
|
||||||
CreateActorWithCachedDrain();
|
CreateActorWithCachedDrain();
|
||||||
|
|
||||||
// Assert — exactly one push containing ONLY the valid row: the orphan
|
// Assert — exactly one push containing ONLY the valid row; the orphan
|
||||||
// has no operational half to send.
|
// is skipped and stays Pending (not in MarkForwardedAsync's id list).
|
||||||
await AwaitAssertAsync(async () =>
|
await AwaitAssertAsync(async () =>
|
||||||
{
|
{
|
||||||
await _client.Received(1).IngestCachedTelemetryAsync(
|
await _client.Received(1).IngestCachedTelemetryAsync(
|
||||||
@@ -426,131 +415,9 @@ public class SiteAuditTelemetryActorTests : TestKit
|
|||||||
Assert.Single(capturedBatch!.Packets);
|
Assert.Single(capturedBatch!.Packets);
|
||||||
Assert.Equal(valid.EventId.ToString(), capturedBatch.Packets[0].AuditEvent.EventId);
|
Assert.Equal(valid.EventId.ToString(), capturedBatch.Packets[0].AuditEvent.EventId);
|
||||||
|
|
||||||
// The valid row is marked Forwarded because central ack'd it...
|
|
||||||
await _queue.Received(1).MarkForwardedAsync(
|
await _queue.Received(1).MarkForwardedAsync(
|
||||||
Arg.Is<IReadOnlyList<Guid>>(g => g.Count == 1 && g[0] == valid.EventId),
|
Arg.Is<IReadOnlyList<Guid>>(g => g.Count == 1 && g[0] == valid.EventId),
|
||||||
Arg.Any<CancellationToken>());
|
Arg.Any<CancellationToken>());
|
||||||
|
|
||||||
// ...and the orphan is marked Forwarded too, in its own call, so it
|
|
||||||
// LEAVES the cached queue. Its audit half is still delivered by the
|
|
||||||
// reconciliation pull, which covers Forwarded rows as well as Pending.
|
|
||||||
await _queue.Received(1).MarkForwardedAsync(
|
|
||||||
Arg.Is<IReadOnlyList<Guid>>(g => g.Count == 1 && g[0] == orphan.EventId),
|
|
||||||
Arg.Any<CancellationToken>());
|
|
||||||
}
|
|
||||||
|
|
||||||
[Fact]
|
|
||||||
public async Task CachedDrain_OrphanRow_InsideGrace_IsRetried_NotAbandoned()
|
|
||||||
{
|
|
||||||
// The other side of the grace window, and the reason abandonment is not
|
|
||||||
// immediate: a missing snapshot is normally a brief write race (the
|
|
||||||
// audit row lands microseconds before the tracking row). Abandoning on
|
|
||||||
// the first failed lookup would throw away the operational half of
|
|
||||||
// every cached call that happened to lose that race.
|
|
||||||
var fresh = ScadaBridgeAuditEventFactory.Create(
|
|
||||||
channel: AuditChannel.ApiOutbound,
|
|
||||||
kind: AuditKind.CachedSubmit,
|
|
||||||
status: AuditStatus.Submitted,
|
|
||||||
eventId: Guid.NewGuid(),
|
|
||||||
occurredAtUtc: DateTime.UtcNow, // inside the grace window
|
|
||||||
target: "ERP.GetOrder",
|
|
||||||
sourceSiteId: "site-1",
|
|
||||||
correlationId: Guid.NewGuid());
|
|
||||||
|
|
||||||
_queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
|
|
||||||
.Returns(Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));
|
|
||||||
_queue.ReadPendingCachedTelemetryAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
|
|
||||||
.Returns(Task.FromResult<IReadOnlyList<AuditEvent>>(new[] { fresh }));
|
|
||||||
_trackingStore.GetStatusAsync(
|
|
||||||
new TrackedOperationId(fresh.CorrelationId!.Value), Arg.Any<CancellationToken>())
|
|
||||||
.Returns(Task.FromResult<TrackingStatusSnapshot?>(null));
|
|
||||||
|
|
||||||
CreateActorWithCachedDrain();
|
|
||||||
|
|
||||||
// Give the drain several ticks to prove it keeps retrying rather than
|
|
||||||
// abandoning. A positive control guards against vacuity: the row must
|
|
||||||
// actually have been READ more than once, otherwise "never marked
|
|
||||||
// Forwarded" would also be true of a drain that never ran at all.
|
|
||||||
await AwaitAssertAsync(async () =>
|
|
||||||
{
|
|
||||||
await _queue.Received(Quantity.Within(2, int.MaxValue))
|
|
||||||
.ReadPendingCachedTelemetryAsync(Arg.Any<int>(), Arg.Any<CancellationToken>());
|
|
||||||
}, TimeSpan.FromSeconds(6));
|
|
||||||
|
|
||||||
await _queue.DidNotReceive().MarkForwardedAsync(
|
|
||||||
Arg.Any<IReadOnlyList<Guid>>(), Arg.Any<CancellationToken>());
|
|
||||||
}
|
|
||||||
|
|
||||||
[Fact]
|
|
||||||
public async Task CachedDrain_FullBatchOfUnresolvableRows_DoesNotStarveTheQueue()
|
|
||||||
{
|
|
||||||
// The actual regression. The queue is read oldest-first with a fixed
|
|
||||||
// BatchSize, so before the fix a batch's worth of permanently
|
|
||||||
// unresolvable rows at the head meant every subsequent drain re-read
|
|
||||||
// exactly those rows, failed identically, and never reached the newer
|
|
||||||
// rows behind them — a permanent stall of the cached-telemetry path,
|
|
||||||
// plus one log line per row per pass (~2,800/min measured on a rig).
|
|
||||||
var stuck = Enumerable.Range(0, 4).Select(_ => NewCachedEvent()).ToList();
|
|
||||||
|
|
||||||
_queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
|
|
||||||
.Returns(Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));
|
|
||||||
_queue.ReadPendingCachedTelemetryAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
|
|
||||||
.Returns(Task.FromResult<IReadOnlyList<AuditEvent>>(stuck));
|
|
||||||
foreach (var row in stuck)
|
|
||||||
{
|
|
||||||
_trackingStore.GetStatusAsync(
|
|
||||||
new TrackedOperationId(row.CorrelationId!.Value), Arg.Any<CancellationToken>())
|
|
||||||
.Returns(Task.FromResult<TrackingStatusSnapshot?>(null));
|
|
||||||
}
|
|
||||||
|
|
||||||
CreateActorWithCachedDrain();
|
|
||||||
|
|
||||||
// Every row is abandoned in ONE call, so the head of the queue clears
|
|
||||||
// and the next drain can see past it.
|
|
||||||
await AwaitAssertAsync(async () =>
|
|
||||||
{
|
|
||||||
await _queue.Received(1).MarkForwardedAsync(
|
|
||||||
Arg.Is<IReadOnlyList<Guid>>(g => g.Count == stuck.Count),
|
|
||||||
Arg.Any<CancellationToken>());
|
|
||||||
}, TimeSpan.FromSeconds(5));
|
|
||||||
|
|
||||||
// Nothing was pushed — there was no operational half to build for any
|
|
||||||
// of them — but the drain still made progress rather than spinning.
|
|
||||||
await _client.DidNotReceive().IngestCachedTelemetryAsync(
|
|
||||||
Arg.Any<CachedTelemetryBatch>(), Arg.Any<CancellationToken>());
|
|
||||||
}
|
|
||||||
|
|
||||||
[Fact]
|
|
||||||
public async Task CachedDrain_TrackingStoreThrow_DoesNotAbandonTheRow()
|
|
||||||
{
|
|
||||||
// A throw is a STORE fault (locked, corrupt, mid-restore), not a
|
|
||||||
// verdict about this row: the snapshot may well exist and be readable
|
|
||||||
// a second later. Abandoning here would discard the operational half
|
|
||||||
// of every in-flight cached call during a transient SQLite lock — so
|
|
||||||
// the throw path deliberately never abandons, however old the row is.
|
|
||||||
var row = NewCachedEvent(); // deliberately older than the grace window
|
|
||||||
|
|
||||||
_queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
|
|
||||||
.Returns(Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));
|
|
||||||
_queue.ReadPendingCachedTelemetryAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
|
|
||||||
.Returns(Task.FromResult<IReadOnlyList<AuditEvent>>(new[] { row }));
|
|
||||||
_trackingStore.GetStatusAsync(
|
|
||||||
new TrackedOperationId(row.CorrelationId!.Value), Arg.Any<CancellationToken>())
|
|
||||||
.ThrowsAsync(new InvalidOperationException("tracking store unavailable"));
|
|
||||||
|
|
||||||
CreateActorWithCachedDrain();
|
|
||||||
|
|
||||||
// Positive control again: prove the drain actually ran repeatedly, so
|
|
||||||
// "never abandoned" is a real observation rather than an artefact of a
|
|
||||||
// drain that never executed.
|
|
||||||
await AwaitAssertAsync(async () =>
|
|
||||||
{
|
|
||||||
await _queue.Received(Quantity.Within(2, int.MaxValue))
|
|
||||||
.ReadPendingCachedTelemetryAsync(Arg.Any<int>(), Arg.Any<CancellationToken>());
|
|
||||||
}, TimeSpan.FromSeconds(6));
|
|
||||||
|
|
||||||
await _queue.DidNotReceive().MarkForwardedAsync(
|
|
||||||
Arg.Any<IReadOnlyList<Guid>>(), Arg.Any<CancellationToken>());
|
|
||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
|
|||||||
Reference in New Issue
Block a user