diff --git a/docs/DEVICE_INTERFACES.md b/docs/DEVICE_INTERFACES.md index 937488b..e19eb67 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,55 @@ 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. 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()`) 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. + ### 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 0000000..51b1039 --- /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 0000000..20f9fa6 --- /dev/null +++ b/src/Daqifi.Core.Tests/Communication/Producers/MessageProducerHealthReportingTests.cs @@ -0,0 +1,132 @@ +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 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() + { + 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 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"); + } + + 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 0000000..612d643 --- /dev/null +++ b/src/Daqifi.Core.Tests/Communication/Transport/SerialStreamTransportDropDetectionTests.cs @@ -0,0 +1,347 @@ +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 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() + { + 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 0000000..6bf70c9 --- /dev/null +++ b/src/Daqifi.Core.Tests/Communication/Transport/SerialUnplugValidationTests.cs @@ -0,0 +1,315 @@ +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 const long NotObserved = -1; + + private readonly CancellationTokenSource _cts = new(); + private readonly Thread _thread; + + /// + /// 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) + { + _thread = new Thread(() => + { + while (!_cts.IsCancellationRequested) + { + try + { + if (!SerialStreamTransport.IsPortEnumerated(portName)) + { + Interlocked.CompareExchange( + ref _portGoneAtTicks, DateTime.UtcNow.Ticks, NotObserved); + 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); + } + }) + { + IsBackground = true, + Name = "PortDisappearanceWatcher" + }; + + _thread.Start(); + } + + public DateTime? PortGoneAtUtc + { + get + { + var ticks = Interlocked.Read(ref _portGoneAtTicks); + return ticks == NotObserved ? null : new DateTime(ticks, DateTimeKind.Utc); + } + } + + 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 0000000..3c2bc81 --- /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 0000000..4477fbf --- /dev/null +++ b/src/Daqifi.Core.Tests/Communication/Transport/TransportConnectionWatchdogTests.cs @@ -0,0 +1,330 @@ +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 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() + { + 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 686f2d2..4539b33 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 44c0955..75115ba 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,28 @@ 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. + // + // 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 new file mode 100644 index 0000000..3aef28a --- /dev/null +++ b/src/Daqifi.Core/Communication/Transport/ITransportHealthSink.cs @@ -0,0 +1,47 @@ +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. +/// +/// +/// 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 +{ + /// + /// 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 bdb868b..ac077d1 100644 --- a/src/Daqifi.Core/Communication/Transport/SerialStreamTransport.cs +++ b/src/Daqifi.Core/Communication/Transport/SerialStreamTransport.cs @@ -7,8 +7,47 @@ 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 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 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. +/// +/// +/// +/// 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 the two consecutive misses required to act, 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 +55,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 +76,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 +95,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 +252,8 @@ await ConnectRetryExecutor.ExecuteAsync( _serialPort = null; }, onStatusChanged: OnStatusChanged); + + StartDropDetection(); } /// @@ -193,6 +262,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 +314,159 @@ public static string[] GetAvailablePortNames() return SerialPort.GetPortNames(); } + /// + /// + /// 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) + { + 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. + // + // 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; + } + + _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. + /// + /// + /// 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()) + { + if (string.Equals(name, portName, StringComparison.OrdinalIgnoreCase)) + { + return true; + } + } + } + catch (Exception) when (isDeviceNodePath) + { + // 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); + } + + // 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); + } + + /// + /// 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 +502,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 dea4396..8a5be80 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 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 (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. +/// +/// +/// +/// 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,47 @@ public void Disconnect() DisconnectAsync().GetAwaiter().GetResult(); } + /// + /// + /// 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) + { + 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 +428,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 0000000..7332bd2 --- /dev/null +++ b/src/Daqifi.Core/Communication/Transport/TransportConnectionWatchdog.cs @@ -0,0 +1,309 @@ +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 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. + // + // 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 + { + 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 70a954a..72e583f 100644 --- a/src/Daqifi.Core/Device/DaqifiDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiDevice.cs @@ -476,16 +476,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); } } @@ -947,7 +952,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 7d1718b..e8a3f98 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,