fix(diagnostics): parse after queued document changes

lib/ruby_lsp/base_server.rb

@@ -38,47 +38,13 @@ def initialize(**options)
     def start
       @reader.each_message do |message|
         method = message[:method]
+        uri = message.dig(:params, :textDocument, :uri)
 
-        # We must parse the document under a mutex lock or else we might switch threads and accept text edits in the
-        # source. Altering the source reference during parsing will put the parser in an invalid internal state, since
-        # it started parsing with one source but then it changed in the middle. We don't want to do this for text
-        # synchronization notifications
-        @global_state.synchronize do
-          uri = message.dig(:params, :textDocument, :uri)
-
-          if uri
-            begin
-              parsed_uri = URI(uri)
-              message[:params][:textDocument][:uri] = parsed_uri
-
-              # We don't want to try to parse documents on text synchronization notifications
-              unless method.start_with?("textDocument/did")
-                document = @store.get(parsed_uri)
-
-                # If the client supports request delegation and we're working with an ERB document and there was
-                # something to parse, then we have to maintain the client updated about the virtual state of the host
-                # language source
-                if document.parse! && @global_state.client_capabilities.supports_request_delegation &&
-                    document.is_a?(ERBDocument)
-
-                  send_message(
-                    Notification.new(
-                      method: "delegate/textDocument/virtualState",
-                      params: {
-                        textDocument: {
-                          uri: uri,
-                          text: document.host_language_source,
-                        },
-                      },
-                    ),
-                  )
-                end
-              end
-            rescue URI::Error
-              # A client can send a document URI with a scheme Ruby's URI parser rejects, so we don't want to fail
-            rescue Store::NonExistingDocumentError
-              # If we receive a request for a file that no longer exists, we don't want to fail
-            end
+        if uri
+          begin
+            message[:params][:textDocument][:uri] = URI(uri)
+          rescue URI::Error
+            # A client can send a document URI with a scheme Ruby's URI parser rejects, so we don't want to fail
           end
         end
 
@@ -165,6 +131,10 @@ def new_worker
     def handle_incoming_message(message)
       id = message[:id]
 
+      # Parse in the worker after earlier document notifications have been applied. This also keeps document edits from
+      # changing the source while it is being parsed.
+      parse_document(message)
+
       # Check if the request was cancelled before trying to process it
       @global_state.synchronize do
         if id && @cancelled_requests.delete(id)
@@ -183,6 +153,39 @@ def handle_incoming_message(message)
       @cancelled_requests.delete(id)
     end
 
+    #: (Hash[Symbol, untyped] message) -> void
+    def parse_document(message)
+      method = message[:method]
+      return unless method.is_a?(String) && !method.start_with?("textDocument/did")
+
+      uri = message.dig(:params, :textDocument, :uri)
+      return unless uri.is_a?(URI::Generic)
+
+      @global_state.synchronize do
+        document = @store.get(uri)
+
+        # If the client supports request delegation and we're working with an ERB document and there was something to
+        # parse, then we have to maintain the client updated about the virtual state of the host language source
+        if document.parse! && @global_state.client_capabilities.supports_request_delegation &&
+            document.is_a?(ERBDocument)
+
+          send_message(
+            Notification.new(
+              method: "delegate/textDocument/virtualState",
+              params: {
+                textDocument: {
+                  uri: uri.to_s,
+                  text: document.host_language_source,
+                },
+              },
+            ),
+          )
+        end
+      end
+    rescue Store::NonExistingDocumentError
+      # If we receive a request for a file that no longer exists, we don't want to fail
+    end
+
     #: ((Result | Error | Notification | Request) message) -> void
     def send_message(message)
       # When we're shutting down the server, there's a small race condition between closing the thread queues and

test/server_test.rb

@@ -278,6 +278,86 @@ def test_returns_nil_diagnostics_and_formatting_for_files_outside_workspace
     assert_nil(result.response)
   end
 
+  def test_diagnostics_parse_after_a_queued_document_change
+    release_worker = Queue.new
+    pipe_reader = nil #: IO?
+    pipe_writer = nil #: IO?
+    reader_thread = nil #: Thread?
+
+    source = <<~RUBY.dup
+      def foo
+        x = 1
+        x
+      end
+    RUBY
+    uri = URI::Generic.from_path(path: File.join(Dir.pwd, "diagnostics.rb"))
+    @server.global_state.apply_options({
+      workspaceFolders: [{ uri: URI::Generic.from_path(path: Dir.pwd).to_s }],
+      initializationOptions: { linters: [] },
+      capabilities: { general: { positionEncodings: ["utf-8"] } },
+    })
+    @server.process_message({
+      method: "textDocument/didOpen",
+      params: {
+        textDocument: { uri: uri, text: source, version: 1, languageId: "ruby" },
+      },
+    })
+
+    worker_started = Queue.new
+    original_process_message = @server.method(:process_message)
+    @server.define_singleton_method(:process_message) do |message|
+      if message[:method] == "textDocument/didChange"
+        worker_started << true
+        release_worker.pop
+      end
+      original_process_message.call(message)
+    end
+
+    messages = [
+      {
+        method: "textDocument/didChange",
+        params: {
+          textDocument: { uri: uri.to_s, version: 2 },
+          contentChanges: [
+            {
+              range: { start: { line: 1, character: 0 }, end: { line: 1, character: 0 } },
+              text: "  y = \"abc\"\n",
+            },
+          ],
+        },
+      },
+      {
+        id: 2,
+        method: "textDocument/diagnostic",
+        params: { textDocument: { uri: uri.to_s } },
+      },
+    ].map do |message|
+      json = message.to_json
+      "Content-Length: #{json.bytesize}\r\n\r\n#{json}"
+    end.join
+    pipe_reader, pipe_writer = IO.pipe
+    pipe_writer.write(messages)
+    pipe_writer.close
+    @server.instance_variable_set(:@reader, RubyLsp::MessageReader.new(pipe_reader))
+
+    reader_thread = Thread.new { @server.start }
+    assert(worker_started.pop(timeout: 20), "worker did not receive the document change")
+    reader_thread.value
+    release_worker << true
+
+    result = @server.pop_response
+    result = @server.pop_response until result.is_a?(RubyLsp::Result) && result.id == 2
+    assert_includes(
+      result.response.items.map(&:message),
+      "assigned but unused variable - y",
+    )
+  ensure
+    release_worker&.close
+    reader_thread&.join
+    pipe_reader&.close unless pipe_reader&.closed?
+    pipe_writer&.close unless pipe_writer&.closed?
+  end
+
   def test_did_close_clears_diagnostics
     @server.process_message({
       method: "textDocument/didClose",