[dds/dap] Ensure DAP always responds to initialize() before sending initializedEvent

Changes to support DAP in DDS introduced a subtle bug where we could send the `InitializedEvent` before the response to the `Initialize` request because `sendResponse` used a `Completer` and would delay adding the response to the stream until the next iteration of the task queue.

Although there were comments stating this order must be preserved, there were no tests that verified this and it breaks VS Code (https://github.com/dart-lang/sdk/issues/52222).

I've added tests for this, and changed how this works slightly so instead of using a `Completer` we pass in a "responseWriter" function that must write the response to the stream synchronously.

Since the DDS path isn't using events yet and it would require a little more refactoring, I've used a Completer there and added a TODO.

Fixes https://github.com/dart-lang/sdk/issues/52222.

Change-Id: I1832126f5015b466c721dbda192da4b8a5d83b62
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/300601
Reviewed-by: Ben Konyi <bkonyi@google.com>
Commit-Queue: Ben Konyi <bkonyi@google.com>

pkg/dds/lib/src/dap/adapters/dart.dart

@@ -631,7 +631,7 @@ abstract class DartDebugAdapter<TL extends LaunchRequestArguments,
     }
 
     logger?.call('Connecting to debugger at $uri');
-    sendConsoleOutput('Connecting to VM Service at $uri\n');
+    sendConsoleOutput('Connecting to VM Service at $uri');
     final vmService = await _vmServiceConnectUri(uri.toString());
     logger?.call('Connected to debugger at $uri!');
 
@@ -1065,10 +1065,9 @@ abstract class DartDebugAdapter<TL extends LaunchRequestArguments,
     _hasSentTerminatedEvent = true;
 
     // Always add a leading newline since the last written text might not have
-    // had one. Send directly via sendEvent and not sendOutput to ensure no
-    // async since we're about to terminate.
+    // had one.
     final reason = isDetaching ? 'Detached' : 'Exited';
-    sendEvent(OutputEventBody(output: '\n$reason$exitSuffix.'));
+    sendConsoleOutput('\n$reason$exitSuffix.');
     sendEvent(TerminatedEventBody());
   }
 
@@ -1272,9 +1271,14 @@ abstract class DartDebugAdapter<TL extends LaunchRequestArguments,
     sendResponse(ScopesResponseBody(scopes: scopes));
   }
 
-  /// Sends an OutputEvent with a newline to the console.
-  void sendConsoleOutput(String message) {
-    sendOutput('console', '\n$message');
+  /// Sends an OutputEvent with a trailing newline to the console.
+  ///
+  /// This method sends output directly and does not go through [sendOutput]
+  /// because that method is async and queues output. Console output is for
+  /// adapter-level output that does not require this and we want to ensure
+  /// it's sent immediately (for example during shutdown/exit).
+  void sendConsoleOutput(String? message) {
+    sendEvent(OutputEventBody(output: '$message\n'));
   }
 
   /// Sends an OutputEvent (without a newline, since calls to this method

pkg/dds/lib/src/dap/adapters/dart_cli_adapter.dart

@@ -202,9 +202,8 @@ class DartCliDebugAdapter extends DartDebugAdapter<DartLaunchRequestArguments,
     final vmServiceInfoFile = args.vmServiceInfoFile;
 
     if ((vmServiceUri == null) == (vmServiceInfoFile == null)) {
-      sendOutput(
-        'console',
-        '\nTo attach, provide exactly one of vmServiceUri/vmServiceInfoFile',
+      sendConsoleOutput(
+        'To attach, provide exactly one of vmServiceUri/vmServiceInfoFile',
       );
       handleSessionTerminate();
       return;
@@ -253,7 +252,7 @@ class DartCliDebugAdapter extends DartDebugAdapter<DartLaunchRequestArguments,
       );
     } catch (e) {
       logger?.call('Client failed to spawn process $e');
-      sendOutput('console', '\nFailed to spawn process: $e');
+      sendConsoleOutput('Failed to spawn process: $e');
       handleSessionTerminate();
     }
 

pkg/dds/lib/src/dap/adapters/dart_test_adapter.dart

@@ -169,7 +169,7 @@ class DartTestDebugAdapter extends DartDebugAdapter<DartLaunchRequestArguments,
   /// to be debugged.
   @override
   Future<void> attachImpl() async {
-    sendOutput('console', '\nAttach is not supported for test runs');
+    sendConsoleOutput('Attach is not supported for test runs');
     handleSessionTerminate();
   }
 

pkg/dds/lib/src/dap/adapters/dds_hosted_adapter.dart

@@ -105,7 +105,7 @@ class DdsHostedAdapter extends DartDebugAdapter<DartLaunchRequestArguments,
     terminatePids(ProcessSignal.sigterm);
   }
 
-  Future<Response> handleMessage(String message) async {
+  void handleMessage(String message, void Function(Response) responseWriter) {
     final potentialException =
         DebugAdapterException('Message does not conform to DAP spec: $message');
 
@@ -113,10 +113,11 @@ class DdsHostedAdapter extends DartDebugAdapter<DartLaunchRequestArguments,
       final Map<String, Object?> json = jsonDecode(message);
       final type = json['type'] as String;
       if (type == 'request') {
-        return handleIncomingRequest(Request.fromJson(json));
+        handleIncomingRequest(Request.fromJson(json), responseWriter);
         // TODO(helin24): Handle event and response?
+      } else {
+        throw potentialException;
       }
-      throw potentialException;
     } catch (e) {
       throw potentialException;
     }

pkg/dds/lib/src/dap/base_debug_adapter.dart

@@ -102,12 +102,17 @@ abstract class BaseDebugAdapter<TLaunchArgs extends LaunchRequestArguments,
   /// [handler] must _always_ call [sendResponse], even if the response does not
   /// require a body.
   ///
+  /// [responseWriter] is a function that will be provided the response to be
+  /// sent on the stream. It is critical that this sends the response
+  /// synchronously because additional events may be sent by handlers that must
+  /// come after it.
+  ///
   /// If [handler] throws, its exception will be sent as an error response.
   Future<void> handle<TArg, TResp>(
     Request request,
     RequestHandler<TArg, TResp> handler,
     TArg Function(Map<String, Object?>) fromJson,
-    Completer<Response> completer,
+    void Function(Response) responseWriter,
   ) async {
     try {
       final args = request.arguments != null
@@ -131,7 +136,7 @@ abstract class BaseDebugAdapter<TLaunchArgs extends LaunchRequestArguments,
           command: request.command,
           body: responseBody,
         );
-        completer.complete(response);
+        responseWriter(response);
       }
 
       await handler(request, args, sendResponse);
@@ -146,7 +151,7 @@ abstract class BaseDebugAdapter<TLaunchArgs extends LaunchRequestArguments,
         message: e is DebugAdapterException ? e.message : '$e',
         body: '$s',
       );
-      completer.complete(response);
+      responseWriter(response);
     }
   }
 
@@ -271,7 +276,7 @@ abstract class BaseDebugAdapter<TLaunchArgs extends LaunchRequestArguments,
   /// Handles incoming messages from the client editor.
   void _handleIncomingMessage(ProtocolMessage message) {
     if (message is Request) {
-      handleIncomingRequest(message).then(_channel.sendResponse);
+      handleIncomingRequest(message, _channel.sendResponse);
     } else if (message is Response) {
       _handleIncomingResponse(message);
     } else {
@@ -280,152 +285,156 @@ abstract class BaseDebugAdapter<TLaunchArgs extends LaunchRequestArguments,
   }
 
   /// Handles an incoming request, calling the appropriate method to handle it.
-  Future<Response> handleIncomingRequest(Request request) {
-    final completer = Completer<Response>();
-
+  ///
+  /// [responseWriter] is a function that will be provided the response to be
+  /// sent on the stream. It is critical that this sends the response
+  /// synchronously because additional events may be sent by handlers that must
+  /// come after it.
+  void handleIncomingRequest(
+    Request request,
+    void Function(Response) responseWriter,
+  ) {
     if (request.command == 'initialize') {
       handle(
         request,
         initializeRequest,
         InitializeRequestArguments.fromJson,
-        completer,
+        responseWriter,
       );
     } else if (request.command == 'launch') {
       handle(
         request,
         _withVoidResponse(launchRequest),
         parseLaunchArgs,
-        completer,
+        responseWriter,
       );
     } else if (request.command == 'attach') {
       handle(
         request,
         _withVoidResponse(attachRequest),
         parseAttachArgs,
-        completer,
+        responseWriter,
       );
     } else if (request.command == 'restart') {
       handle(
         request,
         _withVoidResponse(restartRequest),
         _allowNullArg(RestartArguments.fromJson),
-        completer,
+        responseWriter,
       );
     } else if (request.command == 'terminate') {
       handle(
         request,
         _withVoidResponse(terminateRequest),
         _allowNullArg(TerminateArguments.fromJson),
-        completer,
+        responseWriter,
       );
     } else if (request.command == 'disconnect') {
       handle(
         request,
         _withVoidResponse(disconnectRequest),
         _allowNullArg(DisconnectArguments.fromJson),
-        completer,
+        responseWriter,
       );
     } else if (request.command == 'configurationDone') {
       handle(
         request,
         _withVoidResponse(configurationDoneRequest),
         _allowNullArg(ConfigurationDoneArguments.fromJson),
-        completer,
+        responseWriter,
       );
     } else if (request.command == 'setBreakpoints') {
       handle(
         request,
         setBreakpointsRequest,
         SetBreakpointsArguments.fromJson,
-        completer,
+        responseWriter,
       );
     } else if (request.command == 'setExceptionBreakpoints') {
       handle(
         request,
         setExceptionBreakpointsRequest,
         SetExceptionBreakpointsArguments.fromJson,
-        completer,
+        responseWriter,
       );
     } else if (request.command == 'continue') {
       handle(
         request,
         continueRequest,
         ContinueArguments.fromJson,
-        completer,
+        responseWriter,
       );
     } else if (request.command == 'next') {
       handle(
         request,
         _withVoidResponse(nextRequest),
         NextArguments.fromJson,
-        completer,
+        responseWriter,
       );
     } else if (request.command == 'stepIn') {
       handle(
         request,
         _withVoidResponse(stepInRequest),
         StepInArguments.fromJson,
-        completer,
+        responseWriter,
       );
     } else if (request.command == 'stepOut') {
       handle(
         request,
         _withVoidResponse(stepOutRequest),
         StepOutArguments.fromJson,
-        completer,
+        responseWriter,
       );
     } else if (request.command == 'threads') {
       handle(
         request,
         threadsRequest,
         _voidArgs,
-        completer,
+        responseWriter,
       );
     } else if (request.command == 'stackTrace') {
       handle(
         request,
         stackTraceRequest,
         StackTraceArguments.fromJson,
-        completer,
+        responseWriter,
       );
     } else if (request.command == 'source') {
       handle(
         request,
         sourceRequest,
         SourceArguments.fromJson,
-        completer,
+        responseWriter,
       );
     } else if (request.command == 'scopes') {
       handle(
         request,
         scopesRequest,
         ScopesArguments.fromJson,
-        completer,
+        responseWriter,
       );
     } else if (request.command == 'variables') {
       handle(
         request,
         variablesRequest,
         VariablesArguments.fromJson,
-        completer,
+        responseWriter,
       );
     } else if (request.command == 'evaluate') {
       handle(
         request,
         evaluateRequest,
         EvaluateArguments.fromJson,
-        completer,
+        responseWriter,
       );
     } else {
       handle(
         request,
         customRequest,
         _allowNullArg(RawRequestArguments.fromJson),
-        completer,
+        responseWriter,
       );
     }
-
-    return completer.future;
   }
 
   void _handleIncomingResponse(Response response) {

pkg/dds/lib/src/dap/isolate_manager.dart

@@ -422,21 +422,18 @@ class IsolateManager {
         return result;
       } else if (result is vm.ErrorRef) {
         final message = result.message ?? '<error ref>';
-        _adapter.sendOutput(
-          'console',
-          'Debugger failed to evaluate breakpoint $type "$expression": $message\n',
+        _adapter.sendConsoleOutput(
+          'Debugger failed to evaluate breakpoint $type "$expression": $message',
         );
       } else if (result is vm.Sentinel) {
         final message = result.valueAsString ?? '<collected>';
-        _adapter.sendOutput(
-          'console',
-          'Debugger failed to evaluate breakpoint $type "$expression": $message\n',
+        _adapter.sendConsoleOutput(
+          'Debugger failed to evaluate breakpoint $type "$expression": $message',
         );
       }
     } catch (e) {
-      _adapter.sendOutput(
-        'console',
-        'Debugger failed to evaluate breakpoint $type "$expression": $e\n',
+      _adapter.sendConsoleOutput(
+        'Debugger failed to evaluate breakpoint $type "$expression": $e',
       );
     }
     return null;
@@ -682,7 +679,7 @@ class IsolateManager {
 
     for (final messageResult in results) {
       // TODO(dantup): Format this using other existing code in protocol converter?
-      _adapter.sendOutput('console', '${messageResult?.valueAsString}\n');
+      _adapter.sendConsoleOutput(messageResult?.valueAsString);
     }
   }
 

pkg/dds/lib/src/dap_handler.dart

@@ -26,7 +26,14 @@ class DapHandler {
     // overlapping sequence numbers with startup requests.
     final message = parameters['message'].asString;
 
-    final result = await adapter.handleMessage(message);
+    // TODO(dantup): If/when DAP needs to care about ordering (eg. it handles
+    //  both requests and events), this will need to be changed to have the
+    //  caller provide a "responseWriter" function so the the result can be
+    //  written directly to the stream synchronously, to avoid future events
+    //  being able to be inserted before the response (eg. initializedEvent).
+    final responseCompleter = Completer<Response>();
+    adapter.handleMessage(message, responseCompleter.complete);
+    final result = await responseCompleter.future;
 
     return <String, dynamic>{
       'type': 'DapResponse',

pkg/dds/test/dap/integration/dart_test_test.dart

@@ -157,6 +157,7 @@ main() {
       final output = outputEvents.output.map((e) => e.output).join();
       expectLines(output, [
         'Attach is not supported for test runs',
+        '',
         'Exited.',
       ]);
     });

pkg/dds/test/dap/integration/debug_attach_test.dart

@@ -41,7 +41,7 @@ main() {
       final vmConnection = outputEvents.first;
       expect(vmConnection.output,
           startsWith('Connecting to VM Service at ws://127.0.0.1:'));
-      expect(vmConnection.category, equals('console'));
+      expect(vmConnection.category, anyOf('console', isNull));
 
       // Expect the normal applications output.
       final output = outputEvents
@@ -94,7 +94,7 @@ main() {
         vmConnection.output,
         startsWith('Connecting to VM Service at ws://127.0.0.1:'),
       );
-      expect(vmConnection.category, equals('console'));
+      expect(vmConnection.category, anyOf('console', isNull));
 
       // Expect the normal applications output.
       final output = outputEvents

pkg/dds/test/dap/integration/debug_logging_test.dart

@@ -83,7 +83,7 @@ void main(List<String> args) async {
       final expectedLogMessage = '[log] $longMessage\n';
 
       final consoleOutputs = dap.client.outputEvents
-          .where((event) => event.category == 'console')
+          .where((event) => (event.category ?? 'console') == 'console')
           .map((event) => event.output);
 
       await Future.wait([

pkg/dds/test/dap/integration/debug_test.dart

@@ -37,7 +37,7 @@ main() {
       final vmConnection = outputEvents.first;
       expect(vmConnection.output,
           startsWith('Connecting to VM Service at ws://127.0.0.1:'));
-      expect(vmConnection.category, equals('console'));
+      expect(vmConnection.category, anyOf('console', isNull));
 
       // Expect the normal applications output.
       final output = outputEvents.skip(1).map((e) => e.output).join();

pkg/dds/test/dap/integration/test_client.dart

@@ -468,7 +468,8 @@ class DapTestClient {
 
   /// Handles an incoming message from the server, completing the relevant request
   /// of raising the appropriate event.
-  Future<void> _handleMessage(message) async {
+  Future<void> _handleMessage(ProtocolMessage message) async {
+    _verifyMessageOrdering(message);
     if (message is Response) {
       final pendingRequest = _pendingRequests.remove(message.requestSeq);
       if (pendingRequest == null) {
@@ -520,6 +521,43 @@ class DapTestClient {
     return future.whenComplete(timer.cancel);
   }
 
+  /// Ensures that protocol messages are received in the correct order where
+  /// ordering matters.
+  void _verifyMessageOrdering(ProtocolMessage message) {
+    if (message is Event) {
+      if (message.event == 'initialized' &&
+          !_receivedResponses.contains('initialize')) {
+        throw StateError(
+          'Adapter sent "initialized" event before the "initialize" request had '
+          'been responded to',
+        );
+      }
+
+      _receivedEvents.add(message.event);
+    } else if (message is Response) {
+      if (message.command == 'initialize' &&
+          _receivedEvents.contains('initialized')) {
+        throw StateError(
+          'Adapter sent a response to an "initialize" request after it had '
+          'already send the "initialized" event',
+        );
+      }
+
+      _receivedResponses.add(message.command);
+    }
+  }
+
+  /// A list of all event names that have been received during this session.
+  ///
+  /// Used by [_verifyMessageOrdering].
+  final _receivedEvents = <String>{};
+
+  /// A list of all request names that have been responded to during this
+  /// session.
+  ///
+  /// Used by [_verifyMessageOrdering].
+  final _receivedResponses = <String>{};
+
   /// Creates a [DapTestClient] that connects the server listening on
   /// [host]:[port].
   static Future<DapTestClient> connect(