Improve prefetcher

RogerPodacter · RogerPodacter · commit b260fa97ca01 · 2025-10-13T13:26:41.000-04:00
diff --git a/app/services/eth_block_importer.rb b/app/services/eth_block_importer.rb
@@ -329,9 +329,14 @@ def import_single_block(block_number)
     start = Time.current
 
     # Fetch block data from prefetcher
-    ImportProfiler.start("prefetch_fetch")
-    response = @prefetcher.fetch(block_number)
-    ImportProfiler.stop("prefetch_fetch")
+    begin
+      ImportProfiler.start('prefetcher_fetch')
+      response = prefetcher.fetch(block_number)
+    rescue L1RpcPrefetcher::BlockFetchError => e
+      raise BlockNotReadyToImportError.new(e.message)
+    ensure
+      ImportProfiler.stop('prefetcher_fetch')
+    end
 
     # Handle cancellation, fetch failure, or block not ready
     if response.nil?
diff --git a/config/derive_ethscriptions_blocks.rb b/config/derive_ethscriptions_blocks.rb
@@ -116,95 +116,86 @@ 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
     start_time = Time.now
 
-    loop do
-      begin
-        initial_block = importer.current_max_eth_block_number
+    begin
+      loop do
+        begin
+          initial_block = importer.current_max_eth_block_number
 
-        # Import blocks
-        importer.import_blocks_until_done
+          # Import blocks
+          importer.import_blocks_until_done
 
-        final_block = importer.current_max_eth_block_number
-        blocks_imported = final_block - initial_block
+          final_block = importer.current_max_eth_block_number
+          blocks_imported = final_block - initial_block
 
-        if blocks_imported > 0
-          total_blocks_imported += blocks_imported
+          if blocks_imported > 0
+            total_blocks_imported += blocks_imported
 
-          puts "[#{Time.now}] Imported #{blocks_imported} blocks (#{initial_block + 1} to #{final_block})"
+            puts "[#{Time.now}] Imported #{blocks_imported} blocks (#{initial_block + 1} to #{final_block})"
 
-          # Show validation summary if enabled
-          if ENV.fetch('VALIDATION_ENABLED').casecmp?('true')
-            puts importer.validation_summary
-          end
-        else
-          # We're caught up
-          elapsed = (Time.now - start_time).round(2)
+            # Show validation summary if enabled
+            if ENV.fetch('VALIDATION_ENABLED').casecmp?('true')
+              puts importer.validation_summary
+            end
+          else
+            # We're caught up
+            elapsed = (Time.now - start_time).round(2)
+
+            if total_blocks_imported > 0
+              puts "[#{Time.now}] Session summary: Imported #{total_blocks_imported} blocks in #{elapsed}s"
 
-          if total_blocks_imported > 0
-            puts "[#{Time.now}] Session summary: Imported #{total_blocks_imported} blocks in #{elapsed}s"
+              # Reset counters
+              total_blocks_imported = 0
+              start_time = Time.now
+            end
 
-            # Reset counters
-            total_blocks_imported = 0
-            start_time = Time.now
+            puts "[#{Time.now}] Caught up at block #{final_block}. Waiting #{import_interval}s..."
           end
 
-          puts "[#{Time.now}] Caught up at block #{final_block}. Waiting #{import_interval}s..."
+        rescue EthBlockImporter::BlockNotReadyToImportError => e
+          # This is normal when caught up
+          current = importer.current_max_eth_block_number
+          puts "[#{Time.now}] Waiting for new blocks (current: #{current})..."
+
+        rescue EthBlockImporter::ReorgDetectedError => e
+          Rails.logger.warn "[#{Time.now}] ⚠️  Reorg detected! Reinitializing importer..."
+          puts "[#{Time.now}] ⚠️  Reorg detected at block #{importer.current_max_eth_block_number}"
+
+          # Reinitialize importer to handle reorg
+          importer.shutdown
+          importer = EthBlockImporter.new
+          puts "[#{Time.now}] Importer reinitialized. Continuing from block #{importer.current_max_eth_block_number}"
+
+        rescue EthBlockImporter::ValidationFailureError => e
+          Rails.logger.fatal "[#{Time.now}] 🛑 VALIDATION FAILURE: #{e.message}"
+          puts "[#{Time.now}] 🛑 VALIDATION FAILURE - System stopping for investigation"
+          puts "[#{Time.now}] Fix the validation issue and restart manually"
+          exit 1
+
+        rescue EthBlockImporter::ValidationStalledError => e
+          # Validation is behind - wait longer and keep trying
+          Rails.logger.info "[#{Time.now}] ⏸️  VALIDATION BEHIND: #{e.message}"
+          puts "[#{Time.now}] ⏸️  Validation is behind - waiting #{import_interval * 2}s for validation to catch up..."
+          sleep import_interval * 2  # Wait longer when validation is behind
+          # Don't exit - continue the loop to retry
+
+        rescue => e
+          Rails.logger.error "Import error: #{e.class} - #{e.message}"
+          Rails.logger.error e.backtrace.first(20).join("\n")
+
+          puts "[#{Time.now}] ❌ Error: #{e.message}"
+
+          # For other errors, wait and retry
+          puts "[#{Time.now}] Retrying in #{import_interval}s..."
         end
 
-      rescue EthBlockImporter::BlockNotReadyToImportError => e
-        # This is normal when caught up
-        current = importer.current_max_eth_block_number
-        puts "[#{Time.now}] Waiting for new blocks (current: #{current})..."
-
-      rescue EthBlockImporter::ReorgDetectedError => e
-        Rails.logger.warn "[#{Time.now}] ⚠️  Reorg detected! Reinitializing importer..."
-        puts "[#{Time.now}] ⚠️  Reorg detected at block #{importer.current_max_eth_block_number}"
-
-        # Reinitialize importer to handle reorg
-        importer = EthBlockImporter.new
-        puts "[#{Time.now}] Importer reinitialized. Continuing from block #{importer.current_max_eth_block_number}"
-
-      rescue EthBlockImporter::ValidationFailureError => e
-        Rails.logger.fatal "[#{Time.now}] 🛑 VALIDATION FAILURE: #{e.message}"
-        puts "[#{Time.now}] 🛑 VALIDATION FAILURE - System stopping for investigation"
-        puts "[#{Time.now}] Fix the validation issue and restart manually"
-        exit 1
-
-      rescue EthBlockImporter::ValidationStalledError => e
-        # Validation is behind - wait longer and keep trying
-        Rails.logger.info "[#{Time.now}] ⏸️  VALIDATION BEHIND: #{e.message}"
-        puts "[#{Time.now}] ⏸️  Validation is behind - waiting #{import_interval * 2}s for validation to catch up..."
-        sleep import_interval * 2  # Wait longer when validation is behind
-        # Don't exit - continue the loop to retry
-
-      rescue => e
-        Rails.logger.error "Import error: #{e.class} - #{e.message}"
-        Rails.logger.error e.backtrace.first(20).join("\n")
-
-        puts "[#{Time.now}] ❌ Error: #{e.message}"
-
-        # For other errors, wait and retry
-        puts "[#{Time.now}] Retrying in #{import_interval}s..."
+        sleep import_interval
       end
-
-      sleep import_interval
+    ensure
+      importer&.shutdown
     end
   end
 end
diff --git a/lib/l1_rpc_prefetcher.rb b/lib/l1_rpc_prefetcher.rb
@@ -3,8 +3,9 @@
 
 class L1RpcPrefetcher
   include Memery
+  class BlockFetchError < StandardError; end
   def initialize(ethereum_client:,
-                 ahead: ENV.fetch('L1_PREFETCH_FORWARD', Rails.env.test? ? 5 : 200).to_i,
+                 ahead: ENV.fetch('L1_PREFETCH_FORWARD', Rails.env.test? ? 5 : 20).to_i,
                  threads: ENV.fetch('L1_PREFETCH_THREADS', 2).to_i)
     @eth = ethereum_client
     @ahead = ahead
@@ -13,13 +14,18 @@ def initialize(ethereum_client:,
     # Thread-safe collections and pool
     @pool = Concurrent::FixedThreadPool.new(threads)
     @promises = Concurrent::Map.new
+    @last_chain_tip = current_l1_block_number
 
     Rails.logger.info "L1RpcPrefetcher initialized with #{threads} threads"
   end
 
   def ensure_prefetched(from_block)
-    # Check current chain tip first
-    latest = get_latest_block_number
+    distance_from_last_tip = @last_chain_tip - from_block
+    latest = if distance_from_last_tip > 10
+               cached_l1_block_number
+             else
+               current_l1_block_number
+             end
 
     # Don't prefetch beyond chain tip
     to_block = [from_block + @ahead, latest].min
@@ -45,16 +51,18 @@ def fetch(block_number)
 
     Rails.logger.debug "Fetching block #{block_number}, promise state: #{promise.state}"
 
-    begin
-      result = promise.value!(timeout)
-      Rails.logger.debug "Got result for block #{block_number}"
-
-      result
-    rescue Concurrent::TimeoutError => e
-      Rails.logger.error "Timeout fetching block #{block_number} after #{timeout}s"
+    result = promise.value!(timeout)
+    
+    if result.nil? || result == :not_ready_sentinel
       @promises.delete(block_number)
-      raise
+      message = result.nil? ?
+        "Block #{block_number} fetch timed out after #{timeout}s" :
+        "Block #{block_number} not yet available on L1"
+      raise BlockFetchError.new(message)
     end
+
+    Rails.logger.debug "Got result for block #{block_number}"
+    result
   end
 
   def clear_older_than(min_keep)
@@ -96,21 +104,31 @@ def stats
 
   def shutdown
     @pool.shutdown
-    if @pool.wait_for_termination(30)
-      Rails.logger.info "L1 RPC Prefetcher thread pool shut down successfully"
-    else
-      Rails.logger.warn "L1 RPC Prefetcher shutdown timed out, forcing kill"
-      @pool.kill
+    terminated = @pool.wait_for_termination(3)
+    @pool.kill unless terminated
+    @promises.each_pair do |_, promise|
+      begin
+        if promise.pending? && promise.respond_to?(:cancel)
+          promise.cancel
+        end
+      rescue StandardError => e
+        Rails.logger.warn "Failed cancelling promise during shutdown: #{e.message}"
+      end
     end
+    @promises.clear
+    Rails.logger.info(
+      terminated ?
+        'L1 RPC Prefetcher thread pool shut down successfully' :
+        'L1 RPC Prefetcher shutdown timed out after 3s, pool killed'
+    )
+    terminated
+  rescue StandardError => e
+    Rails.logger.error("Error during L1RpcPrefetcher shutdown: #{e.message}\n#{e.backtrace.join("\n")}")
+    false
   end
 
   private
 
-  def get_latest_block_number
-    @eth.get_block_number
-  end
-  memoize :get_latest_block_number, ttl: 12.seconds
-
   def enqueue_single(block_number)
     @promises.compute_if_absent(block_number) do
       Rails.logger.debug "Creating promise for block #{block_number}"
@@ -132,13 +150,12 @@ def fetch_job(block_number)
     client = @eth
 
     Retriable.retriable(tries: 3, base_interval: 1, max_interval: 4) do
-
       block = client.get_block(block_number, true)
 
       # Handle case where block doesn't exist yet (normal when caught up)
       if block.nil?
         Rails.logger.debug "Block #{block_number} not yet available on L1"
-        return { error: :not_ready, block_number: block_number }
+        return :not_ready_sentinel
       end
 
       receipts = client.get_transaction_receipts(block_number)
@@ -154,4 +171,13 @@ def fetch_job(block_number)
       }
     end
   end
-end
+
+  def current_l1_block_number
+    @last_chain_tip = @eth.get_block_number
+  end
+
+  def cached_l1_block_number
+    current_l1_block_number
+  end
+  memoize :cached_l1_block_number, ttl: 12.seconds
+end
