From 29dc822b9da28c31d212189cb7618debf01f726d Mon Sep 17 00:00:00 2001 From: Tom Lehman Date: Thu, 16 Oct 2025 11:37:30 -0400 Subject: [PATCH] Use IPC instead of HTTP for performance --- app/jobs/gap_detection_job.rb | 4 +- app/services/eth_block_importer.rb | 10 +- app/services/geth_driver.rb | 35 ++++- lib/eth_rpc_client.rb | 225 +++++++++++++++++++++++------ lib/geth_client.rb | 91 ------------ 5 files changed, 219 insertions(+), 146 deletions(-) delete mode 100644 lib/geth_client.rb diff --git a/app/jobs/gap_detection_job.rb b/app/jobs/gap_detection_job.rb index 49981f4..d43f1ae 100644 --- a/app/jobs/gap_detection_job.rb +++ b/app/jobs/gap_detection_job.rb @@ -60,7 +60,7 @@ def get_import_range return nil if latest_l2_block_number == 0 # Get L1 attributes to find the corresponding L1 block - l1_attributes = GethDriver.client.get_l1_attributes(latest_l2_block_number) + l1_attributes = GethDriver.get_l1_attributes(latest_l2_block_number) current_l1_block = l1_attributes[:number] # Check the last validated block @@ -92,7 +92,7 @@ def get_l2_blocks_for_l1_block(l1_block_number) # Search backwards from current L2 tip to find blocks from this L1 block # This is expensive but necessary for gap filling (0..latest_l2_block_number).reverse_each do |l2_block_num| - l1_attributes = GethDriver.client.get_l1_attributes(l2_block_num) + l1_attributes = GethDriver.get_l1_attributes(l2_block_num) if l1_attributes[:number] == l1_block_number l2_block = GethDriver.client.call("eth_getBlockByNumber", ["0x#{l2_block_num.to_s(16)}", false]) diff --git a/app/services/eth_block_importer.rb b/app/services/eth_block_importer.rb index d2710ca..bd1fdf9 100644 --- a/app/services/eth_block_importer.rb +++ b/app/services/eth_block_importer.rb @@ -74,7 +74,7 @@ def populate_ethscriptions_block_cache current_block = EthscriptionsBlock.from_rpc_result(block_data) ImportProfiler.start("l1_attributes_fetch") - l1_attributes = GethDriver.client.get_l1_attributes(current_block.number) + l1_attributes = GethDriver.get_l1_attributes(current_block.number) ImportProfiler.stop("l1_attributes_fetch") current_block.assign_l1_attributes(l1_attributes) @@ -107,7 +107,7 @@ def current_block_number # Removed batch processing - now imports one block at a time def find_first_l2_block_in_epoch(l2_block_number_candidate) - l1_attributes = GethDriver.client.get_l1_attributes(l2_block_number_candidate) + l1_attributes = GethDriver.get_l1_attributes(l2_block_number_candidate) if l1_attributes[:sequence_number] == 0 return l2_block_number_candidate @@ -124,7 +124,7 @@ def set_eth_block_starting_points l1_block = EthRpcClient.l1.get_block(SysConfig.l1_genesis_block_number) eth_block = EthBlock.from_rpc_result(l1_block) ethscriptions_block = EthscriptionsBlock.from_rpc_result(latest_l2_block) - l1_attributes = GethDriver.client.get_l1_attributes(latest_l2_block_number) + l1_attributes = GethDriver.get_l1_attributes(latest_l2_block_number) ethscriptions_block.assign_l1_attributes(l1_attributes) @@ -134,7 +134,7 @@ def set_eth_block_starting_points return [eth_block.number, 0] end - l1_attributes = GethDriver.client.get_l1_attributes(latest_l2_block_number) + l1_attributes = GethDriver.get_l1_attributes(latest_l2_block_number) l1_candidate = l1_attributes[:number] l2_candidate = latest_l2_block_number @@ -148,7 +148,7 @@ def set_eth_block_starting_points l1_result = ethereum_client.get_block(l1_candidate) l1_hash = Hash32.from_hex(l1_result['hash']) - l1_attributes = GethDriver.client.get_l1_attributes(l2_candidate) + l1_attributes = GethDriver.get_l1_attributes(l2_candidate) l2_block = GethDriver.client.call("eth_getBlockByNumber", ["0x#{l2_candidate.to_s(16)}", false]) diff --git a/app/services/geth_driver.rb b/app/services/geth_driver.rb index c6ea99d..01c0030 100644 --- a/app/services/geth_driver.rb +++ b/app/services/geth_driver.rb @@ -1,14 +1,41 @@ module GethDriver extend self attr_reader :password - + include Memery + def client - @_client ||= GethClient.new(ENV.fetch('GETH_RPC_URL')) + @_client ||= EthRpcClient.l2_engine end - + def non_auth_client - @_non_auth_client ||= GethClient.new(non_authed_rpc_url) + @_non_auth_client ||= EthRpcClient.l2 + end + + def get_l1_attributes(l2_block_number) + if l2_block_number > 0 + l2_block = EthRpcClient.l2.call("eth_getBlockByNumber", ["0x#{l2_block_number.to_s(16)}", true]) + l2_attributes_tx = l2_block['transactions'].first + L1AttributesTxCalldata.decode( + ByteString.from_hex(l2_attributes_tx['input']), + l2_block_number + ) + else + l1_block = EthRpcClient.l1.get_block(SysConfig.l1_genesis_block_number) + eth_block = EthBlock.from_rpc_result(l1_block) + { + timestamp: eth_block.timestamp, + number: eth_block.number, + base_fee: eth_block.base_fee_per_gas, + blob_base_fee: 1, + hash: eth_block.block_hash, + batcher_hash: Hash32.from_bin("\x00".b * 32), + sequence_number: 0, + base_fee_scalar: 0, + blob_base_fee_scalar: 1 + }.with_indifferent_access + end end + memoize :get_l1_attributes def non_authed_rpc_url ENV.fetch('NON_AUTH_GETH_RPC_URL') diff --git a/lib/eth_rpc_client.rb b/lib/eth_rpc_client.rb index 61148ca..bb9e7e2 100644 --- a/lib/eth_rpc_client.rb +++ b/lib/eth_rpc_client.rb @@ -1,7 +1,9 @@ class EthRpcClient + include Memery + class HttpError < StandardError attr_reader :code, :http_message - + def initialize(code, http_message) @code = code @http_message = http_message @@ -13,18 +15,48 @@ class ExecutionRevertedError < StandardError; end class MethodRequiredError < StandardError; end attr_accessor :base_url, :http - def initialize(base_url = ENV['L1_RPC_URL']) + def initialize(base_url = ENV['L1_RPC_URL'], jwt_secret: nil, retry_config: {}) self.base_url = base_url - @uri = URI(base_url) - @http = Net::HTTP::Persistent.new( - name: "eth_rpc_#{@uri.host}:#{@uri.port}", - pool_size: 100 # Increase pool size from default 64 - ) - @http.open_timeout = 10 # 10 seconds to establish connection - @http.read_timeout = 30 # 30 seconds for slow eth_call operations - @http.idle_timeout = 30 # Keep connections alive for 30 seconds + @request_id = 0 + @mutex = Mutex.new + + # JWT support (optional, only for HTTP) + @jwt_secret = jwt_secret + @jwt_enabled = !jwt_secret.nil? + + if @jwt_enabled + @jwt_secret_decoded = ByteString.from_hex(jwt_secret).to_bin + end + + # Customizable retry configuration + @retry_config = { + tries: 7, + base_interval: 1, + max_interval: 32, + multiplier: 2, + rand_factor: 0.4 + }.merge(retry_config) + + # Detect transport mode + @mode = detect_mode(base_url) + + if @mode == :ipc + @ipc_path = base_url.start_with?("ipc://") ? base_url.delete_prefix("ipc://") : base_url + @ipc_socket = nil + @ipc_mutex = Mutex.new + Rails.logger.info "EthRpcClient using IPC at: #{@ipc_path}" + else + @uri = URI(base_url) + @http = Net::HTTP::Persistent.new( + name: "eth_rpc_#{@uri.host}:#{@uri.port}", + pool_size: 100 # Increase pool size from default 64 + ) + @http.open_timeout = 10 # 10 seconds to establish connection + @http.read_timeout = 30 # 30 seconds for slow eth_call operations + @http.idle_timeout = 30 # Keep connections alive for 30 seconds + end end - + def self.l1 @_l1_client ||= new(ENV.fetch('L1_RPC_URL')) end @@ -33,6 +65,14 @@ def self.l2 @_l2_client ||= new(ENV.fetch('NON_AUTH_GETH_RPC_URL')) end + def self.l2_engine + @_l2_engine_client ||= new( + ENV.fetch('GETH_RPC_URL'), + jwt_secret: ENV.fetch('JWT_SECRET'), + retry_config: { tries: 5, base_interval: 0.5, max_interval: 4 } + ) + end + def get_block(block_number, include_txs = false) if block_number.is_a?(String) return query_api( @@ -117,59 +157,108 @@ def query_api(method = nil, params = [], **kwargs) method = kwargs[:method] params = kwargs[:params] end - + unless method raise MethodRequiredError, "Method is required" end - + data = { - id: 1, + id: next_request_id, jsonrpc: "2.0", method: method, params: params } - url = base_url - + # Unified retry logic for both HTTP and IPC Retriable.retriable( - tries: 7, - base_interval: 1, - max_interval: 32, - multiplier: 2, - rand_factor: 0.4, - on: [Net::ReadTimeout, Net::OpenTimeout, HttpError, ApiError], + tries: @retry_config[:tries], + base_interval: @retry_config[:base_interval], + max_interval: @retry_config[:max_interval], + multiplier: @retry_config[:multiplier], + rand_factor: @retry_config[:rand_factor], + on: [Net::ReadTimeout, Net::OpenTimeout, HttpError, ApiError, Errno::EPIPE, EOFError, Errno::ECONNREFUSED], on_retry: ->(exception, try, elapsed_time, next_interval) { Rails.logger.info "Retrying #{method} (attempt #{try}, next delay: #{next_interval.round(2)}s) - #{exception.message}" + # Reset IPC connection on retry if it's broken + if @mode == :ipc && [Errno::EPIPE, EOFError, ApiError].include?(exception.class) + @ipc_mutex.synchronize do + ensure_ipc_connected!(force: true) + end + end } ) do - uri = URI(url) - request = Net::HTTP::Post.new(uri) - request.body = data.to_json - headers.each { |key, value| request[key] = value } - - response = @http.request(uri, request) - - if response.code.to_i != 200 - raise HttpError.new(response.code.to_i, response.message) + if @mode == :ipc + send_ipc_request_simple(data) + else + send_http_request_simple(data) end + end + rescue Errno::EACCES, Errno::EPERM => e + # Permission errors should not be retried - fail immediately with clear message + raise "IPC socket permission denied at #{@ipc_path}: #{e.message}. Check socket permissions (chmod 666) or use HTTP instead." + rescue ApiError => e + # Engine API methods not available on IPC - fail with clear message + if e.message.include?("Method not found") && method.start_with?("engine_") + raise "Engine API method '#{method}' not available on IPC. Use authenticated HTTP endpoint instead." + else + raise + end + end - parsed_response = JSON.parse(response.body, max_nesting: false) - - if parsed_response['error'] - error_message = parsed_response.dig('error', 'message') || 'Unknown API error' - - # Don't retry execution reverted errors as they're deterministic failures - if error_message.include?('execution reverted') - raise ExecutionRevertedError, "API error: #{error_message}" - end - - raise ApiError, "API error: #{error_message}" + def send_ipc_request_simple(data) + @ipc_mutex.synchronize do + ensure_ipc_connected! + + request_body = data.to_json + + ImportProfiler.start("send_ipc_request") + @ipc_socket.write(request_body) + @ipc_socket.write("\n") + @ipc_socket.flush + # Wait for response with timeout to prevent hanging if geth dies + timeout = 10 # seconds + readable = IO.select([@ipc_socket], nil, nil, timeout) + + if readable.nil? + # Timeout occurred - force reconnection on retry + ensure_ipc_connected!(force: true) + raise ApiError, "IPC response timeout after #{timeout} seconds" end - parsed_response['result'] + response_raw = @ipc_socket.gets # single JSON object per line + ImportProfiler.stop("send_ipc_request") + raise ApiError, "empty IPC response" unless response_raw + + parse_response_and_handle_errors(response_raw) end end + def ensure_ipc_connected!(force: false) + if force && @ipc_socket + @ipc_socket.close unless @ipc_socket.closed? + @ipc_socket = nil + end + + return if @ipc_socket && !@ipc_socket.closed? + @ipc_socket = UNIXSocket.new(@ipc_path) + end + + def send_http_request_simple(data) + url = base_url + uri = URI(url) + request = Net::HTTP::Post.new(uri) + request.body = data.to_json + headers.each { |key, value| request[key] = value } + + response = @http.request(uri, request) + + if response.code.to_i != 200 + raise HttpError.new(response.code.to_i, response.message) + end + + parse_response_and_handle_errors(response.body) + end + def call(method, params = []) query_api(method: method, params: params) end @@ -182,11 +271,20 @@ def eth_call(to:, data:, block_number: "latest") end def headers - { + h = { 'Accept' => 'application/json', 'Content-Type' => 'application/json' } + # Add JWT authorization if enabled + h['Authorization'] = "Bearer #{jwt}" if @jwt_enabled && @mode == :http + h + end + + def jwt + return nil unless @jwt_enabled + JWT.encode({ iat: Time.now.to_i }, @jwt_secret_decoded, 'HS256') end + memoize :jwt, ttl: 55 # 55 seconds to refresh before 60 second expiry def get_code(address, block_number = "latest") query_api( @@ -201,4 +299,43 @@ def get_storage_at(address, slot, block_number = "latest") params: [address, slot, block_number] ) end + + private + + def parse_response_and_handle_errors(response_text) + parsed_response = JSON.parse(response_text, max_nesting: false) + + if parsed_response['error'] + error_message = parsed_response.dig('error', 'message') || 'Unknown API error' + + # Don't retry execution reverted errors as they're deterministic failures + if error_message.include?('execution reverted') + raise ExecutionRevertedError, "API error: #{error_message}" + end + + raise ApiError, "API error: #{error_message}" + end + + parsed_response['result'] + end + + def detect_mode(url) + begin + uri = URI.parse(url) + return :http if %w[http https].include?(uri.scheme) + rescue URI::InvalidURIError + # Not a valid URI, might be IPC path + end + + # Check if it's an IPC path + if url.start_with?("ipc://") || url.include?(".ipc") || File.socket?(url.delete_prefix("ipc://")) + :ipc + else + :http + end + end + + def next_request_id + @mutex.synchronize { @request_id += 1 } + end end diff --git a/lib/geth_client.rb b/lib/geth_client.rb deleted file mode 100644 index 0e07de8..0000000 --- a/lib/geth_client.rb +++ /dev/null @@ -1,91 +0,0 @@ -class GethClient - class ClientError < StandardError; end - - attr_reader :node_url, :jwt_secret, :http - - def initialize(node_url) - @node_url = node_url - @jwt_secret = ENV.fetch('JWT_SECRET') - @http = Net::HTTP::Persistent.new(name: "geth_client_#{node_url}") - end - - def call(command, args = []) - payload = { - jsonrpc: "2.0", - method: command, - params: args, - id: 1 - } - - # Benchmark.msr("Call: #{command}") do - send_request(payload) - # end - end - alias :send_command :call - - sig { params(l2_block_number: Integer).returns(Hash) } - def get_l1_attributes(l2_block_number) - if l2_block_number > 0 - l2_block = EthRpcClient.l2.call("eth_getBlockByNumber", ["0x#{l2_block_number.to_s(16)}", true]) - l2_attributes_tx = l2_block['transactions'].first - L1AttributesTxCalldata.decode( - ByteString.from_hex(l2_attributes_tx['input']), - l2_block_number - ) - else - l1_block = EthRpcClient.l1.get_block(SysConfig.l1_genesis_block_number) - eth_block = EthBlock.from_rpc_result(l1_block) - { - timestamp: eth_block.timestamp, - number: eth_block.number, - base_fee: eth_block.base_fee_per_gas, - blob_base_fee: 1, - hash: eth_block.block_hash, - batcher_hash: Hash32.from_bin("\x00".b * 32), - sequence_number: 0, - base_fee_scalar: 0, - blob_base_fee_scalar: 1 - }.with_indifferent_access - end - end - - def send_request(payload) - uri = URI(@node_url) - request = Net::HTTP::Post.new(uri) - request['Content-Type'] = 'application/json' - request['Authorization'] = "Bearer #{jwt}" - request.body = payload.to_json - - response = @http.request(uri, request) - - unless response.code.to_i == 200 - raise ClientError, response - end - - parsed_response = JSON.parse(response.body) - - if parsed_response['error'] - raise ClientError.new(parsed_response['error']) - end - - parsed_response['result'] - end - - def jwt_payload - { - iat: current_time.to_i - } - end - - def current_time - Time.zone.now - end - - def jwt - JWT.encode(jwt_payload, ByteString.from_hex(jwt_secret).to_bin, 'HS256') - end - - def shutdown - @http.shutdown - end -end