feat(comm): batch site→central stream events over gRPC (R2)

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.
2026-08-15 03:39:59 -04:00
parent 2b74851f96
commit 9b5cb3dd9d
16 changed files with 2150 additions and 131 deletions
@@ -580,4 +580,216 @@ public class SiteStreamGrpcServerTests : TestKit
        var server = CreateServer();
        Assert.Equal(0, server.DroppedStreamEventCount);
    }
+
+    // ── R2: gRPC event batching, and its negotiation ────────────────────────────
+
+    [Fact]
+    public void BatchOptions_AreBoundFromOptions_AndClampDegenerateValues()
+    {
+        var options = Microsoft.Extensions.Options.Options.Create(new CommunicationOptions());
+        var server = new SiteStreamGrpcServer(_subscriber, _logger, options);
+
+        Assert.Equal(100, server.StreamBatchMaxEvents);
+        Assert.Equal(TimeSpan.FromMilliseconds(25), server.StreamBatchWindow);
+
+        // CommunicationOptionsValidator fails the boot on these, but a host composed
+        // without validation must not blow up deep inside a live RPC.
+        var degenerate = new SiteStreamGrpcServer(_subscriber, _logger,
+            Microsoft.Extensions.Options.Options.Create(new CommunicationOptions
+            {
+                GrpcStreamBatchMaxEvents = 0,
+                GrpcStreamBatchWindow = TimeSpan.FromMilliseconds(-5),
+            }));
+        Assert.Equal(1, degenerate.StreamBatchMaxEvents);
+        Assert.Equal(TimeSpan.Zero, degenerate.StreamBatchWindow);
+    }
+
+    [Fact]
+    public async Task UnnegotiatedSubscription_NeverEmitsABatchFrame()
+    {
+        // OLD-CENTRAL ↔ NEW-SITE skew. proto3 defaults batching_supported to false, which
+        // is exactly what a central built before R2 sends. The site must then keep to one
+        // event per frame — a Batch frame would arrive at that central as
+        // EventOneofCase.None and be silently dropped by its ConvertToDomainEvent.
+        var (server, capture, cts, streamTask, relay) =
+            await StartCapturingStreamAsync(batchingSupported: false);
+
+        for (var i = 0; i < 50; i++)
+        {
+            relay.Tell(new Commons.Messages.Streaming.AttributeValueChanged(
+                "Site1.Pump01", "Path", "Attr", i, "Good", DateTimeOffset.UtcNow));
+        }
+
+        await WaitForConditionAsync(() => CountEvents(capture) >= 50, 10_000);
+
+        cts.Cancel();
+        await streamTask;
+
+        lock (capture)
+        {
+            Assert.All(capture, f => Assert.NotEqual(SiteStreamEvent.EventOneofCase.Batch, f.EventCase));
+            Assert.Equal(50, capture.Count);
+        }
+
+        GC.KeepAlive(server);
+    }
+
+    [Fact]
+    public async Task NegotiatedSubscription_CoalescesABacklogIntoFewerFramesThanEvents()
+    {
+        // NEW-CENTRAL ↔ NEW-SITE. A burst pushed at the relay faster than the pump drains
+        // it must come out in strictly fewer frames than events, with every event
+        // preserved in order.
+        const int burst = 400;
+        var (server, capture, cts, streamTask, relay) =
+            await StartCapturingStreamAsync(batchingSupported: true);
+
+        for (var i = 0; i < burst; i++)
+        {
+            relay.Tell(new Commons.Messages.Streaming.AttributeValueChanged(
+                "Site1.Pump01", "Path", "Attr", i, "Good", DateTimeOffset.UtcNow));
+        }
+
+        await WaitForConditionAsync(() => CountEvents(capture) >= burst, 15_000);
+
+        cts.Cancel();
+        await streamTask;
+
+        List<SiteStreamEvent> frames;
+        lock (capture) { frames = [.. capture]; }
+
+        Assert.Equal(burst, frames.Sum(CountFrameEvents));
+        Assert.True(frames.Count < burst,
+            $"batching produced {frames.Count} frames for {burst} events — no coalescing happened");
+        Assert.Contains(frames, f => f.EventCase == SiteStreamEvent.EventOneofCase.Batch);
+
+        // Order is preserved end to end: the values arrive 0..burst-1 exactly once each.
+        var values = frames.SelectMany(FlattenAttributeValues).ToArray();
+        Assert.Equal(Enumerable.Range(0, burst).Select(i => i.ToString()), values);
+
+        GC.KeepAlive(server);
+    }
+
+    [Fact]
+    public async Task BatchSizeHistogram_IsRecordedOnlyForNegotiatedStreams()
+    {
+        // scadabridge.site.stream.batch_size rides ScadaBridgeTelemetry.MeterName, which is
+        // already in SiteServiceRegistration.ObservedMeters — an unlisted meter exports
+        // nothing, silently. Assert the instrument actually fires, and that it does NOT
+        // fire on an un-negotiated stream (where it would degenerate into a per-event
+        // instrument on the hottest path in the product).
+        var measurements = new List<int>();
+        using var listener = new MeterListener();
+        listener.InstrumentPublished = (instrument, l) =>
+        {
+            if (instrument.Meter.Name == ScadaBridgeTelemetry.MeterName &&
+                instrument.Name == "scadabridge.site.stream.batch_size")
+            {
+                l.EnableMeasurementEvents(instrument);
+            }
+        };
+        listener.SetMeasurementEventCallback<int>((_, m, _, _) =>
+        {
+            lock (measurements) { measurements.Add(m); }
+        });
+        listener.Start();
+
+        // Un-negotiated: no measurements at all.
+        var (_, plainCapture, plainCts, plainTask, plainRelay) =
+            await StartCapturingStreamAsync(batchingSupported: false, correlationId: "corr-hist-off");
+        plainRelay.Tell(new Commons.Messages.Streaming.AttributeValueChanged(
+            "Site1.Pump01", "Path", "Attr", 1, "Good", DateTimeOffset.UtcNow));
+        await WaitForConditionAsync(() => CountEvents(plainCapture) >= 1);
+        plainCts.Cancel();
+        await plainTask;
+
+        lock (measurements) { Assert.Empty(measurements); }
+
+        // Negotiated: one measurement per emitted frame, each within the size cap.
+        var (_, capture, cts, streamTask, relay) =
+            await StartCapturingStreamAsync(batchingSupported: true, correlationId: "corr-hist-on");
+        for (var i = 0; i < 20; i++)
+        {
+            relay.Tell(new Commons.Messages.Streaming.AttributeValueChanged(
+                "Site1.Pump01", "Path", "Attr", i, "Good", DateTimeOffset.UtcNow));
+        }
+        await WaitForConditionAsync(() => CountEvents(capture) >= 20, 10_000);
+        cts.Cancel();
+        await streamTask;
+
+        lock (measurements)
+        {
+            Assert.NotEmpty(measurements);
+            Assert.Equal(20, measurements.Sum());
+            Assert.All(measurements, m => Assert.InRange(m, 1, SiteStreamGrpcServer.DefaultStreamBatchMaxEvents));
+        }
+    }
+
+    /// <summary>Total events carried across all captured frames (unpacking batch frames).</summary>
+    private static int CountEvents(List<SiteStreamEvent> capture)
+    {
+        lock (capture) { return capture.Sum(CountFrameEvents); }
+    }
+
+    private static int CountFrameEvents(SiteStreamEvent frame) =>
+        frame.EventCase == SiteStreamEvent.EventOneofCase.Batch ? frame.Batch.Events.Count : 1;
+
+    private static IEnumerable<string> FlattenAttributeValues(SiteStreamEvent frame)
+    {
+        if (frame.EventCase == SiteStreamEvent.EventOneofCase.Batch)
+        {
+            foreach (var inner in frame.Batch.Events)
+                yield return inner.AttributeChanged.Value;
+            yield break;
+        }
+
+        yield return frame.AttributeChanged.Value;
+    }
+
+    /// <summary>
+    /// Starts a SubscribeInstance stream with the given batch negotiation, capturing every
+    /// written frame and handing back the relay actor so the test can drive domain events.
+    /// </summary>
+    private async Task<(SiteStreamGrpcServer Server, List<SiteStreamEvent> Capture,
+        CancellationTokenSource Cts, Task StreamTask, IActorRef Relay)>
+        StartCapturingStreamAsync(bool batchingSupported, string correlationId = "corr-batch")
+    {
+        IActorRef? capturedActor = null;
+        var subscriber = Substitute.For<ISiteStreamSubscriber>();
+        subscriber.Subscribe(Arg.Any<string>(), Arg.Any<IActorRef>())
+            .Returns(ci =>
+            {
+                capturedActor = ci.Arg<IActorRef>();
+                return "sub-batch";
+            });
+
+        var server = new SiteStreamGrpcServer(subscriber, _logger,
+            Microsoft.Extensions.Options.Options.Create(new CommunicationOptions()));
+        server.SetReady(Sys);
+
+        var capture = new List<SiteStreamEvent>();
+        var writer = Substitute.For<IServerStreamWriter<SiteStreamEvent>>();
+        writer.WriteAsync(Arg.Any<SiteStreamEvent>(), Arg.Any<CancellationToken>())
+            .Returns(Task.CompletedTask)
+            .AndDoes(ci =>
+            {
+                var frame = ci.Arg<SiteStreamEvent>();
+                lock (capture) { capture.Add(frame); }
+            });
+
+        var cts = new CancellationTokenSource();
+        var context = CreateMockContext(cts.Token);
+
+        var request = new InstanceStreamRequest
+        {
+            CorrelationId = correlationId,
+            InstanceUniqueName = "Site1.Pump01",
+            BatchingSupported = batchingSupported
+        };
+
+        var streamTask = Task.Run(() => server.SubscribeInstance(request, writer, context));
+        await WaitForConditionAsync(() => capturedActor != null);
+
+        return (server, capture, cts, streamTask, capturedActor!);
+    }
 }