diff --git a/lib/rubydex/cli.rb b/lib/rubydex/cli.rb index 541150d62..a6ec5e595 100644 --- a/lib/rubydex/cli.rb +++ b/lib/rubydex/cli.rb @@ -26,6 +26,14 @@ def start(argv = ARGV) require "rubydex" dispatch(argv.shift, argv) + rescue StandardError => e + # A command loads `rubydex/server` only when it needs it, so the constant can be absent. + # A direct reference here would raise `NameError` and hide the real error. + raise unless defined?(Rubydex::Server::Error) && e.is_a?(Rubydex::Server::Error) + + # A server that does not start or answer is a runtime condition, not a defect in rdx. + warn("rdx server: #{e.message}") + exit(1) end # Reports `message`, then the usage text, and exits non-zero. Public because the subcommands diff --git a/lib/rubydex/cli/command.rb b/lib/rubydex/cli/command.rb index 1e23e1560..0a1f1be2b 100644 --- a/lib/rubydex/cli/command.rb +++ b/lib/rubydex/cli/command.rb @@ -2,6 +2,8 @@ require "optparse" +require "rubydex/progress" + module Rubydex module CLI # Base class for `rdx` subcommands. A subcommand parses its own options out of `argv` and, when @@ -109,16 +111,17 @@ def abort_with_usage(message) CLI.abort_with_usage(message) end - # Parses this command's options out of `argv`, with a banner derived from the command's own - # declaration. `-h`/`--help` prints the parser and exits, so every subcommand documents itself - # the same way. Pass `options: true` when the command accepts options beyond `--help`. + # A command with subactions passes its own `banner`. Every other command builds one from its + # declaration. # # A bad option reports the message and the usage text, so every subcommand rejects bad input # the same way. - #: (?options: bool) ?{ (OptionParser parser) -> void } -> void - def parse_options!(options: false) - banner = +"Usage: rdx #{self.class.usage_form}" - banner << " [options]" if options + #: (?options: bool, ?banner: String?) ?{ (OptionParser parser) -> void } -> void + def parse_options!(options: false, banner: nil) + unless banner + banner = +"Usage: rdx #{self.class.usage_form}" + banner << " [options]" if options + end parser = OptionParser.new do |p| p.banner = banner @@ -138,19 +141,10 @@ def parse_options!(options: false) #: (IO progress_io) -> Rubydex::Graph def build_graph(progress_io) graph = Rubydex::Graph.configure_for_workspace(Dir.pwd) - with_timer(progress_io, "Indexing workspace...") { graph.index_workspace } - with_timer(progress_io, "Resolving graph...") { graph.resolve } + Progress.with_timer(progress_io, "Indexing workspace...") { graph.index_workspace } + Progress.with_timer(progress_io, "Resolving graph...") { graph.resolve } graph end - - #: (IO io, String message) { -> void } -> void - def with_timer(io, message) - io.print(message) - start = Process.clock_gettime(Process::CLOCK_MONOTONIC, :float_millisecond) - yield - duration = Process.clock_gettime(Process::CLOCK_MONOTONIC, :float_millisecond) - start - io.puts(" finished in #{duration.round(2)}ms") - end end end end diff --git a/lib/rubydex/cli/command/query.rb b/lib/rubydex/cli/command/query.rb index 40a4b1e37..cf5a412be 100644 --- a/lib/rubydex/cli/command/query.rb +++ b/lib/rubydex/cli/command/query.rb @@ -5,7 +5,7 @@ module Rubydex module CLI # `rdx query ` — runs a Cypher query against the workspace graph and prints the result. - # `--schema` describes the queryable schema instead, which needs no graph. + # `--schema` needs no graph. `--server` sends the query to the resident server. class Command class Query < Command command "query" @@ -20,12 +20,16 @@ class Query < Command def run schema = false format = "table" + use_server = false parse_options!(options: true) do |parser| parser.on("--schema", "Describe the queryable schema instead of running a query") { schema = true } parser.on("--format FORMAT", ["table", "json"], "Output format (table or json)") do |value| format = value end + parser.on("--server", "Run the query through the resident server for this workspace") do + use_server = true + end end query = argv.shift @@ -39,6 +43,34 @@ def run abort_with_usage("`query` requires a Cypher query argument (or pass `--schema`)") if query.nil? || query.empty? + if use_server && server_available? + query_through_server(query, format) + else + run_inline(query, format) + end + end + + private + + # The require is cheap, because the client side loads no native extension. An unsupported + # platform falls back to the inline path. + #: -> bool + def server_available? + require "rubydex/server" + + Rubydex::Server.supported? + end + + # The server parses and runs the query, so this process forwards a string and loads no + # native extension. + #: (String query, String format) -> bot + def query_through_server(query, format) + state = Rubydex::Server::State.new(workspace_path: Dir.pwd) + exit(Rubydex::Server::Client.query(state, { query: query, query_format: format })) + end + + #: (String query, String format) -> void + def run_inline(query, format) # Parse the query up front so a malformed query fails fast, before the expensive indexing. parsed = parse_query(query) @@ -48,8 +80,6 @@ def run render(parsed, graph, format) end - private - #: (String query) -> Rubydex::Query def parse_query(query) Rubydex::Query.parse(query) diff --git a/lib/rubydex/cli/command/server.rb b/lib/rubydex/cli/command/server.rb new file mode 100644 index 000000000..108f297eb --- /dev/null +++ b/lib/rubydex/cli/command/server.rb @@ -0,0 +1,70 @@ +# frozen_string_literal: true + +require "rubydex/cli/command" + +module Rubydex + module CLI + class Command + class Server < Command + command "server" + arguments "" + summary "Manage the resident server (start, stop, restart, status)" + + ACTIONS = ["start", "stop", "restart", "status"].freeze #: Array[String] + + USAGE = <<~TEXT #: String + Usage: rdx server [options] + + Actions: + start Start the server for this workspace + stop Stop the running server for this workspace + restart Restart the server for this workspace + status Print the status of the server for this workspace + TEXT + + #: -> void + def run + # Options are parsed before the action is shifted, so a flag is not read as the action. + # `OptionParser#parse!` permutes, so the action may come before or after a flag. + parse_options!(options: true, banner: USAGE) + + action = argv.shift + abort_with_actions("unknown server action: #{action.inspect}") unless ACTIONS.include?(action) + + require "rubydex/server" + + unless Rubydex::Server.supported? + abort("rdx server mode is not supported on this platform " \ + "(requires fork, UNIX sockets and O_NOFOLLOW)") + end + + exit(dispatch_action(action)) + end + + private + + #: (String action) -> Integer + def dispatch_action(action) + state = Rubydex::Server::State.new(workspace_path: Dir.pwd) + + case action + when "start" then Rubydex::Server::Commands.start(state) + when "stop" then Rubydex::Server::Commands.stop(state) + when "restart" then Rubydex::Server::Commands.restart(state) + else Rubydex::Server::Commands.status(state) + end + end + + # Uses the action list, not the top-level command list, because the error is about an action + # of this command. + #: (String message) -> void + def abort_with_actions(message) + warn(message) + warn("") + warn(USAGE) + exit(1) + end + end + end + end +end diff --git a/lib/rubydex/progress.rb b/lib/rubydex/progress.rb new file mode 100644 index 000000000..efb114c6a --- /dev/null +++ b/lib/rubydex/progress.rb @@ -0,0 +1,24 @@ +# frozen_string_literal: true + +module Rubydex + # The CLI and the server print the same progress lines, so the measurement is defined here, not + # in either of them. + module Progress + class << self + # The server passes a `nil` `io` when it has no log. + #: (IO? io, String message) { -> void } -> void + def with_timer(io, message) + unless io + yield + return + end + + io.print(message) + start = Process.clock_gettime(Process::CLOCK_MONOTONIC, :float_millisecond) + yield + duration = Process.clock_gettime(Process::CLOCK_MONOTONIC, :float_millisecond) - start + io.puts(" finished in #{duration.round(2)}ms") + end + end + end +end diff --git a/lib/rubydex/server.rb b/lib/rubydex/server.rb new file mode 100644 index 000000000..da69948eb --- /dev/null +++ b/lib/rubydex/server.rb @@ -0,0 +1,42 @@ +# frozen_string_literal: true + +require "rubydex/progress" +require "rubydex/version" + +module Rubydex + # Client/server mode for the `rdx` executable. + # + # A resident server process indexes and resolves the workspace once, and keeps the graph in + # memory. Later commands reach it over a UNIX domain socket and skip that work. + module Server + # Increase this after any incompatible change to the request or response shape. + PROTOCOL = 1 + + class Error < StandardError; end + + class << self + # Without `O_NOFOLLOW`, a symlink can redirect the permissions this code sets on its runtime + # directory. + #: -> bool + def supported? + Process.respond_to?(:fork) && defined?(::UNIXSocket) && !Gem.win_platform? && + !State::NOFOLLOW.nil? + end + + #: (workspace_path: String, ?progress_io: IO?) -> Rubydex::Graph + 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 } + Progress.with_timer(progress_io, "Resolving graph...") { graph.resolve } + graph + end + end + end +end + +require "rubydex/server/state" +require "rubydex/server/frame" +require "rubydex/server/core" +require "rubydex/server/client" +require "rubydex/server/commands" diff --git a/lib/rubydex/server/client.rb b/lib/rubydex/server/client.rb new file mode 100644 index 000000000..2828c2bef --- /dev/null +++ b/lib/rubydex/server/client.rb @@ -0,0 +1,287 @@ +# frozen_string_literal: true + +require "socket" + +module Rubydex + module Server + # The short-lived client side. It requires no native extension when a server answers, which + # is the source of the time saving. + module Client + HANDSHAKE_TIMEOUT = 10.0 + + # A new server indexes the whole workspace first, so this limit is far larger. + BOOT_TIMEOUT = 120.0 + + STOP_TIMEOUT = 5.0 + + class << self + #: (State state, Hash[Symbol, untyped] options, ?stdout: IO, ?stderr: IO) -> Integer + def query(state, options, stdout: $stdout, stderr: $stderr) + request( + state, + { + "command" => "query", + "query" => options[:query], + "query_format" => options[:query_format] || "table", + }, + stdout: stdout, + stderr: stderr, + ) + end + + # The server answers only when the query completes, so this read has no overall deadline. + #: (State state, Hash[String, untyped] payload, ?stdout: IO, ?stderr: IO) -> Integer + def request(state, payload, stdout: $stdout, stderr: $stderr) + socket = connection(state) + + begin + exchange(state, socket, payload, stdout: stdout, stderr: stderr) + rescue Frame::ReadTimeout, Frame::Malformed => e + # The server failed, so the client reports the failure and does not raise. + stderr.puts("rdx server: #{e.message}") + 1 + end + end + + # Talks to a server that already exists, and changes nothing. `status` uses this to diagnose + # a server that does not answer, so one budget covers every step. Returns `nil` on no answer. + #: (State state, Hash[String, untyped] payload, ?stdout: IO, ?stderr: IO, ?timeout: Float) -> Integer? + def probe(state, payload, stdout: $stdout, stderr: $stderr, timeout: Frame::REQUEST_TIMEOUT) + deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout + + socket = connect(state, timeout: remaining(deadline)) + return unless socket + + exchange(state, socket, payload, stdout: stdout, stderr: stderr, total_timeout: remaining(deadline)) + rescue Frame::ReadTimeout, Frame::Malformed, Errno::EPIPE, Errno::ECONNRESET + nil + end + + #: (State state) -> UNIXSocket + def connection(state) + ensure_server(state) + + socket = connect(state) + return socket if socket + + # A restart repairs a stale socket, but not a live owner that does not answer: a new + # server cannot take the lock from it. + unless state.server_running? + restart(state) + socket = connect(state) + return socket if socket + end + + if state.server_running? + raise Error, "the rdx server for #{state.workspace_path} is not answering; run `rdx server status`" + end + + raise Error, "could not connect to the rdx server at #{state.socket_path}" + end + + #: (State state) -> void + def ensure_server(state) + if state.server_running? && File.socket?(state.socket_path) + return if state.version_compatible? + + stop(state) + end + + state.clean! + start(state) + end + + # A held lock means another process has already started a server here, so this waits for it. + # + # Take the lock before `fork`. `server_running?` acquires the lock to test it, so the + # parent's readiness check can otherwise acquire it before the child does. + #: (State state) -> void + def start(state) + previous = state.read + + return wait_until_ready(state, previous: previous) if state.server_running? + + state.ensure_dir! + lock = state.open_lock + + # Another process acquired the lock after the check above. + unless lock.flock(File::LOCK_EX | File::LOCK_NB) + lock.close + return wait_until_ready(state, previous: previous) + end + + begin + pid = fork { run_server(state, lock) } + rescue StandardError + lock.flock(File::LOCK_UN) + lock.close + raise + end + + Process.detach(pid) if pid + # The child inherits the open file description, so the lock outlives this handle. + lock.close + wait_until_ready(state, previous: previous) + end + + # Returns false when the old server keeps its lock, because a new one cannot start then. + #: (State state) -> bool + def restart(state) + return false unless stop(state) + + state.clean! + start(state) + true + end + + # An authenticated request is the only stop. `Process.daemon` forks a second time, so a + # recorded pid can belong to an unrelated process, and no caller may signal it. + #: (State state) -> bool + def stop(state) + return true unless state.server_running? + + socket = connect(state) + + if socket + begin + exchange( + state, + socket, + { "command" => "stop" }, + stdout: $stdout, + stderr: $stderr, + total_timeout: Frame::REQUEST_TIMEOUT, + ) + rescue Errno::EPIPE, Errno::ECONNRESET, Frame::ReadTimeout, Frame::Malformed + # A server that dies during its answer has stopped. The lock proves that, below. + nil + end + end + + stopped = wait_until_stopped(state) + state.clean! if stopped + stopped + end + + private + + # One exchange with a server that already answers. The three callers differ only in how they + # reached it, and in what a failure means, so every failure raises and each caller decides. + #: (State state, UNIXSocket socket, Hash[String, untyped] payload, stdout: IO, stderr: IO, ?total_timeout: Float?) -> Integer + def exchange(state, socket, payload, stdout:, stderr:, total_timeout: nil) + Frame.write(socket, payload.merge(base_payload(state))) + + response = Frame.read_response(socket, total_timeout: total_timeout) + raise Frame::Malformed, "the server closed the connection without answering" unless response + + stdout.print(response["stdout"]) if response["stdout"] + stderr.print(response["stderr"]) if response["stderr"] + + status = response["status"] + raise Frame::Malformed, "the response carried no status" unless status.is_a?(Integer) + + status + ensure + socket.close unless socket.closed? + end + + # The daemon inherits the held lock across both forks. + #: (State state, File lock) -> void + def run_server(state, lock) + Process.daemon(true) + Core.new(state, lock: lock).run + end + + #: (State state, ?timeout: Float) -> UNIXSocket? + def connect(state, timeout: HANDSHAKE_TIMEOUT) + # A relative socket name keeps the path below the `sockaddr_un` limit. + socket = Dir.chdir(File.dirname(state.socket_path)) do + UNIXSocket.new(File.basename(state.socket_path)) + end + version = Frame.read_line(socket, timeout: timeout).chomp + + if version == state.expected_version + socket + else + socket.close + nil + end + rescue Errno::ENOENT, + Errno::ECONNREFUSED, + Errno::ECONNRESET, + Frame::ReadTimeout, + Frame::Malformed + begin + socket&.close + rescue IOError + nil + end + nil + end + + # `previous` is the record from before the spawn. A held lock, or a changed record, proves + # that a boot began. A free lock after that proves the server died, so this stops early. + #: (State state, previous: Hash[String, untyped]?) -> void + def wait_until_ready(state, previous:) + deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + BOOT_TIMEOUT + seen_lock = false + + loop do + running = state.server_running? + seen_lock ||= running + + # The lock is part of the test: a crashed server leaves a socket and a valid record. + break if running && File.socket?(state.socket_path) && state.version_compatible? + + if (seen_lock || state.read != previous) && !running + raise Error, "the rdx server exited during startup; see #{server_log_hint(state)} for details" + end + + raise Error, "timed out waiting for the rdx server to start (see #{server_log_hint(state)})" if Process.clock_gettime(Process::CLOCK_MONOTONIC) > deadline + + sleep(0.05) + end + end + + # The kernel frees the lock when the process exits, however it exited. + #: (State state) -> bool + def wait_until_stopped(state) + deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + STOP_TIMEOUT + + while state.server_running? + return false if Process.clock_gettime(Process::CLOCK_MONOTONIC) > deadline + + sleep(0.02) + end + + true + end + + # Floored, so a read never gets a zero or negative timeout. + #: (Float deadline) -> Float + def remaining(deadline) + left = deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC) + [left, 0.05].max + end + + # Where the detached server writes its output, so a user can read a boot failure. + #: (State state) -> String + def server_log_hint(state) + override = ENV["RDX_SERVER_LOG"] + override && !override.empty? ? override : state.log_path + end + + #: (State state) -> Hash[String, untyped] + def base_payload(state) + { + "protocol" => PROTOCOL, + "token" => state.token, + "cwd" => Dir.pwd, + "argv" => ARGV, + "env" => {}, + } + end + end + end + end +end diff --git a/lib/rubydex/server/commands.rb b/lib/rubydex/server/commands.rb new file mode 100644 index 000000000..c95173aa1 --- /dev/null +++ b/lib/rubydex/server/commands.rb @@ -0,0 +1,97 @@ +# frozen_string_literal: true + +module Rubydex + module Server + module Commands + class << self + #: (State state, ?stdout: IO) -> Integer + def start(state, stdout: $stdout) + if state.server_running? && state.version_compatible? && File.socket?(state.socket_path) + stdout.puts("rdx server already running") + return 0 + end + + Client.ensure_server(state) + stdout.puts("rdx server started") + 0 + end + + #: (State state, ?stdout: IO) -> Integer + def stop(state, stdout: $stdout) + unless state.server_running? + stdout.puts("rdx server: no server running for #{state.workspace_path}") + state.clean! + return 0 + end + + if Client.stop(state) + stdout.puts("rdx server stopped") + return 0 + end + + stdout.puts(unresponsive_report(state, "it did not stop within #{Client::STOP_TIMEOUT}s")) + 1 + end + + #: (State state, ?stdout: IO) -> Integer + def restart(state, stdout: $stdout) + unless Client.restart(state) + stdout.puts(unresponsive_report(state, "it did not stop within #{Client::STOP_TIMEOUT}s")) + return 1 + end + + stdout.puts("rdx server restarted") + 0 + end + + # `timeout` exists so a test can exercise the unanswered path without the real wait. + #: (State state, ?stdout: IO, ?stderr: IO, ?timeout: Float) -> Integer + def status(state, stdout: $stdout, stderr: $stderr, timeout: Frame::REQUEST_TIMEOUT) + unless state.server_running? + stdout.puts("rdx server: not running for #{state.workspace_path}") + return 0 + end + + unless File.socket?(state.socket_path) + stdout.puts(unresponsive_report(state, "it has not created its socket")) + return 1 + end + + # `probe` must not change the server or spend a fresh timeout on the handshake and the + # answer, because it diagnoses a server that will not answer. + status = Client.probe( + state, + { "command" => "status" }, + stdout: stdout, + stderr: stderr, + timeout: timeout, + ) + return status if status + + stdout.puts(unresponsive_report(state, "it did not answer within #{timeout}s")) + 1 + end + + private + + # The report states an observation, not a diagnosis: the server answers one request at a + # time, so a long query looks the same as a stuck process. + #: (State state, String reason) -> String + def unresponsive_report(state, reason) + <<~REPORT + rdx server: a server holds #{state.workspace_path}, but #{reason} + recorded pid: #{state.server_pid || "unknown"} + recorded start: #{state.started_at || "unknown"} + socket: #{state.socket_path} + log: #{state.log_path} + The server answers one request at a time, so a long query looks the same from here. The + log records the boot and any failure, but not the request in flight, so it cannot tell + you which of the two this is. rdx never signals a recorded pid: check the log for errors, + and stop that process yourself only if it stays silent for longer than the work should + take. + REPORT + end + end + end + end +end diff --git a/lib/rubydex/server/core.rb b/lib/rubydex/server/core.rb new file mode 100644 index 000000000..bc7def658 --- /dev/null +++ b/lib/rubydex/server/core.rb @@ -0,0 +1,217 @@ +# frozen_string_literal: true + +require "socket" + +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 + #: (State state, ?lock: File?) -> void + def initialize(state, lock: nil) + @state = state + @lock = lock + @running = true + @started_at = Process.clock_gettime(Process::CLOCK_MONOTONIC) + end + + # Blocks for the lifetime of the server. + #: -> void + def run + redirect_output + # Recorded before the slow index, so a client can separate a boot crash from a slow start. + # The socket is the readiness signal, and it appears only when the graph is ready. + @state.record! + + require "rubydex" + @graph = Server.build_graph(workspace_path: @state.workspace_path) + + server = open_socket + log("rdx server ready (pid=#{Process.pid}, workspace=#{@state.workspace_path})") + serve(server) + rescue StandardError => e + # A detached crash reaches nobody, so the log is the only record of it. + log("rdx server crashed: #{e.class}: #{e.message}") + e.backtrace&.each { |frame| log(" #{frame}") } + raise + ensure + cleanup(server) + end + + private + + #: -> UNIXServer + def open_socket + File.unlink(@state.socket_path) if File.exist?(@state.socket_path) + # `sockaddr_un` caps the path near 104 bytes, so the bind uses a short relative name. + server = Dir.chdir(File.dirname(@state.socket_path)) do + UNIXServer.new(File.basename(@state.socket_path)) + end + File.chmod(0o600, @state.socket_path) + server + end + + #: (UNIXServer server) -> void + def serve(server) + while @running + client = begin + server.accept + rescue IOError, Errno::EBADF + break + end + + handle(client) + end + end + + #: (UNIXSocket client) -> void + def handle(client) + # The first write lets the client detect an old server before it sends a request. + client.puts(@state.expected_version) + + request = Frame.read_request(client) + return unless request + + if authorized?(request["token"]) + dispatch(request, client) + else + answer(client, response(stderr: "rdx server: unauthorized request\n", status: 1)) + end + rescue Errno::EPIPE, Errno::ECONNRESET + nil + rescue Frame::Malformed, Frame::ReadTimeout => e + # A peer that breaks the protocol must not stop the server. + log("rdx server: rejected a request: #{e.class}: #{e.message}") + answer(client, response(stderr: "rdx server: #{e.message}\n", status: 1)) + rescue StandardError => e + # One failed connection must not end the accept loop. + log("rdx server: internal error: #{e.class}: #{e.message}") + e.backtrace&.first(20)&.each { |frame| log(" #{frame}") } + answer(client, response(stderr: "rdx server: internal error: #{e.class}: #{e.message}\n", status: 1)) + ensure + begin + client.close + rescue IOError + nil + end + end + + # A client that already closed is not an error here. + #: (UNIXSocket client, Hash[String, untyped] payload) -> void + def answer(client, payload) + Frame.write(client, payload) + rescue Errno::EPIPE, Errno::ECONNRESET, IOError + nil + end + + #: (Hash[untyped, untyped] request, UNIXSocket client) -> void + def dispatch(request, client) + command = request["command"] + + payload = case command + when "query" + handle_query(request) + when "status" + response(stdout: status_report) + when "stop" + response + else + response(stderr: "rdx server: unknown command #{command.inspect}\n", status: 1) + end + + Frame.write(client, payload) + # Set after the write. A failed write leaves the server running, so the caller sees a + # timeout instead of a server that vanished without an answer. + @running = false if command == "stop" + end + + # JSON can carry any type in these two fields. Unchecked, the extension raises `TypeError`, + # and this server logs the caller's mistake as its own fault. + #: (Hash[untyped, untyped] request) -> Hash[String, untyped] + def handle_query(request) + query = request["query"] + unless query.is_a?(String) + return response(stderr: "rdx server: the request carried no query string\n", status: 1) + end + + # Only an absent key means the default. `|| "table"` would also accept a JSON `false`. + format = request["query_format"] + format = "table" if format.nil? + unless format.is_a?(String) + 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) + end + end + + #: (?stdout: String, ?stderr: String, ?status: Integer) -> Hash[String, untyped] + def response(stdout: "", stderr: "", status: 0) + { "stdout" => stdout, "stderr" => stderr, "status" => status } + end + + #: -> String + def status_report + uptime = (Process.clock_gettime(Process::CLOCK_MONOTONIC) - @started_at).round(1) + <<~STATUS + rdx server running + pid: #{Process.pid} + workspace: #{@state.workspace_path} + socket: #{@state.socket_path} + uptime: #{uptime}s + version: #{@state.expected_version} + STATUS + end + + #: (String? candidate) -> bool + def authorized?(candidate) + return false unless candidate.is_a?(String) + + expected = @state.token + return false unless candidate.bytesize == expected.bytesize + + # Constant time, so the comparison leaks no part of the token through its duration. + candidate.bytes.zip(expected.bytes).reduce(0) { |acc, (a, b)| acc | (a ^ b) }.zero? + end + + #: -> void + def redirect_output + override = ENV["RDX_SERVER_LOG"] + target = override && !override.empty? ? override : @state.log_path + $stdout.reopen(target, "a") + $stderr.reopen(target, "a") + $stdout.sync = true + $stderr.sync = true + end + + #: (String message) -> void + def log(message) + $stdout.puts("[#{Time.now.iso8601}] #{message}") + rescue StandardError + nil + end + + #: (UNIXServer? server) -> void + def cleanup(server) + begin + server&.close + rescue IOError + nil + end + # The server holds the lock, so it removes its own socket without taking the lock again. + @state.remove_socket! + @lock&.flock(File::LOCK_UN) + @lock&.close + rescue StandardError + nil + end + end + end +end + +require "time" diff --git a/lib/rubydex/server/frame.rb b/lib/rubydex/server/frame.rb new file mode 100644 index 000000000..7fcdc0bbb --- /dev/null +++ b/lib/rubydex/server/frame.rb @@ -0,0 +1,196 @@ +# frozen_string_literal: true + +require "json" + +module Rubydex + module Server + # Length-prefixed JSON frames: one decimal line with the payload size, then that many bytes. + # + # The two directions carry opposite risks, so each has its own reader. `read_request` bounds the + # whole frame, because a slow client must not hold the server's accept loop. `read_response` + # bounds only the gap between chunks, because the server sends nothing until the query completes. + module Frame + class ReadTimeout < Error; end + + class Malformed < Error; end + + REQUEST_TIMEOUT = 10.0 #: Float + + IDLE_TIMEOUT = 10.0 #: Float + + # The length line and the version line are both short. + MAX_LINE_BYTES = 1024 #: Integer + + MAX_REQUEST_BYTES = 1024 * 1024 #: Integer + + # A result is much larger than the query that asks for it, so the two caps differ. + MAX_RESPONSE_BYTES = 512 * 1024 * 1024 #: Integer + + # How many bytes one read asks for. `IO#read_nonblock` allocates `maxlen` before it reads, so + # a peer that declares 512 MiB and sends one byte would cost 512 MiB in one allocation. + CHUNK_BYTES = 64 * 1024 #: Integer + + # A length line holds decimal digits and an optional newline. `String#to_i` would accept + # `"100garbage"` and return `100`. + LENGTH_LINE = /\A\d+\n?\z/ #: Regexp + + class << self + #: (IO socket, Hash[untyped, untyped] payload) -> void + def write(socket, payload) + data = JSON.dump(payload) + socket.puts(data.bytesize) + socket.write(data) + socket.flush + end + + # `timeout` bounds the whole frame, not the gap between chunks: the server answers one + # client at a time. Returns `nil` when the client closed before it sent a byte. + #: (IO socket, ?timeout: Float) -> Hash[untyped, untyped]? + def read_request(socket, timeout: REQUEST_TIMEOUT) + read_frame( + socket, + max_bytes: MAX_REQUEST_BYTES, + total_timeout: timeout, + idle_timeout: nil, + first_wait: nil, + ) + end + + # The client waits for the first byte with no limit, because the query runs before it. + # `idle_timeout` then bounds each gap, which frees the client if the server dies mid-answer. + # A control command needs no work, so its caller passes `total_timeout` instead. + # Returns `nil` when the server closed before it sent a byte. + #: (IO socket, ?idle_timeout: Float, ?total_timeout: Float?) -> Hash[untyped, untyped]? + def read_response(socket, idle_timeout: IDLE_TIMEOUT, total_timeout: nil) + read_frame( + socket, + max_bytes: MAX_RESPONSE_BYTES, + total_timeout: total_timeout, + idle_timeout: idle_timeout, + first_wait: nil, + ) + end + + # Reads one line for the version handshake. `timeout` bounds the whole line. + #: (IO socket, ?timeout: Float) -> String + def read_line(socket, timeout: REQUEST_TIMEOUT) + deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout + line = read_line_until(socket, deadline: deadline, idle_timeout: nil, first_wait: nil) + raise ReadTimeout, "the peer sent no line" unless line + + line + end + + private + + # Every failure becomes `Malformed` or `ReadTimeout`. A `JSON::ParserError` never leaves + # this module. + #: (IO socket, max_bytes: Integer, total_timeout: Float?, idle_timeout: Float?, first_wait: Float?) -> Hash[untyped, untyped]? + def read_frame(socket, max_bytes:, total_timeout:, idle_timeout:, first_wait:) + deadline = total_timeout && (Process.clock_gettime(Process::CLOCK_MONOTONIC) + total_timeout) + + line = read_line_until(socket, deadline: deadline, idle_timeout: idle_timeout, first_wait: first_wait) + return unless line + + length = parse_length(line, max_bytes) + body = read_bytes(socket, length, deadline: deadline, idle_timeout: idle_timeout, first_wait: idle_timeout) + raise Malformed, "the peer closed the connection inside a frame" unless body + + payload = parse_body(body) + raise Malformed, "expected a JSON object, got #{payload.class}" unless payload.is_a?(Hash) + + payload + end + + # Returns `nil` at a clean end of file, so a caller can separate a close from a broken frame. + #: (IO socket, deadline: Float?, idle_timeout: Float?, first_wait: Float?) -> String? + def read_line_until(socket, deadline:, idle_timeout:, first_wait:) + buffer = +"" + wait = first_wait + + loop do + byte = read_bytes(socket, 1, deadline: deadline, idle_timeout: idle_timeout, first_wait: wait) + return if byte.nil? && buffer.empty? + raise ReadTimeout, "the peer sent a partial line" unless byte + + buffer << byte + wait = idle_timeout + return buffer if byte == "\n" + raise Malformed, "a line exceeded #{MAX_LINE_BYTES} bytes" if buffer.bytesize >= MAX_LINE_BYTES + end + end + + # Each chunk resets the idle budget, so a slow but steady peer succeeds. + # + # A close after a partial read is a broken frame, never a short payload: otherwise a peer + # could declare 100 bytes, send `{}`, close, and the caller would accept that valid JSON. + #: (IO socket, Integer length, deadline: Float?, idle_timeout: Float?, first_wait: Float?) -> String? + def read_bytes(socket, length, deadline:, idle_timeout:, first_wait:) + buffer = +"" + idle = first_wait + + while buffer.bytesize < length + wait = wait_budget(deadline, idle) + expired(deadline, idle) if IO.select([socket], nil, nil, wait).nil? + + want = length - buffer.bytesize + chunk = socket.read_nonblock([want, CHUNK_BYTES].min, exception: false) + + if chunk.nil? + return if buffer.empty? + + raise Malformed, "the peer closed the connection after #{buffer.bytesize} of #{length} bytes" + end + + next if chunk == :wait_readable + + buffer << chunk + idle = idle_timeout + end + + buffer + end + + # How long one wait lasts. `nil` blocks until the peer sends a byte. + #: (Float? deadline, Float? idle) -> Float? + def wait_budget(deadline, idle) + return idle unless deadline + + remaining = deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC) + expired(deadline, idle) if remaining <= 0 + + idle ? [remaining, idle].min : remaining + end + + # A peer that sends one byte at a time never idles, so it gets the deadline message. + #: (Float? deadline, Float? idle) -> bot + def expired(deadline, idle) + if deadline && (deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC)) <= 0 + raise ReadTimeout, "the peer did not complete the frame within its deadline" + end + + raise ReadTimeout, "the peer sent nothing for #{idle} seconds" + end + + # The message keeps only the start of the parser text, because that text quotes the payload. + #: (String body) -> untyped + def parse_body(body) + JSON.parse(body) + rescue JSON::ParserError => e + raise Malformed, "the payload is not valid JSON: #{e.message[0, 120]}" + end + + #: (String line, Integer max_bytes) -> Integer + def parse_length(line, max_bytes) + raise Malformed, "expected a decimal length, got #{line.inspect}" unless LENGTH_LINE.match?(line) + + length = Integer(line.chomp, 10) + raise Malformed, "expected a positive length, got #{length}" if length <= 0 + raise Malformed, "a payload of #{length} bytes exceeds the limit of #{max_bytes}" if length > max_bytes + + length + end + end + end + end +end diff --git a/lib/rubydex/server/state.rb b/lib/rubydex/server/state.rb new file mode 100644 index 000000000..bdb5f7ff6 --- /dev/null +++ b/lib/rubydex/server/state.rb @@ -0,0 +1,389 @@ +# frozen_string_literal: true + +require "digest" +require "fileutils" +require "json" +require "securerandom" +require "time" +require "tmpdir" + +module Rubydex + module Server + # Owns the per-workspace runtime directory and the three files in it: `lock`, `state.json` and + # `socket`. An "app id" keys the directory, so a new gem, Ruby or protocol forces a new server. + # + # `lock` and `state.json` stay separate files, for two reasons: + # + # - On Windows `flock` becomes `LockFileEx`, which denies every other process read access to the + # locked range. Data inside `lock` would be unreadable exactly while a server runs. + # - A lock belongs to an inode, so nothing deletes `lock`. A second process would otherwise + # create a new file, lock that, and start a second server for this workspace. + # + # A held lock proves that a live process owns this workspace. The recorded pid is a display + # value, and never a signal target. + class State + # A platform without the flag gets `nil`, and every caller refuses. A path-based fallback + # would follow a planted symlink. `Server.supported?` reports the same fact before a boot. + NOFOLLOW = File.const_defined?(:NOFOLLOW) ? File::NOFOLLOW : nil #: Integer? + + #: String + attr_reader :workspace_path + + #: (?workspace_path: String) -> void + def initialize(workspace_path: Dir.pwd) + @workspace_path = File.expand_path(workspace_path) + end + + #: -> String + def app_id + @app_id ||= Digest::SHA256.hexdigest( + [@workspace_path, PROTOCOL, RUBY_VERSION, Rubydex::VERSION, ext_fingerprint].join("\0"), + )[0, 16] + end + + # A mismatch makes the client restart the server. + #: -> String + def expected_version + "#{Rubydex::VERSION}:#{ext_fingerprint}" + end + + #: -> String + def dir + @dir ||= File.join(runtime_root, app_id) + end + + #: -> String + def socket_path + File.join(dir, "socket") + end + + #: -> String + def lock_path + File.join(dir, "lock") + end + + #: -> String + def state_path + File.join(dir, "state.json") + end + + # Where a detached server writes its output, so a boot crash leaves a backtrace. + #: -> String + def log_path + File.join(dir, "server.log") + end + + # The base keeps its own permissions, because the user owns it. This validates it, and refuses + # an unsafe one, instead of a chmod that could break other files there. + #: -> void + def ensure_dir! + validate_base! + ensure_private_dir(runtime_root) + ensure_private_dir(dir) + end + + # The caller takes `LOCK_EX | LOCK_NB` and keeps the handle. Nothing writes through it. + #: -> File + def open_lock + ensure_dir! + refusing_symlink(lock_path) do + File.open(lock_path, File::RDWR | File::CREAT | nofollow!, 0o600) + end + end + + # A fresh token per boot stops a client that cached the token of a previous server. + #: -> void + def record! + ensure_dir! + @token = SecureRandom.hex(32) + write_state( + "pid" => Process.pid, + "token" => @token, + "version" => expected_version, + "started_at" => Time.now.iso8601, + ) + end + + #: -> Hash[String, untyped]? + def read + return unless runtime_trusted? + + payload = JSON.parse(File.read(state_path)) + payload.is_a?(Hash) ? payload : nil + rescue Errno::ENOENT, JSON::ParserError + nil + end + + # The server knows its own token. A client reads it from the record. + #: -> String? + def token + @token || recorded("token") + end + + #: -> Integer? + def server_pid + pid = recorded("pid") + pid.is_a?(Integer) && pid.positive? ? pid : nil + end + + #: -> String? + def started_at + value = recorded("started_at") + value.is_a?(String) ? value : nil + end + + #: -> bool + def version_compatible? + recorded("version") == expected_version + end + + # The lock answers this, and not the recorded pid: the kernel frees a lock when its holder + # dies, so a recycled pid can never look alive. + #: -> bool + def server_running? + return false unless runtime_trusted? + + refusing_symlink(lock_path) do + File.open(lock_path, File::RDWR | nofollow!) do |file| + if file.flock(File::LOCK_EX | File::LOCK_NB) + file.flock(File::LOCK_UN) + false + else + true + end + end + end + rescue Errno::ENOENT + false + end + + # Takes the lock first, so a live or a starting server keeps its socket. + #: -> void + def clean! + return unless runtime_trusted? + + File.open(lock_path, File::RDWR | nofollow!) do |file| + next unless file.flock(File::LOCK_EX | File::LOCK_NB) + + begin + remove_socket! + ensure + file.flock(File::LOCK_UN) + end + end + rescue Errno::ENOENT + nil + end + + # Only the server may call this, because it already holds the lock. + #: -> void + def remove_socket! + File.unlink(socket_path) + rescue Errno::ENOENT + nil + end + + private + + #: (String key) -> untyped + def recorded(key) + record = read + record && record[key] + end + + # A symlink in place of a runtime file means somebody put it there, so the error names that. + #: [T] (String path) { -> T } -> T + def refusing_symlink(path) + yield + rescue Errno::ELOOP + raise Error, "#{path} is a symlink, and the server runtime files must be real files" + end + + # Every open below is a security control, so a platform without the flag stops here. + #: -> Integer + def nofollow! + NOFOLLOW || raise(Error, "this platform cannot open a path without following symlinks") + end + + # `O_NOFOLLOW` protects only the component it opens, and Ruby has no `openat`, so every later + # path-based open depends on a trustworthy base. + # + # Trustworthy means owned by this user or root, and either closed to other writers or sticky. + # A sticky bit stops another user from a rename. `/tmp` passes; `chmod 777 /shared` does not. + # This runs once per instance, because the canonical chain cannot change during one command. + #: -> void + def validate_base! + return if @base_validated + + ensure_safe_base(base_dir) + @base_validated = true + end + + #: (String path) -> void + def ensure_safe_base(path) + current = File.expand_path(path) + + loop do + check_base_component(current) + + parent = File.dirname(current) + break if parent == current + + current = parent + end + end + + # Every component exists, because `realpath` resolved the base before this walk. + #: (String path) -> void + def check_base_component(path) + stat = begin + File.stat(path) + rescue SystemCallError => error + raise Error, "#{path} cannot be checked: #{error.message}" + end + + raise Error, "#{path} is not a directory" unless stat.directory? + + if Process.respond_to?(:uid) && !stat.uid.zero? && stat.uid != Process.uid + raise Error, "#{path} belongs to another user, so it cannot hold the server runtime directory" + end + + if (stat.mode & 0o022) != 0 && (stat.mode & 0o1000) == 0 + raise Error, "#{path} is writable by other users and is not sticky, so the server runtime " \ + "directory would not be safe inside it" + end + end + + # `path` can survive from an earlier run, so this validates it. A path-based `chmod` would + # follow a planted symlink, and a swap could beat it, so the work goes through one descriptor. + #: (String path) -> void + def ensure_private_dir(path) + FileUtils.mkdir_p(path, mode: 0o700) + + with_private_dir(path) do |directory, stat| + directory.chmod(0o700) unless (stat.mode & 0o777) == 0o700 + end + end + + # `server_running?`, `clean!` and `read` run before `ensure_dir!`, and `O_NOFOLLOW` guards only + # their last component. A runtime root owned by another user would choose the socket and the + # token the client trusts, so they check the directories first, and create nothing. + # + # An absent directory answers `false`. One that exists but is not ours raises, because that + # is tampering. + #: -> bool + def runtime_trusted? + validate_base! + trusted_dir?(runtime_root) && trusted_dir?(dir) + end + + #: (String path) -> bool + def trusted_dir?(path) + with_private_dir(path) do |_directory, stat| + raise Error, "#{path} is not private to this user" unless (stat.mode & 0o777) == 0o700 + end + + true + rescue Errno::ENOENT, Errno::ENOTDIR + false + end + + # Yields the stat, so one caller corrects the mode and the other insists on it. + #: [T] (String path) { (File directory, File::Stat stat) -> T } -> T + def with_private_dir(path) + refusing_symlink(path) do + File.open(path, File::RDONLY | nofollow!) do |directory| + stat = directory.stat + raise Error, "#{path} is not a directory" unless stat.directory? + + if Process.respond_to?(:uid) && stat.uid != Process.uid + raise Error, "#{path} belongs to another user" + end + + yield(directory, stat) + end + end + end + + # The temporary file shares the directory, so the rename stays on one filesystem and is atomic. + #: (Hash[String, untyped] payload) -> void + def write_state(payload) + temp = "#{state_path}.#{Process.pid}" + refusing_symlink(temp) do + File.open(temp, File::WRONLY | File::CREAT | File::TRUNC | nofollow!, 0o600) do |file| + file.write(JSON.dump(payload)) + file.flush + file.fsync + end + end + File.rename(temp, state_path) + rescue StandardError + begin + File.unlink(temp) + rescue Errno::ENOENT + nil + end + raise + end + + # One subdirectory per workspace lives here. It nests inside `base_dir`, which this never + # modifies. + #: -> String + def runtime_root + File.join(base_dir, "rubydex-#{uid}") + end + + # The base is canonical, and it must already exist. + # + # Every runtime path derives from this string, so it holds no symlink component. The owner of a + # link entry can swap it even when its target is safe, and a `stat` would approve the target. + # A missing base is refused, because its creation would race the other writers in its parent. + #: -> String + def base_dir + @base_dir ||= begin + configured = configured_base + + begin + File.realpath(configured) + rescue SystemCallError => error + raise Error, "#{configured} cannot hold the server runtime directory: #{error.message}" + end + end + end + + #: -> String + def configured_base + override = ENV["RDX_SERVER_DIR"] + return File.expand_path(override) if override && !override.empty? + + xdg = ENV["XDG_RUNTIME_DIR"] + xdg && !xdg.empty? ? xdg : Dir.tmpdir + end + + #: -> (Integer | String) + def uid + Process.respond_to?(:uid) ? Process.uid : "nobody" + end + + # A recompiled extension invalidates a running server, because Ruby cannot reload it in place. + #: -> String + def ext_fingerprint + @ext_fingerprint ||= begin + lib_dir = File.expand_path("../..", __dir__) + artifacts = Dir.glob(File.join(lib_dir, "**", "rubydex.{bundle,so}")) + + Dir.glob(File.join(lib_dir, "librubydex_sys.*")) + + if artifacts.empty? + "noext" + else + parts = artifacts.sort.map do |path| + stat = File.stat(path) + "#{File.basename(path)}:#{stat.size}:#{stat.mtime.to_i}" + end + Digest::SHA256.hexdigest(parts.join("|"))[0, 16] + end + end + end + end + end +end diff --git a/test/cli_test.rb b/test/cli_test.rb index d121d5575..970d0cfb0 100644 --- a/test/cli_test.rb +++ b/test/cli_test.rb @@ -26,6 +26,7 @@ def test_commands_are_discovered_from_subclasses assert_includes(commands, Rubydex::CLI::Command::Query) assert_includes(commands, Rubydex::CLI::Command::Console) assert_includes(commands, Rubydex::CLI::Command::Mcp) + assert_includes(commands, Rubydex::CLI::Command::Server) # The declared name is what the class reports, and drives its usage line. assert_equal("query", Rubydex::CLI::Command::Query.command_name) @@ -94,6 +95,7 @@ def test_usage_is_generated_from_the_declared_commands Rubydex::CLI::Command::Query, Rubydex::CLI::Command::Console, Rubydex::CLI::Command::Mcp, + Rubydex::CLI::Command::Server, ].each do |command| assert_stdout_includes_pattern(result, /^ #{Regexp.escape(command.usage_form)}\s{2,}\S/) end @@ -199,7 +201,7 @@ def test_query_supports_json_output end def test_command_help_is_available_per_subcommand - ["query", "console", "mcp"].each do |command| + ["query", "console", "mcp", "server"].each do |command| result = rdx(command, "--help") assert_success_status(result) @@ -208,7 +210,7 @@ def test_command_help_is_available_per_subcommand end def test_every_command_reports_an_invalid_option_with_the_usage - ["query", "console", "mcp"].each do |command| + ["query", "console", "mcp", "server"].each do |command| result = rdx(command, "--bogus-flag") refute_success_status(result) diff --git a/test/server/client_test.rb b/test/server/client_test.rb new file mode 100644 index 000000000..5a07c7b22 --- /dev/null +++ b/test/server/client_test.rb @@ -0,0 +1,284 @@ +# frozen_string_literal: true + +require "test_helper" +require "rubydex/server" +require "socket" +require "stringio" +require "timeout" +require "tmpdir" + +module Rubydex + module Server + # A child process takes the place of a real server, so each test controls when the lock, the + # record, and the socket appear. + class ClientTest < Minitest::Test + def setup + skip("server mode is not supported on this platform") unless Server.supported? + + @runtime_dir = Dir.mktmpdir("rdx-client-test") + @previous_server_dir = ENV["RDX_SERVER_DIR"] + ENV["RDX_SERVER_DIR"] = @runtime_dir + @children = [] #: Array[Integer] + end + + def teardown + @children.each do |pid| + Process.kill("KILL", pid) + Process.wait(pid) + rescue Errno::ESRCH, Errno::ECHILD + nil + end + ENV["RDX_SERVER_DIR"] = @previous_server_dir + FileUtils.rm_rf(@runtime_dir) + end + + # A lock held during startup marks a booting server, so a second client must wait, not + # fail. + def test_start_waits_for_a_server_that_is_still_booting + skip("fork is unavailable on this platform") unless Process.respond_to?(:fork) + + state = State.new(workspace_path: "/workspace") + spawn_owner(state, bind_after: 0.4) + + started = monotonic + Client.start(state) + waited = monotonic - started + + assert(File.socket?(state.socket_path), "expected the owner's socket to exist") + assert_operator(waited, :>=, 0.3, "expected the client to wait for the slow boot") + assert_operator(waited, :<, 30.0, "expected the client to return as soon as the boot finished") + end + + def test_start_fails_fast_when_a_booting_server_dies + skip("fork is unavailable on this platform") unless Process.respond_to?(:fork) + + state = State.new(workspace_path: "/workspace") + spawn_owner(state, bind: false, die_after: 0.3) + + started = monotonic + error = assert_raises(Error) { Client.start(state) } + waited = monotonic - started + + assert_match(/exited during startup/, error.message) + assert_operator(waited, :<, 30.0, "expected a fail-fast, not the full boot timeout") + end + + # The readiness rule must reject a dead server's socket and compatible record, or a client + # connects to a dead socket. + def test_a_stale_socket_beside_a_compatible_record_is_not_ready + skip("fork is unavailable on this platform") unless Process.respond_to?(:fork) + + state = State.new(workspace_path: "/workspace") + pid = spawn_owner(state) + wait_for { File.socket?(state.socket_path) } + kill_owner(pid) + wait_for { !state.server_running? } + + assert(File.socket?(state.socket_path)) + assert(state.version_compatible?) + refute(state.server_running?) + + error = assert_raises(Error) do + Client.send(:wait_until_ready, state, previous: nil) + end + + assert_match(/exited during startup/, error.message) + end + + # The server is single-threaded, so one stuck query keeps it inside `handle` and it never + # reaches `accept` again. + def test_status_reports_a_server_that_never_answers + skip("fork is unavailable on this platform") unless Process.respond_to?(:fork) + + state = State.new(workspace_path: "/workspace") + pid = spawn_owner(state, accept: false) + wait_for { File.socket?(state.socket_path) } + + out = StringIO.new + started = monotonic + status = nil #: Integer? + calls = watching_client_mutations do + Timeout.timeout(20) { status = Commands.status(state, stdout: out, timeout: 0.3) } + end + waited = monotonic - started + + assert_equal(1, status) + assert_match(/did not answer within/, out.string) + assert_match(/recorded pid:\s+#{pid}\b/, out.string) + assert_match(/recorded start:\s+\d{4}-\d{2}-\d{2}T/, out.string) + assert_operator(waited, :<, 5.0, "status must not wait beyond its budget") + # A long query looks the same as a stuck one from outside, so the report must say so or + # the reader kills a live process. + assert_match(/a long query looks the same/, out.string) + assert_match(/only if it stays silent/, out.string) + # The log holds the boot and any failure, not the current request, so the report must not + # send the reader there for it. + assert_match(/not the request in flight/, out.string) + # A new server cannot take the lock from a live one, so `status` must not restart. + assert_empty(calls, "status must not start, stop or restart a server") + end + + # The server accepts and sends the version line, then answers nothing; the handshake + # succeeds, so the response budget bounds this wait. + def test_status_reports_a_server_that_greets_and_then_goes_silent + skip("fork is unavailable on this platform") unless Process.respond_to?(:fork) + + state = State.new(workspace_path: "/workspace") + pid = spawn_owner(state, greet: true) + wait_for { File.socket?(state.socket_path) } + + out = StringIO.new + started = monotonic + status = nil #: Integer? + calls = watching_client_mutations do + Timeout.timeout(20) { status = Commands.status(state, stdout: out, timeout: 0.3) } + end + waited = monotonic - started + + assert_equal(1, status) + assert_match(/did not answer within/, out.string) + assert_match(/recorded pid:\s+#{pid}\b/, out.string) + assert_operator(waited, :<, 5.0, "the response budget must end this wait") + assert_empty(calls, "status must not start, stop or restart a server") + end + + # The owner holds the lock but has not bound its socket, so `status` must read the record + # and not block on a socket that does not exist. + def test_status_reports_an_owner_without_a_socket + skip("fork is unavailable on this platform") unless Process.respond_to?(:fork) + + state = State.new(workspace_path: "/workspace") + pid = spawn_owner(state, bind: false) + + out = StringIO.new + calls = watching_client_mutations { @status = Commands.status(state, stdout: out, timeout: 0.3) } + + assert_equal(1, @status) + assert_match(/has not created its socket/, out.string) + assert_match(/recorded pid:\s+#{pid}\b/, out.string) + assert_empty(calls) + end + + def test_status_reports_a_workspace_that_no_server_owns + state = State.new(workspace_path: "/workspace") + + out = StringIO.new + status = Commands.status(state, stdout: out, timeout: 0.3) + + assert_equal(0, status) + assert_match(/not running/, out.string) + end + + # `server_running?` acquires the lock to test it. A child that locks after the fork can find + # the lock held by its own parent, and then it exits without serving. + def test_start_holds_the_lock_before_it_forks + state = State.new(workspace_path: "/workspace") + held = nil #: bool? + + with_stubbed_fork(-> { held = state.server_running? }) do + Client.start(state) + end + + assert(held, "the lock must already be held when the fork happens") + end + + private + + # `fork` is a private `Kernel` method, so `Minitest#stub` cannot reach it. + #: [T] (^() -> void observer) { -> T } -> T + def with_stubbed_fork(observer) + singleton = Client.singleton_class + singleton.send(:alias_method, :original_wait_until_ready, :wait_until_ready) + singleton.send(:define_method, :wait_until_ready) { |*, **| nil } + singleton.send(:define_method, :fork) { |&_block| observer.call && nil } + yield + ensure + singleton.send(:remove_method, :fork) + singleton.send(:remove_method, :wait_until_ready) + singleton.send(:alias_method, :wait_until_ready, :original_wait_until_ready) + singleton.send(:remove_method, :original_wait_until_ready) + end + + # `status` must not start, stop, or restart a server. + #: [T] { -> T } -> Array[Symbol] + def watching_client_mutations + names = [:ensure_server, :start, :restart, :stop] + originals = names.to_h { |name| [name, Client.method(name)] } + calls = [] #: Array[Symbol] + + names.each do |name| + Client.define_singleton_method(name) do |*_args, **_kwargs| + calls << name + nil + end + end + + yield + calls + ensure + originals&.each { |name, method| Client.define_singleton_method(name, method) } + end + + #: -> Float + def monotonic + Process.clock_gettime(Process::CLOCK_MONOTONIC) + end + + # Each keyword simulates a distinct real server state. + #: (State state, ?bind: bool, ?bind_after: Float, ?accept: bool, ?greet: bool, ?die_after: Float?) -> Integer + def spawn_owner(state, bind: true, bind_after: 0.0, accept: true, greet: false, die_after: nil) + reader, writer = IO.pipe + + pid = fork do + reader.close + lock = state.open_lock + lock.flock(File::LOCK_EX) + state.record! + writer.puts("owned") + writer.close + + if die_after + sleep(die_after) + exit!(1) + end + + if bind + sleep(bind_after) if bind_after.positive? + # Bind by basename so a long runtime path does not overflow `sockaddr_un`. + server = Dir.chdir(File.dirname(state.socket_path)) do + UNIXServer.new(File.basename(state.socket_path)) + end + if accept + client = server.accept + client.puts(state.expected_version) if greet + end + end + + sleep(60) + end + + @children << pid + writer.close + reader.gets # Waits until the child holds the lock and writes its record. + reader.close + pid + end + + #: (Integer pid) -> void + def kill_owner(pid) + Process.kill("KILL", pid) + Process.wait(pid) + rescue Errno::ESRCH, Errno::ECHILD + nil + end + + # A test must not race the child it spawned. + #: (?Float timeout) { -> boolish } -> void + def wait_for(timeout = 5.0) + deadline = monotonic + timeout + sleep(0.02) until yield || monotonic > deadline + flunk("the condition never became true") unless yield + end + end + end +end diff --git a/test/server/core_test.rb b/test/server/core_test.rb new file mode 100644 index 000000000..d082d66ed --- /dev/null +++ b/test/server/core_test.rb @@ -0,0 +1,137 @@ +# frozen_string_literal: true + +require "test_helper" +require "rubydex/server" +require "socket" +require "tmpdir" + +module Rubydex + module Server + # A resident server serves every client, so one bad request, unreadable file, or bug must not + # stop it for the rest. + class CoreTest < Minitest::Test + def setup + skip("server mode is not supported on this platform") unless Server.supported? + + @runtime_dir = Dir.mktmpdir("rdx-core-test") + @workspace = Dir.mktmpdir("rdx-core-workspace") + @previous_server_dir = ENV["RDX_SERVER_DIR"] + ENV["RDX_SERVER_DIR"] = @runtime_dir + @state = State.new(workspace_path: @workspace) + @state.record! + end + + def teardown + ENV["RDX_SERVER_DIR"] = @previous_server_dir + FileUtils.rm_rf(@runtime_dir) + FileUtils.rm_rf(@workspace) + end + + # An error in one command must not reach `run` and stop the server for every other client; + # it must cost one connection. + def test_an_unexpected_error_answers_the_client_and_keeps_the_server + core = Core.new(@state) + core.define_singleton_method(:dispatch) { |_request, _client| raise "boom" } + + response = exchange_with(core, { "command" => "query", "query" => "MATCH (n) RETURN n" }) + + assert_equal(1, response["status"]) + assert_match(/internal error/, response["stderr"]) + assert_match(/boom/, response["stderr"]) + + assert_match(/internal error: RuntimeError: boom/, @log) + assert_match(/core_test\.rb/, @log, "the log must keep the backtrace") + end + + def test_an_unauthorized_request_is_refused_and_never_dispatched + core = Core.new(@state) + dispatched = false + core.define_singleton_method(:dispatch) { |_request, _client| dispatched = true } + + response = exchange_with(core, { "command" => "query" }, token: "not-the-token") + + assert_equal(1, response["status"]) + assert_match(/unauthorized/, response["stderr"]) + refute(dispatched, "an unauthorized request must never reach dispatch") + end + + # The extension raises `TypeError` for a non-string, and no rescue caught it, so it reached + # `run` and stopped the server. + def test_a_request_without_a_query_string_is_a_client_error + core = Core.new(@state) + + [nil, 12345, ["MATCH (n) RETURN n"], { "q" => 1 }].each do |query| + response = core.send(:handle_query, { "command" => "query", "query" => query }) + + assert_equal(1, response["status"], "expected #{query.inspect} to be refused") + assert_match(/no query string/, response["stderr"]) + end + end + + def test_a_request_with_a_malformed_query_format_is_a_client_error + # A real graph ensures the request reaches the extension and raises the `TypeError` this + # test prevents, not a missing-graph error. + core = with_graph + + # `false` is in the list because `|| "table"` would turn it into the default. + [12345, ["json"], { "format" => "json" }, false].each do |format| + response = core.send( + :handle_query, + { "command" => "query", "query" => "MATCH (n) RETURN n", "query_format" => format }, + ) + + assert_equal(1, response["status"], "expected #{format.inspect} to be refused") + assert_match(/no query format string/, response["stderr"]) + end + end + + def test_a_request_without_a_query_format_uses_the_default + core = with_graph + + response = core.send(:handle_query, { "command" => "query", "query" => "MATCH (c:Class) RETURN c.name" }) + + assert_equal(0, response["status"]) + end + + # Invalid Cypher is the caller's mistake, so it must return the parser's message, not an + # internal error. + def test_invalid_cypher_is_reported_as_a_user_error + core = with_graph + + response = exchange_with(core, { "command" => "query", "query" => "NOT A QUERY" }) + + assert_equal(1, response["status"]) + assert_match(/Cypher syntax error/, response["stderr"]) + refute_match(/internal error/, response["stderr"]) + end + + private + + # The graph is empty because these tests check request handling, not what the graph holds. + #: -> Core + def with_graph + core = Core.new(@state) + core.instance_variable_set(:@graph, Rubydex::Graph.configure_for_workspace(@state.workspace_path)) + core + end + + #: (Core core, Hash[String, untyped] payload, ?token: String?) -> Hash[untyped, untyped] + def exchange_with(core, payload, token: nil) + server_side, client_side = UNIXSocket.pair + + client = Thread.new do + client_side.gets + Frame.write(client_side, payload.merge("token" => token || @state.token)) + Frame.read_response(client_side, total_timeout: 10.0) + ensure + client_side.close + end + + @log, _err = capture_io { core.send(:handle, server_side) } + client.value + ensure + client&.join + end + end + end +end diff --git a/test/server/frame_test.rb b/test/server/frame_test.rb new file mode 100644 index 000000000..8346f32cc --- /dev/null +++ b/test/server/frame_test.rb @@ -0,0 +1,289 @@ +# frozen_string_literal: true + +require "test_helper" +require "rubydex/server" +require "socket" + +module Rubydex + module Server + class FrameTest < Minitest::Test + def setup + # Framing needs a socket pair and nothing else: no `State`, no fork, no `O_NOFOLLOW`. + skip("this platform has no UNIX sockets") unless defined?(::UNIXSocket) + end + + def test_write_then_read_round_trips_a_payload + a, b = UNIXSocket.pair + + payload = { "command" => "query", "query" => "MATCH (n) RETURN n", "token" => "abc" } + Frame.write(a, payload) + + assert_equal(payload, Frame.read_request(b)) + ensure + a&.close + b&.close + end + + def test_read_returns_nil_on_eof + a, b = UNIXSocket.pair + a.close + + assert_nil(Frame.read_request(b)) + ensure + b&.close + end + + def test_read_handles_large_payloads + a, b = UNIXSocket.pair + + payload = { "stdout" => "x" * 100_000 } + Thread.new { Frame.write(a, payload) } + + assert_equal(payload, Frame.read_request(b)) + ensure + a&.close + b&.close + end + + def test_read_line_raises_when_nothing_arrives + a, b = UNIXSocket.pair + + assert_raises(Frame::ReadTimeout) do + Frame.read_line(b, timeout: 0.05) + end + ensure + a&.close + b&.close + end + + def test_read_line_returns_the_line + a, b = UNIXSocket.pair + a.puts("0.2.5:fingerprint") + + assert_equal("0.2.5:fingerprint", Frame.read_line(b, timeout: 1.0).chomp) + ensure + a&.close + b&.close + end + + def test_read_rejects_a_length_line_that_is_not_a_number + a, b = UNIXSocket.pair + a.puts("100garbage") + a.write("x" * 100) + a.flush + + # `String#to_i` would accept this line and return 100. + error = assert_raises(Frame::Malformed) { Frame.read_request(b, timeout: 1.0) } + + assert_match(/expected a decimal length/, error.message) + ensure + a&.close + b&.close + end + + def test_read_rejects_a_payload_that_is_not_an_object + a, b = UNIXSocket.pair + body = JSON.dump([1, 2, 3]) + a.puts(body.bytesize) + a.write(body) + a.flush + + error = assert_raises(Frame::Malformed) { Frame.read_request(b, timeout: 1.0) } + + assert_match(/expected a JSON object/, error.message) + ensure + a&.close + b&.close + end + + # The JSON is valid on its own, so only the length check rejects this short body as a complete + # frame. + def test_read_rejects_a_truncated_frame + a, b = UNIXSocket.pair + a.puts(100) + a.write("{}") + a.flush + a.close + + error = assert_raises(Frame::Malformed) { Frame.read_request(b, timeout: 1.0) } + + assert_match(/closed the connection after 2 of 100 bytes/, error.message) + ensure + a&.close unless a&.closed? + b&.close + end + + def test_read_request_times_out_on_a_short_body + a, b = UNIXSocket.pair + a.puts(1000) + a.write("only a few bytes") + a.flush + + assert_raises(Frame::ReadTimeout) { Frame.read_request(b, timeout: 0.2) } + ensure + a&.close + b&.close + end + + # A response may be slow but steady, so the read must survive the gap between two chunks, not a + # stall. + def test_read_response_accepts_a_body_that_arrives_slowly_but_steadily + a, b = UNIXSocket.pair + payload = { "stdout" => "abcdefgh" } + body = JSON.dump(payload) + + writer = Thread.new do + a.puts(body.bytesize) + body.each_char do |char| + a.write(char) + a.flush + sleep(0.05) + end + end + + # The transfer takes about 0.4 seconds in total, which is longer than the idle timeout. + assert_equal(payload, Frame.read_response(b, idle_timeout: 0.25)) + ensure + writer&.join + a&.close + b&.close + end + + # A server sends nothing until the query finishes, so a client must wait for the first byte + # however long the query takes. + def test_read_response_waits_without_a_limit_for_the_first_byte + a, b = UNIXSocket.pair + payload = { "stdout" => "the query finished" } + + writer = Thread.new do + sleep(0.3) + Frame.write(a, payload) + end + + # The idle timeout is far shorter than the wait for the first byte. + assert_equal(payload, Frame.read_response(b, idle_timeout: 0.05)) + ensure + writer&.join + a&.close + b&.close + end + + def test_read_response_still_times_out_after_the_first_byte_arrives + a, b = UNIXSocket.pair + a.puts(1000) + a.write("a partial body") + a.flush + + assert_raises(Frame::ReadTimeout) do + Frame.read_response(b, idle_timeout: 0.2) + end + ensure + a&.close + b&.close + end + + # A query result may be larger than any request, so the two directions carry different caps. + def test_read_response_accepts_a_payload_above_the_request_cap + assert_operator(Frame::MAX_RESPONSE_BYTES, :>, Frame::MAX_REQUEST_BYTES) + + a, b = UNIXSocket.pair + payload = { "stdout" => "x" * (Frame::MAX_REQUEST_BYTES + 1) } + writer = Thread.new { Frame.write(a, payload) } + + assert_equal(payload, Frame.read_response(b)) + ensure + writer&.join + a&.close + b&.close + end + + def test_read_rejects_a_response_sized_payload_on_the_request_path + a, b = UNIXSocket.pair + a.puts(Frame::MAX_REQUEST_BYTES + 1) + a.flush + + error = assert_raises(Frame::Malformed) { Frame.read_request(b, timeout: 1.0) } + + assert_match(/exceeds the limit of #{Frame::MAX_REQUEST_BYTES}/, error.message) + ensure + a&.close + b&.close + end + + # A request needs an absolute deadline, because a client that drips within the idle limit could + # otherwise hold the single-client server forever. + def test_read_request_stops_a_client_that_drips_bytes + a, b = UNIXSocket.pair + body = JSON.dump({ "command" => "query", "query" => "MATCH (n) RETURN n" }) + + writer = Thread.new do + a.puts(body.bytesize) + body.each_char do |char| + a.write(char) + a.flush + sleep(0.05) + end + rescue IOError, Errno::EPIPE + nil + end + + assert_raises(Frame::ReadTimeout) { Frame.read_request(b, timeout: 0.3) } + ensure + b&.close + writer&.kill + a&.close + end + + # The module raises its own error types, so a caller never rescues `JSON::ParserError` for a + # body that is not JSON. + def test_read_reports_a_body_that_is_not_json_as_a_malformed_frame + a, b = UNIXSocket.pair + body = "this is not json" + a.puts(body.bytesize) + a.write(body) + a.flush + + error = assert_raises(Frame::Malformed) { Frame.read_request(b, timeout: 1.0) } + + assert_match(/not valid JSON/, error.message) + ensure + a&.close + b&.close + end + + def test_read_response_reports_a_body_that_is_not_json_as_a_malformed_frame + a, b = UNIXSocket.pair + body = "{ \"stdout\": " + a.puts(body.bytesize) + a.write(body) + a.flush + + assert_raises(Frame::Malformed) { Frame.read_response(b) } + ensure + a&.close + b&.close + end + + # `IO#read_nonblock` allocates its `maxlen` before it reads, so the reader must cap each read at + # one chunk, not the advertised length. + def test_read_asks_for_no_more_than_one_chunk_per_read + a, b = UNIXSocket.pair + requested = [] + b.define_singleton_method(:read_nonblock) do |maxlen, **kwargs| + requested << maxlen + super(maxlen, **kwargs) + end + + payload = { "stdout" => "x" * (Frame::CHUNK_BYTES * 3) } + writer = Thread.new { Frame.write(a, payload) } + + assert_equal(payload, Frame.read_response(b)) + assert_operator(requested.max, :<=, Frame::CHUNK_BYTES) + ensure + writer&.join + a&.close + b&.close + end + end + end +end diff --git a/test/server/integration_test.rb b/test/server/integration_test.rb new file mode 100644 index 000000000..c8aa0f12a --- /dev/null +++ b/test/server/integration_test.rb @@ -0,0 +1,202 @@ +# frozen_string_literal: true + +require "test_helper" +require "helpers/context" +require "rubydex/server" +require "open3" +require "rbconfig" +require "tmpdir" + +module Rubydex + module Server + # The server runs as a fresh subprocess, not a fork, so it does not inherit the loaded native + # extension, as the real CLI does. + class IntegrationTest < Minitest::Test + include Test::Helpers::WithContext + + LIB_DIR = File.expand_path("../../lib", __dir__) #: String + EXE = File.expand_path("../../exe/rdx", __dir__) #: String + + def setup + skip("server mode unsupported on this platform") unless Server.supported? + + @runtime_dir = Dir.mktmpdir("rdx-server-integration") + @previous_server_dir = ENV["RDX_SERVER_DIR"] + ENV["RDX_SERVER_DIR"] = @runtime_dir + @contexts = [] + end + + def teardown + @contexts&.each do |context| + rdx(["server", "stop"], context) + rescue StandardError + nil + end + + # The runtime directory goes away on the next line, and the daemon's log is inside it. + report_server_logs unless passed? + + ENV["RDX_SERVER_DIR"] = @previous_server_dir if @runtime_dir + FileUtils.rm_rf(@runtime_dir) if @runtime_dir + end + + def test_warm_query_matches_repeated_calls + with_context do |context| + track(context) + context.write!("zoo.rb", <<~RUBY) + class Animal; end + class Dog < Animal; end + class Cat < Animal; end + RUBY + + query = "MATCH (c:Class)-[:HAS_PARENT]->(p:Class) WHERE p.name = 'Animal' RETURN c.name ORDER BY c.name" + output = query!(context, query) + + assert_match(/Cat/, output) + assert_match(/Dog/, output) + assert_match(/2 rows/, output) + + assert(server_running?(context)) + assert_equal(output, query!(context, query)) + end + end + + def test_boots_and_indexes_files_in_subdirectories + with_context do |context| + track(context) + # Nested directories reach the workspace manifest's recursive walk, which can crash the + # daemon at boot on a non-flat codebase. + context.write!("app/models/animal.rb", "class Animal; end\n") + context.write!("app/models/dog.rb", "class Dog < Animal; end\n") + + query = "MATCH (c:Class)-[:HAS_PARENT]->(p:Class) WHERE p.name = 'Animal' RETURN c.name" + output = query!(context, query) + + assert_match(/Dog/, output) + assert(server_running?(context)) + end + end + + def test_query_output_matches_inline + with_context do |context| + track(context) + context.write!("zoo.rb", "class Animal; end\nclass Dog < Animal; end\n") + + query = "MATCH (c:Class {name: 'Dog'}) RETURN c.name" + inline, _, inline_status = rdx(["query", query], context) + warm = query!(context, query) + + assert_predicate(inline_status, :success?) + assert_equal(inline, warm) + end + end + + def test_start_status_and_stop + with_context do |context| + track(context) + context.write!("foo.rb", "class Foo; end") + + out, err, status = rdx(["server", "start"], context) + assert_predicate(status, :success?, "start failed: #{err}") + assert_match(/rdx server started/, out) + assert(server_running?(context)) + + status_out, _, _ = rdx(["server", "status"], context) + assert_match(/rdx server running/, status_out) + assert_match(/pid:/, status_out) + + stop_out, _, _ = rdx(["server", "stop"], context) + assert_match(/rdx server stopped/, stop_out) + refute(server_running?(context)) + + status_out, _, _ = rdx(["server", "status"], context) + assert_match(/not running/, status_out) + end + end + + def test_restart_replaces_the_server + with_context do |context| + track(context) + context.write!("foo.rb", "class Foo; end") + + rdx!(["server", "start"], context) + first_pid = state(context).server_pid + refute_nil(first_pid, "the started server recorded no pid") + + rdx!(["server", "restart"], context) + second_pid = state(context).server_pid + + refute_equal(first_pid, second_pid) + assert(server_running?(context)) + end + end + + private + + # Never prints `state.json`: it holds the per-boot token, and CI logs are public. + #: -> void + def report_server_logs + return unless @runtime_dir + + # Names and sizes only, so an absent log is distinguishable from an empty one. + Dir.glob(File.join(@runtime_dir, "**", "*"), File::FNM_DOTMATCH).sort.each do |path| + warn(" #{File.directory?(path) ? "dir " : "file"} #{File.size(path).to_s.rjust(7)} #{path}") + end + + Dir.glob(File.join(@runtime_dir, "**", "server.log")).sort.each do |path| + warn("--- #{path} ---") + warn(File.read(path)) + rescue StandardError => e + warn("--- #{path} unreadable: #{e.class} ---") + end + end + + #: (Test::Helpers::Context context) -> void + def track(context) + @contexts << context + end + + #: (Test::Helpers::Context context) -> State + def state(context) + State.new(workspace_path: context.absolute_path) + end + + #: (Test::Helpers::Context context) -> bool + def server_running?(context) + state(context).server_running? + end + + #: (Array[String] args, Test::Helpers::Context context) -> [String, String, Process::Status] + def rdx(args, context) + Open3.capture3( + RbConfig.ruby, + "-I", + LIB_DIR, + EXE, + *args, + chdir: context.absolute_path, + ) + end + + # Reports what a failed command printed, because an unresponsive server reports on stdout and + # a refusal on stderr. + #: (Array[String] args, Test::Helpers::Context context) -> String + def rdx!(args, context) + out, err, status = rdx(args, context) + said = { "stdout" => out, "stderr" => err }.filter_map do |name, text| + "#{name}: #{text.strip}" unless text.strip.empty? + end + + assert_predicate(status, :success?, ["`rdx #{args.join(" ")}` failed", *said].join("\n")) + out + end + + #: (Test::Helpers::Context context, String query) -> String + def query!(context, query) + out, err, status = rdx(["query", query, "--server"], context) + assert_predicate(status, :success?, "query failed: #{err}") + out + end + end + end +end diff --git a/test/server/state_test.rb b/test/server/state_test.rb new file mode 100644 index 000000000..7127cc8d2 --- /dev/null +++ b/test/server/state_test.rb @@ -0,0 +1,477 @@ +# frozen_string_literal: true + +require "test_helper" +require "rubydex/server" +require "tmpdir" + +module Rubydex + module Server + # Shared: the runtime suite proves the directory code refuses without the flag, and the + # platform suite proves that `supported?` reports the same. + module NofollowStub + private + + # A value of `nil` represents a platform that lacks the flag. + #: [T] (Integer? value) { -> T } -> T + def with_nofollow(value) + original = State::NOFOLLOW + State.send(:remove_const, :NOFOLLOW) + State.const_set(:NOFOLLOW, value) + yield + ensure + State.send(:remove_const, :NOFOLLOW) + State.const_set(:NOFOLLOW, original) + end + end + + class StateIdentityTest < Minitest::Test + def test_app_id_is_stable_for_the_same_workspace + a = State.new(workspace_path: "/some/workspace") + b = State.new(workspace_path: "/some/workspace") + + assert_equal(a.app_id, b.app_id) + end + + def test_app_id_differs_between_workspaces + a = State.new(workspace_path: "/workspace/a") + b = State.new(workspace_path: "/workspace/b") + + refute_equal(a.app_id, b.app_id) + end + + # A protocol change must map the workspace to a different runtime directory, or a new client + # reaches a server that speaks the older wire format. + def test_app_id_differs_between_protocol_versions + before = State.new(workspace_path: "/some/workspace").app_id + after = with_protocol(Server::PROTOCOL + 1) do + State.new(workspace_path: "/some/workspace").app_id + end + + refute_equal(before, after) + end + + def test_expected_version_includes_gem_version + state = State.new(workspace_path: "/workspace") + assert_match(/\A#{Regexp.escape(Rubydex::VERSION)}:/, state.expected_version) + end + + private + + # `remove_const` prevents the Ruby warning about the constant reassignment. + #: [T] (Integer value) { -> T } -> T + def with_protocol(value) + original = Server::PROTOCOL + Server.send(:remove_const, :PROTOCOL) + Server.const_set(:PROTOCOL, value) + yield + ensure + Server.send(:remove_const, :PROTOCOL) + Server.const_set(:PROTOCOL, original) + end + end + + class StateTest < Minitest::Test + include NofollowStub + + def setup + skip("server mode is not supported on this platform") unless Server.supported? + + @runtime_dir = File.realpath(Dir.mktmpdir("rdx-server-test")) + @previous_server_dir = ENV["RDX_SERVER_DIR"] + ENV["RDX_SERVER_DIR"] = @runtime_dir + end + + def teardown + ENV["RDX_SERVER_DIR"] = @previous_server_dir + FileUtils.rm_rf(@runtime_dir) + end + + def test_ensure_dir_creates_directory_with_restrictive_permissions + state = State.new(workspace_path: "/workspace") + state.ensure_dir! + + assert(File.directory?(state.dir)) + + skip("POSIX permissions are not enforced on this platform") if Gem.win_platform? + + assert_equal(0o700, File.stat(state.dir).mode & 0o777) + end + + def test_open_lock_creates_the_file_with_restrictive_permissions + state = State.new(workspace_path: "/workspace") + state.open_lock.close + + assert(File.exist?(state.lock_path)) + + skip("POSIX permissions are not enforced on this platform") if Gem.win_platform? + + assert_equal(0o600, File.stat(state.lock_path).mode & 0o777) + end + + def test_record_writes_the_state_file_with_restrictive_permissions + state = State.new(workspace_path: "/workspace") + state.record! + + assert(File.exist?(state.state_path)) + + skip("POSIX permissions are not enforced on this platform") if Gem.win_platform? + + assert_equal(0o600, File.stat(state.state_path).mode & 0o777) + end + + # A rename replaces the whole file, so no reader sees a partial record and no temporary file + # survives the write. + def test_record_leaves_no_temporary_file_behind + state = State.new(workspace_path: "/workspace") + state.record! + + assert_equal(["state.json"], Dir.children(state.dir).sort) + end + + def test_record_stores_the_identity_of_this_process + state = State.new(workspace_path: "/workspace") + state.record! + + assert_equal(Process.pid, state.server_pid) + assert(state.version_compatible?) + refute_empty(state.token) + assert_match(/\A\d{4}-\d{2}-\d{2}T/, state.started_at) + end + + # The token authenticates a request, so a client that cached a previous server's token must + # not talk to the new one. + def test_record_issues_a_fresh_token_on_every_boot + state = State.new(workspace_path: "/workspace") + + state.record! + first = State.new(workspace_path: "/workspace").token + state.record! + second = State.new(workspace_path: "/workspace").token + + refute_empty(first) + refute_equal(first, second) + end + + def test_read_is_nil_without_a_state_file + state = State.new(workspace_path: "/workspace") + + assert_nil(state.read) + assert_nil(state.token) + assert_nil(state.server_pid) + assert_nil(state.started_at) + refute(state.version_compatible?) + end + + # `write_state` renames a temporary file, so this record came from somewhere else. `read` + # must return `nil` for it, and must not raise. + def test_read_tolerates_a_half_written_record + state = State.new(workspace_path: "/workspace") + state.ensure_dir! + File.write(state.state_path, '{"pid": 12') + + assert_nil(state.read) + assert_nil(state.server_pid) + end + + def test_server_pid_rejects_an_implausible_value + state = State.new(workspace_path: "/workspace") + state.ensure_dir! + + [0, -1, "4711", nil].each do |value| + File.write(state.state_path, JSON.dump({ "pid" => value })) + assert_nil(state.server_pid, "expected #{value.inspect} to be rejected") + end + end + + def test_server_running_is_false_without_a_state_file + state = State.new(workspace_path: "/workspace") + + refute(state.server_running?) + end + + # The lock decides ownership, so a live recorded pid without a lock is not a running server. + def test_server_running_ignores_a_live_but_unrelated_recorded_pid + state = State.new(workspace_path: "/workspace") + state.ensure_dir! + File.write(state.state_path, JSON.dump({ "pid" => Process.pid, "version" => state.expected_version })) + + assert_equal(Process.pid, state.server_pid) + refute(state.server_running?) + end + + def test_server_running_follows_the_lock_across_processes + skip("fork is unavailable on this platform") unless Process.respond_to?(:fork) + + state = State.new(workspace_path: "/workspace") + + with_locking_child(state) do + assert(state.server_running?, "expected the lock held by the child to count as running") + end + + refute(state.server_running?, "expected the lock to go free once the holder died") + end + + def test_clean_removes_the_socket_but_keeps_the_lock_file + state = State.new(workspace_path: "/workspace") + state.open_lock.close + File.write(state.socket_path, "") + + state.clean! + + refute(File.exist?(state.socket_path)) + # A lock belongs to an inode, so deleting this path would let two servers start at once. + assert(File.exist?(state.lock_path)) + end + + def test_clean_leaves_the_socket_of_a_locked_workspace + skip("fork is unavailable on this platform") unless Process.respond_to?(:fork) + + state = State.new(workspace_path: "/workspace") + + with_locking_child(state) do + File.write(state.socket_path, "") + + state.clean! + + assert(File.exist?(state.socket_path)) + end + end + + # On Windows an exclusive lock denies read access to the whole file, so the locked file stays + # empty and the record lives outside it. + def test_the_record_stays_readable_while_a_server_holds_the_lock + skip("fork is unavailable on this platform") unless Process.respond_to?(:fork) + + state = State.new(workspace_path: "/workspace") + + with_locking_child(state) do |child_pid| + refute_equal(state.lock_path, state.state_path) + assert_equal(0, File.size(state.lock_path), "the locked file must never carry data") + + assert_equal(child_pid, state.server_pid) + assert(state.version_compatible?) + refute_nil(state.token) + end + end + + # The runtime files go in a subdirectory this code creates, so `RDX_SERVER_DIR` keeps the + # permissions its owner set. + def test_ensure_dir_leaves_the_override_directory_alone + skip("POSIX permissions are not enforced on this platform") if Gem.win_platform? + + File.chmod(0o755, @runtime_dir) + state = State.new(workspace_path: "/workspace") + state.ensure_dir! + + assert_equal( + 0o755, + File.stat(@runtime_dir).mode & 0o777, + "the directory named by RDX_SERVER_DIR must keep its own permissions", + ) + assert_equal(0o700, File.stat(state.dir).mode & 0o777) + assert_equal(0o700, File.stat(File.dirname(state.dir)).mode & 0o777) + end + + def test_the_runtime_directory_nests_under_the_override + state = State.new(workspace_path: "/workspace") + uid = Process.respond_to?(:uid) ? Process.uid : "nobody" + + assert_equal(File.join(@runtime_dir, "rubydex-#{uid}", state.app_id), state.dir) + end + + # `chmod` follows symlinks and the runtime root lives in a world-writable temp dir, so a + # planted link is refused before it changes the target's permissions. + def test_ensure_dir_refuses_a_symlinked_runtime_root + skip("symlinks are unavailable on this platform") if Gem.win_platform? + + target = Dir.mktmpdir("rdx-symlink-target") + File.chmod(0o755, target) + uid = Process.respond_to?(:uid) ? Process.uid : "nobody" + File.symlink(target, File.join(@runtime_dir, "rubydex-#{uid}")) + + state = State.new(workspace_path: "/workspace") + error = assert_raises(Error) { state.ensure_dir! } + + assert_match(/is a symlink/, error.message) + assert_equal( + 0o755, + File.stat(target).mode & 0o777, + "an unrelated directory must not have its permissions changed", + ) + ensure + FileUtils.rm_rf(target) if target + end + + def test_runtime_dir_honors_override + state = State.new(workspace_path: "/workspace") + assert(state.dir.start_with?(@runtime_dir)) + end + + def test_a_runtime_root_open_to_others_is_refused_before_locking + state = State.new(workspace_path: "/workspace") + state.ensure_dir! + File.chmod(0o755, File.dirname(state.dir)) + + error = assert_raises(Error) { state.server_running? } + + assert_match(/is not private to this user/, error.message) + end + + # The lock and state file reject a symlink only on their own last component, so the runtime + # root is checked before the client trusts its socket or token. + def test_a_symlinked_runtime_root_is_refused_before_reading + state = State.new(workspace_path: "/workspace") + state.ensure_dir! + root = File.dirname(state.dir) + elsewhere = File.join(@runtime_dir, "elsewhere") + FileUtils.mv(root, elsewhere) + File.symlink(elsewhere, root) + + assert_raises(Error) { state.token } + assert_raises(Error) { state.server_running? } + end + + def test_liveness_is_false_before_anything_is_created + refute_predicate(State.new(workspace_path: "/workspace"), :server_running?) + end + + def test_a_base_writable_by_others_is_refused + open_base = File.join(@runtime_dir, "shared") + FileUtils.mkdir_p(open_base) + File.chmod(0o777, open_base) + ENV["RDX_SERVER_DIR"] = open_base + + error = assert_raises(Error) { State.new(workspace_path: "/workspace").ensure_dir! } + + assert_match(/writable by other users and is not sticky/, error.message) + end + + def test_a_sticky_base_writable_by_others_is_accepted + sticky_base = File.join(@runtime_dir, "sticky") + FileUtils.mkdir_p(sticky_base) + File.chmod(0o1777, sticky_base) + ENV["RDX_SERVER_DIR"] = sticky_base + + state = State.new(workspace_path: "/workspace") + state.ensure_dir! + + assert_equal(0o700, File.stat(state.dir).mode & 0o777) + end + + def test_a_base_that_does_not_exist_is_refused_rather_than_created + missing = File.join(@runtime_dir, "absent") + ENV["RDX_SERVER_DIR"] = missing + + error = assert_raises(Error) { State.new(workspace_path: "/workspace").ensure_dir! } + + assert_match(/cannot hold the server runtime directory/, error.message) + refute_path_exists(missing) + end + + # The base is resolved first because the link entry itself, not its target, is what an owner + # can swap. + def test_the_base_is_resolved_before_it_is_checked + open_parent = File.join(@runtime_dir, "open") + real_base = File.join(open_parent, "real") + FileUtils.mkdir_p(real_base) + File.chmod(0o777, open_parent) + link = File.join(@runtime_dir, "link") + File.symlink(real_base, link) + ENV["RDX_SERVER_DIR"] = link + + error = assert_raises(Error) { State.new(workspace_path: "/workspace").ensure_dir! } + + assert_match(%r{#{Regexp.escape(File.realpath(open_parent))} is writable by other users}, error.message) + end + + def test_a_planted_symlink_never_becomes_the_state_file + state = State.new(workspace_path: "/workspace") + state.ensure_dir! + victim = File.join(@runtime_dir, "victim") + File.write(victim, "precious") + File.symlink(victim, "#{state.state_path}.#{Process.pid}") + + assert_raises(Error) { state.record! } + + assert_equal("precious", File.read(victim)) + end + + def test_a_planted_symlink_never_becomes_the_lock + state = State.new(workspace_path: "/workspace") + state.ensure_dir! + victim = File.join(@runtime_dir, "victim") + File.write(victim, "precious") + File.symlink(victim, state.lock_path) + + assert_raises(Error) { state.server_running? } + + assert_equal("precious", File.read(victim)) + end + + def test_a_platform_without_nofollow_refuses_to_create_the_directory + state = State.new(workspace_path: "/workspace") + + error = assert_raises(Error) do + with_nofollow(nil) { state.ensure_dir! } + end + + assert_match(/cannot open a path without following symlinks/, error.message) + end + + private + + # The child is killed so the kernel releases the lock, not the child. + #: [T] (State state) { (Integer child_pid) -> T } -> T + def with_locking_child(state) + reader, writer = IO.pipe + + pid = fork do + reader.close + file = state.open_lock + file.flock(File::LOCK_EX) + state.record! + writer.puts("locked") + writer.close + sleep(60) + end + + writer.close + reader.gets # Waits until the child holds the lock and writes its record. + yield(pid) + ensure + reader&.close + if pid + begin + Process.kill("KILL", pid) + Process.wait(pid) + rescue Errno::ESRCH, Errno::ECHILD + nil + end + end + end + end + + # Runs on every platform, including the ones that cannot serve. The suites above skip there, so + # without this nothing would check what an unsupported platform reports. + class PlatformSupportTest < Minitest::Test + include NofollowStub + + def test_a_platform_without_nofollow_does_not_support_server_mode + with_nofollow(nil) { refute_predicate(Server, :supported?) } + end + + def test_windows_does_not_support_server_mode + skip("this platform is not Windows") unless Gem.win_platform? + + refute_predicate(Server, :supported?) + end + + # Every path that runs only when `supported?` is true relies on `nofollow!` never raising. + def test_a_supported_platform_always_has_the_nofollow_flag + skip("server mode is not supported on this platform") unless Server.supported? + + refute_nil(State::NOFOLLOW) + end + end + end +end