[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 <bkonyi@google.com>
Commit-Queue: Derek Xu <derekx@google.com>

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 = <IsolateTest>[
-  hasPausedAtStart,
-  (VmService service, IsolateRef isolateRef) async {
-    final completer = Completer<void>();
-    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<void>();
     int i = 1;
     late final StreamSubscription subscription;
@@ -71,6 +55,5 @@ void main([args = const <String>[]]) => runIsolateTests(
       tests,
       'extension_event_history_test.dart',
       testeeConcurrent: testMain,
-      pauseOnStart: true,
       pauseOnExit: true,
     );

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 = <IsolateTest>[
+  (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<void>();
+    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 <String>[]]) => runIsolateTests(
+      args,
+      tests,
+      'logging_event_history_test.dart',
+      testeeConcurrent: testMain,
+      pauseOnExit: true,
+    );

pkg/dds/test/on_event_with_history_script.dart

@@ -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();
-}

pkg/dds/test/on_event_with_history_test.dart

@@ -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<void>();
-    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<void>();
-    final isolateId = (await service.getVM()).isolates!.first.id!;
-    await service.resume(isolateId);
-
-    await completer.future;
-    expect(count, 20);
-  });
-}

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`.

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<Event> onEventWithHistory(String stream) {
     late StreamController<Event> controller;
-    late StreamQueue<Event> streamEvents;
+    late StreamSubscription<Event> subscription;
 
     controller = StreamController<Event>(onListen: () async {
-      streamEvents = StreamQueue<Event>(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<void>();
+
+      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.
       }

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'