+6

constellation/src/bin/main.rs

reviewed

··· 34 34 #[arg(short, long)] 35 35 #[clap(value_enum, default_value_t = StorageBackend::Memory)] 36 36 backend: StorageBackend, 37 37 + /// Initiate a database backup into this dir, if supported by the storage 38 38 + #[arg(long)] 39 39 + backup: Option<PathBuf>, 37 40 /// Saved jsonl from jetstream to use instead of a live subscription 38 41 #[arg(short, long)] 39 42 fixture: Option<PathBuf>, ··· 76 79 let storage_dir = args.data.clone().unwrap_or("rocks.test".into()); 77 80 println!("starting rocksdb..."); 78 81 let rocks = RocksStorage::new(storage_dir)?; 82 82 + if let Some(backup_dir) = args.backup { 83 83 + rocks.start_backup(backup_dir)?; 84 84 + } 79 85 println!("rocks ready."); 80 86 run(rocks, fixture, args.data, stream) 81 87 }

+22

constellation/src/storage/rocks_store.rs

reviewed

··· 292 292 }) 293 293 } 294 294 295 295 + pub fn start_backup(&self, path: impl AsRef<Path>) -> Result<()> { 296 296 + use rocksdb::backup::{BackupEngine, BackupEngineOptions}; 297 297 + eprintln!("getting ready to start backup..."); 298 298 + let mut engine = 299 299 + BackupEngine::open(&BackupEngineOptions::new(path)?, &rocksdb::Env::new()?)?; 300 300 + std::thread::spawn({ 301 301 + let db = self.db.clone(); 302 302 + move || { 303 303 + eprintln!("backup starting."); 304 304 + let t0 = Instant::now(); 305 305 + if let Err(e) = engine.create_new_backup(&db) { 306 306 + eprintln!("oh no, backup failed: {e:?}"); 307 307 + } else { 308 308 + eprintln!("yay, backup worked?"); 309 309 + } 310 310 + eprintln!("backup finished after {:.2}s", t0.elapsed().as_secs_f32()); 311 311 + } 312 312 + }); 313 313 + eprintln!("backups should be happening in bg thread."); 314 314 + Ok(()) 315 315 + } 316 316 + 295 317 fn describe_metrics() { 296 318 describe_histogram!( 297 319 "storage_rocksdb_read_seconds",