Update the graph when the workspace changes

lib/rubydex/server.rb

@@ -23,13 +23,25 @@ def supported?
           !State::NOFOLLOW.nil?
       end
 
-      #: (workspace_path: String, ?progress_io: IO?) -> Rubydex::Graph
+      # Builds a fully indexed + resolved graph for the workspace, and returns it with the errors the
+      # indexer reported. `progress_io`, when given, receives human-readable progress messages.
+      #
+      # A caller that discards the errors records a file the indexer never read as successfully
+      # indexed, which is why they come back rather than vanishing here.
+      #: (workspace_path: String, ?progress_io: IO?) -> [Rubydex::Graph, Array[String]]
       def build_graph(workspace_path:, progress_io: nil)
         # The server boot must build the same graph as the inline CLI path.
         graph = Rubydex::Graph.configure_for_workspace(workspace_path)
-        Progress.with_timer(progress_io, "Indexing workspace...") { graph.index_workspace }
+
+        # `workspace_paths` lists every root to index, and it names gem directories that this install
+        # may not have. Each absent root costs one error, and those phantom errors would drown the
+        # ones that concern real files, so they never reach the indexer.
+        roots = graph.workspace_paths.select { |path| File.exist?(path) }
+
+        errors = [] #: Array[String]
+        Progress.with_timer(progress_io, "Indexing workspace...") { errors = graph.index_all(roots) }
         Progress.with_timer(progress_io, "Resolving graph...") { graph.resolve }
-        graph
+        [graph, errors]
       end
     end
   end

lib/rubydex/server/core.rb

@@ -5,15 +5,32 @@
 module Rubydex
   module Server
     # The resident server process. It answers one client at a time, in process, over a UNIX socket.
-    #
-    # The graph is a snapshot from boot. A file edited after that answers with its boot content.
     class Core
+      # Errors that describe one path, and not the health of this process. The walk skips the entry
+      # they name and carries on.
+      #
+      # A resource failure such as `EMFILE`, `ENFILE` or `EIO` is deliberately absent. Swallowing one
+      # of those would make the walk skip every entry, and a refresh would then read the empty result
+      # as "every file was deleted" and erase the graph. Those errors travel on instead, and the
+      # request fails without the manifest moving.
+      PATH_ERROR_NAMES = [:ENOENT, :EACCES, :ELOOP, :ENAMETOOLONG, :ENOTDIR].freeze #: Array[Symbol]
+
+      # Resolved by lookup rather than named directly, because the `Errno` constants a build defines
+      # are platform-dependent and this file still loads on Windows, where server mode cannot run. A
+      # missing constant would otherwise break the load itself. A test pins that every name resolves
+      # here, so a typo cannot hide behind the lookup.
+      PATH_ERRORS = PATH_ERROR_NAMES.filter_map do |name|
+        Errno.const_get(name, false) if Errno.const_defined?(name, false)
+      end.freeze #: Array[Class]
+
       #: (State state, ?lock: File?) -> void
       def initialize(state, lock: nil)
         @state = state
         @lock = lock
+        @mutex = Mutex.new
         @running = true
         @started_at = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+        @manifest = {} #: Hash[String, Float]
       end
 
       # Blocks for the lifetime of the server.
@@ -25,7 +42,8 @@ def run
         @state.record!
 
         require "rubydex"
-        @graph = Server.build_graph(workspace_path: @state.workspace_path)
+        @graph, boot_errors = Server.build_graph(workspace_path: @state.workspace_path)
+        @manifest = initial_manifest(boot_errors)
 
         server = open_socket
         log("rdx server ready (pid=#{Process.pid}, workspace=#{@state.workspace_path})")
@@ -142,14 +160,200 @@ def handle_query(request)
           return response(stderr: "rdx server: the request carried no query format string\n", status: 1)
         end
 
-        # Only the parse and the render answer for user input.
-        begin
-          response(stdout: Rubydex::Query.parse(query).render(@graph, format))
-        rescue ArgumentError => e
-          response(stderr: "#{e.message}\n", status: 1)
+        @mutex.synchronize do
+          refresh_if_stale
+
+          # Only the parse and the render answer for user input. An `ArgumentError` from the refresh
+          # above is a server fault, and reporting it as a bad query would blame the caller.
+          begin
+            response(stdout: Rubydex::Query.parse(query).render(@graph, format))
+          rescue ArgumentError => e
+            response(stderr: "#{e.message}\n", status: 1)
+          end
         end
       end
 
+      # Detects workspace files that changed since the graph was built and applies incremental
+      # updates before answering. Always correct, occasionally slow (Phase 1 freshness model).
+      #: -> void
+      def refresh_if_stale
+        current, unreadable = workspace_manifest
+        previous = @manifest
+
+        # A path the walk could not read this time contributes no entries, and whatever lives under
+        # it is still there. Without this they would all look deleted, and one `chmod`, or one moment
+        # during a checkout, would erase a whole subtree from the graph.
+        hidden = unreadable.empty? ? [] : previous.keys.select { |path| under_any?(path, unreadable) }
+
+        changed = current.select { |path, mtime| previous[path] != mtime }.keys
+        deleted = previous.keys - current.keys - hidden
+        return if changed.empty? && deleted.empty?
+
+        failed = changed - index(changed)
+        deleted.each { |path| @graph.delete_document(uri_for(path)) }
+        @graph.resolve
+
+        # Only a file that indexed cleanly becomes fresh. One that failed keeps its previous mtime,
+        # so the next walk sees it as changed and tries it again. Recording the new mtime would call
+        # a file the server never read "fresh" for the rest of its life. A failed file that is new
+        # has no previous mtime, and stays out of the manifest for the same reason.
+        @manifest = current.reject { |path, _| failed.include?(path) }
+        failed.each { |path| @manifest[path] = previous[path] if previous.key?(path) }
+        hidden.each { |path| @manifest[path] = previous[path] }
+      end
+
+      # Whether `path` is one of `prefixes` or sits beneath one. Equality matters because a single
+      # file can be the thing that could not be read, and not only a directory above it.
+      #: (String path, Array[String] prefixes) -> bool
+      def under_any?(path, prefixes)
+        prefixes.any? { |prefix| path == prefix || path.start_with?("#{prefix}/") }
+      end
+
+      # Indexes `paths` and returns the ones that indexed cleanly.
+      #
+      # `index_all` reports opaque messages for a whole batch, so a failure cannot be attributed to a
+      # file. The graph cannot answer it either: a failed update leaves the previous document in
+      # place, and mapping a path to the URI a document is stored under is exactly the parity
+      # question that Group D still owns.
+      #
+      # So a failing batch is halved until each failure sits alone. A batch indexes far faster per
+      # file than single calls do, which makes this much cheaper than asking file by file. Measured
+      # on 1081 files with one unreadable among them: 21 calls in 75ms, against 1081 calls in 311ms.
+      #: (Array[String] paths) -> Array[String]
+      def index(paths)
+        return paths if paths.empty?
+
+        errors = @graph.index_all(paths)
+        return paths if errors.empty?
+
+        errors.each { |message| log("rdx server: index error: #{message}") }
+        isolate(paths)
+      end
+
+      # Halves a batch that failed until every failure is isolated, and returns what indexed cleanly.
+      #: (Array[String] paths) -> Array[String]
+      def isolate(paths)
+        return [] if paths.size <= 1
+
+        middle = paths.size / 2
+        [paths[0...middle], paths[middle..] || []].flat_map do |half|
+          @graph.index_all(half).empty? ? half : isolate(half)
+        end
+      end
+
+      # The manifest a fresh server starts from.
+      #
+      # With no errors, every file the walk found is fresh. With errors, the indexer cannot say which
+      # file failed, so the walk's files go through `index` and only those that come back clean are
+      # recorded. The rest stay out, and the first request retries them. Recording them from the walk
+      # alone would call a file the server never read "fresh" for the rest of its life.
+      #: (Array[String] errors) -> Hash[String, Float]
+      def initial_manifest(errors)
+        files, = workspace_manifest
+        return files if errors.empty?
+
+        # Logged before anything else. An unreadable workspace root leaves no files to attribute, and
+        # a server that started on an empty graph must still say why.
+        errors.each { |message| log("rdx server: boot index error: #{message}") }
+        return files if files.empty?
+
+        indexed = index(files.keys).to_h { |path| [path, true] }
+
+        # `build_graph` resolved before this ran, and `index` has replaced documents since, so the
+        # graph is resolved again before it serves anything.
+        @graph.resolve
+        files.select { |path, _| indexed.key?(path) }
+      end
+
+      # The indexable files under the workspace, and the directories the walk could not read.
+      #: -> [Hash[String, Float], Array[String]]
+      def workspace_manifest
+        manifest = {} #: Hash[String, Float]
+        unreadable = [] #: Array[String]
+        # Mirror the indexer's discovery: recurse everything except paths matching the configured
+        # exclude globs. The Rust indexer has no ignore-by-name list; it applies these glob patterns
+        # to every entry (pruning directories and skipping files alike).
+        patterns = @graph.excluded_patterns
+        collect_files(@state.workspace_path, manifest, patterns, unreadable, top_level: true)
+        [manifest, unreadable]
+      end
+
+      # The rescues sit at three levels on purpose:
+      #
+      # - The inner one isolates a single entry, so a file that vanished mid-walk cannot hide every
+      #   entry after it in the same directory.
+      # - `EACCES` on the directory itself records it as unreadable. Its files are still there, and
+      #   the caller keeps them rather than treating the subtree as deleted.
+      # - `ENOENT` on the directory means it really is gone, and so are its files.
+      #
+      # `each_child` streams. `Dir.children` would materialise every name in the directory, which
+      # this project cannot afford on a hyper-scale workspace.
+      #: (String dir, Hash[String, Float] manifest, Array[String] patterns, Array[String] unreadable, ?top_level: bool) -> void
+      def collect_files(dir, manifest, patterns, unreadable, top_level: false)
+        Dir.each_child(dir) do |entry|
+          full = File.join(dir, entry)
+          next if excluded_by_patterns?(full, patterns)
+
+          begin
+            if directory_to_walk?(full, top_level)
+              collect_files(full, manifest, patterns, unreadable)
+            elsif Rubydex::Graph::INDEXABLE_EXTENSIONS.include?(File.extname(entry))
+              manifest[full] = File.mtime(full).to_f
+            end
+          rescue *PATH_ERRORS => error
+            # One entry the platform will not answer for: it vanished mid-walk, it cannot be read, or
+            # it is a symlink loop, where `lstat` succeeds but `mtime` follows the link and raises
+            # `ELOOP`. Skipping it must not hide the rest of the directory.
+            #
+            # A permission error is different from the others, because the entry is still there and
+            # may be a whole subtree. A directory that is readable but not searchable, mode `0400`,
+            # lands here for every child: `each_child` lists the names and each `lstat` is refused.
+            # The path is recorded so the caller keeps what it already knew, rather than reading the
+            # gap as a deletion and erasing the subtree.
+            unreadable << full if error.is_a?(Errno::EACCES)
+            next
+          end
+        end
+      rescue Errno::ENOENT
+        # The directory is gone, and so are its files.
+        nil
+      rescue *PATH_ERRORS
+        # Present but unreadable right now. Its files are not gone, so the caller keeps their entries
+        # instead of erasing the subtree.
+        unreadable << dir
+      end
+
+      # Whether the walk descends into `path`. The answer differs by depth, because the two sides of
+      # discovery do:
+      #
+      # - `Graph#workspace_paths` asks `File.directory?` about the workspace's own children, so a
+      #   symlinked directory there becomes an explicit root, and the Rust walker does traverse an
+      #   explicit root (`collect_files_indexes_symlinked_directory_roots`).
+      # - Below that the walker asks a `DirEntry` for its type, which never follows a symlink
+      #   (`collect_files_does_not_follow_symlinked_directories`).
+      #
+      # Following at every depth is what let `ln -s .. sub/loop` record one file 32 times under ever
+      # longer paths, until the platform refused. Following at neither depth would hide a whole
+      # top-level symlinked directory that the indexer does read.
+      #: (String path, bool top_level) -> bool
+      def directory_to_walk?(path, top_level)
+        top_level ? File.directory?(path) : File.lstat(path).directory?
+      end
+
+      # Mirrors the Rust indexer's `is_excluded`: an entry is skipped when any exclude glob matches
+      # its path. `FNM_PATHNAME` keeps `*` from crossing `/` and enables `**` recursion, matching the
+      # `glob` crate's `Pattern::matches_path` semantics.
+      #: (String path, Array[String] patterns) -> bool
+      def excluded_by_patterns?(path, patterns)
+        patterns.any? { |pattern| File.fnmatch?(pattern, path, File::FNM_PATHNAME) }
+      end
+
+      #: (String path) -> String
+      def uri_for(path)
+        path = "/#{path}" if Gem.win_platform?
+        URI::File.build(path: path).to_s
+      end
+
       #: (?stdout: String, ?stderr: String, ?status: Integer) -> Hash[String, untyped]
       def response(stdout: "", stderr: "", status: 0)
         { "stdout" => stdout, "stderr" => stderr, "status" => status }