From 9e50790fec3723ea67177a59ad920fe9f6b24144 Mon Sep 17 00:00:00 2001 From: Derek Xu Date: Wed, 21 May 2025 09:14:35 -0700 Subject: [PATCH] [DDS] Fix DdsExtension.onEventWithHistory TEST=confirmed that pkg/dds/test/extension_event_history_test and pkg/dds/test/logging_event_history_test pass with the changes in this CL and fail without them Fixes: https://github.com/dart-lang/sdk/issues/60672 Change-Id: Ie8c7e5908e80763e5d6e260fdb07f5fe65cb6232 Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/429100 Reviewed-by: Ben Konyi Commit-Queue: Derek Xu --- .../test/extension_event_history_test.dart | 31 +++------- pkg/dds/test/logging_event_history_test.dart | 57 ++++++++++++++++++ .../test/on_event_with_history_script.dart | 12 ---- pkg/dds/test/on_event_with_history_test.dart | 60 ------------------- pkg/dds_service_extensions/CHANGELOG.md | 3 + .../lib/dds_service_extensions.dart | 42 ++++++++----- pkg/dds_service_extensions/pubspec.yaml | 1 - 7 files changed, 94 insertions(+), 112 deletions(-) create mode 100644 pkg/dds/test/logging_event_history_test.dart delete mode 100644 pkg/dds/test/on_event_with_history_script.dart delete mode 100644 pkg/dds/test/on_event_with_history_test.dart diff --git a/pkg/dds/test/extension_event_history_test.dart b/pkg/dds/test/extension_event_history_test.dart index 30759cdc359..2ca1062b712 100644 --- a/pkg/dds/test/extension_event_history_test.dart +++ b/pkg/dds/test/extension_event_history_test.dart @@ -9,7 +9,6 @@ import 'package:dds_service_extensions/dds_service_extensions.dart'; import 'package:test/test.dart'; import 'package:vm_service/vm_service.dart'; -import 'common/service_test_common.dart'; import 'common/test_helper.dart'; Future testMain() async { @@ -18,34 +17,19 @@ Future testMain() async { postEvent('Test', { 'id': i, }); + // Wait between posting events to make it more likely for the test below to + // exercise the logic that makes [service.onExtensionEventWithHistory] + // return both historical and future events. + await Future.delayed(const Duration(milliseconds: 50)); } } final tests = [ - hasPausedAtStart, - (VmService service, IsolateRef isolateRef) async { - final completer = Completer(); - int i = 1; - service.onExtensionEvent.listen((event) async { - expect(event.extensionKind, 'Test'); - expect(event.extensionData!.data['id'], i); - i++; - - if (i == 10) { - await service.streamCancel(EventStreams.kExtension); - completer.complete(); - } else if (i > 10) { - fail('Too many "Test" extension events'); - } - }); - await service.streamListen(EventStreams.kExtension); - - resumeIsolate(service, isolateRef); - await completer.future; - }, (VmService service, _) async { // Confirm that all events in the history buffer get sent on a stream - // returned by [service.onExtensionEventWithHistory]. + // returned by [service.onExtensionEventWithHistory], and that all events + // posted after a listener has been added to the returned stream get sent to + // that listener. final completer = Completer(); int i = 1; late final StreamSubscription subscription; @@ -71,6 +55,5 @@ void main([args = const []]) => runIsolateTests( tests, 'extension_event_history_test.dart', testeeConcurrent: testMain, - pauseOnStart: true, pauseOnExit: true, ); diff --git a/pkg/dds/test/logging_event_history_test.dart b/pkg/dds/test/logging_event_history_test.dart new file mode 100644 index 00000000000..86f523b4439 --- /dev/null +++ b/pkg/dds/test/logging_event_history_test.dart @@ -0,0 +1,57 @@ +// Copyright (c) 2025, the Dart project authors. Please see the AUTHORS file +// for details. All rights reserved. Use of this source code is governed by a +// BSD-style license that can be found in the LICENSE file. + +import 'dart:async'; +import 'dart:developer'; + +import 'package:dds_service_extensions/dds_service_extensions.dart'; +import 'package:test/test.dart'; +import 'package:vm_service/vm_service.dart'; + +import 'common/test_helper.dart'; + +Future testMain() async { + // Trigger the posting of 9 logging events. + for (int i = 1; i <= 9; ++i) { + log(i.toString()); + // Wait between posting events to make it more likely for the test below to + // exercise the logic that makes [service.onLoggingEventWithHistory] return + // both historical and future events. + await Future.delayed(const Duration(milliseconds: 50)); + } +} + +final tests = [ + (VmService service, _) async { + // Confirm that all events in the history buffer get sent on a stream + // returned by [service.onLoggingEventWithHistory], and that all events + // posted after a listener has been added to the returned stream get sent to + // that listener. + final completer = Completer(); + int i = 1; + late final StreamSubscription subscription; + subscription = service.onLoggingEventWithHistory.listen((event) async { + expect(event.kind, 'Logging'); + expect(event.logRecord!.message!.valueAsString!, i.toString()); + i++; + + if (i == 10) { + await subscription.cancel(); + completer.complete(); + } else if (i > 10) { + fail('Too many "Logging" events encountered'); + } + }); + await service.streamListen(EventStreams.kLogging); + await completer.future; + } +]; + +void main([args = const []]) => runIsolateTests( + args, + tests, + 'logging_event_history_test.dart', + testeeConcurrent: testMain, + pauseOnExit: true, + ); diff --git a/pkg/dds/test/on_event_with_history_script.dart b/pkg/dds/test/on_event_with_history_script.dart deleted file mode 100644 index 3f27df0c347..00000000000 --- a/pkg/dds/test/on_event_with_history_script.dart +++ /dev/null @@ -1,12 +0,0 @@ -import 'dart:developer'; - -void main() { - for (int i = 1; i <= 10; ++i) { - log(i.toString()); - } - debugger(); - for (int i = 11; i <= 20; ++i) { - log(i.toString()); - } - debugger(); -} diff --git a/pkg/dds/test/on_event_with_history_test.dart b/pkg/dds/test/on_event_with_history_test.dart deleted file mode 100644 index 5365b03f6b3..00000000000 --- a/pkg/dds/test/on_event_with_history_test.dart +++ /dev/null @@ -1,60 +0,0 @@ -// Copyright (c) 2021, the Dart project authors. Please see the AUTHORS file -// for details. All rights reserved. Use of this source code is governed by a -// BSD-style license that can be found in the LICENSE file. - -import 'dart:async'; -import 'dart:io'; - -import 'package:dds/dds.dart'; -import 'package:dds_service_extensions/dds_service_extensions.dart'; -import 'package:test/test.dart'; -import 'package:vm_service/vm_service_io.dart'; -import 'common/test_helper.dart'; - -void main() { - late Process process; - late DartDevelopmentService dds; - - setUp(() async { - process = await spawnDartProcess( - 'on_event_with_history_script.dart', - ); - }); - - tearDown(() async { - await dds.shutdown(); - process.kill(); - }); - - test('onEventWithHistory returns stream including log history', () async { - dds = await DartDevelopmentService.startDartDevelopmentService( - remoteVmServiceUri, - ); - expect(dds.isRunning, true); - final service = await vmServiceConnectUri(dds.wsUri.toString()); - - // Wait until the test script has finished writing its initial logs. - await executeUntilNextPause(service); - - await service.streamListen('Logging'); - final stream = service.onLoggingEventWithHistory; - - var completer = Completer(); - int count = 0; - stream.listen((event) { - count++; - expect(event.logRecord!.message!.valueAsString, count.toString()); - if (count % 10 == 0) { - completer.complete(); - } - }); - await completer.future; - - completer = Completer(); - final isolateId = (await service.getVM()).isolates!.first.id!; - await service.resume(isolateId); - - await completer.future; - expect(count, 20); - }); -} diff --git a/pkg/dds_service_extensions/CHANGELOG.md b/pkg/dds_service_extensions/CHANGELOG.md index 519cb930ca9..a98bd9a1617 100644 --- a/pkg/dds_service_extensions/CHANGELOG.md +++ b/pkg/dds_service_extensions/CHANGELOG.md @@ -1,6 +1,9 @@ # 2.0.2-wip - Require dart sdk v. 3.5.0 or higher. - Add `DdsExtension.onTimerEventWithHistory`. +- Fix a bug that could make any `DdsExtension.on*EventWithHistory` stream + contain an incorrect set of events. +- Remove dependency on `package:async`. # 2.0.1 - Update `vm_service` to `>=14.0.0 <16.0.0`. diff --git a/pkg/dds_service_extensions/lib/dds_service_extensions.dart b/pkg/dds_service_extensions/lib/dds_service_extensions.dart index 15230382ec6..92096775315 100644 --- a/pkg/dds_service_extensions/lib/dds_service_extensions.dart +++ b/pkg/dds_service_extensions/lib/dds_service_extensions.dart @@ -5,7 +5,6 @@ import 'dart:async'; import 'dart:collection'; -import 'package:async/async.dart'; // ignore: implementation_imports import 'package:vm_service/src/vm_service.dart'; @@ -141,26 +140,39 @@ extension DdsExtension on VmService { /// sent over the returned [Stream]. Stream onEventWithHistory(String stream) { late StreamController controller; - late StreamQueue streamEvents; + late StreamSubscription subscription; controller = StreamController(onListen: () async { - streamEvents = StreamQueue(onEvent(stream)); - final history = (await getStreamHistory(stream)).history; - Event? firstStreamEvent; - unawaited(streamEvents.peek.then((e) { - firstStreamEvent = e; - })); - for (final event in history) { - if (firstStreamEvent != null && - event.timestamp! > firstStreamEvent!.timestamp!) { - break; + bool isFirstEvent = true; + late final Event? lastHistoricalEvent; + final historyRetrievedCompleter = Completer(); + + subscription = onEvent(stream).listen((event) async { + if (isFirstEvent) { + await historyRetrievedCompleter.future; + isFirstEvent = false; } - controller.sink.add(event); + // In practice, it should be impossible for two distinct events to have + // the same timestamp. + if (lastHistoricalEvent == null || + event.timestamp! > lastHistoricalEvent.timestamp!) { + controller.sink.add(event); + } + }, onDone: () { + controller.sink.close(); + }, onError: (error, stackTrace) { + controller.addError(error, stackTrace); + }); + + final history = (await getStreamHistory(stream)).history; + lastHistoricalEvent = history.isNotEmpty ? history.last : null; + for (final historicalEvent in history) { + controller.sink.add(historicalEvent); } - unawaited(controller.sink.addStream(streamEvents.rest)); + historyRetrievedCompleter.complete(); }, onCancel: () { try { - streamEvents.cancel(); + subscription.cancel(); } on StateError { // Underlying stream may have already been cancelled. } diff --git a/pkg/dds_service_extensions/pubspec.yaml b/pkg/dds_service_extensions/pubspec.yaml index 5378711d322..a801070fdaa 100644 --- a/pkg/dds_service_extensions/pubspec.yaml +++ b/pkg/dds_service_extensions/pubspec.yaml @@ -11,7 +11,6 @@ environment: resolution: workspace dependencies: - async: ^2.4.1 dap: ^1.0.0 vm_service: '>=14.0.0 <16.0.0'