dohertj2/lmxopcua
1
0
Fork 0
Code Issues 17 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(mtconnect): host-connectivity probe + instanceId rediscover (Task 13)

Browse Source
This commit is contained in:
Joseph Doherty
2026-07-24 16:56:31 -04:00
parent 13dd9fcd70
commit fa0e40796f
2 changed files with 1094 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectHostAndRediscoverTests.cs
+629
View File
@@ -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;
/// <summary>
/// Task 13 — <see cref="MTConnectDriver"/>'s two optional capabilities:
/// <see cref="IHostConnectivityProbe"/> (a periodic reachability tick that flips one host
/// Running ↔ Stopped) and <see cref="IRediscoverable"/> (the Agent <c>instanceId</c> watch that
/// is the ONLY thing making <see cref="DiscoveryRediscoverPolicy.Once"/> safe).
/// </summary>
/// <remarks>
/// <para>
/// <b>Nothing here waits on wall-clock time, and the probe is a timer.</b> The driver takes
/// its inter-tick delay through an injected scheduler seam, and this suite supplies
/// <see cref="ManualTicker"/>: 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
/// <see cref="CannedAgentClient"/> stays timer-free.
/// </para>
/// <para>
/// <b>The load-bearing tests are the anti-inertness ones.</b> Both capabilities have a
/// plausible-looking implementation that compiles, ships, and does nothing:
/// a connectivity probe routed through the driver's cached <c>/probe</c> 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.
/// </para>
/// </remarks>
public sealed class MTConnectHostAndRediscoverTests
{
private const string AgentUri = "http://fixture-agent:5000";
/// <summary>The <c>instanceId</c> every canned fixture shares.</summary>
private const long FixtureInstanceId = 1655000000L;
/// <summary>A DIFFERENT <c>instanceId</c> — the Agent came back as a new process.</summary>
private const long RestartedInstanceId = 1655999999L;
/// <summary>A third <c>instanceId</c> — the Agent restarted twice.</summary>
private const long SecondRestartInstanceId = 1656111111L;
/// <summary>The cursor <c>Fixtures/current.xml</c> primes the pump with.</summary>
private const long PrimedNextSequence = 108L;
/// <summary>
/// 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.
/// </summary>
private static readonly TimeSpan Watchdog = TimeSpan.FromSeconds(10);
private static readonly DateTime ObservedAt = new(2026, 7, 24, 12, 30, 0, DateTimeKind.Utc);
/// <summary>The interval the probe tests author, asserted to reach the scheduler verbatim.</summary>
private static readonly TimeSpan AuthoredInterval = TimeSpan.FromSeconds(7);
private static CancellationToken Ct => TestContext.Current.CancellationToken;
// ---- fixtures ----
/// <summary>
/// Options with the background probe <b>disabled</b> — the default for the rediscovery half
/// of this suite, so no probe loop can add calls behind a <c>ProbeCallCount</c> assertion.
/// </summary>
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);
}
/// <summary>
/// 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).
/// </summary>
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);
}
/// <summary>
/// Releases exactly one probe tick and returns once the probe it triggered has completed —
/// proven by the loop parking again, not by a delay.
/// </summary>
private static async Task TickAsync(ManualTicker ticker)
{
ticker.Tick();
await ticker.ParkedAsync().WaitAsync(Watchdog, Ct);
}
private static ConcurrentQueue<RediscoveryEventArgs> RecordRediscovery(MTConnectDriver driver)
{
var seen = new ConcurrentQueue<RediscoveryEventArgs>();
((IRediscoverable)driver).OnRediscoveryNeeded += (_, e) => seen.Enqueue(e);
return seen;
}
private static ConcurrentQueue<HostStatusChangedEventArgs> RecordHostStatus(MTConnectDriver driver)
{
var seen = new ConcurrentQueue<HostStatusChangedEventArgs>();
((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))]);
/// <summary>A device model that is recognisably NOT the canned fixture's.</summary>
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 ----
/// <summary>
/// The plan's TDD case. <see cref="MTConnectDriver.RediscoverPolicy"/> is
/// <see cref="DiscoveryRediscoverPolicy.Once"/>, 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.
/// </summary>
[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();
}
/// <summary>
/// Once per <b>change</b>, not once per chunk. The chunks that follow a restart all carry the
/// new <c>instanceId</c>; a driver that compared each chunk against the id it was
/// <i>initialized</i> with — or that simply re-raised on every chunk — would ask Core to
/// rebuild the whole address space on every heartbeat, forever.
/// </summary>
[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);
}
/// <summary>
/// An unchanged <c>instanceId</c> 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.
/// </summary>
[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();
}
/// <summary>
/// <b>The anti-inertness pin for <see cref="IRediscoverable"/>.</b> A restart invalidates the
/// cached <c>/probe</c> device model — it describes a process that no longer exists. Raising
/// the event while keeping that cache would have Core dutifully re-run
/// <see cref="MTConnectDriver.DiscoverAsync"/> and receive the <i>identical stale tree</i>
/// from <c>GetOrFetchProbeModelAsync</c>'s warm cache: a feature that fires, logs, and
/// changes nothing.
/// </summary>
[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"]);
}
/// <summary>
/// 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.
/// </summary>
[Fact]
public async Task A_throwing_rediscovery_handler_does_not_kill_the_pump()
{
var (driver, client) = await InitializedDriverAsync();
var seen = new ConcurrentQueue<DataChangeEventArgs>();
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 ----
/// <summary>
/// One host per Agent, named by the authored <c>AgentUri</c> — 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 <see cref="HostState.Unknown"/>: the probe loop is the sole writer of
/// this field, and it has not run yet.
/// </summary>
[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);
}
/// <summary>
/// <b>The anti-inertness pin for <see cref="IHostConnectivityProbe"/>.</b> Every tick must
/// reach the Agent. A probe routed through the driver's cached-model accessor
/// (<c>GetOrFetchProbeModelAsync</c>) 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.
/// </summary>
[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);
}
/// <summary>
/// Running ↔ Stopped, and the event fires on the <b>transition</b> 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.
/// </summary>
[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);
}
/// <summary>
/// <b>The probe is a liveness check, never a cache writer.</b> Publishing its answer into
/// <c>AgentSession.ProbeModel</c> would put a second writer on the field the lifecycle owns
/// under a compare-and-swap, and would silently re-warm a cache that
/// <c>FlushOptionalCachesAsync</c> (or an Agent restart) deliberately dropped.
/// </summary>
[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();
}
/// <summary>
/// <b>Host connectivity is deliberately NOT an input to <c>DriverHealth</c>.</b> The two real
/// data paths (<c>/current</c> reads, the <c>/sample</c> pump) own health and will report the
/// same outage with the failure that actually mattered. The probe's deadline
/// (<c>Probe.Timeout</c>, 2 s) is far tighter than <c>RequestTimeoutMs</c> and its request is
/// the <i>largest</i> document the Agent serves, so letting it degrade the driver would flap
/// a perfectly working instance on a slow-but-alive Agent.
/// </summary>
[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);
}
/// <summary>
/// <c>Probe.Enabled = false</c> means no loop, no scheduler wait, and not one extra request.
/// An operator who turned probing off must not still be paying for it.
/// </summary>
[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);
}
/// <summary>
/// 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.
/// </summary>
[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);
}
/// <summary>
/// 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.
/// </summary>
[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) ----
/// <summary>
/// arch-review 01/S-6, applied to the two knobs Task 13 is the first consumer of. A
/// <c>0</c> interval turns the "periodic" probe into an unthrottled hot loop against the
/// Agent; a <c>0</c> 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 <c>RequirePositive</c> posture as the four timing knobs Task 9 guarded.
/// </summary>
[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<ArgumentOutOfRangeException>(() => driver.InitializeAsync("{}", Ct));
driver.GetHealth().State.ShouldBe(DriverState.Faulted);
}
/// <summary>
/// …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
/// <c>Enabled</c> back on is a config edit that re-runs the same validation.
/// </summary>
[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();
}
/// <summary>
/// A hand-driven replacement for <see cref="Task.Delay(TimeSpan, CancellationToken)"/>: 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.
/// </summary>
/// <remarks>
/// <see cref="Tick"/> installs the <i>next</i> park signal <b>before</b> releasing the
/// current wait, so a loop that races ahead and parks again immediately cannot lose the
/// signal a test is about to await.
/// </remarks>
private sealed class ManualTicker
{
private readonly Lock _gate = new();
private TaskCompletionSource _parked = new(TaskCreationOptions.RunContinuationsAsynchronously);
private TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously);
private int _waits;
/// <summary>How many times the loop has parked in this scheduler.</summary>
public int Waits => Volatile.Read(ref _waits);
/// <summary>The interval the loop most recently asked to wait for.</summary>
public TimeSpan LastInterval { get; private set; }
/// <summary>The scheduler seam handed to the driver.</summary>
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);
}
/// <summary>Completes once the loop is parked in the current wait.</summary>
public Task ParkedAsync()
{
lock (_gate)
{
return _parked.Task;
}
}
/// <summary>Releases the current wait so exactly one probe runs.</summary>
public void Tick()
{
lock (_gate)
{
var release = _release;
_release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
_parked = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
release.TrySetResult();
}
}
}
}