Prevent race condition when saving file during initial indexing (#3269)

Author: vinistock

lib/ruby_indexer/lib/ruby_indexer/index.rb

@@ -13,6 +13,9 @@ class IndexNotEmptyError < StandardError; end
     #: Configuration
     attr_reader :configuration
 
+    #: bool
+    attr_reader :initial_indexing_completed
+
     class << self
       # Returns the real nesting of a constant name taking into account top level
       # references that may be included anywhere in the name or nesting where that
@@ -366,8 +369,6 @@ def index_all(uris: @configuration.indexable_uris, &block)
           "The index is not empty. To prevent invalid entries, `index_all` can only be called once."
       end
 
-      @initial_indexing_completed = true
-
       RBSIndexer.new(self).index_ruby_core
       # Calculate how many paths are worth 1% of progress
       progress_step = (uris.length / 100.0).ceil
@@ -380,6 +381,8 @@ def index_all(uris: @configuration.indexable_uris, &block)
 
         index_file(uri, collect_comments: false)
       end
+
+      @initial_indexing_completed = true
     end
 
     #: (URI::Generic uri, String source, ?collect_comments: bool) -> void

lib/ruby_lsp/server.rb

@@ -1012,6 +1012,21 @@ def text_document_definition(message)
 
     #: (Hash[Symbol, untyped] message) -> void
     def workspace_did_change_watched_files(message)
+      # If indexing is not complete yet, delay processing did change watched file notifications. We need initial
+      # indexing to be in place so that we can handle file changes appropriately without risking duplicates. We also
+      # have to sleep before re-inserting the notification in the queue otherwise the worker can get stuck in its own
+      # loop of pushing and popping the same notification
+      unless @global_state.index.initial_indexing_completed
+        Thread.new do
+          sleep(2)
+          # We have to ensure that the queue is not closed yet, since nothing stops the user from saving a file and then
+          # immediately telling the LSP to shutdown
+          @incoming_queue << message unless @incoming_queue.closed?
+        end
+
+        return
+      end
+
       changes = message.dig(:params, :changes)
       # We allow add-ons to register for watching files and we have no restrictions for what they register for. If the
       # same pattern is registered more than once, the LSP will receive duplicate change notifications. Receiving them
@@ -1049,27 +1064,29 @@ def workspace_did_change_watched_files(message)
 
     #: (RubyIndexer::Index index, String file_path, Integer change_type) -> void
     def handle_ruby_file_change(index, file_path, change_type)
-      load_path_entry = $LOAD_PATH.find { |load_path| file_path.start_with?(load_path) }
-      uri = URI::Generic.from_path(load_path_entry: load_path_entry, path: file_path)
-
-      case change_type
-      when Constant::FileChangeType::CREATED
-        content = File.read(file_path)
-        # If we receive a late created notification for a file that has already been claimed by the client, we want to
-        # handle change for that URI so that the require path tree is updated
-        @store.key?(uri) ? index.handle_change(uri, content) : index.index_single(uri, content)
-      when Constant::FileChangeType::CHANGED
-        content = File.read(file_path)
-        # We only handle changes on file watched notifications if the client is not the one managing this URI.
-        # Otherwise, these changes are handled when running the combined requests
-        index.handle_change(uri, content) unless @store.key?(uri)
-      when Constant::FileChangeType::DELETED
-        index.delete(uri)
-      end
-    rescue Errno::ENOENT
-      # If a file is created and then delete immediately afterwards, we will process the created notification before we
-      # receive the deleted one, but the file no longer exists. This may happen when running a test suite that creates
-      # and deletes files automatically.
+      @global_state.synchronize do
+        load_path_entry = $LOAD_PATH.find { |load_path| file_path.start_with?(load_path) }
+        uri = URI::Generic.from_path(load_path_entry: load_path_entry, path: file_path)
+
+        case change_type
+        when Constant::FileChangeType::CREATED
+          content = File.read(file_path)
+          # If we receive a late created notification for a file that has already been claimed by the client, we want to
+          # handle change for that URI so that the require path tree is updated
+          @store.key?(uri) ? index.handle_change(uri, content) : index.index_single(uri, content)
+        when Constant::FileChangeType::CHANGED
+          content = File.read(file_path)
+          # We only handle changes on file watched notifications if the client is not the one managing this URI.
+          # Otherwise, these changes are handled when running the combined requests
+          index.handle_change(uri, content) unless @store.key?(uri)
+        when Constant::FileChangeType::DELETED
+          index.delete(uri)
+        end
+      rescue Errno::ENOENT
+        # If a file is created and then delete immediately afterwards, we will process the created notification before
+        # we receive the deleted one, but the file no longer exists. This may happen when running a test suite that
+        # creates and deletes files automatically.
+      end
     end
 
     #: (URI::Generic uri) -> void

test/server_test.rb

@@ -475,6 +475,7 @@ def test_changed_file_only_indexes_ruby
     path = File.join(Dir.pwd, "lib", "foo.rb")
     File.write(path, "class Foo\nend")
 
+    @server.global_state.index.index_all(uris: [])
     @server.global_state.index.expects(:index_single).once.with do |uri|
       uri.full_path == path
     end
@@ -501,6 +502,7 @@ def test_changed_file_only_indexes_ruby
   end
 
   def test_did_change_watched_files_does_not_fail_for_non_existing_files
+    @server.global_state.index.index_all(uris: [])
     @server.process_message({
       method: "workspace/didChangeWatchedFiles",
       params: {
@@ -530,6 +532,7 @@ def test_did_change_watched_files_handles_deletions
 
     uri = URI::Generic.from_path(path: path)
 
+    @server.global_state.index.index_all(uris: [])
     @server.process_message({
       method: "workspace/didChangeWatchedFiles",
       params: {
@@ -585,6 +588,7 @@ def version
     bar.expects(:workspace_did_change_watched_files).once
 
     begin
+      @server.global_state.index.index_all(uris: [])
       @server.process_message({
         method: "workspace/didChangeWatchedFiles",
         params: {
@@ -606,6 +610,7 @@ def version
   end
 
   def test_did_change_watched_files_processes_unique_change_entries
+    @server.global_state.index.index_all(uris: [])
     @server.expects(:handle_rubocop_config_change).once
     @server.process_message({
       method: "workspace/didChangeWatchedFiles",
@@ -1155,6 +1160,7 @@ def test_rubocop_config_changes_trigger_workspace_diagnostic_refresh
       },
     })
 
+    @server.global_state.index.index_all(uris: [])
     @server.process_message({
       method: "workspace/didChangeWatchedFiles",
       params: {
@@ -1316,6 +1322,7 @@ class Foo
         },
       })
 
+      @server.global_state.index.index_all(uris: [])
       @server.global_state.index.expects(:handle_change).never
       @server.process_message({
         method: "workspace/didChangeWatchedFiles",
@@ -1370,6 +1377,7 @@ class Foo
         params: { textDocument: { uri: uri } },
       })
 
+      @server.global_state.index.index_all(uris: [])
       # Then send a late did change watched files notification for the creation of the file
       @server.process_message({
         method: "workspace/didChangeWatchedFiles",
@@ -1414,8 +1422,59 @@ def test_diagnose_state
     assert_equal(0, result.response[:incomingQueueSize])
   end
 
+  def test_modifying_files_during_initial_indexing_does_not_duplicate_entries
+    path = File.join(Dir.pwd, "lib", "foo.rb")
+    uri = URI::Generic.from_path(path: path)
+
+    begin
+      @server.process_message({
+        id: 1,
+        method: "initialize",
+        params: {
+          initializationOptions: {},
+          capabilities: { general: { positionEncodings: ["utf-8"] }, window: { workDoneProgress: true } },
+        },
+      })
+
+      # Start indexing
+      File.write(path, "class Foo\nend")
+      @server.process_message({ method: "initialized", params: {} })
+
+      # Then immediately notify that a file was modified before indexing is finished
+      File.write(path, "class Foo\n  def bar\n  end\nend")
+      @server.process_message({
+        method: "workspace/didChangeWatchedFiles",
+        params: {
+          changes: [
+            {
+              uri: uri.to_s,
+              type: RubyLsp::Constant::FileChangeType::CHANGED,
+            },
+          ],
+        },
+      })
+
+      wait_for_indexing
+
+      # There should not be a duplicate declaration
+      index = @server.global_state.index
+      assert_equal(1, index["Foo"]&.length)
+    ensure
+      FileUtils.rm(path)
+    end
+  end
+
   private
 
+  def wait_for_indexing
+    message = @server.pop_response
+    until message.is_a?(RubyLsp::Notification) && message.method == "$/progress" &&
+        T.unsafe(message).params.value.kind == "end"
+
+      message = @server.pop_response
+    end
+  end
+
   def with_uninstalled_rubocop(&block)
     rubocop_paths = $LOAD_PATH.select { |path| path.include?("gems/rubocop") }
     rubocop_paths.each { |path| $LOAD_PATH.delete(path) }