20260723 修复 explorer 新发布对象 parsed 不可用(#128 M8):repo-bytes.db 只读打开冻结于启动时刻,watch 模式改用 rocksdb secondary 并在 watcher 循环周期 catch-up;blob_store 新增 open_as_secondary/try_catch_up_with_primary 及 live primary 场景单测

This commit is contained in:
yuyr 2026-07-24 11:15:16 +08:00
parent b5c033426e
commit 001f13209f
2 changed files with 111 additions and 3 deletions

View File

@ -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<QueryDb>,
repo_bytes: Option<Arc<ExternalRepoBytesDb>>,
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}"),
}

View File

@ -78,6 +78,7 @@ pub struct ExternalRepoBytesDb {
path: PathBuf,
db: Arc<DB>,
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<PathBuf>,
secondary_path: impl Into<PathBuf>,
) -> StorageResult<Self> {
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<u8>)]) -> 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)
);
}
}