Alarm Summary is now live-cache-driven via a transient per-site central live alarm cache (ISiteAlarmLiveCache) fed by a site-wide, alarm-only SubscribeSite gRPC stream (seed-then-stream), with the 15s poll retained as fallback + NotReporting authority. No persisted central alarm store ([PERM]). Updated Component-CentralUI, Component-Communication (new §6.1 + SubscribeSite RPC), M7 design (T13 delivered note), native-alarms [PERM] note, stillpending follow-up note, CLAUDE.md, and moved register row #10 to Resolved (left #22 in place for the sibling branch). Claude-Session: https://claude.ai/code/session_01MtdgwpEeCUn6cUA5f1LMPj
This commit is contained in:
@@ -67,6 +67,17 @@ Both central and site clusters. Each side has communication actors that handle m
|
||||
- **Accepted limitation (central-side session locality):** debug sessions are process-local state on the central node hosting the `DebugStreamBridgeActor`. A central **restart or failover drops all active sessions** with no `OnStreamTerminated` signal to the operator — the engineer simply re-establishes the debug session from the UI. Debug streaming is an interactive, transient diagnostic aid, so this is an accepted trade-off rather than a durability gap (arch review 02, U7).
|
||||
- **Backpressure is lossy-by-design and now observable:** the per-session `Channel<SiteStreamEvent>` (bounded 1000, `DropOldest`) silently evicts the oldest event when a slow consumer falls behind. The eviction is counted via the channel's `itemDropped` callback and logged at Warning (first eviction, then every 500th) so real event loss on a debug stream is visible instead of silent.
|
||||
|
||||
### 6.1 Aggregated Live Alarm Stream (Site → Central)
|
||||
|
||||
Delivered 2026-07-10 (`docs/plans/2026-07-10-aggregated-live-alarm-stream-plan.md`) for the operator Alarm Summary page (Central UI #9). Where §6 streams **one instance** for the interactive Debug View, this feeds a **transient, per-site central live alarm cache** so the Alarm Summary reflects alarm transitions in near-real-time instead of re-polling the whole site every 15s.
|
||||
|
||||
- **Site-wide, alarm-only stream**: a single `SubscribeSite` gRPC stream per site carries `AlarmStateChanged` events for **all** instances (attribute events are dropped — the summary never shows them, and attributes are far higher-volume). On the site this reuses the existing `SiteStreamManager` broadcast hub via `SiteStreamManager.SubscribeSiteAlarms(...)` (same wiring as the per-instance `Subscribe`, minus the `InstanceUniqueName` filter and filtered to alarms). `StreamRelayActor` is reused unchanged, so placeholder rows (`is_configured_placeholder`) are still dropped at the relay.
|
||||
- **Central live cache** (`ISiteAlarmLiveCache`, singleton `SiteAlarmLiveCacheService`): a DI singleton on the active central node. For each site with ≥1 active viewer it runs ONE shared, **reference-counted** per-site aggregator (`SiteAlarmAggregatorActor`); the first `Subscribe(siteId, onChanged)` starts it, the last subscriber leaving stops it after a short **linger** to avoid re-seed thrash. `GetCurrentAlarms(siteId)` returns the current immutable snapshot; `IsLive(siteId)` reports whether the aggregator has seeded and published at least once.
|
||||
- **Seed-then-stream** (copied from `DebugStreamBridgeActor` ordering): open the `SubscribeSite` stream first (buffer live deltas), run the snapshot fan-out once via the existing `DebugViewSnapshot` path (bounded by `LiveAlarmCacheSeedConcurrency`), flush the buffer with **dedup by `(InstanceUniqueName, AlarmName, SourceReference)`**, then live pass-through. Placeholders are seeded from the snapshot and never expected on the live stream.
|
||||
- **Failover & drift**: NodeA↔NodeB reconnect (client via `SiteStreamGrpcClient.SubscribeSiteAsync`, factory endpoint failover) re-seeds rather than serving stale; a periodic **reconcile snapshot** (default 60s) corrects any missed enable/disable so the cache can't drift indefinitely. `[PERM]` (`docs/plans/2026-05-29-native-alarms-design.md`): the cache is **purely in-memory** — no EF entity/table/migration, no persisted central alarm store — so a new active node simply re-seeds from scratch.
|
||||
- **Options** (`Communication` section, `CommunicationOptions`; eagerly validated by `CommunicationOptionsValidator` / `ValidateOnStart`): `LiveAlarmCacheLinger` (default 30s), `LiveAlarmCacheReconcileInterval` (default 60s), `LiveAlarmCacheSeedConcurrency` (default 8), `LiveAlarmCacheMaxSubscribersPerSite` (default 200).
|
||||
- **Telemetry** (`ScadaBridgeTelemetry` meter): observable gauge `scadabridge.site.alarm_cache.aggregators.active` (running per-site aggregators) and counter `scadabridge.site.alarm_cache.reconnects` (site-wide stream reconnects — a NodeA↔NodeB flip or reconcile-driven reopen; a sustained climb signals a flapping site link).
|
||||
|
||||
#### Site-Side gRPC Streaming Components
|
||||
|
||||
- **SiteStreamGrpcServer**: gRPC service (`SiteStreamService.SiteStreamServiceBase`) hosted on each site node via Kestrel HTTP/2 on a dedicated port (default 8083). Implements the `SubscribeInstance` RPC. For each subscription, creates a `StreamRelayActor` that subscribes to `SiteStreamManager`, bridges events through a `Channel<SiteStreamEvent>` to the gRPC response stream. Tracks active subscriptions by `correlation_id` — duplicate IDs cancel the old stream. Enforces a max concurrent stream limit (default 100). Rejects streams with `StatusCode.Unavailable` before the actor system is ready.
|
||||
@@ -84,13 +95,14 @@ Both central and site clusters. Each side has communication actors that handle m
|
||||
|
||||
The streaming protocol is defined in `sitestream.proto` (`src/ZB.MOM.WW.ScadaBridge.Communication/Protos/sitestream.proto`):
|
||||
|
||||
- **Service**: `SiteStreamService` — hosted on each site node by `SiteStreamGrpcServer` — exposes five RPCs. One is the original real-time **server-streaming** subscription; the other four are **unary request/response** calls added by the Audit Log (#23) and Site Call Audit (#22) components. A unary call is request/response and is distinct from the command/control ClusterClient channel — gRPC on this service is no longer real-time-stream-only:
|
||||
- `SubscribeInstance(InstanceStreamRequest) returns (stream SiteStreamEvent)` — the real-time debug stream (§6); the only server-streaming RPC.
|
||||
- **Service**: `SiteStreamService` — hosted on each site node by `SiteStreamGrpcServer` — exposes six RPCs. Two are real-time **server-streaming** subscriptions (`SubscribeInstance`, `SubscribeSite`); the other four are **unary request/response** calls added by the Audit Log (#23) and Site Call Audit (#22) components. A unary call is request/response and is distinct from the command/control ClusterClient channel — gRPC on this service is no longer real-time-stream-only:
|
||||
- `SubscribeInstance(InstanceStreamRequest) returns (stream SiteStreamEvent)` — the per-instance real-time debug stream (§6).
|
||||
- `SubscribeSite(SiteStreamRequest) returns (stream SiteStreamEvent)` — the **site-wide, alarm-only** aggregated stream (§6.1) added for the operator Alarm Summary live cache. Additive (new RPC + new `SiteStreamRequest { correlation_id }` message; no field renumbering). The server handler mirrors `SubscribeInstance` — same bounded `Channel(1000, DropOldest)`, `GrpcMaxConcurrentStreams`/`GrpcMaxStreamLifetime` limits, `SiteConnectionOpened/Closed` telemetry, and `StreamRelayActor` mapping — but subscribes via `SiteStreamManager.SubscribeSiteAlarms` (all instances, `AlarmStateChanged` only; attribute events dropped, no per-instance filter).
|
||||
- `IngestAuditEvents(AuditEventBatch) returns (IngestAck)` — central-side **ingest** receiving surface for Audit Log (#23) telemetry; routes the batch to the central `AuditLogIngestActor` proxy and returns the accepted `EventId`s. (The production *push* path is still ClusterClient via `ClusterClientSiteAuditClient`; this RPC is the gRPC-receiving counterpart.)
|
||||
- `IngestCachedTelemetry(CachedTelemetryBatch) returns (IngestAck)` — ingest receiving surface for the combined cached-call telemetry packet (audit row + `SiteCalls` operational upsert written in one transaction).
|
||||
- `PullAuditEvents(PullAuditEventsRequest) returns (PullAuditEventsResponse)` — central→site **reconciliation pull** for the Audit Log self-heal feed; the site serves `Pending`/`Forwarded` rows from its `ISiteAuditQueue`.
|
||||
- `PullSiteCalls(PullSiteCallsRequest) returns (PullSiteCallsResponse)` — central→site reconciliation pull for the Site Call Audit (#22) self-heal feed; the site serves operation-tracking rows changed since a cursor from its `IOperationTrackingStore`. A separate RPC from `PullAuditEvents` because the tracking store is the operational source of truth, distinct from the site audit queue.
|
||||
- **Messages**: `InstanceStreamRequest` (correlation_id, instance_unique_name), `SiteStreamEvent` (correlation_id, oneof event: `AttributeValueUpdate`, `AlarmStateUpdate`); `AuditEventDto`/`AuditEventBatch`/`IngestAck` for ingest; `CachedTelemetryPacket`/`CachedTelemetryBatch` (each packet pairing an `AuditEventDto` with a `SiteCallOperationalDto`); `PullAuditEventsRequest`/`PullAuditEventsResponse` and `PullSiteCallsRequest`/`PullSiteCallsResponse` (each request carries `since_utc` + `batch_size`; each response carries `more_available` to signal a saturated batch).
|
||||
- **Messages**: `InstanceStreamRequest` (correlation_id, instance_unique_name), `SiteStreamRequest` (correlation_id only — no instance name; drives the site-wide `SubscribeSite` stream), `SiteStreamEvent` (correlation_id, oneof event: `AttributeValueUpdate`, `AlarmStateUpdate`); `AuditEventDto`/`AuditEventBatch`/`IngestAck` for ingest; `CachedTelemetryPacket`/`CachedTelemetryBatch` (each packet pairing an `AuditEventDto` with a `SiteCallOperationalDto`); `PullAuditEventsRequest`/`PullAuditEventsResponse` and `PullSiteCallsRequest`/`PullSiteCallsResponse` (each request carries `since_utc` + `batch_size`; each response carries `more_available` to signal a saturated batch).
|
||||
- The `oneof event` pattern is extensible — future event types (health metrics, connection state changes) are added as new fields without breaking existing consumers.
|
||||
- Proto field numbers are never reused; new RPCs and message fields are appended additively. Old clients ignore unknown `oneof` variants.
|
||||
|
||||
|
||||
Reference in New Issue
Block a user