From 233dea5d8faaecafbb8f104a3f9796fa1812b018 Mon Sep 17 00:00:00 2001 From: Tyler Kron Date: Wed, 29 Jul 2026 08:56:53 -0600 Subject: [PATCH 1/4] fix(transport): detect a physically dropped serial/TCP connection (closes #382) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A DAQiFi device kept reporting IsConnected == true after its USB cable was pulled, so ConnectionStatus.Lost — which DaqifiDevice is fully wired to raise — was unreachable on serial. SerialPort.IsOpen reflects the OS handle, which stays open after the device vanishes, and neither the consumer's read failures nor the producer's write failures were ever fed back to the transport. Two detectors now sit on top of the handle, sharing one TransportConnectionWatchdog so "declare the connection lost exactly once" lives in a single place: - Port-presence polling (serial), 1 s cadence, 2 consecutive misses required. Presence is SerialPort.GetPortNames() falling back to the Unix device node — no WMI, identical on Windows/macOS/Linux. Armed only when the port is visible to that probe at connect time, so a spelling the probe cannot see disables the check instead of reporting a false drop. Bounds unplug detection at ~3 s. - I/O fault escalation via the new ITransportHealthSink: the reader and writer loops report every read/write outcome, and 5 consecutive failures with no successful transfer between them are a drop. A blip that recovers resets the run and never disconnects. TCP had the same gap — nothing raised StatusChanged(false) on an unexpected drop — and is fixed alongside: the same escalation (including a zero-byte socket read, which means the peer closed), plus TCP keep-alive so a silently severed link is caught within ~20 s of idleness. An intentional Disconnect() disarms both detectors first, so it still reports Disconnected, never Lost. #381's "know this limit" note is removed from docs/DEVICE_INTERFACES.md and DaqifiDeviceRegistry's XML docs, replaced by a documented detection-bound table. Co-Authored-By: Claude Opus 5 --- docs/DEVICE_INTERFACES.md | 48 ++- ...reamMessageConsumerHealthReportingTests.cs | 243 ++++++++++++++ .../MessageProducerHealthReportingTests.cs | 105 ++++++ ...SerialStreamTransportDropDetectionTests.cs | 270 ++++++++++++++++ .../Transport/SerialUnplugValidationTests.cs | 289 +++++++++++++++++ .../TcpStreamTransportDropDetectionTests.cs | 195 +++++++++++ .../TransportConnectionWatchdogTests.cs | 303 ++++++++++++++++++ .../Consumers/StreamMessageConsumer.cs | 32 +- .../Producers/MessageProducer.cs | 21 +- .../Transport/ITransportHealthSink.cs | 42 +++ .../Transport/SerialStreamTransport.cs | 218 ++++++++++++- .../Transport/TcpStreamTransport.cs | 137 +++++++- .../Transport/TransportConnectionWatchdog.cs | 301 +++++++++++++++++ src/Daqifi.Core/Device/DaqifiDevice.cs | 12 +- .../Device/DaqifiDeviceRegistry.cs | 12 +- 15 files changed, 2205 insertions(+), 23 deletions(-) create mode 100644 src/Daqifi.Core.Tests/Communication/Consumers/StreamMessageConsumerHealthReportingTests.cs create mode 100644 src/Daqifi.Core.Tests/Communication/Producers/MessageProducerHealthReportingTests.cs create mode 100644 src/Daqifi.Core.Tests/Communication/Transport/SerialStreamTransportDropDetectionTests.cs create mode 100644 src/Daqifi.Core.Tests/Communication/Transport/SerialUnplugValidationTests.cs create mode 100644 src/Daqifi.Core.Tests/Communication/Transport/TcpStreamTransportDropDetectionTests.cs create mode 100644 src/Daqifi.Core.Tests/Communication/Transport/TransportConnectionWatchdogTests.cs create mode 100644 src/Daqifi.Core/Communication/Transport/ITransportHealthSink.cs create mode 100644 src/Daqifi.Core/Communication/Transport/TransportConnectionWatchdog.cs diff --git a/docs/DEVICE_INTERFACES.md b/docs/DEVICE_INTERFACES.md index 937488b4..fc87b0e4 100644 --- a/docs/DEVICE_INTERFACES.md +++ b/docs/DEVICE_INTERFACES.md @@ -315,14 +315,10 @@ The policy is asked once per connection attempt, not again after connecting. **Ownership and liveness.** The registry disconnects and disposes every device it removes, including duplicates it rejects — once a device is passed to `ConnectAsync` or `Register`, don't dispose it yourself. Registrations whose device stops reporting `IsConnected` are pruned before -every registration attempt (and by `PruneDisconnected()` on demand). The registry does not -reconnect on its own. - -> **Know this limit:** pruning is only as good as the transport's drop detection. -> `SerialStreamTransport.IsConnected` reports the OS handle's state, which stays open after a USB -> device is physically unplugged, so that device keeps reporting `IsConnected` and is *not* pruned — -> a later `ConnectAsync` under the same key hands back the dead handle rather than reconnecting. -> Devices you `Disconnect()` yourself, and drops a transport does report, prune as described. +every registration attempt (and by `PruneDisconnected()` on demand). Pruning covers a physically +unplugged USB device: the transport detects the drop itself (see +[Detecting a dropped connection](#detecting-a-dropped-connection)) and stops reporting +`IsConnected`, within the bounds documented there. The registry does not reconnect on its own. All members are safe to call from any thread; reads return snapshots, and the duplicate policy is always invoked without the internal lock held, so a policy that blocks on a user prompt never @@ -398,6 +394,42 @@ device.StatusChanged += (sender, args) => }; ``` +### Detecting a dropped connection + +`ConnectionStatus.Lost` means the connection ended without anyone asking it to — the USB cable was +pulled, the device lost power, the network dropped. `Disconnect()` reports +`ConnectionStatus.Disconnected` instead, never `Lost`, so the two are always distinguishable. + +Neither OS handle is a liveness signal on its own: `SerialPort.IsOpen` stays `true` after a USB +device is physically unplugged, and `TcpClient.Connected` describes the last completed operation +rather than the current link. The transports therefore watch for a drop directly, and the detection +time is bounded: + +| Transport | Drop | Reported within | +|---|---|---| +| Serial | Cable unplugged / device re-enumerated | **~3 s** — port-presence poll (1 s cadence, 2 consecutive misses) | +| Serial | Reads or writes failing while traffic flows | **< 1 s** — 5 consecutive I/O failures with no successful transfer between them | +| TCP | Peer closed, connection reset | **< 1 s** — same I/O fault escalation | +| TCP | Link silently severed (power loss, WiFi drop) | **~20 s** — TCP keep-alive (10 s idle, 3 s probes, 3 retries) | + +Both serial detectors are cross-platform and behave identically on Windows, macOS, and Linux: port +presence is `SerialPort.GetPortNames()` (falling back to the device node on Unix), with no WMI or +other OS-specific device-change watcher. The presence poll is armed only if the port is visible to +that probe at connect time, so an exotic port-name spelling disables the check rather than +producing a false drop. + +A single failed read never disconnects anything. Escalation requires a *run* of failures with no +successful transfer in between, so a stream that glitches and recovers stays connected. To disable +the serial presence poll and rely on I/O escalation alone, construct the transport with +`livenessCheckInterval: TimeSpan.Zero`. + +When a drop is detected the transport closes its handle, so `IsConnected` already reads `false` by +the time `StatusChanged` fires. The device's message consumer and producer are *not* stopped +automatically — call `Disconnect()` (or `Dispose()`) from your `Lost` handler to release them. + +Custom transports can opt into the same escalation by implementing `ITransportHealthSink`; the +reader and writer loops report every read/write outcome to a transport that does. + ### Working with Device Metadata After initialization, device metadata is populated: diff --git a/src/Daqifi.Core.Tests/Communication/Consumers/StreamMessageConsumerHealthReportingTests.cs b/src/Daqifi.Core.Tests/Communication/Consumers/StreamMessageConsumerHealthReportingTests.cs new file mode 100644 index 00000000..51b10398 --- /dev/null +++ b/src/Daqifi.Core.Tests/Communication/Consumers/StreamMessageConsumerHealthReportingTests.cs @@ -0,0 +1,243 @@ +using Daqifi.Core.Communication.Consumers; +using Daqifi.Core.Communication.Transport; +using System.Text; + +namespace Daqifi.Core.Tests.Communication.Consumers; + +/// +/// Issue #382: the reader loop is the first component to see a device that has gone away, so it +/// must tell the transport — both when reads keep failing (a disconnect) and when a read succeeds +/// again (a blip that must not disconnect anything). +/// +public class StreamMessageConsumerHealthReportingTests +{ + [Fact] + public void ReaderLoop_WhenStreamStartsFailing_ReportsFaultsToTheTransport() + { + using var stream = new ScriptedStream(); + var sink = new RecordingHealthSink(); + using var consumer = new StreamMessageConsumer( + stream, new LineBasedMessageParser(), healthSink: sink); + + stream.FailReads = true; + consumer.Start(); + + Assert.True(WaitUntil(() => sink.FaultCount >= 3, TimeSpan.FromSeconds(5)), + $"expected repeated fault reports, saw {sink.FaultCount}"); + Assert.Equal(0, sink.SuccessCount); + Assert.All(sink.Faults, ex => Assert.IsType(ex)); + + consumer.StopSafely(timeoutMs: 2000); + } + + [Fact] + public void ReaderLoop_WhenAFailingStreamRecovers_ReportsSuccessSoTheBlipIsNotADisconnect() + { + using var stream = new ScriptedStream(); + var sink = new RecordingHealthSink(); + using var consumer = new StreamMessageConsumer( + stream, new LineBasedMessageParser(), healthSink: sink); + + stream.FailReads = true; + consumer.Start(); + + Assert.True(WaitUntil(() => sink.FaultCount >= 1, TimeSpan.FromSeconds(5))); + + // The blip ends: reads work again. + stream.Enqueue("$DAQiFi\r\n"); + stream.FailReads = false; + + Assert.True(WaitUntil(() => sink.SuccessCount >= 1, TimeSpan.FromSeconds(5)), + "a successful read must be reported so the transport clears the failure run"); + + consumer.StopSafely(timeoutMs: 2000); + } + + [Fact] + public void ReaderLoop_OnATimeout_ReportsNeitherFaultNorSuccess() + { + // An idle read timeout is the normal state of a quiet device, not evidence of anything. + using var stream = new ScriptedStream(); + var sink = new RecordingHealthSink(); + using var consumer = new StreamMessageConsumer( + stream, new LineBasedMessageParser(), healthSink: sink); + + stream.TimeoutReads = true; + consumer.Start(); + + Assert.True(WaitUntil(() => stream.ReadCount >= 5, TimeSpan.FromSeconds(5))); + Assert.Equal(0, sink.FaultCount); + Assert.Equal(0, sink.SuccessCount); + + consumer.StopSafely(timeoutMs: 2000); + } + + [Fact] + public void ReaderLoop_OnAZeroByteReadFromANonSocketStream_ReportsNoFault() + { + // Plenty of stream types return 0 for "nothing right now"; only a socket's 0 means closed. + using var stream = new ScriptedStream(); + var sink = new RecordingHealthSink(); + using var consumer = new StreamMessageConsumer( + stream, new LineBasedMessageParser(), healthSink: sink); + + consumer.Start(); + + Assert.True(WaitUntil(() => stream.ReadCount >= 5, TimeSpan.FromSeconds(5))); + Assert.Equal(0, sink.FaultCount); + + consumer.StopSafely(timeoutMs: 2000); + } + + [Fact] + public void ReaderLoop_WithNoHealthSink_BehavesExactlyAsBefore() + { + // The feedback path is opt-in; a consumer constructed without one still just raises + // ErrorOccurred and keeps going. + using var stream = new ScriptedStream(); + var errors = 0; + using var consumer = new StreamMessageConsumer(stream, new LineBasedMessageParser()); + consumer.ErrorOccurred += (_, _) => Interlocked.Increment(ref errors); + + stream.FailReads = true; + consumer.Start(); + + Assert.True(WaitUntil(() => Volatile.Read(ref errors) >= 2, TimeSpan.FromSeconds(5))); + Assert.True(consumer.IsRunning); + + consumer.StopSafely(timeoutMs: 2000); + } + + private static bool WaitUntil(Func condition, TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + while (DateTime.UtcNow < deadline) + { + if (condition()) + { + return true; + } + + Thread.Sleep(10); + } + + return condition(); + } + + /// + /// Records what the reader loop reports, standing in for a real transport. + /// + private sealed class RecordingHealthSink : ITransportHealthSink + { + private readonly object _gate = new(); + private readonly List _faults = []; + private int _successCount; + + public int FaultCount + { + get + { + lock (_gate) + { + return _faults.Count; + } + } + } + + public IReadOnlyList Faults + { + get + { + lock (_gate) + { + return _faults.ToArray(); + } + } + } + + public int SuccessCount => Volatile.Read(ref _successCount); + + public void ReportIoFault(Exception error) + { + lock (_gate) + { + _faults.Add(error); + } + } + + public void ReportIoSuccess() => Interlocked.Increment(ref _successCount); + } + + /// + /// A stream whose reads can be made to fail, time out, or deliver scripted data, so a + /// "connection that starts failing" can be reproduced without hardware. + /// + private sealed class ScriptedStream : Stream + { + private readonly Queue _pending = new(); + private readonly object _gate = new(); + private int _readCount; + + public volatile bool FailReads; + public volatile bool TimeoutReads; + + public int ReadCount => Volatile.Read(ref _readCount); + + public void Enqueue(string text) + { + lock (_gate) + { + _pending.Enqueue(Encoding.UTF8.GetBytes(text)); + } + } + + public override int Read(byte[] buffer, int offset, int count) + { + Interlocked.Increment(ref _readCount); + + if (TimeoutReads) + { + Thread.Sleep(5); + throw new TimeoutException("no data within the read timeout"); + } + + if (FailReads) + { + Thread.Sleep(5); + throw new IOException("the device is gone"); + } + + lock (_gate) + { + if (_pending.Count == 0) + { + return 0; + } + + var chunk = _pending.Dequeue(); + var length = Math.Min(chunk.Length, count); + Array.Copy(chunk, 0, buffer, offset, length); + return length; + } + } + + public override bool CanRead => true; + public override bool CanSeek => false; + public override bool CanWrite => false; + public override long Length => throw new NotSupportedException(); + + public override long Position + { + get => throw new NotSupportedException(); + set => throw new NotSupportedException(); + } + + public override void Flush() + { + } + + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + public override void SetLength(long value) => throw new NotSupportedException(); + public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + } +} diff --git a/src/Daqifi.Core.Tests/Communication/Producers/MessageProducerHealthReportingTests.cs b/src/Daqifi.Core.Tests/Communication/Producers/MessageProducerHealthReportingTests.cs new file mode 100644 index 00000000..49d5aa73 --- /dev/null +++ b/src/Daqifi.Core.Tests/Communication/Producers/MessageProducerHealthReportingTests.cs @@ -0,0 +1,105 @@ +using Daqifi.Core.Communication.Messages; +using Daqifi.Core.Communication.Producers; +using Daqifi.Core.Communication.Transport; + +namespace Daqifi.Core.Tests.Communication.Producers; + +/// +/// Issue #382: a write that keeps failing is, like a read that keeps failing, evidence the device +/// is gone. The producer must report both outcomes to the transport instead of silently draining +/// the queue into a dead stream. +/// +public class MessageProducerHealthReportingTests +{ + [Fact] + public void WhenWritesFail_TheProducerReportsFaultsAndKeepsDraining() + { + using var stream = new FailingWriteStream { FailWrites = true }; + var sink = new CountingHealthSink(); + using var producer = new MessageProducer(stream, healthSink: sink); + + producer.Start(); + for (var i = 0; i < 5; i++) + { + producer.Send(new ScpiMessage($"CMD{i}")); + } + + Assert.True(WaitUntil(() => sink.FaultCount >= 5, TimeSpan.FromSeconds(5)), + $"expected a fault report per failed write, saw {sink.FaultCount}"); + Assert.Equal(0, sink.SuccessCount); + + // Failures must not stall the loop: the queue still drains. + Assert.True(WaitUntil(() => producer.QueuedMessageCount == 0, TimeSpan.FromSeconds(5))); + Assert.True(producer.IsRunning); + } + + [Fact] + public void WhenWritesSucceed_TheProducerReportsSuccess() + { + using var stream = new FailingWriteStream(); + var sink = new CountingHealthSink(); + using var producer = new MessageProducer(stream, healthSink: sink); + + producer.Start(); + producer.Send(new ScpiMessage("CMD")); + + Assert.True(WaitUntil(() => sink.SuccessCount >= 1, TimeSpan.FromSeconds(5))); + Assert.Equal(0, sink.FaultCount); + } + + [Fact] + public void WithNoHealthSink_TheProducerBehavesExactlyAsBefore() + { + using var stream = new FailingWriteStream { FailWrites = true }; + using var producer = new MessageProducer(stream); + + producer.Start(); + producer.Send(new ScpiMessage("CMD")); + + Assert.True(WaitUntil(() => producer.QueuedMessageCount == 0, TimeSpan.FromSeconds(5))); + Assert.True(producer.IsRunning); + } + + private static bool WaitUntil(Func condition, TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + while (DateTime.UtcNow < deadline) + { + if (condition()) + { + return true; + } + + Thread.Sleep(10); + } + + return condition(); + } + + private sealed class CountingHealthSink : ITransportHealthSink + { + private int _faultCount; + private int _successCount; + + public int FaultCount => Volatile.Read(ref _faultCount); + public int SuccessCount => Volatile.Read(ref _successCount); + + public void ReportIoFault(Exception error) => Interlocked.Increment(ref _faultCount); + public void ReportIoSuccess() => Interlocked.Increment(ref _successCount); + } + + private sealed class FailingWriteStream : MemoryStream + { + public volatile bool FailWrites; + + public override void Write(byte[] buffer, int offset, int count) + { + if (FailWrites) + { + throw new IOException("the device is gone"); + } + + base.Write(buffer, offset, count); + } + } +} diff --git a/src/Daqifi.Core.Tests/Communication/Transport/SerialStreamTransportDropDetectionTests.cs b/src/Daqifi.Core.Tests/Communication/Transport/SerialStreamTransportDropDetectionTests.cs new file mode 100644 index 00000000..d500d1ea --- /dev/null +++ b/src/Daqifi.Core.Tests/Communication/Transport/SerialStreamTransportDropDetectionTests.cs @@ -0,0 +1,270 @@ +using Daqifi.Core.Communication.Transport; + +namespace Daqifi.Core.Tests.Communication.Transport; + +/// +/// Issue #382 for the serial transport: a physically unplugged device must be reported as a drop +/// within a bounded time, an intentional disconnect must not be, and the port-presence check must +/// never fire on a platform where it cannot see the port it is watching. +/// +/// +/// The port-presence probe is substituted so "the cable was pulled" is reproducible without +/// hardware; the real probe is covered separately by +/// and the bench validation on the PR. +/// +public class SerialStreamTransportDropDetectionTests +{ + private static readonly TimeSpan FastInterval = TimeSpan.FromMilliseconds(50); + + [Fact] + public void WhenThePortStopsBeingPresent_TransportReportsTheDrop() + { + var present = true; + using var transport = new SerialStreamTransport("/dev/ttyTest382", livenessCheckInterval: FastInterval) + { + PortPresenceProbe = _ => Volatile.Read(ref present) + }; + + var dropped = new ManualResetEventSlim(false); + TransportStatusEventArgs? captured = null; + transport.StatusChanged += (_, e) => + { + if (e.IsConnected) + { + return; + } + + captured = e; + dropped.Set(); + }; + + transport.StartDropDetection(); + Assert.True(transport.IsLivenessMonitorActive); + + // The cable is pulled: the port stops being enumerated. + Volatile.Write(ref present, false); + + Assert.True(dropped.Wait(TimeSpan.FromSeconds(10)), "the transport never reported the unplug"); + Assert.False(transport.IsConnected); + Assert.IsType(captured!.Error); + Assert.Contains("/dev/ttyTest382", captured.Error!.Message); + Assert.False(transport.IsLivenessMonitorActive); + } + + [Fact] + public void WhenThePortMomentarilyDisappearsAndComesBack_NoDropIsReported() + { + // A single hiccup in the enumeration is not an unplug. + var present = 1; + using var transport = new SerialStreamTransport("/dev/ttyTest382", livenessCheckInterval: TimeSpan.FromHours(1)) + { + // Alternates absent/present on every poll, so the miss run never reaches the threshold. + PortPresenceProbe = _ => Interlocked.Increment(ref present) % 2 == 0 + }; + + var drops = 0; + transport.StatusChanged += (_, e) => + { + if (!e.IsConnected) + { + Interlocked.Increment(ref drops); + } + }; + + transport.StartDropDetection(); + + for (var i = 0; i < 50; i++) + { + transport.PollLivenessForTesting(); + } + + Assert.Equal(0, Volatile.Read(ref drops)); + Assert.True(transport.IsLivenessMonitorActive); + } + + [Fact] + public void WhenTheProbeCannotSeeThePortAtConnectTime_TheCheckIsNotArmed() + { + // A platform (or port-name spelling) the probe cannot see must disable the check rather + // than report a drop on every poll. I/O fault escalation still covers such a port. + using var transport = new SerialStreamTransport("COM-not-enumerated", livenessCheckInterval: FastInterval) + { + PortPresenceProbe = _ => false + }; + + var drops = 0; + transport.StatusChanged += (_, e) => + { + if (!e.IsConnected) + { + Interlocked.Increment(ref drops); + } + }; + + transport.StartDropDetection(); + + Assert.False(transport.IsLivenessMonitorActive); + Thread.Sleep(400); + Assert.Equal(0, Volatile.Read(ref drops)); + } + + [Fact] + public void WhenTheLivenessIntervalIsZero_TheCheckIsDisabled() + { + using var transport = new SerialStreamTransport("/dev/ttyTest382", livenessCheckInterval: TimeSpan.Zero) + { + PortPresenceProbe = _ => true + }; + + transport.StartDropDetection(); + + Assert.False(transport.IsLivenessMonitorActive); + } + + [Fact] + public void AfterAnIntentionalDisconnect_ThePortDisappearingReportsNothing() + { + // Releasing the port makes it stop being "ours"; that must never be re-reported as a loss + // on top of the disconnect the caller asked for. + var present = true; + using var transport = new SerialStreamTransport("/dev/ttyTest382", livenessCheckInterval: TimeSpan.FromHours(1)) + { + PortPresenceProbe = _ => Volatile.Read(ref present) + }; + + transport.StartDropDetection(); + Assert.True(transport.IsLivenessMonitorActive); + + transport.Disconnect(); + + var drops = 0; + transport.StatusChanged += (_, e) => + { + if (!e.IsConnected) + { + Interlocked.Increment(ref drops); + } + }; + + Volatile.Write(ref present, false); + for (var i = 0; i < 10; i++) + { + transport.PollLivenessForTesting(); + } + + Assert.False(transport.IsLivenessMonitorActive); + Assert.Equal(0, Volatile.Read(ref drops)); + } + + [Fact] + public void PersistentIoFaults_ReportTheDrop_ButABlipDoesNot() + { + using var transport = new SerialStreamTransport("/dev/ttyTest382", livenessCheckInterval: TimeSpan.Zero); + + var drops = 0; + TransportStatusEventArgs? captured = null; + transport.StatusChanged += (_, e) => + { + if (e.IsConnected) + { + return; + } + + captured = e; + Interlocked.Increment(ref drops); + }; + + transport.StartDropDetection(); + + // The blip: repeated failures that keep recovering must never disconnect. + for (var round = 0; round < 10; round++) + { + for (var i = 0; i < TransportConnectionWatchdog.ConsecutiveFaultThreshold - 1; i++) + { + transport.ReportIoFault(new IOException("blip")); + } + + transport.ReportIoSuccess(); + } + + Assert.Equal(0, Volatile.Read(ref drops)); + + // The real thing: an unbroken run of failures. + for (var i = 0; i < TransportConnectionWatchdog.ConsecutiveFaultThreshold; i++) + { + transport.ReportIoFault(new IOException("device gone")); + } + + Assert.Equal(1, Volatile.Read(ref drops)); + Assert.IsType(captured!.Error); + } + + [Fact] + public void ReportIoFault_BeforeAnyConnection_IsIgnored() + { + using var transport = new SerialStreamTransport("/dev/ttyTest382"); + + var drops = 0; + transport.StatusChanged += (_, e) => + { + if (!e.IsConnected) + { + Interlocked.Increment(ref drops); + } + }; + + for (var i = 0; i < TransportConnectionWatchdog.ConsecutiveFaultThreshold * 3; i++) + { + transport.ReportIoFault(new IOException("no connection to lose")); + } + + Assert.Equal(0, Volatile.Read(ref drops)); + } + + [Fact] + public void ReportIoFault_WithNullError_Throws() + { + using var transport = new SerialStreamTransport("/dev/ttyTest382"); + + Assert.Throws(() => transport.ReportIoFault(null!)); + } + + [Fact] + public void Constructor_WithNegativeLivenessInterval_Throws() + { + Assert.Throws(() => + new SerialStreamTransport("/dev/ttyTest382", livenessCheckInterval: TimeSpan.FromSeconds(-1))); + } + + [Fact] + public void IsPortEnumerated_ForAPortThatDoesNotExist_ReportsAbsent() + { + // The real probe, no substitution: neither the framework enumeration nor the filesystem + // knows this port on any platform. + Assert.False(SerialStreamTransport.IsPortEnumerated("COM-daqifi-382-does-not-exist")); + Assert.False(SerialStreamTransport.IsPortEnumerated("/dev/daqifi-382-does-not-exist")); + } + + [Fact] + public void IsPortEnumerated_ForAnExistingDeviceNodePath_ReportsPresent() + { + // The Unix fallback: a port the framework enumeration happens not to return is still + // present if its device node exists. Windows port names are not paths, so it does not + // apply there. + if (OperatingSystem.IsWindows()) + { + return; + } + + var path = Path.GetTempFileName(); + try + { + Assert.StartsWith("/", path); + Assert.True(SerialStreamTransport.IsPortEnumerated(path)); + } + finally + { + File.Delete(path); + } + } +} diff --git a/src/Daqifi.Core.Tests/Communication/Transport/SerialUnplugValidationTests.cs b/src/Daqifi.Core.Tests/Communication/Transport/SerialUnplugValidationTests.cs new file mode 100644 index 00000000..4c682252 --- /dev/null +++ b/src/Daqifi.Core.Tests/Communication/Transport/SerialUnplugValidationTests.cs @@ -0,0 +1,289 @@ +using Daqifi.Core.Communication.Transport; +using Daqifi.Core.Device; +using System.Diagnostics; +using Xunit.Abstractions; + +namespace Daqifi.Core.Tests.Communication.Transport; + +/// +/// Hardware-in-the-loop validation for issue #382: a human pulls a real USB cable and this +/// measures how long Core takes to report . +/// +/// +/// +/// Does nothing unless DAQIFI_UNPLUG_PORT names a serial port, so CI and ordinary local +/// runs are unaffected. Each test needs its own unplug, so run them one at a time, plugging the +/// device back in between: +/// +/// +/// DAQIFI_UNPLUG_PORT=/dev/cu.usbmodem1101 \ +/// dotnet test src/Daqifi.Core.Tests/Daqifi.Core.Tests.csproj -f net9.0 \ +/// --filter "FullyQualifiedName~UnpluggedSerialDevice_ReportsConnectionLost" \ +/// --logger "console;verbosity=detailed" +/// +/// DAQIFI_UNPLUG_PORT=/dev/cu.usbmodem1101 \ +/// dotnet test src/Daqifi.Core.Tests/Daqifi.Core.Tests.csproj -f net9.0 \ +/// --filter "FullyQualifiedName~UnpluggedSerialDevice_IsPrunedFromTheRegistry" \ +/// --logger "console;verbosity=detailed" +/// +/// +/// The first connects, holds the connection open for a warm-up window that must produce no status +/// change at all (no false positives), then prints >>> UNPLUG THE CABLE NOW and +/// waits. It watches the device node itself, so the reported detection latency is measured from +/// the moment the port actually vanished — not from the prompt. The second re-verifies PR #381's +/// registry pruning against the same event. Environment overrides: +/// DAQIFI_UNPLUG_BAUD (default 115200), DAQIFI_UNPLUG_WARMUP_SECONDS (default 20), +/// DAQIFI_UNPLUG_TIMEOUT_SECONDS (default 180). +/// +/// +public class SerialUnplugValidationTests +{ + private readonly ITestOutputHelper _output; + + public SerialUnplugValidationTests(ITestOutputHelper output) + { + _output = output; + } + + [Fact] + public void UnpluggedSerialDevice_ReportsConnectionLost_WithinTheDocumentedBound() + { + if (!TryGetHardwarePort(out var portName)) + { + return; + } + + var baudRate = ReadInt("DAQIFI_UNPLUG_BAUD", 115200); + var warmup = TimeSpan.FromSeconds(ReadInt("DAQIFI_UNPLUG_WARMUP_SECONDS", 20)); + var timeout = TimeSpan.FromSeconds(ReadInt("DAQIFI_UNPLUG_TIMEOUT_SECONDS", 180)); + + var transport = new SerialStreamTransport(portName, baudRate); + using var device = new DaqifiDevice("Unplug Validation", transport); + + var statuses = new List<(DateTime At, ConnectionStatus Status)>(); + var lost = new ManualResetEventSlim(false); + var lostAt = DateTime.MinValue; + + device.StatusChanged += (_, e) => + { + var now = DateTime.UtcNow; + lock (statuses) + { + statuses.Add((now, e.Status)); + } + + Log($"StatusChanged: {e.Status}"); + + if (e.Status == ConnectionStatus.Lost) + { + lostAt = now; + lost.Set(); + } + }; + + Log($"Connecting to {portName} @ {baudRate} baud..."); + device.Connect(); + Assert.Equal(ConnectionStatus.Connected, device.Status); + Log("Connected."); + + // Phase 1 — a healthy connection must produce no status change at all. This is where a + // liveness check that is too eager, or fault escalation that is too twitchy, shows up. + Log($"Holding the connection open for {warmup.TotalSeconds:0}s; no status change is expected."); + var beforeWarmup = SnapshotCount(statuses); + Thread.Sleep(warmup); + Assert.Equal(beforeWarmup, SnapshotCount(statuses)); + Assert.Equal(ConnectionStatus.Connected, device.Status); + Assert.True(device.IsConnected); + Log("No false positives during the healthy window."); + + // Phase 2 — the human pulls the cable. Watch the device node so detection latency is + // measured from when the port really went away. + using var watcher = new PortDisappearanceWatcher(portName); + + Log(string.Empty); + Log(">>> UNPLUG THE CABLE NOW <<<"); + Log($" (waiting up to {timeout.TotalSeconds:0}s for ConnectionStatus.Lost)"); + Log(string.Empty); + + var sawLost = lost.Wait(timeout); + + var portGoneAt = watcher.PortGoneAtUtc; + if (portGoneAt.HasValue) + { + Log($"Port vanished from the system at {portGoneAt:HH:mm:ss.fff}Z"); + } + + Assert.True(sawLost, + $"No ConnectionStatus.Lost within {timeout.TotalSeconds:0}s. " + + (portGoneAt.HasValue + ? "The port did disappear, so detection failed." + : "The port never disappeared — was the cable actually unplugged?")); + + Log($"ConnectionStatus.Lost observed at {lostAt:HH:mm:ss.fff}Z"); + + if (portGoneAt.HasValue) + { + var detection = lostAt - portGoneAt.Value; + Log($"DETECTION LATENCY: {detection.TotalMilliseconds:0} ms"); + + // The documented bound is ~3s (1s poll cadence x 2 consecutive misses, plus up to one + // interval of phase). Assert with slack so a loaded machine does not produce a false + // failure; the printed number above is the real result. + Assert.True(detection < TimeSpan.FromSeconds(10), + $"Detection took {detection.TotalSeconds:0.00}s, well past the documented ~3s bound."); + } + + Assert.Equal(ConnectionStatus.Lost, device.Status); + Assert.False(device.IsConnected); + Assert.False(transport.IsConnected); + + lock (statuses) + { + Assert.DoesNotContain(ConnectionStatus.Disconnected, statuses.Select(s => s.Status)); + } + + Log("PASS — unplug reported as Lost, not Disconnected."); + } + + [Fact] + public void UnpluggedSerialDevice_IsPrunedFromTheRegistry() + { + // Re-verifies PR #381's stale-registration pruning against a real unplug: the registry + // prunes registrations whose device stops reporting IsConnected, which before #382 never + // happened for a physically unplugged USB device. + if (!TryGetHardwarePort(out var portName)) + { + return; + } + + var baudRate = ReadInt("DAQIFI_UNPLUG_BAUD", 115200); + var timeout = TimeSpan.FromSeconds(ReadInt("DAQIFI_UNPLUG_TIMEOUT_SECONDS", 180)); + + using var registry = new DaqifiDeviceRegistry(); + var transport = new SerialStreamTransport(portName, baudRate); + var device = new DaqifiDevice("Unplug Validation", transport); + + var removals = new List(); + registry.DeviceRemoved += (_, e) => + { + lock (removals) + { + removals.Add(e.Reason); + } + }; + + device.Connect(); + var registration = registry.Register(device, key: "unplug-validation"); + Assert.Equal(DeviceRegistrationOutcome.Registered, registration.Outcome); + Assert.Single(registry.Devices); + + Log(string.Empty); + Log(">>> UNPLUG THE CABLE NOW <<<"); + Log(string.Empty); + + var sw = Stopwatch.StartNew(); + while (sw.Elapsed < timeout && device.IsConnected) + { + Thread.Sleep(100); + } + + Assert.False(device.IsConnected, $"Device still reported IsConnected after {timeout.TotalSeconds:0}s."); + Log($"Device stopped reporting IsConnected after {sw.Elapsed.TotalSeconds:0.00}s."); + + Assert.Equal(1, registry.PruneDisconnected()); + Assert.Empty(registry.Devices); + + lock (removals) + { + Assert.Contains(DeviceRemovalReason.Disconnected, removals); + } + + Log("PASS — the unplugged device was pruned from the registry."); + } + + private static int SnapshotCount(List<(DateTime At, ConnectionStatus Status)> statuses) + { + lock (statuses) + { + return statuses.Count; + } + } + + /// + /// Resolves the port to validate against, or reports that this run is not a hardware run. + /// + /// + /// xUnit 2.x has no dynamic skip (Assert.Skip arrived in v3), and a statically skipped + /// [Fact(Skip = ...)] cannot be turned on by the operator at all — so an unconfigured + /// run logs why it did nothing and returns. The hardware assertions below are the point of the + /// test; they run whenever DAQIFI_UNPLUG_PORT is set. + /// + private bool TryGetHardwarePort(out string portName) + { + portName = Environment.GetEnvironmentVariable("DAQIFI_UNPLUG_PORT") ?? string.Empty; + if (!string.IsNullOrWhiteSpace(portName)) + { + return true; + } + + Log("SKIPPED — set DAQIFI_UNPLUG_PORT to a connected DAQiFi serial port to run this " + + "hardware validation (see the class remarks for the full command line)."); + return false; + } + + private static int ReadInt(string variable, int fallback) + { + var raw = Environment.GetEnvironmentVariable(variable); + return int.TryParse(raw, out var value) && value > 0 ? value : fallback; + } + + private void Log(string message) + { + var line = message.Length == 0 ? string.Empty : $"[{DateTime.UtcNow:HH:mm:ss.fff}Z] {message}"; + _output.WriteLine(line); + Console.WriteLine(line); + } + + /// + /// Records the moment the serial port stops being present, so detection latency is measured + /// from the unplug itself rather than from the operator's prompt. + /// + private sealed class PortDisappearanceWatcher : IDisposable + { + private readonly CancellationTokenSource _cts = new(); + private readonly Thread _thread; + private DateTime? _portGoneAtUtc; + + public PortDisappearanceWatcher(string portName) + { + _thread = new Thread(() => + { + while (!_cts.IsCancellationRequested) + { + if (!SerialStreamTransport.IsPortEnumerated(portName)) + { + _portGoneAtUtc = DateTime.UtcNow; + return; + } + + Thread.Sleep(20); + } + }) + { + IsBackground = true, + Name = "PortDisappearanceWatcher" + }; + + _thread.Start(); + } + + public DateTime? PortGoneAtUtc => _portGoneAtUtc; + + public void Dispose() + { + _cts.Cancel(); + _thread.Join(TimeSpan.FromSeconds(2)); + _cts.Dispose(); + } + } +} diff --git a/src/Daqifi.Core.Tests/Communication/Transport/TcpStreamTransportDropDetectionTests.cs b/src/Daqifi.Core.Tests/Communication/Transport/TcpStreamTransportDropDetectionTests.cs new file mode 100644 index 00000000..3c2bc816 --- /dev/null +++ b/src/Daqifi.Core.Tests/Communication/Transport/TcpStreamTransportDropDetectionTests.cs @@ -0,0 +1,195 @@ +using Daqifi.Core.Communication.Consumers; +using Daqifi.Core.Communication.Transport; +using Daqifi.Core.Device; +using System.Net; +using System.Net.Sockets; + +namespace Daqifi.Core.Tests.Communication.Transport; + +/// +/// Issue #382 for the TCP transport: an unexpected drop must raise StatusChanged(false) — +/// which is what makes reachable — while an intentional +/// disconnect must stay an ordinary disconnect. Driven end to end over a real loopback socket +/// with a real reader loop, so the whole feedback path is exercised. +/// +public class TcpStreamTransportDropDetectionTests +{ + [Fact] + public void WhenThePeerClosesTheConnection_TransportReportsTheDrop() + { + using var listener = new LoopbackListener(); + using var transport = new TcpStreamTransport(IPAddress.Loopback, listener.Port); + + var drops = new List(); + var dropped = new ManualResetEventSlim(false); + transport.StatusChanged += (_, e) => + { + if (e.IsConnected) + { + return; + } + + lock (drops) + { + drops.Add(e); + } + + dropped.Set(); + }; + + transport.Connect(); + using var server = listener.AcceptOne(); + + using var consumer = new StreamMessageConsumer( + transport.Stream, new LineBasedMessageParser(), healthSink: transport); + consumer.Start(); + + // The peer goes away. The reader loop is the only thing that can notice. + server.Close(); + + Assert.True(dropped.Wait(TimeSpan.FromSeconds(10)), "the transport never reported the drop"); + Assert.False(transport.IsConnected); + + TransportStatusEventArgs drop; + lock (drops) + { + drop = Assert.Single(drops); + } + + Assert.IsType(drop.Error); + + consumer.StopSafely(timeoutMs: 2000); + } + + [Fact] + public void WhenDisconnectedIntentionally_TransportReportsAnOrdinaryDisconnect() + { + using var listener = new LoopbackListener(); + using var transport = new TcpStreamTransport(IPAddress.Loopback, listener.Port); + + var drops = new List(); + transport.StatusChanged += (_, e) => + { + if (e.IsConnected) + { + return; + } + + lock (drops) + { + drops.Add(e); + } + }; + + transport.Connect(); + using var server = listener.AcceptOne(); + + using var consumer = new StreamMessageConsumer( + transport.Stream, new LineBasedMessageParser(), healthSink: transport); + consumer.Start(); + + transport.Disconnect(); + + // Tearing down the socket makes the reader loop fail repeatedly; none of that may be + // reported as a drop on top of the disconnect the caller asked for. + Thread.Sleep(500); + consumer.StopSafely(timeoutMs: 2000); + + lock (drops) + { + var status = Assert.Single(drops); + Assert.Null(status.Error); + } + } + + [Fact] + public void DeviceOverTcp_WhenThePeerDisappears_ReportsConnectionLost() + { + // The acceptance criterion end to end: a real transport, a real device, an unexpected + // drop, and ConnectionStatus.Lost coming out the other side. + using var listener = new LoopbackListener(); + var transport = new TcpStreamTransport(IPAddress.Loopback, listener.Port); + using var device = new DaqifiDevice("Loopback Device", transport); + + var lost = new ManualResetEventSlim(false); + var statuses = new List(); + device.StatusChanged += (_, e) => + { + lock (statuses) + { + statuses.Add(e.Status); + } + + if (e.Status == ConnectionStatus.Lost) + { + lost.Set(); + } + }; + + device.Connect(); + using var server = listener.AcceptOne(); + Assert.Equal(ConnectionStatus.Connected, device.Status); + + server.Close(); + + Assert.True(lost.Wait(TimeSpan.FromSeconds(10)), "the device never reported ConnectionStatus.Lost"); + Assert.Equal(ConnectionStatus.Lost, device.Status); + Assert.False(device.IsConnected); + } + + [Fact] + public void DeviceOverTcp_WhenDisconnectedIntentionally_ReportsDisconnectedNotLost() + { + using var listener = new LoopbackListener(); + var transport = new TcpStreamTransport(IPAddress.Loopback, listener.Port); + using var device = new DaqifiDevice("Loopback Device", transport); + + device.Connect(); + using var server = listener.AcceptOne(); + Assert.Equal(ConnectionStatus.Connected, device.Status); + + var statuses = new List(); + device.StatusChanged += (_, e) => + { + lock (statuses) + { + statuses.Add(e.Status); + } + }; + + device.Disconnect(); + + // Give any lingering reader-loop failure a chance to be (wrongly) escalated. + Thread.Sleep(500); + + lock (statuses) + { + Assert.DoesNotContain(ConnectionStatus.Lost, statuses); + Assert.Contains(ConnectionStatus.Disconnected, statuses); + } + + Assert.Equal(ConnectionStatus.Disconnected, device.Status); + } + + /// + /// A loopback TCP listener on an OS-assigned port, so these tests never collide with each + /// other or with anything else on the machine. + /// + private sealed class LoopbackListener : IDisposable + { + private readonly TcpListener _listener; + + public LoopbackListener() + { + _listener = new TcpListener(IPAddress.Loopback, 0); + _listener.Start(); + Port = ((IPEndPoint)_listener.LocalEndpoint).Port; + } + + public int Port { get; } + + public TcpClient AcceptOne() => _listener.AcceptTcpClient(); + + public void Dispose() => _listener.Stop(); + } +} diff --git a/src/Daqifi.Core.Tests/Communication/Transport/TransportConnectionWatchdogTests.cs b/src/Daqifi.Core.Tests/Communication/Transport/TransportConnectionWatchdogTests.cs new file mode 100644 index 00000000..e1a77480 --- /dev/null +++ b/src/Daqifi.Core.Tests/Communication/Transport/TransportConnectionWatchdogTests.cs @@ -0,0 +1,303 @@ +using Daqifi.Core.Communication.Transport; + +namespace Daqifi.Core.Tests.Communication.Transport; + +/// +/// Coverage for issue #382's decision logic: when a run of I/O failures — or a port that stopped +/// being enumerated — means the connection is gone, and, just as importantly, when it does not. +/// +public class TransportConnectionWatchdogTests +{ + private static TransportConnectionWatchdog CreateWatchdog(List losses) + { + return new TransportConnectionWatchdog("Test transport (X)", losses.Add); + } + + [Fact] + public void RecordFault_BelowThreshold_DoesNotSignalLoss() + { + var losses = new List(); + using var watchdog = CreateWatchdog(losses); + watchdog.Arm(); + + for (var i = 0; i < TransportConnectionWatchdog.ConsecutiveFaultThreshold - 1; i++) + { + watchdog.RecordFault(new IOException("transient")); + } + + Assert.Empty(losses); + Assert.True(watchdog.IsArmed); + } + + [Fact] + public void RecordFault_AtThreshold_SignalsLossOnceWithTheFailureAsInnerException() + { + var losses = new List(); + using var watchdog = CreateWatchdog(losses); + watchdog.Arm(); + + var last = new IOException("device gone"); + for (var i = 0; i < TransportConnectionWatchdog.ConsecutiveFaultThreshold; i++) + { + watchdog.RecordFault(i == TransportConnectionWatchdog.ConsecutiveFaultThreshold - 1 + ? last + : new IOException("earlier")); + } + + var loss = Assert.Single(losses); + Assert.IsType(loss); + Assert.Same(last, loss.InnerException); + Assert.Contains("Test transport (X)", loss.Message); + + // Disarmed by the signal: further failures on the same dead connection must not re-notify. + Assert.False(watchdog.IsArmed); + for (var i = 0; i < TransportConnectionWatchdog.ConsecutiveFaultThreshold * 2; i++) + { + watchdog.RecordFault(new IOException("still gone")); + } + + Assert.Single(losses); + } + + [Fact] + public void RecordSuccess_BetweenFaults_PreventsEscalation() + { + // The "recoverable blip": a stream that fails, recovers, fails again, and so on must never + // be torn down, no matter how many total failures accumulate. + var losses = new List(); + using var watchdog = CreateWatchdog(losses); + watchdog.Arm(); + + for (var round = 0; round < 20; round++) + { + for (var i = 0; i < TransportConnectionWatchdog.ConsecutiveFaultThreshold - 1; i++) + { + watchdog.RecordFault(new IOException("blip")); + } + + watchdog.RecordSuccess(); + } + + Assert.Empty(losses); + Assert.True(watchdog.IsArmed); + } + + [Fact] + public void RecordFault_WhenNotArmed_IsIgnored() + { + // Before a connect and after an intentional disconnect there is nothing to lose. + var losses = new List(); + using var watchdog = CreateWatchdog(losses); + + for (var i = 0; i < TransportConnectionWatchdog.ConsecutiveFaultThreshold * 2; i++) + { + watchdog.RecordFault(new IOException("noise")); + } + + Assert.Empty(losses); + } + + [Fact] + public void Arm_AfterAPreviousLoss_ClearsTheFailureRun() + { + var losses = new List(); + using var watchdog = CreateWatchdog(losses); + watchdog.Arm(); + + for (var i = 0; i < TransportConnectionWatchdog.ConsecutiveFaultThreshold; i++) + { + watchdog.RecordFault(new IOException("gone")); + } + + Assert.Single(losses); + + // Reconnect: the next cycle must start from zero, not one failure away from disconnecting. + watchdog.Arm(); + watchdog.RecordFault(new IOException("first of the new cycle")); + + Assert.Single(losses); + Assert.True(watchdog.IsArmed); + } + + [Fact] + public void Disarm_SuppressesAnInFlightFailureRun() + { + // Closing a port during an intentional disconnect makes the in-flight read fail. That must + // report a disconnect, never a loss. + var losses = new List(); + using var watchdog = CreateWatchdog(losses); + watchdog.Arm(); + + for (var i = 0; i < TransportConnectionWatchdog.ConsecutiveFaultThreshold - 1; i++) + { + watchdog.RecordFault(new IOException("closing")); + } + + watchdog.Disarm(); + + for (var i = 0; i < TransportConnectionWatchdog.ConsecutiveFaultThreshold * 2; i++) + { + watchdog.RecordFault(new IOException("closing")); + } + + Assert.Empty(losses); + } + + [Fact] + public void PollPresence_BelowMissThreshold_DoesNotSignalLoss() + { + var losses = new List(); + using var watchdog = CreateWatchdog(losses); + watchdog.Arm(); + + var present = true; + watchdog.StartPresencePolling(() => present, "port gone", TimeSpan.FromHours(1)); + + present = false; + for (var i = 0; i < TransportConnectionWatchdog.PresenceMissThreshold - 1; i++) + { + watchdog.PollPresence(); + } + + Assert.Empty(losses); + } + + [Fact] + public void PollPresence_AtMissThreshold_SignalsLoss() + { + var losses = new List(); + using var watchdog = CreateWatchdog(losses); + watchdog.Arm(); + + var present = true; + watchdog.StartPresencePolling(() => present, "port gone", TimeSpan.FromHours(1)); + + present = false; + for (var i = 0; i < TransportConnectionWatchdog.PresenceMissThreshold; i++) + { + watchdog.PollPresence(); + } + + var loss = Assert.Single(losses); + Assert.IsType(loss); + Assert.Equal("port gone", loss.Message); + Assert.False(watchdog.IsPollingPresence); + } + + [Fact] + public void PollPresence_AfterASingleMissThatRecovers_DoesNotSignalLoss() + { + // A one-off enumeration hiccup is not an unplug. + var losses = new List(); + using var watchdog = CreateWatchdog(losses); + watchdog.Arm(); + + var present = true; + watchdog.StartPresencePolling(() => present, "port gone", TimeSpan.FromHours(1)); + + for (var round = 0; round < 10; round++) + { + present = false; + watchdog.PollPresence(); + present = true; + watchdog.PollPresence(); + } + + Assert.Empty(losses); + } + + [Fact] + public void PollPresence_WhenProbeThrows_DoesNotSignalLoss() + { + // A broken observation is not evidence of a drop. + var losses = new List(); + using var watchdog = CreateWatchdog(losses); + watchdog.Arm(); + + watchdog.StartPresencePolling(() => throw new UnauthorizedAccessException(), "port gone", + TimeSpan.FromHours(1)); + + for (var i = 0; i < TransportConnectionWatchdog.PresenceMissThreshold * 5; i++) + { + watchdog.PollPresence(); + } + + Assert.Empty(losses); + Assert.True(watchdog.IsArmed); + } + + [Fact] + public void StartPresencePolling_WhenNotArmed_DoesNotStart() + { + var losses = new List(); + using var watchdog = CreateWatchdog(losses); + + watchdog.StartPresencePolling(() => false, "port gone", TimeSpan.FromHours(1)); + + Assert.False(watchdog.IsPollingPresence); + } + + [Fact] + public void StartPresencePolling_WithNonPositiveInterval_DoesNotStart() + { + var losses = new List(); + using var watchdog = CreateWatchdog(losses); + watchdog.Arm(); + + watchdog.StartPresencePolling(() => false, "port gone", TimeSpan.Zero); + + Assert.False(watchdog.IsPollingPresence); + } + + [Fact] + public void PollPresence_OnTimerCadence_SignalsLossWithinTheDocumentedBound() + { + // The bound the docs promise: miss-threshold intervals of absence, plus up to one interval + // of phase between the drop and the next poll. + var signalled = new ManualResetEventSlim(false); + using var watchdog = new TransportConnectionWatchdog("Test transport (X)", _ => signalled.Set()); + watchdog.Arm(); + + var interval = TimeSpan.FromMilliseconds(50); + watchdog.StartPresencePolling(() => false, "port gone", interval); + + var bound = interval * (TransportConnectionWatchdog.PresenceMissThreshold + 1); + Assert.True(signalled.Wait(bound + TimeSpan.FromSeconds(5))); + } + + [Fact] + public void Disarm_StopsPresencePolling() + { + var losses = new List(); + using var watchdog = CreateWatchdog(losses); + watchdog.Arm(); + watchdog.StartPresencePolling(() => false, "port gone", TimeSpan.FromHours(1)); + + Assert.True(watchdog.IsPollingPresence); + + watchdog.Disarm(); + + Assert.False(watchdog.IsPollingPresence); + Assert.False(watchdog.IsArmed); + } + + [Fact] + public void ConcurrentFaults_SignalLossExactlyOnce() + { + // Reader and writer loops report from different threads; the loss must be raised once. + var losses = new List(); + var gate = new object(); + using var watchdog = new TransportConnectionWatchdog("Test transport (X)", ex => + { + lock (gate) + { + losses.Add(ex); + } + }); + watchdog.Arm(); + + Parallel.For(0, 200, _ => watchdog.RecordFault(new IOException("gone"))); + + Assert.Single(losses); + } +} diff --git a/src/Daqifi.Core/Communication/Consumers/StreamMessageConsumer.cs b/src/Daqifi.Core/Communication/Consumers/StreamMessageConsumer.cs index 686f2d24..4539b336 100644 --- a/src/Daqifi.Core/Communication/Consumers/StreamMessageConsumer.cs +++ b/src/Daqifi.Core/Communication/Consumers/StreamMessageConsumer.cs @@ -1,4 +1,5 @@ using Daqifi.Core.Communication.Messages; +using Daqifi.Core.Communication.Transport; using System.Net.Sockets; using System.Text; @@ -13,6 +14,7 @@ public class StreamMessageConsumer : IMessageConsumer { private readonly Stream _stream; private readonly IMessageParser _messageParser; + private readonly ITransportHealthSink? _healthSink; private readonly byte[] _buffer; private readonly List _messageBuffer; @@ -63,10 +65,19 @@ public class StreamMessageConsumer : IMessageConsumer /// The stream to read messages from. /// The parser to convert raw data to messages. /// The size of the read buffer in bytes. - public StreamMessageConsumer(Stream stream, IMessageParser messageParser, int bufferSize = 4096) + /// + /// Optional transport to report read outcomes to. This reader loop is the first thing to see a + /// device that has gone away, and a transport cannot tell from its own OS handle — so when + /// supplied, every failed read and every successful read is reported, letting the transport + /// escalate a persistently failing stream to a lost connection (issue #382). When null, read + /// failures are only surfaced through , exactly as before. + /// + public StreamMessageConsumer(Stream stream, IMessageParser messageParser, int bufferSize = 4096, + ITransportHealthSink? healthSink = null) { _stream = stream ?? throw new ArgumentNullException(nameof(stream)); _messageParser = messageParser ?? throw new ArgumentNullException(nameof(messageParser)); + _healthSink = healthSink; _buffer = new byte[bufferSize]; _messageBuffer = new List(); } @@ -348,6 +359,11 @@ private void ProcessMessages() } catch (Exception ex) { + // Tell the transport before raising the error event: a read that keeps + // throwing is how a physically disconnected device announces itself, and the + // transport is the only thing that can turn a run of them into a lost + // connection (issue #382). One failure escalates nothing. + _healthSink?.ReportIoFault(ex); OnErrorOccurred(ex); Thread.Sleep(100); // Back off on error continue; @@ -355,10 +371,24 @@ private void ProcessMessages() if (bytesRead == 0) { + // A NetworkStream returns 0 only after the peer has performed an orderly + // shutdown — the connection is over, and every subsequent read will also + // return 0. Report it so the transport can escalate; other stream types + // legitimately return 0 for "nothing right now" and must not be escalated. + if (_stream is NetworkStream) + { + _healthSink?.ReportIoFault(new EndOfStreamException( + "The remote endpoint closed the connection (a socket read returned 0 bytes).")); + } + Thread.Sleep(10); // No data available, wait briefly continue; } + // A successful read clears any run of failures the transport has accumulated, so + // a stream that glitches and recovers is never mistaken for a disconnected device. + _healthSink?.ReportIoSuccess(); + // Add received data to message buffer (guarded: a caller may be reading // QueuedMessageCount and ClearBuffer's drain runs on this same thread). lock (_bufferLock) diff --git a/src/Daqifi.Core/Communication/Producers/MessageProducer.cs b/src/Daqifi.Core/Communication/Producers/MessageProducer.cs index 44c09555..c1358fa9 100644 --- a/src/Daqifi.Core/Communication/Producers/MessageProducer.cs +++ b/src/Daqifi.Core/Communication/Producers/MessageProducer.cs @@ -1,4 +1,5 @@ using Daqifi.Core.Communication.Messages; +using Daqifi.Core.Communication.Transport; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; using System.Collections.Concurrent; @@ -14,6 +15,7 @@ public class MessageProducer : IMessageProducer { private readonly Stream _stream; private readonly ILogger> _logger; + private readonly ITransportHealthSink? _healthSink; private readonly ConcurrentQueue> _messageQueue; private readonly ManualResetEventSlim _messageAvailable = new(false); private volatile bool _isRunning; @@ -29,11 +31,19 @@ public class MessageProducer : IMessageProducer /// events. When omitted, a is used so existing /// consumers behave exactly as before. /// + /// + /// Optional transport to report write outcomes to. A write that keeps failing is, like a read + /// that keeps failing, evidence the device is gone; reporting it lets the transport escalate a + /// dead link to a lost connection instead of the producer quietly draining into nothing + /// (issue #382). When null, write failures are only logged, exactly as before. + /// /// Thrown when stream is null. - public MessageProducer(Stream stream, ILogger>? logger = null) + public MessageProducer(Stream stream, ILogger>? logger = null, + ITransportHealthSink? healthSink = null) { _stream = stream ?? throw new ArgumentNullException(nameof(stream)); _logger = logger ?? NullLogger>.Instance; + _healthSink = healthSink; _messageQueue = new ConcurrentQueue>(); } @@ -161,11 +171,18 @@ private void ProcessMessages() try { WriteMessageToStream(message); + + // A successful write clears any run of failures the transport has + // accumulated: the link is demonstrably alive. + _healthSink?.ReportIoSuccess(); } catch (Exception ex) { // Surface the failure but keep draining the queue so a single - // bad write doesn't stall the remaining messages. + // bad write doesn't stall the remaining messages. Tell the transport + // too — it is the only component that can decide a run of failures + // means the device is gone rather than glitching. + _healthSink?.ReportIoFault(ex); SafeLog(() => _logger.LogWarning(ex, "Failed to write message to the stream; continuing with remaining queued messages.")); } } diff --git a/src/Daqifi.Core/Communication/Transport/ITransportHealthSink.cs b/src/Daqifi.Core/Communication/Transport/ITransportHealthSink.cs new file mode 100644 index 00000000..b0c22343 --- /dev/null +++ b/src/Daqifi.Core/Communication/Transport/ITransportHealthSink.cs @@ -0,0 +1,42 @@ +namespace Daqifi.Core.Communication.Transport; + +/// +/// Optional capability implemented by a transport that wants to be told how the I/O on its +/// is actually going, so it can notice a connection that +/// died underneath it. +/// +/// +/// +/// A transport only ever sees its own handle, and an OS handle is a poor liveness signal: +/// stays true after a USB device is +/// physically unplugged, and reflects the +/// last completed operation rather than the current link. The component that actually drives the +/// stream — a reader loop or a +/// writer loop — is the first to know, and this +/// interface is the path back from it to the transport (issue #382). +/// +/// +/// Implementations must be safe to call from any thread and must tolerate being called at a high +/// rate: runs once per successful read. +/// +/// +/// A single failure means nothing — serial and socket reads fail transiently. Implementations are +/// expected to escalate only on a run of consecutive failures with no successful transfer in +/// between, so a recoverable blip never tears down a healthy connection. +/// +/// +public interface ITransportHealthSink +{ + /// + /// Reports that a read or write against the transport's stream failed for a reason other than + /// an expected idle timeout. + /// + /// The exception the failed operation raised. + void ReportIoFault(Exception error); + + /// + /// Reports that a read or write against the transport's stream completed successfully, which + /// clears any run of failures accumulated so far. + /// + void ReportIoSuccess(); +} diff --git a/src/Daqifi.Core/Communication/Transport/SerialStreamTransport.cs b/src/Daqifi.Core/Communication/Transport/SerialStreamTransport.cs index bdb868b2..ce50d5cf 100644 --- a/src/Daqifi.Core/Communication/Transport/SerialStreamTransport.cs +++ b/src/Daqifi.Core/Communication/Transport/SerialStreamTransport.cs @@ -7,8 +7,48 @@ namespace Daqifi.Core.Communication.Transport; /// over serial connections. Handles connection lifecycle and provides the underlying /// SerialPort BaseStream for message producers and consumers. /// -public class SerialStreamTransport : IStreamTransport +/// +/// +/// Drop detection. reflects the OS handle, which stays open +/// after a USB device is physically unplugged, so it is not on its own a liveness signal. Two +/// detectors sit on top of it (issue #382), and whichever notices first closes the port and raises +/// with IsConnected == false, which a +/// surfaces as : +/// +/// +/// +/// Port-presence polling at (1 s), requiring +/// consecutive misses. This bounds +/// detection of an unplug at roughly three seconds even on a completely idle connection. +/// Presence is containing the port, falling back to the +/// device node still existing on Unix — no WMI, so it behaves identically on Windows, macOS, and +/// Linux. It is armed only if the port is visible to that probe at connect time, so a platform or +/// port-name spelling the probe cannot see disables the check rather than reporting a false drop. +/// +/// +/// I/O fault escalation via : the reader/writer loops +/// report failures, and +/// consecutive failures with no successful transfer in between are treated as a drop. Under +/// traffic this reports within a few hundred milliseconds; a single failed read that then recovers +/// never disconnects. +/// +/// +/// +/// An intentional disarms both detectors first, so it reports a normal +/// disconnect and never a loss. +/// +/// +public class SerialStreamTransport : IStreamTransport, ITransportHealthSink { + /// + /// Default cadence at which an established connection is checked for the continued presence of + /// its serial port. Combined with + /// consecutive misses, an + /// unplugged device is reported within roughly three seconds. + /// + public static readonly TimeSpan DefaultLivenessCheckInterval = + TransportConnectionWatchdog.DefaultPresencePollInterval; + private readonly string _portName; private readonly int _baudRate; private readonly Parity _parity; @@ -16,9 +56,17 @@ public class SerialStreamTransport : IStreamTransport private readonly StopBits _stopBits; private readonly bool _enableDtr; private readonly bool _enableRts; + private readonly TimeSpan _livenessCheckInterval; + private readonly TransportConnectionWatchdog _watchdog; private SerialPort? _serialPort; private bool _disposed; + /// + /// Test seam: replaces the "is this port still enumerated?" probe so the liveness check can be + /// exercised without unplugging real hardware. Never set in production. + /// + internal Func? PortPresenceProbe { get; set; } + /// /// Initializes a new instance of the SerialStreamTransport class. /// @@ -29,8 +77,17 @@ public class SerialStreamTransport : IStreamTransport /// The stop bits setting. /// Whether to enable Data Terminal Ready (DTR) signal. Default is true. /// Whether to enable Request To Send (RTS) signal. Default is false. + /// + /// How often an established connection re-checks that its port is still present. Defaults to + /// ; pass to disable the + /// check and rely on I/O fault escalation alone. + /// + /// + /// Thrown when is negative. + /// public SerialStreamTransport(string portName, int baudRate = 9600, Parity parity = Parity.None, - int dataBits = 8, StopBits stopBits = StopBits.One, bool enableDtr = true, bool enableRts = false) + int dataBits = 8, StopBits stopBits = StopBits.One, bool enableDtr = true, bool enableRts = false, + TimeSpan? livenessCheckInterval = null) { _portName = portName ?? throw new ArgumentNullException(nameof(portName)); _baudRate = baudRate; @@ -39,6 +96,17 @@ public SerialStreamTransport(string portName, int baudRate = 9600, Parity parity _stopBits = stopBits; _enableDtr = enableDtr; _enableRts = enableRts; + + _livenessCheckInterval = livenessCheckInterval ?? DefaultLivenessCheckInterval; + if (_livenessCheckInterval < TimeSpan.Zero) + { + throw new ArgumentOutOfRangeException(nameof(livenessCheckInterval), _livenessCheckInterval, + "Liveness check interval cannot be negative. Use TimeSpan.Zero to disable the check."); + } + + _watchdog = new TransportConnectionWatchdog( + $"Serial transport ({_portName})", + HandleConnectionLost); } /// @@ -185,6 +253,8 @@ await ConnectRetryExecutor.ExecuteAsync( _serialPort = null; }, onStatusChanged: OnStatusChanged); + + StartDropDetection(); } /// @@ -193,6 +263,11 @@ await ConnectRetryExecutor.ExecuteAsync( /// A task representing the asynchronous disconnect operation. public async Task DisconnectAsync() { + // Disarm before touching the port. Closing it makes any in-flight read fail, and the + // port stops being "present" the moment it is released, so a still-armed watchdog could + // report an intentional disconnect as a lost connection. + _watchdog.Disarm(); + if (!IsConnected) return; @@ -240,6 +315,144 @@ public static string[] GetAvailablePortNames() return SerialPort.GetPortNames(); } + /// + /// + /// Escalates to a lost connection only after + /// failures with no + /// successful transfer in between, so a transient read error cannot tear down a healthy port. + /// + public void ReportIoFault(Exception error) + { + ArgumentNullException.ThrowIfNull(error); + _watchdog.RecordFault(error); + } + + /// + public void ReportIoSuccess() + { + _watchdog.RecordSuccess(); + } + + /// + /// Gets a value indicating whether port-presence polling is currently running. Exposed for + /// tests and diagnostics; polling is skipped when the interval is + /// or the port is not visible to the presence probe at connect time. + /// + internal bool IsLivenessMonitorActive => _watchdog.IsPollingPresence; + + /// + /// Test seam: runs one port-presence check immediately instead of waiting for the timer. + /// + internal void PollLivenessForTesting() => _watchdog.PollPresence(); + + /// + /// Arms drop detection after a successful open and, when possible, starts port-presence polling. + /// + /// + /// Internal rather than private so the arming decision and the presence-driven loss can be + /// exercised without a real port to unplug (paired with ). + /// + internal void StartDropDetection() + { + _watchdog.Arm(); + + if (_livenessCheckInterval <= TimeSpan.Zero) + { + return; + } + + // Only arm the presence check if the probe can actually see the port we just opened. + // If a platform enumerates it under a different spelling than the caller passed, every + // poll would read as "gone" and would disconnect a perfectly healthy connection — far + // worse than not having the check. Fault escalation still covers that case. + if (!IsPortPresent()) + { + return; + } + + _watchdog.StartPresencePolling( + IsPortPresent, + $"Serial port {_portName} is no longer present; the device appears to have been disconnected.", + _livenessCheckInterval); + } + + /// + /// Reports whether the configured port is still present on the system. + /// + private bool IsPortPresent() + { + var probe = PortPresenceProbe; + return probe != null ? probe(_portName) : IsPortEnumerated(_portName); + } + + /// + /// Default presence probe: the port is present if the framework enumerates it, or — on Unix, + /// where a port name is a device node path — if that node still exists. + /// + /// + /// Deliberately requires both answers to be "absent" before reporting absence. The + /// enumeration is the portable signal (and the only one on Windows); the filesystem check + /// covers a Unix port whose spelling the enumeration happens not to return. Errors from either + /// are treated as "no evidence", never as evidence of a drop. + /// + internal static bool IsPortEnumerated(string portName) + { + try + { + foreach (var name in SerialPort.GetPortNames()) + { + if (string.Equals(name, portName, StringComparison.OrdinalIgnoreCase)) + { + return true; + } + } + } + catch (Exception) + { + // Enumeration can fail transiently (a /dev scan racing a device change, a registry + // read on Windows). Fall through to the filesystem check rather than concluding the + // port has gone away. + } + + try + { + if (portName.StartsWith('/')) + { + return File.Exists(portName); + } + } + catch (Exception) + { + // Same reasoning: an unreadable path is not evidence the device was unplugged. + } + + return false; + } + + /// + /// Tears down a connection that was detected as dropped and reports it as a loss. + /// + /// The condition that identified the drop. + private void HandleConnectionLost(Exception error) + { + // Drop the reference before anything that can block: IsConnected must already read false + // when StatusChanged fires, so a handler (or the device registry's pruning) sees the truth. + // Closing a port whose device has physically vanished is the one call here that can stall, + // so it happens after the notification. + var port = Interlocked.Exchange(ref _serialPort, null); + + OnStatusChanged(false, error); + + try + { + port?.Dispose(); + } + catch (Exception) + { + // The device is already gone; failing to close its handle changes nothing. + } + } + /// /// Raises the StatusChanged event. /// @@ -275,6 +488,7 @@ public void Dispose() // Ignore errors during disposal } + _watchdog.Dispose(); _serialPort?.Dispose(); _disposed = true; } diff --git a/src/Daqifi.Core/Communication/Transport/TcpStreamTransport.cs b/src/Daqifi.Core/Communication/Transport/TcpStreamTransport.cs index dea4396d..55e472a7 100644 --- a/src/Daqifi.Core/Communication/Transport/TcpStreamTransport.cs +++ b/src/Daqifi.Core/Communication/Transport/TcpStreamTransport.cs @@ -8,8 +8,53 @@ namespace Daqifi.Core.Communication.Transport; /// over TCP connections. Handles connection lifecycle and provides the underlying /// NetworkStream for message producers and consumers. /// -public class TcpStreamTransport : IStreamTransport +/// +/// +/// Drop detection. describes the last completed operation +/// rather than the current link, and nothing used to raise for an +/// unexpected drop, so was unreachable here too +/// (issue #382). Two things now cover it: +/// +/// +/// +/// I/O fault escalation via . The reader loop reports +/// failed reads — including a zero-byte read, which on a socket means the peer closed — and +/// consecutive failures with +/// no successful transfer in between close the socket and raise with +/// IsConnected == false. A closed or reset connection is therefore reported within a +/// fraction of a second. +/// +/// +/// TCP keep-alive (see ), so a link that is silently +/// severed — a device losing power or dropping off WiFi, where no FIN or RST ever arrives — still +/// fails a read instead of hanging forever. This bounds detection of a silent drop at roughly +/// twenty seconds of idleness. +/// +/// +/// +/// An intentional disarms detection first, so it reports a normal +/// disconnect and never a loss. +/// +/// +public class TcpStreamTransport : IStreamTransport, ITransportHealthSink { + /// + /// Idle time, in seconds, before TCP keep-alive probing starts on an established connection. + /// + internal const int KeepAliveTimeSeconds = 10; + + /// + /// Interval, in seconds, between TCP keep-alive probes once probing has started. + /// + internal const int KeepAliveIntervalSeconds = 3; + + /// + /// Number of unanswered TCP keep-alive probes before the OS fails the connection. With + /// and this bounds + /// detection of a silently severed link at roughly twenty seconds. + /// + internal const int KeepAliveRetryCount = 3; + /// /// Socket receive timeout applied once the connection is established, in milliseconds. /// @@ -27,6 +72,7 @@ public class TcpStreamTransport : IStreamTransport private readonly IPEndPoint _endPoint; private readonly IPAddress? _localInterface; + private readonly TransportConnectionWatchdog _watchdog; private TcpClient? _tcpClient; private NetworkStream? _networkStream; private bool _disposed; @@ -52,6 +98,7 @@ public TcpStreamTransport(IPAddress ipAddress, int port, IPAddress? localInterfa { _endPoint = new IPEndPoint(ipAddress, port); _localInterface = localInterface; + _watchdog = CreateWatchdog(); } /// @@ -77,6 +124,17 @@ public TcpStreamTransport(string host, int port, IPAddress? localInterface = nul Hostname = host; } _localInterface = localInterface; + _watchdog = CreateWatchdog(); + } + + /// + /// Creates the drop detector shared by both constructors. + /// + private TransportConnectionWatchdog CreateWatchdog() + { + return new TransportConnectionWatchdog( + $"TCP transport ({Hostname ?? _endPoint.Address.ToString()}:{_endPoint.Port})", + HandleConnectionLost); } /// @@ -203,6 +261,8 @@ await ConnectRetryExecutor.ExecuteAsync( // download path uses ReadAsync, which ignores SO_RCVTIMEO and keeps its own // long timeout. See OperationalReceiveTimeoutMs. _tcpClient.ReceiveTimeout = OperationalReceiveTimeoutMs; + + EnableKeepAlive(_tcpClient); }, onAttemptFailed: () => { @@ -211,6 +271,34 @@ await ConnectRetryExecutor.ExecuteAsync( _networkStream = null; }, onStatusChanged: OnStatusChanged); + + _watchdog.Arm(); + } + + /// + /// Turns on TCP keep-alive so a silently severed link (device powered off, WiFi dropped) fails + /// a read instead of blocking forever with no FIN or RST ever arriving. + /// + /// + /// The per-socket tuning knobs are supported on Windows, macOS, and Linux, but a platform or + /// socket that rejects one must not fail the connect: keep-alive is a detection improvement, + /// not a requirement, and I/O fault escalation still covers a link that produces read errors. + /// + private static void EnableKeepAlive(TcpClient client) + { + try + { + var socket = client.Client; + socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.KeepAlive, true); + socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveTime, KeepAliveTimeSeconds); + socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveInterval, KeepAliveIntervalSeconds); + socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveRetryCount, KeepAliveRetryCount); + } + catch (Exception) + { + // Not every platform/socket accepts every knob. A connection without keep-alive tuning + // is still a working connection. + } } /// @@ -219,6 +307,10 @@ await ConnectRetryExecutor.ExecuteAsync( /// A task representing the asynchronous disconnect operation. public async Task DisconnectAsync() { + // Disarm before touching the socket: closing it makes any in-flight read fail, and a + // still-armed watchdog would report an intentional disconnect as a lost connection. + _watchdog.Disarm(); + if (!IsConnected) return; @@ -260,6 +352,48 @@ public void Disconnect() DisconnectAsync().GetAwaiter().GetResult(); } + /// + /// + /// Escalates to a lost connection only after + /// failures with no + /// successful transfer in between, so a transient read error cannot tear down a healthy socket. + /// + public void ReportIoFault(Exception error) + { + ArgumentNullException.ThrowIfNull(error); + _watchdog.RecordFault(error); + } + + /// + public void ReportIoSuccess() + { + _watchdog.RecordSuccess(); + } + + /// + /// Tears down a connection that was detected as dropped and reports it as a loss. + /// + /// The condition that identified the drop. + private void HandleConnectionLost(Exception error) + { + // Drop the references before anything that can block: IsConnected must already read false + // when StatusChanged fires, so a handler (or the device registry's pruning) sees the truth. + var stream = Interlocked.Exchange(ref _networkStream, null); + var client = Interlocked.Exchange(ref _tcpClient, null); + + OnStatusChanged(false, error); + + try + { + stream?.Dispose(); + client?.Dispose(); + } + catch (Exception) + { + // The peer is already gone; failing to close the socket changes nothing. + } + } + /// /// Raises the StatusChanged event. /// @@ -295,6 +429,7 @@ public void Dispose() // Ignore errors during disposal } + _watchdog.Dispose(); _networkStream?.Dispose(); _tcpClient?.Dispose(); _disposed = true; diff --git a/src/Daqifi.Core/Communication/Transport/TransportConnectionWatchdog.cs b/src/Daqifi.Core/Communication/Transport/TransportConnectionWatchdog.cs new file mode 100644 index 00000000..1a6b81c9 --- /dev/null +++ b/src/Daqifi.Core/Communication/Transport/TransportConnectionWatchdog.cs @@ -0,0 +1,301 @@ +namespace Daqifi.Core.Communication.Transport; + +/// +/// Shared drop-detection logic for stream transports. Decides when an established connection +/// should be declared lost and raises that decision exactly once per armed cycle. +/// +/// +/// +/// Two independent signals feed the decision, mirroring the two ways a transport learns its device +/// is gone (issue #382): +/// +/// +/// +/// I/O fault escalation. The reader/writer loop reports failures through +/// ; consecutive +/// failures with no successful transfer in between mean the link is gone rather than glitching. +/// A single successful read resets the run, so a recoverable blip never disconnects. +/// +/// +/// Presence polling. An optional periodic probe (for serial: "is this port still +/// enumerated?") bounds detection latency for an idle connection that is not producing read +/// failures because nothing is being read. consecutive misses +/// are required so a single hiccup in the enumeration cannot tear down a live connection. +/// +/// +/// +/// Both live here rather than in each transport so the idempotency — a connection is declared lost +/// once, whichever signal wins the race — is implemented in one place. +/// +/// +internal sealed class TransportConnectionWatchdog : IDisposable +{ + /// + /// Number of consecutive I/O failures, with no successful transfer in between, that mean the + /// connection is gone rather than glitching. + /// + /// + /// The reader loop backs off 100 ms after a failed read, so five failures is roughly half a + /// second of an unbrokenly failing stream — long enough that a one-off fault cannot trip it, + /// short enough that a real drop is reported effectively immediately while traffic is flowing. + /// + internal const int ConsecutiveFaultThreshold = 5; + + /// + /// Number of consecutive presence-probe misses required before the connection is declared lost. + /// + internal const int PresenceMissThreshold = 2; + + /// + /// Default cadence for presence polling. With this bounds + /// detection of a silent drop at roughly three seconds (two intervals plus up to one interval + /// of phase between the drop and the next poll). + /// + internal static readonly TimeSpan DefaultPresencePollInterval = TimeSpan.FromSeconds(1); + + private readonly Action _onConnectionLost; + private readonly string _description; + private readonly object _timerLock = new(); + + /// + /// 1 while a connection is established and a loss may still be signalled, 0 otherwise. The + /// armed-to-disarmed transition is the one-shot gate: whichever signal wins the race performs + /// it, and every later signal sees 0 and does nothing. + /// + private int _armed; + + private int _consecutiveFaults; + private int _presenceMisses; + private int _pollInFlight; + private Timer? _presenceTimer; + private Func? _presenceProbe; + private string _presenceLostMessage = string.Empty; + private bool _disposed; + + /// + /// Initializes a new watchdog. + /// + /// + /// How the transport describes itself in the exception attached to a loss, e.g. + /// "Serial transport (/dev/cu.usbmodem1101)". + /// + /// + /// Invoked at most once per armed cycle when the connection is declared lost. Invoked on the + /// reporting thread (a reader/writer loop) or on a timer thread, never with an internal lock + /// held. + /// + public TransportConnectionWatchdog(string description, Action onConnectionLost) + { + _description = description ?? throw new ArgumentNullException(nameof(description)); + _onConnectionLost = onConnectionLost ?? throw new ArgumentNullException(nameof(onConnectionLost)); + } + + /// + /// Gets a value indicating whether a connection loss can still be signalled. + /// + public bool IsArmed => Volatile.Read(ref _armed) == 1; + + /// + /// Gets a value indicating whether presence polling is currently running. + /// + public bool IsPollingPresence + { + get + { + lock (_timerLock) + { + return _presenceTimer != null; + } + } + } + + /// + /// Arms the watchdog after a successful connect, clearing any state left by a previous cycle. + /// + public void Arm() + { + Volatile.Write(ref _consecutiveFaults, 0); + Volatile.Write(ref _presenceMisses, 0); + Volatile.Write(ref _armed, 1); + } + + /// + /// Disarms the watchdog and stops presence polling. Used for an intentional disconnect, where + /// the transport reports Disconnected itself and a concurrent in-flight poll or read + /// failure must not also report a loss. + /// + public void Disarm() + { + Volatile.Write(ref _armed, 0); + StopPresencePolling(); + Volatile.Write(ref _consecutiveFaults, 0); + Volatile.Write(ref _presenceMisses, 0); + } + + /// + /// Starts polling for the continued presence of the underlying device. + /// No-ops when the watchdog is not armed or is not positive. + /// + /// Returns true while the device is still present. + /// Message for the exception attached to a presence-driven loss. + /// Polling cadence. + public void StartPresencePolling(Func probe, string lostMessage, TimeSpan interval) + { + ArgumentNullException.ThrowIfNull(probe); + + if (!IsArmed || interval <= TimeSpan.Zero) + { + return; + } + + lock (_timerLock) + { + if (_disposed) + { + return; + } + + _presenceProbe = probe; + _presenceLostMessage = lostMessage; + Volatile.Write(ref _presenceMisses, 0); + + _presenceTimer?.Dispose(); + _presenceTimer = new Timer(_ => PollPresence(), null, interval, interval); + } + } + + /// + /// Stops presence polling, leaving the armed state untouched. + /// + public void StopPresencePolling() + { + lock (_timerLock) + { + _presenceTimer?.Dispose(); + _presenceTimer = null; + } + } + + /// + /// Records a failed read or write. Declares the connection lost once + /// failures have occurred with no successful transfer + /// in between. + /// + /// The exception the failed operation raised. + public void RecordFault(Exception error) + { + if (!IsArmed) + { + return; + } + + var faults = Interlocked.Increment(ref _consecutiveFaults); + if (faults < ConsecutiveFaultThreshold) + { + return; + } + + SignalConnectionLost(new TransportNotConnectedException( + $"{_description} saw {ConsecutiveFaultThreshold} consecutive I/O failures with no " + + "successful transfer in between; the connection is treated as lost.", + error)); + } + + /// + /// Records a successful read or write, clearing the current failure run. + /// + public void RecordSuccess() + { + // Read first: the overwhelmingly common case is a healthy stream with the counter already + // at zero, and this runs once per successful read. + if (Volatile.Read(ref _consecutiveFaults) != 0) + { + Interlocked.Exchange(ref _consecutiveFaults, 0); + } + } + + /// + /// Runs one presence probe. Exposed for tests so the polling decision can be exercised without + /// waiting on a timer. + /// + internal void PollPresence() + { + // A probe that runs long (an enumeration that blocks) must not stack callbacks. + if (Interlocked.Exchange(ref _pollInFlight, 1) != 0) + { + return; + } + + try + { + if (!IsArmed) + { + return; + } + + var probe = _presenceProbe; + if (probe == null) + { + return; + } + + if (probe()) + { + Volatile.Write(ref _presenceMisses, 0); + return; + } + + if (Interlocked.Increment(ref _presenceMisses) < PresenceMissThreshold) + { + return; + } + + SignalConnectionLost(new TransportNotConnectedException(_presenceLostMessage)); + } + catch + { + // A probe that throws is a broken observation, not evidence of a drop: never let it + // disconnect a healthy transport, and never let it kill the timer thread. + } + finally + { + Volatile.Write(ref _pollInFlight, 0); + } + } + + /// + /// Declares the connection lost, at most once per armed cycle. + /// + private void SignalConnectionLost(Exception error) + { + // Disarm and signal in one step. Whichever of the two detectors gets here first owns the + // notification; the other sees 0 and returns. + if (Interlocked.Exchange(ref _armed, 0) != 1) + { + return; + } + + StopPresencePolling(); + _onConnectionLost(error); + } + + /// + /// Disposes the watchdog, stopping presence polling. + /// + public void Dispose() + { + lock (_timerLock) + { + if (_disposed) + { + return; + } + + _disposed = true; + _presenceTimer?.Dispose(); + _presenceTimer = null; + } + + Volatile.Write(ref _armed, 0); + } +} diff --git a/src/Daqifi.Core/Device/DaqifiDevice.cs b/src/Daqifi.Core/Device/DaqifiDevice.cs index 29a3bf30..0c2bfd58 100644 --- a/src/Daqifi.Core/Device/DaqifiDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiDevice.cs @@ -475,16 +475,21 @@ public void Connect() // Create message producer and consumer from transport if needed if (_transport != null) { + // The reader/writer loops are the first thing to notice a device that has + // gone away; a transport that can act on that gets told (issue #382). + var healthSink = _transport as ITransportHealthSink; + if (_messageProducer == null) { - _messageProducer = new MessageProducer(_transport.Stream); + _messageProducer = new MessageProducer(_transport.Stream, healthSink: healthSink); } if (_messageConsumer == null) { _messageConsumer = new StreamMessageConsumer( _transport.Stream, - new ProtobufMessageParser()); + new ProtobufMessageParser(), + healthSink: healthSink); } } @@ -897,7 +902,8 @@ private async Task> ExecuteTextCommandCoreAsync( // Create a temporary text consumer on the same stream using var textConsumer = new StreamMessageConsumer( _transport.Stream, - new LineBasedMessageParser()); + new LineBasedMessageParser(), + healthSink: _transport as ITransportHealthSink); textConsumer.MessageReceived += (_, e) => { diff --git a/src/Daqifi.Core/Device/DaqifiDeviceRegistry.cs b/src/Daqifi.Core/Device/DaqifiDeviceRegistry.cs index 7d1718b4..e8a3f98f 100644 --- a/src/Daqifi.Core/Device/DaqifiDeviceRegistry.cs +++ b/src/Daqifi.Core/Device/DaqifiDeviceRegistry.cs @@ -31,12 +31,12 @@ namespace Daqifi.Core.Device; /// /// Liveness. Registrations whose device stops reporting /// are pruned before every registration attempt (and by on demand). -/// Pruning is only as good as the transport's drop detection, which is a real limit today: the -/// serial transport reports the OS handle's state, and that stays open after a USB device is -/// physically unplugged, so such a device keeps reporting IsConnected and is not pruned. -/// Devices you disconnect yourself, and drops a transport does report, are pruned as described. -/// The registry does not subscribe to device status events or reconnect on its own; automatic -/// reconnect is issue #379. +/// Pruning is only as good as the transport's drop detection, and that now covers a physically +/// unplugged USB device: polls for its +/// port's continued presence and escalates persistently failing I/O, so an unplugged device stops +/// reporting IsConnected within roughly three seconds and is pruned like any other drop +/// (issue #382). The registry does not subscribe to device status events or reconnect on its own; +/// automatic reconnect is issue #379. /// /// /// Concurrency. All members are safe to call from any thread. Reads snapshot the live set, From 5b0e51614eb4caa29839fc97e72ae9db1e0825c0 Mon Sep 17 00:00:00 2001 From: Tyler Kron Date: Wed, 29 Jul 2026 09:02:47 -0600 Subject: [PATCH 2/4] refactor(transport): tighten drop detection after self-review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - A write TIMEOUT is no longer reported as an I/O fault. It means the device is not draining its receive buffer right now (busy or flow-controlled), not that the link is gone — escalating it could disconnect a healthy device. Mirrors how the reader loop already treats a read timeout as benign. - Public XML docs no longer cref members of the internal TransportConnectionWatchdog; the thresholds are stated in prose so the generated documentation resolves for consumers. - docs: document that Lost is raised on an internal reader/timer thread, with the pattern for tearing the device down off the callback. Co-Authored-By: Claude Opus 5 --- docs/DEVICE_INTERFACES.md | 21 +++++++++--- .../MessageProducerHealthReportingTests.cs | 27 ++++++++++++++++ .../Producers/MessageProducer.cs | 12 ++++++- .../Transport/ITransportHealthSink.cs | 5 +++ .../Transport/SerialStreamTransport.cs | 32 +++++++++---------- .../Transport/TcpStreamTransport.cs | 21 ++++++------ 6 files changed, 85 insertions(+), 33 deletions(-) diff --git a/docs/DEVICE_INTERFACES.md b/docs/DEVICE_INTERFACES.md index fc87b0e4..e19eb67e 100644 --- a/docs/DEVICE_INTERFACES.md +++ b/docs/DEVICE_INTERFACES.md @@ -419,13 +419,26 @@ that probe at connect time, so an exotic port-name spelling disables the check r producing a false drop. A single failed read never disconnects anything. Escalation requires a *run* of failures with no -successful transfer in between, so a stream that glitches and recovers stays connected. To disable -the serial presence poll and rely on I/O escalation alone, construct the transport with -`livenessCheckInterval: TimeSpan.Zero`. +successful transfer in between, so a stream that glitches and recovers stays connected. A read or +write *timeout* is not a failure at all — it is what an idle or momentarily busy device looks +like. To disable the serial presence poll and rely on I/O escalation alone, construct the transport +with `livenessCheckInterval: TimeSpan.Zero`. When a drop is detected the transport closes its handle, so `IsConnected` already reads `false` by the time `StatusChanged` fires. The device's message consumer and producer are *not* stopped -automatically — call `Disconnect()` (or `Dispose()`) from your `Lost` handler to release them. +automatically — call `Disconnect()` (or `Dispose()`) once you have handled `Lost` to release them. + +`Lost` is raised on an internal thread — the reader loop or the liveness timer, whichever detected +the drop — so treat the handler like any background callback: do the minimum, and push teardown or +reconnection onto your own thread rather than blocking inside it. + +```csharp +device.StatusChanged += (_, e) => +{ + if (e.Status != ConnectionStatus.Lost) return; + _ = Task.Run(() => device.Disconnect()); // don't tear down inside the callback +}; +``` Custom transports can opt into the same escalation by implementing `ITransportHealthSink`; the reader and writer loops report every read/write outcome to a transport that does. diff --git a/src/Daqifi.Core.Tests/Communication/Producers/MessageProducerHealthReportingTests.cs b/src/Daqifi.Core.Tests/Communication/Producers/MessageProducerHealthReportingTests.cs index 49d5aa73..20f9fa62 100644 --- a/src/Daqifi.Core.Tests/Communication/Producers/MessageProducerHealthReportingTests.cs +++ b/src/Daqifi.Core.Tests/Communication/Producers/MessageProducerHealthReportingTests.cs @@ -47,6 +47,27 @@ public void WhenWritesSucceed_TheProducerReportsSuccess() Assert.Equal(0, sink.FaultCount); } + [Fact] + public void WhenWritesTimeOut_NoFaultIsReported() + { + // A write timeout means the device is not draining its buffer right now, not that the + // link is gone. Escalating it would disconnect a healthy but momentarily busy device. + using var stream = new FailingWriteStream { TimeoutWrites = true }; + var sink = new CountingHealthSink(); + using var producer = new MessageProducer(stream, healthSink: sink); + + producer.Start(); + for (var i = 0; i < 5; i++) + { + producer.Send(new ScpiMessage($"CMD{i}")); + } + + Assert.True(WaitUntil(() => producer.QueuedMessageCount == 0, TimeSpan.FromSeconds(5))); + Thread.Sleep(100); + Assert.Equal(0, sink.FaultCount); + Assert.Equal(0, sink.SuccessCount); + } + [Fact] public void WithNoHealthSink_TheProducerBehavesExactlyAsBefore() { @@ -91,9 +112,15 @@ private sealed class CountingHealthSink : ITransportHealthSink private sealed class FailingWriteStream : MemoryStream { public volatile bool FailWrites; + public volatile bool TimeoutWrites; public override void Write(byte[] buffer, int offset, int count) { + if (TimeoutWrites) + { + throw new TimeoutException("the device is not draining its buffer"); + } + if (FailWrites) { throw new IOException("the device is gone"); diff --git a/src/Daqifi.Core/Communication/Producers/MessageProducer.cs b/src/Daqifi.Core/Communication/Producers/MessageProducer.cs index c1358fa9..75115ba2 100644 --- a/src/Daqifi.Core/Communication/Producers/MessageProducer.cs +++ b/src/Daqifi.Core/Communication/Producers/MessageProducer.cs @@ -182,7 +182,17 @@ private void ProcessMessages() // bad write doesn't stall the remaining messages. Tell the transport // too — it is the only component that can decide a run of failures // means the device is gone rather than glitching. - _healthSink?.ReportIoFault(ex); + // + // A write TIMEOUT is deliberately excluded: it means the device is not + // draining its receive buffer right now (busy, or flow-controlled), not + // that the link is gone. Treating it as evidence of a disconnect could + // tear down a healthy connection to a momentarily busy device — the + // same reason the reader loop treats a read timeout as benign. + if (ex is not TimeoutException) + { + _healthSink?.ReportIoFault(ex); + } + SafeLog(() => _logger.LogWarning(ex, "Failed to write message to the stream; continuing with remaining queued messages.")); } } diff --git a/src/Daqifi.Core/Communication/Transport/ITransportHealthSink.cs b/src/Daqifi.Core/Communication/Transport/ITransportHealthSink.cs index b0c22343..3aef28a0 100644 --- a/src/Daqifi.Core/Communication/Transport/ITransportHealthSink.cs +++ b/src/Daqifi.Core/Communication/Transport/ITransportHealthSink.cs @@ -24,6 +24,11 @@ namespace Daqifi.Core.Communication.Transport; /// expected to escalate only on a run of consecutive failures with no successful transfer in /// between, so a recoverable blip never tears down a healthy connection. /// +/// +/// Callers must not report an operation that merely hit its configured timeout: that is what an +/// idle device (or one that is momentarily not draining its receive buffer) looks like, and +/// treating it as a failure would disconnect a healthy connection. +/// /// public interface ITransportHealthSink { diff --git a/src/Daqifi.Core/Communication/Transport/SerialStreamTransport.cs b/src/Daqifi.Core/Communication/Transport/SerialStreamTransport.cs index ce50d5cf..6eef1db9 100644 --- a/src/Daqifi.Core/Communication/Transport/SerialStreamTransport.cs +++ b/src/Daqifi.Core/Communication/Transport/SerialStreamTransport.cs @@ -17,20 +17,20 @@ namespace Daqifi.Core.Communication.Transport; /// /// /// -/// Port-presence polling at (1 s), requiring -/// consecutive misses. This bounds -/// detection of an unplug at roughly three seconds even on a completely idle connection. -/// Presence is containing the port, falling back to the -/// device node still existing on Unix — no WMI, so it behaves identically on Windows, macOS, and -/// Linux. It is armed only if the port is visible to that probe at connect time, so a platform or -/// port-name spelling the probe cannot see disables the check rather than reporting a false drop. +/// Port-presence polling at (1 s), requiring two +/// consecutive misses. This bounds detection of an unplug at roughly three seconds even on +/// a completely idle connection. Presence is containing the +/// port, falling back to the device node still existing on Unix — no WMI, so it behaves +/// identically on Windows, macOS, and Linux. It is armed only if the port is visible to that probe +/// at connect time, so a platform or port-name spelling the probe cannot see disables the check +/// rather than reporting a false drop. /// /// /// I/O fault escalation via : the reader/writer loops -/// report failures, and -/// consecutive failures with no successful transfer in between are treated as a drop. Under -/// traffic this reports within a few hundred milliseconds; a single failed read that then recovers -/// never disconnects. +/// report failures, and five consecutive failures with no successful transfer in between are +/// treated as a drop. Under traffic this reports within a few hundred milliseconds; a single +/// failed read that then recovers never disconnects, and a read or write that merely hits its +/// timeout is not a failure at all. /// /// /// @@ -42,9 +42,8 @@ public class SerialStreamTransport : IStreamTransport, ITransportHealthSink { /// /// Default cadence at which an established connection is checked for the continued presence of - /// its serial port. Combined with - /// consecutive misses, an - /// unplugged device is reported within roughly three seconds. + /// its serial port. Combined with the two consecutive misses required to act, an unplugged + /// device is reported within roughly three seconds. /// public static readonly TimeSpan DefaultLivenessCheckInterval = TransportConnectionWatchdog.DefaultPresencePollInterval; @@ -317,9 +316,8 @@ public static string[] GetAvailablePortNames() /// /// - /// Escalates to a lost connection only after - /// failures with no - /// successful transfer in between, so a transient read error cannot tear down a healthy port. + /// Escalates to a lost connection only after five failures with no successful transfer in + /// between, so a transient read error cannot tear down a healthy port. /// public void ReportIoFault(Exception error) { diff --git a/src/Daqifi.Core/Communication/Transport/TcpStreamTransport.cs b/src/Daqifi.Core/Communication/Transport/TcpStreamTransport.cs index 55e472a7..8a5be806 100644 --- a/src/Daqifi.Core/Communication/Transport/TcpStreamTransport.cs +++ b/src/Daqifi.Core/Communication/Transport/TcpStreamTransport.cs @@ -18,16 +18,16 @@ namespace Daqifi.Core.Communication.Transport; /// /// /// I/O fault escalation via . The reader loop reports -/// failed reads — including a zero-byte read, which on a socket means the peer closed — and -/// consecutive failures with -/// no successful transfer in between close the socket and raise with -/// IsConnected == false. A closed or reset connection is therefore reported within a -/// fraction of a second. +/// failed reads — including a zero-byte read, which on a socket means the peer closed — and five +/// consecutive failures with no successful transfer in between close the socket and raise +/// with IsConnected == false. A closed or reset connection is +/// therefore reported within a fraction of a second. A read that merely hits its timeout is not a +/// failure and is never reported. /// /// -/// TCP keep-alive (see ), so a link that is silently -/// severed — a device losing power or dropping off WiFi, where no FIN or RST ever arrives — still -/// fails a read instead of hanging forever. This bounds detection of a silent drop at roughly +/// TCP keep-alive (10 s idle, 3 s probes, 3 retries), so a link that is silently severed — +/// a device losing power or dropping off WiFi, where no FIN or RST ever arrives — still fails a +/// read instead of hanging forever. This bounds detection of a silent drop at roughly /// twenty seconds of idleness. /// /// @@ -354,9 +354,8 @@ public void Disconnect() /// /// - /// Escalates to a lost connection only after - /// failures with no - /// successful transfer in between, so a transient read error cannot tear down a healthy socket. + /// Escalates to a lost connection only after five failures with no successful transfer in + /// between, so a transient read error cannot tear down a healthy socket. /// public void ReportIoFault(Exception error) { From d5f2c6b391dbf0a1ce8c8d587e6864c5ccec7965 Mon Sep 17 00:00:00 2001 From: Tyler Kron Date: Wed, 29 Jul 2026 09:29:41 -0600 Subject: [PATCH 3/4] fix(transport): don't let a failed presence probe look like an absent port MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Qodo review: IsPortEnumerated swallowed exceptions from GetPortNames() and the filesystem check and returned false, so a failure to OBSERVE was indistinguishable from the port actually being gone. Two transient enumeration failures would count as two consecutive misses and close a healthy connection — the watchdog's "a throwing probe is no evidence" guard could never engage for the default probe. A probe that cannot answer now throws: - The enumeration failure is swallowed only when the filesystem check can still answer for that port name (a Unix device-node path); otherwise it propagates. - StartDropDetection catches a throwing baseline probe and leaves presence polling off rather than guessing, and never lets it escape a successful connect. - The unplug harness's port watcher treats a throwing probe as "keep watching". Two regression tests: a probe that starts throwing after arming reports no drop and stays armed, and a probe that throws at connect time leaves the check unarmed while fault escalation still covers the port. Co-Authored-By: Claude Opus 5 --- ...SerialStreamTransportDropDetectionTests.cs | 77 +++++++++++++++++++ .../Transport/SerialUnplugValidationTests.cs | 14 +++- .../Transport/SerialStreamTransport.cs | 56 +++++++++----- .../Transport/TransportConnectionWatchdog.cs | 6 +- 4 files changed, 128 insertions(+), 25 deletions(-) diff --git a/src/Daqifi.Core.Tests/Communication/Transport/SerialStreamTransportDropDetectionTests.cs b/src/Daqifi.Core.Tests/Communication/Transport/SerialStreamTransportDropDetectionTests.cs index d500d1ea..612d6432 100644 --- a/src/Daqifi.Core.Tests/Communication/Transport/SerialStreamTransportDropDetectionTests.cs +++ b/src/Daqifi.Core.Tests/Communication/Transport/SerialStreamTransportDropDetectionTests.cs @@ -108,6 +108,83 @@ public void WhenTheProbeCannotSeeThePortAtConnectTime_TheCheckIsNotArmed() Assert.Equal(0, Volatile.Read(ref drops)); } + [Fact] + public void WhenThePresenceProbeStartsThrowing_NoDropIsReported() + { + // A probe that cannot answer is a failure to observe, not evidence the port went away. + // Before this was fixed the default probe swallowed its own exceptions and returned + // "absent", so two transient enumeration failures looked like two consecutive misses and + // closed a healthy connection. + var firstCall = true; + using var transport = new SerialStreamTransport("/dev/ttyTest382", livenessCheckInterval: TimeSpan.FromHours(1)) + { + PortPresenceProbe = _ => + { + if (firstCall) + { + firstCall = false; + return true; + } + + throw new UnauthorizedAccessException("the probe could not run"); + } + }; + + var drops = 0; + transport.StatusChanged += (_, e) => + { + if (!e.IsConnected) + { + Interlocked.Increment(ref drops); + } + }; + + transport.StartDropDetection(); + Assert.True(transport.IsLivenessMonitorActive); + + for (var i = 0; i < TransportConnectionWatchdog.PresenceMissThreshold * 5; i++) + { + transport.PollLivenessForTesting(); + } + + Assert.Equal(0, Volatile.Read(ref drops)); + Assert.True(transport.IsLivenessMonitorActive); + } + + [Fact] + public void WhenThePresenceProbeThrowsAtConnectTime_TheCheckIsNotArmedAndConnectStillSucceeds() + { + // No baseline observation means nothing to compare later polls against, so the check stays + // off — and the failure must not escape into the caller's successful connect. + using var transport = new SerialStreamTransport("/dev/ttyTest382", livenessCheckInterval: FastInterval) + { + PortPresenceProbe = _ => throw new UnauthorizedAccessException("the probe could not run") + }; + + var drops = 0; + transport.StatusChanged += (_, e) => + { + if (!e.IsConnected) + { + Interlocked.Increment(ref drops); + } + }; + + transport.StartDropDetection(); + + Assert.False(transport.IsLivenessMonitorActive); + Thread.Sleep(400); + Assert.Equal(0, Volatile.Read(ref drops)); + + // Fault escalation still covers the port. + for (var i = 0; i < TransportConnectionWatchdog.ConsecutiveFaultThreshold; i++) + { + transport.ReportIoFault(new IOException("device gone")); + } + + Assert.Equal(1, Volatile.Read(ref drops)); + } + [Fact] public void WhenTheLivenessIntervalIsZero_TheCheckIsDisabled() { diff --git a/src/Daqifi.Core.Tests/Communication/Transport/SerialUnplugValidationTests.cs b/src/Daqifi.Core.Tests/Communication/Transport/SerialUnplugValidationTests.cs index 4c682252..ceee117b 100644 --- a/src/Daqifi.Core.Tests/Communication/Transport/SerialUnplugValidationTests.cs +++ b/src/Daqifi.Core.Tests/Communication/Transport/SerialUnplugValidationTests.cs @@ -260,10 +260,18 @@ public PortDisappearanceWatcher(string portName) { while (!_cts.IsCancellationRequested) { - if (!SerialStreamTransport.IsPortEnumerated(portName)) + try { - _portGoneAtUtc = DateTime.UtcNow; - return; + if (!SerialStreamTransport.IsPortEnumerated(portName)) + { + _portGoneAtUtc = DateTime.UtcNow; + return; + } + } + catch (Exception) + { + // The probe could not answer this time; that is not the port going away. + // Keep watching rather than recording a bogus disappearance time. } Thread.Sleep(20); diff --git a/src/Daqifi.Core/Communication/Transport/SerialStreamTransport.cs b/src/Daqifi.Core/Communication/Transport/SerialStreamTransport.cs index 6eef1db9..ac077d10 100644 --- a/src/Daqifi.Core/Communication/Transport/SerialStreamTransport.cs +++ b/src/Daqifi.Core/Communication/Transport/SerialStreamTransport.cs @@ -363,7 +363,18 @@ internal void StartDropDetection() // If a platform enumerates it under a different spelling than the caller passed, every // poll would read as "gone" and would disconnect a perfectly healthy connection — far // worse than not having the check. Fault escalation still covers that case. - if (!IsPortPresent()) + // + // A probe that throws here is treated the same way: without a baseline observation there + // is nothing to compare later polls against, so the check stays off rather than guessing. + // The throw must not escape either — the port is open and the connect succeeded. + try + { + if (!IsPortPresent()) + { + return; + } + } + catch (Exception) { return; } @@ -388,13 +399,26 @@ private bool IsPortPresent() /// where a port name is a device node path — if that node still exists. /// /// + /// /// Deliberately requires both answers to be "absent" before reporting absence. The /// enumeration is the portable signal (and the only one on Windows); the filesystem check - /// covers a Unix port whose spelling the enumeration happens not to return. Errors from either - /// are treated as "no evidence", never as evidence of a drop. + /// covers a Unix port whose spelling the enumeration happens not to return. + /// + /// + /// A failure to observe is not absence, and this method must never conflate the two: a + /// probe that cannot answer throws, and the caller treats that as no evidence of a drop. + /// Returning false when the enumeration merely failed would let two transient failures + /// look like two consecutive misses and close a healthy connection. A failed enumeration is + /// therefore only swallowed when the filesystem check can still answer for this port name. + /// /// + /// + /// Propagates whatever the underlying probe raised when no source could answer. + /// internal static bool IsPortEnumerated(string portName) { + var isDeviceNodePath = portName.StartsWith('/'); + try { foreach (var name in SerialPort.GetPortNames()) @@ -405,26 +429,18 @@ internal static bool IsPortEnumerated(string portName) } } } - catch (Exception) - { - // Enumeration can fail transiently (a /dev scan racing a device change, a registry - // read on Windows). Fall through to the filesystem check rather than concluding the - // port has gone away. - } - - try - { - if (portName.StartsWith('/')) - { - return File.Exists(portName); - } - } - catch (Exception) + catch (Exception) when (isDeviceNodePath) { - // Same reasoning: an unreadable path is not evidence the device was unplugged. + // Enumeration can fail transiently (a /dev scan racing a device change). A Unix port + // name is also a filesystem path, so an independent answer is still available — the + // failure is only swallowed because that second source can actually answer. When it + // cannot (a Windows-style name), the exception propagates as "no observation". + return File.Exists(portName); } - return false; + // The enumeration answered and did not list this port. On Unix it may simply not enumerate + // the spelling the caller opened, so the device node is the tie-breaker. + return isDeviceNodePath && File.Exists(portName); } /// diff --git a/src/Daqifi.Core/Communication/Transport/TransportConnectionWatchdog.cs b/src/Daqifi.Core/Communication/Transport/TransportConnectionWatchdog.cs index 1a6b81c9..3b7a8d14 100644 --- a/src/Daqifi.Core/Communication/Transport/TransportConnectionWatchdog.cs +++ b/src/Daqifi.Core/Communication/Transport/TransportConnectionWatchdog.cs @@ -254,8 +254,10 @@ internal void PollPresence() } catch { - // A probe that throws is a broken observation, not evidence of a drop: never let it - // disconnect a healthy transport, and never let it kill the timer thread. + // A probe that throws is a failure to observe, not evidence of a drop: never let it + // disconnect a healthy transport, and never let it kill the timer thread. Probes are + // required to surface "could not answer" this way rather than returning false, which + // would be indistinguishable from the device having actually gone. } finally { From 5dd7d7d20f0ed99baf6858b5ec1508cb3556ed4c Mon Sep 17 00:00:00 2001 From: Tyler Kron Date: Wed, 29 Jul 2026 09:47:25 -0600 Subject: [PATCH 4/4] fix(transport): a probe exception must break the presence-miss run MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Qodo review, two findings. 1. PollPresence swallowed a probe exception but kept the miss run, so "absent, could-not-observe, absent" satisfied a two-miss threshold that never actually saw the port absent twice in a row. The threshold is about CONSECUTIVE observed absences, so the exception now clears the run rather than merely failing to extend it. Regression test drives that exact sequence. 2. The unplug harness's PortDisappearanceWatcher published its timestamp across threads unsynchronized, so the reader could see a stale value and print a wrong detection latency — undermining the one number that harness exists to produce. Now a long ticks field published with Interlocked.CompareExchange, which also makes "record the first absence, once" explicit. Co-Authored-By: Claude Opus 5 --- .../Transport/SerialUnplugValidationTests.cs | 24 ++++++++++++++--- .../TransportConnectionWatchdogTests.cs | 27 +++++++++++++++++++ .../Transport/TransportConnectionWatchdog.cs | 6 +++++ 3 files changed, 54 insertions(+), 3 deletions(-) diff --git a/src/Daqifi.Core.Tests/Communication/Transport/SerialUnplugValidationTests.cs b/src/Daqifi.Core.Tests/Communication/Transport/SerialUnplugValidationTests.cs index ceee117b..6bf70c92 100644 --- a/src/Daqifi.Core.Tests/Communication/Transport/SerialUnplugValidationTests.cs +++ b/src/Daqifi.Core.Tests/Communication/Transport/SerialUnplugValidationTests.cs @@ -250,9 +250,19 @@ private void Log(string message) /// private sealed class PortDisappearanceWatcher : IDisposable { + private const long NotObserved = -1; + private readonly CancellationTokenSource _cts = new(); private readonly Thread _thread; - private DateTime? _portGoneAtUtc; + + /// + /// UTC ticks of the first observed absence, or . Held as a + /// and published with so the reading thread + /// cannot observe a stale value — the whole point of this watcher is a trustworthy latency + /// number, and an unsynchronized DateTime? would undermine exactly that. The + /// compare-exchange also makes "record the FIRST absence, once" explicit. + /// + private long _portGoneAtTicks = NotObserved; public PortDisappearanceWatcher(string portName) { @@ -264,7 +274,8 @@ public PortDisappearanceWatcher(string portName) { if (!SerialStreamTransport.IsPortEnumerated(portName)) { - _portGoneAtUtc = DateTime.UtcNow; + Interlocked.CompareExchange( + ref _portGoneAtTicks, DateTime.UtcNow.Ticks, NotObserved); return; } } @@ -285,7 +296,14 @@ public PortDisappearanceWatcher(string portName) _thread.Start(); } - public DateTime? PortGoneAtUtc => _portGoneAtUtc; + public DateTime? PortGoneAtUtc + { + get + { + var ticks = Interlocked.Read(ref _portGoneAtTicks); + return ticks == NotObserved ? null : new DateTime(ticks, DateTimeKind.Utc); + } + } public void Dispose() { diff --git a/src/Daqifi.Core.Tests/Communication/Transport/TransportConnectionWatchdogTests.cs b/src/Daqifi.Core.Tests/Communication/Transport/TransportConnectionWatchdogTests.cs index e1a77480..4477fbfa 100644 --- a/src/Daqifi.Core.Tests/Communication/Transport/TransportConnectionWatchdogTests.cs +++ b/src/Daqifi.Core.Tests/Communication/Transport/TransportConnectionWatchdogTests.cs @@ -226,6 +226,33 @@ public void PollPresence_WhenProbeThrows_DoesNotSignalLoss() Assert.True(watchdog.IsArmed); } + [Fact] + public void PollPresence_WhenAProbeExceptionInterruptsAMissRun_DoesNotSignalLoss() + { + // The threshold is about CONSECUTIVE observed absences. "absent, could-not-observe, + // absent" never saw the port absent twice in a row, so it must not satisfy a two-miss + // threshold — the exception has to break the run, not merely fail to extend it. + var losses = new List(); + using var watchdog = CreateWatchdog(losses); + watchdog.Arm(); + + var step = 0; + watchdog.StartPresencePolling( + () => (step++ % 2 == 0) + ? false + : throw new UnauthorizedAccessException("the probe could not run"), + "port gone", + TimeSpan.FromHours(1)); + + for (var i = 0; i < TransportConnectionWatchdog.PresenceMissThreshold * 10; i++) + { + watchdog.PollPresence(); + } + + Assert.Empty(losses); + Assert.True(watchdog.IsArmed); + } + [Fact] public void StartPresencePolling_WhenNotArmed_DoesNotStart() { diff --git a/src/Daqifi.Core/Communication/Transport/TransportConnectionWatchdog.cs b/src/Daqifi.Core/Communication/Transport/TransportConnectionWatchdog.cs index 3b7a8d14..7332bd2d 100644 --- a/src/Daqifi.Core/Communication/Transport/TransportConnectionWatchdog.cs +++ b/src/Daqifi.Core/Communication/Transport/TransportConnectionWatchdog.cs @@ -258,6 +258,12 @@ internal void PollPresence() // disconnect a healthy transport, and never let it kill the timer thread. Probes are // required to surface "could not answer" this way rather than returning false, which // would be indistinguishable from the device having actually gone. + // + // Clearing the run matters as much as not incrementing it. The threshold is about + // CONSECUTIVE observed absences; letting a miss survive an exception would make + // "absent, could-not-observe, absent" satisfy a two-miss threshold that never actually + // saw the port absent twice in a row. + Volatile.Write(ref _presenceMisses, 0); } finally {