[DDS] Start caching events sent on the 'Timer' stream

TEST=pkg/dds/test/timer_event_history_test

Change-Id: Id1b867e5f582aca9d57c6d52978880bbbd889b27
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/411160
Reviewed-by: Ben Konyi <[email protected]>
Commit-Queue: Derek Xu <[email protected]>

Author: derekxu16

pkg/dds/CHANGELOG.md

@@ -2,6 +2,7 @@
 
 - Widen the dependency on `package:shelf_web_socket`.
 - Require Dart SDK v. 3.5.0 or higher.
+- Started caching events sent on the 'Timer' stream. The cached events can be retrieved using the `getStreamHistory` RPC.
 
 # 5.0.0
 - [DAP] The debug adapter no longer spawns its own in-process copy of DDS, instead relying on one started by the Dart VM (or `Flutter`). This means the `enableDds` and `enableAuthCodes` arguments to the `DartDebugAdapter` base class have been deprecated and have any effect. Suppressing DDS (or auth codes) should be done in launch configuration (for example using `vmAdditionalArgs` or `toolArgs` depending on the target tool).

pkg/dds/lib/src/stream_manager.dart

@@ -427,6 +427,7 @@ class StreamManager {
   static const kEchoStream = '_Echo';
   static const kDebugStream = 'Debug';
   static const kExtensionStream = 'Extension';
+  static const kTimerStream = 'Timer';
   static const kHeapSnapshotStream = 'HeapSnapshot';
   static const kIsolateStream = 'Isolate';
   static const kGCStream = 'GC';
@@ -463,10 +464,11 @@ class StreamManager {
     kIsolateStream,
   };
 
-  // Never cancel the logging and extension event streams as `LoggingRepository`
-  // requires them keep history.
+  // Never cancel the logging, timer, and extension event streams as
+  // `LoggingRepository` requires them keep history.
   static const loggingRepositoryStreams = <String>{
     kExtensionStream,
+    kTimerStream,
     kLoggingStream,
     kStderrStream,
     kStdoutStream,

pkg/dds/test/extension_event_history_test.dart

@@ -5,6 +5,7 @@
 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';
 
@@ -22,7 +23,6 @@ Future testMain() async {
 
 final tests = <IsolateTest>[
   hasPausedAtStart,
-  resumeIsolate,
   (VmService service, IsolateRef isolateRef) async {
     final completer = Completer<void>();
     int i = 1;
@@ -39,8 +39,31 @@ final tests = <IsolateTest>[
       }
     });
     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].
+    final completer = Completer<void>();
+    int i = 1;
+    late final StreamSubscription subscription;
+    subscription = service.onExtensionEventWithHistory.listen((event) async {
+      expect(event.extensionKind, 'Test');
+      expect(event.extensionData!.data['id'], i);
+      i++;
+
+      if (i == 10) {
+        await subscription.cancel();
+        completer.complete();
+      } else if (i > 10) {
+        fail('Too many "Test" extension events');
+      }
+    });
+    await service.streamListen(EventStreams.kExtension);
+    await completer.future;
+  }
 ];
 
 void main([args = const <String>[]]) => runIsolateTests(

pkg/dds/test/timer_event_history_test.dart

@@ -0,0 +1,72 @@
+// 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:io' show sleep;
+
+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<void> testeeMain() async {
+  final completer = Completer<void>();
+  late final Timer t;
+  t = Timer(
+    const Duration(milliseconds: 100),
+    () {
+      t.cancel();
+      completer.complete();
+    },
+  );
+
+  // Sleep for 201 ms to force [t] to fire at least 100 ms late. This allows us
+  // to expect to receive at least one 'TimerSignificantlyOverdue' event in
+  // [tests] below, because a 'TimerSignificantlyOverdue' event should be fired
+  // whenever a timer is identified to be at least 100 ms overdue.
+  sleep(const Duration(milliseconds: 201));
+  await completer.future;
+}
+
+final tests = <IsolateTest>[
+  hasPausedAtStart,
+  (VmService service, IsolateRef isolateRef) async {
+    final completer = Completer<void>();
+    service.onTimerEvent.listen((event) async {
+      expect(event.kind, 'TimerSignificantlyOverdue');
+
+      await service.streamCancel(EventStreams.kTimer);
+      completer.complete();
+    });
+    await service.streamListen(EventStreams.kTimer);
+
+    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.onTimerEventWithHistory].
+    final completer = Completer<void>();
+    late final StreamSubscription subscription;
+    subscription = service.onTimerEventWithHistory.listen((event) async {
+      expect(event.kind, 'TimerSignificantlyOverdue');
+
+      await subscription.cancel();
+      completer.complete();
+    });
+    await service.streamListen(EventStreams.kTimer);
+    await completer.future;
+  }
+];
+
+void main([args = const <String>[]]) => runIsolateTests(
+      args,
+      tests,
+      'timer_event_history_test.dart',
+      testeeConcurrent: testeeMain,
+      pauseOnStart: true,
+      pauseOnExit: true,
+    );

pkg/dds_service_extensions/CHANGELOG.md

@@ -1,5 +1,6 @@
 # 2.0.2-wip
 - Require dart sdk v. 3.5.0 or higher.
+- Add `DdsExtension.onTimerEventWithHistory`.
 
 # 2.0.1
 - Update `vm_service` to `>=14.0.0 <16.0.0`.

pkg/dds_service_extensions/lib/dds_service_extensions.dart

@@ -202,6 +202,13 @@ extension DdsExtension on VmService {
   Stream<Event> get onExtensionEventWithHistory =>
       onEventWithHistory('Extension');
 
+  /// Returns a new [Stream<Event>] of events sent on the `Timer` stream which
+  /// outputs historical events before streaming real-time events.
+  ///
+  /// Note: unlike [onTimerEvent], the returned stream is a single subscription
+  /// stream and a new stream is created for each invocation of this getter.
+  Stream<Event> get onTimerEventWithHistory => onEventWithHistory('Timer');
+
   /// The [getClientName] RPC is used to retrieve the name associated with the
   /// currently connected VM service client.
   ///