From c035f54395532dc5c12ceebcbebbb8c970089f27 Mon Sep 17 00:00:00 2001 From: mctursh Date: Fri, 4 Sep 2026 21:57:28 +0100 Subject: [PATCH 1/2] feat(backfill): --resume a halted run from the store's last checkpoint A long replay that halted (bad block, vote mismatch, crash) had to start over from the snapshot. Now the disk store periodically commits a checkpoint and --resume picks up from it, no snapshot needed. A checkpoint is the buffered accounts plus the bank-hash roll state (2048 LE lattice bytes ++ the 32-byte hash, same layout as the snapshot trailer) written in one durable redb commit, so a crash can't land the accounts without the checkpoint that describes them. What makes it sound is checkpoint_mode: once seeding is done, put() stops auto-flushing and flush() no-ops, leaving checkpoint_flush as the only committer. The redb therefore never holds writes past the last checkpoint, so resuming from it can't replay onto state that already moved. The checkpoint is written even when a chunk buffered nothing, so a no-write chunk still advances the slot. --- slate-backfill/src/main.rs | 73 ++++++++---- slate-replay/src/backfill.rs | 223 ++++++++++++++++++++++++++++------- slate-replay/src/bankhash.rs | 39 ++++++ slate-replay/src/lib.rs | 10 ++ slate-replay/src/store.rs | 90 +++++++++++++- 5 files changed, 366 insertions(+), 69 deletions(-) diff --git a/slate-backfill/src/main.rs b/slate-backfill/src/main.rs index d398e22..109410d 100644 --- a/slate-backfill/src/main.rs +++ b/slate-backfill/src/main.rs @@ -15,8 +15,8 @@ use solana_pubkey::Pubkey; #[derive(Parser)] #[command(about = "Reconstruct a program's historical account state by replaying a slot range")] struct Args { - /// Path to the full snapshot the range starts from (omit with --dry-run). - #[arg(required_unless_present = "dry_run")] + /// Path to the full snapshot the range starts from (omit with --dry-run or --resume). + #[arg(required_unless_present_any = ["dry_run", "resume"])] snapshot: Option, /// Slot the snapshot was taken at. Replay covers (from, to]. #[arg(long)] @@ -37,6 +37,11 @@ struct Args { /// report what it looks like. A preflight to run before downloading a snapshot. #[arg(long)] dry_run: bool, + /// Resume a crashed or halted run from the store's last checkpoint instead of seeding + /// fresh. Needs --store disk at the same --store-path (and the same --block-cache to + /// skip re-fetching). No snapshot needed. + #[arg(long)] + resume: bool, /// Account store: `memory` (RAM, small ranges) or `disk` (redb, large ranges). #[arg(long, default_value = "memory")] store: String, @@ -91,38 +96,57 @@ fn main() -> anyhow::Result<()> { return dry_run_report(&blocks); } - // Validate program, snapshot, config, and ClickHouse before the expensive fetch. + // Validate program, config, and ClickHouse before the expensive fetch. let program_str = args .program .as_ref() .context("--program is required for a real run")?; let program = Pubkey::from_str(program_str) .with_context(|| format!("invalid program pubkey {program_str}"))?; - let snapshot_path = args - .snapshot - .as_ref() - .context(" is required for a real run")?; - check_snapshot_file(snapshot_path)?; let cfg = Config::load(&args.config)?; check_clickhouse(&cfg.clickhouse.url)?; - println!("preflight ok: RPC, program, snapshot, config, and ClickHouse all check out"); - // Bootstrap the bank-hash roll from the manifest (lattice + bank hash at s_snap) so SlotHashes rolls real hashes. - let manifest = read_manifest_hashes( - File::open(snapshot_path).with_context(|| format!("opening snapshot {snapshot_path}"))?, - args.from, - ) - .context("reading the snapshot manifest bank hash")?; - let lt_hash = read_manifest_lt_hash( - File::open(snapshot_path).with_context(|| format!("opening snapshot {snapshot_path}"))?, - args.from, - ) - .context("reading the snapshot manifest lattice hash")? - .context("snapshot has no accounts_lt_hash (a pre-lattice snapshot?)")?; - let bootstrap = Some((lt_hash, manifest.bank_hash)); + // A fresh run reads its seed and roll bootstrap from the snapshot; --resume takes the roll state from the store's checkpoint, so no snapshot. + let snapshot_path = if args.resume { + None + } else { + let path = args + .snapshot + .as_ref() + .context(" is required for a fresh run")?; + check_snapshot_file(path)?; + Some(path) + }; + if args.resume { + println!("preflight ok: RPC, program, config, and ClickHouse check out (resume: no snapshot)"); + } else { + println!("preflight ok: RPC, program, config, ClickHouse, and snapshot all check out"); + } + + let bootstrap = match snapshot_path { + None => None, + Some(path) => { + let manifest = read_manifest_hashes( + File::open(path).with_context(|| format!("opening snapshot {path}"))?, + args.from, + ) + .context("reading the snapshot manifest bank hash")?; + let lt_hash = read_manifest_lt_hash( + File::open(path).with_context(|| format!("opening snapshot {path}"))?, + args.from, + ) + .context("reading the snapshot manifest lattice hash")? + .context("snapshot has no accounts_lt_hash (a pre-lattice snapshot?)")?; + Some((lt_hash, manifest.bank_hash)) + } + }; - let snapshot = - File::open(snapshot_path).with_context(|| format!("opening snapshot {snapshot_path}"))?; + let snapshot: Box = match snapshot_path { + None => Box::new(std::io::empty()), + Some(path) => { + Box::new(File::open(path).with_context(|| format!("opening snapshot {path}"))?) + } + }; let account_store = match args.store.as_str() { "memory" => AccountStoreChoice::Memory, @@ -164,6 +188,7 @@ fn main() -> anyhow::Result<()> { account_store, args.chunk_slots, verify_end, + args.resume, ) .await?; match &result.replay.halt { diff --git a/slate-replay/src/backfill.rs b/slate-replay/src/backfill.rs index 2ff6fd3..e68951a 100644 --- a/slate-replay/src/backfill.rs +++ b/slate-replay/src/backfill.rs @@ -1,6 +1,6 @@ use std::{collections::HashSet, io::Read, path::PathBuf, sync::Arc}; -use anyhow::Result; +use anyhow::{Context, Result}; use slate_store::ClickHouseClient; use solana_hash::Hash; use solana_lattice_hash::lt_hash::LtHash; @@ -41,6 +41,7 @@ pub async fn backfill( account_store: AccountStoreChoice, chunk_slots: usize, verify_end: Option>, + resume: bool, ) -> Result { let chunk_slots = chunk_slots.max(1); let source: Arc = match block_cache { @@ -53,18 +54,19 @@ pub async fn backfill( tokio::task::spawn_blocking(move || src.confirmed_slots(from, to)).await?? }; - // Footprint pass: stream the range once; never more than a chunk of blocks resident. + // Footprint pass: stream the range once to build the seed set. A resume skips the seed, so this only runs then if the boundary diff needs it to filter the end snapshot. let mut footprint = HashSet::new(); - for chunk in slots.chunks(chunk_slots) { - let blocks = fetch_chunk(&source, chunk).await?; - block::extend_footprint(&mut footprint, &blocks); + if !resume || verify_end.is_some() { + for chunk in slots.chunks(chunk_slots) { + let blocks = fetch_chunk(&source, chunk).await?; + block::extend_footprint(&mut footprint, &blocks); + } + block::footprint_fixed(&mut footprint); + // programData PDAs and SlotHashes are read, not declared as keys, so the footprint misses them. + let programdata = block::programdata_addresses(&footprint); + footprint.extend(programdata); + footprint.insert(solana_sdk_ids::sysvar::slot_hashes::id()); } - block::footprint_fixed(&mut footprint); - // Seed programData accounts: PDAs of the program ids, never declared keys, so the footprint misses them. - let programdata = block::programdata_addresses(&footprint); - footprint.extend(programdata); - // Seed SlotHashes: read via syscall not as an account, so the footprint misses it; the first block needs it. - footprint.insert(solana_sdk_ids::sysvar::slot_hashes::id()); // Remember the store backing so a boundary diff loads the end snapshot into the same kind. let end_store_mode = match &account_store { @@ -74,41 +76,71 @@ pub async fn backfill( } }; - // Seed the bank from the snapshot into the chosen store: a RAM map, or redb on disk. - let (mut bank, baseline) = match account_store { - AccountStoreChoice::Memory => { - let accounts = snapshot::load_accounts(snapshot, Some(&footprint), Some(program))?; - let baseline = persist::baseline_rows(&accounts, program, s_snap); - let mut bank = ReplayBank::default(); - for (pubkey, (account, slot)) in &accounts { - bank.insert(*pubkey, account.clone(), *slot); - } - (bank, baseline) + // Fresh run: seed the bank from the snapshot and roll from the manifest hashes. Resume: reopen the store and roll from its checkpoint. Third value is the slot to resume after (S_snap for a fresh run). + let (mut bank, baseline, resume_from) = if resume { + let AccountStoreChoice::Disk { path, cache_bytes } = account_store else { + anyhow::bail!("--resume requires --store disk"); + }; + let mut disk = crate::store::DiskStore::create(&path, cache_bytes)?; + let (slot, roll) = disk + .read_checkpoint() + .context("--resume: the store has no checkpoint to resume from")?; + disk.set_checkpoint_mode(true); + let mut bank = ReplayBank::with_store(Box::new(disk)); + // Empty roll = the original run had the bank-hash roll off; keep it off. + if !roll.is_empty() { + let (lt_hash, bank_hash) = crate::bankhash::deserialize_roll_state(&roll) + .context("--resume: checkpoint roll state is corrupt")?; + bank.bootstrap_bankhash(lt_hash, bank_hash); } - AccountStoreChoice::Disk { path, cache_bytes } => { - let mut disk = crate::store::DiskStore::create(&path, cache_bytes)?; - let (written, owned) = - snapshot::stream_into_store(snapshot, &mut disk, Some(&footprint), Some(program))?; - eprintln!( - "seeded {written} accounts into disk store {}", - path.display() - ); - // The seed's program-owned accounts ARE the S_snap baseline (same set the memory path derives). - let baseline = persist::baseline_rows(&owned, program, s_snap); - (ReplayBank::with_store(Box::new(disk)), baseline) + eprintln!("resuming after checkpoint at slot {slot}"); + (bank, Vec::new(), slot) + } else { + let (mut bank, baseline) = match account_store { + AccountStoreChoice::Memory => { + let accounts = snapshot::load_accounts(snapshot, Some(&footprint), Some(program))?; + let baseline = persist::baseline_rows(&accounts, program, s_snap); + let mut bank = ReplayBank::default(); + for (pubkey, (account, slot)) in &accounts { + bank.insert(*pubkey, account.clone(), *slot); + } + (bank, baseline) + } + AccountStoreChoice::Disk { path, cache_bytes } => { + let mut disk = crate::store::DiskStore::create(&path, cache_bytes)?; + let (written, owned) = snapshot::stream_into_store( + snapshot, + &mut disk, + Some(&footprint), + Some(program), + )?; + eprintln!("seeded {written} accounts into disk store {}", path.display()); + // Seed's done (auto-flushed in batches); from here hold writes so only checkpoint_flush commits. + disk.set_checkpoint_mode(true); + let baseline = persist::baseline_rows(&owned, program, s_snap); + (ReplayBank::with_store(Box::new(disk)), baseline) + } + }; + if let Some((lt_hash, bank_hash)) = bootstrap { + bank.bootstrap_bankhash(lt_hash, bank_hash); } + (bank, baseline, s_snap) }; - // Start the bank-hash roll from the manifest's lattice + bank hash so SlotHashes rolls real hashes. - if let Some((lt_hash, bank_hash)) = bootstrap { - bank.bootstrap_bankhash(lt_hash, bank_hash); + + // Fresh run: checkpoint at s_snap right after seeding, so a crash before chunk 1's checkpoint still resumes (skips the ~expensive re-seed) instead of finding no checkpoint. + if !resume { + bank.checkpoint(s_snap)?; } - // Baseline first, then replay each chunk and persist its writes, clearing the log so it stays bounded. - store.insert_accounts(&baseline).await?; + // Fresh run inserts the S_snap baseline; a resume already has it. + if !baseline.is_empty() { + store.insert_accounts(&baseline).await?; + } - // covered_hi comes from the fetched blocks, not indexing slots, the candidate list includes skipped slots. - let mut covered_hi = s_snap; - let result = if let Some(&first_slot) = slots.first() { + // On resume, replay only the slots past the checkpoint; earlier ones are already persisted. + let replay_slots: Vec = slots.into_iter().filter(|&s| s > resume_from).collect(); + let mut covered_hi = resume_from; + let result = if let Some(&first_slot) = replay_slots.first() { let epoch = first_slot / 432_000; let feature_set = build_feature_set(&bank, first_slot); let replayer = Replayer::new_with_feature_set(first_slot, epoch, feature_set); @@ -118,7 +150,7 @@ pub async fn backfill( let mut completed = 0usize; let mut halt = None; - for chunk in slots.chunks(chunk_slots) { + for chunk in replay_slots.chunks(chunk_slots) { let blocks = fetch_chunk(&source, chunk).await?; let chunk_replay = replayer.replay_range(&mut bank, &blocks); let done = chunk_replay.blocks_completed; @@ -139,6 +171,8 @@ pub async fn backfill( halt = Some(h); break; } + // Checkpoint clean chunks only. On a halt the roller sits one slot past covered_hi, so we skip it and let the halting chunk's buffered writes drop unflushed; resume re-runs that chunk. + bank.checkpoint(covered_hi)?; } RangeReplay { blocks_completed: completed, @@ -153,8 +187,8 @@ pub async fn backfill( store.record_coverage(s_snap, covered_hi).await?; - // Boundary diff (verification only, doesn't gate): byte-exact end-state vs the real snapshot over the footprint. - let boundary = if let Some(end_snapshot) = verify_end { + // Boundary diff (verification only): byte-exact end-state vs the real snapshot over the footprint. Skipped on a halt: the end-state is incomplete, and skipping it avoids touching the store past the last checkpoint. + let boundary = if let (Some(end_snapshot), None) = (verify_end, &result.halt) { bank.flush(); let mut end_store: Box = match &end_store_mode { None => Box::new(MemStore::default()), @@ -280,6 +314,7 @@ mod tests { AccountStoreChoice::Memory, 2000, None, + false, ) .await .expect("backfill"); @@ -358,6 +393,7 @@ mod tests { AccountStoreChoice::Memory, 2000, Some(Box::new(SNAPSHOT)), + false, ) .await .expect("backfill"); @@ -374,4 +410,105 @@ mod tests { &boundary.mismatches[..boundary.mismatches.len().min(5)] ); } + + // Resume: run the first half fresh, --resume the rest from the checkpoint; the final per-slot + // state must match a straight run. Disk store, since resume needs a checkpoint to reopen. + #[tokio::test] + #[ignore = "needs a local ClickHouse (slate_test db)"] + async fn resume_continues_from_a_checkpoint() { + let system = solana_sdk_ids::system_program::id(); + let accounts = snapshot::load_accounts(SNAPSHOT, None, None).unwrap(); + let (src, src_balance) = accounts + .iter() + .filter(|(_, (a, _))| *a.owner() == system && a.data().is_empty()) + .map(|(k, (a, _))| (*k, a.lamports())) + .max_by_key(|&(_, bal)| bal) + .expect("a fundable wallet"); + let (w1, w2, w3) = (Pubkey::new_unique(), Pubkey::new_unique(), Pubkey::new_unique()); + let fee = 5_000u64; + + let transfer = |from: &Pubkey, to: &Pubkey, amount: u64, from_pre: u64, slot: u64| -> Block { + let mut data = vec![2u8, 0, 0, 0]; + data.extend_from_slice(&amount.to_le_bytes()); + let ix = Instruction { + program_id: system, + accounts: vec![AccountMeta::new(*from, true), AccountMeta::new(*to, false)], + data, + }; + let message = Message::new_with_blockhash(&[ix], Some(from), &Hash::default()); + Block { + slot, + parent_slot: slot - 1, + blockhash: Hash::default(), + previous_blockhash: Hash::default(), + block_time: 1_700_000_000, + transactions: vec![BlockTx { + transaction: VersionedTransaction { + signatures: vec![Signature::default()], + message: VersionedMessage::Legacy(message), + }, + meta: TxMeta { + err: None, + fee, + compute_units_consumed: 150, + pre_balances: vec![from_pre, 0, 1], + post_balances: vec![from_pre - amount - fee, amount, 1], + loaded_addresses: LoadedAddresses::default(), + post_token_balances: vec![], + }, + }], + fee_reward: None, + } + }; + + // One hop per slot: src -> w1 (201), w1 -> w2 (202), w2 -> w3 (203). + let blocks = vec![ + transfer(&src, &w1, 3_000_000, src_balance, 201), + transfer(&w1, &w2, 2_000_000, 3_000_000, 202), + transfer(&w2, &w3, 1_000_000, 2_000_000, 203), + ]; + + let store = ClickHouseClient::with_database("http://localhost:8123", "slate_test"); + let path = std::env::temp_dir().join("slate_resume_test.redb"); + let _ = std::fs::remove_file(&path); + let _ = std::fs::remove_file(path.with_extension("end.redb")); + let disk = || AccountStoreChoice::Disk { + path: path.clone(), + cache_bytes: 32 * 1024 * 1024, + }; + let source = || Arc::new(VecBlockSource::new(blocks.clone())) as Arc; + + // Fresh run over (200, 202]: replays 201, 202 and checkpoints each (chunk_slots = 1). + backfill( + SNAPSHOT, 200, source(), None, 200, 202, &system, &store, None, disk(), 1, None, false, + ) + .await + .expect("fresh run"); + + // --resume over (200, 203]: picks up from the checkpoint at 202 and replays only 203. + let out = backfill( + SNAPSHOT, 200, source(), None, 200, 203, &system, &store, None, disk(), 1, None, true, + ) + .await + .expect("resume run"); + assert!(out.replay.is_complete(), "resume halted: {:?}", out.replay.halt); + + // w3 is funded only at 203, past the checkpoint: proves the resume replayed on. + let w3_end = store + .get_account_info(&w3.to_bytes(), 203) + .await + .unwrap() + .expect("w3 present at 203"); + assert_eq!(w3_end.lamports, 1_000_000); + // w1's slot-202 value from the fresh run survived the reopen. + let w1_end = store + .get_account_info(&w1.to_bytes(), 203) + .await + .unwrap() + .expect("w1 present"); + assert_eq!(w1_end.lamports, 3_000_000 - 2_000_000 - fee); + + let _ = std::fs::remove_file(&path); + let _ = std::fs::remove_file(path.with_extension("end.redb")); + } } diff --git a/slate-replay/src/bankhash.rs b/slate-replay/src/bankhash.rs index 5307318..494bc46 100644 --- a/slate-replay/src/bankhash.rs +++ b/slate-replay/src/bankhash.rs @@ -32,6 +32,27 @@ fn lt_hash_bytes(lt: &LtHash) -> [u8; 2048] { bytes } +// Roll state for a checkpoint: 2048 LE lattice bytes ++ 32 hash bytes (the snapshot-trailer layout). +pub(crate) fn serialize_roll_state(lt: &LtHash, bank_hash: &Hash) -> Vec { + let mut out = Vec::with_capacity(2048 + 32); + out.extend_from_slice(<_hash_bytes(lt)); + out.extend_from_slice(bank_hash.as_ref()); + out +} + +// Inverse of serialize_roll_state; None on a wrong length. +pub(crate) fn deserialize_roll_state(bytes: &[u8]) -> Option<(LtHash, Hash)> { + if bytes.len() != 2048 + 32 { + return None; + } + let mut lanes = [0u16; 1024]; + for (lane, chunk) in lanes.iter_mut().zip(bytes[..2048].chunks_exact(2)) { + *lane = u16::from_le_bytes([chunk[0], chunk[1]]); + } + let bank_hash = Hash::new_from_array(bytes[2048..2080].try_into().ok()?); + Some((LtHash(lanes), bank_hash)) +} + // Lattice-regime bank hash: SHA256(SHA256(parent||sig_count_LE||blockhash)||lt_hash[2048]). No accounts-delta (SIMD-0223), no epoch-accounts-hash (SIMD-0215). pub fn bank_hash( parent_bank_hash: &Hash, @@ -67,6 +88,11 @@ impl BankHashRoller { self.bank_hash } + // By reference: LtHash is large and deliberately not Copy. + pub fn lt_hash(&self) -> &LtHash { + &self.lt_hash + } + pub fn roll_slot( &mut self, changes: &[SlotChange], @@ -114,6 +140,19 @@ mod tests { assert_eq!(lt.checksum().0, expected); } + #[test] + fn roll_state_round_trips() { + let mut lanes = [0u16; 1024]; + for (i, l) in lanes.iter_mut().enumerate() { + *l = (i as u16).wrapping_mul(7).wrapping_add(1); + } + let lt = LtHash(lanes); + let hash = Hash::new_from_array([3u8; 32]); + let (lt2, hash2) = deserialize_roll_state(&serialize_roll_state(<, &hash)).unwrap(); + assert_eq!(lt.0, lt2.0); + assert_eq!(hash, hash2); + } + // The bank-hash combine is two nested SHA-256s over the four inputs. #[test] fn bank_hash_is_two_nested_sha256() { diff --git a/slate-replay/src/lib.rs b/slate-replay/src/lib.rs index bdad657..add4930 100644 --- a/slate-replay/src/lib.rs +++ b/slate-replay/src/lib.rs @@ -184,6 +184,16 @@ impl ReplayBank { self.bankhash_roller.as_ref().map(|r| r.bank_hash()) } + // Durably checkpoint at `slot`: flush accounts + roll state in one atomic commit, so --resume can continue from here. A no-op roll (tests) serializes to empty. + pub fn checkpoint(&mut self, slot: u64) -> anyhow::Result<()> { + let roll = self + .bankhash_roller + .as_ref() + .map(|r| crate::bankhash::serialize_roll_state(r.lt_hash(), &r.bank_hash())) + .unwrap_or_default(); + self.store.checkpoint_flush(slot, &roll) + } + // Roll the lattice over this slot's changes and compute its bank hash; None (no-op) if the roll isn't active. pub fn finalize_slot_bankhash( &mut self, diff --git a/slate-replay/src/store.rs b/slate-replay/src/store.rs index 34b89d8..0659aee 100644 --- a/slate-replay/src/store.rs +++ b/slate-replay/src/store.rs @@ -11,6 +11,9 @@ pub trait AccountStore: Send + Sync { fn contains(&self, key: &Pubkey) -> bool; // Commit buffered writes; no-op for write-through stores. fn flush(&mut self); + // Atomically flush buffered accounts + a resume checkpoint (slot + roll bytes) in one durable commit; no-op if the store can't resume. + fn checkpoint_flush(&mut self, slot: u64, roll: &[u8]) -> anyhow::Result<()>; + fn read_checkpoint(&self) -> Option<(u64, Vec)>; } // In-RAM HashMap; for tests and ranges small enough to fit. @@ -33,9 +36,18 @@ impl AccountStore for MemStore { } fn flush(&mut self) {} + + fn checkpoint_flush(&mut self, _slot: u64, _roll: &[u8]) -> anyhow::Result<()> { + Ok(()) + } + + fn read_checkpoint(&self) -> Option<(u64, Vec)> { + None + } } const ACCOUNTS: TableDefinition<&[u8], &[u8]> = TableDefinition::new("accounts"); +const META: TableDefinition<&str, &[u8]> = TableDefinition::new("meta"); // Flush the write-buffer once it holds this many bytes, so it never grows unbounded. const FLUSH_THRESHOLD_BYTES: usize = 256 * 1024 * 1024; @@ -45,6 +57,8 @@ pub struct DiskStore { db: Database, buffer: HashMap, buffered_bytes: usize, + // Set after seeding: put() stops auto-flushing and flush() no-ops, so checkpoint_flush is the only committer and the redb never holds writes past the last checkpoint (what makes --resume sound). + checkpoint_mode: bool, } impl DiskStore { @@ -57,6 +71,7 @@ impl DiskStore { db, buffer: HashMap::new(), buffered_bytes: 0, + checkpoint_mode: false, }) } @@ -67,6 +82,10 @@ impl DiskStore { let value = table.get(key.as_ref()).ok()??; decode(value.value()) } + + pub fn set_checkpoint_mode(&mut self, on: bool) { + self.checkpoint_mode = on; + } } impl AccountStore for DiskStore { @@ -83,7 +102,8 @@ impl AccountStore for DiskStore { self.buffered_bytes = self.buffered_bytes.saturating_sub(57 + old.data().len()); } self.buffered_bytes += added; - if self.buffered_bytes >= FLUSH_THRESHOLD_BYTES { + // No mid-slot auto-flush in checkpoint mode; it would put the redb ahead of the last checkpoint. + if !self.checkpoint_mode && self.buffered_bytes >= FLUSH_THRESHOLD_BYTES { self.flush(); } } @@ -93,7 +113,8 @@ impl AccountStore for DiskStore { } fn flush(&mut self) { - if self.buffer.is_empty() { + // No-op in checkpoint mode (replay_range calls this per chunk); checkpoint_flush is the only committer. + if self.checkpoint_mode || self.buffer.is_empty() { return; } let mut txn = self.db.begin_write().expect("begin write txn"); @@ -110,6 +131,38 @@ impl AccountStore for DiskStore { txn.commit().expect("commit account writes"); self.buffered_bytes = 0; } + + fn checkpoint_flush(&mut self, slot: u64, roll: &[u8]) -> anyhow::Result<()> { + let mut txn = self.db.begin_write()?; + // Immediate: accounts + checkpoint land durably together, so a crash can't split them. + txn.set_durability(Durability::Immediate); + { + let mut accounts = txn.open_table(ACCOUNTS)?; + for (key, (account, acct_slot)) in self.buffer.drain() { + accounts.insert(key.as_ref(), encode(&account, acct_slot).as_slice())?; + } + let mut meta = txn.open_table(META)?; + // slot(8 LE) ++ roll; written even on an empty buffer, so a no-write chunk still advances the slot. + let mut value = slot.to_le_bytes().to_vec(); + value.extend_from_slice(roll); + meta.insert("checkpoint", value.as_slice())?; + } + txn.commit()?; + self.buffered_bytes = 0; + Ok(()) + } + + fn read_checkpoint(&self) -> Option<(u64, Vec)> { + let txn = self.db.begin_read().ok()?; + let table = txn.open_table(META).ok()?; // no META table yet = never checkpointed + let value = table.get("checkpoint").ok()??; + let bytes = value.value(); + if bytes.len() < 8 { + return None; + } + let slot = u64::from_le_bytes(bytes[0..8].try_into().ok()?); + Some((slot, bytes[8..].to_vec())) + } } // slot(8) | lamports(8) | rent_epoch(8) | executable(1) | owner(32) | data(rest): 57-byte head + data. @@ -182,4 +235,37 @@ mod tests { let _ = std::fs::remove_file(&path); } + + #[test] + fn checkpoint_survives_reopen() { + let path = std::env::temp_dir().join("slate_diskstore_checkpoint.redb"); + let _ = std::fs::remove_file(&path); + let roll = vec![7u8; 40]; // opaque here; real roll serialization lives in bankhash + + let key = Pubkey::new_from_array([5u8; 32]); + let account = AccountSharedData::from(Account { + lamports: 9_000, + data: vec![9, 9, 9], + owner: Pubkey::new_from_array([1u8; 32]), + executable: false, + rent_epoch: 0, + }); + + { + let mut store = DiskStore::create(&path, 16 * 1024 * 1024).unwrap(); + store.set_checkpoint_mode(true); + store.put(key, account, 4242); + store.checkpoint_flush(4242, &roll).unwrap(); + } // drop closes the db, standing in for a process exit + + let store = DiskStore::create(&path, 16 * 1024 * 1024).unwrap(); + let (slot, got_roll) = store.read_checkpoint().expect("checkpoint survived reopen"); + assert_eq!(slot, 4242); + assert_eq!(got_roll, roll); + let (acct, s) = store.get(&key).expect("account present after reopen"); + assert_eq!(s, 4242); + assert_eq!(acct.lamports(), 9_000); + + let _ = std::fs::remove_file(&path); + } } From 1440c740558abd21d3abf10aaefa0344bd7b496f Mon Sep 17 00:00:00 2001 From: mctursh Date: Fri, 4 Sep 2026 22:05:04 +0100 Subject: [PATCH 2/2] ran fmt --- slate-backfill/src/main.rs | 4 +- slate-replay/src/backfill.rs | 110 +++++++++++++++++++++++------------ 2 files changed, 76 insertions(+), 38 deletions(-) diff --git a/slate-backfill/src/main.rs b/slate-backfill/src/main.rs index 109410d..b99dc99 100644 --- a/slate-backfill/src/main.rs +++ b/slate-backfill/src/main.rs @@ -118,7 +118,9 @@ fn main() -> anyhow::Result<()> { Some(path) }; if args.resume { - println!("preflight ok: RPC, program, config, and ClickHouse check out (resume: no snapshot)"); + println!( + "preflight ok: RPC, program, config, and ClickHouse check out (resume: no snapshot)" + ); } else { println!("preflight ok: RPC, program, config, ClickHouse, and snapshot all check out"); } diff --git a/slate-replay/src/backfill.rs b/slate-replay/src/backfill.rs index e68951a..1d834f5 100644 --- a/slate-replay/src/backfill.rs +++ b/slate-replay/src/backfill.rs @@ -114,7 +114,10 @@ pub async fn backfill( Some(&footprint), Some(program), )?; - eprintln!("seeded {written} accounts into disk store {}", path.display()); + eprintln!( + "seeded {written} accounts into disk store {}", + path.display() + ); // Seed's done (auto-flushed in batches); from here hold writes so only checkpoint_flush commits. disk.set_checkpoint_mode(true); let baseline = persist::baseline_rows(&owned, program, s_snap); @@ -424,42 +427,47 @@ mod tests { .map(|(k, (a, _))| (*k, a.lamports())) .max_by_key(|&(_, bal)| bal) .expect("a fundable wallet"); - let (w1, w2, w3) = (Pubkey::new_unique(), Pubkey::new_unique(), Pubkey::new_unique()); + let (w1, w2, w3) = ( + Pubkey::new_unique(), + Pubkey::new_unique(), + Pubkey::new_unique(), + ); let fee = 5_000u64; - let transfer = |from: &Pubkey, to: &Pubkey, amount: u64, from_pre: u64, slot: u64| -> Block { - let mut data = vec![2u8, 0, 0, 0]; - data.extend_from_slice(&amount.to_le_bytes()); - let ix = Instruction { - program_id: system, - accounts: vec![AccountMeta::new(*from, true), AccountMeta::new(*to, false)], - data, + let transfer = + |from: &Pubkey, to: &Pubkey, amount: u64, from_pre: u64, slot: u64| -> Block { + let mut data = vec![2u8, 0, 0, 0]; + data.extend_from_slice(&amount.to_le_bytes()); + let ix = Instruction { + program_id: system, + accounts: vec![AccountMeta::new(*from, true), AccountMeta::new(*to, false)], + data, + }; + let message = Message::new_with_blockhash(&[ix], Some(from), &Hash::default()); + Block { + slot, + parent_slot: slot - 1, + blockhash: Hash::default(), + previous_blockhash: Hash::default(), + block_time: 1_700_000_000, + transactions: vec![BlockTx { + transaction: VersionedTransaction { + signatures: vec![Signature::default()], + message: VersionedMessage::Legacy(message), + }, + meta: TxMeta { + err: None, + fee, + compute_units_consumed: 150, + pre_balances: vec![from_pre, 0, 1], + post_balances: vec![from_pre - amount - fee, amount, 1], + loaded_addresses: LoadedAddresses::default(), + post_token_balances: vec![], + }, + }], + fee_reward: None, + } }; - let message = Message::new_with_blockhash(&[ix], Some(from), &Hash::default()); - Block { - slot, - parent_slot: slot - 1, - blockhash: Hash::default(), - previous_blockhash: Hash::default(), - block_time: 1_700_000_000, - transactions: vec![BlockTx { - transaction: VersionedTransaction { - signatures: vec![Signature::default()], - message: VersionedMessage::Legacy(message), - }, - meta: TxMeta { - err: None, - fee, - compute_units_consumed: 150, - pre_balances: vec![from_pre, 0, 1], - post_balances: vec![from_pre - amount - fee, amount, 1], - loaded_addresses: LoadedAddresses::default(), - post_token_balances: vec![], - }, - }], - fee_reward: None, - } - }; // One hop per slot: src -> w1 (201), w1 -> w2 (202), w2 -> w3 (203). let blocks = vec![ @@ -480,18 +488,46 @@ mod tests { // Fresh run over (200, 202]: replays 201, 202 and checkpoints each (chunk_slots = 1). backfill( - SNAPSHOT, 200, source(), None, 200, 202, &system, &store, None, disk(), 1, None, false, + SNAPSHOT, + 200, + source(), + None, + 200, + 202, + &system, + &store, + None, + disk(), + 1, + None, + false, ) .await .expect("fresh run"); // --resume over (200, 203]: picks up from the checkpoint at 202 and replays only 203. let out = backfill( - SNAPSHOT, 200, source(), None, 200, 203, &system, &store, None, disk(), 1, None, true, + SNAPSHOT, + 200, + source(), + None, + 200, + 203, + &system, + &store, + None, + disk(), + 1, + None, + true, ) .await .expect("resume run"); - assert!(out.replay.is_complete(), "resume halted: {:?}", out.replay.halt); + assert!( + out.replay.is_complete(), + "resume halted: {:?}", + out.replay.halt + ); // w3 is funded only at 203, past the checkpoint: proves the resume replayed on. let w3_end = store