fix: Properly propagate transport errors (#110)

Summary
Right now, if a transport (e.g., the H4Transport) fails, we only log the error but never propagate it to the user, meaning, e.g., an OnAdvertisement call might hang forever, and the user has no way of knowing. This PR introduces error propagation to the BleObserver.

Changes
Added an onError for HciTransports
Added onError to the OnAdvertisement of IBleObserver
Raising errors in HciTransports, Windows Observers, and Android observers
Testing
Added unit tests
Impact
Breaking the onAdvertisement function
OnAdvertisement (IObservable) might now return errors

Author: OleRoss

.editorconfig

@@ -40,6 +40,8 @@ dotnet_diagnostic.CS9124.severity = error
 dotnet_diagnostic.CA1819.severity = none
 dotnet_diagnostic.CA1716.severity = none
 dotnet_diagnostic.CA1043.severity = none
+# Do not warn catching general exception types. With c#, you never know which exceptions occur
+dotnet_diagnostic.CA1031.severity = none
 
 # ReSharper properties
 resharper_formatter_off_tag = @formatter:off

src/Darp.Ble.Android/AndroidBleObserver.cs

@@ -40,8 +40,15 @@ protected override Task StartObservingAsyncCore(CancellationToken cancellationTo
             },
             failure =>
             {
-                Logger.LogError("Scan failure because of {Failure}", failure);
-                _ = StopObservingAsync();
+                if (_scanCallback is not null)
+                {
+                    _bluetoothLeScanner.StopScan(_scanCallback);
+                    _scanCallback.Dispose();
+                    _scanCallback = null;
+                }
+                _ = OnErrorAsync(
+                    new BleObservationException(this, $"Scan failed because of {failure}", innerException: null)
+                );
             }
         );
         using var settingsBuilder = new ScanSettings.Builder();

src/Darp.Ble.Hci/Host/HciHost.cs

@@ -15,6 +15,13 @@
 
 namespace Darp.Ble.Hci.Host;
 
+/// <summary> Provides the fatal exception raised by the transport layer. </summary>
+public sealed class HciTransportFailedEventArgs(Exception exception) : EventArgs
+{
+    /// <summary> The fatal transport exception. </summary>
+    public Exception Exception { get; } = exception;
+}
+
 /// <summary>
 /// The <see cref="HciHost"/> is responsible for all host-related commands.
 /// </summary>
@@ -29,6 +36,9 @@ public sealed partial class HciHost(HciDevice hciDevice, ITransportLayer transpo
     private AclPacketQueue? _leAclPacketQueue;
     private bool _isResetDoneAtLeastOnce;
 
+    /// <summary> An event that notifies when the transport has failed </summary>
+    public event EventHandler<HciTransportFailedEventArgs>? TransportFailed;
+
     /// <summary> The HCI Device </summary>
     public HciDevice Device { get; } = hciDevice;
 
@@ -44,7 +54,7 @@ public sealed partial class HciHost(HciDevice hciDevice, ITransportLayer transpo
     public async Task ResetAsync(CancellationToken token)
     {
         ObjectDisposedException.ThrowIf(Device.IsDisposed, this);
-        await _transportLayer.InitializeAsync(OnReceivedPacket, token).ConfigureAwait(false);
+        await _transportLayer.InitializeAsync(OnReceivedPacket, OnTransportFailed, token).ConfigureAwait(false);
         Activity? activity = Logging.StartInitializeHciHostActivity();
         try
         {
@@ -234,6 +244,11 @@ when HciAclPacket.TryReadLittleEndian(packet.Pdu, out HciAclPacket x):
         }
     }
 
+    private void OnTransportFailed(Exception exception)
+    {
+        TransportFailed?.Invoke(this, new HciTransportFailedEventArgs(exception));
+    }
+
     private void OnReceivedHciEventPacket(HciPacketEvent packet)
     {
         if (

src/Darp.Ble.Hci/Transport/H4TransportLayer.cs

@@ -26,23 +26,57 @@ public sealed class H4TransportLayer(string portName, ILogger<H4TransportLayer>?
     private readonly SerialPort _serialPort = new(portName);
     private readonly Channel<IHciPacket> _txQueue = Channel.CreateUnbounded<IHciPacket>();
     private readonly CancellationTokenSource _cancelSource = new();
+    private Action<Exception>? _onError;
     private bool _isDisposing;
+    private int _isFaulted;
     private Task? _rxTask;
     private Task? _txTask;
 
     private CancellationToken StopToken => _cancelSource.Token;
 
     /// <inheritdoc />
-    public ValueTask InitializeAsync(Action<HciPacket> onReceived, CancellationToken cancellationToken)
+    public ValueTask InitializeAsync(
+        Action<HciPacket> onReceived,
+        Action<Exception> onError,
+        CancellationToken cancellationToken
+    )
     {
         if (_txTask is not null || _rxTask is not null)
             throw new InvalidOperationException("Initialization can only be done once");
+        _onError = onError;
         _serialPort.Open();
         _txTask = Task.Run(RunTx, cancellationToken);
         _rxTask = Task.Run(() => RunRx(onReceived), cancellationToken);
         return ValueTask.CompletedTask;
     }
 
+    private void ReportTransportFailure(Exception exception, string direction)
+    {
+        if (Interlocked.Exchange(ref _isFaulted, 1) != 0)
+            return;
+
+        _logger?.LogH4TransportWithError(exception, direction, exception.Message);
+        try
+        {
+            _cancelSource.Cancel();
+        }
+        catch
+        {
+            // Ignore cancellation errors while reporting a fatal transport failure
+        }
+
+        try
+        {
+            _serialPort.Close();
+        }
+        catch
+        {
+            // Ignore close errors while reporting a fatal transport failure
+        }
+
+        _onError?.Invoke(exception);
+    }
+
     private async Task RunTx()
     {
         try
@@ -77,7 +111,7 @@ private async Task RunTx()
                 _logger?.LogH4TransportDisconnected("Tx");
                 return;
             }
-            _logger?.LogH4TransportWithError(e, "Tx", e.Message);
+            ReportTransportFailure(e, "Tx");
         }
 #pragma warning restore CA1031
     }
@@ -130,7 +164,7 @@ private async Task RunRx(Action<HciPacket> onReceived)
                 return;
             }
 
-            _logger?.LogH4TransportWithError(e, "Rx", e.Message);
+            ReportTransportFailure(e, "Rx");
         }
 #pragma warning restore CA1031
     }

src/Darp.Ble.Hci/Transport/ITransportLayer.cs

@@ -10,5 +10,9 @@ public interface ITransportLayer : IAsyncDisposable
     void Enqueue(IHciPacket packet);
 
     /// <summary> Initialize the transport layer </summary>
-    ValueTask InitializeAsync(Action<HciPacket> onReceived, CancellationToken cancellationToken);
+    ValueTask InitializeAsync(
+        Action<HciPacket> onReceived,
+        Action<Exception> onError,
+        CancellationToken cancellationToken
+    );
 }

src/Darp.Ble.HciHost/HciHostBleObserver.cs

@@ -27,6 +27,14 @@ public HciHostBleObserver(HciHostBleDevice device, ILogger<HciHostBleObserver> l
     {
         _device = device;
         _subscription = Host.Subscribe(this);
+        Host.TransportFailed += OnTransportFailed;
+    }
+
+    private void OnTransportFailed(object? sender, HciTransportFailedEventArgs args)
+    {
+        _ = OnErrorAsync(
+            new BleObservationException(this, "The HCI transport failed while observing advertisements", args.Exception)
+        );
     }
 
     [MessageSink]
@@ -121,6 +129,7 @@ await Host.QueryCommandCompletionAsync<HciLeSetExtendedScanEnableCommand, HciSet
 
     protected override ValueTask DisposeAsyncCore()
     {
+        Host.TransportFailed -= OnTransportFailed;
         _subscription.Dispose();
         return base.DisposeAsyncCore();
     }

src/Darp.Ble.WinRT/WinBleObserver.cs

@@ -53,8 +53,12 @@ protected override Task StartObservingAsyncCore(CancellationToken cancellationTo
         {
             if (args.Error is BluetoothError.Success)
                 return;
-            Logger.LogError("Watcher stopped with error {Error}", args.Error);
-            await StopObservingAsync().ConfigureAwait(false);
+            _observableSubscription?.Dispose();
+            _watcher = null;
+            await OnErrorAsync(
+                    new BleObservationException(this, $"Watcher stopped with error {args.Error}", innerException: null)
+                )
+                .ConfigureAwait(false);
         };
         _observableSubscription = Observable
             .FromEventPattern<
@@ -63,7 +67,7 @@ protected override Task StartObservingAsyncCore(CancellationToken cancellationTo
                 BluetoothLEAdvertisementReceivedEventArgs
             >(addHandler => _watcher.Received += addHandler, removeHandler => _watcher.Received -= removeHandler)
             .Select(adv => OnAdvertisementReport(this, adv))
-            .Subscribe(OnNext);
+            .Subscribe(adv => OnNext(adv));
         return Task.CompletedTask;
     }
 

src/Darp.Ble/BleObserverExtensions.cs

@@ -10,13 +10,16 @@ public static class BleObserverExtensions
 {
     /// <summary>
     /// Observes advertisements broadcast by BLE devices using the provided <see cref="IBleObserver"/>.
+    /// Will error when a fatal error has occurred
     /// </summary>
     /// <param name="observer">The instance of <see cref="IBleObserver"/> that will monitor BLE advertisements.</param>
     /// <returns>An observable sequence of <see cref="IGapAdvertisement"/> instances representing BLE advertisements.</returns>
     public static IObservable<IGapAdvertisement> OnAdvertisement(this IBleObserver observer)
     {
         ArgumentNullException.ThrowIfNull(observer);
-        return Observable.Create<IGapAdvertisement>(advObserver => observer.OnAdvertisement(advObserver.OnNext));
+        return Observable.Create<IGapAdvertisement>(advObserver =>
+            observer.OnAdvertisement(advObserver.OnNext, advObserver.OnError)
+        );
     }
 
     /// <summary> Publish the observer to allow observing advertisements without having to start/stop observation manually </summary>
@@ -28,7 +31,10 @@ public static IConnectableObservable<IGapAdvertisement> Publish(this IBleObserve
 
         IObservable<IGapAdvertisement> inner = Observable.Create<IGapAdvertisement>(async observer =>
         {
-            IDisposable unhook = bleObserver.OnAdvertisement(onAdvertisement: observer.OnNext);
+            IDisposable unhook = bleObserver.OnAdvertisement(
+                onAdvertisement: observer.OnNext,
+                onError: observer.OnError
+            );
 
             await bleObserver.StartObservingAsync().ConfigureAwait(false);
 

src/Darp.Ble/IBleObserver.cs

@@ -26,8 +26,9 @@ public interface IBleObserver
 
     /// <summary> Register a callback called when an advertisement was received </summary>
     /// <param name="onAdvertisement"> The callback </param>
+    /// <param name="onError"> The callback for fatal observation errors </param>
     /// <returns> A disposable to unsubscribe the callback </returns>
-    IDisposable OnAdvertisement(Action<IGapAdvertisement> onAdvertisement);
+    IDisposable OnAdvertisement(Action<IGapAdvertisement> onAdvertisement, Action<Exception>? onError = null);
 
     /// <summary> Start observing for advertisements. </summary>
     /// <param name="cancellationToken"> The CancellationToken to cancel the initial starting process </param>

src/Darp.Ble/Implementation/BleObserver.cs

@@ -24,10 +24,15 @@ internal enum ObserverState
 /// <param name="logger"> The logger </param>
 public abstract class BleObserver(BleDevice device, ILogger<BleObserver> logger) : IBleObserver, IAsyncDisposable
 {
+    private readonly record struct AdvertisementHandlerSubscription(
+        Action<IGapAdvertisement> OnAdvertisement,
+        Action<Exception>? OnError
+    );
+
     private readonly BleDevice _bleDevice = device;
     private readonly SemaphoreSlim _startStopSemaphore = new(1, 1);
     private readonly Lock _handlersLock = new();
-    private Action<IGapAdvertisement>[] _handlers = [];
+    private AdvertisementHandlerSubscription[] _handlers = [];
 
     private volatile ObserverState _observerState = ObserverState.Stopped;
 
@@ -103,25 +108,27 @@ public async Task StartObservingAsync(CancellationToken cancellationToken = defa
     }
 
     /// <inheritdoc />
-    public IDisposable OnAdvertisement(Action<IGapAdvertisement> onAdvertisement)
+    public IDisposable OnAdvertisement(Action<IGapAdvertisement> onAdvertisement, Action<Exception>? onError = null)
     {
         ObjectDisposedException.ThrowIf(_bleDevice.IsDisposing, nameof(BleObserver));
 
+        var subscription = new AdvertisementHandlerSubscription(onAdvertisement, onError);
+
         // Extend handlers list
         lock (_handlersLock)
         {
-            Action<IGapAdvertisement>[] oldHandlers = _handlers;
-            var newArr = new Action<IGapAdvertisement>[oldHandlers.Length + 1];
+            AdvertisementHandlerSubscription[] oldHandlers = _handlers;
+            var newArr = new AdvertisementHandlerSubscription[oldHandlers.Length + 1];
             Array.Copy(oldHandlers, newArr, oldHandlers.Length);
-            newArr[^1] = onAdvertisement;
+            newArr[^1] = subscription;
             Volatile.Write(ref _handlers, newArr);
         }
 
         return Disposable.Create(
-            (this, onAdvertisement),
+            (this, subscription),
             static tuple =>
             {
-                (BleObserver self, Action<IGapAdvertisement> handler) = tuple;
+                (BleObserver self, AdvertisementHandlerSubscription handler) = tuple;
                 lock (self._handlersLock)
                 {
                     if (Helpers.TryRemoveImmutable(self._handlers, handler, out var newHandlers))
@@ -143,12 +150,12 @@ protected void OnNext(IGapAdvertisement advertisement)
         // Taking the current snapshot of the handlers.
         // In case of an unsubscription of a handler we might have taken the reference here already and call it afterward.
         // This is a known tradeoff
-        Action<IGapAdvertisement>[] handlers = Volatile.Read(ref _handlers);
-        foreach (Action<IGapAdvertisement> handler in handlers)
+        AdvertisementHandlerSubscription[] handlers = Volatile.Read(ref _handlers);
+        foreach (AdvertisementHandlerSubscription handler in handlers)
         {
             try
             {
-                handler(advertisement);
+                handler.OnAdvertisement(advertisement);
             }
             catch (Exception e)
             {
@@ -158,6 +165,52 @@ protected void OnNext(IGapAdvertisement advertisement)
         }
     }
 
+    /// <summary> Notify subscribers of a fatal observation error. Existing subscriptions are terminated. </summary>
+    /// <param name="exception"> The exception that triggered the error </param>
+    protected async Task OnErrorAsync(Exception exception)
+    {
+        ArgumentNullException.ThrowIfNull(exception);
+
+        if (_bleDevice.IsDisposing)
+            return;
+
+        AdvertisementHandlerSubscription[] handlers;
+        await _startStopSemaphore.WaitAsync().ConfigureAwait(false);
+        try
+        {
+            handlers = Volatile.Read(ref _handlers);
+            if (handlers.Length == 0 && _observerState is ObserverState.Stopped)
+                return;
+
+            _observerState = ObserverState.Stopped;
+            lock (_handlersLock)
+            {
+                _handlers = [];
+            }
+        }
+        finally
+        {
+            _startStopSemaphore.Release();
+        }
+
+        Logger.LogObservationFailed(exception);
+
+        foreach (AdvertisementHandlerSubscription handler in handlers)
+        {
+            if (handler.OnError is null)
+                continue;
+
+            try
+            {
+                handler.OnError(exception);
+            }
+            catch (Exception e)
+            {
+                Logger.LogObservationErrorDuringErrorHandling(e);
+            }
+        }
+    }
+
     /// <summary> Core implementation to start observing async </summary>
     /// <param name="cancellationToken"> The cancellationToken to cancel the operation </param>
     /// <returns> A task that completes when observation has started </returns>