diff --git a/docs/DEVICE_INTERFACES.md b/docs/DEVICE_INTERFACES.md index bb719fb..5582f04 100644 --- a/docs/DEVICE_INTERFACES.md +++ b/docs/DEVICE_INTERFACES.md @@ -523,12 +523,21 @@ device.StatusChanged += (sender, args) => Console.WriteLine("Device connected successfully"); break; case ConnectionStatus.Lost: - Console.WriteLine("Connection lost - may attempt reconnection"); + Console.WriteLine("Connection lost"); + break; + case ConnectionStatus.Retrying: + Console.WriteLine("Reconnecting..."); // only with reconnect enabled + break; + case ConnectionStatus.Failed: + Console.WriteLine("Reconnection gave up"); break; } }; ``` +See [Reconnecting automatically after a drop](#reconnecting-automatically-after-a-drop) for the +`Retrying` and `Failed` states, which a device only ever reports once reconnection is turned on. + ### Detecting a dropped connection `ConnectionStatus.Lost` means the connection ended without anyone asking it to — the USB cable was @@ -582,6 +591,102 @@ The failures that feed this escalation are also reported individually, as they h `IDevice.ErrorOccurred` — see [Error Surface](#error-surface). That event is diagnostics only; `ConnectionStatus.Lost` remains the single signal that means "the connection is over". +### Reconnecting automatically after a drop + +By default a drop is where the story ends: `Lost` is reported and nothing else happens. Set +`ReconnectOptions` and the device will rebuild the session by itself — reconnect the transport, +re-initialize, put the channel configuration back, and restart a stream that was interrupted — with +no code from you in the loop. + +```csharp +device.ReconnectOptions = ReconnectOptions.Default; // 5 attempts, 1 s backing off to 30 s + +device.Reconnected += (_, e) => + Console.WriteLine($"back after {e.Outage.TotalSeconds:0.#}s (attempt {e.AttemptNumber})"); + +device.ReconnectFailed += (_, e) => + Console.WriteLine($"gave up after {e.AttemptsMade} attempts: {e.LastError?.Message}"); +``` + +`ReconnectOptions.Fast` and `ReconnectOptions.Resilient` are ready-made policies for links that +blip briefly and for unattended long runs respectively; build your own for anything else. +`ReconnectOptions.Disabled` (the default) says so explicitly. + +**What you can watch.** `ReconnectAttempt` fires before each attempt with its number and the wait +that precedes it; `Reconnected` fires once the session is fully back; `ReconnectFailed` fires when +it stops without one. The status follows along, so a UI can show progress without subscribing to +anything new: + +| Status | Meaning | +|---|---| +| `Lost` | The drop was detected. Also where a cancelled reconnect leaves the device. | +| `Retrying` | Waiting out the backoff before the next attempt. | +| `Connected` | The session is back, configuration and stream included. | +| `Failed` | Every attempt failed. Terminal — nothing more will be tried. | + +Running out of attempts is deliberately hard to miss: as well as `ReconnectFailed` and the `Failed` +status, it is logged as an error and raised on `ErrorOccurred` with source +`DeviceErrorSource.Reconnect`, carrying a `DeviceReconnectFailedException` whose inner exception is +whatever ended the final attempt. + +**What gets restored.** Only what the library itself owns: + +- the set of enabled channels (analog and digital), +- the streaming frequency, and +- an active stream, unless `ResumeStreaming` is turned off. + +It does not matter whether you established that state through the typed API +(`EnableChannels`, `StartStreaming`) or by sending the SCPI yourself with +`Send(ScpiMessageProducer.StartStreaming(...))` — the device recognizes its own streaming and +ADC-enable commands whichever way they were sent, so a session driven entirely by raw commands is +restored just the same. The one exception is the global DIO enable: it is a single switch for the +whole port rather than a per-channel mask, so sending it directly tells the device nothing about +*which* digital channels you wanted. Use `EnableChannels` for those. + +**What does not.** Everything else is the device's own state, and Core does not presume to know +what it should be after an outage of unknown length: + +- DIO directions and output levels, PWM enable/duty/frequency, analog outputs, and calibration + written only to device RAM; +- an SD card logging session — the device keeps logging or does not, entirely on its own; +- **any operation that was in flight.** An SD card download interrupted by a drop fails, and is + neither resumed nor retried; run it again once `Reconnected` says the device is back. + +A resumed stream is a genuinely new session: timestamp reconstruction re-anchors and the gap +detector resets, because the device's tick counter may well have restarted while it was away. +`Reconnected.Outage` is the measure of the interruption, not a `GapDetected` event. + +**Same endpoint only.** Reconnection re-opens the endpoint the device was already using. It cannot +follow a device that moved: a serial device that comes back on a different port path, or one whose +IP address changed after a reboot, is a new endpoint and needs a fresh `DaqifiDeviceFactory` +connect. Failing over from one transport to another (USB to WiFi, say) is out of scope. + +**Stopping it.** `CancelReconnect()` stops the loop at its next checkpoint and leaves the device on +`Lost`; `Disconnect()` and `Dispose()` do the same and then tear down. A caller-issued `Connect()` +or `Disconnect()` always wins — the loop unwinds without touching the session the caller +established, and a `Disconnect()` issued from inside a `Lost` handler stops the reconnect before it +even starts. + +Which is the other half of the rule: with reconnect enabled, **stop tearing down on `Lost` +yourself**. The teardown shown under [Detecting a dropped +connection](#detecting-a-dropped-connection) is for devices without a reconnect policy. Here the +device does it for you, between attempts, and doing it as well just cancels the recovery you asked +for. + +```csharp +device.ReconnectOptions = new ReconnectOptions +{ + Enabled = true, + MaxAttempts = 10, + InitialDelay = TimeSpan.FromSeconds(2), + MaxDelay = TimeSpan.FromMinutes(1), + ResumeStreaming = true +}; +``` + +All three events are raised on a background thread, and a handler that throws is caught and +ignored — it cannot stop a reconnect in progress. + ### Working with Device Metadata After initialization, device metadata is populated: @@ -942,6 +1047,8 @@ streaming. - **Event-Driven**: Status changes and messages handled via events - **Observable Failures**: Background read and decode errors surface on `ErrorOccurred` instead of failing silently +- **Opt-in Auto-Reconnect**: A dropped connection can rebuild itself — transport, initialization, + channel configuration and stream — with no consumer code - **Type Safety**: Generic message types provide compile-time safety - **Retry Support**: Built-in connection retry with exponential backoff - **Thread-Safe Sending**: Background message queue for thread-safe command sending diff --git a/src/Daqifi.Core.Tests/Device/DeviceReconnectTests.cs b/src/Daqifi.Core.Tests/Device/DeviceReconnectTests.cs new file mode 100644 index 0000000..06e94c9 --- /dev/null +++ b/src/Daqifi.Core.Tests/Device/DeviceReconnectTests.cs @@ -0,0 +1,1747 @@ +using Daqifi.Core.Channel; +using Daqifi.Core.Communication.Messages; +using Daqifi.Core.Communication.Producers; +using Daqifi.Core.Communication.Transport; +using Daqifi.Core.Device; +using Google.Protobuf; +using System.Diagnostics; + +namespace Daqifi.Core.Tests.Device; + +/// +/// Issue #379: after a mid-stream drop, a device with reconnect enabled has to rebuild the whole +/// session by itself — transport, initialization, channel configuration, and the stream — with no +/// consumer code involved. With reconnect left at its default it must do exactly what it has always +/// done: report and stop. +/// +/// +/// Driven by a scripted transport that can be dropped on command and told to refuse the next N +/// reconnects, so the loop's success, retry, give-up and cancellation paths are all reachable +/// without hardware. The one test that goes the long way round — real read failures escalated by +/// the production — is +/// . +/// +public class DeviceReconnectTests +{ + private static readonly TimeSpan EventTimeout = TimeSpan.FromSeconds(15); + + /// + /// Safety valve on the gates that park a scripted connect or initialization mid-flight. A test + /// that is going to pass releases its gate within milliseconds, so this only ever fires on a + /// test that has already failed — which is exactly why it has to stay well under + /// . A gate outliving the assertion that gave up on it leaves a + /// background thread parked inside the transport long after the test finished, where it can + /// interleave with whatever runs next. + /// + private static readonly TimeSpan GateTimeout = TimeSpan.FromSeconds(5); + + /// A policy that reconnects promptly, so tests do not spend their time waiting. + private static ReconnectOptions FastPolicy(int maxAttempts = 4) => new() + { + Enabled = true, + MaxAttempts = maxAttempts, + InitialDelay = TimeSpan.FromMilliseconds(10), + MaxDelay = TimeSpan.FromMilliseconds(60), + BackoffMultiplier = 2.0 + }; + + #region Default behaviour is unchanged + + [Fact] + public void WithReconnectAtItsDefault_ADropStopsAtLost() + { + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Default Device", transport); + + ConnectAndInitialize(device); + + // Subscribed after the connect, so the recorded transitions are only what the drop caused. + var statuses = new List(); + device.StatusChanged += (_, e) => + { + lock (statuses) + { + statuses.Add(e.Status); + } + }; + + var reconnectEvents = 0; + device.ReconnectAttempt += (_, _) => Interlocked.Increment(ref reconnectEvents); + device.Reconnected += (_, _) => Interlocked.Increment(ref reconnectEvents); + device.ReconnectFailed += (_, _) => Interlocked.Increment(ref reconnectEvents); + + var connectsBeforeDrop = transport.ConnectCount; + + transport.SimulateDrop(); + + // Long enough that any reconnect worth having would have started by now. + Thread.Sleep(500); + + Assert.Equal(ConnectionStatus.Lost, device.Status); + Assert.False(device.IsReconnecting); + Assert.Equal(0, Volatile.Read(ref reconnectEvents)); + Assert.Equal(connectsBeforeDrop, transport.ConnectCount); + Assert.Equal(1, device.InitializeCount); + + lock (statuses) + { + // Exactly one transition, to Lost. No Retrying, no Failed, nothing else. + Assert.Equal(new[] { ConnectionStatus.Lost }, statuses); + } + } + + [Fact] + public void ReconnectIsOffOnAFreshDevice() + { + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Fresh Device", transport); + + Assert.False(device.ReconnectOptions.Enabled); + Assert.False(device.IsReconnecting); + } + + #endregion + + #region The session comes back + + [Fact] + public async Task AfterADrop_TheSessionIsRebuiltWithNoConsumerInvolvement() + { + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Resuming Device", transport); + device.ReconnectOptions = FastPolicy(); + + ConnectAndInitialize(device); + + var analog1 = device.Channels.Single(c => c.Type == ChannelType.Analog && c.ChannelNumber == 1); + var analog3 = device.Channels.Single(c => c.Type == ChannelType.Analog && c.ChannelNumber == 3); + var digital0 = device.Channels.Single(c => c.Type == ChannelType.Digital && c.ChannelNumber == 0); + device.EnableChannels(new[] { analog1, analog3, digital0 }); + + device.StreamingFrequency = 250; + device.StartStreaming(); + + var reconnected = WaitFor(h => device.Reconnected += h); + device.ClearSentCommands(); + + transport.SimulateDrop(); + + var result = await reconnected; + + // The device is connected, re-initialized and streaming again — nobody was asked to help. + Assert.Equal(ConnectionStatus.Connected, device.Status); + Assert.True(device.IsConnected); + Assert.Equal(2, device.InitializeCount); + Assert.True(device.IsStreaming); + Assert.True(result.StreamingResumed); + Assert.Equal(1, result.AttemptNumber); + Assert.True(result.Outage > TimeSpan.Zero); + Assert.False(device.IsReconnecting); + + // The channel configuration was replayed onto the reconnected device. It has to have been + // sent, not merely remembered: the scripted device reports every analog channel disabled + // when it comes back, exactly as a rebooted one does. + var sent = device.SentCommands; + var expectedMask = (1u << 1) | (1u << 3); + Assert.Contains($"ENAble:VOLTage:DC {expectedMask}", sent); + Assert.Contains("DIO:PORt:ENAble 1", sent); + + // ...and the stream restarted at the frequency it was running at before the drop. + Assert.Contains("SYSTem:StartStreamData 250", sent); + + Assert.True(device.Channels.Single(c => c.Type == ChannelType.Analog && c.ChannelNumber == 1).IsEnabled); + Assert.True(device.Channels.Single(c => c.Type == ChannelType.Analog && c.ChannelNumber == 3).IsEnabled); + Assert.False(device.Channels.Single(c => c.Type == ChannelType.Analog && c.ChannelNumber == 0).IsEnabled); + Assert.True(device.Channels.Single(c => c.Type == ChannelType.Digital && c.ChannelNumber == 0).IsEnabled); + } + + [Fact] + public async Task ASessionDrivenEntirelyByRawCommands_IsStillRestored() + { + // The shape the example CLI actually uses, and the one that failed on a real cable pull: + // channels enabled and streaming started with Send(...) rather than EnableChannels() and + // StartStreaming(). Core had no idea a session existed, so a recovered link reported + // StreamingResumed:false — having just stopped, during re-initialization, the stream that + // was still running. Every other test here drives the device through its typed API, which + // is why the whole suite stayed green while the bench failed. + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Raw Driven Device", transport); + device.ReconnectOptions = FastPolicy(); + + ConnectAndInitialize(device); + + device.Send(ScpiMessageProducer.EnableAdcChannels("1")); + device.Send(ScpiMessageProducer.StartStreaming(250)); + + // Core's view tracks the commands regardless of which API sent them. + Assert.True(device.IsStreaming); + Assert.Equal(250, device.StreamingFrequency); + Assert.True(device.Channels.Single(c => c.Type == ChannelType.Analog && c.ChannelNumber == 0).IsEnabled); + + var reconnected = WaitFor(h => device.Reconnected += h); + device.ClearSentCommands(); + + transport.SimulateDrop(); + var result = await reconnected; + + Assert.True(result.StreamingResumed); + Assert.True(device.IsStreaming); + + var sent = device.SentCommands; + Assert.Contains("ENAble:VOLTage:DC 1", sent); + Assert.Contains("SYSTem:StartStreamData 250", sent); + } + + [Fact] + public void ARawStartedStream_ReAnchorsTimestampsInsteadOfContinuingTheLastSession() + { + // The silent-wrong-data one. Timestamp reconstruction advances a per-session anchor by the + // device-tick delta between frames, so a session that reuses the previous anchor stamps its + // samples with times that never happened. StartStreaming() resets it; a raw start used to + // skip that entirely. + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Re-anchoring Device", transport); + + ConnectAndInitialize(device); + + var channel = (IAnalogChannel)device.Channels.First(c => c.Type == ChannelType.Analog); + var samples = new List(); + channel.SampleReceived += (_, e) => + { + lock (samples) + { + samples.Add(e.Sample); + } + }; + + device.EnableChannel(channel); + + // A first session, ending with the device clock near zero. + device.StartStreaming(); + device.InvokeStreamMessage(AnalogFrame(1_000, 1.0f)); + device.StopStreaming(); + + // A second session started by raw command, with the device clock far ahead — a device that + // has been running a while, or has rebooted its counter. Continuing the old anchor would + // push the timestamp roughly a minute into the future at the 50 MHz default tick rate. + device.Send(ScpiMessageProducer.StartStreaming(100)); + device.InvokeStreamMessage(AnalogFrame(3_000_000_000, 2.0f)); + + lock (samples) + { + Assert.Equal(2, samples.Count); + var elapsed = samples[1].Timestamp - samples[0].Timestamp; + Assert.True( + elapsed < TimeSpan.FromSeconds(10), + $"the second session's first sample landed {elapsed.TotalSeconds:0.#}s after the first " + + "session's, so it was anchored to the previous session rather than to now"); + } + } + + [Fact] + public void ARawStartedStream_StartsWithAFreshGapDetector() + { + // The gap detector compares device-clock deltas between consecutive frames. Carried across + // a session boundary it reports the boundary itself as a huge dropped-sample gap. + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Fresh Gap Device", transport); + + ConnectAndInitialize(device); + device.EnableChannel(device.Channels.First(c => c.Type == ChannelType.Analog)); + + var gaps = 0; + device.GapDetected += (_, _) => Interlocked.Increment(ref gaps); + + device.StartStreaming(); + device.InvokeStreamMessage(AnalogFrame(1_000, 1.0f)); + device.InvokeStreamMessage(AnalogFrame(501_000, 1.0f)); + device.StopStreaming(); + + var gapsBefore = Volatile.Read(ref gaps); + + device.Send(ScpiMessageProducer.StartStreaming(100)); + device.InvokeStreamMessage(AnalogFrame(3_000_000_000, 2.0f)); + + Assert.Equal(gapsBefore, Volatile.Read(ref gaps)); + } + + [Fact] + public void ARawStartedStream_StartsItsDecodeFailureCountAtZero() + { + // DecodeFailureCount is documented as describing the current session, and StartStreaming + // resets it. A raw start that inherited the previous session's tally would make a healthy + // stream look broken. + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Fresh Count Device", transport); + + ConnectAndInitialize(device); + + var channel = (IAnalogChannel)device.Channels.First(c => c.Type == ChannelType.Analog); + channel.SampleReceived += (_, _) => throw new InvalidOperationException("decode consumer is broken"); + device.EnableChannel(channel); + + device.StartStreaming(); + device.InvokeStreamMessage(AnalogFrame(1_000, 1.0f)); + device.InvokeStreamMessage(AnalogFrame(2_000, 1.0f)); + Assert.Equal(2, device.DecodeFailureCount); + + device.StopStreaming(); + device.Send(ScpiMessageProducer.StartStreaming(100)); + + Assert.Equal(0, device.DecodeFailureCount); + } + + [Fact] + public void ARawStartWhileAlreadyStreaming_RecordsTheRateWithoutRestartingTheSession() + { + // The typed API cannot express a restart-in-place (StartStreaming returns early), so there + // is no session boundary here and nothing to re-anchor — but the new rate is still real and + // must be what a reconnect would replay. + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Restarting Device", transport); + + ConnectAndInitialize(device); + + var channel = (IAnalogChannel)device.Channels.First(c => c.Type == ChannelType.Analog); + channel.SampleReceived += (_, _) => throw new InvalidOperationException("decode consumer is broken"); + device.EnableChannel(channel); + + device.StartStreaming(); + device.InvokeStreamMessage(AnalogFrame(1_000, 1.0f)); + Assert.Equal(1, device.DecodeFailureCount); + + device.Send(ScpiMessageProducer.StartStreaming(400)); + + Assert.True(device.IsStreaming); + Assert.Equal(400, device.StreamingFrequency); + + // Not a new session: the running tally is left alone. + Assert.Equal(1, device.DecodeFailureCount); + } + + [Fact] + public async Task TrackingARate_NeverThrowsOutOfSend_WhileCapabilitiesChangeUnderneath() + { + // MaxSamplingRate is a mutable public property, and a status message can update it at any + // moment on the consumer thread. Validating a rate against one read and then assigning it + // through the public setter — which reads it again and throws on a value it dislikes — + // leaves a window where tracking turns a command that has ALREADY reached the device into + // an exception at the caller's Send(). This drives that window directly. + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Shifting Ceiling Device", transport); + + ConnectAndInitialize(device); + + using var stop = new CancellationTokenSource(TimeSpan.FromSeconds(3)); + Exception? escaped = null; + + var churn = Task.Run(() => + { + // Flip the ceiling between "accepts 900" and "accepts almost nothing" as fast as + // possible, so it changes between the two reads. + while (!stop.IsCancellationRequested) + { + device.Metadata.Capabilities.MaxSamplingRate = 1000; + device.Metadata.Capabilities.MaxSamplingRate = 1; + } + }); + + try + { + for (var i = 0; i < 20_000 && escaped == null; i++) + { + escaped = Record.Exception(() => device.Send(ScpiMessageProducer.StartStreaming(900))); + } + } + finally + { + stop.Cancel(); + await churn; + } + + Assert.Null(escaped); + } + + [Theory] + [InlineData("SYSTem:StartStreamData")] // no argument at all + [InlineData("SYSTem:StartStreamData ")] // argument present but empty + [InlineData("SYSTem:StartStreamData abc")] // not a number + [InlineData("SYSTem:StartStreamData 100 extra")] // trailing junk + [InlineData("SYSTem:StartStreamData 0")] // below the usable range + [InlineData("SYSTem:StartStreamData -5")] // negative + [InlineData("SYSTem:StartStreamData 99999999")] // beyond the device's sampling rate + public void AStartStreamingCommandWithAnUnusableRate_DoesNotMarkTheDeviceStreaming(string command) + { + // The bench proved the happy path; a valid rate never reaches this branch. Marking the + // device as streaming here would leave StreamingFrequency holding a rate from some earlier + // session, and a reconnect would then restore that rate — resuming at a number nobody asked + // for, which is precisely the silent-wrong-data mode this feature exists to prevent. + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Malformed Command Device", transport); + + ConnectAndInitialize(device); + + var frequencyBefore = device.StreamingFrequency; + + device.Send(new ScpiMessage(command)); + + Assert.False(device.IsStreaming); + Assert.Equal(frequencyBefore, device.StreamingFrequency); + } + + [Fact] + public void AnUnusableStartCommand_LeavesARunningSessionExactlyAsItWas() + { + // The firmware rejects the malformed command and keeps streaming at the rate it already + // had, so clearing the flags would swap one inaccuracy for another. + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Still Streaming Device", transport); + + ConnectAndInitialize(device); + + device.Send(ScpiMessageProducer.StartStreaming(250)); + Assert.True(device.IsStreaming); + Assert.Equal(250, device.StreamingFrequency); + + device.Send(new ScpiMessage("SYSTem:StartStreamData wat")); + + Assert.True(device.IsStreaming); + Assert.Equal(250, device.StreamingFrequency); + } + + [Fact] + public void AfterAnUnusableStartCommand_AGenuineStartStreamingStillWorks() + { + // The stale-flag trap from issue #118: StartStreaming() returns early while IsStreaming is + // set, so a spuriously-true flag would silently swallow the caller's real request. + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Recoverable Device", transport); + + ConnectAndInitialize(device); + + device.Send(new ScpiMessage("SYSTem:StartStreamData not-a-rate")); + device.ClearSentCommands(); + + device.StreamingFrequency = 200; + device.StartStreaming(); + + Assert.True(device.IsStreaming); + Assert.Contains("SYSTem:StartStreamData 200", device.SentCommands); + } + + [Fact] + public async Task AnUnusableStartCommand_LeavesNothingForAReconnectToResume() + { + // The end-to-end consequence: no session was established, so a reconnect must not invent + // one at whatever rate happened to be left in StreamingFrequency. + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Nothing To Resume Device", transport); + device.ReconnectOptions = FastPolicy(); + + ConnectAndInitialize(device); + device.Send(new ScpiMessage("SYSTem:StartStreamData oops")); + + var reconnected = WaitFor(h => device.Reconnected += h); + device.ClearSentCommands(); + + transport.SimulateDrop(); + var result = await reconnected; + + Assert.False(result.StreamingResumed); + Assert.False(device.IsStreaming); + Assert.DoesNotContain( + device.SentCommands, + c => c.StartsWith("SYSTem:StartStreamData", StringComparison.Ordinal)); + } + + [Fact] + public async Task ADeviceThatWasNotStreaming_ComesBackIdle() + { + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Idle Device", transport); + device.ReconnectOptions = FastPolicy(); + + ConnectAndInitialize(device); + device.EnableChannel(device.Channels.First(c => c.Type == ChannelType.Analog)); + + var reconnected = WaitFor(h => device.Reconnected += h); + device.ClearSentCommands(); + + transport.SimulateDrop(); + var result = await reconnected; + + Assert.False(result.StreamingResumed); + Assert.False(device.IsStreaming); + Assert.DoesNotContain(device.SentCommands, c => c.StartsWith("SYSTem:StartStreamData", StringComparison.Ordinal)); + + // The channel configuration is restored regardless of whether a stream was running. + Assert.Contains("ENAble:VOLTage:DC 1", device.SentCommands); + } + + [Fact] + public async Task WithResumeStreamingOff_TheChannelsComeBackButTheStreamDoesNot() + { + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Manual Resume Device", transport); + var policy = FastPolicy(); + policy.ResumeStreaming = false; + device.ReconnectOptions = policy; + + ConnectAndInitialize(device); + device.EnableChannel(device.Channels.First(c => c.Type == ChannelType.Analog)); + device.StartStreaming(); + + var reconnected = WaitFor(h => device.Reconnected += h); + device.ClearSentCommands(); + + transport.SimulateDrop(); + var result = await reconnected; + + Assert.False(result.StreamingResumed); + Assert.Contains("ENAble:VOLTage:DC 1", device.SentCommands); + Assert.DoesNotContain(device.SentCommands, c => c.StartsWith("SYSTem:StartStreamData", StringComparison.Ordinal)); + + // The device is genuinely idle, and says so. Reporting a stale IsStreaming here would both + // lie and make the caller's own StartStreaming() a silent no-op. + Assert.False(device.IsStreaming); + device.StartStreaming(); + Assert.True(device.IsStreaming); + Assert.Contains(device.SentCommands, c => c.StartsWith("SYSTem:StartStreamData", StringComparison.Ordinal)); + } + + [Fact] + public async Task ADeviceThatRefusesTheFirstAttempts_IsRetriedUntilItAnswers() + { + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Slow-To-Return Device", transport); + device.ReconnectOptions = FastPolicy(maxAttempts: 5); + + ConnectAndInitialize(device); + device.EnableChannel(device.Channels.First(c => c.Type == ChannelType.Analog)); + + var attempts = new List(); + device.ReconnectAttempt += (_, e) => + { + lock (attempts) + { + attempts.Add(e); + } + }; + + var reconnected = WaitFor(h => device.Reconnected += h); + + // The device is still coming back up: the next two connects are refused. + transport.FailNextConnects(2); + transport.SimulateDrop(); + + var result = await reconnected; + + Assert.Equal(3, result.AttemptNumber); + + lock (attempts) + { + Assert.Equal(3, attempts.Count); + Assert.Equal(new[] { 1, 2, 3 }, attempts.Select(a => a.AttemptNumber)); + Assert.All(attempts, a => Assert.Equal(5, a.MaxAttempts)); + + // Backing off, not hammering: each wait is longer than the last. + Assert.True(attempts[1].Delay > attempts[0].Delay); + Assert.True(attempts[2].Delay > attempts[1].Delay); + + // The first attempt has nothing to report; the later ones say why the previous failed. + Assert.Null(attempts[0].PreviousError); + Assert.NotNull(attempts[1].PreviousError); + Assert.NotNull(attempts[2].PreviousError); + } + } + + [Fact] + public async Task AMidStreamReadFailure_DrivesTheWholeLoopEndToEnd() + { + // The long way round: no simulated drop, just reads that start failing. The production + // watchdog escalates them to Lost, which is what the reconnect loop hangs off. + using var transport = new ScriptedReconnectTransport(useWatchdog: true); + using var device = new ScriptedStreamingDevice("Unplugged Device", transport); + device.ReconnectOptions = FastPolicy(); + + ConnectAndInitialize(device); + device.EnableChannel(device.Channels.First(c => c.Type == ChannelType.Analog)); + device.StartStreaming(); + + var reconnected = WaitFor(h => device.Reconnected += h); + device.ClearSentCommands(); + + transport.FailReadsUntilNextConnect(); + + var result = await reconnected; + + Assert.True(result.StreamingResumed); + Assert.Equal(ConnectionStatus.Connected, device.Status); + Assert.Contains("ENAble:VOLTage:DC 1", device.SentCommands); + Assert.Contains(device.SentCommands, c => c.StartsWith("SYSTem:StartStreamData", StringComparison.Ordinal)); + } + + #endregion + + #region Giving up + + [Fact] + public async Task WhenEveryAttemptFails_ItGivesUpLoudly() + { + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Gone Device", transport); + device.ReconnectOptions = FastPolicy(maxAttempts: 3); + + ConnectAndInitialize(device); + + DeviceErrorEventArgs? deviceError = null; + device.ErrorOccurred += (_, e) => deviceError = e; + + var failed = WaitFor(h => device.ReconnectFailed += h); + + transport.FailNextConnects(int.MaxValue); + transport.SimulateDrop(); + + var result = await failed; + + Assert.Equal(3, result.AttemptsMade); + Assert.False(result.WasCanceled); + Assert.NotNull(result.LastError); + + // Terminal, and impossible to miss: a distinct status plus the device error surface. + Assert.Equal(ConnectionStatus.Failed, device.Status); + Assert.False(device.IsConnected); + + Assert.NotNull(deviceError); + Assert.Equal(DeviceErrorSource.Reconnect, deviceError!.Source); + var reconnectFailure = Assert.IsType(deviceError.Error); + Assert.Equal(3, reconnectFailure.AttemptsMade); + Assert.Equal("Gone Device", reconnectFailure.DeviceName); + Assert.NotNull(reconnectFailure.InnerException); + + // It really did stop: no further attempts after the report. + var connectsAtGiveUp = transport.ConnectCount; + Thread.Sleep(300); + Assert.Equal(connectsAtGiveUp, transport.ConnectCount); + Assert.False(device.IsReconnecting); + } + + [Fact] + public async Task AnInitializationThatKeepsFailing_AlsoExhaustsTheAttempts() + { + // The transport comes back every time; it is the device that will not initialize. The loop + // must not treat a reachable-but-unusable device as a success. + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Unresponsive Device", transport); + device.ReconnectOptions = FastPolicy(maxAttempts: 2); + + ConnectAndInitialize(device); + + // Set after the first (successful) initialization, so only the reconnect's attempts fail. + device.InitializeFailure = _ => + new TimeoutException("the device never reported its channel configuration"); + + var failed = WaitFor(h => device.ReconnectFailed += h); + transport.SimulateDrop(); + + var result = await failed; + + Assert.Equal(2, result.AttemptsMade); + Assert.IsType(result.LastError); + Assert.Equal(ConnectionStatus.Failed, device.Status); + + // Nothing half-open is left behind: the transport it managed to open was closed again. + Assert.False(transport.IsConnected); + } + + #endregion + + #region Cancellation + + [Fact] + public async Task CancelReconnect_StopsTheLoopAndLeavesTheDeviceLost() + { + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Abandoned Device", transport); + var policy = FastPolicy(maxAttempts: 20); + policy.InitialDelay = TimeSpan.FromSeconds(30); + policy.MaxDelay = TimeSpan.FromSeconds(30); + device.ReconnectOptions = policy; + + ConnectAndInitialize(device); + + var failed = WaitFor(h => device.ReconnectFailed += h); + + transport.FailNextConnects(int.MaxValue); + transport.SimulateDrop(); + + // Wait until the loop is parked in its 30 s backoff, so cancellation lands at a known point + // rather than racing the loop's own teardown. + WaitUntilRetrying(device); + Assert.True(device.IsReconnecting); + + device.CancelReconnect(); + + var result = await failed; + + // The 30 s backoff means cancellation, not exhaustion, is what ended this. + Assert.True(result.WasCanceled); + Assert.Equal(1, result.AttemptsMade); + Assert.Equal(ConnectionStatus.Lost, device.Status); + Assert.False(device.IsReconnecting); + Assert.False(device.IsConnected); + } + + [Fact] + public async Task Disconnect_DuringAReconnect_WinsAndTheLoopStopsQuietly() + { + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Departing Device", transport); + var policy = FastPolicy(maxAttempts: 20); + policy.InitialDelay = TimeSpan.FromSeconds(30); + policy.MaxDelay = TimeSpan.FromSeconds(30); + device.ReconnectOptions = policy; + + ConnectAndInitialize(device); + + var failed = WaitFor(h => device.ReconnectFailed += h); + + transport.FailNextConnects(int.MaxValue); + transport.SimulateDrop(); + WaitUntilRetrying(device); + + device.Disconnect(); + + var result = await failed; + Assert.True(result.WasCanceled); + + // The caller's teardown owns the outcome: the loop must not have overwritten it, nor + // re-opened the transport behind it. + Thread.Sleep(300); + Assert.Equal(ConnectionStatus.Disconnected, device.Status); + Assert.False(transport.IsConnected); + Assert.False(device.IsReconnecting); + } + + [Fact] + public void Disconnect_WhileAConnectAttemptIsInFlight_DoesNotLeaveTheDeviceQuietlyAlive() + { + // Opening a port blocks for as long as it takes, so a caller pulling the plug on the device + // lands *inside* an attempt rather than tidily between two. The attempt still finishes and + // brings the transport back up; if the loop merely bails out at that point, the caller is + // left holding a device they closed that is silently open again, reporting Connected, with + // a reader thread running on it. + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Resurrected Device", transport); + device.ReconnectOptions = FastPolicy(); + + ConnectAndInitialize(device); + + var connectGate = transport.BlockConnects(); + transport.ConnectEntered.Reset(); + + transport.SimulateDrop(); + + Assert.True( + transport.ConnectEntered.Wait(EventTimeout), + "the reconnect never reached the transport's connect"); + + // Release the parked attempt from another thread. The caller's Disconnect now waits for an + // in-flight connect rather than running alongside it, so releasing after it returns would + // wait on a call that is itself waiting on the gate. + ReleaseAfter(connectGate, TimeSpan.FromMilliseconds(100)); + + // The caller shuts the device down while that connect is in flight. The attempt still + // finishes and re-opens the transport. + device.Disconnect(); + Assert.Equal(ConnectionStatus.Disconnected, device.Status); + + WaitUntil(() => !device.IsReconnecting, "the reconnect loop never finished"); + + // The caller's decision has to survive it. + Assert.Equal(ConnectionStatus.Disconnected, device.Status); + Assert.False(device.IsConnected); + Assert.False(transport.IsConnected); + } + + [Fact] + public void ACallerDisconnect_NeverDrivesTheTransportAlongsideAnInFlightReconnect() + { + // Cancelling is not synchronizing. SupersedeReconnect asks the loop to stop and returns + // immediately, but a loop already inside a blocking transport connect cannot be + // interrupted — so without a lifecycle lock the caller's Disconnect closes the port while + // the reconnect is still opening it, and both threads race to build a message consumer. + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Contended Device", transport); + device.ReconnectOptions = FastPolicy(); + + ConnectAndInitialize(device); + + var connectGate = transport.BlockConnects(); + transport.ConnectEntered.Reset(); + + transport.SimulateDrop(); + + Assert.True( + transport.ConnectEntered.Wait(EventTimeout), + "the reconnect never reached the transport's connect"); + + ReleaseAfter(connectGate, TimeSpan.FromMilliseconds(100)); + + device.Disconnect(); + + WaitUntil(() => !device.IsReconnecting, "the reconnect loop never finished"); + + Assert.False( + transport.SawConcurrentLifecycleCalls, + "a caller's Disconnect was inside the transport at the same time as the reconnect's connect"); + + // The round-one guarantee still holds on top of the new one. + Assert.Equal(ConnectionStatus.Disconnected, device.Status); + Assert.False(device.IsConnected); + Assert.False(transport.IsConnected); + } + + [Fact] + public void AConnectThatCannotTakeTheLifecycleLock_FailsRatherThanRunningAlongside() + { + // The escape hatch: an earlier revision logged a warning on timeout and ran the operation + // anyway, which is precisely the double-open it was added to prevent — two threads both + // finding no message consumer, both starting one, two readers on one stream. Contention + // now has to fail, not proceed. + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Contended Connect Device", transport) + { + LifecycleLockTimeoutOverride = TimeSpan.FromMilliseconds(200) + }; + device.ReconnectOptions = FastPolicy(); + + ConnectAndInitialize(device); + + var connectGate = transport.BlockConnects(); + transport.ConnectEntered.Reset(); + + transport.SimulateDrop(); + Assert.True( + transport.ConnectEntered.Wait(EventTimeout), + "the reconnect never reached the transport's connect"); + + // The reconnect is parked inside the transport holding the lifecycle lock, so a caller + // connect cannot have it. It must give up rather than open a second connection alongside. + var thrown = Assert.Throws(() => device.Connect()); + Assert.Contains("Nothing was opened", thrown.Message, StringComparison.OrdinalIgnoreCase); + + Assert.False( + transport.SawConcurrentLifecycleCalls, + "a caller's Connect ran inside the transport alongside the reconnect's connect"); + + connectGate.Set(); + WaitUntil(() => !device.IsReconnecting, "the reconnect loop never finished"); + } + + [Fact] + public void ADisconnectThatCannotTakeTheLifecycleLock_WaitsRatherThanFailing() + { + // The other half of the policy. Teardown is the resource-release path and Dispose depends + // on it, so contention must not turn it into an exception — it waits instead, and the + // short connect-side timeout must not leak into it. + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Patient Teardown Device", transport) + { + LifecycleLockTimeoutOverride = TimeSpan.FromMilliseconds(200) + }; + device.ReconnectOptions = FastPolicy(); + + ConnectAndInitialize(device); + + var connectGate = transport.BlockConnects(); + transport.ConnectEntered.Reset(); + + transport.SimulateDrop(); + Assert.True( + transport.ConnectEntered.Wait(EventTimeout), + "the reconnect never reached the transport's connect"); + + // Held well past the connect-side timeout, so a shared budget would have expired. + ReleaseAfter(connectGate, TimeSpan.FromMilliseconds(600)); + + device.Disconnect(); + + Assert.Equal(ConnectionStatus.Disconnected, device.Status); + Assert.False(transport.SawConcurrentLifecycleCalls); + + WaitUntil(() => !device.IsReconnecting, "the reconnect loop never finished"); + Assert.False(transport.IsConnected); + } + + [Fact] + public void ADisconnectBehindAWedgedConnect_GivesUpRatherThanHangingForever() + { + // SerialPort.Open is synchronous and uncancellable, and this repo already knows it can + // wedge — SerialDeviceFinder carries a process-wide port quarantine built for exactly that. + // So a holder is NOT guaranteed to be bounded, and a teardown that waits on one without a + // bound inherits the hang, turning Dispose into a permanent block. + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Wedged Device", transport) + { + LifecycleLockTimeoutOverride = TimeSpan.FromMilliseconds(200), + TeardownLockTimeoutOverride = TimeSpan.FromMilliseconds(300) + }; + device.ReconnectOptions = FastPolicy(); + + ConnectAndInitialize(device); + + // Never released: this connect is wedged, exactly as a hung port would be. + transport.BlockConnects(); + transport.ConnectEntered.Reset(); + + transport.SimulateDrop(); + Assert.True( + transport.ConnectEntered.Wait(EventTimeout), + "the reconnect never reached the transport's connect"); + + var stopwatch = Stopwatch.StartNew(); + device.Disconnect(); + stopwatch.Stop(); + + // It gave up on the stuck holder instead of waiting for a return that may never come. + Assert.True( + stopwatch.Elapsed < TimeSpan.FromSeconds(3), + $"Disconnect blocked for {stopwatch.ElapsedMilliseconds}ms behind a wedged connect; " + + "it should have abandoned the wait"); + + // The caller's intent is still recorded, even though the transport could not be touched. + Assert.Equal(ConnectionStatus.Disconnected, device.Status); + Assert.False(device.IsConnected); + + // And it did not race the wedged connect to get there. + Assert.False(transport.SawConcurrentLifecycleCalls); + } + + [Fact] + public async Task ACallerConnectOutlivingAnAbandonedTeardown_DoesNotBringTheDeviceBackToLife() + { + // Reconnect is deliberately left OFF here: the lifecycle lock and its abandon path apply to + // every device, so this failure needs no reconnect loop at all. A caller's connect wedges + // holding the lock, the caller's own Disconnect abandons its wait and returns reporting + // Disconnected, and then the wedged connect finishes — setting Connected and starting a + // reader behind a caller who was already told the device was down. AbandonIfSuperseded does + // not help here: it belongs to the reconnect loop and never runs on a caller's connect. + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Overruled Device", transport) + { + TeardownLockTimeoutOverride = TimeSpan.FromMilliseconds(300) + }; + + var connectGate = transport.BlockConnects(); + transport.ConnectEntered.Reset(); + + var connecting = Task.Run(() => device.Connect()); + + Assert.True( + transport.ConnectEntered.Wait(EventTimeout), + "the caller's connect never reached the transport"); + + // Released well after the teardown gives up, so Disconnect abandons and returns first. + ReleaseAfter(connectGate, TimeSpan.FromMilliseconds(800)); + + device.Disconnect(); + Assert.Equal(ConnectionStatus.Disconnected, device.Status); + + await connecting; + + // The caller asked for teardown; it has to stand whoever was holding the lock. + Assert.Equal(ConnectionStatus.Disconnected, device.Status); + Assert.False(device.IsConnected); + Assert.False(transport.IsConnected); + + // And it was put back down sequentially, never raced. + Assert.False(transport.SawConcurrentLifecycleCalls); + } + + [Fact] + public async Task ACancelledDisconnectAsync_StillTearsTheConnectionDown() + { + // A defect that exists only in the merged result. #341 defines the token for + // DisconnectAsync as shortening the wait for an in-flight command exchange — the disconnect + // itself always completes and never throws. #379's lifecycle lock added a second wait that + // the token was then handed to, and a cancelled token there was treated as "abandon + // teardown". SemaphoreSlim throws for an already-cancelled token even when it is free, so + // this hit EVERY cancelled disconnect, not just a contended one: the device reported + // Disconnected with the transport still open and the message pumps still running. + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Cancelled Teardown Device", transport); + + ConnectAndInitialize(device); + Assert.True(transport.IsConnected); + + using var alreadyCancelled = new CancellationTokenSource(); + await alreadyCancelled.CancelAsync(); + + // #341's contract: this must not throw, however cancelled the token is. + await device.DisconnectAsync(alreadyCancelled.Token); + + Assert.Equal(ConnectionStatus.Disconnected, device.Status); + Assert.False(device.IsConnected); + + // The wire-level evidence: teardown really ran rather than being skipped. + Assert.False(transport.IsConnected); + + // And the pumps really stopped — a device whose producer was left running would still + // accept a send instead of refusing it. + Assert.Throws( + () => device.Send(ScpiMessageProducer.StopStreaming)); + } + + [Fact] + public async Task ACancelledDisconnectAsync_StillWaitsOutAConnectInFlight() + { + // The token must not shortcut the lifecycle wait either: tearing down alongside a connect + // is the corruption the lock exists to prevent, so a cancelled teardown still waits for the + // holder — bounded by TeardownLockTimeout, which is what covers a genuinely wedged one. + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Cancelled Contended Device", transport); + + ConnectAndInitialize(device); + + var connectGate = transport.BlockConnects(); + transport.ConnectEntered.Reset(); + + var connecting = Task.Run(() => device.Connect()); + Assert.True( + transport.ConnectEntered.Wait(EventTimeout), + "the connect never reached the transport"); + + ReleaseAfter(connectGate, TimeSpan.FromMilliseconds(200)); + + using var alreadyCancelled = new CancellationTokenSource(); + await alreadyCancelled.CancelAsync(); + + await device.DisconnectAsync(alreadyCancelled.Token); + await connecting; + + Assert.Equal(ConnectionStatus.Disconnected, device.Status); + Assert.False(transport.IsConnected); + Assert.False(transport.SawConcurrentLifecycleCalls); + } + + [Fact] + public void ANormalConnectDisconnectCycle_ReportsNoLifecycleContention() + { + // The uncontended path is unchanged by the lock: nothing waits, nothing overlaps. + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Ordinary Device", transport); + + ConnectAndInitialize(device); + device.Disconnect(); + + Assert.False(transport.SawConcurrentLifecycleCalls); + Assert.Equal(ConnectionStatus.Disconnected, device.Status); + } + + [Fact] + public async Task Disconnect_WhileInitializationIsInFlight_IsNotReportedAsASuccessfulReconnect() + { + // The same race one step later: the transport came back and initialization was most of the + // way through when the caller disconnected. Announcing a recovered session at that point + // would hand them a device they had just closed. + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedBaseDevice("Late Abandoned Device", transport); + device.ReconnectOptions = FastPolicy(); + + device.Connect(); + await device.InitializeAsync(); + + var initGate = new ManualResetEventSlim(false); + device.InitializeEntered.Reset(); + device.InitializeGate = initGate; + + var reconnectedCount = 0; + device.Reconnected += (_, _) => Interlocked.Increment(ref reconnectedCount); + + transport.SimulateDrop(); + + Assert.True( + device.InitializeEntered.Wait(EventTimeout), + "the reconnect never reached initialization"); + + device.Disconnect(); + initGate.Set(); + + WaitUntil(() => !device.IsReconnecting, "the reconnect loop never finished"); + + Assert.Equal(0, Volatile.Read(ref reconnectedCount)); + Assert.Equal(ConnectionStatus.Disconnected, device.Status); + Assert.False(device.IsConnected); + Assert.False(transport.IsConnected); + } + + [Fact] + public void DisposingDuringAReconnect_DoesNotThrow() + { + var transport = new ScriptedReconnectTransport(); + var device = new ScriptedStreamingDevice("Disposed Device", transport); + device.ReconnectOptions = FastPolicy(maxAttempts: 20); + + ConnectAndInitialize(device); + + transport.FailNextConnects(int.MaxValue); + transport.SimulateDrop(); + WaitUntilRetrying(device); + + device.Dispose(); + transport.Dispose(); + + // Whatever the loop was in the middle of, it unwinds without surfacing anything. + Thread.Sleep(400); + } + + #endregion + + #region Robustness + + [Fact] + public async Task AThrowingReconnectSubscriber_DoesNotStopTheLoop() + { + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Badly Observed Device", transport); + device.ReconnectOptions = FastPolicy(); + + ConnectAndInitialize(device); + device.ReconnectAttempt += (_, _) => throw new InvalidOperationException("a badly behaved subscriber"); + + var reconnected = WaitFor(h => device.Reconnected += h); + transport.SimulateDrop(); + + var result = await reconnected; + Assert.Equal(1, result.AttemptNumber); + Assert.Equal(ConnectionStatus.Connected, device.Status); + } + + [Fact] + public async Task ASecondDropAfterARecovery_StartsAFreshLoop() + { + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Flaky Device", transport); + device.ReconnectOptions = FastPolicy(); + + ConnectAndInitialize(device); + device.EnableChannel(device.Channels.First(c => c.Type == ChannelType.Analog)); + device.StartStreaming(); + + var first = WaitFor(h => device.Reconnected += h); + transport.SimulateDrop(); + await first; + + // Reconnected is raised from inside the loop; let it finish unwinding so the second drop is + // unambiguously a fresh one rather than one folded into the loop still in flight. + WaitUntil(() => !device.IsReconnecting, "the first reconnect loop never finished"); + + var second = WaitFor(h => device.Reconnected += h); + device.ClearSentCommands(); + transport.SimulateDrop(); + var result = await second; + + Assert.True(result.StreamingResumed); + Assert.Equal(3, device.InitializeCount); + Assert.Contains("ENAble:VOLTage:DC 1", device.SentCommands); + } + + [Fact] + public void AHandlerThatDisconnectsOnLost_IsNotOverruledByAReconnect() + { + // The teardown-on-Lost pattern the docs show for devices without reconnect. If a consumer + // still does it, their decision has to stand — a reconnect starting behind it would reopen + // a device they just closed. + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Torn Down Device", transport); + device.ReconnectOptions = FastPolicy(); + + ConnectAndInitialize(device); + + device.StatusChanged += (_, e) => + { + if (e.Status == ConnectionStatus.Lost) + { + device.Disconnect(); + } + }; + + var reconnectEvents = 0; + device.ReconnectAttempt += (_, _) => Interlocked.Increment(ref reconnectEvents); + + var connectsBeforeDrop = transport.ConnectCount; + transport.SimulateDrop(); + + Thread.Sleep(500); + + Assert.Equal(ConnectionStatus.Disconnected, device.Status); + Assert.Equal(0, Volatile.Read(ref reconnectEvents)); + Assert.Equal(connectsBeforeDrop, transport.ConnectCount); + Assert.False(transport.IsConnected); + Assert.False(device.IsReconnecting); + } + + [Fact] + public void SettingReconnectOptionsToNull_IsRejected() + { + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Picky Device", transport); + + Assert.Throws(() => device.ReconnectOptions = null!); + } + + [Fact] + public void CancelReconnect_WithNothingRunning_IsHarmless() + { + using var transport = new ScriptedReconnectTransport(); + using var device = new ScriptedStreamingDevice("Quiet Device", transport); + + device.CancelReconnect(); + device.CancelReconnect(); + + Assert.False(device.IsReconnecting); + } + + #endregion + + #region Helpers + + private static void ConnectAndInitialize(ScriptedStreamingDevice device) + { + device.Connect(); + device.InitializeAsync().GetAwaiter().GetResult(); + } + + /// + /// Blocks until the reconnect loop has torn the dead session down and is waiting out its + /// backoff, which is the only point at which a test can cancel without racing that teardown. + /// + private static void WaitUntilRetrying(DaqifiStreamingDevice device) => + WaitUntil( + () => device.Status == ConnectionStatus.Retrying, + "the reconnect loop never reached its backoff wait"); + + /// + /// Releases a gate from another thread after a short delay, for tests where the call that would + /// otherwise release it is itself waiting on that gate. + /// + private static void ReleaseAfter(ManualResetEventSlim gate, TimeSpan delay) + { + _ = Task.Run(async () => + { + await Task.Delay(delay).ConfigureAwait(false); + gate.Set(); + }); + } + + /// Builds a single-channel analog frame carrying a device-clock timestamp. + private static DaqifiOutMessage AnalogFrame(uint deviceTimestamp, float value) + { + var frame = new DaqifiOutMessage { MsgTimeStamp = deviceTimestamp }; + frame.AnalogInDataFloat.Add(value); + return frame; + } + + private static void WaitUntil(Func condition, string because) + { + var deadline = DateTime.UtcNow + EventTimeout; + while (DateTime.UtcNow < deadline) + { + if (condition()) + { + return; + } + + Thread.Sleep(5); + } + + Assert.True(condition(), because); + } + + /// + /// Subscribes to an event and returns a task that completes with its first payload, so a test + /// can arm the wait before provoking the drop that satisfies it. + /// + private static Task WaitFor(Action> subscribe) + where TArgs : EventArgs + { + var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + subscribe((_, e) => tcs.TrySetResult(e)); + + return WaitWithTimeoutAsync(tcs.Task); + } + + private static async Task WaitWithTimeoutAsync(Task task) + { + var completed = await Task.WhenAny(task, Task.Delay(EventTimeout)).ConfigureAwait(false); + Assert.True(completed == task, $"timed out waiting for {typeof(TArgs).Name}"); + return await task.ConfigureAwait(false); + } + + /// + /// A streaming device with no real SCPI: commands are captured rather than written, and + /// initialization synthesizes the status message a device would have sent. + /// + private sealed class ScriptedStreamingDevice : DaqifiStreamingDevice + { + private readonly List _sent = new(); + private int _initializeCount; + + public ScriptedStreamingDevice(string name, IStreamTransport transport) + : base(name, transport) + { + } + + /// + /// Shortens the connect-side lifecycle wait so the contention path is reachable in a test + /// without a ten-second pause. Mirrors how SdCardDownloadTimeout is overridden. + /// + public TimeSpan? LifecycleLockTimeoutOverride { get; init; } + + internal override TimeSpan LifecycleLockTimeout => + LifecycleLockTimeoutOverride ?? base.LifecycleLockTimeout; + + /// + /// Shortens the teardown-side wait so the abandon path is reachable without a thirty-second + /// pause. + /// + public TimeSpan? TeardownLockTimeoutOverride { get; init; } + + internal override TimeSpan TeardownLockTimeout => + TeardownLockTimeoutOverride ?? base.TeardownLockTimeout; + + /// Number of analog channels the scripted device reports. + public int AnalogChannelCount { get; init; } = 4; + + /// Number of digital channels the scripted device reports. + public int DigitalChannelCount { get; init; } = 2; + + /// + /// Optional failure injector, keyed by 1-based initialization count, so a test can let the + /// first (pre-drop) initialization succeed and fail every later one. + /// + public Func? InitializeFailure { get; set; } + + public int InitializeCount => Volatile.Read(ref _initializeCount); + + public IReadOnlyList SentCommands + { + get + { + lock (_sent) + { + return _sent.ToArray(); + } + } + } + + public void ClearSentCommands() + { + lock (_sent) + { + _sent.Clear(); + } + } + + /// Feeds a streaming frame straight into the decode path. + public void InvokeStreamMessage(DaqifiOutMessage message) => OnStreamMessageReceived(message); + + public override void Send(IOutboundMessage message) + { + if (message is IOutboundMessage stringMessage) + { + lock (_sent) + { + _sent.Add(stringMessage.Data); + } + } + + // Still goes through the production Send. That is where a session driven by raw + // commands is tracked (#379), so a double that recorded the command and swallowed it + // would be testing a device that does not exist — which is exactly how a passing + // suite missed the bench failure. The transport's stream accepts writes and discards + // them, so nothing actually leaves. + base.Send(message); + } + + public override Task InitializeAsync( + TimeSpan? channelPopulationTimeout = null, + CancellationToken cancellationToken = default) + { + var attempt = Interlocked.Increment(ref _initializeCount); + + cancellationToken.ThrowIfCancellationRequested(); + + if (!IsConnected) + { + return Task.FromException(new DeviceNotConnectedException()); + } + + var failure = InitializeFailure?.Invoke(attempt); + if (failure != null) + { + return Task.FromException(failure); + } + + PopulateChannelsFromStatus(BuildStatusMessage()); + return Task.CompletedTask; + } + + /// + /// The status a device sends after a fresh boot: channels present, and — crucially — + /// analog_in_port_enabled reported as all-zero. A restored enable set therefore + /// cannot be an in-memory leftover; it has to have been re-applied. + /// + private DaqifiOutMessage BuildStatusMessage() + { + var status = new DaqifiOutMessage + { + AnalogInPortNum = (uint)AnalogChannelCount, + AnalogInRes = 65535, + DigitalPortNum = (uint)DigitalChannelCount, + AnalogInPortEnabled = ByteString.CopyFrom(new byte[] { 0x00, 0x00 }) + }; + + for (var i = 0; i < AnalogChannelCount; i++) + { + status.AnalogInPortRange.Add(1.0f); + status.AnalogInCalM.Add(1.0f); + status.AnalogInCalB.Add(0.0f); + status.AnalogInIntScaleM.Add(1.0f); + } + + return status; + } + } + + /// + /// A plain (non-streaming) device on a scripted transport, whose initialization can be parked + /// mid-flight. Deriving from rather than the streaming subclass is + /// the point: its session restore is the base no-op, so nothing in it re-checks the connection + /// on the loop's behalf, and the loop's own final guard is what has to catch a caller who + /// disconnected while initialization was running. + /// + private sealed class ScriptedBaseDevice : DaqifiDevice + { + private int _initializeCount; + + public ScriptedBaseDevice(string name, IStreamTransport transport) + : base(name, transport) + { + } + + /// Signalled once initialization is past its entry checks. + public ManualResetEventSlim InitializeEntered { get; } = new(false); + + /// Parks initialization after its entry checks until set. + public ManualResetEventSlim? InitializeGate { get; set; } + + public int InitializeCount => Volatile.Read(ref _initializeCount); + + public override void Send(IOutboundMessage message) + { + } + + public override Task InitializeAsync( + TimeSpan? channelPopulationTimeout = null, + CancellationToken cancellationToken = default) + { + Interlocked.Increment(ref _initializeCount); + cancellationToken.ThrowIfCancellationRequested(); + + if (!IsConnected) + { + return Task.FromException(new DeviceNotConnectedException()); + } + + // Entry checks are done; everything past here is the seconds of SCPI round-trips a real + // initialization spends, which is the window a caller's Disconnect lands in. + InitializeEntered.Set(); + InitializeGate?.Wait(GateTimeout); + + return Task.CompletedTask; + } + } + + /// + /// A transport that can be dropped on command and told to refuse the next N reconnects, so the + /// whole reconnect loop is reachable without hardware. Each connect hands out a fresh + /// stream, matching the real serial transport, whose BaseStream is a new instance after a + /// reopen. + /// + private sealed class ScriptedReconnectTransport : IStreamTransport, ITransportHealthSink + { + private readonly object _gate = new(); + private readonly TransportConnectionWatchdog? _watchdog; + private IdleStream _stream = new(); + private bool _isConnected; + private bool _disposed; + private int _connectFailuresRemaining; + private int _connectCount; + + public ScriptedReconnectTransport(bool useWatchdog = false) + { + if (useWatchdog) + { + _watchdog = new TransportConnectionWatchdog("Scripted transport", HandleConnectionLost); + } + } + + public int ConnectCount => Volatile.Read(ref _connectCount); + + public Stream Stream + { + get + { + lock (_gate) + { + ObjectDisposedException.ThrowIf(_disposed, this); + return _stream; + } + } + } + + public bool IsConnected + { + get + { + lock (_gate) + { + return _isConnected && !_disposed; + } + } + } + + public string ConnectionInfo => IsConnected ? "Scripted: Connected" : "Scripted: Disconnected"; + + public event EventHandler? StatusChanged; + + /// Refuses the next connect attempts. + public void FailNextConnects(int count) + { + lock (_gate) + { + _connectFailuresRemaining = count; + } + } + + /// + /// Makes reads start failing, as a pulled cable does, until the next successful connect + /// hands out a fresh stream. + /// + public void FailReadsUntilNextConnect() + { + lock (_gate) + { + _stream.FailReads = true; + } + } + + /// Reports a drop the way a transport's own detection would. + public void SimulateDrop() + { + lock (_gate) + { + _isConnected = false; + } + + _watchdog?.Disarm(); + StatusChanged?.Invoke( + this, + new TransportStatusEventArgs(false, ConnectionInfo, new IOException("the device went away"))); + } + + /// + /// Signalled as soon as a connect attempt begins, before it does any work. + /// + public ManualResetEventSlim ConnectEntered { get; } = new(false); + + private volatile ManualResetEventSlim? _connectGate; + + /// + /// Parks every subsequent connect attempt until the returned gate is set, so a test can + /// act — pull the device out from under the loop, say — while one is genuinely in flight. + /// Real connects block for as long as opening a port takes; this makes that window + /// controllable instead of a race to lose. + /// + public ManualResetEventSlim BlockConnects() + { + var gate = new ManualResetEventSlim(false); + _connectGate = gate; + return gate; + } + + private int _lifecycleDepth; + private volatile bool _sawConcurrentLifecycleCalls; + + /// + /// True if a connect and a disconnect were ever inside this transport at the same time. + /// A real transport is a serial port or a socket: opening one while closing it is + /// undefined, so the device must never do it. + /// + public bool SawConcurrentLifecycleCalls => _sawConcurrentLifecycleCalls; + + private void EnterLifecycle() + { + if (Interlocked.Increment(ref _lifecycleDepth) > 1) + { + _sawConcurrentLifecycleCalls = true; + } + } + + private void ExitLifecycle() => Interlocked.Decrement(ref _lifecycleDepth); + + public Task ConnectAsync() => ConnectAsync(null); + + public Task ConnectAsync(ConnectionRetryOptions? retryOptions) + { + Interlocked.Increment(ref _connectCount); + EnterLifecycle(); + try + { + ConnectCore(retryOptions); + } + finally + { + ExitLifecycle(); + } + + return Task.CompletedTask; + } + + private void ConnectCore(ConnectionRetryOptions? retryOptions) + { + // Outside the lock: a test holding the gate must still be able to inspect the + // transport and drive the device while the connect is parked here. + ConnectEntered.Set(); + _connectGate?.Wait(GateTimeout); + + lock (_gate) + { + ObjectDisposedException.ThrowIf(_disposed, this); + + if (_connectFailuresRemaining > 0) + { + _connectFailuresRemaining--; + throw new IOException("the device is not back yet"); + } + + // A reopened transport is a new stream; binding a consumer to the old one is + // exactly the bug the device's Disconnect/Connect cycle exists to avoid. + _stream.Dispose(); + _stream = new IdleStream(); + _isConnected = true; + } + + _watchdog?.Arm(); + StatusChanged?.Invoke(this, new TransportStatusEventArgs(true, ConnectionInfo)); + } + + public Task DisconnectAsync() + { + EnterLifecycle(); + try + { + // Disarm first: closing the handle is what makes in-flight reads fail, and none of + // that is a lost connection. + _watchdog?.Disarm(); + + lock (_gate) + { + _isConnected = false; + } + + StatusChanged?.Invoke(this, new TransportStatusEventArgs(false, ConnectionInfo)); + } + finally + { + ExitLifecycle(); + } + + return Task.CompletedTask; + } + + public void Connect() => ConnectAsync().GetAwaiter().GetResult(); + + public void Disconnect() => DisconnectAsync().GetAwaiter().GetResult(); + + public void ReportIoFault(Exception error) => _watchdog?.RecordFault(error); + + public void ReportIoSuccess() => _watchdog?.RecordSuccess(); + + private void HandleConnectionLost(Exception error) + { + lock (_gate) + { + _isConnected = false; + } + + StatusChanged?.Invoke(this, new TransportStatusEventArgs(false, ConnectionInfo, error)); + } + + public void Dispose() + { + lock (_gate) + { + if (_disposed) + { + return; + } + + _disposed = true; + _isConnected = false; + _stream.Dispose(); + } + + _watchdog?.Dispose(); + } + } + + /// + /// A stream that idles quietly — the shape a connected device with nothing to say presents to + /// the reader loop — until told to fail its reads. + /// + private sealed class IdleStream : Stream + { + public volatile bool FailReads; + + public override int Read(byte[] buffer, int offset, int count) + { + if (FailReads) + { + Thread.Sleep(5); + throw new IOException("the device is gone"); + } + + // Not a socket: a zero-byte read means "nothing yet", never a fault. + Thread.Sleep(20); + return 0; + } + + public override bool CanRead => true; + public override bool CanSeek => false; + public override bool CanWrite => true; + 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) + { + } + } + + #endregion +} diff --git a/src/Daqifi.Core.Tests/Device/ReconnectOptionsTests.cs b/src/Daqifi.Core.Tests/Device/ReconnectOptionsTests.cs new file mode 100644 index 0000000..64beeee --- /dev/null +++ b/src/Daqifi.Core.Tests/Device/ReconnectOptionsTests.cs @@ -0,0 +1,166 @@ +using Daqifi.Core.Device; + +namespace Daqifi.Core.Tests.Device; + +/// +/// The reconnect policy itself (issue #379): its defaults, its backoff arithmetic, and the values +/// it refuses. A policy that accepted "zero attempts" or a negative delay would turn into a device +/// that quietly never reconnects, or a loop that never waits. +/// +public class ReconnectOptionsTests +{ + [Fact] + public void ANewPolicyIsOff() + { + var options = new ReconnectOptions(); + + Assert.False(options.Enabled); + Assert.True(options.ResumeStreaming); + Assert.Equal(5, options.MaxAttempts); + } + + [Fact] + public void ThePresetsSayWhatTheyMean() + { + Assert.False(ReconnectOptions.Disabled.Enabled); + Assert.True(ReconnectOptions.Default.Enabled); + Assert.True(ReconnectOptions.Fast.Enabled); + Assert.True(ReconnectOptions.Resilient.Enabled); + + // Resilient keeps trying for far longer than Fast does. + Assert.True(ReconnectOptions.Resilient.MaxAttempts > ReconnectOptions.Fast.MaxAttempts); + Assert.True(ReconnectOptions.Resilient.MaxDelay > ReconnectOptions.Fast.MaxDelay); + } + + [Fact] + public void TheFirstAttemptStillWaits() + { + // Unlike the initial-connect retry policy: at the instant a drop is detected the endpoint is + // gone by definition, so trying immediately is guaranteed to fail. + var options = new ReconnectOptions { InitialDelay = TimeSpan.FromSeconds(2) }; + + Assert.Equal(TimeSpan.FromSeconds(2), options.CalculateDelay(1)); + } + + [Fact] + public void TheDelayBacksOffAndThenStopsGrowing() + { + var options = new ReconnectOptions + { + InitialDelay = TimeSpan.FromSeconds(1), + BackoffMultiplier = 2.0, + MaxDelay = TimeSpan.FromSeconds(5) + }; + + Assert.Equal(TimeSpan.FromSeconds(1), options.CalculateDelay(1)); + Assert.Equal(TimeSpan.FromSeconds(2), options.CalculateDelay(2)); + Assert.Equal(TimeSpan.FromSeconds(4), options.CalculateDelay(3)); + + // Capped from here on, however far the attempt count runs. + Assert.Equal(TimeSpan.FromSeconds(5), options.CalculateDelay(4)); + Assert.Equal(TimeSpan.FromSeconds(5), options.CalculateDelay(50)); + Assert.Equal(TimeSpan.FromSeconds(5), options.CalculateDelay(10_000)); + } + + [Fact] + public void AMultiplierOfOneGivesAFixedDelay() + { + var options = new ReconnectOptions + { + InitialDelay = TimeSpan.FromMilliseconds(750), + BackoffMultiplier = 1.0 + }; + + Assert.Equal(TimeSpan.FromMilliseconds(750), options.CalculateDelay(1)); + Assert.Equal(TimeSpan.FromMilliseconds(750), options.CalculateDelay(9)); + } + + [Fact] + public void AZeroInitialDelayStaysZero() + { + var options = new ReconnectOptions { InitialDelay = TimeSpan.Zero }; + + Assert.Equal(TimeSpan.Zero, options.CalculateDelay(1)); + Assert.Equal(TimeSpan.Zero, options.CalculateDelay(7)); + } + + [Theory] + [InlineData(64)] + [InlineData(1024)] + [InlineData(1075)] + [InlineData(4096)] + [InlineData(int.MaxValue)] + public void AZeroInitialDelayStaysZero_EvenWhereTheBackoffFactorOverflows(int attemptNumber) + { + // Regression: the exponential factor overflows to +Infinity past roughly attempt 1075 at a + // multiplier of 2, and 0 x Infinity is NaN. That NaN used to be answered with MaxDelay, so a + // policy configured for immediate retries silently became a 30-second wait — the opposite of + // what it asked for. + var options = new ReconnectOptions + { + InitialDelay = TimeSpan.Zero, + BackoffMultiplier = 2.0, + MaxDelay = TimeSpan.FromSeconds(30) + }; + + Assert.Equal(TimeSpan.Zero, options.CalculateDelay(attemptNumber)); + } + + [Fact] + public void AnOverflowingBackoffOnARealDelay_SettlesAtTheCap() + { + // The other side of the same overflow: with a positive InitialDelay the product is + // +Infinity rather than NaN, and the cap is the right answer. + var options = new ReconnectOptions + { + InitialDelay = TimeSpan.FromSeconds(1), + BackoffMultiplier = 2.0, + MaxDelay = TimeSpan.FromSeconds(30) + }; + + Assert.Equal(TimeSpan.FromSeconds(30), options.CalculateDelay(2000)); + Assert.Equal(TimeSpan.FromSeconds(30), options.CalculateDelay(int.MaxValue)); + } + + [Fact] + public void MaxDelayCapsTheFirstAttemptToo() + { + // A ceiling that exempted the very first wait would not be a ceiling. + var options = new ReconnectOptions + { + InitialDelay = TimeSpan.FromSeconds(10), + MaxDelay = TimeSpan.FromSeconds(2) + }; + + Assert.Equal(TimeSpan.FromSeconds(2), options.CalculateDelay(1)); + Assert.Equal(TimeSpan.FromSeconds(2), options.CalculateDelay(5)); + } + + [Theory] + [InlineData(0)] + [InlineData(-1)] + public void APolicyThatWouldNeverTry_IsRejected(int maxAttempts) + { + Assert.Throws( + () => new ReconnectOptions { MaxAttempts = maxAttempts }); + } + + [Fact] + public void NegativeDelaysAreRejected() + { + Assert.Throws( + () => new ReconnectOptions { InitialDelay = TimeSpan.FromSeconds(-1) }); + Assert.Throws( + () => new ReconnectOptions { MaxDelay = TimeSpan.FromSeconds(-1) }); + } + + [Theory] + [InlineData(0.5)] + [InlineData(0.0)] + [InlineData(double.NaN)] + public void ABackoffThatShrinksOrIsNotANumber_IsRejected(double multiplier) + { + Assert.Throws( + () => new ReconnectOptions { BackoffMultiplier = multiplier }); + } +} diff --git a/src/Daqifi.Core/Device/ConnectionStatus.cs b/src/Daqifi.Core/Device/ConnectionStatus.cs index 90ebb40..b434594 100644 --- a/src/Daqifi.Core/Device/ConnectionStatus.cs +++ b/src/Daqifi.Core/Device/ConnectionStatus.cs @@ -26,12 +26,18 @@ public enum ConnectionStatus Lost, /// - /// The device is retrying connection after a failure. + /// The device is retrying connection after a failure — in particular, it is between + /// automatic reconnect attempts after a drop (see ). Only + /// ever seen when reconnect is enabled; with the default policy a drop stops at + /// . /// Retrying, /// - /// The device connection failed after all retry attempts. + /// The device connection failed after all retry attempts — automatic reconnection ran out + /// of attempts and gave up (see ). Terminal: + /// nothing further will be attempted without a new Connect(). Only ever seen when + /// reconnect is enabled. /// Failed } diff --git a/src/Daqifi.Core/Device/DaqifiDevice.cs b/src/Daqifi.Core/Device/DaqifiDevice.cs index a817d50..00f25f6 100644 --- a/src/Daqifi.Core/Device/DaqifiDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiDevice.cs @@ -625,15 +625,375 @@ public DaqifiDevice(string name, IStreamTransport transport, ILogger? logger = n _transport.StatusChanged += OnTransportStatusChanged; } + #region Lifecycle serialization (issue #379) + + /// + /// Serializes connect against disconnect, on both the synchronous and the asynchronous + /// paths. + /// + /// + /// + /// Automatic reconnection (issue #379) introduced a second thread that opens and closes the + /// transport, and cancellation is not synchronization: + /// asks the loop to stop and returns immediately, but a loop already inside a blocking + /// Connect() cannot be interrupted and will run to completion. Without this, a + /// caller's could be opening and closing the same serial port + /// concurrently, and both threads could build and start a message consumer — leaving two + /// readers on one stream, the framing corruption this class refuses to risk anywhere else. + /// + /// + /// Narrow on purpose. This is an internal lifecycle invariant — the device never drives its + /// own transport from two threads at once — and deliberately not the general + /// per-device operation serialization of issue #342, which has to decide ordering across + /// the whole public API and interacts with _textExchangeLock. Nothing here changes + /// what any public method does when uncontended. + /// + /// + /// A semaphore rather than a monitor because and + /// hold it across await, which a monitor cannot do — + /// its continuation may resume on a different thread. Semaphores are not reentrant, so + /// re-entry is tracked separately by . + /// + /// + private readonly SemaphoreSlim _lifecycleLock = new(1, 1); + + /// + /// True while the current logical flow already holds . + /// + /// + /// Both connect and disconnect raise from inside their critical + /// section, and a consumer handler calling from there is re-entry + /// on the same flow — which runs nested today with no lock at all and must keep working + /// rather than deadlock against a non-reentrant semaphore. + /// rather than a thread id so it survives an await resuming on another thread, the + /// same technique _isInsideTextExchange already uses in this class. + /// + private readonly AsyncLocal _isInsideLifecycleOperation = new(); + + /// + /// How long waits for a lifecycle operation already in flight before + /// giving up. Overridable for tests, mirroring SdCardDownloadTimeout. + /// + internal virtual TimeSpan LifecycleLockTimeout => TimeSpan.FromSeconds(10); + + /// + /// How long waits for a lifecycle operation already in flight + /// before abandoning the wait. Far more generous than + /// because a teardown that gives up early is a teardown + /// that did not happen. Overridable for tests, mirroring SdCardDownloadTimeout. + /// + internal virtual TimeSpan TeardownLockTimeout => TimeSpan.FromSeconds(30); + + /// + /// What a lifecycle operation does when it cannot have to + /// itself. Running anyway is deliberately not an option: the whole point of the lock is + /// that two threads must never drive the transport at once, and a guarantee with a + /// "proceed regardless" branch is not a guarantee. + /// + private enum LifecycleContention + { + /// + /// Give up and throw rather than run alongside. For : nothing has + /// been opened, so failing costs the caller a retry and nothing else. + /// + Fail, + + /// + /// Give up and report it, leaving the operation in flight alone. For + /// : the caller is told nothing was torn down rather than being + /// blocked forever behind a holder that may never return. + /// + Abandon + } + + /// + /// The wait a contention policy allows, and what to do when it runs out. + /// + /// + /// + /// The two callers want opposite things from contention, so neither a shared timeout nor a + /// shared fallback would suit both. + /// + /// + /// fails. It waits and then + /// throws. Opening a second connection alongside one already in flight is exactly what this + /// lock exists to prevent — both threads would find no message consumer, both would build + /// and start one, and the loser's reader would be left running on the same stream, silently + /// corrupting frame boundaries for the rest of the session. A caller who gets a + /// instead has lost nothing: no handle was opened, no state + /// changed, and they can try again. + /// + /// + /// abandons. It waits — + /// far longer, because a teardown that gives up early is a teardown that did not happen — + /// and then reports that it did not run, leaving the holder alone. It must not throw + /// (Dispose depends on it) and must not run alongside (that is the corruption + /// above). An unbounded wait is not an option either: SerialPort.Open is called + /// synchronously with no timeout and can wedge in uncancellable native I/O — a hazard this + /// codebase already knows well enough to have built a process-wide port quarantine around + /// it in — so waiting on it forever would turn + /// Dispose into a permanent block. Abandoning the stuck operation is the house + /// answer to uncancellable native I/O here. + /// + /// + private TimeSpan ContentionWait(LifecycleContention onContention) => + onContention == LifecycleContention.Abandon ? TeardownLockTimeout : LifecycleLockTimeout; + + /// + /// Builds the failure for a connect that could not have the lock to itself. + /// + private TimeoutException LifecycleTimeout(TimeSpan timeout) + { + SafeLog(() => _logger.LogError( + "[Lifecycle] Device '{DeviceName}' could not take the connect/disconnect lock " + + "within {TimeoutSeconds}s; refusing to connect alongside the operation in flight.", + Name, + timeout.TotalSeconds)); + + return new TimeoutException( + $"Device '{Name}' could not start connecting within " + + $"{timeout.TotalSeconds:0.#}s because another connect or disconnect " + + "was still in progress. Nothing was opened; retry once it has finished."); + } + + /// Reports a teardown that gave up waiting for a stuck lifecycle operation. + private void LogAbandonedTeardown(TimeSpan timeout) => + SafeLog(() => _logger.LogError( + "[Lifecycle] Device '{DeviceName}' could not take the connect/disconnect lock " + + "within {TimeoutSeconds}s, so nothing was torn down. A connect is most likely " + + "wedged in uncancellable native I/O; it will release its own session when it " + + "returns.", + Name, + timeout.TotalSeconds)); + + /// + /// Runs a lifecycle operation under , never alongside another. + /// See for the per-policy semantics. + /// + /// true if the operation ran; false if the wait was abandoned. + /// + /// Thrown when is and + /// another lifecycle operation held the lock for the whole timeout. + /// + private bool RunLifecycleExclusive(Action operation, LifecycleContention onContention) + { + // Re-entry from inside the critical section (a StatusChanged handler calling back in) + // proceeds without acquiring, exactly as a reentrant monitor would. + if (_isInsideLifecycleOperation.Value) + { + operation(); + return true; + } + + var timeout = ContentionWait(onContention); + var acquired = false; + + try + { + acquired = _lifecycleLock.Wait(timeout); + } + catch (ObjectDisposedException) + { + // Disposed underneath us; there is nothing left to serialize against. + operation(); + return true; + } + + if (!acquired) + { + if (onContention == LifecycleContention.Abandon) + { + LogAbandonedTeardown(timeout); + return false; + } + + throw LifecycleTimeout(timeout); + } + + _isInsideLifecycleOperation.Value = true; + try + { + operation(); + return true; + } + finally + { + _isInsideLifecycleOperation.Value = false; + ReleaseLifecycleLock(); + } + } + + /// + private async Task RunLifecycleExclusiveAsync( + Func operation, + LifecycleContention onContention, + CancellationToken cancellationToken) + { + if (_isInsideLifecycleOperation.Value) + { + await operation().ConfigureAwait(false); + return true; + } + + var timeout = ContentionWait(onContention); + var isTeardown = onContention == LifecycleContention.Abandon; + var acquired = false; + + // A teardown's token is NOT allowed to govern this wait. Issue #341 defines what the + // token means for DisconnectAsync — it shortens the courtesy wait for an in-flight + // command exchange, never aborts the disconnect, and never surfaces as an + // OperationCanceledException — and this lock is a second, later wait that contract + // never covered. Passing the token here made a cancelled DisconnectAsync skip teardown + // altogether and report Disconnected with the transport still open and the message + // pumps still running; and because SemaphoreSlim throws for an already-cancelled token + // even when the semaphore is free, that happened on every cancelled disconnect, not + // just a contended one. The wait stays bounded by TeardownLockTimeout, which is what + // protects against a genuinely wedged holder (issue #379). The token still reaches + // AcquireTextExchangeLockForTeardownAsync inside the teardown, where it means what + // #341 says it means. + // + // The connect path is the opposite case and does honour the token: ConnectAsync is + // documented to be abandonable and to throw OperationCanceledException. + var acquireToken = isTeardown ? CancellationToken.None : cancellationToken; + + try + { + acquired = await _lifecycleLock.WaitAsync(timeout, acquireToken).ConfigureAwait(false); + } + catch (ObjectDisposedException) + { + await operation().ConfigureAwait(false); + return true; + } + + if (!acquired) + { + if (isTeardown) + { + LogAbandonedTeardown(timeout); + return false; + } + + throw LifecycleTimeout(timeout); + } + + _isInsideLifecycleOperation.Value = true; + try + { + await operation().ConfigureAwait(false); + return true; + } + finally + { + _isInsideLifecycleOperation.Value = false; + ReleaseLifecycleLock(); + } + } + + private void ReleaseLifecycleLock() + { + try + { + _lifecycleLock.Release(); + } + catch (ObjectDisposedException) + { + // Raced a Dispose that already tore the semaphore down. + } + } + + #endregion + /// /// Connects to the device. /// /// + /// /// Opens the transport on the calling thread. is the /// non-blocking, cancellable equivalent and is preferred on a UI thread; this overload is /// kept for existing callers and behaves exactly as it always has. + /// + /// + /// A caller-issued connect supersedes any automatic reconnect in progress: the loop is + /// cancelled and unwinds without touching the session this call establishes. Cancelling it + /// does not stop it instantly — an attempt already inside a blocking transport connect runs + /// to completion — so this waits for any connect or disconnect in flight rather than + /// running alongside it, and throws if one is still in flight after + /// . Nothing has been opened in that case, so the call is + /// safe to retry. + /// /// + /// + /// Thrown when another connect or disconnect was still in progress after + /// . Nothing was opened. + /// public void Connect() + { + _callerWantsDisconnected = false; + SupersedeReconnect(); + ConnectCore(); + } + + /// + /// The body of , without the reconnect-supersede step — the reconnect + /// loop calls this so it does not cancel itself. + /// + private void ConnectCore() + { + RunLifecycleExclusive(ConnectCoreUnsynchronized, LifecycleContention.Fail); + HonourTeardownRaisedDuringConnect(); + } + + /// + private async Task ConnectCoreAsync(CancellationToken cancellationToken) + { + await RunLifecycleExclusiveAsync( + () => ConnectCoreUnsynchronizedAsync(cancellationToken), + LifecycleContention.Fail, + cancellationToken).ConfigureAwait(false); + + HonourTeardownRaisedDuringConnect(); + } + + /// + /// Closes a connection this call established if a teardown landed while it was in flight. + /// + /// + /// + /// Matters in the one case where the caller's could not do it + /// itself: a connect wedged in uncancellable native I/O holds the lifecycle lock long enough + /// for the teardown to abandon its wait, so the teardown returns having deliberately left + /// the transport alone — and whatever this connect goes on to build would otherwise be live, + /// with a reader running, after the caller was told the device was disconnected. + /// + /// + /// Shared by both connect entry points rather than living in the reconnect loop: + /// covers the same ground for a reconnect attempt, but it + /// is part of the loop and never runs for a caller's own connect. + /// + /// + /// The ordering is not a race: sets the flag before it + /// contends for the lock, and this reads it after releasing it. A teardown that + /// abandoned must therefore have been waiting while the connect still held the lock, so its + /// write always happens-before this read. + /// + /// + private void HonourTeardownRaisedDuringConnect() + { + if (!_callerWantsDisconnected && !_disposed) + { + return; + } + + SafeLog(() => _logger.LogWarning( + "[Lifecycle] Device '{DeviceName}' was disconnected while this connect was in flight; " + + "closing the connection it established.", + Name)); + + DisconnectCore(ConnectionStatus.Disconnected); + } + + private void ConnectCoreUnsynchronized() { BeginConnect(); @@ -666,10 +1026,21 @@ public void Connect() /// A cancellation token to observe while connecting. /// A task representing the asynchronous connect operation. /// Thrown when the attempt is canceled. + /// + /// Thrown when another connect or disconnect was still in progress after + /// . Nothing was opened. + /// // Deliberately not virtual, matching Connect(). A virtual async twin of a non-virtual sync // method is a trap: a subclass would override this one, leave Connect() unintercepted, and // get different behavior depending on which entry point the caller reached for. public async Task ConnectAsync(CancellationToken cancellationToken = default) + { + _callerWantsDisconnected = false; + SupersedeReconnect(); + await ConnectCoreAsync(cancellationToken).ConfigureAwait(false); + } + + private async Task ConnectCoreUnsynchronizedAsync(CancellationToken cancellationToken) { cancellationToken.ThrowIfCancellationRequested(); @@ -823,8 +1194,94 @@ private async Task SafeDisconnectTransportAsync() /// sees _isDisconnecting == true via the post-acquisition /// validation and bails out cleanly. Callers wanting a non-blocking /// disconnect should use . + /// + /// Also waits for any connect or disconnect already in flight — an automatic reconnect + /// attempt parked inside a blocking transport connect, typically — rather than tearing down + /// alongside it. That wait is bounded by , generously, + /// because SerialPort.Open is uncancellable and can wedge indefinitely; waiting on + /// it without a bound would make this method — and therefore — hang + /// forever. If the wait is abandoned, the device still reports + /// , but the transport is deliberately left to + /// the stuck operation, which releases it when it finally returns. That outcome is logged + /// at error level: it is a rare safety fallback rather than routine, it means a port + /// is wedged and the transport was not released on the caller's schedule, and it is exactly + /// the line an operator needs to still be there after log filtering. This never throws on + /// contention. + /// /// public void Disconnect() + { + // A user-issued teardown always beats an automatic reconnect: record the intent and + // stop the loop before tearing anything down. Setting the flag first matters — a + // reconnect already inside a blocking Connect() will finish it regardless, and this is + // what tells it to put the session straight back down instead of leaving the device + // quietly alive behind the caller's back. + _callerWantsDisconnected = true; + SupersedeReconnect(); + DisconnectCore(ConnectionStatus.Disconnected); + } + + /// + /// The body of , parameterized by the status the device settles on. + /// + /// + /// The status to report once teardown is done. + /// for a caller-issued disconnect; for the teardown + /// the reconnect loop performs between attempts, which must not look to consumers like the + /// session ended on purpose. + /// + private void DisconnectCore(ConnectionStatus finalStatus) + { + if (RunLifecycleExclusive( + () => DisconnectCoreUnsynchronized(finalStatus), + LifecycleContention.Abandon)) + { + return; + } + + MarkDisconnectedWithoutTeardown(finalStatus); + } + + /// + private async Task DisconnectCoreAsync(ConnectionStatus finalStatus, CancellationToken cancellationToken) + { + if (await RunLifecycleExclusiveAsync( + () => DisconnectCoreUnsynchronizedAsync(finalStatus, cancellationToken), + LifecycleContention.Abandon, + cancellationToken).ConfigureAwait(false)) + { + return; + } + + MarkDisconnectedWithoutTeardown(finalStatus); + } + + /// + /// Settles the device's reported state when teardown had to be abandoned. + /// + /// + /// The wait was abandoned because a lifecycle operation is stuck, most likely a + /// SerialPort.Open wedged in uncancellable native I/O. Racing it would be the stream + /// corruption the lifecycle lock exists to prevent, so the transport is left to the holder — + /// which releases it once it unwedges, because was + /// set before this wait began and every connect path re-reads it after dropping the lock: + /// for a caller's own connect, + /// for a reconnect attempt. + /// + /// What can still be done safely is record the caller's intent at the device level. These + /// are this class's own fields, not the transport, so setting them cannot corrupt anything + /// the stuck operation is doing — and without them the device would keep reporting itself + /// connected after the caller had asked it not to. + /// + /// + private void MarkDisconnectedWithoutTeardown(ConnectionStatus finalStatus) + { + State = DeviceState.Disconnected; + _isInitialized = false; + Status = finalStatus; + } + + private void DisconnectCoreUnsynchronized(ConnectionStatus finalStatus) { _isDisconnecting = true; var lockAcquired = AcquireTextExchangeLockForTeardown(); @@ -838,7 +1295,7 @@ public void Disconnect() } finally { - FinishDisconnect(lockAcquired); + FinishDisconnect(lockAcquired, finalStatus); } } @@ -870,6 +1327,16 @@ public void Disconnect() /// A task representing the asynchronous disconnect operation. // Not virtual, for the same reason as ConnectAsync. public async Task DisconnectAsync(CancellationToken cancellationToken = default) + { + // Same intent-first ordering as Disconnect: see the comment there. + _callerWantsDisconnected = true; + SupersedeReconnect(); + await DisconnectCoreAsync(ConnectionStatus.Disconnected, cancellationToken).ConfigureAwait(false); + } + + private async Task DisconnectCoreUnsynchronizedAsync( + ConnectionStatus finalStatus, + CancellationToken cancellationToken) { _isDisconnecting = true; var lockAcquired = await AcquireTextExchangeLockForTeardownAsync(cancellationToken) @@ -890,7 +1357,7 @@ public async Task DisconnectAsync(CancellationToken cancellationToken = default) } finally { - FinishDisconnect(lockAcquired); + FinishDisconnect(lockAcquired, finalStatus); } } @@ -981,9 +1448,15 @@ private void StopMessagePumps() /// never be left reporting Connected because teardown threw. /// /// Whether the text-exchange lock was acquired before teardown. - private void FinishDisconnect(bool lockAcquired) + /// + /// The status to settle on. Parameterized for the reconnect loop, whose teardown between + /// attempts reports rather than + /// — nobody asked for that teardown, and it + /// must not look to consumers like the session ended on purpose (issue #379). + /// + private void FinishDisconnect(bool lockAcquired, ConnectionStatus finalStatus) { - Status = ConnectionStatus.Disconnected; + Status = finalStatus; State = DeviceState.Disconnected; _isInitialized = false; _isDisconnecting = false; @@ -1860,11 +2333,542 @@ private void OnTransportStatusChanged(object? sender, TransportStatusEventArgs e // not during an intentional Disconnect() call if (Status == ConnectionStatus.Connected && !_isDisconnecting) { + // Read before raising Lost: a handler is free to call Connect or Disconnect + // synchronously from inside it, and if it does, this drop must not start a + // reconnect for a session the caller has already moved on from. + var epochAtDrop = Volatile.Read(ref _sessionEpoch); + + // Snapshot the session BEFORE anything observes the drop. Raising Lost runs + // consumer handlers synchronously on this thread, and a handler is entirely + // entitled to start tearing the device down — by which point what the session + // looked like is no longer recoverable. No-op unless reconnect is enabled, so + // the default path does exactly what it did before (issue #379). + // + // Skipped while a reconnect is already running: a drop during an attempt would + // otherwise overwrite the snapshot of the session being restored with the empty + // half-built one, and the loop would go on to restore nothing. + if (ReconnectOptions.Enabled && !IsReconnecting) + { + try + { + CaptureSessionSnapshot(); + } + catch (Exception ex) + { + SafeLog(() => _logger.LogWarning( + ex, "[Reconnect] Capturing the session state after a drop failed; the session cannot be restored.")); + } + } + Status = ConnectionStatus.Lost; + + BeginReconnectIfEnabled(epochAtDrop); } } } + #region Automatic reconnection (issue #379) + + private ReconnectOptions _reconnectOptions = new(); + + /// + /// Gets or sets the policy for re-establishing a session after + /// . Disabled by default: a drop is reported and nothing + /// else happens, exactly as it always has been. + /// + /// + /// + /// With set, a detected drop starts a background + /// loop that reconnects the same endpoint, re-runs , + /// and restores the session state Core owns — the channel enable set, the streaming + /// frequency, and an interrupted stream. Progress arrives on + /// / / and as + /// between attempts. + /// + /// + /// Deliberately not restored: anything the device owns rather than Core (DIO + /// directions and output levels, PWM state, analog outputs, calibration written to RAM), an + /// SD logging session, and any in-flight operation — an SD download interrupted by a drop + /// fails, and is not resumed or retried. + /// + /// + /// Same-endpoint only. A serial device that comes back on a different port path, or a + /// device whose IP address changed, is a new endpoint and needs a fresh + /// DaqifiDeviceFactory connect. Failing over to a different transport is out of + /// scope entirely. + /// + /// + /// Thrown when set to null. + public ReconnectOptions ReconnectOptions + { + get => _reconnectOptions; + set + { + ArgumentNullException.ThrowIfNull(value); + _reconnectOptions = value; + } + } + + /// + /// Occurs before each automatic reconnect attempt, carrying the attempt number and the + /// backoff wait that precedes it. + /// + /// + /// Raised on a background thread, like every event in this group. Handlers should do the + /// minimum; one that throws is caught and ignored. + /// + public event EventHandler? ReconnectAttempt; + + /// + /// Occurs once an automatic reconnect has restored the session — the device is connected, + /// re-initialized, its channel configuration re-applied, and an interrupted stream running + /// again. + /// + public event EventHandler? Reconnected; + + /// + /// Occurs when automatic reconnection stops without restoring the session, because it ran + /// out of attempts or was cancelled. + /// + /// + /// Running out of attempts is also raised on with + /// and leaves the device on + /// , so giving up is impossible to miss even with + /// nothing subscribed here. Cancellation is not an error and does neither. + /// + public event EventHandler? ReconnectFailed; + + // 0 = idle, 1 = a reconnect loop is running. Guards against a second loop being started by + // the Lost that a failing attempt's own teardown can produce. + private int _reconnectRunning; + + // Volatile: written by the thread that starts a loop and by the loop's own cleanup, read by + // CancelReconnect from any thread. A stale read only ever delays a cancellation, never + // corrupts one — the epoch check is what actually stops the loop — but there is no reason + // to leave even that on the table. + private volatile CancellationTokenSource? _reconnectCts; + + // Bumped by every caller-issued Connect/Disconnect/Dispose. The reconnect loop captures it + // when it starts and re-checks before each step that touches device state; a change means + // the caller has moved on and the loop must unwind. This is what keeps the loop from + // re-opening a transport the caller just closed, without Disconnect() having to block on a + // loop that may be calling back into it. + private int _sessionEpoch; + + // Which way the caller last pointed the device. The epoch says *that* a caller superseded + // the loop; this says *what they wanted*, which is what decides how the loop unwinds. + // Connect() is a blocking, multi-second operation, so a caller can always land in the + // middle of one — no amount of checking beforehand avoids that, and by the time the loop + // looks again it may be holding a session it has just brought up. If the caller wants the + // device down, that session is the loop's own doing and has to go back down with it; + // if the caller wants it up, it is theirs and must be left strictly alone. + private volatile bool _callerWantsDisconnected; + + /// + /// Gets a value indicating whether an automatic reconnect is currently in progress. + /// + public bool IsReconnecting => Volatile.Read(ref _reconnectRunning) != 0; + + /// + /// Stops any automatic reconnect in progress. Safe to call at any time, including when + /// nothing is reconnecting. + /// + /// + /// The loop unwinds at its next checkpoint — it does not interrupt a connect attempt + /// already in flight — and reports with + /// set. The device is left on + /// : the connection really is gone, and nothing is + /// trying to bring it back. This returns immediately rather than waiting for the loop, so + /// it is safe to call from a device event handler. + /// + public void CancelReconnect() + { + try + { + _reconnectCts?.Cancel(); + } + catch (ObjectDisposedException) + { + // The loop finished and disposed its own token source. Nothing to cancel. + } + } + + /// + /// Cancels any reconnect in progress and declares the session it was rebuilding + /// obsolete, so the unwinding loop leaves the caller's new session strictly alone. + /// + private void SupersedeReconnect() + { + Interlocked.Increment(ref _sessionEpoch); + CancelReconnect(); + } + + /// + /// Starts the reconnect loop on a background thread, if the policy allows one and none is + /// already running. + /// + /// + /// Called from the transport's status callback, which runs on the reader loop or the + /// liveness timer — so the work is handed to the thread pool rather than done inline. + /// + /// + /// The session epoch observed before the drop was announced. A mismatch means a caller + /// connected or disconnected in the meantime — including from inside their own + /// handler — and this drop is no longer theirs to recover from. + /// + private void BeginReconnectIfEnabled(int expectedEpoch) + { + if (!ReconnectOptions.Enabled || _transport == null || _disposed || _isDisconnecting) + { + return; + } + + // Only ever start from a device that is actually sitting on a lost connection with the + // session it was lost from still current. + if (Status != ConnectionStatus.Lost || Volatile.Read(ref _sessionEpoch) != expectedEpoch) + { + return; + } + + // One loop at a time. A failing attempt tears the transport down again, which can + // produce another Lost; without this that would fork a second loop racing the first. + if (Interlocked.CompareExchange(ref _reconnectRunning, 1, 0) != 0) + { + return; + } + + var cts = new CancellationTokenSource(); + _reconnectCts = cts; + + var epoch = Volatile.Read(ref _sessionEpoch); + var options = ReconnectOptions; + + _ = Task.Run(async () => + { + var wasCanceled = false; + + try + { + await RunReconnectLoopAsync(options, epoch, cts.Token).ConfigureAwait(false); + } + catch (Exception ex) + { + // The loop handles its own failures; anything escaping is a bug in it, and must + // not become an unobserved task exception. + SafeLog(() => _logger.LogError(ex, "[Reconnect] The reconnect loop terminated unexpectedly.")); + } + finally + { + wasCanceled = cts.IsCancellationRequested; + + // Clear the token source before releasing the running flag, so a loop started + // by the next drop cannot have its own source nulled out from under it. + _reconnectCts = null; + cts.Dispose(); + Interlocked.Exchange(ref _reconnectRunning, 0); + } + + // A drop that landed in the moments between this loop finishing its work and + // releasing the running flag would have been skipped by the single-flight guard, + // stranding the device on Lost with nothing trying to bring it back. Pick it up. + // Exhausting the attempts settles on Failed, and cancellation is excluded here, so + // neither can retrigger. + if (!wasCanceled) + { + BeginReconnectIfEnabled(Volatile.Read(ref _sessionEpoch)); + } + }); + } + + /// + /// Attempts, with backoff, to rebuild the session that was just lost. + /// + /// + /// The policy as it stood when the drop was detected. Assigning a new + /// mid-flight therefore leaves this loop on the terms it + /// started under and applies from the next drop; mutating the instance already assigned + /// does reach it, since the loop holds that same object. + /// + /// The session epoch at the time of the drop. + /// Cancelled by . + private async Task RunReconnectLoopAsync( + ReconnectOptions options, + int epoch, + CancellationToken cancellationToken) + { + var startedAt = DateTime.UtcNow; + Exception? lastError = null; + var attempt = 0; + + SafeLog(() => _logger.LogInformation( + "[Reconnect] Device '{DeviceName}' lost its connection; reconnecting (up to {MaxAttempts} attempt(s)).", + Name, + options.MaxAttempts)); + + while (attempt < options.MaxAttempts) + { + attempt++; + + if (IsSessionStale(epoch) || cancellationToken.IsCancellationRequested) + { + ReportReconnectStopped(epoch, attempt - 1, lastError, wasCanceled: true); + return; + } + + var delay = options.CalculateDelay(attempt); + RaiseReconnectEvent(ReconnectAttempt, new ReconnectAttemptEventArgs( + attempt, options.MaxAttempts, delay, lastError), nameof(ReconnectAttempt)); + + try + { + // Tear down what is left of the dead session first: the producer and consumer + // are still bound to a stream that is gone, and Connect() only rebuilds them + // once they have been nulled. Reported as Retrying, not Disconnected — nobody + // asked for this teardown. + DisconnectCore(ConnectionStatus.Retrying); + + if (delay > TimeSpan.Zero) + { + await Task.Delay(delay, cancellationToken).ConfigureAwait(false); + } + + if (IsSessionStale(epoch)) + { + ReportReconnectStopped(epoch, attempt, lastError, wasCanceled: true); + return; + } + + cancellationToken.ThrowIfCancellationRequested(); + + ConnectCore(); + + // Connect() blocks for as long as opening the port takes, so a caller can — and + // on a slow serial open, will — land squarely in the middle of it. There is now + // a live transport and a running reader that this loop built, so bailing out + // here is not enough on its own: if the caller wants the device down, the + // session has to be taken back down with it. + if (AbandonIfSuperseded(epoch, attempt, lastError)) + { + return; + } + + await InitializeAsync(cancellationToken: cancellationToken).ConfigureAwait(false); + + var streamingResumed = await RestoreSessionSnapshotAsync( + options, cancellationToken).ConfigureAwait(false); + + // Same again before declaring victory: initialization and restore are several + // seconds of SCPI round-trips, and a session the caller has since disowned must + // not be handed to them as a successful reconnect. + if (AbandonIfSuperseded(epoch, attempt, lastError)) + { + return; + } + + SafeLog(() => _logger.LogInformation( + "[Reconnect] Device '{DeviceName}' reconnected on attempt {Attempt}.", Name, attempt)); + + RaiseReconnectEvent(Reconnected, new ReconnectedEventArgs( + attempt, DateTime.UtcNow - startedAt, streamingResumed), nameof(Reconnected)); + return; + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + ReportReconnectStopped(epoch, attempt, lastError, wasCanceled: true); + return; + } + catch (Exception ex) + { + lastError = ex; + SafeLog(() => _logger.LogWarning( + ex, + "[Reconnect] Attempt {Attempt} of {MaxAttempts} to reconnect device '{DeviceName}' failed.", + attempt, + options.MaxAttempts, + Name)); + } + } + + ReportReconnectStopped(epoch, attempt, lastError, wasCanceled: false); + } + + /// + /// Checks whether a caller has superseded the session this loop is rebuilding and, if so, + /// unwinds whatever the loop has already brought up before reporting that it stopped. + /// + /// + /// + /// Called at the points where the loop is holding a live session of its own making — after + /// , and again once initialization and restore are done. Both are + /// preceded by long blocking work, which is exactly when a caller's + /// lands, so a check beforehand cannot substitute for this one. + /// + /// + /// What "unwind" means depends on what the caller wanted, which is why the epoch alone is + /// not enough to decide it. After a or a disposal the live session + /// is the loop's own doing and is torn straight back down — leaving it up is a device the + /// caller closed quietly coming back to life. After a caller's own + /// the session belongs to them, and tearing it down would be the same bug in reverse, so it + /// is left alone and only logged. + /// + /// + /// true if the loop must stop. + private bool AbandonIfSuperseded(int epoch, int attempt, Exception? lastError) + { + if (!IsSessionStale(epoch)) + { + return false; + } + + if (_callerWantsDisconnected || _disposed) + { + SafeLog(() => _logger.LogInformation( + "[Reconnect] Device '{DeviceName}' was disconnected while a reconnect attempt was in flight; " + + "closing the connection the attempt had established.", + Name)); + + try + { + DisconnectCore(ConnectionStatus.Disconnected); + } + catch (Exception ex) + { + // Best-effort unwind of an abandoned attempt. A transport already disposed out + // from under us can throw here, and there is nothing left to salvage by + // letting that escape into the loop's retry logic. + SafeLog(() => _logger.LogDebug( + ex, "[Reconnect] Closing the abandoned reconnect attempt's connection failed.")); + } + } + else + { + SafeLog(() => _logger.LogWarning( + "[Reconnect] Device '{DeviceName}' was reconnected by its caller while an automatic " + + "reconnect attempt was in flight; leaving the caller's connection alone.", + Name)); + } + + ReportReconnectStopped(epoch, attempt, lastError, wasCanceled: true); + return true; + } + + /// + /// True once a caller-issued connect, disconnect or disposal has superseded the session the + /// reconnect loop was started for. + /// + /// + /// Deliberately does not consult _isDisconnecting: the loop's own teardown between + /// attempts sets that flag, so reading it here would make the loop consider itself stale. + /// A caller-issued bumps the epoch before it sets the flag, which + /// is what this actually needs to see. + /// + private bool IsSessionStale(int epoch) => + _disposed || Volatile.Read(ref _sessionEpoch) != epoch; + + /// + /// Settles the device after a reconnect loop that did not restore the session, and reports + /// why. + /// + /// + /// Exhausting the attempts is terminal and loud: , a + /// logged error, and an raise. Being cancelled is neither — the + /// caller asked for it — so the device is simply left reporting + /// , which is the truth. Neither touches + /// at all once the session is stale, since by then the status belongs + /// to whatever the caller did next. + /// + private void ReportReconnectStopped(int epoch, int attemptsMade, Exception? lastError, bool wasCanceled) + { + if (!IsSessionStale(epoch)) + { + // An attempt can fail after the transport is back up (a re-initialization that + // times out, say), so tear the half-built session down rather than leaving a live + // handle and a running reader behind a terminal status. + DisconnectCore(wasCanceled ? ConnectionStatus.Lost : ConnectionStatus.Failed); + } + + if (wasCanceled) + { + SafeLog(() => _logger.LogInformation( + "[Reconnect] Reconnection of device '{DeviceName}' was cancelled after {AttemptsMade} attempt(s).", + Name, + attemptsMade)); + } + else + { + SafeLog(() => _logger.LogError( + lastError, + "[Reconnect] Device '{DeviceName}' could not be reconnected after {AttemptsMade} attempt(s); giving up.", + Name, + attemptsMade)); + + // Terminal failure has to be impossible to miss, so it goes to the device error + // surface as well as to this group's own event (issue #379 / #378). + RaiseDeviceError( + DeviceErrorSource.Reconnect, + new DeviceReconnectFailedException(Name, attemptsMade, lastError)); + } + + RaiseReconnectEvent( + ReconnectFailed, + new ReconnectFailedEventArgs(attemptsMade, lastError, wasCanceled), + nameof(ReconnectFailed)); + } + + /// + /// Raises one of the reconnect events, isolating the loop from a subscriber that throws — + /// the same guarantee and SendFailed give. + /// + private void RaiseReconnectEvent( + EventHandler? handler, + TArgs args, + string eventName) + where TArgs : EventArgs + { + if (handler == null) + { + return; + } + + try + { + handler(this, args); + } + catch (Exception ex) + { + SafeLog(() => _logger.LogWarning(ex, "[{EventName}] subscriber threw", eventName)); + } + } + + /// + /// When overridden in a derived class, records the session state that a reconnect should + /// restore. Called on the thread that detected the drop, before anything else observes it, + /// and only when reconnect is enabled. + /// + /// + /// The base device owns nothing session-shaped — its channel collection is repopulated from + /// the device's own status message on every connect — so this does nothing here. + /// overrides it to record which channels were enabled + /// and whether a stream was running. + /// + protected virtual void CaptureSessionSnapshot() + { + } + + /// + /// When overridden in a derived class, re-applies the state recorded by + /// to a device that has just been reconnected and + /// re-initialized. + /// + /// The policy governing this reconnect. + /// Cancelled if reconnection is cancelled. + /// true if an interrupted stream was restarted. + protected virtual Task RestoreSessionSnapshotAsync( + ReconnectOptions options, + CancellationToken cancellationToken) => Task.FromResult(false); + + #endregion + /// /// Logs a message that the producer's background thread could not deliver. /// diff --git a/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs b/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs index c9ed15a..241a235 100644 --- a/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs +++ b/src/Daqifi.Core/Device/DaqifiStreamingDevice.cs @@ -389,6 +389,25 @@ public void StartStreaming() if (IsStreaming) return; + BeginStreamingSession(); + + IsStreaming = true; + Send(ScpiMessageProducer.StartStreaming(StreamingFrequency)); + } + + /// + /// Resets everything that is scoped to one streaming session, so the frames that follow are + /// decoded against this session rather than the last one. + /// + /// + /// Shared by and by the tracking of a start-streaming command + /// sent directly through . It is one method precisely so the two cannot + /// drift: a raw-started stream that skipped this would reconstruct timestamps from the + /// previous session's anchor and re-use its gap detector, producing samples stamped with + /// times that never happened — silently. + /// + private void BeginStreamingSession() + { // Re-anchor per-session timestamp reconstruction: the first frame of this session // anchors to the current host time, and subsequent frames advance by the device-tick // delta. Apply the device-reported tick frequency (falls back to the 50 MHz default @@ -406,9 +425,6 @@ public void StartStreaming() _awaitingFirstFullAnalogFrame = CountEnabledAnalogChannels(SnapshotChannels()) > 0; _suppressedWarmupFrameCount = 0; Interlocked.Exchange(ref _decodeFailureCount, 0); - - IsStreaming = true; - Send(ScpiMessageProducer.StartStreaming(StreamingFrequency)); } /// @@ -428,6 +444,350 @@ public void StopStreaming() Send(ScpiMessageProducer.StopStreaming); } + #region Session state tracking for commands sent directly (issue #379) + + /// The command emits. + private const string StartStreamingCommand = "SYSTem:StartStreamData"; + + /// The command emits. + private const string StopStreamingCommand = "SYSTem:StopStreamData"; + + /// The command emits. + private const string EnableAdcChannelsCommand = "ENAble:VOLTage:DC"; + + /// + /// Sends a command, and keeps this device's view of the streaming session in step with it. + /// + /// + /// + /// is public and is a perfectly ordinary way to drive a device — the + /// example CLI does the whole job that way — but a session driven through it used to be + /// completely invisible to this class: stayed false while + /// data poured in, and the enabled-channel set stayed empty. That mattered once reconnect + /// arrived (issue #379). A physical cable pull on the bench recovered the link and then + /// reported StreamingResumed: false, because as far as Core was concerned nothing had + /// ever been streaming — while re-initialization had in fact just stopped the stream that + /// was running. The session looked restored and was not. + /// + /// + /// So the two commands that define a streaming session are recognized here regardless of + /// which API produced them, and the same state is updated that + /// / and + /// would have set. This is the same principle as #409, where + /// analog IsEnabled is resynced from the device's own reported mask: what Core + /// believes about a session has to track what is actually true of it. + /// + /// + /// Only these commands are interpreted, and only after the send itself has succeeded. + /// Everything else passes through untouched. + /// + /// + /// What equivalence with the typed API covers. Each typed method was walked and every + /// effect beyond setting a flag accounted for, so this is a stated scope rather than a + /// hopeful one: + /// + /// + /// + /// — the whole per-session reset (timestamp anchor and tick + /// frequency, gap detector, warmup guard and its counter, decode-failure count) is shared + /// through and runs here too. Skipping it was a real + /// defect: frames decoded against a previous session's anchor carry times that never + /// happened. + /// + /// + /// — does nothing beyond clearing the flag, so clearing it here + /// is complete. + /// + /// + /// — assigns under the + /// channels lock and derives the outbound mask from it. The mask has already been sent by + /// the time this runs, so only the assignment is replayed, and it is applied to every + /// analog channel because the firmware treats the mask as a set-replace. + /// + /// + /// + /// Deliberately outside it. The global DIO enable is one switch for the whole port + /// rather than a per-channel mask, so a raw DIO:PORt:ENAble carries no information + /// about which digital channels were wanted and none is inferred. Argument validation + /// is not replayed either: by the time a command is seen here it has already gone to the + /// device, and the device is the authority on whether it accepted it. + /// + /// + /// Running after the send is safe rather than merely convenient. A string command is handed + /// to the background producer, so it has not reached the wire when this runs, and the device + /// cannot answer a command it has not received; on the producer-less path that writes + /// synchronously there is no message consumer decoding frames at all. Either way no frame + /// can be decoded between the send and the state it should be decoded against. + /// + /// + /// The type of the message data payload. + /// The message to send. + public override void Send(IOutboundMessage message) + { + base.Send(message); + + // Only after the send has actually gone through: a command that threw never reached + // the device and must not move this device's idea of the session. + if (message is IOutboundMessage textCommand) + { + TrackSessionCommand(textCommand.Data); + } + } + + /// + /// Updates the streaming-session view from a command that has just been sent. + /// + private void TrackSessionCommand(string? command) + { + if (string.IsNullOrWhiteSpace(command)) + { + return; + } + + var trimmed = command.Trim(); + + if (trimmed.StartsWith(StopStreamingCommand, StringComparison.OrdinalIgnoreCase)) + { + IsStreaming = false; + return; + } + + if (trimmed.StartsWith(StartStreamingCommand, StringComparison.OrdinalIgnoreCase)) + { + TrackStreamingStart(trimmed.AsSpan(StartStreamingCommand.Length)); + return; + } + + if (trimmed.StartsWith(EnableAdcChannelsCommand, StringComparison.OrdinalIgnoreCase)) + { + TrackAdcEnableMask(trimmed.AsSpan(EnableAdcChannelsCommand.Length)); + } + } + + /// + /// Records a start-streaming command, but only one carrying a rate this device can model. + /// + /// + /// + /// A command whose argument is missing, unparseable, or outside the device's sampling range + /// is not treated as the start of a session. Marking one as streaming anyway would be + /// wrong three times over: the firmware rejects such a command and does not start streaming, + /// so the flag would not describe the device; would be left + /// holding a rate from some earlier session, which a reconnect would then faithfully restore + /// — resuming at a rate nobody asked for is the silent-wrong-data failure this whole feature + /// exists to prevent; and a stale makes the next legitimate + /// a silent no-op, which is the same stale-flag trap that + /// issue #118 and the defensive stops scattered through the SD paths already guard against. + /// + /// + /// The existing state is left alone rather than cleared. A device already streaming at a + /// good rate goes on doing exactly that when the firmware rejects a malformed start, so + /// and both remain true of it; + /// forcing them off would swap one inaccuracy for another. + /// + /// + /// Because of this, is never true alongside a rate that was + /// not validated — so session restore has no "streaming at an unknown rate" case to decide + /// what to do about. The state it replays is always one that was really commanded. + /// + /// + private void TrackStreamingStart(ReadOnlySpan argument) + { + var rate = argument.Trim(); + + // One read of the ceiling, used for both the check and the assignment below. The public + // StreamingFrequency setter re-reads it and throws when it does not like the value, and + // MaxSamplingRate is a mutable public property — so validating against one read and + // then assigning through a setter that takes another would let a concurrent + // capabilities update throw out of a Send whose command has already gone to the device. + // Tracking a command must never be able to fail the send that carried it. + var maxSamplingRate = Math.Max(1, Metadata.Capabilities.MaxSamplingRate); + + if (!int.TryParse(rate, NumberStyles.Integer, CultureInfo.InvariantCulture, out var frequency) + || frequency < 1 + || frequency > maxSamplingRate) + { + Trace.WriteLine( + $"[{nameof(TrackStreamingStart)}] Ignoring a start-streaming command with an unusable rate " + + $"('{rate.ToString()}'); the session state is unchanged."); + return; + } + + // Frequency first: anything observing IsStreaming must never catch it true next to a + // rate belonging to a previous session. Assigned to the backing field, not through the + // validating setter, for the reason above — the value has just been validated against + // the same rule. + _streamingFrequency = frequency; + + if (IsStreaming) + { + // A restart while already streaming. The typed API cannot even express this + // (StartStreaming returns early), so there is no session boundary to re-anchor at + // and no equivalence to preserve; recording the new rate is all that is warranted. + return; + } + + // A session is beginning, so it gets exactly the preparation StartStreaming would have + // given it. Ordering matches too: the state is ready before the flag flips. + BeginStreamingSession(); + IsStreaming = true; + } + + /// + /// Applies a sent ADC enable bitmask to this device's analog channels, so a caller who + /// enabled channels with a raw command has the same restorable session as one who used + /// . + /// + /// + /// The mask is a set-replace, exactly as the firmware treats it, so every analog channel is + /// assigned from it rather than only the set bits. A device-reported mask still wins on the + /// next status frame (#409) — that is the device's own view, and it outranks what was asked + /// for. + /// + private void TrackAdcEnableMask(ReadOnlySpan argument) + { + if (!uint.TryParse(argument.Trim(), NumberStyles.Integer, CultureInfo.InvariantCulture, out var mask)) + { + return; + } + + WithChannelsLock(() => + { + foreach (var channel in SnapshotChannels()) + { + if (channel.Type != ChannelType.Analog || channel.ChannelNumber > MaxAdcBitmaskChannel) + { + continue; + } + + channel.IsEnabled = (mask & (1u << channel.ChannelNumber)) != 0; + } + }); + } + + #endregion + + #region Session restore after an automatic reconnect (issue #379) + + /// + /// What the streaming session looked like at the instant the connection dropped. Null until + /// a drop is detected with reconnect enabled. + /// + private volatile StreamingSessionSnapshot? _sessionSnapshot; + + /// + /// The subset of a streaming session that Core owns and can therefore put back: which + /// channels were enabled, and whether data was flowing. + /// + private sealed class StreamingSessionSnapshot + { + public StreamingSessionSnapshot(HashSet<(ChannelType Type, int Number)> enabledChannels, bool wasStreaming) + { + EnabledChannels = enabledChannels; + WasStreaming = wasStreaming; + } + + /// + /// The enabled channels, held by identity rather than by reference: a reconnect can + /// replace the channel objects, and a device that came back with a different channel + /// count should restore the intersection rather than fail. + /// + public HashSet<(ChannelType Type, int Number)> EnabledChannels { get; } + + public bool WasStreaming { get; } + } + + /// + protected override void CaptureSessionSnapshot() + { + var enabled = new HashSet<(ChannelType, int)>(); + foreach (var channel in GetChannelsSnapshot()) + { + if (channel.IsEnabled) + { + enabled.Add((channel.Type, channel.ChannelNumber)); + } + } + + _sessionSnapshot = new StreamingSessionSnapshot(enabled, IsStreaming); + } + + /// + /// Re-applies the enabled-channel set recorded at the drop and, if the policy says so, + /// restarts a stream that was interrupted. + /// + /// + /// + /// The enable set has to be replayed from the snapshot rather than read back off the + /// channel objects: resyncs analog + /// IsEnabled from the device's own enabled mask on every status message (#409), so by + /// the time re-initialization is done the in-memory view reflects the freshly reconnected + /// device, not the session that was lost. + /// + /// + /// The streaming frequency needs no replay — it is a host-side setting that the drop never + /// touched — but it does have to reach the device again, which is what the resumed + /// does. + /// + /// + /// A resumed stream is a genuinely new session: timestamp reconstruction re-anchors and the + /// gap detector resets, because the device's tick counter may well have restarted while it + /// was away, and carrying the old anchor across would manufacture a nonsense gap. + /// is the marker for the outage, and it carries its + /// duration. + /// + /// + protected override Task RestoreSessionSnapshotAsync( + ReconnectOptions options, + CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(options); + + // A reconnected device is never streaming: re-initialization has just sent it + // StopStreamData. The flag, though, is still set from before the drop — nothing stopped + // the stream, the connection simply ended — and leaving it that way would report a + // device as streaming while it sits idle, and make StartStreaming() a silent no-op. + IsStreaming = false; + + var snapshot = _sessionSnapshot; + if (snapshot == null) + { + return Task.FromResult(false); + } + + cancellationToken.ThrowIfCancellationRequested(); + + // Normalize to a known state before re-applying: whatever the device came back with is + // not necessarily what it had, and the enable commands are set-replace anyway. + DisableAllChannels(); + + var toEnable = new List(); + foreach (var channel in GetChannelsSnapshot()) + { + if (snapshot.EnabledChannels.Contains((channel.Type, channel.ChannelNumber))) + { + toEnable.Add(channel); + } + } + + if (toEnable.Count > 0) + { + EnableChannels(toEnable); + } + + cancellationToken.ThrowIfCancellationRequested(); + + var resumeStreaming = snapshot.WasStreaming && options.ResumeStreaming; + if (resumeStreaming) + { + StartStreaming(); + } + + return Task.FromResult(resumeStreaming); + } + + #endregion + /// /// The default bounded-buffer capacity (in samples) used by . /// diff --git a/src/Daqifi.Core/Device/DeviceErrorSource.cs b/src/Daqifi.Core/Device/DeviceErrorSource.cs index 84f5df6..432443e 100644 --- a/src/Daqifi.Core/Device/DeviceErrorSource.cs +++ b/src/Daqifi.Core/Device/DeviceErrorSource.cs @@ -32,4 +32,16 @@ public enum DeviceErrorSource /// (). The frame is dropped; the stream keeps running. /// StreamDecode = 2, + + /// + /// Automatic reconnection after a dropped connection (issue #379): every allowed attempt + /// failed, so the session is gone for good. + /// + /// + /// Unlike the other sources this one is terminal rather than incidental — it is raised exactly + /// once, when the device gives up, and it is the loud counterpart to + /// . A reconnect that is cancelled does not raise it: + /// stopping on request is not a failure. + /// + Reconnect = 3, } diff --git a/src/Daqifi.Core/Device/DeviceReconnectFailedException.cs b/src/Daqifi.Core/Device/DeviceReconnectFailedException.cs new file mode 100644 index 0000000..bd9c19a --- /dev/null +++ b/src/Daqifi.Core/Device/DeviceReconnectFailedException.cs @@ -0,0 +1,35 @@ +namespace Daqifi.Core.Device; + +/// +/// Reports that automatic reconnection exhausted every allowed attempt without restoring the +/// session (issue #379). +/// +/// +/// This is never thrown — nobody is on the other end of a background reconnect loop to catch it. +/// It exists so that giving up arrives on as a typed +/// failure carrying the attempt count, with whatever ended the final attempt as its +/// . +/// +public class DeviceReconnectFailedException : Exception +{ + /// + /// Initializes a new instance of the class. + /// + /// The device that could not be reconnected. + /// How many attempts were made before giving up. + /// The failure that ended the final attempt, if there was one. + public DeviceReconnectFailedException(string deviceName, int attemptsMade, Exception? innerException = null) + : base( + $"Device '{deviceName}' could not be reconnected after {attemptsMade} attempt(s).", + innerException) + { + DeviceName = deviceName; + AttemptsMade = attemptsMade; + } + + /// Gets the name of the device that could not be reconnected. + public string DeviceName { get; } + + /// Gets how many reconnect attempts were made before giving up. + public int AttemptsMade { get; } +} diff --git a/src/Daqifi.Core/Device/ReconnectAttemptEventArgs.cs b/src/Daqifi.Core/Device/ReconnectAttemptEventArgs.cs new file mode 100644 index 0000000..b21351d --- /dev/null +++ b/src/Daqifi.Core/Device/ReconnectAttemptEventArgs.cs @@ -0,0 +1,49 @@ +namespace Daqifi.Core.Device; + +/// +/// Reports that an automatic reconnect attempt is about to be made (issue #379). +/// +/// +/// Raised once per attempt, before the backoff wait, so a UI can show both which attempt is +/// running and how long it will be until anything happens. +/// +public class ReconnectAttemptEventArgs : EventArgs +{ + /// + /// Initializes a new instance of the class. + /// + /// The 1-based attempt number. + /// The total number of attempts the policy allows. + /// How long the device will wait before making this attempt. + /// Why the previous attempt failed, or null for the first. + public ReconnectAttemptEventArgs( + int attemptNumber, + int maxAttempts, + TimeSpan delay, + Exception? previousError = null) + { + AttemptNumber = attemptNumber; + MaxAttempts = maxAttempts; + Delay = delay; + PreviousError = previousError; + Timestamp = DateTime.UtcNow; + } + + /// Gets the 1-based number of the attempt about to be made. + public int AttemptNumber { get; } + + /// Gets how many attempts the policy allows in total. + public int MaxAttempts { get; } + + /// Gets the backoff wait that precedes this attempt. + public TimeSpan Delay { get; } + + /// + /// Gets the failure that ended the previous attempt, or null when this is the first + /// attempt after the drop. + /// + public Exception? PreviousError { get; } + + /// Gets the UTC time at which the attempt was scheduled. + public DateTime Timestamp { get; } +} diff --git a/src/Daqifi.Core/Device/ReconnectFailedEventArgs.cs b/src/Daqifi.Core/Device/ReconnectFailedEventArgs.cs new file mode 100644 index 0000000..a2bfb9a --- /dev/null +++ b/src/Daqifi.Core/Device/ReconnectFailedEventArgs.cs @@ -0,0 +1,48 @@ +namespace Daqifi.Core.Device; + +/// +/// Reports that automatic reconnection has stopped without restoring the session (issue #379) — +/// either because every allowed attempt failed, or because it was cancelled. +/// +/// +/// Giving up is the loud outcome: alongside this event the failure is logged, raised on +/// with +/// , and the device settles on +/// . Cancellation is quieter — it is what the caller asked +/// for — and leaves the device on . +/// +public class ReconnectFailedEventArgs : EventArgs +{ + /// + /// Initializes a new instance of the class. + /// + /// How many attempts were made. + /// The failure that ended the final attempt, if there was one. + /// Whether reconnection stopped because it was cancelled. + public ReconnectFailedEventArgs(int attemptsMade, Exception? lastError, bool wasCanceled) + { + AttemptsMade = attemptsMade; + LastError = lastError; + WasCanceled = wasCanceled; + Timestamp = DateTime.UtcNow; + } + + /// Gets how many reconnect attempts were made before giving up. + public int AttemptsMade { get; } + + /// + /// Gets the failure that ended the last attempt, or null when reconnection was + /// cancelled before any attempt failed. + /// + public Exception? LastError { get; } + + /// + /// Gets a value indicating whether reconnection stopped because it was cancelled — by + /// , , or + /// disposal — rather than by exhausting . + /// + public bool WasCanceled { get; } + + /// Gets the UTC time at which reconnection stopped. + public DateTime Timestamp { get; } +} diff --git a/src/Daqifi.Core/Device/ReconnectOptions.cs b/src/Daqifi.Core/Device/ReconnectOptions.cs new file mode 100644 index 0000000..fc40e21 --- /dev/null +++ b/src/Daqifi.Core/Device/ReconnectOptions.cs @@ -0,0 +1,206 @@ +namespace Daqifi.Core.Device; + +/// +/// Policy for automatically re-establishing a session after +/// (issue #379). Assign to +/// . +/// +/// +/// +/// Off by default. A freshly constructed instance has set to +/// false, which is exactly the behaviour a device has always had: a drop is reported as +/// and nothing else happens. Reconnect has to be asked for. +/// +/// +/// The shape mirrors , which governs +/// the initial connect, so the two read the same way. They are separate settings for +/// separate moments: that one retries a connect the caller asked for, this one retries a session +/// the caller never asked to lose. +/// +/// +public class ReconnectOptions +{ + private int _maxAttempts = 5; + private TimeSpan _initialDelay = TimeSpan.FromSeconds(1); + private TimeSpan _maxDelay = TimeSpan.FromSeconds(30); + private double _backoffMultiplier = 2.0; + + /// + /// Gets or sets a value indicating whether a lost connection is reconnected automatically. + /// Default is false — reconnect is opt-in. + /// + public bool Enabled { get; set; } + + /// + /// Gets or sets how many reconnect attempts are made before giving up. Default is 5. + /// + /// Thrown when the value is less than 1. + public int MaxAttempts + { + get => _maxAttempts; + set + { + if (value < 1) + { + throw new ArgumentOutOfRangeException( + nameof(MaxAttempts), value, "At least one reconnect attempt must be allowed."); + } + + _maxAttempts = value; + } + } + + /// + /// Gets or sets the wait before the first reconnect attempt. Default is 1 second. + /// + /// + /// There is always a wait before the first attempt, unlike + /// : at the instant a drop is + /// detected the endpoint is, by definition, gone. A serial port that has just been unplugged + /// has not finished disappearing from the OS yet, let alone come back. + /// + /// Thrown when the value is negative. + public TimeSpan InitialDelay + { + get => _initialDelay; + set + { + if (value < TimeSpan.Zero) + { + throw new ArgumentOutOfRangeException( + nameof(InitialDelay), value, "The delay cannot be negative."); + } + + _initialDelay = value; + } + } + + /// + /// Gets or sets the ceiling the backoff grows to. Default is 30 seconds. + /// + /// Thrown when the value is negative. + public TimeSpan MaxDelay + { + get => _maxDelay; + set + { + if (value < TimeSpan.Zero) + { + throw new ArgumentOutOfRangeException( + nameof(MaxDelay), value, "The delay cannot be negative."); + } + + _maxDelay = value; + } + } + + /// + /// Gets or sets the factor each successive delay is multiplied by. Default is 2.0; 1.0 gives a + /// fixed delay. + /// + /// Thrown when the value is less than 1. + public double BackoffMultiplier + { + get => _backoffMultiplier; + set + { + // Negated rather than `value < 1.0` so NaN — which compares false against everything — + // is rejected too, instead of turning every backoff into NaN milliseconds. + if (!(value >= 1.0)) + { + throw new ArgumentOutOfRangeException( + nameof(BackoffMultiplier), value, "The backoff multiplier must be at least 1.0."); + } + + _backoffMultiplier = value; + } + } + + /// + /// Gets or sets a value indicating whether a stream that was running when the connection + /// dropped is restarted once the session is back. Default is true. + /// + /// + /// Set to false to restore the channel configuration but leave the device idle — useful + /// when the consumer wants to decide for itself whether resuming acquisition still makes sense + /// after an outage of unknown length. + /// + public bool ResumeStreaming { get; set; } = true; + + /// + /// A policy that never reconnects. Identical to a default-constructed instance; named for + /// callers that want to say so explicitly. + /// + public static ReconnectOptions Disabled => new(); + + /// + /// A reasonable enabled policy: five attempts, 1 s growing to 30 s, resuming an active stream. + /// + public static ReconnectOptions Default => new() { Enabled = true }; + + /// + /// A policy for links that blip briefly and often — six quick attempts inside about 10 seconds. + /// + public static ReconnectOptions Fast => new() + { + Enabled = true, + MaxAttempts = 6, + InitialDelay = TimeSpan.FromMilliseconds(500), + MaxDelay = TimeSpan.FromSeconds(4), + BackoffMultiplier = 1.5 + }; + + /// + /// A policy for unattended long runs: keeps trying for roughly ten minutes before giving up. + /// + public static ReconnectOptions Resilient => new() + { + Enabled = true, + MaxAttempts = 15, + InitialDelay = TimeSpan.FromSeconds(2), + MaxDelay = TimeSpan.FromSeconds(60), + BackoffMultiplier = 2.0 + }; + + /// + /// Calculates how long to wait before attempt . + /// + /// The 1-based attempt number. + /// + /// multiplied by once per attempt + /// already made, capped at . Always when + /// is zero, however many attempts have been made. + /// + public TimeSpan CalculateDelay(int attemptNumber) + { + // Zero in, zero out. A caller who asked for immediate retries must get them at every + // attempt number: multiplying zero by the growing backoff factor is still zero, right up + // until that factor overflows Math.Pow to infinity, where 0 × ∞ is NaN and the arithmetic + // below would hand back MaxDelay — the exact opposite of the configured policy. + if (InitialDelay == TimeSpan.Zero) + { + return TimeSpan.Zero; + } + + // MaxDelay is a ceiling on every wait, including the first: a policy whose MaxDelay is + // below its InitialDelay means what it says rather than exempting attempt 1. + if (attemptNumber <= 1) + { + return InitialDelay < MaxDelay ? InitialDelay : MaxDelay; + } + + var delayMs = InitialDelay.TotalMilliseconds * Math.Pow(BackoffMultiplier, attemptNumber - 1); + + // Math.Pow overflows to +Infinity for a large enough attempt count. With a positive + // InitialDelay the product is then +Infinity too, and Math.Min yields MaxDelay — the + // intended cap. NaN is unreachable from here now that a zero InitialDelay returns above + // and BackoffMultiplier rejects NaN, so this is a backstop against a future relaxation of + // either, not a live path. + if (double.IsNaN(delayMs)) + { + return MaxDelay; + } + + return TimeSpan.FromMilliseconds(Math.Min(delayMs, MaxDelay.TotalMilliseconds)); + } +} diff --git a/src/Daqifi.Core/Device/ReconnectedEventArgs.cs b/src/Daqifi.Core/Device/ReconnectedEventArgs.cs new file mode 100644 index 0000000..920bac1 --- /dev/null +++ b/src/Daqifi.Core/Device/ReconnectedEventArgs.cs @@ -0,0 +1,45 @@ +namespace Daqifi.Core.Device; + +/// +/// Reports that a lost session has been re-established and its state restored (issue #379). +/// +/// +/// Raised after the transport is back, the device has been re-initialized, and the channel +/// configuration — and the stream, if one was running and the policy resumes it — have been +/// re-applied. By the time this fires, samples are flowing again. +/// +public class ReconnectedEventArgs : EventArgs +{ + /// + /// Initializes a new instance of the class. + /// + /// The 1-based attempt that succeeded. + /// How long the session was down, measured from the drop. + /// Whether an interrupted stream was restarted. + public ReconnectedEventArgs(int attemptNumber, TimeSpan outage, bool streamingResumed) + { + AttemptNumber = attemptNumber; + Outage = outage; + StreamingResumed = streamingResumed; + Timestamp = DateTime.UtcNow; + } + + /// Gets the 1-based number of the attempt that succeeded. + public int AttemptNumber { get; } + + /// + /// Gets how long the session was down, from the drop being detected to the state being + /// restored. + /// + public TimeSpan Outage { get; } + + /// + /// Gets a value indicating whether a stream that was running at the time of the drop was + /// restarted. false when nothing was streaming, or when + /// is off. + /// + public bool StreamingResumed { get; } + + /// Gets the UTC time at which the session was restored. + public DateTime Timestamp { get; } +}