9b5cb3dd9d
Every AttributeValueChanged/AlarmStateChanged rode its own gRPC message on
SiteStreamService; at target scale that is ~37.5k messages/s/site of pure
framing overhead. Coalesce them, additively, with no new RPC.
Wire (sitestream.proto, regenerated via docker/regen-proto.sh sitestream):
InstanceStreamRequest.batching_supported = 3
SiteStreamRequest.batching_supported = 2
SiteStreamEvent.batch = 4 (new oneof case)
SiteStreamEventBatch { repeated SiteStreamEvent events = 1 }
The proto3 default of batching_supported IS the negotiation, and it is
load-bearing: a batch frame reaches a pre-R2 central as EventOneofCase.None,
whose ConvertToDomainEvent returns null — the whole batch would vanish with
no error anywhere. An old central cannot set the flag so it never receives
one; an old site ignores the unknown request field and keeps sending
per-event frames, which the new client's ForEachEvent handles as the
single-event case. Both skew directions are covered by tests that go through
a real proto serialize/parse round-trip.
Server: SiteStreamEventBatcher, a per-subscriber pump replacing the handler's
await-foreach/WriteAsync loop and byte-identical to it at a cap of 1. It
never delays a lone event — it drains the already-queued backlog for free and
lingers only once a backlog is proven — emits a single-event buffer as a
plain frame, preserves order and per-event Timestamps exactly, and flushes
what is buffered when the send channel's writer completes. It sits DOWNSTREAM
of StreamRelayActor's bounded DropOldest channel, so it changes framing only
and does not move the burst ceiling (deferred register row 31, which lives in
the shared publish stage upstream of the BroadcastHub).
Client: sets the flag on both subscriptions and unpacks in order into the
existing per-event pipeline, so SiteAlarmAggregatorActor,
DebugStreamBridgeActor, consumer-keepalive/orphan logic,
reconnect-on-graceful-completion, generation fencing, the (siteId, endpoint)
factory key and IsLive semantics are untouched.
Options (validated): GrpcStreamBatchMaxEvents 100 (1 disables),
GrpcStreamBatchWindow 25 ms — validated strictly under 250 ms, the load
test's end-to-end P99 threshold. Measured worst case (trickle-with-backlog,
window-bound rather than cap-bound): P50 13.8 ms, P99 25.4 ms, max 25.8 ms;
cap-bound case sent 600 queued events in 6 frames.
Telemetry: histogram scadabridge.site.stream.batch_size tagged by stream
kind, recorded only on negotiated streams (per-event otherwise). It rides
ScadaBridgeTelemetry.MeterName, already in the ObservedMeters allowlist.
Docs: Component-Communication.md gains an Event Batching section; CLAUDE.md's
gRPC streaming bullet records the wire shape and the negotiation rationale.
139 lines
9.4 KiB
C#
139 lines
9.4 KiB
C#
using ZB.MOM.WW.Configuration;
|
|
|
|
namespace ZB.MOM.WW.ScadaBridge.Communication;
|
|
|
|
/// <summary>
|
|
/// Validates <see cref="CommunicationOptions"/> at startup (ValidateOnStart) so a
|
|
/// malformed "ScadaBridge:Communication" appsettings section fails fast at boot with a
|
|
/// key-naming message instead of surfacing later at first Ask/gRPC use. Every
|
|
/// timeout feeds a per-pattern Ask deadline or a gRPC keepalive/stream-lifetime
|
|
/// setting, and a zero/negative value there produces an opaque runtime failure
|
|
/// far from the offending config key.
|
|
/// </summary>
|
|
public sealed class CommunicationOptionsValidator : OptionsValidatorBase<CommunicationOptions>
|
|
{
|
|
/// <summary>
|
|
/// Exclusive upper bound on <see cref="CommunicationOptions.GrpcStreamBatchWindow"/> —
|
|
/// the end-to-end site→central event latency budget the target-scale load test asserts
|
|
/// a P99 against. The coalescing window is the only latency batching introduces, so it
|
|
/// must stay strictly inside that budget rather than consuming it whole.
|
|
/// </summary>
|
|
internal static readonly TimeSpan StreamBatchWindowCeiling = TimeSpan.FromMilliseconds(250);
|
|
|
|
/// <inheritdoc />
|
|
protected override void Validate(ValidationBuilder builder, CommunicationOptions options)
|
|
{
|
|
builder.RequireThat(options.DeploymentTimeout > TimeSpan.Zero,
|
|
$"ScadaBridge:Communication:DeploymentTimeout must be a positive duration (was {options.DeploymentTimeout}).");
|
|
|
|
builder.RequireThat(options.LifecycleTimeout > TimeSpan.Zero,
|
|
$"ScadaBridge:Communication:LifecycleTimeout must be a positive duration (was {options.LifecycleTimeout}).");
|
|
|
|
builder.RequireThat(options.ArtifactDeploymentTimeout > TimeSpan.Zero,
|
|
$"ScadaBridge:Communication:ArtifactDeploymentTimeout must be a positive duration (was {options.ArtifactDeploymentTimeout}).");
|
|
|
|
builder.RequireThat(options.QueryTimeout > TimeSpan.Zero,
|
|
$"ScadaBridge:Communication:QueryTimeout must be a positive duration (was {options.QueryTimeout}).");
|
|
|
|
builder.RequireThat(options.IntegrationTimeout > TimeSpan.Zero,
|
|
$"ScadaBridge:Communication:IntegrationTimeout must be a positive duration (was {options.IntegrationTimeout}).");
|
|
|
|
builder.RequireThat(options.DebugViewTimeout > TimeSpan.Zero,
|
|
$"ScadaBridge:Communication:DebugViewTimeout must be a positive duration (was {options.DebugViewTimeout}).");
|
|
|
|
builder.RequireThat(options.HealthReportTimeout > TimeSpan.Zero,
|
|
$"ScadaBridge:Communication:HealthReportTimeout must be a positive duration (was {options.HealthReportTimeout}).");
|
|
|
|
builder.RequireThat(options.NotificationForwardTimeout > TimeSpan.Zero,
|
|
$"ScadaBridge:Communication:NotificationForwardTimeout must be a positive duration (was {options.NotificationForwardTimeout}).");
|
|
|
|
// Outermost rung of the audit-ingest timeout ladder. It must be strictly greater than the
|
|
// gRPC deadline / central Ask beneath it, or a slow-but-succeeding central write gets acked
|
|
// to a caller that already gave up and re-sent the batch. See CommunicationOptions.
|
|
builder.RequireThat(
|
|
options.AuditForwardTimeout > Grpc.SiteStreamGrpcServer.AuditIngestAskTimeout,
|
|
$"ScadaBridge:Communication:AuditForwardTimeout must be strictly greater than the "
|
|
+ $"central audit-ingest Ask timeout ({Grpc.SiteStreamGrpcServer.AuditIngestAskTimeout}) "
|
|
+ $"so the ingest timeout ladder stays monotonic (was {options.AuditForwardTimeout}).");
|
|
|
|
builder.RequireThat(options.GrpcKeepAlivePingDelay > TimeSpan.Zero,
|
|
$"ScadaBridge:Communication:GrpcKeepAlivePingDelay must be a positive duration (was {options.GrpcKeepAlivePingDelay}).");
|
|
|
|
builder.RequireThat(options.GrpcKeepAlivePingTimeout > TimeSpan.Zero,
|
|
$"ScadaBridge:Communication:GrpcKeepAlivePingTimeout must be a positive duration (was {options.GrpcKeepAlivePingTimeout}).");
|
|
|
|
builder.RequireThat(options.GrpcMaxStreamLifetime > TimeSpan.Zero,
|
|
$"ScadaBridge:Communication:GrpcMaxStreamLifetime must be a positive duration (was {options.GrpcMaxStreamLifetime}).");
|
|
|
|
builder.RequireThat(options.TransportHeartbeatInterval > TimeSpan.Zero,
|
|
$"ScadaBridge:Communication:TransportHeartbeatInterval must be a positive duration (was {options.TransportHeartbeatInterval}).");
|
|
|
|
builder.RequireThat(options.ApplicationHeartbeatInterval > TimeSpan.Zero,
|
|
$"ScadaBridge:Communication:ApplicationHeartbeatInterval must be a positive duration (was {options.ApplicationHeartbeatInterval}).");
|
|
|
|
builder.RequireThat(options.TransportFailureThreshold > TimeSpan.Zero,
|
|
$"ScadaBridge:Communication:TransportFailureThreshold must be a positive duration (was {options.TransportFailureThreshold}).");
|
|
|
|
builder.RequireThat(options.PendingDeploymentTtl > TimeSpan.Zero,
|
|
$"ScadaBridge:Communication:PendingDeploymentTtl must be a positive duration (was {options.PendingDeploymentTtl}).");
|
|
|
|
builder.RequireThat(options.PendingDeploymentPurgeInterval > TimeSpan.Zero,
|
|
$"ScadaBridge:Communication:PendingDeploymentPurgeInterval must be a positive duration (was {options.PendingDeploymentPurgeInterval}).");
|
|
|
|
builder.RequireThat(options.GrpcMaxConcurrentStreams > 0,
|
|
$"ScadaBridge:Communication:GrpcMaxConcurrentStreams must be positive (was {options.GrpcMaxConcurrentStreams}).");
|
|
|
|
// ── Site→central stream event batching (R2) ─────────────────────────────
|
|
// 1 is legal and means "disabled" (one event per frame, the un-negotiated shape).
|
|
builder.RequireThat(options.GrpcStreamBatchMaxEvents > 0,
|
|
$"ScadaBridge:Communication:GrpcStreamBatchMaxEvents must be positive — 1 disables "
|
|
+ $"batching (was {options.GrpcStreamBatchMaxEvents}).");
|
|
|
|
// The coalescing window is the ONLY latency batching can add, and the target-scale
|
|
// load test holds end-to-end event latency to a 250 ms P99. Validate it strictly
|
|
// below that so a misconfigured window cannot silently spend the entire budget.
|
|
builder.RequireThat(
|
|
options.GrpcStreamBatchWindow >= TimeSpan.Zero
|
|
&& options.GrpcStreamBatchWindow < StreamBatchWindowCeiling,
|
|
$"ScadaBridge:Communication:GrpcStreamBatchWindow must be non-negative and strictly "
|
|
+ $"below {StreamBatchWindowCeiling.TotalMilliseconds:0} ms (the end-to-end stream "
|
|
+ $"latency budget the coalescing window spends from); zero means \"pack only what is "
|
|
+ $"already queued\" (was {options.GrpcStreamBatchWindow}).");
|
|
|
|
// The gRPC site→central transport needs at least one central endpoint to dial. gRPC is now
|
|
// the only site→central transport (ClusterClient was removed in the migration's Phase 4), so
|
|
// every site node must declare its central endpoints — there is no Akka fallback to ignore
|
|
// the list. Central nodes leave it empty (they host CentralControlService, they don't dial it).
|
|
builder.RequireThat(
|
|
options.CentralGrpcEndpoints.All(e => !string.IsNullOrWhiteSpace(e)),
|
|
"ScadaBridge:Communication:CentralGrpcEndpoints must not contain empty or whitespace "
|
|
+ "entries (a site node must list at least one central gRPC endpoint; central nodes "
|
|
+ "leave it empty).");
|
|
|
|
// ── Aggregated live alarm cache (plan #10, Task 6) ───────────────────────
|
|
// Linger drives a Timer dueTime; TimeSpan.Zero is valid (stop the aggregator
|
|
// immediately when the last viewer leaves), only a negative value is invalid.
|
|
builder.RequireThat(options.LiveAlarmCacheLinger >= TimeSpan.Zero,
|
|
$"ScadaBridge:Communication:LiveAlarmCacheLinger must be zero or a positive duration (was {options.LiveAlarmCacheLinger}).");
|
|
|
|
// Reconcile interval is a periodic timer cadence AND the start-retry cadence; a
|
|
// zero/negative value would spin or never fire.
|
|
builder.RequireThat(options.LiveAlarmCacheReconcileInterval > TimeSpan.Zero,
|
|
$"ScadaBridge:Communication:LiveAlarmCacheReconcileInterval must be a positive duration (was {options.LiveAlarmCacheReconcileInterval}).");
|
|
|
|
// Seed fan-out concurrency gates a SemaphoreSlim; at least 1, capped so a
|
|
// fat-fingered config can't open an unbounded burst of cross-cluster Asks.
|
|
builder.RequireThat(options.LiveAlarmCacheSeedConcurrency is >= 1 and <= 64,
|
|
$"ScadaBridge:Communication:LiveAlarmCacheSeedConcurrency must be between 1 and 64 (was {options.LiveAlarmCacheSeedConcurrency}).");
|
|
|
|
// Per-site viewer cap must admit at least one viewer, else the page could never go live.
|
|
builder.RequireThat(options.LiveAlarmCacheMaxSubscribersPerSite >= 1,
|
|
$"ScadaBridge:Communication:LiveAlarmCacheMaxSubscribersPerSite must be at least 1 (was {options.LiveAlarmCacheMaxSubscribersPerSite}).");
|
|
|
|
// Publish-coalescing window drives a single-shot timer; TimeSpan.Zero is valid
|
|
// (publish per delta — legacy), only a negative value is invalid.
|
|
builder.RequireThat(options.LiveAlarmCachePublishCoalesce >= TimeSpan.Zero,
|
|
$"ScadaBridge:Communication:LiveAlarmCachePublishCoalesce must be zero or a positive duration (was {options.LiveAlarmCachePublishCoalesce}).");
|
|
}
|
|
}
|