a5256e9b12
'Pattern 4: Integration Routing' — RouteIntegrationCallAsync → SiteEnvelope(IntegrationCallRequest) → SiteCommunicationActor integration handler — was plumbed end to end but connected at neither end: no producer (zero callers) and no handler (AkkaHostedService never registered LocalHandlerType.Integration). It was an early scaffold the architecture routed around — the brokered External→Central→Site→Central round-trip is served by the Inbound API's routed-site-script path (the RouteTo* verbs, driven from CommunicationServiceInstanceRouter), which is live, tested, and shares IntegrationTimeout. Decision (#32): delete. Removed the IntegrationCall{Request,Response} messages, RouteIntegrationCallAsync, the SiteCommunicationActor receive block + _integrationHandler field + LocalHandlerType.Integration, and the four tests that covered them (2 actor, 2 message-contract, 1 dispatcher- reject, 1 mapper-reject). KEPT IntegrationTimeout — it is the live timeout for the RouteTo* verbs. Updated the exclusion-narrative comments (proto/mapper/dispatcher), design §4, the components doc timeout table, and marked the known-issue RESOLVED. Full solution build clean (0/0); Communication 634 + Host 421 green. Net -172/+20 across 14 files. Not in the gRPC proto (was deliberately excluded there), so no wire-format change.
317 lines
33 KiB
Markdown
317 lines
33 KiB
Markdown
# Central–Site Communication
|
||
|
||
The Central–Site Communication component is the transport layer that connects the central cluster to every site cluster. It provides three independent transports, **all now gRPC-based across the boundary** (Akka `ClusterClient` was removed in Phase 4 of the ClusterClient→gRPC migration, 2026-07-23): gRPC command/control in both directions (`CentralControlService` on central via `GrpcCentralTransport`; `SiteCommandService` on the site via `GrpcSiteTransport`), gRPC server-streaming for real-time data (`SiteStreamService`), and plain token-gated HTTP for the deployment-config fetch. Cross-cluster Akka remoting and `ClusterClientReceptionist` are gone; Akka remoting is now intra-cluster only.
|
||
|
||
## Overview
|
||
|
||
Communication (#5) runs on every node in every cluster. The component code lives in `src/ZB.MOM.WW.ScadaBridge.Communication/`, organised as follows:
|
||
|
||
- `Actors/` — `CentralCommunicationActor`, `SiteCommunicationActor`, `DebugStreamBridgeActor`, `StreamRelayActor`.
|
||
- `Grpc/` — `SiteStreamGrpcServer`, `SiteStreamGrpcClient`, `SiteStreamGrpcClientFactory`, `ISiteStreamSubscriber`, and the proto DTO mappers.
|
||
- `Protos/` — `sitestream.proto` (proto source; generated C# is vendored in `SiteStreamGrpc/`).
|
||
- `CommunicationService.cs` — typed Ask-pattern façade used by callers on the central side.
|
||
- `DebugStreamService.cs` — session manager for debug stream bridge actors.
|
||
- `CommunicationOptions.cs` — configuration options class.
|
||
- `ServiceCollectionExtensions.cs` — DI registration (`AddCommunication`).
|
||
|
||
DI registration is called from the Host composition root via `AddCommunication`. The actors themselves are created inside `AkkaHostedService.RegisterCentralActors` / `RegisterSiteActors` because they must be created within the actor system, not by the DI container.
|
||
|
||
## Key Concepts
|
||
|
||
### Three transports, three concerns
|
||
|
||
| Transport | Who dials | Data direction | Purpose |
|
||
|-----------|-----------|----------------|---------|
|
||
| gRPC command/control (`CentralControlService`) | **site dials central** | site → central | Heartbeats, health reports, notification submit/status, audit + cached-telemetry ingest, reconcile |
|
||
| gRPC command/control (`SiteCommandService`) | **central dials the site** | central → site | Deploy notifies, lifecycle, OPC UA, remote queries, subscribe/unsubscribe handshake, snapshots, parked, route, failover |
|
||
| gRPC (`SiteStreamService`) | **central dials the site** | mostly site → central | Real-time attribute value and alarm state changes (server-streaming), plus the audit ingest/pull unary RPCs |
|
||
| HTTP `GET` (`/api/internal/deployments/{id}/config`) | **site dials central** | central → site | The flattened deployment config itself (notify-and-fetch), gated by a per-deployment `X-Deployment-Token` |
|
||
|
||
The transports are independent. A gRPC stream interruption does not affect in-flight command/control calls, and vice versa.
|
||
|
||
**`SiteStreamService`'s gRPC dial direction is inverted from its data direction.** Values flow site → central, but each **site node hosts the `SiteStreamService` server** and **central is the client**. Command/control is different: it runs on its own two services — `CentralControlService` (central-hosted, site dials in) and `SiteCommandService` (site-hosted, central dials in). So a central node now DOES host a gRPC server (`CentralControlService`), unlike before Phase 4. The two legacy `Ingest*` unary RPCs on the site-hosted `SiteStreamService` remain dead in the shipped topology (no site dials a site for ingest); audit telemetry is pushed site → central over `CentralControlService`, and central pulls with `PullAuditEvents` / `PullSiteCalls` by dialling the site's `SiteStreamService`.
|
||
|
||
**No transport carries transport encryption; two of the three now carry authentication.**
|
||
|
||
- **Akka remoting — unauthenticated, but intra-cluster only.** `BuildHocon` emits no `enable-ssl`, no secure cookie and no `trusted-selection-paths`, so remoting is plaintext and open to anything that can reach the remoting port — but as of Phase 4 remoting no longer crosses the site↔central boundary, so this exposure is pair-internal.
|
||
- **gRPC — authenticated by preshared key since 2026-07-22.** The listeners are still **h2c** — `ListenAnyIP(grpcPort, o => o.Protocols = HttpProtocols.Http2)` with no `UseHttps` — but `ControlPlaneAuthInterceptor` gates the command/control and streaming services alike: every method under `/sitestream.SiteStreamService/` (including the `PullAuditEvents` / `PullSiteCalls` RPCs that return audit rows) **and** the Phase-4 `CentralControlService` (site→central) / `SiteCommandService` (central→site). It is fail-closed (no key ⇒ everything refused, and `StartupValidator` will not boot a site node in that state), compares with `CryptographicOperations.FixedTimeEquals`, and rejects with `PermissionDenied`. The caller attaches the key via `ControlPlaneCredentials`, which binds `CallCredentials` to each channel so unary and streaming calls are covered uniformly. Keys are **per site** (`SB-GRPC-PSK-<siteId>`), so a compromised site yields only its own. `LocalDbSyncAuthInterceptor` shares the listener and keeps its own separate key on `/localdb_sync.v1.LocalDbSync/` — the two authenticate different peers (central vs. the pair partner) and are never shared.
|
||
- **HTTP config fetch — token-authenticated.** Its per-deployment token is the *entire* security boundary (the endpoint is `AllowAnonymous`).
|
||
|
||
A bearer PSK over plaintext h2c is readable and replayable by anyone on the path, so the design still assumes a trusted network between central and sites — but the bar is now "read the traffic" rather than "reach the port". TLS on these listeners is the follow-on hardening and would not change the key design. Operational detail: [`docs/deployment/topology-guide.md`](../deployment/topology-guide.md).
|
||
|
||
### Notify-and-fetch: the deployment-config HTTP path
|
||
|
||
An instance deployment does not carry its flattened configuration inside the command message. Central stages a `PendingDeployment` row (config JSON + a freshly generated `DeploymentFetchToken` + a TTL) and sends only a small `RefreshDeploymentCommand` over gRPC command/control (`SiteCommandService`), carrying the deployment id, revision hash, `CentralFetchBaseUrl` and the fetch token. The site's Deployment Manager singleton then calls back to central over plain HTTP:
|
||
|
||
```csharp
|
||
// SiteRuntime/Deployment/HttpDeploymentConfigFetcher.cs
|
||
var url = $"{centralFetchBaseUrl.TrimEnd('/')}/api/internal/deployments/{Uri.EscapeDataString(deploymentId)}/config";
|
||
using var req = new HttpRequestMessage(HttpMethod.Get, url);
|
||
req.Headers.Add("X-Deployment-Token", token);
|
||
```
|
||
|
||
`DeploymentConfigEndpoints.Resolve` (`ManagementService/DeploymentConfigEndpoints.cs:101`) checks existence and TTL *before* the token, so unknown, superseded and expired deployments are all indistinguishable `404`s; a live row with a wrong or missing token is `401`. The token comparison is constant-time. This path was originally introduced because a flattened config could exceed the default 128 KB Akka frame size when command/control rode ClusterClient — the oversized message was silently dropped and the deploy hung to its timeout (see `docs/known-issues/2026-06-26-deploy-config-exceeds-akka-frame-size.md`). That Akka frame limit no longer applies now that command/control rides gRPC (`SiteCommandService`, 4 MB cap), but notify-and-fetch remains the deploy path. `DeployArtifactsCommand` was **not** moved to this path and still carries its payload inline — now over gRPC, so it is no longer at risk of the 128 KB drop.
|
||
|
||
### Hub-and-spoke topology
|
||
|
||
Sites do not communicate with each other. All inter-cluster traffic flows through central. Central maintains one `GrpcSiteTransport` channel pair per site (dialling `SiteCommandService`); each site maintains a single `GrpcCentralTransport` channel pair pointed at both central nodes (dialling `CentralControlService`).
|
||
|
||
### `SiteEnvelope` routing
|
||
|
||
Central-side callers wrap outbound messages in a `SiteEnvelope(SiteId, Message)`. `CentralCommunicationActor` resolves the site's `ClusterClient` by `SiteId` and forwards the inner message to `/user/site-communication` on the site:
|
||
|
||
```csharp
|
||
// CommunicationService.cs — deployment pattern (notify-and-fetch)
|
||
public async Task<DeploymentStatusResponse> RefreshDeploymentAsync(
|
||
string siteId, RefreshDeploymentCommand command, CancellationToken cancellationToken = default)
|
||
{
|
||
var envelope = new SiteEnvelope(siteId, command);
|
||
return await GetActor().Ask<DeploymentStatusResponse>(
|
||
envelope, _options.DeploymentTimeout, cancellationToken);
|
||
}
|
||
```
|
||
|
||
`CentralCommunicationActor.HandleSiteEnvelope` extracts the inner message and routes it via the cached `ClusterClient`:
|
||
|
||
```csharp
|
||
private void HandleSiteEnvelope(SiteEnvelope envelope)
|
||
{
|
||
if (!_siteClients.TryGetValue(envelope.SiteId, out var entry))
|
||
{
|
||
_log.Warning("No ClusterClient for site {0}, cannot route message {1}",
|
||
envelope.SiteId, envelope.Message.GetType().Name);
|
||
return; // caller's Ask times out — no central buffering
|
||
}
|
||
|
||
entry.Client.Tell(
|
||
new ClusterClient.Send("/user/site-communication", envelope.Message),
|
||
Sender);
|
||
}
|
||
```
|
||
|
||
### No central buffering
|
||
|
||
If a site is unreachable when a command arrives, the caller's Ask times out. Central never queues command/control messages on behalf of a site. This is deliberate: it keeps the central coordinator stateless with respect to site availability and pushes retry responsibility to the operator or to the Store-and-Forward Engine for messages that tolerate it.
|
||
|
||
## Architecture
|
||
|
||
> **Phase 4 note (2026-07-23).** The `ClusterClient` mechanics described in this section — `SiteEnvelope` routing via `ClusterClient.Send("/user/site-communication", …)`, one `ClusterClient` per site on `CentralCommunicationActor`, the site's outbound `_centralClient`, and `ClusterClientReceptionist` registration — were **replaced by gRPC command/control** in Phase 4 of the ClusterClient→gRPC migration. Central→site commands now go over `GrpcSiteTransport` → the site-hosted `SiteCommandService` (per-site NodeA→NodeB failover channel pair), and site→central messages over `GrpcCentralTransport` → the central-hosted `CentralControlService` (sticky central-a→central-b pair). Endpoints are dialled directly (no receptionist, no cross-cluster actor discovery). The per-site address load below still runs, but it now feeds a **per-site gRPC-endpoint cache** (from `GrpcNodeAAddress`/`GrpcNodeBAddress`) consumed by a `SitePairChannelProvider`, not a `ClusterClient` contact set. See [Component-Communication.md](../requirements/Component-Communication.md) for the current design.
|
||
|
||
### Central-side: `CentralCommunicationActor`
|
||
|
||
`CentralCommunicationActor` is a `ReceiveActor` created at `/user/central-communication` and registered with `ClusterClientReceptionist` so the site's `ClusterClient` can locate it. It owns:
|
||
|
||
- A `Dictionary<string, (IActorRef Client, ImmutableHashSet<string> ContactAddresses)>` keyed by site identifier — one `ClusterClient` per site.
|
||
- A `RefreshSiteAddresses` periodic timer (60-second cadence, starting immediately). Each tick fires `LoadSiteAddressesFromDb`, which reads every `Site` row from the database and (as of Phase 4) builds a **per-site gRPC-endpoint cache** from `GrpcNodeAAddress`/`GrpcNodeBAddress`, piping a `SiteAddressCacheLoaded` message back to Self. `HandleSiteAddressCacheLoaded` reconciles the diff, updating the `SitePairChannelProvider` that `GrpcSiteTransport` uses to dial each site's `SiteCommandService`. (Before Phase 4 this parsed the Akka `NodeAAddress`/`NodeBAddress` into receptionist paths and created/updated/stopped a `ClusterClient` per site.)
|
||
- Proxy references to `NotificationOutboxActor` and `AuditLogIngestActor` cluster singletons, injected post-construction via `RegisterNotificationOutbox` / `RegisterAuditIngest` messages from the Host. Messages that arrive before the proxy is registered are answered with a non-accepted ack (notifications) or an empty reply (audit), so the site retries without data loss.
|
||
- Fanout of `SiteHealthReport` to the peer central node via `DistributedPubSub`, keyed on the `site-health-replica` topic, so both central nodes' aggregators stay in sync regardless of which central node the site's `ClusterClient` load-balanced the report to.
|
||
|
||
`ISiteClientFactory` / `DefaultSiteClientFactory` abstract `ClusterClient` construction for testability.
|
||
|
||
### Site-side: `SiteCommunicationActor`
|
||
|
||
`SiteCommunicationActor` is a `ReceiveActor` created at `/user/site-communication` and registered with `ClusterClientReceptionist`. It owns:
|
||
|
||
- An `IActorRef? _centralClient` — the site's outbound `ClusterClient` to central. Injected post-construction via `RegisterCentralClient`.
|
||
- A `Timers`-based heartbeat on `CommunicationOptions.ApplicationHeartbeatInterval` (default 5 s; deliberately distinct from the Akka.Remote `TransportHeartbeatInterval`, so retuning the transport failure detector cannot silently retune the health heartbeat). Each tick sends a `HeartbeatMessage` whose `IsActive` is stamped from `ActiveNodeEvaluator.SelfIsOldestUp` — the node is active when it is the **oldest `Up` member**, *not* when it holds cluster leadership (`SiteCommunicationActor.cs:517-518`). A throwing active-check is caught and reported as `IsActive = false`.
|
||
- Dispatch to local handlers for every inbound command pattern. Handlers for event-log, parked-message, integration, and artifact patterns are registered post-construction via `RegisterLocalHandler`; unregistered patterns receive an inline error reply so the central Ask does not stall.
|
||
|
||
Site-to-central messages (health reports, audit batches, notification submissions) are sent via:
|
||
|
||
```csharp
|
||
_centralClient.Tell(
|
||
new ClusterClient.Send("/user/central-communication", msg), Sender);
|
||
```
|
||
|
||
For request/response messages, the original `Sender` is forwarded as the `ClusterClient.Send` sender so any reply from central routes straight back to the waiting Ask on the site, not through `SiteCommunicationActor`. For fire-and-forget messages (e.g. `SiteHealthReport`), `Self` is used as the sender instead, because no reply is expected.
|
||
|
||
### Address loading and the 60-second refresh
|
||
|
||
`CentralCommunicationActor` calls `ISiteRepository.GetAllSitesAsync` inside a background `Task.Run` (to avoid blocking the actor thread on a database round-trip) and pipes the result as `SiteAddressCacheLoaded`. The actor-lifecycle `CancellationTokenSource` is threaded into the repository call so a slow MS SQL query is cancelled when the actor stops.
|
||
|
||
A malformed address for one site does not abort the refresh loop — the actor catches the parse failure, logs a warning, skips that site, and processes the rest. The refresh also runs immediately on startup (`TimeSpan.Zero` initial delay) so the cache is populated before the first command arrives.
|
||
|
||
`CommunicationService.RefreshSiteAddresses()` triggers an on-demand refresh when a site record is added, edited, or deleted from the Central UI or CLI.
|
||
|
||
### gRPC real-time data transport
|
||
|
||
Real-time attribute value and alarm state changes are delivered over `SiteStreamService`, defined in `sitestream.proto`. The **server runs on every site node and the client runs on central** — central dials in to receive the stream (see the transport table above).
|
||
|
||
**Site-side** — `SiteStreamGrpcServer` (Kestrel h2c, HTTP/2 only, port 8083):
|
||
|
||
- Implements `SiteStreamService.SiteStreamServiceBase`.
|
||
- For each `SubscribeInstance` call, creates a `StreamRelayActor` (named `stream-relay-{correlationId}-{seq}`) and subscribes it to `ISiteStreamSubscriber` (implemented by `SiteStreamManager` in the Site Runtime project — `SiteStreamGrpcServer` holds only the interface so it does not reference `SiteRuntime` directly).
|
||
- Bridges events via a `BoundedChannel<SiteStreamEvent>` (capacity 1000, `DropOldest`) from the actor thread to the async gRPC write loop.
|
||
- Enforces a `GrpcMaxConcurrentStreams` limit (default 100) and a `GrpcMaxStreamLifetime` session timeout (default 4 hours) to evict zombie streams.
|
||
- Validates `correlation_id` against `ActorPath.IsValidPathElement` before use in an actor name, rejecting invalid values with `StatusCode.InvalidArgument`.
|
||
- During `CoordinatedShutdown`, `CancelAllStreams()` flips `_shuttingDown`, refuses new subscriptions with `StatusCode.Unavailable`, and cancels all active `CancellationTokenSource`s.
|
||
|
||
`StreamRelayActor` is a lightweight `ReceiveActor` that converts `AttributeValueChanged` and `AlarmStateChanged` domain events to proto `SiteStreamEvent` messages and writes them to the channel writer:
|
||
|
||
```csharp
|
||
// StreamRelayActor.cs
|
||
private void HandleAttributeValueChanged(AttributeValueChanged msg)
|
||
{
|
||
var protoEvent = new SiteStreamEvent
|
||
{
|
||
CorrelationId = _correlationId,
|
||
AttributeChanged = new AttributeValueUpdate
|
||
{
|
||
InstanceUniqueName = msg.InstanceUniqueName,
|
||
AttributePath = msg.AttributePath,
|
||
AttributeName = msg.AttributeName,
|
||
Value = ValueFormatter.FormatDisplayValue(msg.Value),
|
||
Quality = MapQuality(msg.Quality),
|
||
Timestamp = Timestamp.FromDateTimeOffset(msg.Timestamp)
|
||
}
|
||
};
|
||
WriteToChannel(protoEvent);
|
||
}
|
||
```
|
||
|
||
**Central-side** — `SiteStreamGrpcClient` / `SiteStreamGrpcClientFactory`:
|
||
|
||
- `SiteStreamGrpcClientFactory` (singleton) caches one `SiteStreamGrpcClient` per **`(site, endpoint)` pair** — a `ConcurrentDictionary<(string Site, string Endpoint), SiteStreamGrpcClient>`. The key was widened from site-only to fix an arch-review High: with a site-only key, one debug session's NodeA→NodeB failover flip disposed a channel another session was still using. `GetOrCreate` therefore no longer disposes on endpoint mismatch; both of a site's node channels coexist, and site *removal* (`RemoveSiteAsync`) is the only shared-disposal path. The trade-off is that an edited gRPC address leaves the old endpoint's idle channel cached until site removal or process shutdown — bounded at a handful of entries per site.
|
||
- `SiteStreamGrpcClient` opens a `GrpcChannel` with HTTP/2 keepalive (`KeepAlivePingDelay` default 15 s, `KeepAlivePingTimeout` default 10 s, `KeepAlivePingPolicy.Always`). `SubscribeAsync` is a plain `async Task` that calls `SubscribeInstance` and reads the response stream with `await foreach`, invoking `onEvent` for each received event and `onError` on any non-cancellation exception. The caller (`DebugStreamBridgeActor.OpenGrpcStream`) launches it inside a `Task.Run` so the long-running stream loop runs off the actor thread.
|
||
|
||
### Debug stream session lifecycle
|
||
|
||
`DebugStreamService` manages one `DebugStreamBridgeActor` per active debug session. On `StartStreamAsync`, it resolves the instance's site and gRPC addresses, creates the bridge actor, and holds the session in a `ConcurrentDictionary`.
|
||
|
||
`DebugStreamBridgeActor` (one per session, short-lived, no persistence):
|
||
|
||
1. In `PreStart`, sends `SubscribeDebugViewRequest` to `CentralCommunicationActor` (ClusterClient, for the initial snapshot).
|
||
2. On receiving `DebugViewSnapshot`, fires `onEvent(snapshot)` and calls `OpenGrpcStream`.
|
||
3. `OpenGrpcStream` calls `_grpcFactory.GetOrCreate(siteId, endpoint)` and launches `client.SubscribeAsync(...)` as a background task. Domain events are marshalled back to the actor via `Self.Tell` for thread safety.
|
||
4. On a gRPC error, flips to the other node endpoint and retries (first retry immediate, subsequent retries with `ReconnectDelay` default 5 s). The retry budget (`MaxRetries = 3`) is recovered only after `StabilityWindow` (default 60 s) of uninterrupted connection — a stream that delivers one event then immediately fails does not count as stable.
|
||
5. On `StopDebugStream`, cancels the gRPC subscription and sends `UnsubscribeDebugViewRequest` to the site via ClusterClient.
|
||
|
||
### Proto definition summary
|
||
|
||
```proto
|
||
// Protos/sitestream.proto — six RPCs, all served by the SITE
|
||
service SiteStreamService {
|
||
rpc SubscribeInstance(InstanceStreamRequest) returns (stream SiteStreamEvent);
|
||
rpc SubscribeSite(SiteStreamRequest) returns (stream SiteStreamEvent);
|
||
rpc IngestAuditEvents(AuditEventBatch) returns (IngestAck);
|
||
rpc IngestCachedTelemetry(CachedTelemetryBatch) returns (IngestAck);
|
||
rpc PullAuditEvents(PullAuditEventsRequest) returns (PullAuditEventsResponse);
|
||
rpc PullSiteCalls(PullSiteCallsRequest) returns (PullSiteCallsResponse);
|
||
}
|
||
```
|
||
|
||
Two are server-streaming: `SubscribeInstance` carries the per-instance real-time stream; `SubscribeSite` is the **site-wide, alarm-only** stream (no instance filter, attribute updates never carried) that feeds the aggregated central live alarm cache. The four unary RPCs serve the Audit Log and Site Call Audit push/pull paths — `SiteStreamGrpcServer` hosts them on the same port because sites already listen there. As noted above, the two `Ingest*` RPCs are dead in the shipped topology (no central gRPC server exists for a site to dial); the two `Pull*` RPCs are live, with central as the caller.
|
||
|
||
`SiteStreamEvent` uses a `oneof event { AttributeValueUpdate, AlarmStateUpdate }` discriminator. `AlarmStateUpdate` carries the full native alarm condition (fields 8–23) alongside the base computed-alarm fields (1–7), added additively so old clients ignoring unknown fields continue to work. Field numbers are never reused and evolution is additive only.
|
||
|
||
The generated C# is **vendored** under `Communication/SiteStreamGrpc/` with the `<Protobuf>` include commented out, so editing `sitestream.proto` does not regenerate on build — regeneration is a manual toggle-build-copy-untoggle.
|
||
|
||
## Usage
|
||
|
||
Central callers interact through `CommunicationService`, which wraps each command pattern in a typed `Ask` with a per-pattern timeout:
|
||
|
||
| Pattern | Method | Timeout |
|
||
|---------|--------|---------|
|
||
| Instance deployment (notify-and-fetch) | `RefreshDeploymentAsync` | 120 s |
|
||
| Instance lifecycle | `DisableInstanceAsync`, `EnableInstanceAsync`, `DeleteInstanceAsync` | 30 s |
|
||
| Artifact deployment | `DeployArtifactsAsync` | 60 s |
|
||
| Integration routing (Inbound API routed-site-script) | `RouteToCallAsync`, `RouteToGetAttributesAsync`, `RouteToSetAttributesAsync`, `RouteToWaitForAttributeAsync` | 30 s (`IntegrationTimeout`) |
|
||
| Debug snapshot | `RequestDebugSnapshotAsync` | 30 s |
|
||
| Remote queries | `QueryEventLogsAsync`, `QueryParkedMessagesAsync`, etc. | 30 s |
|
||
| OPC UA tag browse | `BrowseNodeAsync` | 30 s |
|
||
| Notification outbox (central-local) | `QueryNotificationOutboxAsync`, `RetryNotificationAsync`, etc. | 30 s |
|
||
| Site Call Audit (central-local) | `QuerySiteCallsAsync`, `RetrySiteCallAsync`, etc. | 30 s |
|
||
|
||
Notification Outbox and Site Call Audit actors are central-local singletons — their `CommunicationService` methods Ask the proxy directly without wrapping in `SiteEnvelope`.
|
||
|
||
For real-time streaming, callers use `DebugStreamService.StartStreamAsync`, which creates a `DebugStreamBridgeActor` and returns a session handle. Ongoing events arrive via the `onEvent` callback; session teardown is via `StopStreamAsync`.
|
||
|
||
## Configuration
|
||
|
||
All options are bound from the `ScadaBridge:Communication` section via `CommunicationOptions`:
|
||
|
||
| Key | Default | Description |
|
||
|-----|---------|-------------|
|
||
| `DeploymentTimeout` | `00:02:00` | Ask timeout for the `RefreshDeploymentCommand` round-trip (covers the site's HTTP config fetch and apply). |
|
||
| `LifecycleTimeout` | `00:00:30` | Ask timeout for lifecycle commands (disable, enable, delete). |
|
||
| `ArtifactDeploymentTimeout` | `00:01:00` | Ask timeout for system-wide artifact deployment. |
|
||
| `QueryTimeout` | `00:00:30` | Ask timeout for remote queries and management commands. |
|
||
| `IntegrationTimeout` | `00:00:30` | Ask timeout for integration routing and Inbound API routing. |
|
||
| `DebugViewTimeout` | `00:00:10` | Ask timeout for debug subscribe/unsubscribe handshake. |
|
||
| `NotificationForwardTimeout` | `00:00:30` | Ask timeout for notification submission forwarding. |
|
||
| `CentralGrpcEndpoints` | `[]` | Site-side: gRPC h2c endpoints of the central nodes' `CentralControlService`, e.g. `http://scadabridge-central-a:8083` (central's `CentralGrpcPort`, default 8083 — direct, not via Traefik, which is HTTP/1 only). **Required on Site nodes** (at least one); empty on Central nodes, which host `CentralControlService` and do not dial it. Replaces the former `CentralContactPoints` Akka-address list. |
|
||
| `GrpcKeepAlivePingDelay` | `00:00:15` | HTTP/2 keepalive PING interval on `SiteStreamGrpcClient`. |
|
||
| `GrpcKeepAlivePingTimeout` | `00:00:10` | HTTP/2 keepalive PING timeout. |
|
||
| `GrpcMaxStreamLifetime` | `04:00:00` | Per-stream session timeout; forces reconnect of zombie streams. |
|
||
| `GrpcMaxConcurrentStreams` | `100` | Max concurrent `SubscribeInstance` streams per site node. |
|
||
| `ApplicationHeartbeatInterval` | `00:00:05` | `SiteCommunicationActor` site→central heartbeat cadence. |
|
||
| `TransportHeartbeatInterval` | `00:00:05` | Akka.Remote transport failure-detector heartbeat interval (emitted into the HOCON by the Host). Distinct from the application heartbeat above. |
|
||
| `TransportFailureThreshold` | `00:00:15` | Akka remoting failure-detection threshold. |
|
||
| `CentralFetchBaseUrl` | `""` | Base URL (Traefik/LB) the site uses to fetch deploy configs from central. Carried in `RefreshDeploymentCommand` so sites need no standing config; **empty makes a deploy impossible** — `DeploymentService` fails fast. |
|
||
| `PendingDeploymentTtl` | `00:05:00` | How long a staged `PendingDeployment` row and its fetch token stay valid. Must comfortably cover both site nodes' fetches within one deploy window. |
|
||
| `PendingDeploymentPurgeInterval` | `01:00:00` | Cadence of the central `pending-deployment-purge` singleton that sweeps TTL-expired staging rows. Hygiene only — the fetch endpoint already enforces the TTL. |
|
||
|
||
Three layers of dead-client detection protect the gRPC stream path:
|
||
|
||
| Layer | Detects | Timeline |
|
||
|-------|---------|----------|
|
||
| TCP RST | Clean process death, connection close | ~1–5 s |
|
||
| gRPC keepalive PING | Network partition, silent crash | ~25 s |
|
||
| Session timeout (`GrpcMaxStreamLifetime`) | Zombie streams with misconfigured keepalive | 4 h |
|
||
|
||
## Dependencies & Interactions
|
||
|
||
- [Commons (#16)](./Commons.md) — owns all message contracts used by this component: `DeployInstanceCommand`, `SiteEnvelope`, `HeartbeatMessage`, `SiteHealthReport`, `SiteHealthReportReplica`, `RegisterNotificationOutbox`, `RegisterAuditIngest`, `IngestAuditEventsCommand`, `IngestCachedTelemetryCommand`, and all other request/response records. Commons does not hold an Akka package reference, so `RegisterAuditIngest` (which carries an `IActorRef`) lives in this project.
|
||
- [Cluster Infrastructure (#13)](./ClusterInfrastructure.md) — provides ClusterSingleton and the oldest-`Up` active/standby model that `SiteCommunicationActor`'s `IsActive` stamp depends on, plus the single `"scadabridge"` `ActorSystem` name for intra-cluster remoting. (`ClusterClientReceptionist` is no longer used — cross-cluster messaging is gRPC as of Phase 4.) `CentralCommunicationActor`'s `DistributedPubSub` fanout keeps both central nodes in sync regardless of which one a site's report landed on.
|
||
- [Configuration Database (#17)](./ConfigurationDatabase.md) — provides `ISiteRepository.GetAllSitesAsync` for address loading; site records carry `NodeAAddress`, `NodeBAddress`, `GrpcNodeAAddress`, `GrpcNodeBAddress`.
|
||
- [Deployment Manager (#2)](./DeploymentManager.md) — the primary consumer of command/control patterns 1–3. `CommunicationService` is injected into the Deployment Manager actor to send deploy notifies, lifecycle commands, and artifact deployments to sites. It also owns the staging half of the notify-and-fetch HTTP path (`PendingDeployment` rows + fetch tokens); the endpoint itself is served by the Management Service.
|
||
- [Site Runtime (#3)](./SiteRuntime.md) — `SiteCommunicationActor` forwards inbound commands to the `DeploymentManager` singleton proxy. `SiteStreamManager` (in Site Runtime) implements `ISiteStreamSubscriber` so `SiteStreamGrpcServer` can subscribe relay actors to instance event feeds without referencing Site Runtime directly.
|
||
- [Health Monitoring (#11)](./HealthMonitoring.md) — `CentralCommunicationActor` calls `ICentralHealthAggregator.MarkHeartbeat` and `ProcessReport` for every inbound heartbeat and health report. `DistributedPubSub` fanout keeps both central nodes' aggregators in sync.
|
||
- [Audit Log (#23)](./AuditLog.md) — `SiteStreamGrpcServer` hosts the `IngestAuditEvents`, `IngestCachedTelemetry`, `PullAuditEvents` and `PullSiteCalls` RPCs. The `Ingest*` pair on the site-hosted service stays unused in the shipped topology; sites push audit telemetry site→central over gRPC to `CentralControlService`, which routes `IngestAuditEventsCommand` / `IngestCachedTelemetryCommand` to the `AuditLogIngestActor` proxy. The `Pull*` reconciliation RPCs run the other way, with the central `site-audit-reconciliation` singleton dialling each site's `SiteStreamService`.
|
||
- [Notification Outbox (#21)](./NotificationOutbox.md) — `CentralCommunicationActor` routes `NotificationSubmit` / `NotificationStatusQuery` messages from sites to the `NotificationOutboxActor` proxy. `CommunicationService` Asks the proxy directly for central-UI outbox management calls.
|
||
- [Site Call Audit (#22)](./SiteCallAudit.md) — `CommunicationService` Asks the `SiteCallAuditActor` proxy directly for query and relay operations. `SiteCallAuditActor` issues `RetryParkedOperation` / `DiscardParkedOperation` relay commands to sites via `SiteEnvelope`; `SiteCommunicationActor` dispatches them to `_parkedMessageHandler`.
|
||
- [Store-and-Forward Engine (#6)](./StoreAndForward.md) — the site S&F Engine drives `NotificationSubmit` forwarding and cached-call telemetry emission through `SiteCommunicationActor`. Parked-message queries and retry/discard relay commands flow back the other way.
|
||
- [Management Service (#18)](./ManagementService.md) — `ManagementActor` runs at `/user/management` on central and is reached **in-process** through `ManagementActorHolder`; the CLI connects over HTTP, not any cluster transport. (It was `ClusterClientReceptionist`-registered until 2026-07-22, for a CLI that was never built that way.) After Phase 4 this component uses no `ClusterClient` at all — its cross-cluster connections are the gRPC `CentralControlService` / `SiteCommandService` transports. Management Service also hosts `DeploymentConfigEndpoints` — the `GET /api/internal/deployments/{id}/config` route that terminates the HTTP deploy-config transport, mapped in the central-role block alongside `/api/audit/*` and `/management`.
|
||
- Design spec: [Component-Communication.md](../requirements/Component-Communication.md).
|
||
|
||
## Troubleshooting
|
||
|
||
### A site's commands fail immediately
|
||
|
||
Check that `GrpcNodeAAddress` and `GrpcNodeBAddress` are populated in the site configuration — if both are empty, `CentralCommunicationActor` logs a warning and skips that site on every refresh, so no `SiteCommandService` channel is built and all commands timeout. `CommunicationService.RefreshSiteAddresses()` triggers an on-demand refresh after an address is added.
|
||
|
||
### Commands are timing out but the site is reachable
|
||
|
||
A single malformed gRPC endpoint string for one site can silently prevent channel creation for that site while other sites are unaffected. Check the logs for a `Warning` line from `HandleSiteAddressCacheLoaded` naming the offending site. The per-site parse-guard skips the bad entry so the rest of the refresh proceeds.
|
||
|
||
A `Warning` at the `Status.Failure` handler in `CentralCommunicationActor` means `LoadSiteAddressesFromDb` itself threw (typically a SQL connection error); the cache is left stale until the next successful refresh.
|
||
|
||
### gRPC debug stream drops immediately after opening
|
||
|
||
`SiteStreamGrpcServer` rejects `correlation_id` values that contain characters invalid in Akka actor names (`/`, whitespace, etc.) with `StatusCode.InvalidArgument`. Verify that the calling `DebugStreamBridgeActor` generates a safe correlation ID.
|
||
|
||
After a site node failover, the `DebugStreamBridgeActor` attempts to reconnect to the other node endpoint (`_useNodeA` flips on each error). If both nodes are unreachable, the actor exhausts its 3-retry budget and calls `onTerminated`. The engineer must restart the debug session.
|
||
|
||
### Deployments fail immediately with a config-fetch error
|
||
|
||
The site received the `RefreshDeploymentCommand` over gRPC command/control (`SiteCommandService`) but could not complete the HTTP leg. Check `CentralFetchBaseUrl` first — it must be reachable *from the site*, so a value that only resolves inside the central network fails every deploy. A `404` from the fetch means the staged row was unknown, superseded, or past `PendingDeploymentTtl`; a `401` means the row is live but the token did not match. Because the endpoint hides existence, a `404` cannot distinguish "wrong id" from "expired".
|
||
|
||
### Heartbeats arrive but health reports do not
|
||
|
||
`SiteCommunicationActor` sends heartbeats and health reports via separate paths. Health reports are sent only when the site's `HealthReportSender` publishes them (every 30 s by default). If heartbeats arrive but reports do not, the health-report sender on the site may have faulted — check site-side logs for errors in `HealthReportSender`.
|
||
|
||
## Related Documentation
|
||
|
||
- [Central–Site Communication design specification](../requirements/Component-Communication.md)
|
||
- [Commons](./Commons.md)
|
||
- [Cluster Infrastructure](./ClusterInfrastructure.md)
|
||
- [Configuration Database](./ConfigurationDatabase.md)
|
||
- [Deployment Manager](./DeploymentManager.md)
|
||
- [Site Runtime](./SiteRuntime.md)
|
||
- [Health Monitoring](./HealthMonitoring.md)
|
||
- [Audit Log](./AuditLog.md)
|
||
- [Notification Outbox](./NotificationOutbox.md)
|
||
- [Site Call Audit](./SiteCallAudit.md)
|
||
- [Store-and-Forward Engine](./StoreAndForward.md)
|
||
- [Management Service](./ManagementService.md)
|