diff --git a/docs/drivers/MTConnect.md b/docs/drivers/MTConnect.md index 9e338bb0..49f2c72b 100644 --- a/docs/drivers/MTConnect.md +++ b/docs/drivers/MTConnect.md @@ -73,7 +73,7 @@ below). ## Capability surface -`MTConnectDriver : IDriver, IReadable, IWritable, ISubscribable, ITagDiscovery, IHostConnectivityProbe, IRediscoverable` +`MTConnectDriver : IDriver, IReadable, IWritable, ISubscribable, ITagDiscovery, IHostConnectivityProbe, IRediscoverable, IDriverDiagnostics` (`src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs`). `IWritable` landed 2026-08-05 (`b29623cf`), against the Agent's `AllowPut` @@ -92,6 +92,7 @@ request/response handshake — is a different mechanism and remains deferred; se | `ISubscribable` | `SubscribeAsync`/`OnDataChange` — the shared `/sample` long-poll pump | **The production data path** | | `IHostConnectivityProbe` | periodic `/probe` under `Probe.*` | No consumer wires it today — see [Known limitations](#known-limitations) | | `IRediscoverable` | watches the Agent's `Header.instanceId` | No consumer wires it today — see [Known limitations](#known-limitations) | +| `IDriverDiagnostics` | `GetDiagnostics()` — `/sample` effective count, lag, saturated chunks, count raises | Shown on `/hosts` (review 2026-10-21 T-5) | | `IAlarmSource` | CONDITION observations off the same `/sample` pump → Part 9 transitions keyed by the alarm tag's RawPath | Opt-in per raw tag (`alarm` object) — see [CONDITION alarms](#condition-alarms-626) | ### Feature status at a glance (#613) @@ -159,7 +160,7 @@ Read off `MTConnectDriverOptions` / `MTConnectDriverConfigDto` | `DeviceName` | `null` | **Double duty.** (1) Scopes requests to `{AgentUri}/{DeviceName}/...`; `null` = whole Agent. (2) The write path's fallback device — the device a write is addressed to when the probe model does not say which device declared the DataItem (see [Write-back](#write-back--observation-injection-off-by-default)) | | `RequestTimeoutMs` | `5000` | Per-call deadline for `/probe` and `/current` | | `SampleIntervalMs` | `1000` | The Agent's `/sample?interval=` query param | -| `SampleCount` | `1000` | The Agent's `/sample?count=` query param | +| `SampleCount` | `1000` | The Agent's `/sample?count=` query param. **Caps subscription throughput** at `SampleCount / SampleIntervalMs` — see [Subscribe](#subscribe--the-sample-pump) for the adaptive raise | | `HeartbeatMs` | `10000` | The Agent's `/sample?heartbeat=` query param — also the watchdog's liveness window | | `Probe.Enabled` / `Probe.IntervalMs` / `Probe.TimeoutMs` | `true` / `5000` / `2000` | Background connectivity-probe knobs (mirrors `ModbusProbeOptions`) | | `Reconnect.MinBackoffMs` / `MaxBackoffMs` / `BackoffMultiplier` | `0` / `30000` / `2.0` | Geometric backoff after a failed request or a dropped `/sample` stream | @@ -333,6 +334,17 @@ the two need disambiguating, not just OR-ing together). On a changed `/hosts` page — see [Known limitations](#known-limitations) for what that does and does not do. +**Lag measurement and the adaptive `count` (review 2026-10-21 T-5).** Throughput is capped at `SampleCount` +observations per chunk, one chunk per `SampleIntervalMs`; a machine producing more used to fall behind the ring +buffer forever, re-baselining every buffer-length (dropping intermediate transitions) with nothing surfacing it. +Each chunk is now measured: `lag = lastSequence + 1 − nextSequence` (unknown when the Agent omits `lastSequence`) +and whether it was **saturated** (carried the full `count`). Three consecutive saturated chunks with a positive (or +unknown) lag reopen the stream at the cursor reached with `count` doubled, up to 16× `SampleCount` (max 65 536). A +raised stream that delivers nothing (an Agent refusing the larger count) steps back to the last count that delivered +and stops raising; at the ceiling a still-behind stream logs one Warning per pump. Every new pump starts from the +configured `SampleCount`. The driver now implements `IDriverDiagnostics`: `sampleCount` (effective), `sampleLag`, +`sampleSaturatedChunks` and `sampleCountRaises` ride the `/hosts` diagnostics bag. + **A pump carries the session generation it was started against (review 2026-10-21 T-1).** A lifecycle restart retires the running pump (or abandons it after its bounded teardown wait); if that pump finishes after the new session is installed, its cursor publish, a late chunk and a late re-baseline are all dropped — checked under the diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Contracts/MTConnectDriverOptions.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Contracts/MTConnectDriverOptions.cs index da204235..d46bfaf6 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Contracts/MTConnectDriverOptions.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Contracts/MTConnectDriverOptions.cs @@ -44,6 +44,13 @@ public sealed class MTConnectDriverOptions /// The count query parameter passed to the Agent's /sample request — the /// maximum number of DataItem observations returned per chunk. Default 1000, the /// MTConnect Agent's own conventional default. + /// + /// It caps subscription throughput at SampleCount / SampleIntervalMs observations per + /// millisecond (default 1000 a second): a machine producing more falls behind the Agent's ring buffer. The + /// pump measures that lag and, when it is persistently behind, reopens the stream with the count doubled — up + /// to 16× this value, at most 65 536 — and reports the effective count and the lag on the /hosts + /// diagnostics (review 2026-10-21 T-5). Each new stream starts again from this value. + /// /// public int SampleCount { get; init; } = 1000; diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/IMTConnectAgentClient.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/IMTConnectAgentClient.cs index b845e122..3a1e7155 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/IMTConnectAgentClient.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/IMTConnectAgentClient.cs @@ -59,7 +59,7 @@ public sealed record MTConnectWriteOutcome(bool Accepted, string? AgentError); /// . An here would be a /// synchronous method wearing a ValueTask, and it would oblige every canned-XML fake /// to grow one too. The async unwind that a streaming client does need is already -/// covered elsewhere and is not this method's job: the enumerator +/// covered elsewhere and is not this method's job: the enumerator /// is itself and is owned by the pump that enumerates it, /// while on the client converts any read still in flight /// into an ordinary . @@ -79,7 +79,7 @@ public interface IMTConnectAgentClient : IDisposable /// /// Issues an Agent /current request and returns the latest observed value for every /// data item. Used both to prime the observation index at startup and to re-baseline after - /// a detected sequence gap in . + /// a detected sequence gap in . /// /// Cancellation token. /// The latest observation for every data item. @@ -113,6 +113,20 @@ public interface IMTConnectAgentClient : IDisposable /// An asynchronous sequence of streams results, one per multipart chunk. IAsyncEnumerable SampleAsync(long from, CancellationToken ct); + /// + /// As , but asking the Agent for up to + /// observations per chunk instead of the configured + /// — how the pump catches up when it falls behind an Agent + /// producing more than SampleCount / SampleIntervalMs observations a second (R06:T-5). The default + /// ignores , so a client that cannot honour it keeps working at the configured one. + /// + /// The sequence to open the stream at. + /// The per-chunk observation cap for this stream. + /// Cancellation token; the only contracted way the enumeration ends quietly. + /// The stream's chunks. + IAsyncEnumerable SampleAsync(long from, int count, CancellationToken ct) => + SampleAsync(from, ct); + /// /// Sets the Agent's observation for one DataItem, through the Agent's AllowPut /// surface (POST {agent}/{deviceName}?{dataItemId}={value}). @@ -183,7 +197,7 @@ public interface IMTConnectAgentClient : IDisposable /// /// /// is a running cursor, not the original - /// from. It equals the from passed to only + /// from. It equals the from passed to only /// for the first chunk; from the second chunk on it is the previous /// chunk's . Comparing every chunk /// against the original from instead reports a gap on every chunk once the @@ -199,7 +213,7 @@ public interface IMTConnectAgentClient : IDisposable /// a chunk the Agent was willing to send — and cppagent answers a from that has /// already fallen out of its buffer with an MTConnectError / OUT_OF_RANGE /// document served under HTTP 200, which this client surfaces as an - /// out of , never as a chunk. + /// out of , never as a chunk. /// A pump that re-baselines only on IsSequenceGap will therefore fault instead of /// recovering exactly when it has fallen furthest behind. Treat an OUT_OF_RANGE /// parse failure as a re-baseline trigger (re-prime from , diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectAgentClient.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectAgentClient.cs index 2fe22332..c16339c2 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectAgentClient.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectAgentClient.cs @@ -17,7 +17,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect; /// The constructor opens no connection. It only composes the request URIs and builds /// its s (which themselves dial nothing until a request is issued), /// so a throwaway instance is safe to construct — the universal discovery browser's -/// CanBrowse pattern depends on that. likewise dials +/// CanBrowse pattern depends on that. likewise dials /// nothing until it is enumerated. /// /// @@ -54,7 +54,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect; /// /// /// Disposal is safe against an in-flight enumeration. Re-initialization disposes and rebuilds this client, which can happen while a pump is still -/// enumerating . Rather than letting the in-flight read fault with +/// enumerating . Rather than letting the in-flight read fault with /// an unclassified or that no /// caller is written to expect, every request runs under a token linked to an internal /// dispose token, so disposal surfaces as an ordinary @@ -79,6 +79,8 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable private readonly string _agentRoot; private readonly string _sampleUriBase; private readonly string _sampleQuerySuffix; + private readonly int _sampleCount; + private readonly int _heartbeatMs; private readonly TimeSpan _requestTimeout; private readonly TimeSpan _streamIdleTimeout; @@ -151,9 +153,10 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable _probeUri = BuildRequestUri(options, "probe"); _currentUri = BuildRequestUri(options, "current"); _sampleUriBase = BuildRequestUri(options, "sample").AbsoluteUri; - _sampleQuerySuffix = string.Create( - CultureInfo.InvariantCulture, - $"&interval={options.SampleIntervalMs}&count={options.SampleCount}&heartbeat={options.HeartbeatMs}"); + // count is appended per request: the driver may raise it for one stream when it falls behind (R06:T-5). + _sampleQuerySuffix = string.Create(CultureInfo.InvariantCulture, $"&interval={options.SampleIntervalMs}"); + _sampleCount = options.SampleCount; + _heartbeatMs = options.HeartbeatMs; _http = handler is null ? new HttpClient() : new HttpClient(handler, disposeHandler: false); _http.Timeout = _requestTimeout; @@ -195,10 +198,32 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable public async IAsyncEnumerable SampleAsync( long from, [EnumeratorCancellation] CancellationToken ct) + { + await foreach (var chunk in SampleAsync(from, count: null, ct).ConfigureAwait(false)) + { + yield return chunk; + } + } + + /// + public IAsyncEnumerable SampleAsync(long from, int count, CancellationToken ct) + { + if (count <= 0) + { + throw new ArgumentOutOfRangeException(nameof(count), count, "a /sample count must be positive"); + } + + return SampleAsync(from, (int?)count, ct); + } + + private async IAsyncEnumerable SampleAsync( + long from, + int? count, + [EnumeratorCancellation] CancellationToken ct) { ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposed) != 0, this); - var uri = BuildSampleUri(from); + var uri = BuildSampleUri(from, count); // Linked to the caller's token AND to disposal, so tearing this client down mid-enumeration // surfaces as a cancellation rather than as a raw ObjectDisposedException from the socket. @@ -411,8 +436,10 @@ public sealed class MTConnectAgentClient : IMTConnectAgentClient, IDisposable } } - private Uri BuildSampleUri(long from) => - new(string.Create(CultureInfo.InvariantCulture, $"{_sampleUriBase}?from={from}{_sampleQuerySuffix}"), + private Uri BuildSampleUri(long from, int? count) => + new(string.Create( + CultureInfo.InvariantCulture, + $"{_sampleUriBase}?from={from}{_sampleQuerySuffix}&count={count ?? _sampleCount}&heartbeat={_heartbeatMs}"), UriKind.Absolute); private async Task SendAsync(Uri uri, CancellationToken ct) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.SampleLag.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.SampleLag.cs new file mode 100644 index 00000000..611eff06 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.SampleLag.cs @@ -0,0 +1,156 @@ +using System.Globalization; +using Microsoft.Extensions.Logging; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect; + +/// +/// Review 2026-10-21 R06:T-5 — /sample throughput is capped at +/// observations per chunk, one chunk per +/// . A machine producing more than that falls steadily behind +/// the Agent's ring buffer: every buffer-length the driver re-baselines (losing every intermediate transition) and +/// every delivered value is backlog-stale. This half measures the lag, surfaces it, and raises the per-stream +/// count so the pump can catch up. +/// +/// +/// +/// Measured per chunk. lag = lastSequence + 1 − nextSequence — the observations the Agent holds +/// that this stream has not been sent yet (unknown when the Agent omits lastSequence). A chunk is +/// saturated when it carries the full count: the Agent had more and stopped. A stream is behind +/// when consecutive chunks are saturated with a positive (or unknown) lag. +/// +/// +/// Adapted per pump. A behind stream is reopened at the cursor it reached with count doubled, up +/// to . A raised stream that then delivers nothing (an Agent that refuses the +/// larger count) steps back to the last count that delivered and stops raising for that pump. At the ceiling, +/// a stream still behind logs one Warning per pump. A new pump (a re-initialize, a resubscribe) starts again +/// from the configured — the count is an upper bound, so a +/// raised one costs nothing while the Agent has little to send. +/// +/// +/// Surfaced on the bag the /hosts page shows: the effective +/// count, the last measured lag, saturated chunks and count raises. +/// +/// +public sealed partial class MTConnectDriver : IDriverDiagnostics +{ + /// Consecutive behind chunks before the count is raised — one saturated chunk is just a burst. + internal const int BehindChunksBeforeRaise = 3; + + /// The most the per-stream count is ever raised to, relative to the configured one. + internal const int SampleCountMaxGrowth = 16; + + /// Absolute ceiling on a raised count (a stock cppagent buffer holds 131072 observations). + internal const int SampleCountAbsoluteCeiling = 65_536; + + /// The count the current pump is streaming with. Diagnostics only. + private int _effectiveSampleCount; + + /// The last measured lag (observations), or -1 when unknown. Diagnostics only. + private long _sampleLag = -1; + + /// Chunks that carried the full count. Diagnostics only. + private long _saturatedChunks; + + /// Times a pump raised its count. Diagnostics only. + private long _sampleCountRaises; + + /// The ceiling a pump started with may raise its count to. + /// The configured . + /// The ceiling — never below . + internal static int SampleCountCeiling(int configured) => + Math.Max(configured, (int)Math.Min((long)configured * SampleCountMaxGrowth, SampleCountAbsoluteCeiling)); + + /// + public IReadOnlyDictionary GetDiagnostics() + { + var bag = new Dictionary(StringComparer.Ordinal); + var count = Volatile.Read(ref _effectiveSampleCount); + if (count > 0) + { + bag["sampleCount"] = count.ToString(CultureInfo.InvariantCulture); + } + + var lag = Interlocked.Read(ref _sampleLag); + if (lag >= 0) + { + bag["sampleLag"] = lag.ToString(CultureInfo.InvariantCulture); + } + + bag["sampleSaturatedChunks"] = Interlocked.Read(ref _saturatedChunks).ToString(CultureInfo.InvariantCulture); + bag["sampleCountRaises"] = Interlocked.Read(ref _sampleCountRaises).ToString(CultureInfo.InvariantCulture); + + return bag; + } + + /// + /// Measures one delivered chunk against the count its stream was opened with and records the result. + /// + /// The chunk just applied. + /// The count the stream was opened with. + /// Whether this chunk says the stream is behind the Agent. + private bool MeasureSampleLag(MTConnectStreamsResult chunk, int sampleCount) + { + long? lag = chunk.LastSequence is { } last ? Math.Max(0, last + 1 - chunk.NextSequence) : null; + Interlocked.Exchange(ref _sampleLag, lag ?? -1); + + var saturated = chunk.Observations.Count >= sampleCount; + if (saturated) + { + Interlocked.Increment(ref _saturatedChunks); + } + + return saturated && lag is not 0; + } + + /// One pump's adaptive-count state. Only that pump's task touches it. + private sealed class SampleCountState(int configured) + { + /// The count the next stream opens with. + public int Count { get; set; } = configured; + + /// The last count a stream actually delivered at. + public int LastDelivered { get; set; } = configured; + + /// The most may be raised to; lowered to the last good count on a refusal. + public int Ceiling { get; set; } = SampleCountCeiling(configured); + + /// Whether the at-the-ceiling Warning has been logged for this pump. + public bool CeilingWarned { get; set; } + } + + /// + /// The count a behind stream should reopen with, or when it is already at its ceiling + /// (logged once per pump). + /// + private int? NextSampleCount(SampleCountState state, MTConnectDriverOptions options, long cursor) + { + var sampleCount = state.Count; + var next = (int)Math.Min((long)sampleCount * 2, state.Ceiling); + if (next > sampleCount) + { + Interlocked.Increment(ref _sampleCountRaises); + _logger.LogInformation( + "MTConnect driver {DriverInstanceId} is falling behind {AgentUri}'s /sample stream (chunks full at count {Count}, lag {Lag} observation(s)); reopening at sequence {Cursor} with count {NextCount}.", + _driverInstanceId, options.AgentUri, sampleCount, DescribeLag(), cursor, next); + + return next; + } + + if (!state.CeilingWarned) + { + state.CeilingWarned = true; + _logger.LogWarning( + "MTConnect driver {DriverInstanceId} is still behind {AgentUri}'s /sample stream at its count ceiling {Count} (lag {Lag} observation(s)): the machine produces more than count / SampleIntervalMs observations a second, so values are backlog-stale and the stream will re-baseline (dropping intermediate transitions) whenever it falls out of the Agent's buffer. Lower SampleIntervalMs or raise SampleCount.", + _driverInstanceId, options.AgentUri, sampleCount, DescribeLag()); + } + + return null; + } + + private string DescribeLag() + { + var lag = Interlocked.Read(ref _sampleLag); + return lag >= 0 ? lag.ToString(CultureInfo.InvariantCulture) : "unknown"; + } +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs index 717cf2c0..4949de28 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs @@ -1971,6 +1971,8 @@ public sealed partial class MTConnectDriver var cursor = from; var attempt = 0; _reconnectAttempts = 0; + var counts = new SampleCountState(options.SampleCount); + Volatile.Write(ref _effectiveSampleCount, counts.Count); try { @@ -2001,7 +2003,7 @@ public sealed partial class MTConnectDriver } } - var outcome = await ConsumeStreamAsync(client, options, cursor, instanceId, generation, ct).ConfigureAwait(false); + var outcome = await ConsumeStreamAsync(client, options, cursor, instanceId, generation, counts, ct).ConfigureAwait(false); cursor = outcome.Cursor; instanceId = outcome.InstanceId; @@ -2012,6 +2014,30 @@ public sealed partial class MTConnectDriver return; } + // R06:T-5 — the adaptive count. A stream that delivered vindicates the count it was opened with; a + // RAISED count whose stream delivered nothing may be one the Agent refuses (count above its buffer), + // so step back to the last count that worked and stop raising for this pump. + if (outcome.DeliveredAny) + { + counts.LastDelivered = counts.Count; + } + else if (counts.Count > counts.LastDelivered && outcome.Verdict is not StreamVerdict.Cancelled) + { + _logger.LogWarning( + outcome.Error, + "MTConnect driver {DriverInstanceId}: the /sample stream opened with the raised count {Count} delivered nothing; stepping back to {Previous} and not raising it again for this stream.", + _driverInstanceId, counts.Count, counts.LastDelivered); + counts.Count = counts.LastDelivered; + counts.Ceiling = counts.Count; + Volatile.Write(ref _effectiveSampleCount, counts.Count); + } + + if (outcome.RaiseCountTo is { } raised) + { + counts.Count = raised; + Volatile.Write(ref _effectiveSampleCount, raised); + } + switch (outcome.Verdict) { case StreamVerdict.Cancelled: @@ -2118,6 +2144,7 @@ public sealed partial class MTConnectDriver long cursor, long instanceId, long generation, + SampleCountState counts, CancellationToken ct) { // The RUNNING cursor the gap check is made against: the previous chunk's nextSequence, equal @@ -2131,9 +2158,12 @@ public sealed partial class MTConnectDriver // that cannot work at all. They must not share a backoff ladder — see RunSampleStreamAsync. var delivered = false; + // Consecutive chunks that said this stream is behind the Agent (R06:T-5). + var behindChunks = 0; + try { - await foreach (var chunk in client.SampleAsync(cursor, ct).ConfigureAwait(false)) + await foreach (var chunk in client.SampleAsync(cursor, counts.Count, ct).ConfigureAwait(false)) { if (chunk is null) { @@ -2198,6 +2228,21 @@ public sealed partial class MTConnectDriver _streamDegraded = false; _pumpLive = true; PublishHealthy(DateTime.UtcNow); + + // R06:T-5 — a stream persistently behind the Agent reopens at the cursor it reached with a larger + // count, rather than falling out of the ring buffer and re-baselining every buffer-length. + if (!MeasureSampleLag(chunk, counts.Count)) + { + behindChunks = 0; + } + else if (++behindChunks >= BehindChunksBeforeRaise) + { + behindChunks = 0; + if (NextSampleCount(counts, options, expected) is { } next) + { + return new StreamOutcome(StreamVerdict.Reopen, expected, instanceId, delivered, null, next); + } + } } // The seam contracts that cancellation is the ONLY way this enumeration ends quietly, so @@ -3894,8 +3939,12 @@ public sealed partial class MTConnectDriver /// the reconnect backoff from ratcheting monotonically over a process lifetime. /// /// The failure, when there was one. + /// + /// Set on a the adaptive count asked for: the stream was persistently behind + /// the Agent, and the next one opens with this count (R06:T-5). + /// private sealed record StreamOutcome( - StreamVerdict Verdict, long Cursor, long InstanceId, bool DeliveredAny, Exception? Error); + StreamVerdict Verdict, long Cursor, long InstanceId, bool DeliveredAny, Exception? Error, int? RaiseCountTo = null); // ---- config DTOs ---- diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDtos.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDtos.cs index 2e012d22..d9992bf0 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDtos.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDtos.cs @@ -56,7 +56,7 @@ public sealed record MTConnectComponent( /// /// One MTConnect DataItem declaration from the Agent's probe response. This is pure metadata — /// it carries no observed value; values arrive separately via -/// / +/// / /// and are correlated back to a data item by . /// /// @@ -89,7 +89,7 @@ public sealed record MTConnectDataItem( /// /// Result of an Agent /current request, or of a single multipart chunk from a /// /sample stream. Produced by and -/// . +/// . /// /// /// The Agent's current instance id (its MTConnectStreams header instanceId). Changes @@ -98,7 +98,7 @@ public sealed record MTConnectDataItem( /// /// /// The header nextSequence — the sequence number to resume a subsequent -/// call from. +/// call from. /// /// /// The header firstSequence — the oldest sequence number still held in the Agent's @@ -110,11 +110,17 @@ public sealed record MTConnectDataItem( /// The observations carried in this snapshot/chunk, in Agent-supplied order. Never null; /// empty when the chunk carries no new observations (e.g. a heartbeat). /// +/// +/// The header lastSequence — the newest sequence number the Agent holds, or null when the +/// document omits it. With it measures how far a /sample consumer is behind +/// the Agent: LastSequence + 1 - NextSequence observations it has not been sent yet (R06:T-5). +/// public sealed record MTConnectStreamsResult( long InstanceId, long NextSequence, long FirstSequence, - IReadOnlyList Observations); + IReadOnlyList Observations, + long? LastSequence = null); /// /// One observed value for a single DataItem, as reported in a /current snapshot or a diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectStreamExceptions.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectStreamExceptions.cs index fe0b54a9..c1a02259 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectStreamExceptions.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectStreamExceptions.cs @@ -7,7 +7,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect; /// /// /// Why these are exceptions rather than a normal end of enumeration. -/// is contracted to yield indefinitely until +/// is contracted to yield indefinitely until /// its token is cancelled, so a pump written to that contract has no reason to handle normal /// completion — it would either drop the subscription silently or hot-loop reconnecting on /// an enumerator that ends immediately. Returning normally would make "the data path diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectStreamsParser.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectStreamsParser.cs index d2a0fc72..dfa943d7 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectStreamsParser.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectStreamsParser.cs @@ -330,7 +330,8 @@ internal static class MTConnectStreamsParser sequences.InstanceId, sequences.NextSequence, sequences.FirstSequence, - observations); + observations, + sequences.LastSequence); } /// @@ -342,7 +343,14 @@ internal static class MTConnectStreamsParser new( MTConnectXml.RequiredLongAttribute(reader, "instanceId", HeaderOwner, Subject), MTConnectXml.RequiredLongAttribute(reader, "nextSequence", HeaderOwner, Subject), - MTConnectXml.RequiredLongAttribute(reader, "firstSequence", HeaderOwner, Subject)); + MTConnectXml.RequiredLongAttribute(reader, "firstSequence", HeaderOwner, Subject), + // Optional on purpose: only the /sample lag measurement reads it, and a document without it must + // still parse (the pump then falls back to chunk saturation alone — R06:T-5). + long.TryParse( + MTConnectXml.OptionalAttribute(reader, "lastSequence"), + NumberStyles.Integer, + CultureInfo.InvariantCulture, + out var last) ? last : null); /// /// Consumes the element the reader is positioned on, collecting any observation it is and @@ -653,7 +661,7 @@ internal static class MTConnectStreamsParser } /// The three sequence numbers read off a <Header>. - private readonly record struct HeaderSequences(long InstanceId, long NextSequence, long FirstSequence); + private readonly record struct HeaderSequences(long InstanceId, long NextSequence, long FirstSequence, long? LastSequence); /// /// Accumulates one observation's text across however many nodes the parser hands it, without diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/CannedAgentClient.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/CannedAgentClient.cs index 3e752354..9d3cdbcb 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/CannedAgentClient.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/CannedAgentClient.cs @@ -20,7 +20,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests; /// /// /// It honours the seam's stream-end contract (see -/// ): cancelling the token is the only way the +/// ): cancelling the token is the only way the /// enumeration ends without throwing. Closing the scripted stream via /// raises , exactly as the /// production client does when an Agent drops the connection — so a pump written against the @@ -297,6 +297,20 @@ internal sealed class CannedAgentClient : IMTConnectAgentClient, IDisposable return CurrentFailure is null ? Current : throw CurrentFailure; } + /// The count the most recent /sample enumeration was opened with (null = the 2-arg overload). + public int? LastSampleCount { get; private set; } + + /// Every count a /sample enumeration was opened with, in order. + public ConcurrentQueue SampleCounts { get; } = new(); + + /// + public IAsyncEnumerable SampleAsync(long from, int count, CancellationToken ct) + { + LastSampleCount = count; + SampleCounts.Enqueue(count); + return SampleAsync(from, ct); + } + /// public async IAsyncEnumerable SampleAsync( long from, [EnumeratorCancellation] CancellationToken ct) diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectSampleLagTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectSampleLagTests.cs new file mode 100644 index 00000000..517a95de --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectSampleLagTests.cs @@ -0,0 +1,154 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests; + +/// +/// Review 2026-10-21 R06:T-5 — /sample throughput was hard-capped at SampleCount / SampleIntervalMs +/// with no lag detection: a machine producing more fell behind the Agent's ring buffer forever, re-baselining +/// every buffer-length with every value backlog-stale and nothing surfacing it. The pump now measures the lag +/// (lastSequence + 1 − nextSequence, plus chunk saturation), surfaces it on the diagnostics bag, and raises +/// the per-stream count when it is persistently behind. +/// +public sealed class MTConnectSampleLagTests +{ + private const long Instance = 1655000000L; + private const long Primed = 108L; + private static readonly TimeSpan Watchdog = TimeSpan.FromSeconds(30); + + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + /// A chunk of dev1_pos values from . + private static MTConnectStreamsResult Chunk(long first, int observations, long? last) + { + var obs = Enumerable.Range(0, observations) + .Select(i => new MTConnectObservation( + "dev1_pos", (first + i).ToString(System.Globalization.CultureInfo.InvariantCulture), + new DateTime(2026, 10, 22, 12, 0, 0, DateTimeKind.Utc).AddSeconds(first + i))) + .ToList(); + return new MTConnectStreamsResult(Instance, first + observations, first, obs, last); + } + + private static async Task<(MTConnectDriver Driver, CannedAgentClient Client)> SubscribedDriverAsync(int sampleCount) + { + var client = CannedAgentClient.FromFixtures(); + var driver = new MTConnectDriver( + new MTConnectDriverOptions + { + AgentUri = "http://fixture-agent:5000", + RequestTimeoutMs = 5000, + SampleCount = sampleCount, + Tags = [new MTConnectTagDefinition("dev1_pos", DriverDataType.Float64)], + }, + "mt-t5", + _ => client); + await driver.InitializeAsync("{}", Ct); + await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct); + await client.WaitForSampleCallsAsync(1).WaitAsync(Watchdog, Ct); + return (driver, client); + } + + /// Three full chunks with a positive lag reopen the stream, at the cursor reached, with count doubled. + [Fact] + public async Task A_persistently_behind_stream_reopens_with_a_doubled_count() + { + var (driver, client) = await SubscribedDriverAsync(sampleCount: 2); + try + { + await client.PumpAsync(Chunk(Primed, 2, last: 900)).WaitAsync(Watchdog, Ct); + await client.PumpAsync(Chunk(Primed + 2, 2, last: 900)).WaitAsync(Watchdog, Ct); + await client.PumpAsync(Chunk(Primed + 4, 2, last: 900)).WaitAsync(Watchdog, Ct); + + await client.WaitForSampleCallsAsync(2).WaitAsync(Watchdog, Ct); + client.SampleCounts.ShouldBe([2, 4]); + client.LastSampleFrom.ShouldBe(Primed + 6, "the raised stream resumes where the behind one got to"); + + var diag = driver.GetDiagnostics(); + diag["sampleCount"].ShouldBe("4"); + diag["sampleLag"].ShouldBe((900 + 1 - (Primed + 6)).ToString(System.Globalization.CultureInfo.InvariantCulture)); + diag["sampleSaturatedChunks"].ShouldBe("3"); + diag["sampleCountRaises"].ShouldBe("1"); + } + finally + { + await driver.ShutdownAsync(CancellationToken.None); + } + } + + /// An Agent that omits lastSequence: persistent saturation alone is enough. + [Fact] + public async Task Persistent_saturation_without_a_lastSequence_still_raises_the_count() + { + var (driver, client) = await SubscribedDriverAsync(sampleCount: 2); + try + { + for (var i = 0; i < 3; i++) + await client.PumpAsync(Chunk(Primed + (2 * i), 2, last: null)).WaitAsync(Watchdog, Ct); + + await client.WaitForSampleCallsAsync(2).WaitAsync(Watchdog, Ct); + client.SampleCounts.ShouldBe([2, 4]); + driver.GetDiagnostics().ContainsKey("sampleLag").ShouldBeFalse("an unknown lag is not reported as a number"); + } + finally + { + await driver.ShutdownAsync(CancellationToken.None); + } + } + + /// + /// Full chunks with NO lag (the Agent happened to have exactly count observations) are not "behind": the stream + /// is kept. Positive control: the first test, through the same pump, does reopen. + /// + [Fact] + public async Task Full_chunks_with_no_lag_keep_the_stream() + { + var (driver, client) = await SubscribedDriverAsync(sampleCount: 2); + try + { + for (var i = 0; i < 4; i++) + { + var first = Primed + (2 * i); + await client.PumpAsync(Chunk(first, 2, last: first + 1)).WaitAsync(Watchdog, Ct); + } + + client.SampleCallCount.ShouldBe(1, "a caught-up stream must not be reopened"); + driver.GetDiagnostics()["sampleLag"].ShouldBe("0"); + driver.GetDiagnostics()["sampleCountRaises"].ShouldBe("0"); + } + finally + { + await driver.ShutdownAsync(CancellationToken.None); + } + } + + /// A raised count the Agent refuses (its stream delivers nothing) steps back to the last count that worked. + [Fact] + public async Task A_refused_raised_count_steps_back_to_the_last_count_that_delivered() + { + var (driver, client) = await SubscribedDriverAsync(sampleCount: 2); + try + { + // The NEXT enumeration (the raised one) is refused the way a real Agent refuses a count above its buffer. + client.SampleFailures.Enqueue(new InvalidDataException("MTConnectError: INVALID_REQUEST count too large")); + for (var i = 0; i < 3; i++) + await client.PumpAsync(Chunk(Primed + (2 * i), 2, last: 900)).WaitAsync(Watchdog, Ct); + + await client.WaitForSampleCallsAsync(3).WaitAsync(Watchdog, Ct); + client.SampleCounts.Take(3).ShouldBe([2, 4, 2]); + driver.GetDiagnostics()["sampleCount"].ShouldBe("2"); + } + finally + { + await driver.ShutdownAsync(CancellationToken.None); + } + } + + /// The ceiling is 16× the configured count, capped at 65 536, and never below the configured count. + [Theory] + [InlineData(1000, 16000)] + [InlineData(10_000, 65_536)] + [InlineData(100_000, 100_000)] + public void The_count_ceiling_is_bounded(int configured, int expected) => + MTConnectDriver.SampleCountCeiling(configured).ShouldBe(expected); +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectStreamsParseTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectStreamsParseTests.cs index b13e3dee..60f41b44 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectStreamsParseTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectStreamsParseTests.cs @@ -593,6 +593,38 @@ public sealed class MTConnectStreamsParseTests "http://agent:5000/VMC-3Axis/sample?from=108&interval=250&count=64&heartbeat=7000"); } + /// R06:T-5 — the count overload sends the raised count in place of the configured one. + [Fact] + public async Task SampleAsync_with_a_count_sends_that_count_instead_of_the_configured_one() + { + var handler = StubHandler.RespondingWithMultipart("bnd", await File.ReadAllTextAsync(SampleFixture, Ct)); + using var client = NewClient(handler, new MTConnectDriverOptions + { + AgentUri = "http://agent:5000", + DeviceName = "VMC-3Axis", + SampleIntervalMs = 250, + SampleCount = 64, + HeartbeatMs = 7000 + }); + + await DrainAsync(client.SampleAsync(108, 256, Ct)); + + handler.LastRequestUri!.AbsoluteUri.ShouldBe( + "http://agent:5000/VMC-3Axis/sample?from=108&interval=250&count=256&heartbeat=7000"); + } + + /// R06:T-5 — the header's lastSequence is carried (it measures the stream's lag); absent ⇒ null. + [Fact] + public async Task The_header_lastSequence_is_carried_when_present_and_null_when_absent() + { + var current = MTConnectStreamsParser.Parse(await File.ReadAllTextAsync("Fixtures/current.xml", Ct)); + current.LastSequence.ShouldBe(107); + + var stripped = (await File.ReadAllTextAsync("Fixtures/current.xml", Ct)).Replace("lastSequence=\"107\"", string.Empty); + stripped.ShouldNotContain("lastSequence"); + MTConnectStreamsParser.Parse(stripped).LastSequence.ShouldBeNull(); + } + [Fact] public async Task SampleAsync_yields_one_result_per_multipart_chunk() {