Auto: twincat-3.1 — per-tag MaxDelay tuning

Closes #313

docs/Driver.TwinCAT.Cli.md

@@ -103,10 +103,37 @@ otopcua-twincat-cli subscribe -n 192.168.1.40.1.1 -s GVL.Counter -t DInt -i 500
 
 # Fall back to polling for runtimes where native notifications are constrained
 otopcua-twincat-cli subscribe -n 192.168.1.40.1.1 -s GVL.Counter -t DInt -i 500 --poll-only
+
+# Coalesce bursty changes — runtime buffers up to 500 ms before dispatch
+otopcua-twincat-cli subscribe -n 192.168.1.40.1.1 -s GVL.Counter -t DInt -i 50 --max-delay-ms 500
 ```
 
+| Flag | Default | Purpose |
+|---|---|---|
+| `-s` / `--symbol` | **required** | Symbol path — same format as `read` |
+| `-t` / `--type` | `DInt` | IEC type (see Data types section) |
+| `-i` / `--interval-ms` | `1000` | **Cycle time** — minimum interval between change checks the PLC runtime applies |
+| `--max-delay-ms` | `0` | **Max coalescing window** — upper bound on how long the runtime buffers change events before dispatch. `0` = fire ASAP, no coalescing |
+| `--poll-only` | off | Disable native notifications, use `PollGroupEngine` instead |
+
+`-i` / `--interval-ms` and `--max-delay-ms` are different things and both flow
+into the Beckhoff `NotificationSettings` ctor:
+
+- **`--interval-ms`** is the *cycle*: the runtime checks for value changes at
+  most this often. Smaller = lower latency, higher CPU.
+- **`--max-delay-ms`** is the *coalescing ceiling*: once a change is detected,
+  the runtime can hold it for up to this long before dispatching, which lets
+  it batch a burst of changes into a single callback. Default `0` means
+  every detected change fires immediately — same as the pre-PR-3.1 behaviour.
+
+For high-frequency signals (a counter incrementing every 10 ms PLC cycle),
+pair a small `-i` (so latency stays bounded) with a non-zero `--max-delay-ms`
+(so the OPC UA queue downstream doesn't flood). For slow signals just leave
+`--max-delay-ms` at `0`.
+
 The subscribe banner announces which mechanism is in play — "ADS notification"
-or "polling" — so it's obvious in screen-recorded bug reports.
+or "polling" — and includes the `max-delay` value when set, so it's obvious
+in screen-recorded bug reports.
 
 `--poll-only` polls go through the same cached-handle path as `read`, so
 repeated polls of the same symbol carry only a 4-byte handle on the wire

docs/drivers/TwinCAT-Test-Fixture.md

@@ -96,6 +96,16 @@ CPU load or network jitter real notifications can coalesce. The fake fires
 one callback per test invocation — real callback-coalescing behavior is
 untested.
 
+PR 3.1 (#313) makes the per-tag `MaxDelay` configurable via
+`TwinCATTagDefinition.MaxDelayMs` — the runtime can buffer changes for up to
+that many milliseconds before dispatch, deliberately coalescing bursty
+high-frequency signals so the OPC UA queue downstream doesn't flood. Default
+`null` / `0` preserves the pre-PR-3.1 "fire ASAP" behaviour.
+`TwinCATMaxDelayTests.Driver_coalesces_notifications_at_max_delay` exercises
+the wire-side coalescer end-to-end against `GVL_Fixture.nCounter`; the unit
+suite (`TwinCATNativeNotificationTests`) covers the plumbing contract via
+the `FakeTwinCATClient.FakeNotification.MaxDelayMs` capture.
+
 ### 4. TC2 vs TC3 variant handling
 
 TwinCAT 2 (ADS v1) and TwinCAT 3 (ADS v2) have subtly different

src/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Cli/Commands/SubscribeCommand.cs

@@ -20,9 +20,18 @@ public sealed class SubscribeCommand : TwinCATCommandBase
         "String / WString / Time / Date / DateTime / TimeOfDay (default DInt).")]
     public TwinCATDataType DataType { get; init; } = TwinCATDataType.DInt;
 
-    [CommandOption("interval-ms", 'i', Description = "Publishing interval ms (default 1000).")]
+    [CommandOption("interval-ms", 'i', Description =
+        "Cycle time ms — minimum interval between change checks the PLC runtime applies " +
+        "(default 1000). Different from --max-delay-ms; see help for that flag.")]
     public int IntervalMs { get; init; } = 1000;
 
+    [CommandOption("max-delay-ms", Description =
+        "Per-tag MaxDelay in ms (PR 3.1, issue #313) — upper bound on how long the runtime " +
+        "can buffer / coalesce change events before dispatch. 0 (default) = fire ASAP, no " +
+        "coalescing. Larger values (e.g. 500) reduce event rate on bursty signals so the " +
+        "OPC UA queue downstream doesn't flood. Distinct from --interval-ms (the cycle).")]
+    public int MaxDelayMs { get; init; } = 0;
+
     public override async ValueTask ExecuteAsync(IConsole console)
     {
         ConfigureLogging();
@@ -34,7 +43,8 @@ public sealed class SubscribeCommand : TwinCATCommandBase
             DeviceHostAddress: Gateway,
             SymbolPath: SymbolPath,
             DataType: DataType,
-            Writable: false);
+            Writable: false,
+            MaxDelayMs: MaxDelayMs > 0 ? MaxDelayMs : null);
         var options = BuildOptions([tag]);
 
         await using var driver = new TwinCATDriver(options, DriverInstanceId);
@@ -54,8 +64,9 @@ public sealed class SubscribeCommand : TwinCATCommandBase
             handle = await driver.SubscribeAsync([tagName], TimeSpan.FromMilliseconds(IntervalMs), ct);
 
             var mode = PollOnly ? "polling" : "ADS notification";
+            var coalesce = MaxDelayMs > 0 ? $", max-delay {MaxDelayMs}ms" : "";
             await console.Output.WriteLineAsync(
-                $"Subscribed to {SymbolPath} @ {IntervalMs}ms ({mode}). Ctrl+C to stop.");
+                $"Subscribed to {SymbolPath} @ {IntervalMs}ms ({mode}{coalesce}). Ctrl+C to stop.");
             try
             {
                 await Task.Delay(System.Threading.Timeout.InfiniteTimeSpan, ct);

src/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/AdsTwinCATClient.cs

@@ -434,6 +434,7 @@ internal sealed class AdsTwinCATClient : ITwinCATClient
         TwinCATDataType type,
         int? bitIndex,
         TimeSpan cycleTime,
+        int maxDelayMs,
         Action<string, object?> onChange,
         CancellationToken cancellationToken)
     {
@@ -442,8 +443,14 @@ internal sealed class AdsTwinCATClient : ITwinCATClient
         // fires when the value differs; OnCycle fires every cycle. OnChange is the right default
         // for OPC UA data-change semantics — the PLC already has the best view of "has this
         // changed" so we let it decide.
+        //
+        // PR 3.1 (#313) — maxDelay is now per-tag tunable. 0 = "fire ASAP, no coalescing"
+        // (the pre-PR-3.1 default). Larger values let the runtime coalesce bursty changes
+        // so the OPC UA queue downstream doesn't flood under high-frequency signals. Same
+        // unit (100ns ticks) as cycleTicks; convert from the caller-supplied milliseconds.
         var cycleTicks = (uint)Math.Max(1, cycleTime.Ticks / TimeSpan.TicksPerMillisecond * 10_000);
-        var settings = new NotificationSettings(AdsTransMode.OnChange, (int)cycleTicks, 0);
+        var maxDelayTicks = Math.Max(0, maxDelayMs) * 10_000;
+        var settings = new NotificationSettings(AdsTransMode.OnChange, (int)cycleTicks, maxDelayTicks);
 
         // AddDeviceNotificationExAsync returns Task<ResultHandle>; AdsNotificationEx fires
         // with the handle as part of the event args so we use the handle as the correlation

src/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/ITwinCATClient.cs

@@ -96,6 +96,13 @@ public interface ITwinCATClient : IDisposable
     /// <param name="type">Declared type; drives the native layout + callback value boxing.</param>
     /// <param name="bitIndex">For BOOL-within-word tags — the bit to extract from the parent word.</param>
     /// <param name="cycleTime">Minimum interval between change notifications (native-floor depends on target).</param>
+    /// <param name="maxDelayMs">
+    ///     Per-tag <c>MaxDelay</c> in milliseconds — the upper bound on how long the runtime can
+    ///     buffer / coalesce change events before dispatching them (PR 3.1 / issue #313).
+    ///     <c>0</c> means "fire ASAP, no coalescing" (the pre-PR-3.1 default behaviour); larger
+    ///     values let the runtime coalesce bursty high-frequency changes so the OPC UA queue
+    ///     downstream doesn't flood. Plumbs straight into <c>NotificationSettings(mode, cycleTime, maxDelay)</c>.
+    /// </param>
     /// <param name="onChange">Invoked with <c>(symbolPath, boxedValue)</c> per notification.</param>
     /// <param name="cancellationToken">Cancels the initial registration; does not tear down an established notification.</param>
     Task<ITwinCATNotificationHandle> AddNotificationAsync(
@@ -103,6 +110,7 @@ public interface ITwinCATClient : IDisposable
         TwinCATDataType type,
         int? bitIndex,
         TimeSpan cycleTime,
+        int maxDelayMs,
         Action<string, object?> onChange,
         CancellationToken cancellationToken);
 

src/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriver.cs

@@ -472,8 +472,11 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
                 var symbolName = parsed?.ToAdsSymbolName() ?? def.SymbolPath;
                 var bitIndex = parsed?.BitIndex;
 
+                // PR 3.1 (#313) — pass the per-tag MaxDelay (default 0 = current behaviour)
+                // through to NotificationSettings so the PLC can coalesce bursty changes.
+                var maxDelayMs = def.MaxDelayMs ?? 0;
                 var reg = await client.AddNotificationAsync(
-                    symbolName, def.DataType, bitIndex, publishingInterval,
+                    symbolName, def.DataType, bitIndex, publishingInterval, maxDelayMs,
                     (_, value) => OnDataChange?.Invoke(this,
                         new DataChangeEventArgs(handle, reference, new DataValueSnapshot(
                             value, TwinCATStatusMapper.Good, DateTime.UtcNow, DateTime.UtcNow))),

src/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT/TwinCATDriverOptions.cs

@@ -50,6 +50,17 @@ public sealed record TwinCATDeviceOptions(
 ///     only whole-array support; multi-dim shapes flatten to the product on the wire and the
 ///     OPC UA layer reflects the rank via its own <c>ArrayDimensions</c> metadata.
 /// </summary>
+/// <remarks>
+///     <para><b>MaxDelayMs</b> (PR 3.1, issue #313) is the optional per-tag <c>MaxDelay</c>
+///     that flows into the native ADS <c>NotificationSettings(mode, cycleTime, maxDelay)</c>
+///     ctor. <c>maxDelay</c> is the upper bound on how long the runtime can buffer /
+///     coalesce change events before dispatching them. <c>null</c> / <c>0</c> (default)
+///     means "fire ASAP, no coalescing" — same as the pre-PR-3.1 behaviour. Set to a larger
+///     value (e.g. <c>500</c> ms) for bursty high-frequency signals where the OPC UA queue
+///     would otherwise flood. Has no effect when <c>UseNativeNotifications=false</c> (the
+///     polled fallback uses <c>PollGroupEngine</c>'s own publishing interval). Ignored for
+///     whole-array tags — those bypass the native-notification path entirely.</para>
+/// </remarks>
 public sealed record TwinCATTagDefinition(
     string Name,
     string DeviceHostAddress,
@@ -57,7 +68,8 @@ public sealed record TwinCATTagDefinition(
     TwinCATDataType DataType,
     bool Writable = true,
     bool WriteIdempotent = false,
-    int[]? ArrayDimensions = null);
+    int[]? ArrayDimensions = null,
+    int? MaxDelayMs = null);
 
 public sealed class TwinCATProbeOptions
 {

tests/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.IntegrationTests/TwinCATMaxDelayTests.cs

@@ -0,0 +1,83 @@
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Driver.TwinCAT;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.IntegrationTests;
+
+/// <summary>
+///     PR 3.1 (#313) — per-tag <c>MaxDelay</c> coalescing observed on the wire. Subscribes
+///     to <c>GVL_Fixture.nCounter</c> (the cycle-incrementing fixture counter) with a
+///     500 ms <c>MaxDelayMs</c>; over a 1 s observation window the runtime should batch
+///     the per-cycle changes into ≤ 3 callbacks rather than one-per-PLC-cycle (~100 with a
+///     10 ms task) which is what <c>MaxDelay=0</c> would deliver.
+/// </summary>
+/// <remarks>
+///     <para>Skipped via <see cref="TwinCATFactAttribute"/> when the XAR VM isn't reachable
+///     / env vars aren't set. Build-only proof in CI; full runtime cover happens on the
+///     XAR-equipped lab box.</para>
+///
+///     <para>The 1 s window + ≤ 3 events tolerance is generous on purpose — the runtime's
+///     coalescer can still fire mid-window when the cycle accumulates a value-change burst,
+///     and the test mostly cares that <i>coalescing happens at all</i> (i.e. it's
+///     dramatically less than the no-coalescing baseline).</para>
+/// </remarks>
+[Collection("TwinCATXar")]
+[Trait("Category", "Integration")]
+[Trait("Simulator", "TwinCAT-XAR")]
+public sealed class TwinCATMaxDelayTests(TwinCATXarFixture sim)
+{
+    [TwinCATFact]
+    public async Task Driver_coalesces_notifications_at_max_delay()
+    {
+        if (sim.SkipReason is not null) Assert.Skip(sim.SkipReason);
+
+        var options = BuildOptions(sim);
+        await using var drv = new TwinCATDriver(options, driverInstanceId: "tc3-maxdelay");
+        await drv.InitializeAsync("{}", TestContext.Current.CancellationToken);
+
+        var observed = new List<DateTime>();
+        drv.OnDataChange += (_, _) =>
+        {
+            lock (observed) observed.Add(DateTime.UtcNow);
+        };
+
+        var handle = await drv.SubscribeAsync(
+            ["Counter"], TimeSpan.FromMilliseconds(50),
+            TestContext.Current.CancellationToken);
+
+        // Observe for 1 s. With MAIN incrementing nCounter every 10 ms PLC cycle, the
+        // un-coalesced rate would be ~100 events; with MaxDelay=500 ms we expect ≤ 3
+        // (initial fire + at most two coalesce-window flushes).
+        await Task.Delay(TimeSpan.FromMilliseconds(1000), TestContext.Current.CancellationToken);
+
+        await drv.UnsubscribeAsync(handle, TestContext.Current.CancellationToken);
+
+        int count;
+        lock (observed) count = observed.Count;
+        count.ShouldBeLessThanOrEqualTo(
+            3,
+            $"MaxDelayMs=500 must coalesce nCounter changes — observed {count} events in 1 s");
+    }
+
+    private static TwinCATDriverOptions BuildOptions(TwinCATXarFixture sim) => new()
+    {
+        Devices = [
+            new TwinCATDeviceOptions(
+                HostAddress: $"ads://{sim.TargetNetId}:{sim.AmsPort}",
+                DeviceName: "XAR-VM"),
+        ],
+        Tags = [
+            new TwinCATTagDefinition(
+                Name: "Counter",
+                DeviceHostAddress: $"ads://{sim.TargetNetId}:{sim.AmsPort}",
+                SymbolPath: "GVL_Fixture.nCounter",
+                DataType: TwinCATDataType.DInt,
+                MaxDelayMs: 500),
+        ],
+        UseNativeNotifications = true,
+        Timeout = TimeSpan.FromSeconds(5),
+        // Disable probe so it doesn't race with the 1 s observation window.
+        Probe = new TwinCATProbeOptions { Enabled = false },
+    };
+}

tests/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.IntegrationTests/TwinCatProject/README.md

@@ -71,6 +71,12 @@ GVL_Fixture.nCounter := GVL_Fixture.nCounter + 1;
 - `PlcTask` — cyclic, 10 ms interval, priority 20
 - Assigned to `MAIN`
 
+> **Note (PR 3.1 / #313)**: `GVL_Fixture.nCounter` doubles as the
+> coalescing-test driver for `TwinCATMaxDelayTests`. The 10 ms cycle +
+> per-cycle increment in `MAIN` means a no-coalescing subscriber sees ~100
+> events / s; with `MaxDelayMs=500` the test asserts ≤ 3 events / s. No new
+> project state required.
+
 ## Performance scenarios
 
 PR 2.1 (ADS Sum-read / Sum-write) ships an opt-in perf-tier integration test

tests/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/FakeTwinCATClient.cs

@@ -315,12 +315,13 @@ internal class FakeTwinCATClient : ITwinCATClient
 
     public virtual Task<ITwinCATNotificationHandle> AddNotificationAsync(
         string symbolPath, TwinCATDataType type, int? bitIndex, TimeSpan cycleTime,
+        int maxDelayMs,
         Action<string, object?> onChange, CancellationToken cancellationToken)
     {
         if (ThrowOnAddNotification)
             throw Exception ?? new InvalidOperationException("fake AddNotification failure");
 
-        var reg = new FakeNotification(symbolPath, type, bitIndex, onChange, this);
+        var reg = new FakeNotification(symbolPath, type, bitIndex, cycleTime, maxDelayMs, onChange, this);
         Notifications.Add(reg);
         return Task.FromResult<ITwinCATNotificationHandle>(reg);
     }
@@ -352,11 +353,16 @@ internal class FakeTwinCATClient : ITwinCATClient
 
     public sealed class FakeNotification(
         string symbolPath, TwinCATDataType type, int? bitIndex,
+        TimeSpan cycleTime, int maxDelayMs,
         Action<string, object?> onChange, FakeTwinCATClient owner) : ITwinCATNotificationHandle
     {
         public string SymbolPath { get; } = symbolPath;
         public TwinCATDataType Type { get; } = type;
         public int? BitIndex { get; } = bitIndex;
+        /// <summary>Cycle time the driver requested (PR 3.1 — captured for tests).</summary>
+        public TimeSpan CycleTime { get; } = cycleTime;
+        /// <summary>Per-tag MaxDelay in ms (PR 3.1 / #313). 0 = no coalescing.</summary>
+        public int MaxDelayMs { get; } = maxDelayMs;
         public Action<string, object?> OnChange { get; } = onChange;
         public bool Disposed { get; private set; }
 

tests/ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.Tests/TwinCATNativeNotificationTests.cs

@@ -137,12 +137,13 @@ public sealed class TwinCATNativeNotificationTests
 
         public override Task<ITwinCATNotificationHandle> AddNotificationAsync(
             string symbolPath, TwinCATDataType type, int? bitIndex, TimeSpan cycleTime,
+            int maxDelayMs,
             Action<string, object?> onChange, CancellationToken cancellationToken)
         {
             AddCallCount++;
             if (AddCallCount > _succeedBefore)
                 throw new InvalidOperationException($"fake fail on call #{AddCallCount}");
-            return base.AddNotificationAsync(symbolPath, type, bitIndex, cycleTime, onChange, cancellationToken);
+            return base.AddNotificationAsync(symbolPath, type, bitIndex, cycleTime, maxDelayMs, onChange, cancellationToken);
         }
     }
 
@@ -187,6 +188,84 @@ public sealed class TwinCATNativeNotificationTests
         await drv.UnsubscribeAsync(handle, CancellationToken.None);
     }
 
+    // ---- PR 3.1 (#313) — per-tag MaxDelay tuning ----
+
+    [Fact]
+    public async Task Native_subscribe_default_MaxDelay_is_zero_per_tag()
+    {
+        // No MaxDelayMs on the tag → fake captures 0 (the pre-PR-3.1 "fire ASAP" default).
+        var (drv, factory) = NewNativeDriver(
+            new TwinCATTagDefinition("A", "ads://5.23.91.23.1.1:851", "MAIN.A", TwinCATDataType.DInt));
+        await drv.InitializeAsync("{}", CancellationToken.None);
+
+        _ = await drv.SubscribeAsync(["A"], TimeSpan.FromMilliseconds(100), CancellationToken.None);
+
+        factory.Clients[0].Notifications.Count.ShouldBe(1);
+        factory.Clients[0].Notifications[0].MaxDelayMs.ShouldBe(0);
+    }
+
+    [Fact]
+    public async Task Native_subscribe_plumbs_per_tag_MaxDelayMs_into_NotificationSettings()
+    {
+        var (drv, factory) = NewNativeDriver(
+            new TwinCATTagDefinition("A", "ads://5.23.91.23.1.1:851", "MAIN.A", TwinCATDataType.DInt,
+                MaxDelayMs: 500));
+        await drv.InitializeAsync("{}", CancellationToken.None);
+
+        _ = await drv.SubscribeAsync(["A"], TimeSpan.FromMilliseconds(100), CancellationToken.None);
+
+        factory.Clients[0].Notifications.Count.ShouldBe(1);
+        factory.Clients[0].Notifications[0].MaxDelayMs.ShouldBe(500);
+    }
+
+    [Fact]
+    public async Task Native_subscribe_per_tag_MaxDelay_values_flow_independently()
+    {
+        // Mixing tags in a single Subscribe call: each tag's MaxDelayMs lands on its own
+        // NotificationSettings, no cross-talk between subscriptions.
+        var (drv, factory) = NewNativeDriver(
+            new TwinCATTagDefinition("A", "ads://5.23.91.23.1.1:851", "MAIN.A", TwinCATDataType.DInt,
+                MaxDelayMs: 250),
+            new TwinCATTagDefinition("B", "ads://5.23.91.23.1.1:851", "MAIN.B", TwinCATDataType.DInt,
+                MaxDelayMs: 1000),
+            new TwinCATTagDefinition("C", "ads://5.23.91.23.1.1:851", "MAIN.C", TwinCATDataType.DInt));
+        await drv.InitializeAsync("{}", CancellationToken.None);
+
+        _ = await drv.SubscribeAsync(["A", "B", "C"], TimeSpan.FromMilliseconds(50), CancellationToken.None);
+
+        factory.Clients[0].Notifications.Count.ShouldBe(3);
+        var bySymbol = factory.Clients[0].Notifications
+            .ToDictionary(n => n.SymbolPath, n => n.MaxDelayMs, StringComparer.OrdinalIgnoreCase);
+        bySymbol["MAIN.A"].ShouldBe(250);
+        bySymbol["MAIN.B"].ShouldBe(1000);
+        bySymbol["MAIN.C"].ShouldBe(0); // unset = default
+    }
+
+    [Fact]
+    public void TagDefinition_MaxDelayMs_default_is_null()
+    {
+        // Defaulting MaxDelayMs on the record itself — preserves the call sites that don't
+        // set it (every existing TwinCATTagDefinition usage in tests + production).
+        var def = new TwinCATTagDefinition(
+            "X", "ads://5.23.91.23.1.1:851", "MAIN.X", TwinCATDataType.DInt);
+        def.MaxDelayMs.ShouldBeNull();
+    }
+
+    [Fact]
+    public void TagDefinition_MaxDelayMs_round_trips_via_record_with()
+    {
+        // The `with` expression is the closest thing to a DTO round-trip at this layer —
+        // confirms the new property is a regular init member that participates in equality
+        // / copy semantics. Caller-side serialisation layers (server / admin) layer their
+        // own DTOs on top of this record.
+        var def = new TwinCATTagDefinition(
+            "X", "ads://5.23.91.23.1.1:851", "MAIN.X", TwinCATDataType.DInt);
+        var withDelay = def with { MaxDelayMs = 750 };
+        withDelay.MaxDelayMs.ShouldBe(750);
+        // sanity — original is untouched
+        def.MaxDelayMs.ShouldBeNull();
+    }
+
     [Fact]
     public async Task Subscribe_handle_DiagnosticId_indicates_native_vs_poll()
     {