From a68e1f04343848f1f733827651987db61b08fe30 Mon Sep 17 00:00:00 2001 From: mctursh Date: Mon, 31 Aug 2026 23:20:29 +0100 Subject: [PATCH] feat(backfill): fetch each block once, not twice, via a block cache --- Cargo.lock | 1 + slate-backfill/src/main.rs | 7 +++- slate-replay/Cargo.toml | 2 ++ slate-replay/src/backfill.rs | 11 ++++-- slate-replay/src/block.rs | 11 +++--- slate-replay/src/source.rs | 67 ++++++++++++++++++++++++++++++++++-- 6 files changed, 89 insertions(+), 10 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 838630d..ec7ef46 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3984,6 +3984,7 @@ dependencies = [ "ed25519-dalek 2.2.0", "redb", "reqwest", + "serde", "serde_json", "sha2 0.10.9", "slate-store", diff --git a/slate-backfill/src/main.rs b/slate-backfill/src/main.rs index f9a0f26..d398e22 100644 --- a/slate-backfill/src/main.rs +++ b/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, /// 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, diff --git a/slate-replay/Cargo.toml b/slate-replay/Cargo.toml index 7a7d0d0..2a9348b 100644 --- a/slate-replay/Cargo.toml +++ b/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" diff --git a/slate-replay/src/backfill.rs b/slate-replay/src/backfill.rs index 7ec441b..2ff6fd3 100644 --- a/slate-replay/src/backfill.rs +++ b/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, + block_cache: Option, from: u64, to: u64, program: &Pubkey, @@ -42,6 +43,10 @@ pub async fn backfill( verify_end: Option>, ) -> Result { let chunk_slots = chunk_slots.max(1); + let source: Arc = 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, diff --git a/slate-replay/src/block.rs b/slate-replay/src/block.rs index 2a0fabc..3ab2f01 100644 --- a/slate-replay/src/block.rs +++ b/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, 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, pub readonly: Vec, diff --git a/slate-replay/src/source.rs b/slate-replay/src/source.rs index 7869889..0ef31a9 100644 --- a/slate-replay/src/source.rs +++ b/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 = TableDefinition::new("blocks"); + +pub struct CachingBlockSource { + inner: Arc, + db: Database, +} + +impl CachingBlockSource { + pub fn new(inner: Arc, cache_path: PathBuf) -> Result { + 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) -> 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> { + self.inner.confirmed_slots(from, to) + } + + fn fetch(&self, slots: &[u64]) -> Result> { + let mut hits: HashMap = HashMap::new(); + let mut misses: Vec = 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,