feat(backfill): fetch each block once, not twice, via a block cache

slate-backfill/src/main.rs

@@ -1,4 +1,4 @@
-use std::{fs::File, io::Read, str::FromStr, sync::Arc};
+use std::{fs::File, io::Read, path::PathBuf, str::FromStr, sync::Arc};
 
 use anyhow::Context;
 use clap::Parser;
@@ -46,6 +46,10 @@ struct Args {
     /// Path for the disk store's redb file.
     #[arg(long, default_value = "slate-accounts.redb")]
     store_path: String,
+    /// Block cache path (redb). Point runs of the same cluster at one file to skip
+    /// re-fetching on retries. Omit to disable.
+    #[arg(long)]
+    block_cache: Option<String>,
     /// Replay this many slots per chunk before flushing writes to ClickHouse and
     /// clearing the log. Bounds write-log RAM over long ranges.
     #[arg(long, default_value_t = 2000)]
@@ -151,6 +155,7 @@ fn main() -> anyhow::Result<()> {
             snapshot,
             args.from,
             source,
+            args.block_cache.as_ref().map(PathBuf::from),
             args.from,
             args.to,
             &program,

slate-replay/Cargo.toml

@@ -73,6 +73,8 @@ slate-store = { path = "../slate-store" }
 # so it can't panic inside (or stall) the runtime that persists to ClickHouse.
 tokio = { workspace = true }
 
+serde = { workspace = true, features = ["derive"] }
+
 [dev-dependencies]
 solana-message = "3"
 solana-instruction = "3"

slate-replay/src/backfill.rs

@@ -1,4 +1,4 @@
-use std::{collections::HashSet, io::Read, sync::Arc};
+use std::{collections::HashSet, io::Read, path::PathBuf, sync::Arc};
 
 use anyhow::Result;
 use slate_store::ClickHouseClient;
@@ -10,7 +10,7 @@ use crate::{
     RangeReplay, ReplayBank, Replayer, WriteRecord,
     block::{self, Block},
     boundary, build_feature_set, compat, persist, register_builtins, snapshot,
-    source::BlockSource,
+    source::{BlockSource, CachingBlockSource},
     store::{AccountStore, DiskStore, MemStore},
 };
 
@@ -32,6 +32,7 @@ pub async fn backfill(
     snapshot: impl Read,
     s_snap: u64,
     source: Arc<dyn BlockSource>,
+    block_cache: Option<PathBuf>,
     from: u64,
     to: u64,
     program: &Pubkey,
@@ -42,6 +43,10 @@ pub async fn backfill(
     verify_end: Option<Box<dyn Read>>,
 ) -> Result<BackfillReport> {
     let chunk_slots = chunk_slots.max(1);
+    let source: Arc<dyn BlockSource> = match block_cache {
+        None => source,
+        Some(path) => Arc::new(CachingBlockSource::new(source, path)?),
+    };
     // Slots are just u64s (bounded), so hold them all; the blocks themselves never all fit.
     let slots = {
         let src = Arc::clone(&source);
@@ -266,6 +271,7 @@ mod tests {
             SNAPSHOT,
             s_snap,
             source,
+            None,
             s_snap,
             s + 1,
             &system,
@@ -343,6 +349,7 @@ mod tests {
             SNAPSHOT,
             200,
             source,
+            None,
             200,
             200,
             &system,

slate-replay/src/block.rs

@@ -3,6 +3,7 @@ use std::collections::HashSet;
 use agave_reserved_account_keys::ReservedAccountKeys;
 use anyhow::{Context, Result};
 use base64::Engine;
+use serde::{Deserialize, Serialize};
 use solana_hash::Hash;
 use solana_message::{
     AddressLoader,
@@ -15,7 +16,7 @@ use solana_transaction::{
 };
 use solana_transaction_error::AddressLoaderError;
 
-#[derive(Clone)]
+#[derive(Serialize, Deserialize, Clone)]
 pub struct Block {
     pub slot: u64,
     pub parent_slot: u64,
@@ -28,13 +29,13 @@ pub struct Block {
     pub fee_reward: Option<(Pubkey, u64)>,
 }
 
-#[derive(Clone)]
+#[derive(Serialize, Deserialize, Clone)]
 pub struct BlockTx {
     pub transaction: VersionedTransaction,
     pub meta: TxMeta,
 }
 
-#[derive(Clone)]
+#[derive(Serialize, Deserialize, Clone)]
 pub struct TxMeta {
     pub err: Option<String>,
     pub fee: u64,
@@ -51,14 +52,14 @@ impl TxMeta {
     }
 }
 
-#[derive(Clone)]
+#[derive(Serialize, Deserialize, Clone)]
 pub struct TokenBalance {
     pub account_index: u8,
     pub mint: Pubkey,
     pub amount: u64,
 }
 
-#[derive(Default, Clone)]
+#[derive(Serialize, Deserialize, Default, Clone)]
 pub struct LoadedAddresses {
     pub writable: Vec<Pubkey>,
     pub readonly: Vec<Pubkey>,

slate-replay/src/source.rs

@@ -1,18 +1,21 @@
 use std::{
+    collections::HashMap,
+    path::PathBuf,
     sync::{
-        Mutex,
+        Arc, Mutex,
         atomic::{AtomicUsize, Ordering},
     },
     time::Duration,
 };
 
 use anyhow::Result;
+use redb::{Database, Durability, TableDefinition};
 use reqwest::blocking::Client;
 
 use crate::block::{Block, fetch_block_opt, fetch_confirmed_slots};
 
 // Big retry budget: one unrecovered miss aborts a whole pass, and Old Faithful flakes transiently (CDN range-fetch), so it has to outlast a transient window, not just a blip.
-const MAX_RETRIES: usize = 40;
+const MAX_RETRIES: usize = 80;
 
 // Send + Sync so a shared source can be handed to a blocking fetch task while the async loop persists the previous chunk.
 pub trait BlockSource: Send + Sync {
@@ -29,6 +32,23 @@ pub struct RpcBlockSource {
     concurrency: usize,
 }
 
+const BLOCKS: TableDefinition<u64, &[u8]> = TableDefinition::new("blocks");
+
+pub struct CachingBlockSource {
+    inner: Arc<dyn BlockSource>,
+    db: Database,
+}
+
+impl CachingBlockSource {
+    pub fn new(inner: Arc<dyn BlockSource>, cache_path: PathBuf) -> Result<Self> {
+        let db = Database::create(&cache_path)?;
+        let txn = db.begin_write()?;
+        txn.open_table(BLOCKS)?;
+        txn.commit()?;
+        Ok(Self { inner, db })
+    }
+}
+
 impl RpcBlockSource {
     pub fn new(rpc_url: impl Into<String>) -> Self {
         let client = Client::builder()
@@ -126,6 +146,49 @@ impl BlockSource for RpcBlockSource {
     }
 }
 
+impl BlockSource for CachingBlockSource {
+    fn confirmed_slots(&self, from: u64, to: u64) -> Result<Vec<u64>> {
+        self.inner.confirmed_slots(from, to)
+    }
+
+    fn fetch(&self, slots: &[u64]) -> Result<Vec<Block>> {
+        let mut hits: HashMap<u64, Block> = HashMap::new();
+        let mut misses: Vec<u64> = Vec::new();
+
+        {
+            let txn = self.db.begin_read()?;
+            let table = txn.open_table(BLOCKS)?;
+            for &slot in slots {
+                match table.get(slot)? {
+                    Some(g) => {
+                        hits.insert(slot, bincode::deserialize(g.value())?);
+                    }
+                    None => misses.push(slot),
+                }
+            }
+        }
+
+        let fresh = self.inner.fetch(&misses)?;
+        if !fresh.is_empty() {
+            let mut txn = self.db.begin_write()?;
+            txn.set_durability(Durability::None);
+            {
+                let mut table = txn.open_table(BLOCKS)?;
+                for b in &fresh {
+                    table.insert(b.slot, bincode::serialize(&b)?.as_slice())?;
+                }
+            }
+            txn.commit()?;
+        }
+
+        for b in fresh {
+            hits.insert(b.slot, b);
+        }
+
+        Ok(slots.iter().filter_map(|s| hits.remove(s)).collect())
+    }
+}
+
 // In-memory BlockSource for tests and small pre-built ranges; the replay path treats it like a remote source.
 pub struct VecBlockSource {
     blocks: Vec<Block>,