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, AgentDisplay(options), 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, AgentDisplay(options), sampleCount, DescribeLag());
}
return null;
}
private string DescribeLag()
{
var lag = Interlocked.Read(ref _sampleLag);
return lag >= 0 ? lag.ToString(CultureInfo.InvariantCulture) : "unknown";
}
}