perf(dcl): batch subscribe/read/write seam, bounded reconnect, sharded subscriptions

2026-08-14 21:14:04 -04:00
parent ee193cd2bb
commit d15c5f02ea
25 changed files with 3131 additions and 324 deletions
@@ -0,0 +1,146 @@
+using Microsoft.Extensions.Logging.Abstractions;
+using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Alarms;
+using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;
+
+namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Adapters;
+
+/// <summary>
+/// WP2.1b — the MxGateway adapter's batch subscribe seam and the alarm-stream union
+/// filter. The plain-vs-supervisory advise choice itself lives inside
+/// <c>RealMxGatewayClient</c> (it needs a live MXAccess session); its bounded-pipeline
+/// mechanism is pinned separately by <see cref="BatchSeamPrimitiveTests"/>.
+/// </summary>
+[Collection("DataConnectionManagerActor")]
+public class MxGatewayBatchSeamTests
+{
+    private static MxGatewayDataConnection NewAdapter(FakeMxGatewayClient fake) =>
+        new(fake, NullLogger<MxGatewayDataConnection>.Instance);
+
+    private static Dictionary<string, string> Details() => new()
+    {
+        ["Endpoint"] = "http://gw:5000",
+        ["ApiKey"] = "key",
+        ["ClientName"] = "client-a",
+        ["WriteUserId"] = "0",
+        ["ReadTimeoutMs"] = "2000",
+    };
+
+    [Fact]
+    public async Task SubscribeBatch_IssuesExactlyOneBulkCallPerBatch()
+    {
+        var fake = new FakeMxGatewayClient();
+        var adapter = NewAdapter(fake);
+        await adapter.ConnectAsync(Details());
+
+        var results = await adapter.SubscribeBatchAsync(
+            ["A.x", "A.y", "A.z"], (_, _) => { });
+
+        // ONE bulk RPC for the whole batch — replacing the historical AddItem + Advise
+        // pair PER TAG (6 RPCs for these three tags).
+        Assert.Single(fake.BulkSubscribeCalls);
+        Assert.Equal(3, fake.BulkSubscribeCalls[0].Count);
+        Assert.Equal(3, results.Count);
+        Assert.All(results, r => Assert.True(r.Success));
+        Assert.All(results, r => Assert.NotNull(r.SubscriptionId));
+    }
+
+    [Fact]
+    public async Task SubscribeBatch_ReportsPerTagFailuresWithoutFailingTheBatch()
+    {
+        var fake = new FakeMxGatewayClient();
+        fake.BulkSubscribeFailures.Add("A.y");
+        var adapter = NewAdapter(fake);
+        await adapter.ConnectAsync(Details());
+
+        var results = await adapter.SubscribeBatchAsync(["A.x", "A.y"], (_, _) => { });
+
+        Assert.True(results.Single(r => r.TagPath == "A.x").Success);
+        var failed = results.Single(r => r.TagPath == "A.y");
+        Assert.False(failed.Success);
+        Assert.Null(failed.SubscriptionId);
+        Assert.NotNull(failed.ErrorMessage);
+    }
+
+    [Fact]
+    public async Task UnsubscribeBatch_ReleasesEveryIdInOneBulkCall()
+    {
+        var fake = new FakeMxGatewayClient();
+        var adapter = NewAdapter(fake);
+        await adapter.ConnectAsync(Details());
+
+        var results = await adapter.SubscribeBatchAsync(["A.x", "A.y"], (_, _) => { });
+        await adapter.UnsubscribeBatchAsync(results.Select(r => r.SubscriptionId!).ToList());
+
+        Assert.Single(fake.BulkUnsubscribeCalls);
+        Assert.Equal(2, fake.BulkUnsubscribeCalls[0].Count);
+    }
+
+    [Fact]
+    public async Task AlarmStream_OpensOnTheLongestCommonPrefix_AndOnlyRestartsForAnEscapingSource()
+    {
+        var fake = new FakeMxGatewayClient();
+        var adapter = NewAdapter(fake);
+        await adapter.ConnectAsync(Details());
+
+        void NoTransition(NativeAlarmTransition _) { }
+
+        // First source → the stream opens scoped to it.
+        await adapter.SubscribeAlarmsAsync("Area1.Tank1", null, NoTransition);
+        await WaitForPrefixCountAsync(fake, 1);
+        Assert.Equal("Area1.Tank1", fake.AlarmStreamPrefixes[0]);
+
+        // A sibling under the same prefix widens the union → one restart on "Area1.Tank".
+        await adapter.SubscribeAlarmsAsync("Area1.Tank2", null, NoTransition);
+        await WaitForPrefixCountAsync(fake, 2);
+        Assert.Equal("Area1.Tank", fake.AlarmStreamPrefixes[1]);
+
+        // A source ALREADY covered by the live prefix must NOT restart the stream.
+        await adapter.SubscribeAlarmsAsync("Area1.Tank3.Sub", null, NoTransition);
+        await Task.Delay(100);
+        Assert.Equal(2, fake.AlarmStreamPrefixes.Count);
+
+        // A source outside the prefix widens it again — here down to gateway-wide.
+        await adapter.SubscribeAlarmsAsync("Zone9.Pump", null, NoTransition);
+        await WaitForPrefixCountAsync(fake, 3);
+        Assert.Null(fake.AlarmStreamPrefixes[2]); // "" → gateway-wide
+    }
+
+    [Fact]
+    public async Task AlarmStream_UnsubscribeNeverRestartsTheStream()
+    {
+        var fake = new FakeMxGatewayClient();
+        var adapter = NewAdapter(fake);
+        await adapter.ConnectAsync(Details());
+
+        var first = await adapter.SubscribeAlarmsAsync("Area1.Tank1", null, _ => { });
+        // Wait for the first stream to actually open before widening: the open runs on a
+        // Task.Run whose token the restart cancels, so a restart racing an unstarted task
+        // would leave the first prefix unrecorded.
+        await WaitForPrefixCountAsync(fake, 1);
+        await adapter.SubscribeAlarmsAsync("Area1.Tank2", null, _ => { });
+        await WaitForPrefixCountAsync(fake, 2);
+
+        // Dropping a source leaves the prefix too BROAD at worst — bandwidth, not
+        // correctness — so no restart is issued.
+        await adapter.UnsubscribeAlarmsAsync(first);
+        await Task.Delay(100);
+        Assert.Equal(2, fake.AlarmStreamPrefixes.Count);
+    }
+
+    private static async Task WaitForPrefixCountAsync(FakeMxGatewayClient fake, int expected)
+    {
+        var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(5);
+        while (DateTime.UtcNow < deadline)
+        {
+            lock (fake.AlarmStreamPrefixes)
+            {
+                if (fake.AlarmStreamPrefixes.Count >= expected)
+                    return;
+            }
+            await Task.Delay(20);
+        }
+
+        Assert.Fail($"alarm stream was opened {fake.AlarmStreamPrefixes.Count} time(s), expected {expected}");
+    }
+}