20260726 修复rpki-explorer体验问题32项(#130):RocksDB max_open_files+compose ulimits、objects列表流式早停+scope前置(每页2.85s->10ms级)、未知/过期run统一404、runs倒序、导出文件名Content-Disposition、路径段percent_decode、nginx /api/v1代理与400回SPA;前端serialNumber schema、30s请求超时、reason筛选、组合VRP查询、导出反馈与repo/PP入口、键盘可达、cursor进URL、分页/对比度/skip-link等体验项
This commit is contained in:
@@ -542,7 +542,18 @@ impl ApiError {
|
||||
|
||||
impl From<QueryDbError> for ApiError {
|
||||
fn from(value: QueryDbError) -> Self {
|
||||
Self::new(500, value.to_string())
|
||||
match value {
|
||||
// The run is still indexed but its report.json was removed by
|
||||
// retention cleanup: this is a client-visible "gone", not a
|
||||
// server fault.
|
||||
QueryDbError::RunDataGone(run_id) => Self::new(
|
||||
404,
|
||||
format!(
|
||||
"report data for run {run_id} has expired and was cleaned up by retention"
|
||||
),
|
||||
),
|
||||
other => Self::new(500, other.to_string()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -555,11 +566,10 @@ fn route_request(
|
||||
let (path, query) = split_target(target);
|
||||
let query = parse_query(query);
|
||||
let path = path.trim_end_matches('/');
|
||||
let segments = path
|
||||
.trim_start_matches("/api/v1")
|
||||
.trim_start_matches('/')
|
||||
.split('/')
|
||||
.filter(|s| !s.is_empty())
|
||||
let decoded_segments = decode_path_segments(path);
|
||||
let segments = decoded_segments
|
||||
.iter()
|
||||
.map(String::as_str)
|
||||
.collect::<Vec<_>>();
|
||||
if segments.is_empty() {
|
||||
return Ok(json!({"data":{"service":"rpki_query_service","version":1}}));
|
||||
@@ -1002,11 +1012,10 @@ fn route_post_request(
|
||||
) -> Result<Value, ApiError> {
|
||||
let (path, _) = split_target(target);
|
||||
let path = path.trim_end_matches('/');
|
||||
let segments = path
|
||||
.trim_start_matches("/api/v1")
|
||||
.trim_start_matches('/')
|
||||
.split('/')
|
||||
.filter(|s| !s.is_empty())
|
||||
let decoded_segments = decode_path_segments(path);
|
||||
let segments = decoded_segments
|
||||
.iter()
|
||||
.map(String::as_str)
|
||||
.collect::<Vec<_>>();
|
||||
match segments.as_slice() {
|
||||
["runs", raw_run_id, "exports"] => {
|
||||
@@ -1159,11 +1168,10 @@ fn route_raw_request(
|
||||
) -> Option<Result<Vec<u8>, ApiError>> {
|
||||
let (path, _) = split_target(target);
|
||||
let path = path.trim_end_matches('/');
|
||||
let segments = path
|
||||
.trim_start_matches("/api/v1")
|
||||
.trim_start_matches('/')
|
||||
.split('/')
|
||||
.filter(|s| !s.is_empty())
|
||||
let decoded_segments = decode_path_segments(path);
|
||||
let segments = decoded_segments
|
||||
.iter()
|
||||
.map(String::as_str)
|
||||
.collect::<Vec<_>>();
|
||||
match segments.as_slice() {
|
||||
["runs", raw_run_id, "objects", object_instance_id, "raw"] => Some(raw_object_response(
|
||||
@@ -1224,7 +1232,20 @@ fn export_download_response(
|
||||
.as_ref()
|
||||
.ok_or_else(|| ApiError::new(404, "export output path not found"))?;
|
||||
let bytes = fs::read(path).map_err(|err| ApiError::new(404, err.to_string()))?;
|
||||
Ok(binary_response(200, "application/x-tar", bytes))
|
||||
let filename = format!(
|
||||
"rpki-export-{}-{}.tar",
|
||||
sanitize_filename_token(&run_id),
|
||||
sanitize_filename_token(job_id)
|
||||
);
|
||||
Ok(binary_response_with_headers(
|
||||
200,
|
||||
"application/x-tar",
|
||||
&[(
|
||||
"Content-Disposition".to_string(),
|
||||
format!("attachment; filename=\"{filename}\""),
|
||||
)],
|
||||
bytes,
|
||||
))
|
||||
}
|
||||
|
||||
fn export_job_store_key(run_id: &str, job_id: &str) -> String {
|
||||
@@ -2098,6 +2119,18 @@ fn hex_value(byte: u8) -> Option<u8> {
|
||||
}
|
||||
}
|
||||
|
||||
/// Split an API path into percent-decoded segments. Real-world ids are plain
|
||||
/// hex, so decoding is a no-op for them; it only matters for hand-built URLs
|
||||
/// (e.g. an id containing a literal `%2F`).
|
||||
fn decode_path_segments(path: &str) -> Vec<String> {
|
||||
path.trim_start_matches("/api/v1")
|
||||
.trim_start_matches('/')
|
||||
.split('/')
|
||||
.filter(|s| !s.is_empty())
|
||||
.map(|segment| percent_decode(segment).unwrap_or_else(|| segment.to_string()))
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn limit(query: &BTreeMap<String, String>) -> usize {
|
||||
query
|
||||
.get("limit")
|
||||
@@ -2248,13 +2281,26 @@ fn json_response(status: u16, value: &Value) -> Vec<u8> {
|
||||
}
|
||||
|
||||
fn binary_response(status: u16, content_type: &str, body: Vec<u8>) -> Vec<u8> {
|
||||
binary_response_with_headers(status, content_type, &[], body)
|
||||
}
|
||||
|
||||
fn binary_response_with_headers(
|
||||
status: u16,
|
||||
content_type: &str,
|
||||
extra_headers: &[(String, String)],
|
||||
body: Vec<u8>,
|
||||
) -> Vec<u8> {
|
||||
let reason = match status {
|
||||
200 => "OK",
|
||||
404 => "Not Found",
|
||||
_ => "Internal Server Error",
|
||||
};
|
||||
let mut head = format!("HTTP/1.1 {status} {reason}\r\nContent-Type: {content_type}\r\n");
|
||||
for (name, value) in extra_headers {
|
||||
head.push_str(&format!("{name}: {value}\r\n"));
|
||||
}
|
||||
let mut response = format!(
|
||||
"HTTP/1.1 {status} {reason}\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
|
||||
"{head}Content-Length: {}\r\nConnection: close\r\n\r\n",
|
||||
body.len()
|
||||
)
|
||||
.into_bytes();
|
||||
@@ -2262,6 +2308,15 @@ fn binary_response(status: u16, content_type: &str, body: Vec<u8>) -> Vec<u8> {
|
||||
response
|
||||
}
|
||||
|
||||
/// Header-safe token for use inside a `Content-Disposition` filename: strips
|
||||
/// anything that could smuggle CRLF or quotes into the response head.
|
||||
fn sanitize_filename_token(value: &str) -> String {
|
||||
value
|
||||
.chars()
|
||||
.filter(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.'))
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn is_sha256_hex(value: &str) -> bool {
|
||||
value.len() == 64 && value.bytes().all(|byte| byte.is_ascii_hexdigit())
|
||||
}
|
||||
@@ -2626,6 +2681,94 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
fn index_sample_run(temp: &tempfile::TempDir) -> QueryDb {
|
||||
let run_dir = temp.path().join("runs/run_0001");
|
||||
fs::create_dir_all(&run_dir).expect("run dir");
|
||||
let object_bytes =
|
||||
fs::read("tests/fixtures/repository/rpki.cernet.net/repo/cernet/0/AS4538.roa")
|
||||
.expect("fixture roa");
|
||||
let object_sha = hex::encode(Sha256::digest(&object_bytes));
|
||||
write_sample_run(&run_dir, &object_sha);
|
||||
let query_db_path = temp.path().join("query-db");
|
||||
index_artifacts(&ArtifactIndexerConfig {
|
||||
query_db_path: query_db_path.clone(),
|
||||
run_root: Some(temp.path().to_path_buf()),
|
||||
run_dir: None,
|
||||
repo_bytes_db_path: None,
|
||||
projection_entry_limit: 5,
|
||||
min_run_seq: None,
|
||||
retain_indexed_runs: None,
|
||||
})
|
||||
.expect("index artifacts");
|
||||
QueryDb::open(&query_db_path).expect("query db")
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unknown_run_id_returns_404_instead_of_empty_payload() {
|
||||
let temp = tempfile::tempdir().expect("tempdir");
|
||||
let db = index_sample_run(&temp);
|
||||
|
||||
for target in [
|
||||
"/api/v1/runs/run_9999",
|
||||
"/api/v1/runs/run_9999/summary",
|
||||
"/api/v1/runs/run_9999/objects",
|
||||
"/api/v1/runs/run_9999/issues",
|
||||
"/api/v1/runs/run_9999/stats/overview",
|
||||
"/api/v1/runs/run_9999/stats/validation-events?name=manifest",
|
||||
] {
|
||||
let err = match test_route_request(&db, None, target) {
|
||||
Ok(_) => panic!("expected 404 for {target}"),
|
||||
Err(err) => err,
|
||||
};
|
||||
assert_eq!(err.status, 404, "{target}");
|
||||
}
|
||||
// Existing runs and "latest" keep working.
|
||||
assert!(
|
||||
test_route_request(&db, None, "/api/v1/runs/run_0001/stats/overview").is_ok()
|
||||
);
|
||||
assert!(test_route_request(&db, None, "/api/v1/runs/latest/objects").is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn expired_run_report_returns_404_run_data_gone() {
|
||||
let temp = tempfile::tempdir().expect("tempdir");
|
||||
let db = index_sample_run(&temp);
|
||||
assert!(test_route_request(&db, None, "/api/v1/runs/run_0001/objects").is_ok());
|
||||
|
||||
// Retention cleanup removes the run directory while the index record
|
||||
// survives: object routes must report a distinguishable 404.
|
||||
fs::remove_dir_all(temp.path().join("runs/run_0001")).expect("remove run dir");
|
||||
let err = test_route_request(&db, None, "/api/v1/runs/run_0001/objects")
|
||||
.expect_err("run data gone");
|
||||
assert_eq!(err.status, 404);
|
||||
assert!(
|
||||
err.message.contains("expired"),
|
||||
"unexpected message: {}",
|
||||
err.message
|
||||
);
|
||||
let err = test_route_request(&db, None, "/api/v1/runs/run_0001/issues")
|
||||
.expect_err("run data gone");
|
||||
assert_eq!(err.status, 404);
|
||||
// Stats served from the index keep working.
|
||||
assert!(
|
||||
test_route_request(&db, None, "/api/v1/runs/run_0001/stats/overview").is_ok()
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn decode_path_segments_percent_decodes_ids() {
|
||||
let segments = decode_path_segments("/api/v1/runs/run_0001/objects/abc%2Fdef");
|
||||
assert_eq!(segments, vec!["runs", "run_0001", "objects", "abc/def"]);
|
||||
assert!(decode_path_segments("/api/v1").is_empty());
|
||||
assert!(decode_path_segments("/api/v1/").is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sanitize_filename_token_strips_crlf_and_quotes() {
|
||||
assert_eq!(sanitize_filename_token("run_0001"), "run_0001");
|
||||
assert_eq!(sanitize_filename_token("a\r\nb\"c d/e"), "abcde");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn raw_route_downloads_object_bytes_from_repo_bytes_db() {
|
||||
let temp = tempfile::tempdir().expect("tempdir");
|
||||
@@ -2813,6 +2956,31 @@ mod tests {
|
||||
assert_eq!(job.status, "complete");
|
||||
assert_eq!(job.object_count, 1);
|
||||
assert!(job.bytes_written > 0);
|
||||
|
||||
let download = test_route_raw_request(
|
||||
&db,
|
||||
None,
|
||||
&format!("/api/v1/runs/latest/exports/{job_id}/download"),
|
||||
)
|
||||
.expect("download route")
|
||||
.expect("download response");
|
||||
let head_end = download
|
||||
.windows(4)
|
||||
.position(|window| window == b"\r\n\r\n")
|
||||
.expect("header separator");
|
||||
let head = String::from_utf8_lossy(&download[..head_end]);
|
||||
assert!(head.starts_with("HTTP/1.1 200 OK"), "head: {head}");
|
||||
assert!(
|
||||
head.contains("Content-Type: application/x-tar\r\n"),
|
||||
"head: {head}"
|
||||
);
|
||||
let expected_disposition = format!(
|
||||
"Content-Disposition: attachment; filename=\"rpki-export-run_0001-{job_id}.tar\""
|
||||
);
|
||||
assert!(
|
||||
head.contains(&expected_disposition),
|
||||
"head: {head}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -90,6 +90,7 @@ impl ExternalRawStoreDb {
|
||||
let mut opts = Options::default();
|
||||
opts.create_if_missing(true);
|
||||
opts.set_compression_type(rocksdb::DBCompressionType::Lz4);
|
||||
opts.set_max_open_files(512);
|
||||
let db = DB::open(&opts, &path).map_err(|e| StorageError::RocksDb(e.to_string()))?;
|
||||
Ok(Self {
|
||||
path,
|
||||
@@ -187,6 +188,7 @@ impl ExternalRepoBytesDb {
|
||||
let mut opts = Options::default();
|
||||
opts.create_if_missing(true);
|
||||
opts.set_compression_type(rocksdb::DBCompressionType::Lz4);
|
||||
opts.set_max_open_files(512);
|
||||
let db = DB::open(&opts, &path).map_err(|e| StorageError::RocksDb(e.to_string()))?;
|
||||
Ok(Self {
|
||||
path,
|
||||
@@ -200,6 +202,7 @@ impl ExternalRepoBytesDb {
|
||||
let path = path.into();
|
||||
let mut opts = Options::default();
|
||||
opts.set_compression_type(rocksdb::DBCompressionType::Lz4);
|
||||
opts.set_max_open_files(512);
|
||||
let db = DB::open_for_read_only(&opts, &path, false)
|
||||
.map_err(|e| StorageError::RocksDb(e.to_string()))?;
|
||||
Ok(Self {
|
||||
@@ -227,6 +230,7 @@ impl ExternalRepoBytesDb {
|
||||
}
|
||||
let mut opts = Options::default();
|
||||
opts.set_compression_type(rocksdb::DBCompressionType::Lz4);
|
||||
opts.set_max_open_files(512);
|
||||
let db = DB::open_as_secondary(&opts, &path, &secondary_path)
|
||||
.map_err(|e| StorageError::RocksDb(e.to_string()))?;
|
||||
Ok(Self {
|
||||
|
||||
@@ -176,9 +176,14 @@ pub fn list_report_objects_filtered(
|
||||
start_pp_index,
|
||||
start_object_index,
|
||||
);
|
||||
ObjectScanSeed { state: &mut state }
|
||||
.deserialize(&mut deserializer)
|
||||
.map_err(QueryDbError::from)?;
|
||||
let scan = ObjectScanSeed { state: &mut state }.deserialize(&mut deserializer);
|
||||
match scan {
|
||||
Ok(()) => {}
|
||||
// A full page plus its next cursor is all the caller needs; the
|
||||
// visitor aborts the remaining scan with a sentinel error.
|
||||
Err(err) if is_object_scan_complete(&err) => {}
|
||||
Err(err) => return Err(QueryDbError::from(err)),
|
||||
}
|
||||
Ok(QueryPage {
|
||||
data: state.data,
|
||||
next_cursor: state.next_cursor,
|
||||
@@ -215,6 +220,16 @@ pub fn object_cursor(pp_index: u64, object_index: u64) -> String {
|
||||
format!("r1:{pp_index}:{object_index}")
|
||||
}
|
||||
|
||||
/// Sentinel message used to abort the streaming object scan early once a list
|
||||
/// page (plus its next cursor) has been collected. The deserializer is left
|
||||
/// mid-document and dropped, which is fine because the caller only needs the
|
||||
/// state accumulated so far.
|
||||
const OBJECT_SCAN_COMPLETE_MSG: &str = "rpki report object scan complete: page collected";
|
||||
|
||||
fn is_object_scan_complete(err: &serde_json::Error) -> bool {
|
||||
err.is_data() && err.to_string().starts_with(OBJECT_SCAN_COMPLETE_MSG)
|
||||
}
|
||||
|
||||
fn parse_object_cursor(cursor: Option<&str>) -> QueryDbResult<(u64, u64)> {
|
||||
let Some(cursor) = cursor else {
|
||||
return Ok((0, 0));
|
||||
@@ -732,6 +747,15 @@ impl<'a> ObjectScanState<'a> {
|
||||
scanned_objects: 0,
|
||||
}
|
||||
}
|
||||
|
||||
/// True once a list page is full and its next cursor is fixed; everything
|
||||
/// after this point cannot alter the page (cursor semantics: the cursor
|
||||
/// already points at the first item of the next page).
|
||||
fn list_page_complete(&self) -> bool {
|
||||
matches!(self.mode, ObjectScanMode::List)
|
||||
&& self.data.len() >= self.limit
|
||||
&& self.next_cursor.is_some()
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
@@ -807,7 +831,13 @@ impl<'de> Visitor<'de> for ObjectScanPpsVisitor<'_, '_> {
|
||||
A: SeqAccess<'de>,
|
||||
{
|
||||
let mut pp_index = 0u64;
|
||||
while let Some(raw_pp) = seq.next_element::<Box<RawValue>>()? {
|
||||
loop {
|
||||
if self.state.list_page_complete() {
|
||||
return Err(de::Error::custom(OBJECT_SCAN_COMPLETE_MSG));
|
||||
}
|
||||
let Some(raw_pp) = seq.next_element::<Box<RawValue>>()? else {
|
||||
break;
|
||||
};
|
||||
process_raw_publication_point(self.state, pp_index, raw_pp.get())
|
||||
.map_err(de::Error::custom)?;
|
||||
pp_index += 1;
|
||||
@@ -822,6 +852,20 @@ fn process_raw_publication_point(
|
||||
raw_pp: &str,
|
||||
) -> QueryDbResult<()> {
|
||||
state.scanned_publication_points += 1;
|
||||
if matches!(state.mode, ObjectScanMode::List) && pp_index < state.start_pp_index {
|
||||
// Every object in this PP sits before the cursor, so none of them can
|
||||
// be part of the page: skip materializing the subtree entirely.
|
||||
return Ok(());
|
||||
}
|
||||
if state.scope != ObjectScope::All {
|
||||
// Scope is decidable from PP-level fields alone; probe them before
|
||||
// materializing the (potentially huge) objects subtree.
|
||||
let probe: PublicationPointScopeProbe = serde_json::from_str(raw_pp)?;
|
||||
let ctx = PpObjectContext::from_pp(state.run_id, &probe.into_summary());
|
||||
if !state.scope.matches(&ctx) {
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
let pp: PublicationPointForObjects = serde_json::from_str(raw_pp)?;
|
||||
let ctx = PpObjectContext::from_pp(state.run_id, &pp.summary);
|
||||
if !state.scope.matches(&ctx) {
|
||||
@@ -909,6 +953,30 @@ impl<'de> Visitor<'de> for PublicationPointForObjectsVisitor {
|
||||
}
|
||||
}
|
||||
|
||||
/// Lightweight probe of the PP-level fields needed to decide `ObjectScope`
|
||||
/// before the (potentially huge) objects subtree is materialized.
|
||||
#[derive(Clone, Debug, Default, Deserialize)]
|
||||
struct PublicationPointScopeProbe {
|
||||
node_id: Option<u64>,
|
||||
rsync_base_uri: Option<String>,
|
||||
manifest_rsync_uri: Option<String>,
|
||||
publication_point_rsync_uri: Option<String>,
|
||||
rrdp_notification_uri: Option<String>,
|
||||
}
|
||||
|
||||
impl PublicationPointScopeProbe {
|
||||
fn into_summary(self) -> PublicationPointSummary {
|
||||
PublicationPointSummary {
|
||||
node_id: self.node_id,
|
||||
rsync_base_uri: self.rsync_base_uri,
|
||||
manifest_rsync_uri: self.manifest_rsync_uri,
|
||||
publication_point_rsync_uri: self.publication_point_rsync_uri,
|
||||
rrdp_notification_uri: self.rrdp_notification_uri,
|
||||
..PublicationPointSummary::default()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Default)]
|
||||
struct ReportObject {
|
||||
uri: String,
|
||||
@@ -1360,4 +1428,195 @@ mod tests {
|
||||
assert_eq!(found.object.result, "error");
|
||||
assert_eq!(found.resolution.mode, "report_scan");
|
||||
}
|
||||
|
||||
/// Three PPs with two objects each; used to anchor paging and scope
|
||||
/// behavior of the early-stop object scan.
|
||||
fn write_multi_pp_report(report_path: &std::path::Path) {
|
||||
fs::write(
|
||||
report_path,
|
||||
r#"{
|
||||
"meta":{"validation_time_rfc3339_utc":"2026-06-16T00:00:00Z"},
|
||||
"tree":{"warnings":[]},
|
||||
"publication_points":[
|
||||
{
|
||||
"node_id":1,
|
||||
"rsync_base_uri":"rsync://repo-a.example/rpki/",
|
||||
"manifest_rsync_uri":"rsync://repo-a.example/rpki/m.mft",
|
||||
"publication_point_rsync_uri":"rsync://repo-a.example/rpki/",
|
||||
"rrdp_notification_uri":"https://repo-a.example/rrdp/notification.xml",
|
||||
"objects":[
|
||||
{"rsync_uri":"rsync://repo-a.example/rpki/m.mft","sha256_hex":"a1","kind":"manifest","result":"ok"},
|
||||
{"rsync_uri":"rsync://repo-a.example/rpki/a.roa","sha256_hex":"a2","kind":"roa","result":"ok"}
|
||||
]
|
||||
},
|
||||
{
|
||||
"node_id":2,
|
||||
"rsync_base_uri":"rsync://repo-b.example/rpki/",
|
||||
"manifest_rsync_uri":"rsync://repo-b.example/rpki/m.mft",
|
||||
"publication_point_rsync_uri":"rsync://repo-b.example/rpki/",
|
||||
"objects":[
|
||||
{"rsync_uri":"rsync://repo-b.example/rpki/m.mft","sha256_hex":"b1","kind":"manifest","result":"ok"},
|
||||
{"rsync_uri":"rsync://repo-b.example/rpki/b.roa","sha256_hex":"b2","kind":"roa","result":"error","detail":"bad roa"}
|
||||
]
|
||||
},
|
||||
{
|
||||
"node_id":3,
|
||||
"rsync_base_uri":"rsync://repo-c.example/rpki/",
|
||||
"manifest_rsync_uri":"rsync://repo-c.example/rpki/m.mft",
|
||||
"publication_point_rsync_uri":"rsync://repo-c.example/rpki/",
|
||||
"rrdp_notification_uri":"https://repo-c.example/rrdp/notification.xml",
|
||||
"objects":[
|
||||
{"rsync_uri":"rsync://repo-c.example/rpki/m.mft","sha256_hex":"c1","kind":"manifest","result":"ok"},
|
||||
{"rsync_uri":"rsync://repo-c.example/rpki/c.roa","sha256_hex":"c2","kind":"roa","result":"ok"}
|
||||
]
|
||||
}
|
||||
]
|
||||
}"#,
|
||||
)
|
||||
.expect("report");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn object_listing_pages_across_publication_points_with_stable_cursors() {
|
||||
let temp = tempfile::tempdir().expect("tempdir");
|
||||
let report_path = temp.path().join("report.json");
|
||||
write_multi_pp_report(&report_path);
|
||||
|
||||
// Page 1: fills inside PP 0; the cursor must resume at PP 0 object 2.
|
||||
let page1 = list_report_objects(&report_path, "run_0001", ObjectScope::All, 2, None)
|
||||
.expect("page1");
|
||||
assert_eq!(page1.data.len(), 2);
|
||||
assert_eq!(page1.data[0].uri, "rsync://repo-a.example/rpki/m.mft");
|
||||
assert_eq!(page1.data[1].uri, "rsync://repo-a.example/rpki/a.roa");
|
||||
assert_eq!(page1.next_cursor.as_deref(), Some("r1:0:2"));
|
||||
|
||||
// Page 2: PP 0 is exhausted; the two PP 1 objects fill the page.
|
||||
let page2 = list_report_objects(
|
||||
&report_path,
|
||||
"run_0001",
|
||||
ObjectScope::All,
|
||||
2,
|
||||
page1.next_cursor.as_deref(),
|
||||
)
|
||||
.expect("page2");
|
||||
assert_eq!(page2.data.len(), 2);
|
||||
assert_eq!(page2.data[0].uri, "rsync://repo-b.example/rpki/m.mft");
|
||||
assert_eq!(page2.data[1].uri, "rsync://repo-b.example/rpki/b.roa");
|
||||
assert_eq!(page2.next_cursor.as_deref(), Some("r1:1:2"));
|
||||
|
||||
// Page 3: last PP, page fills at the final object.
|
||||
let page3 = list_report_objects(
|
||||
&report_path,
|
||||
"run_0001",
|
||||
ObjectScope::All,
|
||||
2,
|
||||
page2.next_cursor.as_deref(),
|
||||
)
|
||||
.expect("page3");
|
||||
assert_eq!(page3.data.len(), 2);
|
||||
assert_eq!(page3.next_cursor.as_deref(), Some("r1:2:2"));
|
||||
|
||||
// Page 4: nothing left.
|
||||
let page4 = list_report_objects(
|
||||
&report_path,
|
||||
"run_0001",
|
||||
ObjectScope::All,
|
||||
2,
|
||||
page3.next_cursor.as_deref(),
|
||||
)
|
||||
.expect("page4");
|
||||
assert!(page4.data.is_empty());
|
||||
assert_eq!(page4.next_cursor, None);
|
||||
|
||||
// A page larger than the remaining tail must not produce a cursor.
|
||||
let tail = list_report_objects(&report_path, "run_0001", ObjectScope::All, 10, None)
|
||||
.expect("tail");
|
||||
assert_eq!(tail.data.len(), 6);
|
||||
assert_eq!(tail.next_cursor, None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn object_listing_scope_skips_non_matching_publication_points() {
|
||||
let temp = tempfile::tempdir().expect("tempdir");
|
||||
let report_path = temp.path().join("report.json");
|
||||
write_multi_pp_report(&report_path);
|
||||
|
||||
// PP scope: only PP 1 (node_id 2) objects, cursor resumes within PP 1.
|
||||
let page1 = list_report_objects(
|
||||
&report_path,
|
||||
"run_0001",
|
||||
ObjectScope::PublicationPoint("node_2".to_string()),
|
||||
1,
|
||||
None,
|
||||
)
|
||||
.expect("page1");
|
||||
assert_eq!(page1.data.len(), 1);
|
||||
assert_eq!(page1.data[0].uri, "rsync://repo-b.example/rpki/m.mft");
|
||||
assert_eq!(page1.next_cursor.as_deref(), Some("r1:1:1"));
|
||||
let page2 = list_report_objects(
|
||||
&report_path,
|
||||
"run_0001",
|
||||
ObjectScope::PublicationPoint("node_2".to_string()),
|
||||
10,
|
||||
page1.next_cursor.as_deref(),
|
||||
)
|
||||
.expect("page2");
|
||||
assert_eq!(page2.data.len(), 1);
|
||||
assert_eq!(page2.data[0].uri, "rsync://repo-b.example/rpki/b.roa");
|
||||
assert_eq!(page2.next_cursor, None);
|
||||
|
||||
// Repo scope derived from the PP without rrdp uri (repo-b).
|
||||
let probe_pp = list_report_objects(
|
||||
&report_path,
|
||||
"run_0001",
|
||||
ObjectScope::PublicationPoint("node_2".to_string()),
|
||||
10,
|
||||
None,
|
||||
)
|
||||
.expect("probe");
|
||||
let repo_id = probe_pp.data[0].repo_id.clone();
|
||||
let repo_page = list_report_objects(
|
||||
&report_path,
|
||||
"run_0001",
|
||||
ObjectScope::Repo(repo_id),
|
||||
10,
|
||||
None,
|
||||
)
|
||||
.expect("repo page");
|
||||
assert_eq!(repo_page.data.len(), 2);
|
||||
assert!(
|
||||
repo_page
|
||||
.data
|
||||
.iter()
|
||||
.all(|object| object.uri.contains("repo-b.example"))
|
||||
);
|
||||
|
||||
// Unknown scope id: empty page, no cursor, no error.
|
||||
let empty = list_report_objects(
|
||||
&report_path,
|
||||
"run_0001",
|
||||
ObjectScope::PublicationPoint("node_999".to_string()),
|
||||
10,
|
||||
None,
|
||||
)
|
||||
.expect("empty");
|
||||
assert!(empty.data.is_empty());
|
||||
assert_eq!(empty.next_cursor, None);
|
||||
|
||||
// Filtered listing keeps working across skipped PPs.
|
||||
let mut filter = ObjectFilter::default();
|
||||
filter.query = Some("c.roa".to_string());
|
||||
let filtered = list_report_objects_filtered(
|
||||
&report_path,
|
||||
"run_0001",
|
||||
ObjectScope::All,
|
||||
&filter,
|
||||
10,
|
||||
None,
|
||||
)
|
||||
.expect("filtered");
|
||||
assert_eq!(filtered.data.len(), 1);
|
||||
assert_eq!(filtered.data[0].uri, "rsync://repo-c.example/rpki/c.roa");
|
||||
assert_eq!(filtered.next_cursor, None);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -60,6 +60,8 @@ pub enum QueryDbError {
|
||||
MissingColumnFamily(&'static str),
|
||||
#[error("invalid run artifact: {0}")]
|
||||
InvalidArtifact(String),
|
||||
#[error("report data for run {0} is no longer available (retention cleanup)")]
|
||||
RunDataGone(String),
|
||||
#[error("CIR decode error: {0}")]
|
||||
CirDecode(String),
|
||||
}
|
||||
@@ -289,6 +291,7 @@ impl QueryDb {
|
||||
opts.create_if_missing(true);
|
||||
opts.create_missing_column_families(true);
|
||||
opts.set_compression_type(rocksdb::DBCompressionType::Lz4);
|
||||
opts.set_max_open_files(512);
|
||||
let descriptors = QUERY_DB_COLUMN_FAMILIES
|
||||
.iter()
|
||||
.map(|name| ColumnFamilyDescriptor::new(*name, cf_options()))
|
||||
@@ -310,6 +313,7 @@ impl QueryDb {
|
||||
opts.create_if_missing(false);
|
||||
opts.create_missing_column_families(false);
|
||||
opts.set_compression_type(rocksdb::DBCompressionType::Lz4);
|
||||
opts.set_max_open_files(512);
|
||||
let descriptors = QUERY_DB_COLUMN_FAMILIES
|
||||
.iter()
|
||||
.map(|name| ColumnFamilyDescriptor::new(*name, cf_options()))
|
||||
@@ -351,8 +355,10 @@ impl QueryDb {
|
||||
pub fn resolve_run_id(&self, run_id: &str) -> QueryDbResult<Option<String>> {
|
||||
if run_id == "latest" || run_id == "latest_run" {
|
||||
self.latest_ready_run()
|
||||
} else {
|
||||
} else if self.get_run(run_id)?.is_some() {
|
||||
Ok(Some(run_id.to_string()))
|
||||
} else {
|
||||
Ok(None)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -361,7 +367,38 @@ impl QueryDb {
|
||||
limit: usize,
|
||||
cursor: Option<&str>,
|
||||
) -> QueryDbResult<QueryPage<RunRecord>> {
|
||||
self.list_json_by_prefix(CF_RUNS, "run/", limit, cursor)
|
||||
// Runs are listed newest first: zero-padded run ids sort by age, so a
|
||||
// reverse scan over the `run/` prefix yields the latest run on page 1.
|
||||
// Cursor semantics mirror `list_json_by_prefix`: the cursor is the
|
||||
// first key of the next page (inclusive).
|
||||
let limit = limit.clamp(1, 1000);
|
||||
let cf = self.cf(CF_RUNS)?;
|
||||
let prefix = "run/";
|
||||
let start = match cursor {
|
||||
Some(cursor) => cursor.as_bytes().to_vec(),
|
||||
None => prefix_range_end(prefix.as_bytes())
|
||||
.unwrap_or_else(|| prefix.as_bytes().to_vec()),
|
||||
};
|
||||
let mut data = Vec::new();
|
||||
let mut next_cursor = None;
|
||||
let mode = IteratorMode::From(&start, rocksdb::Direction::Reverse);
|
||||
for item in self.db.iterator_cf(cf, mode) {
|
||||
let (key, value) = item?;
|
||||
let key_str = String::from_utf8_lossy(&key);
|
||||
if !key_str.starts_with(prefix) {
|
||||
break;
|
||||
}
|
||||
if data.len() >= limit {
|
||||
next_cursor = Some(key_str.to_string());
|
||||
break;
|
||||
}
|
||||
data.push(serde_json::from_slice(&value)?);
|
||||
}
|
||||
Ok(QueryPage {
|
||||
data,
|
||||
next_cursor,
|
||||
limit,
|
||||
})
|
||||
}
|
||||
|
||||
pub fn list_repos(
|
||||
@@ -691,9 +728,17 @@ impl QueryDb {
|
||||
}
|
||||
|
||||
fn report_path_for_run(&self, run_id: &str) -> QueryDbResult<Option<PathBuf>> {
|
||||
Ok(self
|
||||
.get_run(run_id)?
|
||||
.map(|run| Path::new(&run.run_dir).join("report.json")))
|
||||
let Some(run) = self.get_run(run_id)? else {
|
||||
return Ok(None);
|
||||
};
|
||||
let path = Path::new(&run.run_dir).join("report.json");
|
||||
if !path.exists() {
|
||||
// The run is still indexed but its report.json was removed by
|
||||
// retention cleanup; callers must be able to tell this apart
|
||||
// from a genuine io failure (mapped to 404 upstream).
|
||||
return Err(QueryDbError::RunDataGone(run_id.to_string()));
|
||||
}
|
||||
Ok(Some(path))
|
||||
}
|
||||
|
||||
pub fn get_object_projection(
|
||||
@@ -1094,6 +1139,7 @@ impl QueryDb {
|
||||
fn cf_options() -> Options {
|
||||
let mut opts = Options::default();
|
||||
opts.set_compression_type(rocksdb::DBCompressionType::Lz4);
|
||||
opts.set_max_open_files(512);
|
||||
opts
|
||||
}
|
||||
|
||||
@@ -1951,15 +1997,17 @@ mod tests {
|
||||
db.resolve_run_id("run_0001").unwrap().as_deref(),
|
||||
Some("run_0001")
|
||||
);
|
||||
assert_eq!(db.resolve_run_id("run_9999").unwrap(), None);
|
||||
|
||||
let first_page = db.list_runs(1, None).expect("runs");
|
||||
assert_eq!(first_page.data.len(), 1);
|
||||
assert_eq!(first_page.data[0].run_id, "run_0001");
|
||||
assert_eq!(first_page.data[0].run_id, "run_0002");
|
||||
let second_page = db
|
||||
.list_runs(1, first_page.next_cursor.as_deref())
|
||||
.expect("second runs");
|
||||
assert_eq!(second_page.data.len(), 1);
|
||||
assert_eq!(second_page.data[0].run_id, "run_0002");
|
||||
assert_eq!(second_page.data[0].run_id, "run_0001");
|
||||
assert!(second_page.next_cursor.is_none());
|
||||
|
||||
let repos = db.list_repos("run_0002", 10, None).expect("repos");
|
||||
assert_eq!(repos.data.len(), 1);
|
||||
@@ -2105,6 +2153,43 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn missing_report_file_is_reported_as_run_data_gone() {
|
||||
let temp = tempfile::tempdir().expect("tempdir");
|
||||
let run1 = temp.path().join("runs/run_0001");
|
||||
fs::create_dir_all(&run1).expect("run1");
|
||||
write_sample_run(&run1, "run_0001", 1);
|
||||
|
||||
let query_db_path = temp.path().join("query-db");
|
||||
index_artifacts(&ArtifactIndexerConfig {
|
||||
query_db_path: query_db_path.clone(),
|
||||
run_root: Some(temp.path().to_path_buf()),
|
||||
run_dir: None,
|
||||
repo_bytes_db_path: None,
|
||||
projection_entry_limit: 50,
|
||||
min_run_seq: None,
|
||||
retain_indexed_runs: None,
|
||||
})
|
||||
.expect("index");
|
||||
let db = QueryDb::open(&query_db_path).expect("open query db");
|
||||
assert_eq!(db.list_objects("run_0001", 10, None).unwrap().data.len(), 2);
|
||||
|
||||
// Simulate retention cleanup removing the run directory while the
|
||||
// index record survives: object listing/lookup must fail with a
|
||||
// distinguishable error instead of a bare io error.
|
||||
fs::remove_dir_all(&run1).expect("remove run dir");
|
||||
let err = db.list_objects("run_0001", 10, None).unwrap_err();
|
||||
assert!(
|
||||
matches!(err, QueryDbError::RunDataGone(ref run_id) if run_id == "run_0001"),
|
||||
"unexpected error: {err}"
|
||||
);
|
||||
let err = db.get_object_by_sha256("run_0001", "22").unwrap_err();
|
||||
assert!(
|
||||
matches!(err, QueryDbError::RunDataGone(_)),
|
||||
"unexpected error: {err}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn repeated_indexing_does_not_move_latest_backwards() {
|
||||
let temp = tempfile::tempdir().expect("tempdir");
|
||||
|
||||
Reference in New Issue
Block a user