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 f3839fd1..7a129af8 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect/MTConnectDriver.cs @@ -42,21 +42,24 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect; /// nextSequence. /// /// -/// Scope. This type currently implements , -/// , and . -/// IHostConnectivityProbe/IRediscoverable (Task 13) are added on top of the -/// state this class already caches — the probe model, the Agent instanceId, and the -/// observation index. There is deliberately no IWritable: MTConnect is a read-only -/// protocol, which is also why every discovered leaf is +/// Scope. This type implements , , +/// , , +/// and — the last two +/// built on state this class already caches (the Agent instanceId and the probe +/// model). There is deliberately no IWritable: MTConnect is a read-only protocol, +/// which is also why every discovered leaf is /// . /// /// /// The read path never writes the shared observation index — see /// . That index has exactly one writer: the /sample pump -/// (). +/// (). The same single-writer discipline covers the two +/// Task-13 fields: the connectivity probe loop is the sole writer of +/// , and it publishes nothing into +/// (see ). /// /// -public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDiscovery +public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDiscovery, IHostConnectivityProbe, IRediscoverable { /// /// Coarse per-entry byte weights behind . The contract asks @@ -128,6 +131,13 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis private readonly ILogger _logger; private readonly Func _agentClientFactory; + /// + /// How the connectivity probe loop waits between ticks. Injected so the cadence is a seam + /// rather than a wall clock: the loop is a timer, and a suite that drove it by sleeping would + /// be flaky by construction. Defaults to . + /// + private readonly Func _probeScheduler; + /// /// Serializes the three lifecycle entry points against each other. Initialize / Reinitialize /// / Shutdown all swap the client and the served caches; overlapping them would let a @@ -183,6 +193,48 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis /// Live subscriptions by id. Guarded by . private readonly Dictionary _subscriptions = []; + /// + /// Guards the two host-connectivity fields, which the probe loop writes and + /// reads from an arbitrary caller thread. Held for a field + /// read/write only — never across an await, and never while a status handler runs. + /// + private readonly Lock _probeLock = new(); + + /// + /// The Agent's last observed reachability. Written only by + /// , so the state and the event stream can never disagree: + /// every value this field has ever held was announced by exactly one + /// . That is why a successful + /// does not seed it — raising the event from there would run + /// caller code while the non-reentrant semaphore is held (the same + /// hazard that makes the pump, not , republish to subscribers), + /// and seeding it silently would leave a consumer that tracks the event stream disagreeing + /// with . until the first tick + /// is the honest answer. Guarded by . + /// + private HostState _hostState = HostState.Unknown; + + /// When last changed. Guarded by . + private DateTime _hostStateChangedUtc = DateTime.UtcNow; + + /// + /// Cancels the connectivity probe loop. Written only under , which + /// serializes every start and stop of the loop. + /// + private CancellationTokenSource? _probeCts; + + /// The connectivity probe loop's task, or null when none runs. See . + private Task? _probeTask; + + /// + /// The Agent instanceId the last was raised for — + /// seeded at the commit point of with the id the priming + /// /current reported. This is what makes rediscovery fire once per change + /// rather than once per chunk: every chunk after a restart carries the new id, and a + /// re-baseline that fails leaves the pump still holding the old one. + /// + private long _announcedInstanceId; + private MTConnectDriverOptions _options; private MTConnectObservationIndex _index; private DriverHealth _health = new(DriverState.Unknown, null, null); @@ -263,11 +315,17 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis /// , never from this constructor. /// /// Optional logger; defaults to the null logger. + /// + /// How the connectivity probe loop waits between ticks. Defaults to + /// ; tests inject a hand-driven ticker so + /// the probe's cadence is asserted deterministically instead of slept through. + /// public MTConnectDriver( MTConnectDriverOptions options, string driverInstanceId, Func? agentClientFactory = null, - ILogger? logger = null) + ILogger? logger = null, + Func? probeScheduler = null) { ArgumentNullException.ThrowIfNull(options); ArgumentException.ThrowIfNullOrWhiteSpace(driverInstanceId); @@ -276,6 +334,7 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis _driverInstanceId = driverInstanceId; _logger = logger ?? NullLogger.Instance; _agentClientFactory = agentClientFactory ?? (o => new MTConnectAgentClient(o, logger)); + _probeScheduler = probeScheduler ?? Task.Delay; // Never null, so every consumer (Task 10's ReadAsync included) can ask it for a snapshot // without a null check: an un-primed index answers BadWaitingForInitialData for an authored @@ -333,6 +392,14 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis } } + /// + /// The connectivity probe loop's task while one is running, else null. Exposed to + /// tests as the deterministic teardown barrier — "the probe loop has stopped" is a task + /// completion, never a sleep — and as the proof that Probe.Enabled = false starts + /// nothing at all. + /// + internal Task? ProbeLoopTask => Volatile.Read(ref _probeTask); + /// The options currently in force — the constructor's, or the last document applied. internal MTConnectDriverOptions EffectiveOptions => _options; @@ -362,6 +429,7 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis // enumerating a disposed client reads the disposal as a dropped connection and // reconnects against an object that no longer exists. await StopSampleStreamAsync().ConfigureAwait(false); + await StopProbeLoopAsync().ConfigureAwait(false); await TeardownCoreAsync().ConfigureAwait(false); await StartCoreAsync(options, existingClient: null, cancellationToken).ConfigureAwait(false); @@ -406,6 +474,7 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis var existing = Volatile.Read(ref _session)?.Client; await StopSampleStreamAsync().ConfigureAwait(false); + await StopProbeLoopAsync().ConfigureAwait(false); if (existing is null || RequiresClientRebuild(_options, incoming)) { @@ -438,6 +507,7 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis var lastRead = ReadHealth().LastSuccessfulRead; await StopSampleStreamAsync().ConfigureAwait(false); + await StopProbeLoopAsync().ConfigureAwait(false); await TeardownCoreAsync().ConfigureAwait(false); WriteHealth(new DriverHealth(DriverState.Unknown, lastRead, null)); @@ -969,6 +1039,351 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis private static bool IsCondition(string? category) => category.AsSpan().Trim().Equals("CONDITION", StringComparison.OrdinalIgnoreCase); + // ---- IHostConnectivityProbe: one host per Agent, on its own cheap tick ---- + + /// + /// + /// Raised from the probe loop's task, never under a lock and never under + /// . A handler is arbitrary caller code: one that blocks forever + /// blocks the loop with it (and therefore the teardown that awaits it), exactly as an + /// handler does to the pump. One that throws is absorbed. + /// + public event EventHandler? OnHostStatusChanged; + + /// + /// The single host this driver instance talks to, named by its configured + /// . Mirrors ModbusDriver.HostName's + /// host:port convention: several MTConnect drivers in one server disambiguate on the + /// Admin dashboard by endpoint rather than by driver-instance id. + /// + public string HostName => _options.AgentUri; + + /// + /// + /// One entry, always — an Agent is a single host, and the driver knows of it whether or not + /// it has been reached. Before the first probe tick (and for the whole life of an instance + /// with Probe.Enabled = false) that entry reads , which + /// is the truth: nothing has measured it. See for why a successful + /// initialize does not seed it . + /// + public IReadOnlyList GetHostStatuses() + { + lock (_probeLock) + { + return [new HostConnectivityStatus(HostName, _hostState, _hostStateChangedUtc)]; + } + } + + /// + /// Starts the periodic connectivity probe, if the operator asked for one. Caller must hold + /// . + /// + /// The client this session owns; the loop dials that one, never a later one. + /// The options this session started with. + private void StartProbeLoopCore(IMTConnectAgentClient client, MTConnectDriverOptions options) + { + if (!options.Probe.Enabled) + { + return; + } + + var cts = new CancellationTokenSource(); + _probeCts = cts; + Volatile.Write(ref _probeTask, Task.Run(() => RunProbeLoopAsync(client, options, cts.Token), CancellationToken.None)); + } + + /// + /// The connectivity probe: wait, ask the Agent whether it is answering, publish the verdict, + /// repeat. Never throws — it is a detached background task, and an escaping exception would + /// silently end host reporting while kept serving the last + /// thing it saw. + /// + /// + /// + /// It calls directly and publishes + /// nothing. Both halves of that are deliberate. + /// + /// + /// Routing it through is the obvious shortcut and + /// it is inert: that accessor answers from the cache + /// already warmed, so the loop would issue no request at all and report + /// forever no matter what the Agent was doing — a feature + /// that ticks, logs, and measures nothing. + /// + /// + /// Publishing the answer into is the opposite + /// mistake: it would put a second writer on a field the lifecycle owns under a + /// compare-and-swap, and would silently re-warm a cache that + /// — or an Agent restart, see + /// — deliberately dropped. The parsed model is + /// discarded; only the fact that a request completed is used. + /// + /// + /// Why /probe rather than the /sample heartbeat. The heartbeat is + /// free, but it only exists while something is subscribed — an instance with no + /// subscriptions would report indefinitely, and host + /// reachability would become a function of OPC UA client behaviour. A dedicated tick is + /// the one signal that is true whether or not anyone is watching. Its cost is a parse of + /// the device model per interval, which is why + /// exists. + /// + /// + /// It waits first, then probes. has just completed a + /// successful /probe one moment earlier; a tick that fired immediately would spend + /// a second identical request to learn nothing, and would do it on every re-initialize. + /// + /// + /// It never calls a lifecycle method — see the remarks. + /// awaits this task while holding that semaphore, so + /// reaching back into one would be a permanent hang with no exception and no log line. + /// + /// + private async Task RunProbeLoopAsync( + IMTConnectAgentClient client, MTConnectDriverOptions options, CancellationToken ct) + { + try + { + while (!ct.IsCancellationRequested) + { + try + { + await _probeScheduler(options.Probe.Interval, ct).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + return; + } + + if (ct.IsCancellationRequested) + { + return; + } + + bool reachable; + try + { + using var deadline = CancellationTokenSource.CreateLinkedTokenSource(ct); + deadline.CancelAfter(options.Probe.Timeout); + + // The answer is deliberately discarded — see the remarks. + _ = await client.ProbeAsync(deadline.Token).ConfigureAwait(false); + reachable = true; + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + return; + } + catch (Exception ex) + { + // Unreachable, unusable, disposed, or past its deadline — from the host's point + // of view these are one fact, and the data paths report the detail. + reachable = false; + + _logger.LogDebug( + ex, + "MTConnect driver {DriverInstanceId} could not reach {AgentUri} on its connectivity probe.", + _driverInstanceId, options.AgentUri); + } + + TransitionHostTo(reachable ? HostState.Running : HostState.Stopped); + } + } + catch (Exception ex) + { + _logger.LogError( + ex, + "MTConnect driver {DriverInstanceId} stopped its connectivity probe on an unexpected fault; host status for {AgentUri} will not be updated until the driver is re-initialized.", + _driverInstanceId, options.AgentUri); + } + } + + /// + /// Publishes a host state and announces it — only when it actually changed. Raising on + /// every tick would emit one event per interval per Agent forever: Core would re-fan Bad + /// quality across the host's subtree each time, and an operator watching the feed could not + /// tell a new outage from an ongoing one. + /// + private void TransitionHostTo(HostState newState) + { + HostState old; + lock (_probeLock) + { + old = _hostState; + if (old == newState) + { + return; + } + + _hostState = newState; + _hostStateChangedUtc = DateTime.UtcNow; + } + + _logger.LogInformation( + "MTConnect driver {DriverInstanceId} host {AgentUri} went {Previous} -> {Current}.", + _driverInstanceId, HostName, old, newState); + + // Outside the lock: a handler is caller code. + try + { + OnHostStatusChanged?.Invoke(this, new HostStatusChangedEventArgs(HostName, old, newState)); + } + catch (Exception ex) + { + _logger.LogWarning( + ex, + "MTConnect driver {DriverInstanceId} caught a fault from a host-status subscriber for {AgentUri}; the probe continues.", + _driverInstanceId, HostName); + } + } + + /// + /// Stops the connectivity probe loop and waits for it to be gone. Idempotent, and never + /// throws — it runs on teardown paths where a second exception would mask the first. + /// + /// + /// ⚠️ This runs while is held and it awaits the loop, on + /// exactly the same terms as : safe only because the loop + /// never touches the lifecycle and every await it makes observes the token cancelled below. + /// The one thing outside that guarantee is an handler that + /// blocks forever — a consumer bug this driver cannot paper over. + /// + private async Task StopProbeLoopAsync() + { + var cts = _probeCts; + var task = Volatile.Read(ref _probeTask); + _probeCts = null; + Volatile.Write(ref _probeTask, null); + + if (cts is null) + { + return; + } + + try + { + await cts.CancelAsync().ConfigureAwait(false); + } + catch (ObjectDisposedException) + { + // Raced another stop; the loop is already going away. + } + + if (task is not null) + { + try + { + await task.ConfigureAwait(false); + } + catch (Exception ex) + { + // RunProbeLoopAsync is written not to throw, so this is belt-and-braces. + _logger.LogDebug( + ex, + "MTConnect driver {DriverInstanceId} swallowed a fault from its connectivity probe while stopping it.", + _driverInstanceId); + } + } + + cts.Dispose(); + } + + // ---- IRediscoverable: the Agent instanceId watch ---- + + /// + /// + /// + /// This event is what makes = + /// safe. The runtime runs exactly one + /// discovery pass per connect, so an Agent that restarts mid-run and renumbers, adds, or + /// drops data items would otherwise leave a stale address space sitting behind a + /// perfectly Healthy driver. + /// + /// + /// Raised from the /sample pump's task, outside every lock. A handler that + /// throws is absorbed and logged — a consumer bug must not tear down the stream serving + /// every subscriber. A handler that blocks blocks the pump, and one that + /// synchronously awaits a lifecycle method on this driver deadlocks it: teardown waits + /// for the pump, and the pump is waiting inside the handler. Core's consumer schedules + /// the rediscovery rather than running it inline, which is the only safe shape. + /// + /// + public event EventHandler? OnRediscoveryNeeded; + + /// + /// Handles an Agent restart observed on the wire: drop the device model it invalidated, then + /// tell Core to re-discover — once per distinct instanceId. + /// + /// + /// + /// Dropping the cached /probe model is not housekeeping — it is what stops this + /// whole capability being inert. Core's response to the event is to re-run + /// , which reads + /// ; with the cache still warm that call answers + /// from a model describing a process that no longer exists, and the "rediscovery" would + /// diff the stale tree against itself and change nothing. Dropped through the same + /// compare-and-swap uses, so a lifecycle change + /// landing mid-drop wins rather than being undone. + /// + /// + /// Once per change, not once per chunk. Every chunk after a restart carries the new + /// id, and a re-baseline that fails leaves the pump still comparing against the old one — + /// so the guard is a latch on the id that was last announced, not on the pump's cursor + /// state. A second, different restart is a second announcement; an id the Agent reuses + /// after some other id is one too, because that is genuinely a new process. + /// + /// + /// The instanceId the Agent's chunk reported. + /// The options in force, for the log line's endpoint. + private void AnnounceAgentRestart(long instanceId, MTConnectDriverOptions options) + { + if (Interlocked.Exchange(ref _announcedInstanceId, instanceId) == instanceId) + { + return; + } + + InvalidateProbeModel(); + + _logger.LogInformation( + "MTConnect driver {DriverInstanceId} dropped its cached /probe device model for {AgentUri} and raised rediscovery: agent instanceId is now {InstanceId}.", + _driverInstanceId, options.AgentUri, instanceId); + + try + { + OnRediscoveryNeeded?.Invoke(this, new RediscoveryEventArgs("MTConnect agent instanceId changed", null)); + } + catch (Exception ex) + { + _logger.LogWarning( + ex, + "MTConnect driver {DriverInstanceId} caught a fault from a rediscovery subscriber; the /sample stream continues.", + _driverInstanceId); + } + } + + /// + /// Drops the cached /probe device model so the next consumer re-fetches it. Same + /// compare-and-swap discipline as — retried, because + /// unlike a flush this drop is a correctness requirement rather than a courtesy: losing the + /// race must not leave a model from a dead Agent installed. + /// + private void InvalidateProbeModel() + { + while (true) + { + var session = Volatile.Read(ref _session); + if (session?.ProbeModel is null) + { + return; + } + + var dropped = session with { ProbeModel = null, ProbeModelDataItemCount = 0 }; + if (ReferenceEquals(Interlocked.CompareExchange(ref _session, dropped, session), session)) + { + return; + } + } + } + /// /// Starts the shared /sample pump, if there is anything to pump and anything to pump /// from. Caller must hold . @@ -1166,6 +1581,11 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis ObservationIndex.Clear(); FanOut(SubscribedReferences(), ObservationIndex); + // The driver's own state is consistent by this point, so it is safe to hand the + // restart to arbitrary caller code — which may call straight back into + // DiscoverAsync. + AnnounceAgentRestart(chunk.InstanceId, options); + return await RebaselineAsync(client, options, expected, instanceId, ct).ConfigureAwait(false); } @@ -1632,6 +2052,19 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis RequirePositive(options.SampleIntervalMs, nameof(MTConnectDriverOptions.SampleIntervalMs)); RequirePositive(options.SampleCount, nameof(MTConnectDriverOptions.SampleCount)); + // Same 01/S-6 guard for Task 13's two knobs, gated on the feature that reads them. A 0 + // interval turns the "periodic" probe into an unthrottled hot loop against the Agent, and + // a 0 (or negative) timeout cancels every probe before it can be answered — pinning the + // host to Stopped forever and reporting an outage that does not exist. Checked only when + // probing is ON because a disabled probe reads neither value, and faulting a deployment + // over a field the driver never touches would buy nothing; flipping Enabled back on is a + // config edit, and ReinitializeAsync runs this same validation. + if (options.Probe.Enabled) + { + RequirePositive(options.Probe.Interval, $"{nameof(MTConnectDriverOptions.Probe)}.{nameof(MTConnectProbeOptions.Interval)}"); + RequirePositive(options.Probe.Timeout, $"{nameof(MTConnectDriverOptions.Probe)}.{nameof(MTConnectProbeOptions.Timeout)}"); + } + var client = existingClient; if (client is null) { @@ -1650,6 +2083,11 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis _options = options; built = null; + // The Agent this session talks to, as of now. A restart is measured against THIS, so a + // re-initialize against a different Agent process never re-announces the change the + // operator's own action caused. + Volatile.Write(ref _announcedInstanceId, current.InstanceId); + // One publish, so a concurrent lock-free reader sees the new client, the new probe // cache, and the pump's opening cursor together or none of them. Volatile.Write( @@ -1678,6 +2116,8 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis StartSampleStreamCore(republishOnStart: true); } + StartProbeLoopCore(client, options); + _logger.LogInformation( "MTConnect driver {DriverInstanceId} initialized against {AgentUri}{DeviceScope}: {DeviceCount} device(s), {ObservationCount} primed observation(s), agent instanceId {InstanceId}.", _driverInstanceId, @@ -2084,6 +2524,22 @@ public sealed class MTConnectDriver : IDriver, IReadable, ISubscribable, ITagDis } } + /// + /// The half of the 01/S-6 guard, for the connectivity probe's two + /// knobs. A non-positive interval is an unthrottled loop; a non-positive timeout cancels + /// every request before it can be answered. + /// + private static void RequirePositive(TimeSpan value, string optionName) + { + if (value <= TimeSpan.Zero) + { + throw new ArgumentOutOfRangeException( + nameof(MTConnectDriverOptions), + value, + $"{optionName} must be positive; a non-positive value would either spin the connectivity probe against the Agent or cancel every probe before it could be answered."); + } + } + /// Barrier-protected read of the cross-thread health snapshot. private DriverHealth ReadHealth() => Volatile.Read(ref _health); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectHostAndRediscoverTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectHostAndRediscoverTests.cs new file mode 100644 index 00000000..14be33e9 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectHostAndRediscoverTests.cs @@ -0,0 +1,629 @@ +using System.Collections.Concurrent; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests; + +/// +/// Task 13 — 's two optional capabilities: +/// (a periodic reachability tick that flips one host +/// Running ↔ Stopped) and (the Agent instanceId watch that +/// is the ONLY thing making safe). +/// +/// +/// +/// Nothing here waits on wall-clock time, and the probe is a timer. The driver takes +/// its inter-tick delay through an injected scheduler seam, and this suite supplies +/// : the loop parks in the seam, the test releases exactly one +/// tick, and the ticker's next park is the proof that the probe which followed has already +/// completed. No test sleeps, no test polls a counter, and +/// stays timer-free. +/// +/// +/// The load-bearing tests are the anti-inertness ones. Both capabilities have a +/// plausible-looking implementation that compiles, ships, and does nothing: +/// a connectivity probe routed through the driver's cached /probe accessor never +/// touches the network once the cache is warm (so it reports Running forever, whatever the +/// Agent is doing), and a rediscovery signal raised without dropping that same cache makes +/// Core re-run discovery and receive the identical stale device tree. Both are pinned here +/// by asserting the Agent was actually called. +/// +/// +public sealed class MTConnectHostAndRediscoverTests +{ + private const string AgentUri = "http://fixture-agent:5000"; + + /// The instanceId every canned fixture shares. + private const long FixtureInstanceId = 1655000000L; + + /// A DIFFERENT instanceId — the Agent came back as a new process. + private const long RestartedInstanceId = 1655999999L; + + /// A third instanceId — the Agent restarted twice. + private const long SecondRestartInstanceId = 1656111111L; + + /// The cursor Fixtures/current.xml primes the pump with. + private const long PrimedNextSequence = 108L; + + /// + /// Failure guard for the awaits a defect could turn into a permanent hang (a probe loop that + /// never parks, a teardown that deadlocks on the lifecycle semaphore). It exists to make a + /// hang red, not to assert a duration — no assertion is ever made about elapsed time. + /// + private static readonly TimeSpan Watchdog = TimeSpan.FromSeconds(10); + + private static readonly DateTime ObservedAt = new(2026, 7, 24, 12, 30, 0, DateTimeKind.Utc); + + /// The interval the probe tests author, asserted to reach the scheduler verbatim. + private static readonly TimeSpan AuthoredInterval = TimeSpan.FromSeconds(7); + + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + // ---- fixtures ---- + + /// + /// Options with the background probe disabled — the default for the rediscovery half + /// of this suite, so no probe loop can add calls behind a ProbeCallCount assertion. + /// + private static MTConnectDriverOptions Opts(MTConnectProbeOptions? probe = null) => + new() + { + AgentUri = AgentUri, + RequestTimeoutMs = 5000, + Probe = probe ?? new MTConnectProbeOptions { Enabled = false }, + Tags = + [ + new MTConnectTagDefinition("dev1_pos", DriverDataType.Float64), + new MTConnectTagDefinition("dev1_execution", DriverDataType.String), + ], + }; + + private static MTConnectProbeOptions Probing(bool enabled = true) => + new() { Enabled = enabled, Interval = AuthoredInterval, Timeout = TimeSpan.FromSeconds(2) }; + + private static async Task<(MTConnectDriver Driver, CannedAgentClient Client)> InitializedDriverAsync( + CannedAgentClient? client = null, + MTConnectDriverOptions? options = null, + ManualTicker? ticker = null) + { + var agent = client ?? CannedAgentClient.FromFixtures(); + var driver = new MTConnectDriver( + options ?? Opts(), "mt1", _ => agent, probeScheduler: ticker is null ? null : ticker.WaitAsync); + + await driver.InitializeAsync("{}", Ct); + + return (driver, agent); + } + + /// + /// Brings a driver up with the probe loop wired to a manual ticker, and returns once the loop + /// is demonstrably parked in its first inter-tick wait (i.e. it has run zero probes). + /// + private static async Task<(MTConnectDriver Driver, CannedAgentClient Client, ManualTicker Ticker)> + ProbingDriverAsync(CannedAgentClient? client = null) + { + var ticker = new ManualTicker(); + var (driver, agent) = await InitializedDriverAsync(client, Opts(Probing()), ticker); + + await ticker.ParkedAsync().WaitAsync(Watchdog, Ct); + + return (driver, agent, ticker); + } + + /// + /// Releases exactly one probe tick and returns once the probe it triggered has completed — + /// proven by the loop parking again, not by a delay. + /// + private static async Task TickAsync(ManualTicker ticker) + { + ticker.Tick(); + await ticker.ParkedAsync().WaitAsync(Watchdog, Ct); + } + + private static ConcurrentQueue RecordRediscovery(MTConnectDriver driver) + { + var seen = new ConcurrentQueue(); + ((IRediscoverable)driver).OnRediscoveryNeeded += (_, e) => seen.Enqueue(e); + + return seen; + } + + private static ConcurrentQueue RecordHostStatus(MTConnectDriver driver) + { + var seen = new ConcurrentQueue(); + ((IHostConnectivityProbe)driver).OnHostStatusChanged += (_, e) => seen.Enqueue(e); + + return seen; + } + + private static MTConnectStreamsResult ChunkFrom( + long instanceId, long firstSequence, long nextSequence, params (string Id, string Value)[] observations) => + new( + instanceId, + nextSequence, + firstSequence, + [.. observations.Select(o => new MTConnectObservation(o.Id, o.Value, ObservedAt))]); + + /// A device model that is recognisably NOT the canned fixture's. + private static MTConnectProbeModel OtherModel() => + new([ + new MTConnectDevice( + "other-device", + "other-device", + [], + [new MTConnectDataItem("other_item", "other_item", "EVENT", "AVAILABILITY", null, null, null, null)]), + ]); + + // ---- IRediscoverable: the instanceId watch ---- + + /// + /// The plan's TDD case. is + /// , so the runtime runs exactly one discovery + /// pass per connect — an Agent that restarts and renumbers (or adds) data items would leave a + /// stale address space behind a Healthy driver. This event is the only thing that makes that + /// policy safe. + /// + [Fact] + public async Task InstanceId_change_raises_rediscovery() + { + var (driver, client) = await InitializedDriverAsync(); + var seen = RecordRediscovery(driver); + await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct); + + client.CurrentAnswers.Enqueue(ChunkFrom(RestartedInstanceId, 40, 42, ("dev1_pos", "9.5"))); + + await client + .PumpAsync(ChunkFrom(RestartedInstanceId, 5000, 5005, ("dev1_pos", "5.5"))) + .WaitAsync(Watchdog, Ct); + + var raised = seen.ShouldHaveSingleItem(); + raised.Reason.ShouldBe("MTConnect agent instanceId changed"); + raised.ScopeHint.ShouldBeNull(); + } + + /// + /// Once per change, not once per chunk. The chunks that follow a restart all carry the + /// new instanceId; a driver that compared each chunk against the id it was + /// initialized with — or that simply re-raised on every chunk — would ask Core to + /// rebuild the whole address space on every heartbeat, forever. + /// + [Fact] + public async Task InstanceId_change_raises_rediscovery_once_per_change_not_once_per_chunk() + { + var (driver, client) = await InitializedDriverAsync(); + var seen = RecordRediscovery(driver); + await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct); + + client.CurrentAnswers.Enqueue(ChunkFrom(RestartedInstanceId, 40, 42, ("dev1_pos", "9.5"))); + + await client + .PumpAsync(ChunkFrom(RestartedInstanceId, 5000, 5005, ("dev1_pos", "5.5"))) + .WaitAsync(Watchdog, Ct); + seen.Count.ShouldBe(1); + + // Two more chunks from the SAME restarted Agent, contiguous with the re-baseline. + await client.PumpAsync(ChunkFrom(RestartedInstanceId, 42, 45, ("dev1_pos", "10.5"))).WaitAsync(Watchdog, Ct); + await client.PumpAsync(ChunkFrom(RestartedInstanceId, 45, 48, ("dev1_pos", "11.5"))).WaitAsync(Watchdog, Ct); + + seen.Count.ShouldBe(1); + + // …but a SECOND restart is a second change, and must be announced. + client.CurrentAnswers.Enqueue(ChunkFrom(SecondRestartInstanceId, 10, 12, ("dev1_pos", "1.5"))); + await client + .PumpAsync(ChunkFrom(SecondRestartInstanceId, 6000, 6005, ("dev1_pos", "2.5"))) + .WaitAsync(Watchdog, Ct); + + seen.Count.ShouldBe(2); + } + + /// + /// An unchanged instanceId is the overwhelmingly common case — every ordinary chunk, + /// every heartbeat, and every ring-buffer re-baseline of a still-running Agent. None of them + /// may cost an address-space rebuild. + /// + [Fact] + public async Task Unchanged_instanceId_raises_no_rediscovery() + { + var (driver, client) = await InitializedDriverAsync(); + var seen = RecordRediscovery(driver); + await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct); + + await client + .PumpAsync(ChunkFrom(FixtureInstanceId, PrimedNextSequence, 113, ("dev1_pos", "1.5"))) + .WaitAsync(Watchdog, Ct); + + // A heartbeat… + await client.PumpAsync(ChunkFrom(FixtureInstanceId, 1, 120)).WaitAsync(Watchdog, Ct); + + // …and a ring-buffer gap, which re-baselines but is NOT a restart. + client.CurrentAnswers.Enqueue(ChunkFrom(FixtureInstanceId, 5000, 5005, ("dev1_pos", "2.5"))); + await client + .PumpAsync(ChunkFrom(FixtureInstanceId, 5000, 5005, ("dev1_pos", "2.5"))) + .WaitAsync(Watchdog, Ct); + + seen.ShouldBeEmpty(); + } + + /// + /// The anti-inertness pin for . A restart invalidates the + /// cached /probe device model — it describes a process that no longer exists. Raising + /// the event while keeping that cache would have Core dutifully re-run + /// and receive the identical stale tree + /// from GetOrFetchProbeModelAsync's warm cache: a feature that fires, logs, and + /// changes nothing. + /// + [Fact] + public async Task InstanceId_change_drops_the_cached_probe_model_so_rediscovery_sees_the_new_one() + { + var (driver, client) = await InitializedDriverAsync(); + await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct); + driver.CachedProbeModel.ShouldNotBeNull(); + + // The restarted Agent declares a different device model. + client.Probe = OtherModel(); + client.CurrentAnswers.Enqueue(ChunkFrom(RestartedInstanceId, 40, 42, ("dev1_pos", "9.5"))); + + await client + .PumpAsync(ChunkFrom(RestartedInstanceId, 5000, 5005, ("dev1_pos", "5.5"))) + .WaitAsync(Watchdog, Ct); + + driver.CachedProbeModel.ShouldBeNull(); + + var probeCalls = client.ProbeCallCount; + var builder = new CapturingBuilder(); + await driver.DiscoverAsync(builder, Ct); + + // Discovery went back to the Agent, and streamed the NEW model rather than the old one. + client.ProbeCallCount.ShouldBe(probeCalls + 1); + builder.Variables.Select(v => v.Attr.FullName).ShouldBe(["other_item"]); + } + + /// + /// A rediscovery handler is arbitrary caller code. One that throws is a bug in the consumer, + /// not a reason to stop streaming data to every subscriber on this driver. + /// + [Fact] + public async Task A_throwing_rediscovery_handler_does_not_kill_the_pump() + { + var (driver, client) = await InitializedDriverAsync(); + var seen = new ConcurrentQueue(); + driver.OnDataChange += (_, e) => seen.Enqueue(e); + ((IRediscoverable)driver).OnRediscoveryNeeded += (_, _) => + throw new InvalidOperationException("rediscovery subscriber blew up"); + + await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct); + client.CurrentAnswers.Enqueue(ChunkFrom(RestartedInstanceId, 40, 42, ("dev1_pos", "9.5"))); + + await client + .PumpAsync(ChunkFrom(RestartedInstanceId, 5000, 5005, ("dev1_pos", "5.5"))) + .WaitAsync(Watchdog, Ct); + + // The pump survived the throw, re-baselined, and keeps delivering. + await client.PumpAsync(ChunkFrom(RestartedInstanceId, 42, 45, ("dev1_pos", "12.5"))).WaitAsync(Watchdog, Ct); + + driver.SampleStreamTask!.IsCompleted.ShouldBeFalse(); + seen.Where(e => e.FullReference == "dev1_pos").Last().Snapshot.Value.ShouldBe(12.5d); + } + + // ---- IHostConnectivityProbe ---- + + /// + /// One host per Agent, named by the authored AgentUri — that is the identity the Admin + /// dashboard shows and the key Core scopes its Bad-quality fan-out by. Before the first tick + /// the honest answer is : the probe loop is the sole writer of + /// this field, and it has not run yet. + /// + [Fact] + public async Task Host_statuses_report_one_host_named_by_the_agent_uri() + { + var (driver, _, _) = await ProbingDriverAsync(); + + driver.HostName.ShouldBe(AgentUri); + + var status = ((IHostConnectivityProbe)driver).GetHostStatuses().ShouldHaveSingleItem(); + status.HostName.ShouldBe(AgentUri); + status.State.ShouldBe(HostState.Unknown); + } + + /// + /// The anti-inertness pin for . Every tick must + /// reach the Agent. A probe routed through the driver's cached-model accessor + /// (GetOrFetchProbeModelAsync) compiles, ships, and — with a cache warmed by + /// initialize — never issues a single request, reporting Running forever regardless of what + /// the Agent is doing. The call count is the only thing that can tell the two apart. + /// + [Fact] + public async Task Probe_loop_calls_the_agent_on_every_tick() + { + var (_, client, ticker) = await ProbingDriverAsync(); + + // Parked before the first tick: the loop waits, THEN probes, so initialize's own /probe is + // the only call so far. + client.ProbeCallCount.ShouldBe(1); + + await TickAsync(ticker); + client.ProbeCallCount.ShouldBe(2); + + await TickAsync(ticker); + client.ProbeCallCount.ShouldBe(3); + + await TickAsync(ticker); + client.ProbeCallCount.ShouldBe(4); + + // …and it waits the authored interval between them, rather than a hard-coded cadence. + ticker.LastInterval.ShouldBe(AuthoredInterval); + } + + /// + /// Running ↔ Stopped, and the event fires on the transition only. A driver that raised + /// on every tick would emit one event per interval per Agent forever; Core would re-fan Bad + /// quality across the host's subtree each time, and an operator watching the feed could not + /// tell a new outage from an ongoing one. + /// + [Fact] + public async Task Probe_flips_the_host_on_transition_only() + { + var (driver, client, ticker) = await ProbingDriverAsync(); + var seen = RecordHostStatus(driver); + + client.ProbeFailure = new HttpRequestException("agent unreachable"); + + await TickAsync(ticker); + await TickAsync(ticker); + + var down = seen.ShouldHaveSingleItem(); + down.HostName.ShouldBe(AgentUri); + down.OldState.ShouldBe(HostState.Unknown); + down.NewState.ShouldBe(HostState.Stopped); + ((IHostConnectivityProbe)driver).GetHostStatuses()[0].State.ShouldBe(HostState.Stopped); + + client.ProbeFailure = null; + + await TickAsync(ticker); + await TickAsync(ticker); + + seen.Count.ShouldBe(2); + var up = seen.Last(); + up.OldState.ShouldBe(HostState.Stopped); + up.NewState.ShouldBe(HostState.Running); + ((IHostConnectivityProbe)driver).GetHostStatuses()[0].State.ShouldBe(HostState.Running); + } + + /// + /// The probe is a liveness check, never a cache writer. Publishing its answer into + /// AgentSession.ProbeModel would put a second writer on the field the lifecycle owns + /// under a compare-and-swap, and would silently re-warm a cache that + /// FlushOptionalCachesAsync (or an Agent restart) deliberately dropped. + /// + [Fact] + public async Task Probe_ticks_never_publish_into_the_session_probe_cache() + { + var (driver, client, ticker) = await ProbingDriverAsync(); + var original = driver.CachedProbeModel.ShouldNotBeNull(); + + // The Agent now answers /probe with a different model. Nothing the connectivity probe does + // may let that model reach the driver's cache. + client.Probe = OtherModel(); + + await TickAsync(ticker); + await TickAsync(ticker); + + driver.CachedProbeModel.ShouldBeSameAs(original); + driver.GetMemoryFootprint().ShouldBeGreaterThan(0); + + // …and a flushed cache stays flushed until a real consumer asks for it. + await driver.FlushOptionalCachesAsync(Ct); + await TickAsync(ticker); + driver.CachedProbeModel.ShouldBeNull(); + } + + /// + /// Host connectivity is deliberately NOT an input to DriverHealth. The two real + /// data paths (/current reads, the /sample pump) own health and will report the + /// same outage with the failure that actually mattered. The probe's deadline + /// (Probe.Timeout, 2 s) is far tighter than RequestTimeoutMs and its request is + /// the largest document the Agent serves, so letting it degrade the driver would flap + /// a perfectly working instance on a slow-but-alive Agent. + /// + [Fact] + public async Task An_unreachable_host_does_not_degrade_driver_health() + { + var (driver, client, ticker) = await ProbingDriverAsync(); + client.ProbeFailure = new HttpRequestException("agent unreachable"); + + await TickAsync(ticker); + + ((IHostConnectivityProbe)driver).GetHostStatuses()[0].State.ShouldBe(HostState.Stopped); + driver.GetHealth().State.ShouldBe(DriverState.Healthy); + + // …and the converse: a reachable host does not clear a genuinely failing read path either. + client.ProbeFailure = null; + client.CurrentFailure = new HttpRequestException("connection refused"); + await driver.ReadAsync(["dev1_pos"], Ct); + driver.GetHealth().State.ShouldBe(DriverState.Degraded); + + await TickAsync(ticker); + + ((IHostConnectivityProbe)driver).GetHostStatuses()[0].State.ShouldBe(HostState.Running); + driver.GetHealth().State.ShouldBe(DriverState.Degraded); + } + + /// + /// Probe.Enabled = false means no loop, no scheduler wait, and not one extra request. + /// An operator who turned probing off must not still be paying for it. + /// + [Fact] + public async Task Probe_disabled_starts_no_loop_and_makes_no_calls() + { + var ticker = new ManualTicker(); + var (driver, client) = await InitializedDriverAsync(options: Opts(Probing(enabled: false)), ticker: ticker); + + driver.ProbeLoopTask.ShouldBeNull(); + ticker.Waits.ShouldBe(0); + client.ProbeCallCount.ShouldBe(1); // the initialize prime, and nothing else + + // The capability is still implemented — it just has nothing to report. + ((IHostConnectivityProbe)driver).GetHostStatuses() + .ShouldHaveSingleItem().State.ShouldBe(HostState.Unknown); + } + + /// + /// Teardown stops the timer and waits for it. A loop left running would keep dialling an + /// Agent through a client its owner has already disposed — the "torn down but still + /// reporting" shape the rest of this driver's lifecycle is written to prevent. + /// + [Fact] + public async Task Shutdown_stops_the_probe_loop_and_makes_no_further_calls() + { + var (driver, client, ticker) = await ProbingDriverAsync(); + await TickAsync(ticker); + var loop = driver.ProbeLoopTask.ShouldNotBeNull(); + + await driver.ShutdownAsync(Ct).WaitAsync(Watchdog, Ct); + + loop.IsCompleted.ShouldBeTrue(); + loop.IsFaulted.ShouldBeFalse(); + driver.ProbeLoopTask.ShouldBeNull(); + client.IsDisposed.ShouldBeTrue(); + + // The loop is provably gone (ShutdownAsync awaited it), so releasing another tick can reach + // nobody — no probe against a disposed client, and no further waits on the scheduler. + var calls = client.ProbeCallCount; + var waits = ticker.Waits; + ticker.Tick(); + + client.ProbeCallCount.ShouldBe(calls); + ticker.Waits.ShouldBe(waits); + } + + /// + /// A re-initialize replaces the client, so it must replace the loop that dials it — the old + /// one holds a reference to a client the re-initialize may be about to dispose. + /// + [Fact] + public async Task Reinitialize_replaces_the_probe_loop() + { + var (driver, _, ticker) = await ProbingDriverAsync(); + var first = driver.ProbeLoopTask.ShouldNotBeNull(); + + await driver.ReinitializeAsync("{}", Ct).WaitAsync(Watchdog, Ct); + + first.IsCompleted.ShouldBeTrue(); + var second = driver.ProbeLoopTask.ShouldNotBeNull(); + second.ShouldNotBeSameAs(first); + + // The replacement loop is live: it parks, ticks, and probes like the first. + await ticker.ParkedAsync().WaitAsync(Watchdog, Ct); + await TickAsync(ticker); + } + + // ---- operator-authorable zeroes (arch-review 01/S-6) ---- + + /// + /// arch-review 01/S-6, applied to the two knobs Task 13 is the first consumer of. A + /// 0 interval turns the "periodic" probe into an unthrottled hot loop against the + /// Agent; a 0 timeout cancels every probe before it can be answered, pinning the host + /// to Stopped forever and reporting an outage that does not exist. Both are refused with the + /// same RequirePositive posture as the four timing knobs Task 9 guarded. + /// + [Theory] + [InlineData(0, 2000)] + [InlineData(-1, 2000)] + [InlineData(5000, 0)] + [InlineData(5000, -1)] + public async Task A_non_positive_probe_interval_or_timeout_is_refused(int intervalMs, int timeoutMs) + { + var options = Opts(new MTConnectProbeOptions + { + Enabled = true, + Interval = TimeSpan.FromMilliseconds(intervalMs), + Timeout = TimeSpan.FromMilliseconds(timeoutMs), + }); + + var driver = new MTConnectDriver(options, "mt1", _ => CannedAgentClient.FromFixtures()); + + await Should.ThrowAsync(() => driver.InitializeAsync("{}", Ct)); + driver.GetHealth().State.ShouldBe(DriverState.Faulted); + } + + /// + /// …but only when probing is on. Faulting a deployment over a field the driver never reads + /// would be a new way for a deploy to fail with nothing gained; flipping + /// Enabled back on is a config edit that re-runs the same validation. + /// + [Fact] + public async Task A_non_positive_probe_interval_is_ignored_when_probing_is_disabled() + { + var options = Opts(new MTConnectProbeOptions + { + Enabled = false, Interval = TimeSpan.Zero, Timeout = TimeSpan.Zero, + }); + + var (driver, _) = await InitializedDriverAsync(options: options); + + driver.GetHealth().State.ShouldBe(DriverState.Healthy); + driver.ProbeLoopTask.ShouldBeNull(); + } + + /// + /// A hand-driven replacement for : the + /// probe loop parks here, and a test releases exactly one tick at a time. The park signal is + /// what makes the suite deterministic — when the loop is parked again, the probe that + /// followed the previous release has demonstrably finished. + /// + /// + /// installs the next park signal before releasing the + /// current wait, so a loop that races ahead and parks again immediately cannot lose the + /// signal a test is about to await. + /// + private sealed class ManualTicker + { + private readonly Lock _gate = new(); + private TaskCompletionSource _parked = new(TaskCreationOptions.RunContinuationsAsynchronously); + private TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously); + private int _waits; + + /// How many times the loop has parked in this scheduler. + public int Waits => Volatile.Read(ref _waits); + + /// The interval the loop most recently asked to wait for. + public TimeSpan LastInterval { get; private set; } + + /// The scheduler seam handed to the driver. + public async Task WaitAsync(TimeSpan interval, CancellationToken ct) + { + TaskCompletionSource release; + lock (_gate) + { + LastInterval = interval; + Interlocked.Increment(ref _waits); + release = _release; + _parked.TrySetResult(); + } + + await release.Task.WaitAsync(ct).ConfigureAwait(false); + } + + /// Completes once the loop is parked in the current wait. + public Task ParkedAsync() + { + lock (_gate) + { + return _parked.Task; + } + } + + /// Releases the current wait so exactly one probe runs. + public void Tick() + { + lock (_gate) + { + var release = _release; + _release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _parked = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + release.TrySetResult(); + } + } + } +}