diff --git a/src/bin/rpki_query_service.rs b/src/bin/rpki_query_service.rs index c90b21d..6395c01 100644 --- a/src/bin/rpki_query_service.rs +++ b/src/bin/rpki_query_service.rs @@ -150,9 +150,20 @@ fn real_main() -> Result<(), String> { let db = Arc::new(open_query_db_for_service(&args).map_err(|e| e.to_string())?); let export_jobs = Arc::new(Mutex::new(BTreeMap::new())); let repo_bytes = match args.repo_bytes_db.as_ref() { - Some(path) => Some(Arc::new( - ExternalRepoBytesDb::open_read_only(path).map_err(|e| e.to_string())?, - )), + Some(path) => { + let opened = if args.watch_run_root.is_some() { + // Watch mode tracks a live soak that keeps writing new object + // bytes. A plain read-only open is frozen at open time (new + // manifests/CRLs/objects become unparsable until restart), so + // use a secondary instance and catch up in the watcher loop. + let secondary_path = args.query_db.with_extension("repo-bytes.secondary"); + ExternalRepoBytesDb::open_as_secondary(path, &secondary_path) + .map_err(|e| e.to_string())? + } else { + ExternalRepoBytesDb::open_read_only(path).map_err(|e| e.to_string())? + }; + Some(Arc::new(opened)) + } None => None, }; if let Some(run_root) = args.watch_run_root.clone() { @@ -163,6 +174,7 @@ fn real_main() -> Result<(), String> { )?; spawn_indexer_watcher( Arc::clone(&db), + repo_bytes.clone(), args.query_db.clone(), resolve_indexer_bin(args.indexer_bin.clone())?, run_root, @@ -229,6 +241,7 @@ fn resolve_effective_watch_min_run_seq( fn spawn_indexer_watcher( db: Arc, + repo_bytes: Option>, query_db_path: PathBuf, indexer_bin: PathBuf, run_root: PathBuf, @@ -250,6 +263,9 @@ fn spawn_indexer_watcher( ) { Ok(message) => { let _ = db.try_catch_up_with_primary(); + if let Some(repo_bytes) = repo_bytes.as_ref() { + let _ = repo_bytes.try_catch_up_with_primary(); + } eprintln!("query watcher indexed {}: {message}", run_dir.display()); } Err(err) => eprintln!( @@ -260,6 +276,9 @@ fn spawn_indexer_watcher( } Ok(None) => { let _ = db.try_catch_up_with_primary(); + if let Some(repo_bytes) = repo_bytes.as_ref() { + let _ = repo_bytes.try_catch_up_with_primary(); + } } Err(err) => eprintln!("query watcher index failed: {err}"), } diff --git a/src/blob_store.rs b/src/blob_store.rs index cd65327..df01be1 100644 --- a/src/blob_store.rs +++ b/src/blob_store.rs @@ -78,6 +78,7 @@ pub struct ExternalRepoBytesDb { path: PathBuf, db: Arc, read_only: bool, + secondary: bool, } impl ExternalRawStoreDb { @@ -191,6 +192,7 @@ impl ExternalRepoBytesDb { path, db: Arc::new(db), read_only: false, + secondary: false, }) } @@ -204,9 +206,52 @@ impl ExternalRepoBytesDb { path, db: Arc::new(db), read_only: true, + secondary: false, }) } + /// Open the repo-bytes DB as a RocksDB secondary instance. + /// + /// Unlike `open_read_only` (a frozen point-in-time view), a secondary + /// instance can follow the live primary via `try_catch_up_with_primary`, + /// which is required when the soak keeps writing new object bytes while + /// the query service is running. + pub fn open_as_secondary( + path: impl Into, + secondary_path: impl Into, + ) -> StorageResult { + let path = path.into(); + let secondary_path = secondary_path.into(); + if let Some(parent) = secondary_path.parent() { + std::fs::create_dir_all(parent).map_err(|e| StorageError::RocksDb(e.to_string()))?; + } + let mut opts = Options::default(); + opts.set_compression_type(rocksdb::DBCompressionType::Lz4); + let db = DB::open_as_secondary(&opts, &path, &secondary_path) + .map_err(|e| StorageError::RocksDb(e.to_string()))?; + Ok(Self { + path, + db: Arc::new(db), + read_only: true, + secondary: true, + }) + } + + /// Pull the secondary view up to the primary's current state. No-op for + /// non-secondary handles. + pub fn try_catch_up_with_primary(&self) -> StorageResult<()> { + if self.secondary { + self.db + .try_catch_up_with_primary() + .map_err(|e| StorageError::RocksDb(e.to_string()))?; + } + Ok(()) + } + + pub(crate) fn is_secondary(&self) -> bool { + self.secondary + } + pub fn put_blob_bytes_batch(&self, blobs: &[(String, Vec)]) -> StorageResult<()> { if blobs.is_empty() { return Ok(()); @@ -740,4 +785,48 @@ mod tests { ); assert!(repo_bytes.get_blob_bytes("not-a-valid-hash").is_err()); } + + #[test] + fn external_repo_bytes_db_secondary_catches_up_with_live_primary() { + let td = tempfile::tempdir().expect("tempdir"); + let primary_path = td.path().join("repo-bytes.db"); + let secondary_path = td.path().join("repo-bytes.secondary"); + + let primary = ExternalRepoBytesDb::open(&primary_path).expect("open primary"); + let bytes_a = b"repo-bytes-a".to_vec(); + let hash_a = sha256_hex(&bytes_a); + primary + .put_blob_bytes_batch(&[(hash_a.clone(), bytes_a.clone())]) + .expect("put a"); + + let secondary = + ExternalRepoBytesDb::open_as_secondary(&primary_path, &secondary_path) + .expect("open secondary"); + assert!(secondary.is_secondary()); + + // A secondary open does not necessarily see pre-existing data until it + // catches up with the primary's manifest. + secondary + .try_catch_up_with_primary() + .expect("initial catch up"); + assert_eq!( + secondary.get_blob_bytes(&hash_a).expect("get a via secondary"), + Some(bytes_a) + ); + + // Bytes written by the primary *after* the secondary open become + // visible after another catch-up (this is the live-soak scenario). + let bytes_b = b"repo-bytes-b".to_vec(); + let hash_b = sha256_hex(&bytes_b); + primary + .put_blob_bytes_batch(&[(hash_b.clone(), bytes_b.clone())]) + .expect("put b after secondary open"); + secondary + .try_catch_up_with_primary() + .expect("second catch up"); + assert_eq!( + secondary.get_blob_bytes(&hash_b).expect("get b via secondary"), + Some(bytes_b) + ); + } }