From b879e51541b2560612bcfc1ffbdc943c03cf2216 Mon Sep 17 00:00:00 2001
From: Joseph Doherty <dohejw01@gmail.com>
Date: Fri, 10 Jul 2026 06:47:55 -0400
Subject: [PATCH] =?UTF-8?q?test(perf):=20site-wide=20stream=20throughput?=
 =?UTF-8?q?=20measurement=20=E2=80=94=20first=20real=20perf-envelope=20tes?=
 =?UTF-8?q?t=20(arch-review=2008=20=C2=A72.3)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Claude-Session: https://claude.ai/code/session_01MtdgwpEeCUn6cUA5f1LMPj
---
 .../Streaming/SiteStreamThroughputTests.cs    | 102 ++++++++++++++++++
 1 file changed, 102 insertions(+)
 create mode 100644 tests/ZB.MOM.WW.ScadaBridge.PerformanceTests/Streaming/SiteStreamThroughputTests.cs

diff --git a/tests/ZB.MOM.WW.ScadaBridge.PerformanceTests/Streaming/SiteStreamThroughputTests.cs b/tests/ZB.MOM.WW.ScadaBridge.PerformanceTests/Streaming/SiteStreamThroughputTests.cs
new file mode 100644
index 00000000..07993754
--- /dev/null
+++ b/tests/ZB.MOM.WW.ScadaBridge.PerformanceTests/Streaming/SiteStreamThroughputTests.cs
@@ -0,0 +1,102 @@
+using System.Diagnostics;
+using Akka.Actor;
+using Microsoft.Extensions.Logging.Abstractions;
+using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming;
+using ZB.MOM.WW.ScadaBridge.SiteRuntime;
+using ZB.MOM.WW.ScadaBridge.SiteRuntime.Streaming;
+
+namespace ZB.MOM.WW.ScadaBridge.PerformanceTests.Streaming;
+
+/// <summary>
+/// Throughput measurement for the site-wide attribute/alarm stream
+/// (<see cref="SiteStreamManager"/>: <c>Source.ActorRef</c> → <c>BroadcastHub</c>
+/// with per-subscriber buffering, <c>DropHead</c> overflow). Design envelope: this
+/// stream carries every attribute value change on a site; it must comfortably
+/// sustain tens of thousands of events/sec. Thresholds are deliberately
+/// conservative (CI-machine safe) — this test exists to catch order-of-magnitude
+/// regressions, not to benchmark.
+/// </summary>
+/// <remarks>
+/// <para>
+/// This is the first of the two highest-value real measurements arch-review 08 §2.3
+/// called for. It follows the existing <c>HotPathLatencyTests</c> style:
+/// <see cref="Stopwatch"/>-based, <c>[Trait("Category", "Performance")]</c>,
+/// conservative thresholds so CI stays green on slow machines.
+/// </para>
+/// <para>
+/// It spins up a real (non-clustered) <see cref="ActorSystem"/> rather than using
+/// Akka.TestKit — the PerformanceTests project references SiteRuntime (which brings
+/// in Akka + Akka.Streams transitively) but not Akka.TestKit.Xunit2, so no csproj
+/// change is required.
+/// </para>
+/// </remarks>
+public class SiteStreamThroughputTests
+{
+    private sealed class CountingActor : ReceiveActor
+    {
+        public static long Received;
+
+        public CountingActor()
+        {
+            // The manager Tells the raw ISiteStreamEvent record (AttributeValueChanged)
+            // straight to the subscriber — no wrapper — so we match on the record.
+            Receive<AttributeValueChanged>(_ => Interlocked.Increment(ref Received));
+            ReceiveAny(_ => { });
+        }
+    }
+
+    [Trait("Category", "Performance")]
+    [Fact]
+    public async Task SiteStream_SustainsAtLeast10kEventsPerSecond_ToOneSubscriber()
+    {
+        const int eventCount = 100_000;
+        CountingActor.Received = 0;
+
+        var system = ActorSystem.Create("SiteStreamThroughputTests");
+        try
+        {
+            // Buffer sized to the whole burst so DropHead should not fire.
+            var manager = new SiteStreamManager(
+                new SiteRuntimeOptions { StreamBufferSize = eventCount },
+                NullLogger<SiteStreamManager>.Instance);
+            manager.Initialize(system);
+
+            var counter = system.ActorOf(Props.Create<CountingActor>());
+            manager.Subscribe("PerfInstance", counter);
+
+            var sw = Stopwatch.StartNew();
+            for (var i = 0; i < eventCount; i++)
+            {
+                manager.PublishAttributeValueChanged(new AttributeValueChanged(
+                    "PerfInstance", "Attr", "Attr", i, "Good", DateTimeOffset.UtcNow));
+            }
+
+            // Drain: wait until the counter goes quiet (no growth across a 200ms window).
+            long last = -1;
+            var deadline = DateTime.UtcNow.AddSeconds(30);
+            while (DateTime.UtcNow < deadline)
+            {
+                var now = Interlocked.Read(ref CountingActor.Received);
+                if (now == last && now > 0) break;
+                last = now;
+                await Task.Delay(200);
+            }
+            sw.Stop();
+
+            var received = Interlocked.Read(ref CountingActor.Received);
+            var eventsPerSecond = received / sw.Elapsed.TotalSeconds;
+
+            // DropHead means under extreme pressure some events may drop; with the
+            // buffer sized to the burst, deliver ratio should be ~1.0.
+            Assert.True(received >= eventCount * 0.9,
+                $"Expected >=90% delivery, got {received}/{eventCount}");
+            Assert.True(eventsPerSecond >= 10_000,
+                $"Expected >=10k events/s sustained, got {eventsPerSecond:F0}/s over {sw.Elapsed.TotalSeconds:F1}s " +
+                $"(delivered {received}/{eventCount})");
+        }
+        finally
+        {
+            await system.Terminate();
+        }
+    }
+}