using Grpc.Core; using Microsoft.Extensions.Logging.Abstractions; using Shouldly; using Xunit; using ZB.MOM.WW.HistorianGateway.Contracts.Grpc; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian; namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests; /// /// The gateway has no batch SendEvent RPC, so the alarm writer pipelines its per-event sends — at most /// at once — instead of one round trip after another. Every /// event still gets its own outcome, in input order. /// public sealed class GatewayAlarmWriterPipeliningTests { private static CancellationToken Ct => TestContext.Current.CancellationToken; private static AlarmHistorianEvent Evt(int i) => new( $"alarm-{i}", "Area/Pump", "N", "LimitAlarm", AlarmSeverity.High, "Activated", $"m{i}", "u", null, new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc).AddSeconds(i)); [Fact] public async Task Sends_overlap_up_to_the_bound_and_outcomes_keep_input_order() { var inFlight = 0; var maxInFlight = 0; var fake = new FakeHistorianGatewayClient { SendEventSource = async (e, ct) => { var now = Interlocked.Increment(ref inFlight); InterlockedMax(ref maxInFlight, now); await Task.Delay(20, ct); Interlocked.Decrement(ref inFlight); // Odd messages are soft failures: their outcome must land on their own slot. return e.Properties.TryGetValue("Message", out var m) && int.Parse(m[1..]) % 2 == 1 ? new WriteAck { Success = false } : new WriteAck { Success = true }; }, }; using var writer = new GatewayAlarmHistorianWriter(fake, NullLogger.Instance); var outcomes = await writer.WriteBatchAsync(Enumerable.Range(0, 40).Select(Evt).ToList(), Ct); outcomes.Count.ShouldBe(40); for (var i = 0; i < 40; i++) outcomes[i].ShouldBe(i % 2 == 1 ? HistorianWriteOutcome.RetryPlease : HistorianWriteOutcome.Ack, $"event {i}"); maxInFlight.ShouldBeGreaterThan(1, "the sends must overlap — a sequential drain is 40 round trips"); maxInFlight.ShouldBeLessThanOrEqualTo(GatewayAlarmHistorianWriter.MaxSendsInFlight); fake.SendEventCallCount.ShouldBe(40); } [Fact] public async Task A_cancellation_mid_batch_sends_nothing_more_and_dead_letters_nothing() { using var cts = new CancellationTokenSource(); var sent = 0; var fake = new FakeHistorianGatewayClient { SendEventSource = async (_, ct) => { if (Interlocked.Increment(ref sent) == 3) await cts.CancelAsync(); await Task.Delay(10, ct); return new WriteAck { Success = true }; }, }; using var writer = new GatewayAlarmHistorianWriter(fake, NullLogger.Instance); var outcomes = await writer.WriteBatchAsync(Enumerable.Range(0, 40).Select(Evt).ToList(), cts.Token); outcomes.Count.ShouldBe(40); outcomes.ShouldNotContain(HistorianWriteOutcome.PermanentFail); outcomes.Count(o => o == HistorianWriteOutcome.RetryPlease).ShouldBeGreaterThan(0); sent.ShouldBeLessThanOrEqualTo(GatewayAlarmHistorianWriter.MaxSendsInFlight + 2, "no new send starts after the cancel"); } [Fact] public async Task A_poison_event_dead_letters_alone() { var fake = new FakeHistorianGatewayClient { SendEventSource = (e, _) => e.Properties.TryGetValue("Message", out var m) && m == "m5" ? Task.FromException(new RpcException(new Status(StatusCode.InvalidArgument, "bad"))) : Task.FromResult(new WriteAck { Success = true }), }; using var writer = new GatewayAlarmHistorianWriter(fake, NullLogger.Instance); var outcomes = await writer.WriteBatchAsync(Enumerable.Range(0, 10).Select(Evt).ToList(), Ct); outcomes[5].ShouldBe(HistorianWriteOutcome.PermanentFail); outcomes.Where((_, i) => i != 5).ShouldAllBe(o => o == HistorianWriteOutcome.Ack); } private static void InterlockedMax(ref int target, int value) { int current; while ((current = Volatile.Read(ref target)) < value && Interlocked.CompareExchange(ref target, value, current) != current) { } } }