Skip to content

Make import faster #105

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
RogerPodacter merged 1 commit into from
Sep 18, 2025
Merged

Make import faster #105

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
156 changes: 83 additions & 73 deletions app/services/eth_block_importer.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,14 +7,13 @@ class BlockNotReadyToImportError < StandardError; end
# Raised when a re-org is detected (parent hash mismatch)
class ReorgDetectedError < StandardError; end

attr_accessor :l1_rpc_results, :ethscriptions_block_cache, :ethereum_client, :eth_block_cache, :geth_driver
attr_accessor :ethscriptions_block_cache, :ethereum_client, :eth_block_cache, :geth_driver

def initialize
@l1_rpc_results = {}
@ethscriptions_block_cache = {}
@eth_block_cache = {}

@ethereum_client ||= EthRpcClient.new(ENV.fetch('L1_RPC_URL'))
@ethereum_client ||= EthRpcClient.l1

@geth_driver = GethDriver

Expand All @@ -24,7 +23,7 @@ def initialize

# Create a shared thread pool for validation
if @validation_enabled
@validation_threads = ENV.fetch('VALIDATION_THREADS', '50').to_i
@validation_threads = ENV.fetch('VALIDATION_THREADS', '10').to_i
@validation_executor = Concurrent::ThreadPoolExecutor.new(
min_threads: 1,
max_threads: @validation_threads,
Expand All @@ -34,12 +33,29 @@ def initialize
logger.info "Created validation thread pool with #{@validation_threads} threads"
end

# L1 prefetcher for blocks/receipts/API data
@prefetcher = L1RpcPrefetcher.new(
ethereum_client: @ethereum_client,
ahead: ENV.fetch('L1_PREFETCH_FORWARD', Rails.env.test? ? 5 : 20).to_i,
threads: ENV.fetch('L1_PREFETCH_THREADS', Rails.env.test? ? 2 : 2).to_i

Copilot AI Sep 18, 2025

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Line 40 uses the same value (2) for both test and non-test environments. This should likely be different values, such as a higher thread count for production.

Suggested change
threads: ENV.fetch('L1_PREFETCH_THREADS', Rails.env.test? ? 2 : 2).to_i
threads: ENV.fetch('L1_PREFETCH_THREADS', Rails.env.test? ? 2 : 10).to_i

Copilot uses AI. Check for mistakes.
)
@prefetched_api_data = {}

logger.info "EthBlockImporter initialized - Validation: #{@validation_enabled ? 'ENABLED' : 'disabled'}"

MemeryExtensions.clear_all_caches!

set_eth_block_starting_points
populate_ethscriptions_block_cache

unless Rails.env.test?
max_block = current_max_eth_block_number
if max_block && max_block > 0
ImportProfiler.start('prefetch_warmup')
@prefetcher.ensure_prefetched(max_block + 1)
ImportProfiler.stop('prefetch_warmup')
end
end
end

def current_max_ethscriptions_block_number
Expand All @@ -60,10 +76,14 @@ def populate_ethscriptions_block_cache

while epochs_found < 64 && current_block_number >= 0
hex_block_number = "0x#{current_block_number.to_s(16)}"
ImportProfiler.start("l2_block_fetch")
block_data = geth_driver.client.call("eth_getBlockByNumber", [hex_block_number, false])
ImportProfiler.stop("l2_block_fetch")
current_block = EthscriptionsBlock.from_rpc_result(block_data)


ImportProfiler.start("l1_attributes_fetch")
l1_attributes = GethDriver.client.get_l1_attributes(current_block.number)
ImportProfiler.stop("l1_attributes_fetch")
current_block.assign_l1_attributes(l1_attributes)

ethscriptions_block_cache[current_block.number] = current_block
Expand Down Expand Up @@ -183,50 +203,15 @@ def import_blocks_until_done
raise BlockNotReadyToImportError.new("Block not ready")
end

populate_l1_rpc_results(block_numbers)

import_blocks(block_numbers)
rescue BlockNotReadyToImportError => e
puts "#{e.message}. Stopping import."
ImportProfiler.report if ImportProfiler.enabled?
break
end
end
end

def populate_l1_rpc_results(block_numbers)
next_start_block = block_numbers.last + 1
next_block_numbers = (next_start_block...(next_start_block + import_batch_size)).to_a

blocks_to_import = block_numbers

if blocks_behind > 1
blocks_to_import += next_block_numbers.select do |num|
num <= current_block_number
end
end

blocks_to_import -= l1_rpc_results.keys

l1_rpc_results.reverse_merge!(get_blocks_promises(blocks_to_import))
end

def get_blocks_promises(block_numbers)
block_numbers.map do |block_number|
block_promise = Concurrent::Promise.execute do
ethereum_client.get_block(block_number, true)
end

receipt_promise = Concurrent::Promise.execute do
ethereum_client.get_transaction_receipts(block_number)
end

[block_number, {
block: block_promise,
receipts: receipt_promise
}.with_indifferent_access]
end.to_h
end

def fetch_block_from_cache(block_number)
block_number = [block_number, 0].max

Expand All @@ -246,6 +231,12 @@ def prune_caches
ethscriptions_block_cache.delete_if do |_, ethscriptions_block|
ethscriptions_block.eth_block_number < oldest_eth_block_to_keep
end

# Clean up prefetcher and API data caches
if oldest_eth_block_to_keep
@prefetcher.clear_older_than(oldest_eth_block_to_keep)
@prefetched_api_data.delete_if { |block_num, _| block_num < oldest_eth_block_to_keep }
end
end

def current_ethscriptions_block(type)
Expand Down Expand Up @@ -289,48 +280,50 @@ def current_facet_finalized_block
end

def import_blocks(block_numbers)
ImportProfiler.start("import_blocks_total")

# Initialize seen creates for this batch (needed by prefetcher threads)
EthscriptionTransaction.reset_seen_creates!

logger.info "Block Importer: importing blocks #{block_numbers.join(', ')}"
start = Time.current

block_responses = l1_rpc_results.select do |block_number, _|
block_numbers.include?(block_number)
end.to_h.transform_values do |hsh|
hsh.transform_values(&:value!)
end

l1_rpc_results.reject! { |block_number, _| block_responses.key?(block_number) }


ImportProfiler.start("prefetch_ensure")
@prefetcher.ensure_prefetched(block_numbers.first)
ImportProfiler.stop("prefetch_ensure")

eth_blocks = []
ethscriptions_blocks = []
res = []

block_numbers.each.with_index do |block_number, index|
block_response = block_responses[block_number]

block_result = block_response['block']
receipt_result = block_response['receipts']

block_numbers.each do |block_number|
ImportProfiler.start("prefetch_fetch")
response = @prefetcher.fetch(block_number)
ImportProfiler.stop("prefetch_fetch")

eth_block = response[:eth_block]
ethscriptions_block = response[:ethscriptions_block]
ethscription_txs = response[:ethscription_txs]

ethscription_txs.each { |tx| tx.ethscriptions_block = ethscriptions_block }

if @validation_enabled && response[:api_data]
@prefetched_api_data[block_number] = response[:api_data]
end

parent_eth_block = eth_block_cache[block_number - 1]
if parent_eth_block && parent_eth_block.block_hash != Hash32.from_hex(block_result['parentHash'])

if parent_eth_block && parent_eth_block.block_hash != eth_block.parent_hash
logger.info "Reorg detected at block #{block_number}"
raise ReorgDetectedError
end

eth_block = EthBlock.from_rpc_result(block_result)

ethscriptions_block = EthscriptionsBlock.from_eth_block(eth_block)

ethscription_txs = EthTransaction.ethscription_txs_from_rpc_results(block_result, receipt_result, ethscriptions_block)

ethscription_txs.each do |ethscription_tx|
ethscription_tx.ethscriptions_block = ethscriptions_block
end

ImportProfiler.start("propose_ethscriptions_block")
imported_ethscriptions_blocks = propose_ethscriptions_block(
ethscriptions_block: ethscriptions_block,
ethscription_txs: ethscription_txs
)
ImportProfiler.stop("propose_ethscriptions_block")

logger.debug "Block #{block_number}: Found #{ethscription_txs.length} ethscription txs, created #{imported_ethscriptions_blocks.length} L2 blocks"

Expand Down Expand Up @@ -368,13 +361,22 @@ def import_blocks(block_numbers)
logger.info "Total gas used: #{total_gas_millions} million (avg: #{average_gas_per_block_millions} million / block)"
logger.info "Gas per second: #{gas_per_second_millions} million / s"

if block_numbers.length >= 10
stats = @prefetcher.stats
logger.info "Prefetcher stats: promises=#{stats[:promises_total]} " \
"(fulfilled=#{stats[:promises_fulfilled]}, pending=#{stats[:promises_pending]}, " \
"threads_active=#{stats[:threads_active]}, queued=#{stats[:threads_queued]})"
end

# Validate imported blocks if enabled
if @validation_enabled
if ethscriptions_blocks.any?
ImportProfiler.start("validation_wall_clock")
start_validation = Time.current
logger.info "Starting parallel validation for #{block_numbers.length} blocks with #{ethscriptions_blocks.length} L2 blocks"
validate_imported_blocks(block_numbers, ethscriptions_blocks)
validation_time = Time.current - start_validation
ImportProfiler.stop("validation_wall_clock")
logger.info "Validation completed in #{validation_time.round(2)}s (#{(block_numbers.length / validation_time).round(2)} blocks/s)"
else
logger.info "Validation enabled but no ethscriptions blocks to validate"
Expand All @@ -383,14 +385,19 @@ def import_blocks(block_numbers)
logger.warn "Validation requested but not enabled in importer initialization"
end

ImportProfiler.stop("import_blocks_total")

# Generate profiling report after each batch
if ImportProfiler.enabled?
ImportProfiler.report
ImportProfiler.reset
end

[ethscriptions_blocks, eth_blocks]
end

def import_next_block
block_number = next_block_to_import

populate_l1_rpc_results([block_number])

import_blocks([block_number])
end

Expand Down Expand Up @@ -450,7 +457,8 @@ def validate_imported_blocks(l1_block_numbers, l2_blocks)
else
begin
validator = BlockValidator.new
result = validator.validate_l1_block(l1_block_num, l2_hashes)
api_data = @prefetched_api_data.delete(l1_block_num)
result = validator.validate_l1_block(l1_block_num, l2_hashes, expected_override: api_data)

if result.stats[:api_unavailable]
{
Expand Down Expand Up @@ -544,6 +552,8 @@ def validation_summary
end

def shutdown
@prefetcher&.shutdown

if @validation_executor
logger.info "Shutting down validation thread pool..."
@validation_executor.shutdown
Expand Down
14 changes: 13 additions & 1 deletion app/services/geth_driver.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,12 +22,14 @@ def propose_block(
finalized_block:
)
# Create filler blocks if necessary and update head_block
ImportProfiler.start("create_filler_blocks")
filler_blocks = create_filler_blocks(
head_block: head_block,
new_ethscriptions_block: new_ethscriptions_block,
safe_block: safe_block,
finalized_block: finalized_block
)
ImportProfiler.stop("create_filler_blocks")

head_block = filler_blocks.last || head_block

Expand Down Expand Up @@ -74,7 +76,10 @@ def propose_block(
payload_attributes = ByteString.deep_hexify(payload_attributes)
fork_choice_state = ByteString.deep_hexify(fork_choice_state)

ImportProfiler.start("engine_forkchoiceUpdated_1")
fork_choice_response = client.call("engine_forkchoiceUpdatedV#{version}", [fork_choice_state, payload_attributes])
ImportProfiler.stop("engine_forkchoiceUpdated_1")

if fork_choice_response['error']
raise "Fork choice update failed: #{fork_choice_response['error']}"
end
Expand All @@ -84,7 +89,10 @@ def propose_block(
raise "Fork choice update did not return a payload ID"
end

ImportProfiler.start("engine_getPayload")
get_payload_response = client.call("engine_getPayloadV#{version}", [payload_id])
ImportProfiler.stop("engine_getPayload")

if get_payload_response['error']
raise "Get payload failed: #{get_payload_response['error']}"
end
Expand All @@ -104,8 +112,10 @@ def propose_block(

new_payload_request = ByteString.deep_hexify(new_payload_request)

ImportProfiler.start("engine_newPayload")
new_payload_response = client.call("engine_newPayloadV#{version}", new_payload_request)

ImportProfiler.stop("engine_newPayload")

status = new_payload_response['status']
unless status == 'VALID'
raise "New payload was not valid: #{status}"
Expand All @@ -126,7 +136,9 @@ def propose_block(

fork_choice_state = ByteString.deep_hexify(fork_choice_state)

ImportProfiler.start("engine_forkchoiceUpdated_2")
fork_choice_response = client.call("engine_forkchoiceUpdatedV#{version}", [fork_choice_state, nil])
ImportProfiler.stop("engine_forkchoiceUpdated_2")

status = fork_choice_response['payloadStatus']['status']
unless status == 'VALID'
Expand Down
13 changes: 13 additions & 0 deletions config/derive_ethscriptions_blocks.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,6 +114,19 @@ module Clockwork
every(import_interval.seconds, 'import_ethscriptions_blocks') do
importer = EthBlockImporter.new

# Handle Ctrl+C gracefully
Signal.trap('INT') do
puts "\n[#{Time.now}] Received INT signal, shutting down..."
importer.shutdown
exit 130
end

Signal.trap('TERM') do
puts "\n[#{Time.now}] Received TERM signal, shutting down..."
importer.shutdown
exit 143
end

# Track statistics
total_blocks_imported = 0
total_ethscriptions = 0
Expand Down
Loading