diff --git a/slate-backfill/src/main.rs b/slate-backfill/src/main.rs index d398e22..b99dc99 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,59 @@ 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 snapshot = - File::open(snapshot_path).with_context(|| format!("opening snapshot {snapshot_path}"))?; + 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: 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 +190,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..1d834f5 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,74 @@ 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 +153,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 +174,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 +190,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 +317,7 @@ mod tests { AccountStoreChoice::Memory, 2000, None, + false, ) .await .expect("backfill"); @@ -358,6 +396,7 @@ mod tests { AccountStoreChoice::Memory, 2000, Some(Box::new(SNAPSHOT)), + false, ) .await .expect("backfill"); @@ -374,4 +413,138 @@ 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); + } }