fix: harden SV pacing estimator and isolate unrelated traffic

src/AR.Iec61850.Transports.Npcap/AR.Iec61850.Transports.Npcap.csproj

@@ -8,6 +8,10 @@
     <PackageReference Include="SharpPcap" Version="6.3.1" />
   </ItemGroup>
 
+  <ItemGroup>
+    <InternalsVisibleTo Include="AR.Iec61850.Tests" />
+  </ItemGroup>
+
   <PropertyGroup>
     <Copyright>Copyright (C) 2026 Mas Ari / masarray</Copyright>
     <PackageLicenseExpression>GPL-3.0-or-later</PackageLicenseExpression>

src/AR.Iec61850.Transports.Npcap/NpcapProcessBusDuplexTransport.cs

@@ -20,8 +20,9 @@ public sealed class NpcapProcessBusDuplexTransport : IProcessBusTransport, IProc
 
     private readonly ICaptureDevice _device;
     private readonly IInjectionDevice _injectionDevice;
-    private readonly object _gate = new();
-    private readonly SemaphoreSlim _sendGate = new(1, 1);
+    private readonly object _captureGate = new();
+    private readonly object _clockMapGate = new();
+    private readonly SemaphoreSlim _injectionGate = new(1, 1);
     private readonly Dictionary<SvTransmitKey, SvTransmitClock> _svTransmitClocks = new();
     private bool _capturing;
     private bool _disposed;
@@ -45,21 +46,23 @@ public async ValueTask SendAsync(ReadOnlyMemory<byte> frame, CancellationToken c
         ObjectDisposedException.ThrowIf(_disposed, this);
         cancellationToken.ThrowIfCancellationRequested();
 
-        await _sendGate.WaitAsync(cancellationToken).ConfigureAwait(false);
-        try
+        if (!TryReadSampledValuesKey(frame.Span, out var streamKey))
         {
-            var isSampledValues = TryReadSampledValuesKey(frame.Span, out var streamKey);
-            if (isSampledValues)
-                await PaceSampledValuesAsync(streamKey, cancellationToken).ConfigureAwait(false);
-
-            _injectionDevice.SendPacket(frame.ToArray());
+            await InjectAsync(frame, cancellationToken).ConfigureAwait(false);
+            return;
+        }
 
-            if (isSampledValues)
-                CommitSampledValuesSend(streamKey, Stopwatch.GetTimestamp());
+        var clock = GetOrCreateClock(streamKey);
+        await clock.PacingGate.WaitAsync(cancellationToken).ConfigureAwait(false);
+        try
+        {
+            await PaceSampledValuesAsync(clock, cancellationToken).ConfigureAwait(false);
+            await InjectAsync(frame, cancellationToken).ConfigureAwait(false);
+            clock.Commit(Stopwatch.GetTimestamp());
         }
         finally
         {
-            _sendGate.Release();
+            clock.PacingGate.Release();
         }
     }
 
@@ -83,7 +86,7 @@ public async IAsyncEnumerable<ProcessBusCapturedFrame> CaptureAsync(
 
         try
         {
-            lock (_gate)
+            lock (_captureGate)
             {
                 if (_capturing)
                     throw new InvalidOperationException("This Npcap session is already capturing.");
@@ -130,7 +133,7 @@ public async IAsyncEnumerable<ProcessBusCapturedFrame> CaptureAsync(
                 }
             }
 
-            lock (_gate)
+            lock (_captureGate)
                 _capturing = false;
 
             channel.Writer.TryComplete();
@@ -151,16 +154,56 @@ public void Dispose()
             // Best-effort cleanup only.
         }
 
-        _sendGate.Dispose();
+        _injectionGate.Dispose();
+        lock (_clockMapGate)
+        {
+            foreach (var clock in _svTransmitClocks.Values)
+                clock.Dispose();
+            _svTransmitClocks.Clear();
+        }
+
         _disposed = true;
     }
 
-    private async ValueTask PaceSampledValuesAsync(SvTransmitKey key, CancellationToken cancellationToken)
+    private async ValueTask InjectAsync(ReadOnlyMemory<byte> frame, CancellationToken cancellationToken)
+    {
+        // Keep the device critical section intentionally short. PTP and GOOSE may pass
+        // while another SV stream is waiting for its pacing deadline.
+        await _injectionGate.WaitAsync(cancellationToken).ConfigureAwait(false);
+        try
+        {
+            _injectionDevice.SendPacket(frame.ToArray());
+        }
+        finally
+        {
+            _injectionGate.Release();
+        }
+    }
+
+    private SvTransmitClock GetOrCreateClock(SvTransmitKey key)
     {
-        if (!_svTransmitClocks.TryGetValue(key, out var clock) || clock.NominalIntervalTicks <= 0)
+        lock (_clockMapGate)
+        {
+            if (_svTransmitClocks.TryGetValue(key, out var existing))
+                return existing;
+
+            var created = new SvTransmitClock(
+                MinimumLearnableIntervalTicks,
+                MaximumLearnableIntervalTicks);
+            _svTransmitClocks.Add(key, created);
+            return created;
+        }
+    }
+
+    private static async ValueTask PaceSampledValuesAsync(
+        SvTransmitClock clock,
+        CancellationToken cancellationToken)
+    {
+        var intervalTicks = clock.NominalIntervalTicks;
+        if (clock.LastSentTicks <= 0 || intervalTicks <= 0)
             return;
 
-        var targetTicks = clock.LastSentTicks + clock.NominalIntervalTicks;
+        var targetTicks = clock.LastSentTicks + intervalTicks;
         while (true)
         {
             cancellationToken.ThrowIfCancellationRequested();
@@ -183,30 +226,6 @@ await Task.Delay(
         }
     }
 
-    private void CommitSampledValuesSend(SvTransmitKey key, long sentTicks)
-    {
-        if (!_svTransmitClocks.TryGetValue(key, out var clock))
-        {
-            _svTransmitClocks[key] = new SvTransmitClock(sentTicks, 0);
-            return;
-        }
-
-        var observedInterval = sentTicks - clock.LastSentTicks;
-        var nominalInterval = clock.NominalIntervalTicks;
-        var learnable = observedInterval >= MinimumLearnableIntervalTicks &&
-                        observedInterval <= MaximumLearnableIntervalTicks &&
-                        (nominalInterval <= 0 || observedInterval <= nominalInterval * 3);
-
-        if (learnable)
-        {
-            nominalInterval = nominalInterval <= 0
-                ? observedInterval
-                : (long)Math.Round((nominalInterval * 0.9) + (observedInterval * 0.1));
-        }
-
-        _svTransmitClocks[key] = new SvTransmitClock(sentTicks, nominalInterval);
-    }
-
     private static bool TryReadSampledValuesKey(ReadOnlySpan<byte> frame, out SvTransmitKey key)
     {
         key = default;
@@ -245,5 +264,27 @@ private static DateTimeOffset ToDateTimeOffset(PosixTimeval timeval)
     }
 
     private readonly record struct SvTransmitKey(ulong MacPrefix, uint MacSuffix, ushort AppId, ushort VlanId);
-    private sealed record SvTransmitClock(long LastSentTicks, long NominalIntervalTicks);
+
+    private sealed class SvTransmitClock : IDisposable
+    {
+        private readonly SvTransmitIntervalEstimator _estimator;
+
+        public SvTransmitClock(long minimumIntervalTicks, long maximumIntervalTicks)
+        {
+            _estimator = new SvTransmitIntervalEstimator(minimumIntervalTicks, maximumIntervalTicks);
+        }
+
+        public SemaphoreSlim PacingGate { get; } = new(1, 1);
+        public long LastSentTicks { get; private set; }
+        public long NominalIntervalTicks => _estimator.NominalIntervalTicks;
+
+        public void Commit(long sentTicks)
+        {
+            if (LastSentTicks > 0)
+                _estimator.Observe(sentTicks - LastSentTicks);
+            LastSentTicks = sentTicks;
+        }
+
+        public void Dispose() => PacingGate.Dispose();
+    }
 }

src/AR.Iec61850.Transports.Npcap/SvTransmitIntervalEstimator.cs

@@ -0,0 +1,89 @@
+namespace AR.Iec61850.Transports.Npcap;
+
+/// <summary>
+/// Learns a stable SV transmit interval without allowing one scheduler-late observation
+/// to become the permanent wire rate. The estimator activates only after several
+/// mutually consistent intervals have been observed.
+/// </summary>
+internal sealed class SvTransmitIntervalEstimator
+{
+    private readonly long _minimumIntervalTicks;
+    private readonly long _maximumIntervalTicks;
+    private readonly int _requiredConsistentIntervals;
+    private long _candidateIntervalTicks;
+    private int _candidateCount;
+
+    public SvTransmitIntervalEstimator(
+        long minimumIntervalTicks,
+        long maximumIntervalTicks,
+        int requiredConsistentIntervals = 4)
+    {
+        if (minimumIntervalTicks <= 0)
+            throw new ArgumentOutOfRangeException(nameof(minimumIntervalTicks));
+        if (maximumIntervalTicks < minimumIntervalTicks)
+            throw new ArgumentOutOfRangeException(nameof(maximumIntervalTicks));
+        if (requiredConsistentIntervals < 2)
+            throw new ArgumentOutOfRangeException(nameof(requiredConsistentIntervals));
+
+        _minimumIntervalTicks = minimumIntervalTicks;
+        _maximumIntervalTicks = maximumIntervalTicks;
+        _requiredConsistentIntervals = requiredConsistentIntervals;
+    }
+
+    public long NominalIntervalTicks { get; private set; }
+    public int CandidateCount => _candidateCount;
+
+    public void Observe(long intervalTicks)
+    {
+        if (intervalTicks < _minimumIntervalTicks || intervalTicks > _maximumIntervalTicks)
+        {
+            if (NominalIntervalTicks == 0)
+                ResetCandidate();
+            return;
+        }
+
+        if (NominalIntervalTicks > 0)
+        {
+            var minimumAccepted = NominalIntervalTicks * 3 / 4;
+            var maximumAccepted = NominalIntervalTicks * 3 / 2;
+            if (intervalTicks >= minimumAccepted && intervalTicks <= maximumAccepted)
+            {
+                NominalIntervalTicks = (long)Math.Round(
+                    (NominalIntervalTicks * 0.9) + (intervalTicks * 0.1));
+            }
+
+            return;
+        }
+
+        if (_candidateCount == 0)
+        {
+            _candidateIntervalTicks = intervalTicks;
+            _candidateCount = 1;
+            return;
+        }
+
+        var tolerance = Math.Max(
+            _minimumIntervalTicks / 2,
+            (long)Math.Round(_candidateIntervalTicks * 0.15));
+        if (Math.Abs(intervalTicks - _candidateIntervalTicks) > tolerance)
+        {
+            _candidateIntervalTicks = intervalTicks;
+            _candidateCount = 1;
+            return;
+        }
+
+        _candidateIntervalTicks = (long)Math.Round(
+            ((_candidateIntervalTicks * _candidateCount) + intervalTicks) /
+            (double)(_candidateCount + 1));
+        _candidateCount++;
+
+        if (_candidateCount >= _requiredConsistentIntervals)
+            NominalIntervalTicks = _candidateIntervalTicks;
+    }
+
+    private void ResetCandidate()
+    {
+        _candidateIntervalTicks = 0;
+        _candidateCount = 0;
+    }
+}

tests/AR.Iec61850.Tests/AR.Iec61850.Tests.csproj

@@ -25,6 +25,7 @@
   <ItemGroup>
     <ProjectReference Include="..\..\src\AR.Iec61850\AR.Iec61850.csproj" />
     <ProjectReference Include="..\..\src\AR.Iec61850.Simulation\AR.Iec61850.Simulation.csproj" />
+    <ProjectReference Include="..\..\src\AR.Iec61850.Transports.Npcap\AR.Iec61850.Transports.Npcap.csproj" />
   </ItemGroup>
 
   <ItemGroup>

tests/AR.Iec61850.Tests/Transports/SvTransmitIntervalEstimatorTests.cs

@@ -0,0 +1,61 @@
+using AR.Iec61850.Transports.Npcap;
+
+namespace AR.Iec61850.Tests.Transports;
+
+public sealed class SvTransmitIntervalEstimatorTests
+{
+    [Fact]
+    public void SingleLateObservation_DoesNotBecomePermanentNominalRate()
+    {
+        var estimator = new SvTransmitIntervalEstimator(20, 5_000, requiredConsistentIntervals: 4);
+
+        estimator.Observe(1_000); // Scheduler-late first interval.
+        Assert.Equal(0, estimator.NominalIntervalTicks);
+
+        estimator.Observe(250);
+        estimator.Observe(248);
+        estimator.Observe(252);
+        Assert.Equal(0, estimator.NominalIntervalTicks);
+
+        estimator.Observe(251);
+
+        Assert.InRange(estimator.NominalIntervalTicks, 248, 252);
+    }
+
+    [Fact]
+    public void InconsistentIntervals_DoNotActivatePacing()
+    {
+        var estimator = new SvTransmitIntervalEstimator(20, 5_000, requiredConsistentIntervals: 4);
+
+        foreach (var interval in new long[] { 250, 800, 240, 1_200, 260, 700 })
+            estimator.Observe(interval);
+
+        Assert.Equal(0, estimator.NominalIntervalTicks);
+        Assert.Equal(1, estimator.CandidateCount);
+    }
+
+    [Fact]
+    public void ActiveNominalRate_IgnoresLongSchedulerStall()
+    {
+        var estimator = new SvTransmitIntervalEstimator(20, 5_000, requiredConsistentIntervals: 4);
+        foreach (var interval in new long[] { 250, 249, 251, 250 })
+            estimator.Observe(interval);
+
+        var nominalBeforeStall = estimator.NominalIntervalTicks;
+        estimator.Observe(4_000);
+
+        Assert.Equal(nominalBeforeStall, estimator.NominalIntervalTicks);
+    }
+
+    [Fact]
+    public void ActiveNominalRate_TracksOnlyNearbySteadyIntervals()
+    {
+        var estimator = new SvTransmitIntervalEstimator(20, 5_000, requiredConsistentIntervals: 4);
+        foreach (var interval in new long[] { 250, 250, 250, 250 })
+            estimator.Observe(interval);
+
+        estimator.Observe(260);
+
+        Assert.InRange(estimator.NominalIntervalTicks, 250, 252);
+    }
+}