fix(audio-response): make repeated operations reliable

lib/src/managers/ble_manager.dart

@@ -396,7 +396,14 @@ class BleManager extends BleGattManager {
     }
 
     streamController.onCancel = () async {
-      if (_streamControllers.containsKey(streamIdentifier)) {
+      // A canceled controller may finish closing after a replacement
+      // subscription has already installed a new controller for the same
+      // characteristic. Only the controller that still owns the map entry may
+      // tear down the native notification subscription.
+      if (identical(
+        _streamControllers[streamIdentifier],
+        streamController,
+      )) {
         final canceledController = _streamControllers.remove(streamIdentifier);
         _subscriptionSetups.remove(streamIdentifier);
         if (canceledController != null && !canceledController.isClosed) {

lib/src/models/devices/open_earable_v2_audio_response_manager.dart

@@ -24,7 +24,7 @@ class OpenEarableV2AudioResponseManager implements AudioResponseManager {
   /// Maximum time to wait for each protocol notification.
   final Duration notificationTimeout;
 
-  bool _operationInProgress = false;
+  Future<void> _operationQueue = Future<void>.value();
 
   @override
   Future<void> uploadAudioBuffer({
@@ -79,7 +79,9 @@ class OpenEarableV2AudioResponseManager implements AudioResponseManager {
 
       try {
         reportProgress(AudioResponseUploadPhase.starting, 0);
-        var statusFuture = _nextTransferStatus(statusIterator, transferId);
+        var statusFuture = _guardPendingFuture(
+          _nextTransferStatus(statusIterator, transferId),
+        );
         await _write(
           AudioResponseBleUuids.transferControlCharacteristicUuid,
           AudioResponseTransferControl.start(
@@ -117,7 +119,9 @@ class OpenEarableV2AudioResponseManager implements AudioResponseManager {
             continue;
           }
 
-          statusFuture = _nextTransferStatus(statusIterator, transferId);
+          statusFuture = _guardPendingFuture(
+            _nextTransferStatus(statusIterator, transferId),
+          );
           for (var credit = 0;
               credit < status.credits && sentOffset < samples.length;
               credit++) {
@@ -160,7 +164,9 @@ class OpenEarableV2AudioResponseManager implements AudioResponseManager {
           AudioResponseUploadPhase.committing,
           lastAcknowledgedSamples,
         );
-        statusFuture = _nextTransferStatus(statusIterator, transferId);
+        statusFuture = _guardPendingFuture(
+          _nextTransferStatus(statusIterator, transferId),
+        );
         await _write(
           AudioResponseBleUuids.transferControlCharacteristicUuid,
           AudioResponseTransferControl.commit(
@@ -196,7 +202,9 @@ class OpenEarableV2AudioResponseManager implements AudioResponseManager {
       );
 
       try {
-        final resultFuture = _nextResult(resultIterator, config.id);
+        final resultFuture = _guardPendingFuture(
+          _nextResult(resultIterator, config.id),
+        );
         await _write(
           AudioResponseBleUuids.configCharacteristicUuid,
           config.toBytes(),
@@ -210,17 +218,24 @@ class OpenEarableV2AudioResponseManager implements AudioResponseManager {
     });
   }
 
-  Future<T> _runExclusive<T>(Future<T> Function() operation) async {
-    if (_operationInProgress) {
-      throw StateError('An audio response operation is already in progress');
-    }
+  Future<T> _runExclusive<T>(Future<T> Function() operation) {
+    final result = Completer<T>();
+    _operationQueue = _operationQueue.then((_) async {
+      try {
+        result.complete(await operation());
+      } on Object catch (error, stackTrace) {
+        result.completeError(error, stackTrace);
+      }
+    });
+    return result.future;
+  }
 
-    _operationInProgress = true;
-    try {
-      return await operation();
-    } finally {
-      _operationInProgress = false;
-    }
+  Future<T> _guardPendingFuture<T>(Future<T> future) {
+    // Notifications must be awaited before the corresponding write so a fast
+    // response cannot be missed. If that write fails, keep the pending wait
+    // from surfacing a second, detached asynchronous error during cleanup.
+    future.ignore();
+    return future;
   }
 
   Future<void> _write(

test/ble_manager_subscription_test.dart

@@ -0,0 +1,82 @@
+import 'dart:async';
+import 'dart:typed_data';
+
+import 'package:flutter_test/flutter_test.dart';
+import 'package:open_earable_flutter/src/managers/ble_manager.dart';
+import 'package:universal_ble/universal_ble.dart';
+
+void main() {
+  TestWidgetsFlutterBinding.ensureInitialized();
+
+  test('a stale controller cannot tear down its replacement subscription',
+      () async {
+    const deviceId = 'device';
+    const serviceId = '7467b395-9043-4453-bc5c-2d8e8b10680a';
+    const characteristicId = '7467b39c-9043-4453-bc5c-2d8e8b10680a';
+    final platform = _FakeUniversalBlePlatform();
+    UniversalBle.setInstance(platform);
+    final manager = BleManager();
+
+    platform.updateConnection(deviceId, true);
+    final oldStream = await manager.subscribe(
+      deviceId: deviceId,
+      serviceId: serviceId,
+      characteristicId: characteristicId,
+    );
+    final oldSubscription = oldStream.listen((_) {});
+    oldSubscription.pause();
+
+    // Closing a paused stream defers its onCancel callback. Reconnect and
+    // subscribe again before that callback is allowed to run.
+    platform.updateConnection(deviceId, false);
+    platform.updateConnection(deviceId, true);
+    final replacementStream = await manager.subscribe(
+      deviceId: deviceId,
+      serviceId: serviceId,
+      characteristicId: characteristicId,
+    );
+    final replacementValue = Completer<List<int>>();
+    final replacementSubscription = replacementStream.listen(
+      replacementValue.complete,
+    );
+
+    oldSubscription.resume();
+    await oldSubscription.asFuture<void>();
+    await oldSubscription.cancel();
+    await Future<void>.delayed(Duration.zero);
+
+    platform.updateCharacteristicValue(
+      deviceId,
+      characteristicId,
+      Uint8List.fromList([7]),
+      null,
+    );
+
+    expect(await replacementValue.future, [7]);
+    expect(
+      platform.notificationChanges,
+      [
+        BleInputProperty.notification,
+        BleInputProperty.notification,
+      ],
+    );
+    await replacementSubscription.cancel();
+  });
+}
+
+class _FakeUniversalBlePlatform extends UniversalBlePlatform {
+  final List<BleInputProperty> notificationChanges = [];
+
+  @override
+  Future<void> setNotifiable(
+    String deviceId,
+    String service,
+    String characteristic,
+    BleInputProperty bleInputProperty,
+  ) async {
+    notificationChanges.add(bleInputProperty);
+  }
+
+  @override
+  dynamic noSuchMethod(Invocation invocation) => super.noSuchMethod(invocation);
+}

test/open_earable_v2_audio_response_manager_test.dart

@@ -176,6 +176,72 @@ void main() {
       expect(result.response, [10, 20]);
     });
 
+    test('queues an operation requested while another is finishing', () async {
+      final bleManager = _FakeBleGattManager();
+      final manager = OpenEarableV2AudioResponseManager(
+        bleManager: bleManager,
+        deviceId: 'device',
+      );
+      final firstWrite = Completer<void>();
+      final configWrites = <int>[];
+
+      bleManager.onWrite = (write) {
+        if (write.characteristicId !=
+            AudioResponseBleUuids.configCharacteristicUuid) {
+          return;
+        }
+        final config = AudioResponseConfig.fromBytes(write.bytes);
+        configWrites.add(config.id);
+        if (config.id == 1) {
+          firstWrite.complete();
+          return;
+        }
+        bleManager.emitResult(
+          AudioResponseResult(
+            id: config.id,
+            points: 1,
+            frequencies: [100],
+            response: [config.id],
+          ),
+        );
+      };
+
+      final firstResult = manager.measureAudioResponse(
+        AudioResponseConfig(
+          id: 1,
+          transfer_id: 42,
+          volume: 0.5,
+          points: 1,
+          frequencies: [100],
+        ),
+      );
+      await firstWrite.future;
+      final secondResult = manager.measureAudioResponse(
+        AudioResponseConfig(
+          id: 2,
+          transfer_id: 42,
+          volume: 0.5,
+          points: 1,
+          frequencies: [100],
+        ),
+      );
+      await Future<void>.delayed(Duration.zero);
+
+      expect(configWrites, [1]);
+      bleManager.emitResult(
+        AudioResponseResult(
+          id: 1,
+          points: 1,
+          frequencies: [100],
+          response: [1],
+        ),
+      );
+
+      expect((await firstResult).id, 1);
+      expect((await secondResult).id, 2);
+      expect(configWrites, [1, 2]);
+    });
+
     test('allows device default measurement points', () async {
       final bleManager = _FakeBleGattManager();
       final manager = OpenEarableV2AudioResponseManager(
@@ -392,6 +458,53 @@ void main() {
           .map((write) => AudioResponseTransferControl.fromBytes(write.bytes));
       expect(controls.map(_controlType), [0, 2]);
     });
+
+    test('does not leak a pending status error when a write fails', () async {
+      final bleManager = _FakeBleGattManager();
+      final manager = OpenEarableV2AudioResponseManager(
+        bleManager: bleManager,
+        deviceId: 'device',
+      );
+      var failStartWrite = true;
+
+      bleManager.onWrite = (write) {
+        if (!failStartWrite ||
+            write.characteristicId !=
+                AudioResponseBleUuids.transferControlCharacteristicUuid) {
+          return;
+        }
+        final control = AudioResponseTransferControl.fromBytes(write.bytes);
+        if (_controlType(control) == 0) {
+          failStartWrite = false;
+          throw StateError('start write failed');
+        }
+      };
+
+      await expectLater(
+        manager.uploadAudioBuffer(
+          transferId: 42,
+          samples: const [0],
+          samplingRate: 16000,
+        ),
+        throwsA(
+          isA<StateError>().having(
+            (error) => error.message,
+            'message',
+            'start write failed',
+          ),
+        ),
+      );
+      await Future<void>.delayed(Duration.zero);
+
+      final controls = bleManager.writes
+          .where(
+            (write) =>
+                write.characteristicId ==
+                AudioResponseBleUuids.transferControlCharacteristicUuid,
+          )
+          .map((write) => AudioResponseTransferControl.fromBytes(write.bytes));
+      expect(controls.map(_controlType), [0, 2]);
+    });
   });
 }