[ DWDS ] Disconnect non-DDS clients when DDS connects

bkonyi · bkonyi · commit 7e6db356cf88 · 2025-08-20T14:46:51.000-04:00
The native VM service disconnects all non-DDS clients when DDS connects in order to keep state consistent. DWDS was never configured to work this way and, instead, would not let DDS connect if there were already existing clients. This is fine most of the time, as DDS is typically the first client, but in cases where multiple DDS instances attempt to start at once (e.g., a simultaneous `flutter run` and `flutter attach`), this inconsistency prevents tools from falling back to using an already connected DDS instance, as seen in flutter/flutter#171758. Fixes #2399
diff --git a/dwds/lib/src/services/chrome_proxy_service.dart b/dwds/lib/src/services/chrome_proxy_service.dart
@@ -33,6 +33,38 @@ import 'package:vm_service/vm_service.dart' hide vmServiceVersion;
 import 'package:vm_service_interface/vm_service_interface.dart';
 import 'package:webkit_inspection_protocol/webkit_inspection_protocol.dart';
 
+// This event is identical to the one sent by the VM service from
+// sdk/lib/vmservice/vmservice.dart before existing VM service clients are
+// disconnected.
+final class DartDevelopmentServiceConnectedEvent extends Event {
+  DartDevelopmentServiceConnectedEvent({
+    required super.timestamp,
+    required this.uri,
+  }) : message =
+           'A Dart Developer Service instance has connected and this direct '
+               'connection to the VM service will now be closed. Please reconnect to '
+               'the Dart Development Service at $uri.',
+       super(kind: 'DartDevelopmentServiceConnected');
+
+  final String message;
+  final String uri;
+
+  @override
+  Map<String, Object?> toJson() => {
+    ...super.toJson(),
+    'uri': uri,
+    'message': message,
+  };
+}
+
+final class DisconnectNonDartDevelopmentServiceClients extends RPCError {
+  DisconnectNonDartDevelopmentServiceClients()
+    : super('_yieldControlToDDS', kErrorCode);
+
+  // Arbitrary error code that's unlikely to be used elsewhere.
+  static const kErrorCode = -199328;
+}
+
 /// A proxy from the chrome debug protocol to the dart vm service protocol.
 class ChromeProxyService extends ProxyService {
   /// Signals when isolate starts.
@@ -1563,15 +1595,21 @@ class ChromeProxyService extends ProxyService {
 
   @override
   Future<void> yieldControlToDDS(String uri) async {
-    final canYield = ChromeDebugService.yieldControlToDDS(uri);
+    // This will throw an RPCError if there's already an existing DDS instance.
+    ChromeDebugService.yieldControlToDDS(uri);
+
+    // Notify existing clients that DDS has connected and they're about to be
+    // disconnected.
+    final event = DartDevelopmentServiceConnectedEvent(
+      timestamp: DateTime.now().millisecondsSinceEpoch,
+      uri: uri,
+    );
+    streamNotify(EventStreams.kService, event);
 
-    if (!canYield) {
-      throw RPCError(
-        'yieldControlToDDS',
-        RPCErrorKind.kFeatureDisabled.code,
-        'Existing VM service clients prevent DDS from taking control.',
-      );
-    }
+    // We throw since we have no other way to control what the response content
+    // is for this RPC. The debug service will check for this particular
+    // exception as a signal to close connections to all other clients.
+    throw DisconnectNonDartDevelopmentServiceClients();
   }
 
   Future<InstanceRef> _instanceRef(RemoteObject? obj) async {
diff --git a/dwds/lib/src/services/debug_service.dart b/dwds/lib/src/services/debug_service.dart
@@ -25,57 +25,101 @@ import 'package:shelf/shelf.dart' as shelf;
 import 'package:shelf/shelf.dart' hide Response;
 import 'package:shelf_web_socket/shelf_web_socket.dart';
 import 'package:sse/server/sse_handler.dart';
+import 'package:stream_channel/stream_channel.dart';
+import 'package:vm_service/vm_service.dart';
 import 'package:vm_service_interface/vm_service_interface.dart';
 import 'package:web_socket_channel/web_socket_channel.dart';
 
 const _kSseHandlerPath = '\$debugHandler';
 
 bool _acceptNewConnections = true;
-int _clientsConnected = 0;
+
+final _clientConnections = <int, StreamChannel>{};
+int _clientId = 0;
 
 Logger _logger = Logger('DebugService');
 
+void _handleConnection(
+  StreamChannel channel,
+  ChromeProxyService chromeProxyService,
+  ServiceExtensionRegistry serviceExtensionRegistry, {
+  void Function(Map<String, Object>)? onRequest,
+  void Function(Map<String, Object?>)? onResponse,
+}) {
+  final clientId = _clientId++;
+  final responseController = StreamController<Map<String, Object?>>();
+  responseController.stream
+      .asyncMap((response) async {
+        // This error indicates a successful invocation to _yieldControlToDDS.
+        // We don't have a good way to access the list of connected clients
+        // while also being able to determine which client invoked the RPC
+        // without some form of client ID.
+        //
+        // We can probably do better than this, but it will likely involve some
+        // refactoring.
+        if (response case {
+          'error': {
+            'code': DisconnectNonDartDevelopmentServiceClients.kErrorCode,
+          },
+        }) {
+          final nonDdsClients = _clientConnections.entries
+              .where((MapEntry<int, StreamChannel> e) => e.key != clientId)
+              .map((e) => e.value);
+          await Future.wait([
+            for (final client in nonDdsClients) client.sink.close(),
+          ]);
+          // Remove the artificial error and return Success.
+          response.remove('error');
+          response['result'] = Success().toJson();
+        }
+        if (onResponse != null) onResponse(response);
+        channel.sink.add(jsonEncode(response));
+      })
+      .listen(channel.sink.add, onError: channel.sink.addError);
+  final inputStream = channel.stream.map((value) {
+    if (value is List<int>) {
+      value = utf8.decode(value);
+    } else if (value is! String) {
+      throw StateError(
+        'Got value with unexpected type ${value.runtimeType} from web '
+        'socket, expected a List<int> or String.',
+      );
+    }
+    final request = Map<String, Object>.from(jsonDecode(value));
+    if (onRequest != null) onRequest(request);
+    return request;
+  });
+  VmServerConnection(
+    inputStream,
+    responseController.sink,
+    serviceExtensionRegistry,
+    chromeProxyService,
+  ).done.whenComplete(() {
+    _clientConnections.remove(clientId);
+    if (!_acceptNewConnections && _clientConnections.isEmpty) {
+      // DDS has disconnected so we can allow for clients to connect directly
+      // to DWDS.
+      ChromeDebugService._ddsUri = null;
+      _acceptNewConnections = true;
+    }
+  });
+  _clientConnections[clientId] = channel;
+}
+
 void Function(WebSocketChannel, String?) _createNewConnectionHandler(
   ChromeProxyService chromeProxyService,
   ServiceExtensionRegistry serviceExtensionRegistry, {
   void Function(Map<String, Object>)? onRequest,
   void Function(Map<String, Object?>)? onResponse,
 }) {
   return (webSocket, subprotocol) {
-    final responseController = StreamController<Map<String, Object?>>();
-    webSocket.sink.addStream(
-      responseController.stream.map((response) {
-        if (onResponse != null) onResponse(response);
-        return jsonEncode(response);
-      }),
-    );
-    final inputStream = webSocket.stream.map((value) {
-      if (value is List<int>) {
-        value = utf8.decode(value);
-      } else if (value is! String) {
-        throw StateError(
-          'Got value with unexpected type ${value.runtimeType} from web '
-          'socket, expected a List<int> or String.',
-        );
-      }
-      final request = Map<String, Object>.from(jsonDecode(value));
-      if (onRequest != null) onRequest(request);
-      return request;
-    });
-    ++_clientsConnected;
-    VmServerConnection(
-      inputStream,
-      responseController.sink,
-      serviceExtensionRegistry,
+    _handleConnection(
+      webSocket,
       chromeProxyService,
-    ).done.whenComplete(() {
-      --_clientsConnected;
-      if (!_acceptNewConnections && _clientsConnected == 0) {
-        // DDS has disconnected so we can allow for clients to connect directly
-        // to DWDS.
-        _acceptNewConnections = true;
-      }
-    });
+      serviceExtensionRegistry,
+      onRequest: onRequest,
+      onResponse: onResponse,
+    );
   };
 }
 
@@ -88,41 +132,12 @@ Future<void> _handleSseConnections(
 }) async {
   while (await handler.connections.hasNext) {
     final connection = await handler.connections.next;
-    final responseController = StreamController<Map<String, Object?>>();
-    final sub = responseController.stream
-        .map((response) {
-          if (onResponse != null) onResponse(response);
-          return jsonEncode(response);
-        })
-        .listen(connection.sink.add);
-    safeUnawaited(
-      chromeProxyService.remoteDebugger.onClose.first.whenComplete(() {
-        connection.sink.close();
-        sub.cancel();
-      }),
-    );
-    final inputStream = connection.stream.map((value) {
-      final request = jsonDecode(value) as Map<String, Object>;
-      if (onRequest != null) onRequest(request);
-      return request;
-    });
-    ++_clientsConnected;
-    final vmServerConnection = VmServerConnection(
-      inputStream,
-      responseController.sink,
-      serviceExtensionRegistry,
+    _handleConnection(
+      connection,
       chromeProxyService,
-    );
-    safeUnawaited(
-      vmServerConnection.done.whenComplete(() {
-        --_clientsConnected;
-        if (!_acceptNewConnections && _clientsConnected == 0) {
-          // DDS has disconnected so we can allow for clients to connect directly
-          // to DWDS.
-          _acceptNewConnections = true;
-        }
-        return sub.cancel();
-      }),
+      serviceExtensionRegistry,
+      onRequest: onRequest,
+      onResponse: onResponse,
     );
   }
 }
@@ -224,15 +239,19 @@ class ChromeDebugService implements DebugService {
     return _encodedUri = encoded;
   }
 
-  // TODO(https://github.com/dart-lang/webdev/issues/2399): yieldControlToDDS
-  // should disconnect existing non-DDS clients.
-  static bool yieldControlToDDS(String uri) {
-    if (_clientsConnected > 1) {
-      return false;
+  static void yieldControlToDDS(String uri) {
+    if (_ddsUri != null) {
+      // This exception is identical to the one thrown from
+      // sdk/lib/vmservice/vmservice.dart
+      throw RPCError(
+        '_yieldControlToDDS',
+        RPCErrorKind.kFeatureDisabled.code,
+        'A DDS instance is already connected at $_ddsUri.',
+        {'ddsUri': _ddsUri.toString()},
+      );
     }
-    _ddsUri = uri;
     _acceptNewConnections = false;
-    return true;
+    _ddsUri = uri;
   }
 
   static Future<ChromeDebugService> start(
@@ -441,6 +460,7 @@ class WebSocketDebugService implements DebugService {
     WebSocketProxyService webSocketProxyService,
   ) {
     return webSocketHandler((WebSocketChannel webSocket, String? subprotocol) {
+      final clientId = _clientId++;
       final responseController = StreamController<Map<String, Object?>>();
       webSocket.sink.addStream(responseController.stream.map(jsonEncode));
 
@@ -455,14 +475,15 @@ class WebSocketDebugService implements DebugService {
         return Map<String, Object>.from(jsonDecode(value));
       });
 
-      ++_clientsConnected;
+      _clientConnections[clientId] = webSocket;
+
       VmServerConnection(
         inputStream,
         responseController.sink,
         serviceExtensionRegistry,
         webSocketProxyService,
       ).done.whenComplete(() {
-        --_clientsConnected;
+        _clientConnections.remove(clientId);
       });
     });
   }
diff --git a/dwds/pubspec.yaml b/dwds/pubspec.yaml
@@ -32,6 +32,7 @@ dependencies:
   shelf_web_socket: '>=2.0.0 <4.0.0'
   source_maps: ^0.10.10
   stack_trace: ^1.10.0
+  stream_channel: ^2.1.4
   sse: ^4.1.2
   uuid: ^4.0.0
   vm_service: '>=14.2.4 <16.0.0'
diff --git a/dwds/test/debug_service_test.dart b/dwds/test/debug_service_test.dart
