8000 fix(ble): complete nearby-sensor scans at a registry-owned deadline by shroominic · Pull Request #130 · shroominic/OpenGlucose · GitHub
[go: up one dir, main page]
Skip to content

fix(ble): complete nearby-sensor scans at a registry-owned deadline #130

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
shroominic wants to merge 1 commit into from
Closed

fix(ble): complete nearby-sensor scans at a registry-owned deadline #130

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
139 changes: 129 additions & 10 deletions openhealth/lib/src/cgm_driver_registry.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,20 @@ import 'package:cgm_core/cgm_core.dart';

typedef CgmDiscoveryMapper = DiscoveredSensor? Function(BleScanResult result);

/// Prints a scan lifecycle diagnostic in debug builds only.
///
/// The scan path was debugged against a physical device, so its transitions are
/// worth keeping. Every message stays identity-free, and the call sits inside
/// an `assert` so release builds neither build nor emit them.
void _debugScanStep(Object Function() message) {
assert(() {
// Diagnostic-only helper: the analyzer would otherwise flag the print.
// ignore: avoid_print, intentional debug-only scan diagnostics.
print('[scan] ${message()}');
return true;
}(), 'scan diagnostics run only in debug builds');
}

/// One vendor driver and its pure advertisement classifier.
final class CgmDriverRegistration {
CgmDriverRegistration({
Expand Down Expand Up @@ -109,10 +123,18 @@ final class CgmDriverRegistry implements CgmDriver {
Future<void> _lifecycleTail = Future<void>.value();
Future<void> Function()? _cancelActiveScan;
StreamController<DiscoveredSensor>? _activeScanController;
Timer? _activeScanTimeout;
Object? _scanStopFailure;
StackTrace? _scanStopFailureStackTrace;
var _scanGeneration = 0;

/// Upper bound for one transport scan cancellation.
///
/// A transport that cannot confirm its own cancellation must not be able to
/// block the scan lifecycle (and therefore the UI) forever. Cancelling a
/// healthy transport resolves in milliseconds, so this only bounds failure.
static const _scanCancellationTimeout = Duration(seconds: 2);

@override
String get driverId => 'openglucose-driver-registry';

Expand Down Expand Up @@ -142,11 +164,39 @@ final class CgmDriverRegistry implements CgmDriver {
}
final generation = ++_scanGeneration;
_activeScanController = controller;
var deadlineExpired = false;
_activeScanTimeout?.cancel();
_activeScanTimeout = timeout == null
? null
: Timer(timeout, () {
deadlineExpired = true;
unawaited(
_completeScanAtDeadline(generation, controller),
);
});
// Discovery setup can await a vendor receiver. A scan that reaches
// its deadline while that is pending must still end on time.
bool aborted() {
if (!deadlineExpired && !cancelled && !controller.isClosed) {
return false;
}
if (identical(_activeScanController, controller)) {
_activeScanTimeout?.cancel();
_activeScanTimeout = null;
_activeScanController = null;
_cancelActiveScan = null;
}
if (!controller.isClosed) {
unawaited(controller.close());
}
return true;
}

final seen = <String, String>{};
try {
final availableRegistrations = <CgmDriverRegistration>[];
for (final registration in _registrations) {
if (cancelled || controller.isClosed) return;
if (aborted()) return;
try {
await registration.prepareDiscovery?.call().timeout(
const Duration(seconds: 15),
Expand All @@ -157,7 +207,7 @@ final class CgmDriverRegistry implements CgmDriver {
// manufacturers or reuse its previously cached identity.
}
}
if (cancelled || controller.isClosed) return;
if (aborted()) return;
final source = _transport.scan(
timeout: timeout,
allowDuplicates: true,
Expand Down Expand Up @@ -212,6 +262,10 @@ final class CgmDriverRegistry implements CgmDriver {
},
);
_cancelActiveScan = subscription.cancel;
assert(() {
_debugScanStep(() => 'started a transport scan');
return true;
}(), 'scan diagnostics run only in debug builds');
} on Object catch (error, stackTrace) {
if (_ownsScan(generation, controller)) {
controller.addError(error, stackTrace);
Expand Down Expand Up @@ -247,6 +301,10 @@ final class CgmDriverRegistry implements CgmDriver {
// Keep cleanup fail-closed inside the registry; connect and every
// replacement scan rethrow this latched failure.
_latchScanStopFailure(error, stackTrace);
assert(() {
_debugScanStep(() => 'latched a failed scan cancellation');
return true;
}(), 'scan diagnostics run only in debug builds');
},
);
},
Expand Down Expand Up @@ -280,6 +338,43 @@ final class CgmDriverRegistry implements CgmDriver {
identical(_activeScanController, controller) &&
!controller.isClosed;

/// Ends a scan whose transport stream never closed on its own.
///
/// flutter_blue_plus can stop a physical scan without emitting the
/// `isScanning == false` transition that `FlutterBluePlusTransport` waits
/// for, which leaves the transport stream open past the requested timeout.
/// The registry owns that timeout for every consumer, so the Connect-a-sensor
/// panel and any caller that replaces a scan always observe completion
/// instead of an indefinite spinner.
Future<void> _completeScanAtDeadline(
int generation,
StreamController<DiscoveredSensor> controller,
) async {
if (!_ownsScan(generation, controller)) {
return;
}
_activeScanTimeout?.cancel();
_activeScanTimeout = null;
// Close the result stream first: the same listener drains and releases its
// subscription before the cancellation below is serialized.
unawaited(controller.close());
assert(() {
_debugScanStep(() => 'ended an overrun scan at its deadline');
return true;
}(), 'scan diagnostics run only in debug builds');
try {
await _serialize<void>(() async {
if (identical(_activeScanController, controller)) {
await _stopActiveScanLocked();
}
});
} on Object {
// The result stream is already closed and the transport could not
// confirm its cancellation, which stays latched for the next scan
// attempt. This deadline path must not surface a second unhandled error.
}
}

Future<void> _finishScan(
int generation,
StreamController<DiscoveredSensor> controller,
Expand All @@ -296,6 +391,8 @@ final class CgmDriverRegistry implements CgmDriver {
}
_cancelActiveScan = null;
_activeScanController = null;
_activeScanTimeout?.cancel();
_activeScanTimeout = null;
if (!controller.isClosed) {
await controller.close();
}
Expand All @@ -312,14 +409,33 @@ final class CgmDriverRegistry implements CgmDriver {
final cancelScan = _cancelActiveScan;
final controller = _activeScanController;
_scanGeneration += 1;
_activeScanTimeout?.cancel();
_activeScanTimeout = null;

Object? cancellationError;
StackTrace? cancellationStackTrace;
try {
await cancelScan?.call();
} on Object catch (error, stackTrace) {
cancellationError = error;
cancellationStackTrace = stackTrace;
var cancellationUnconfirmed = false;
if (cancelScan != null) {
// A transport that neither completes nor fails its own cancellation must
// not hold the whole scan lifecycle open.
var cancelled = false;
await Future.any(<Future<void>>[
cancelScan().then<void>(
(_) => cancelled = true,
onError: (Object error, StackTrace stackTrace) {
cancellationError = error;
cancellationStackTrace = stackTrace;
},
),
Future<void>.delayed(_scanCancellationTimeout),
]);
cancellationUnconfirmed = !cancelled && cancellationError == null;
if (cancellationUnconfirmed) {
assert(() {
_debugScanStep(() => 'proceeded past an unconfirmed cancellation');
return true;
}(), 'scan diagnostics run only in debug builds');
}
}

if (controller != null && !controller.isClosed) {
Expand All @@ -329,17 +445,20 @@ final class CgmDriverRegistry implements CgmDriver {
}
}

if (cancellationError != null) {
final cancellationFailure = cancellationError;
if (cancellationFailure != null) {
// StreamSubscription.cancel is not retryable: later calls return the
// same failed future. Latch the failure so no replacement scan or
// routed connection can proceed without proof that the process-global
// BLE scan stopped. Recreating the registry is the recovery boundary.
// An unconfirmed cancellation is deliberately not latched: it is bounded
// in time, and a retry in the same session must stay possible.
_latchScanStopFailure(
cancellationError,
cancellationFailure,
cancellationStackTrace ?? StackTrace.current,
);
Error.throwWithStackTrace(
cancellationError,
cancellationFailure,
cancellationStackTrace ?? StackTrace.current,
);
}
Expand Down
103 changes: 103 additions & 0 deletions openhealth/test/app_controller_persistence_test.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2545,6 +2545,70 @@ void main() {
await scan.timeout(const Duration(seconds: 1));
});

testWidgets('a wedged physical scan cannot leave the controller scanning', (
tester,
) async {
SharedPreferences.setMockInitialValues(<String, Object>{});
final preferences = await SharedPreferences.getInstance();
final transport = _WedgedScanTransport();
final registry = CgmDriverRegistry(
transport: transport,
registrations: <CgmDriverRegistration>[
CgmDriverRegistration(
driver: _ControlledDriver(
const <_ControlledSession>[],
driverId: 'aidex',
),
scanServiceUuids: const <String>['181f'],
discover: (result) => _multiDriverSensor(
driverId: 'aidex',
storageKey: 'aidex:${result.deviceId}',
),
),
],
);
final controller = CgmAppController(
preferences: preferences,
driver: registry,
healthStateStore: _ControllableHealthStateStore(),
);

await controller.initialize();
final scan = controller.scan();
await tester.pump();
expect(transport.scanStarted, isTrue);
transport.emit(
const BleScanResult(
deviceId: 'shared-platform-id',
deviceName: 'alpha',
rssi: -42,
),
);
await tester.pump();
expect(controller.scanning, isTrue);
expect(controller.sensors, hasLength(1));

// Nothing else can end this scan: the transport never closes its stream.
await tester.pump(const Duration(seconds: 7));
await scan;
expect(controller.scanning, isFalse);
expect(controller.sensors, hasLength(1));

// A repeat request must reach the transport again instead of hanging
// behind the wedged scan. The previous cancellation is bounded, so the
// retry starts once that bound elapses.
final repeat = controller.scan();
await tester.pump(const Duration(seconds: 10));
expect(transport.scanCalls, 2);

await tester.pump(const Duration(seconds: 7));
await repeat;
expect(controller.scanning, isFalse);

controller.dispose();
await transport.cancelled.future.timeout(const Duration(seconds: 1));
});

test('initial ready snapshot disables activation on reconnect', () async {
SharedPreferences.setMockInitialValues(<String, Object>{});
final preferences = await SharedPreferences.getInstance();
Expand Down Expand Up @@ -3028,6 +3092,45 @@ final class _BlockingBleTransport implements BleTransport {
}) => throw UnimplementedError();
}

/// A transport that delivers results but never closes its scan stream.
///
/// This mirrors the field failure where the Android radio scan ran, delivered
/// advertisements, and then stopped without the Dart scan stream ever seeing a
/// completion signal.
final class _WedgedScanTransport implements BleTransport {
final Completer<void> cancelled = Completer<void>();
int scanCalls = 0;
bool scanStarted = false;
StreamController<BleScanResult>? _controller;

void emit(BleScanResult result) => _controller?.add(result);

@override
Stream<BleScanResult> scan({
Duration? timeout,
bool allowDuplicates = true,
List<String>? withServices,
}) {
scanCalls += 1;
scanStarted = true;
final controller = StreamController<BleScanResult>(
onCancel: () {
if (!cancelled.isCompleted) {
cancelled.complete();
}
},
);
_controller = controller;
return controller.stream;
}

@override
Future<BleConnection> connect(
String deviceId, {
Duration timeout = const Duration(seconds: 10),
}) => throw UnimplementedError();
}

DiscoveredSensor _multiDriverSensor({
required String driverId,
required String storageKey,
Expand Down
Loading
Loading
0