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

fix(mtconnect): close the silent-refusal + teardown-wedge defects in the pump (Task 11 review)

Browse Source 
C1 (critical): a latched "this endpoint cannot stream" verdict could end with a
GREEN driver holding a permanently silent subscription. DriverInstanceActor
handles a tag-set change as Unsubscribe-then-Subscribe with no re-initialize;
the unsubscribe cleared the stream-degraded flag, the resubscribe was refused by
the latch without a word, and one successful read then reported Healthy.
StartSampleStreamCore now re-asserts the degradation and logs a Warning naming
the remedy, and teardown no longer clears the flag while the latch is set.

NOT promoted to Faulted: DriverHealthReport maps any Faulted driver to a /readyz
503, so that would de-ready the whole node over one misconfigured Agent whose
reads are perfectly healthy. Faulted stays for the case that earns it.

I1: teardown waits are bounded (5 s) and honour the caller's token. They were
unbounded and ignored it, so DriverInstanceActor's 5 s budget was inert and
PostStop — which blocks a dispatcher thread on ShutdownAsync while holding the
lifecycle semaphore — could be wedged forever by one blocking subscriber. The
cancellation source is disposed only when the loop is provably gone. Applied to
StopProbeLoopAsync too: Task 13 reproduced the identical shape, and closing the
class in one method while leaving the other as the pattern to copy is worse than
not closing it.

I2: the reconnect ladder was monotonic for the process lifetime, so ~9 unrelated
drops pinned the driver at MaxBackoffMs forever. A stream that delivered before
dropping now starts a fresh ladder (at attempt 1, not 0, so MinBackoffMs is
still honoured).

I3: MaxBackoffMs <= 0 clamped every delay to zero — an operator-authorable
reconnect spin. Treated as unset, like a multiplier that cannot grow.

I4: the session published instanceId without NextSequence, so after a restart it
held the new agent's id beside the dead one's cursor, and a gap/OUT_OF_RANGE
re-baseline (id unchanged) never published the cursor at all. Both move in one
CAS, and the pump publishes its cursor as it exits.

I5: OnDataChange was a multicast Invoke — the first throwing handler aborted the
rest of the list, starving every later subscriber. Now walked by hand with the
catch inside the loop. The test that claimed to cover this registered the
recorder BEFORE the thrower, so it passed regardless; swapped, it was red.
Faults are latched to one Warning per stream generation (Debug thereafter) so a
consistently-throwing consumer cannot flood the log.

Also: the pump restarts after an unexpected fault (IsCompleted, not null); a
device-scope that matches nothing is a Warning, not a Debug tally; a device or
component with neither name nor id is skipped rather than emitting a blank path
segment; DiscoverAsync's remark no longer claims DriverInstanceActor retries
(it does not for a Once driver, and injection is dormant in v3); and two flake
seeds in the fake are gone (Dispose re-read its own counter; _disposeCts was
disposed under a racing SampleAsync).

439/439. Every changed behaviour falsified by mutation.
This commit is contained in:
Joseph Doherty
2026-07-24 17:16:13 -04:00
parent a92584c45e
commit a3a0cfe9bb
6 changed files with 926 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests/MTConnectSubscribeTests.cs
+271 -6
View File
@@ -84,17 +84,21 @@ public sealed class MTConnectSubscribeTests
// ---- fixtures ----
private static (MTConnectDriver Driver, CannedAgentClient Client) NewDriver(
CannedAgentClient? client = null, MTConnectDriverOptions? options = null)
CannedAgentClient? client = null,
MTConnectDriverOptions? options = null,
RecordingDriverLogger? logger = null)
{
var agent = client ?? CannedAgentClient.FromFixtures();
return (new MTConnectDriver(options ?? Opts(), "mt1", _ => agent), agent);
return (new MTConnectDriver(options ?? Opts(), "mt1", _ => agent, logger), agent);
}
private static async Task<(MTConnectDriver Driver, CannedAgentClient Client)> InitializedDriverAsync(
CannedAgentClient? client = null, MTConnectDriverOptions? options = null)
CannedAgentClient? client = null,
MTConnectDriverOptions? options = null,
RecordingDriverLogger? logger = null)
{
var (driver, agent) = NewDriver(client, options);
var (driver, agent) = NewDriver(client, options, logger);
await driver.InitializeAsync("{}", Ct);
return (driver, agent);
@@ -464,11 +468,24 @@ public sealed class MTConnectSubscribeTests
driver.GetHealth().State.ShouldBe(DriverState.Degraded);
driver.GetHealth().LastError.ShouldNotBeNullOrWhiteSpace();
// A full unsubscribe/resubscribe cycle must not re-open the wound either.
// A full unsubscribe/resubscribe cycle must not re-open the wound either. This is exactly
// what DriverInstanceActor does for a tag-set change: UnsubscribeAsync then SubscribeAsync,
// with NO re-initialize in between.
await driver.UnsubscribeAsync(handle, Ct).WaitAsync(Watchdog, Ct);
await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
client.SampleCallCount.ShouldBe(1);
// ...and the driver must not read back GREEN afterwards. The unsubscribe stopped a stream
// that was never running, the resubscribe was silently refused by the latch, and one
// successful read was then enough to clear the last degradation flag — leaving a Healthy
// driver holding a subscription that can never deliver a value. That is #485 in the
// subscription plane: an established handle, an Established reply, and permanent silence.
var res = await driver.ReadAsync(["dev1_pos"], Ct);
res[0].StatusCode.ShouldBe(Good);
driver.GetHealth().State.ShouldBe(DriverState.Degraded);
driver.GetHealth().LastError.ShouldNotBeNullOrWhiteSpace();
}
/// <summary>
@@ -747,8 +764,12 @@ public sealed class MTConnectSubscribeTests
public async Task A_throwing_subscriber_does_not_kill_the_pump()
{
var (driver, client) = await InitializedDriverAsync();
var seen = Record(driver);
// Registration ORDER is the whole point: the thrower goes FIRST. A multicast invoke aborts
// the rest of the invocation list at the first exception, so with the recorder registered
// first it would run before the throw and its assertions would hold however badly the
// driver isolates subscribers — the test would pass without testing anything.
driver.OnDataChange += (_, _) => throw new InvalidOperationException("subscriber blew up");
var seen = Record(driver);
await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
seen.Clear();
@@ -819,6 +840,250 @@ public sealed class MTConnectSubscribeTests
client.SampleCallCount.ShouldBe(2);
}
/// <summary>
/// <b>C1, the other half.</b> Refusing to start the stream is correct, but it must not be
/// silent: the operator's only other evidence is a subscription that never delivers. The
/// refusal re-asserts the degradation AND says so, naming the remedy.
/// </summary>
[Fact]
public async Task A_refused_stream_start_re_asserts_degraded_and_says_so()
{
var logger = new RecordingDriverLogger();
var (driver, client) = await InitializedDriverAsync(logger: logger);
client.SampleFailure = new MTConnectStreamNotSupportedException("not multipart");
var handle = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
await driver.SampleStreamTask!.WaitAsync(Watchdog, Ct);
await driver.UnsubscribeAsync(handle, Ct).WaitAsync(Watchdog, Ct);
await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
driver.GetHealth().State.ShouldBe(DriverState.Degraded);
logger.WarningsSnapshot()
.ShouldContain(w => w.Contains("refused to start a /sample stream", StringComparison.Ordinal));
}
/// <summary>
/// <b>C1, the window the refusal path cannot cover.</b> Dropping the last subscription stops
/// a stream, and teardown clears the stream degradation with it — but NOT while the endpoint
/// is latched as unable to stream at all. That latch is a standing fact about the Agent,
/// discovered at runtime and true whether or not anyone is subscribed right now; letting a
/// successful read clear it here would make the driver flicker green between subscriptions
/// and report the impairment only while someone happens to be listening.
/// </summary>
[Fact]
public async Task A_latched_endpoint_stays_degraded_even_with_no_subscriptions_left()
{
var (driver, client) = await InitializedDriverAsync();
client.SampleFailure = new MTConnectStreamNotSupportedException("not multipart");
var handle = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
await driver.SampleStreamTask!.WaitAsync(Watchdog, Ct);
// No subscriptions left at all — so nothing re-asserts the degradation on a refused start.
await driver.UnsubscribeAsync(handle, Ct).WaitAsync(Watchdog, Ct);
var res = await driver.ReadAsync(["dev1_pos"], Ct);
res[0].StatusCode.ShouldBe(Good);
driver.GetHealth().State.ShouldBe(DriverState.Degraded);
}
/// <summary>
/// <b>I1 — the caller's teardown deadline must be real.</b> <c>DriverInstanceActor</c> wraps
/// this call in a 5 s <see cref="CancellationTokenSource"/> and blocks an Akka dispatcher
/// thread on the shutdown path, so a wait that ignored the token would let one blocking
/// <c>OnDataChange</c> handler wedge an actor-system thread — while holding the lifecycle
/// semaphore, taking every later lifecycle call on this driver with it.
/// </summary>
[Fact]
public async Task Unsubscribe_honours_the_callers_deadline_when_a_subscriber_blocks()
{
var (driver, client) = await InitializedDriverAsync();
var blocking = new ManualResetEventSlim(false);
var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
// Blocks only on the PUMPED value — the initial-data callback fires on the subscribe
// thread, and blocking there would prove nothing about teardown.
driver.OnDataChange += (_, e) =>
{
if (e.Snapshot.Value is not 1.5d)
{
return;
}
entered.TrySetResult();
blocking.Wait();
};
var handle = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
try
{
_ = client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5")));
await entered.Task.WaitAsync(Watchdog, Ct);
// The pump is now stuck inside caller code. The unsubscribe must still return.
using var deadline = new CancellationTokenSource(TimeSpan.FromMilliseconds(100));
await driver.UnsubscribeAsync(handle, deadline.Token).WaitAsync(Watchdog, Ct);
driver.SampleStreamTask.ShouldBeNull();
}
finally
{
blocking.Set();
}
}
/// <summary>
/// <b>I2 — the backoff ladder is per-outage, not per-process.</b> A stream that delivered
/// before it dropped proves the endpoint works, so its failure starts a fresh ladder. Without
/// this the counter only ever climbs: an agent that drops one connection an hour would be
/// pinned at <c>MaxBackoffMs</c> within a day, having never once failed twice in a row.
/// </summary>
[Fact]
public async Task Reconnect_attempts_reset_when_the_stream_delivered_before_dropping()
{
var (driver, client) = await InitializedDriverAsync();
await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
await client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))).WaitAsync(Watchdog, Ct);
client.EndStream();
await client.PumpAsync(Chunk(113, 118, ("dev1_pos", "2.5"))).WaitAsync(Watchdog, Ct);
client.EndStream();
await client.PumpAsync(Chunk(118, 123, ("dev1_pos", "3.5"))).WaitAsync(Watchdog, Ct);
// Two unrelated drops, each preceded by a delivered chunk: each starts a fresh ladder, so
// the count is 1 (this outage's first retry, which still honours MinBackoffMs) — never 2.
driver.ReconnectAttempts.ShouldBe(1);
client.SampleCallCount.ShouldBe(3);
}
/// <summary>
/// …and the counter still accumulates when nothing is delivered, which is the case the
/// backoff exists for. Asserted as monotonic growth rather than an exact value, because the
/// pump may have advanced again between the barrier and the read.
/// </summary>
[Fact]
public async Task Reconnect_attempts_accumulate_while_the_stream_delivers_nothing()
{
// A 1 ms cap keeps the failing loop from actually sleeping; nothing here asserts a duration.
var options = Opts(new MTConnectReconnectOptions { MinBackoffMs = 0, MaxBackoffMs = 1 });
var (driver, client) = await InitializedDriverAsync(options: options);
client.SampleFailure = new TimeoutException("no chunk within the heartbeat window");
var handle = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
try
{
await client.WaitForSampleCallsAsync(3).WaitAsync(Watchdog, Ct);
var early = driver.ReconnectAttempts;
await client.WaitForSampleCallsAsync(6).WaitAsync(Watchdog, Ct);
early.ShouldBeGreaterThanOrEqualTo(2);
driver.ReconnectAttempts.ShouldBeGreaterThan(early);
}
finally
{
await driver.UnsubscribeAsync(handle, Ct).WaitAsync(Watchdog, Ct);
}
}
/// <summary>
/// <b>I3 — a non-positive cap is an operator-authorable hot loop.</b> Clamping to it (the
/// obvious <c>Math.Max(0, …)</c>) makes EVERY delay zero, so the pump reconnect-spins as fast
/// as a refused connection returns, one Warning per iteration. Treated as "unset" instead,
/// exactly like a multiplier that cannot grow.
/// </summary>
[Theory]
[InlineData(0)]
[InlineData(-1)]
public void Reconnect_backoff_treats_a_non_positive_cap_as_unset(int maxBackoffMs)
{
var options = new MTConnectReconnectOptions
{
MinBackoffMs = 0, MaxBackoffMs = maxBackoffMs, BackoffMultiplier = 2.0,
};
// The first retry is still immediate — that is MinBackoffMs, and it is honoured.
MTConnectDriver.BackoffFor(1, options).ShouldBe(TimeSpan.Zero);
// Every later one must actually back off.
MTConnectDriver.BackoffFor(2, options).ShouldBeGreaterThan(TimeSpan.Zero);
MTConnectDriver.BackoffFor(9, options).ShouldBeGreaterThan(MTConnectDriver.BackoffFor(2, options));
}
/// <summary>
/// <b>I4 — the session's instanceId and cursor are one fact and must move together.</b>
/// Publishing the id alone left the session holding the NEW agent's id beside the OLD
/// agent's cursor, which is precisely the tear the record exists to prevent — and on a gap
/// or <c>OUT_OF_RANGE</c> re-baseline (where the id does not change at all) the cursor was
/// not published even once.
/// </summary>
[Fact]
public async Task Rebaseline_publishes_the_cursor_beside_the_instance_id()
{
var client = CannedAgentClient.WithGapThenResume();
var (driver, _) = await InitializedDriverAsync(client);
await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
driver.AgentNextSequence.ShouldBe(PrimedNextSequence);
await client.PumpOnce().WaitAsync(Watchdog, Ct); // the gap chunk
driver.AgentNextSequence.ShouldBe(5005L);
driver.AgentInstanceId.ShouldBe(FixtureInstanceId);
}
/// <summary>
/// …and the consequence that makes it matter: the NEXT pump opens at the re-baselined
/// cursor. A last-unsubscribe followed by a resubscribe is an ordinary tag-set change
/// (<c>DriverInstanceActor</c> does exactly that, with no re-initialize), and a stale cursor
/// there means reopening at a sequence the Agent has already evicted.
/// </summary>
[Fact]
public async Task A_restarted_pump_resumes_from_the_rebaselined_cursor()
{
var client = CannedAgentClient.WithGapThenResume();
var (driver, _) = await InitializedDriverAsync(client);
var handle = await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
await client.PumpOnce().WaitAsync(Watchdog, Ct); // gap -> re-baseline to 5005
await driver.UnsubscribeAsync(handle, Ct).WaitAsync(Watchdog, Ct);
await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
await client.PumpOnce().WaitAsync(Watchdog, Ct); // the contiguous follow-up
client.LastSampleFrom.ShouldBe(5005L);
}
/// <summary>
/// <b>I5 — one broken subscriber must not starve the others.</b> A multicast
/// <c>Invoke</c> aborts the invocation list at the first exception, so a single try/catch
/// around it absorbs the exception while silently robbing every later subscriber of the
/// value — which reads as "isolated" in the log and is not. The thrower sits BETWEEN the two
/// recorders on purpose.
/// </summary>
[Fact]
public async Task Every_subscriber_receives_a_value_even_when_one_in_the_middle_throws()
{
var (driver, client) = await InitializedDriverAsync();
var first = Record(driver);
driver.OnDataChange += (_, _) => throw new InvalidOperationException("subscriber blew up");
var last = Record(driver);
await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
first.Clear();
last.Clear();
await client.PumpAsync(Chunk(PrimedNextSequence, 113, ("dev1_pos", "1.5"))).WaitAsync(Watchdog, Ct);
For(first, "dev1_pos").Select(e => e.Snapshot.Value).ShouldBe([1.5d]);
For(last, "dev1_pos").Select(e => e.Snapshot.Value).ShouldBe([1.5d]);
}
/// <summary>A handle from some other driver — the type check must not be a cast.</summary>
private sealed class ForeignHandle : ISubscriptionHandle
{