P5 #144: event-driven persistent report receive path

src/AR.Iec61850/Mms/MmsPersistentReportMonitor.cs

@@ -575,9 +575,26 @@ private async Task ReceiveAndDispatchPersistentReportsAsync(
 
             if (IsReceivePumpRunning)
             {
-                var delay = remaining < TimeSpan.FromMilliseconds(25) ? remaining : TimeSpan.FromMilliseconds(25);
-                if (delay > TimeSpan.Zero)
-                    await Task.Delay(delay, cancellationToken).ConfigureAwait(false);
+                // The association-owned receive pump already blocks on the transport.
+                // Do not add a 25 ms application polling loop on top of it. Wait for the
+                // router's binary InformationReport signal, bounded by the next legitimate
+                // scheduler deadline (residual polling) or this receive slice deadline.
+                var wakeAt = deadline;
+                if (pollDirectory != null &&
+                    references.Length > 0 &&
+                    nextPollAt < wakeAt)
+                {
+                    wakeAt = nextPollAt;
+                }
+
+                var waitFor = wakeAt - DateTimeOffset.UtcNow;
+                if (waitFor > TimeSpan.Zero)
+                {
+                    await _receiveRouter.WaitForInformationReportAsync(
+                        waitFor,
+                        cancellationToken).ConfigureAwait(false);
+                }
+
                 continue;
             }
 

src/AR.Iec61850/Mms/MmsReceiveRouter.cs

@@ -25,6 +25,7 @@ public sealed class MmsReceiveRouter
     private readonly Queue<MmsPduEnvelope> _unconfirmed = new();
     private readonly Queue<MmsPduEnvelope> _unmatched = new();
     private readonly Dictionary<Guid, Channel<MmsPduEnvelope>> _informationReportSubscriptions = new();
+    private readonly SemaphoreSlim _informationReportAvailable = new(0, 1);
 
     public int QueuedConfirmedResultCount
     {
@@ -88,6 +89,7 @@ public MmsReceiveRouteResult Route(ReadOnlyMemory<byte> presentationPayload)
             if (envelope.IsInformationReport)
             {
                 _informationReports.Enqueue(envelope);
+                SignalInformationReportAvailable();
                 foreach (var subscription in _informationReportSubscriptions.Values)
                     subscription.Writer.TryWrite(envelope);
 
@@ -152,6 +154,31 @@ public bool TryDequeueInformationReport(out MmsPduEnvelope envelope)
         return false;
     }
 
+    /// <summary>
+    /// Waits until the canonical InformationReport queue has work without polling.
+    /// The signal is binary: multiple reports coalesce into one wake because callers
+    /// drain the queue after waking. Queue contents remain the single routing authority.
+    /// </summary>
+    internal async Task<bool> WaitForInformationReportAsync(
+        TimeSpan timeout,
+        CancellationToken cancellationToken = default)
+    {
+        cancellationToken.ThrowIfCancellationRequested();
+
+        if (QueuedInformationReportCount > 0)
+        {
+            _informationReportAvailable.Wait(0);
+            return true;
+        }
+
+        if (timeout <= TimeSpan.Zero)
+            return false;
+
+        return await _informationReportAvailable
+            .WaitAsync(timeout, cancellationToken)
+            .ConfigureAwait(false);
+    }
+
     internal MmsInformationReportSubscription SubscribeInformationReports(int capacity = 32)
     {
         var boundedCapacity = Math.Clamp(capacity, 1, 1024);
@@ -197,6 +224,18 @@ internal void FaultInformationReportSubscriptions(Exception exception)
             subscription.Writer.TryComplete(exception);
     }
 
+    private void SignalInformationReportAvailable()
+    {
+        try
+        {
+            _informationReportAvailable.Release();
+        }
+        catch (SemaphoreFullException)
+        {
+            // A wake is already pending. The queue remains authoritative and will be drained.
+        }
+    }
+
     public void Clear()
     {
         Channel<MmsPduEnvelope>[] subscriptions;
@@ -210,6 +249,10 @@ public void Clear()
             _informationReportSubscriptions.Clear();
         }
 
+        while (_informationReportAvailable.Wait(0))
+        {
+        }
+
         foreach (var subscription in subscriptions)
             subscription.Writer.TryComplete(new IOException("MMS association receive routing was reset."));
     }

tests/AR.Iec61850.Tests/Mms/MmsReceiveRouterTests.cs

@@ -99,6 +99,69 @@ public async Task Route_FansOutInformationReportWithoutStealingLegacyQueue()
         Assert.True(queued.IsInformationReport);
     }
 
+    [Fact]
+    public async Task WaitForInformationReportAsync_ReturnsImmediately_WhenQueueAlreadyHasReport()
+    {
+        var router = new MmsReceiveRouter();
+        router.Route(BuildInformationReport());
+
+        using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(1));
+        var available = await router.WaitForInformationReportAsync(
+            TimeSpan.FromSeconds(1),
+            timeout.Token);
+
+        Assert.True(available);
+        Assert.True(router.TryDequeueInformationReport(out var queued));
+        Assert.True(queued.IsInformationReport);
+    }
+
+    [Fact]
+    public async Task WaitForInformationReportAsync_WakesFromFutureReport_WithoutPolling()
+    {
+        var router = new MmsReceiveRouter();
+
+        using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(1));
+        var wait = router.WaitForInformationReportAsync(
+            TimeSpan.FromSeconds(1),
+            timeout.Token);
+
+        await Task.Yield();
+        Assert.False(wait.IsCompleted);
+
+        router.Route(BuildInformationReport());
+
+        Assert.True(await wait);
+        Assert.True(router.TryDequeueInformationReport(out var queued));
+        Assert.True(queued.IsInformationReport);
+    }
+
+    [Fact]
+    public async Task WaitForInformationReportAsync_CoalescesBurstIntoSingleWake()
+    {
+        var router = new MmsReceiveRouter();
+        router.Route(BuildInformationReport());
+        router.Route(BuildInformationReport());
+
+        Assert.True(await router.WaitForInformationReportAsync(TimeSpan.FromSeconds(1)));
+        Assert.True(router.TryDequeueInformationReport(out _));
+        Assert.True(router.TryDequeueInformationReport(out _));
+
+        var staleWake = await router.WaitForInformationReportAsync(TimeSpan.FromMilliseconds(20));
+        Assert.False(staleWake);
+    }
+
+    [Fact]
+    public async Task WaitForInformationReportAsync_TimesOut_WhenNoReportArrives()
+    {
+        var router = new MmsReceiveRouter();
+
+        var available = await router.WaitForInformationReportAsync(
+            TimeSpan.FromMilliseconds(20));
+
+        Assert.False(available);
+        Assert.Equal(0, router.QueuedInformationReportCount);
+    }
+
     [Fact]
     public async Task Clear_FaultsActiveInformationReportSubscribers()
     {