NASP/rpki
84
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

20260326 完成数据库model迁移；20260327 增加一键replay脚本

Browse Source
This commit is contained in:
yuyr
2026-03-27 11:24:34 +08:00
parent fe8b89d829
commit cd0ba15286
28 changed files with 1956 additions and 930 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
src/audit_trace.rs
View File

@@ -384,7 +384,7 @@ mod tests {
]
}
fn put_raw(store: &RocksStore, bytes: &[u8], uri: &str, object_type: &str) {
fn put_raw_evidence(store: &RocksStore, bytes: &[u8], uri: &str, object_type: &str) {
let mut entry = RawByHashEntry::from_bytes(sha256_hex(bytes), bytes.to_vec());
entry.origin_uris.push(uri.to_string());
entry.object_type = Some(object_type.to_string());
@@ -432,17 +432,17 @@ mod tests {
.put_audit_rule_index_entry(&rule_entry)
.expect("put rule index");
put_raw(&store, leaf_manifest.as_bytes(), leaf_manifest, "mft");
put_raw(
put_raw_evidence(&store, leaf_manifest.as_bytes(), leaf_manifest, "mft");
put_raw_evidence(
&store,
format!("{}-crl", leaf_manifest).as_bytes(),
&leaf_manifest.replace(".mft", ".crl"),
"crl",
);
put_raw(&store, b"roa-raw", &local.source_object_uri, "roa");
put_raw(&store, b"roa-ee", "rsync://example.test/leaf/a.ee", "cer");
put_raw(&store, root_manifest.as_bytes(), root_manifest, "mft");
put_raw(
put_raw_evidence(&store, b"roa-raw", &local.source_object_uri, "roa");
put_raw_evidence(&store, b"roa-ee", "rsync://example.test/leaf/a.ee", "cer");
put_raw_evidence(&store, root_manifest.as_bytes(), root_manifest, "mft");
put_raw_evidence(
&store,
format!("{}-crl", root_manifest).as_bytes(),
&root_manifest.replace(".mft", ".crl"),

22
src/bin/db_stats.rs
View File

@@ -3,9 +3,9 @@ use std::path::PathBuf;
use rocksdb::{DB, IteratorMode, Options};
use rpki::storage::{
ALL_COLUMN_FAMILY_NAMES, CF_AUDIT_RULE_INDEX, CF_RAW_BY_HASH, CF_RAW_OBJECTS,
CF_REPOSITORY_VIEW, CF_RRDP_OBJECT_INDEX, CF_RRDP_SOURCE, CF_RRDP_SOURCE_MEMBER, CF_RRDP_STATE,
CF_RRDP_URI_OWNER, CF_VCIR, column_family_descriptors,
ALL_COLUMN_FAMILY_NAMES, CF_AUDIT_RULE_INDEX, CF_RAW_BY_HASH, CF_REPOSITORY_VIEW,
CF_RRDP_SOURCE, CF_RRDP_SOURCE_MEMBER, CF_RRDP_URI_OWNER, CF_VCIR,
column_family_descriptors,
};
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -48,8 +48,7 @@ Options:
Output groups:
- current_repository_view: repository_view + raw_by_hash
- current_validation_state: vcir + audit_rule_index
- current_rrdp_state: rrdp_state + rrdp_source + rrdp_source_member + rrdp_uri_owner
- legacy_compatibility: raw_objects + rrdp_object_index
- current_rrdp_state: rrdp_source + rrdp_source_member + rrdp_uri_owner
"
)
}
@@ -78,10 +77,9 @@ fn cf_group(cf_name: &str) -> CfGroup {
match cf_name {
CF_REPOSITORY_VIEW | CF_RAW_BY_HASH => CfGroup::CurrentRepositoryView,
CF_VCIR | CF_AUDIT_RULE_INDEX => CfGroup::CurrentValidationState,
CF_RRDP_STATE | CF_RRDP_SOURCE | CF_RRDP_SOURCE_MEMBER | CF_RRDP_URI_OWNER => {
CF_RRDP_SOURCE | CF_RRDP_SOURCE_MEMBER | CF_RRDP_URI_OWNER => {
CfGroup::CurrentRrdpState
}
CF_RAW_OBJECTS | CF_RRDP_OBJECT_INDEX => CfGroup::LegacyCompatibility,
_ => CfGroup::LegacyCompatibility,
}
}
@@ -175,7 +173,7 @@ mod tests {
);
assert_eq!(cf_group(CF_RRDP_SOURCE), CfGroup::CurrentRrdpState);
assert_eq!(cf_group(CF_RRDP_URI_OWNER), CfGroup::CurrentRrdpState);
assert_eq!(cf_group(CF_RAW_OBJECTS), CfGroup::LegacyCompatibility);
assert_eq!(cf_group("unknown_legacy"), CfGroup::LegacyCompatibility);
}
#[test]
@@ -185,15 +183,13 @@ mod tests {
(CF_RAW_BY_HASH, 7),
(CF_VCIR, 11),
(CF_AUDIT_RULE_INDEX, 13),
(CF_RRDP_STATE, 17),
(CF_RRDP_SOURCE_MEMBER, 19),
(CF_RRDP_OBJECT_INDEX, 29),
]);
assert_eq!(grouped.get(&CfGroup::CurrentRepositoryView), Some(&12));
assert_eq!(grouped.get(&CfGroup::CurrentValidationState), Some(&24));
assert_eq!(grouped.get(&CfGroup::CurrentRrdpState), Some(&36));
assert_eq!(grouped.get(&CfGroup::LegacyCompatibility), Some(&29));
assert_eq!(grouped.get(&CfGroup::CurrentRrdpState), Some(&19));
assert_eq!(grouped.get(&CfGroup::LegacyCompatibility), None);
}
#[test]
@@ -201,6 +197,6 @@ mod tests {
let text = usage();
assert!(text.contains("--exact"), "{text}");
assert!(text.contains("current_validation_state"), "{text}");
assert!(text.contains("legacy_compatibility"), "{text}");
assert!(text.contains("current_rrdp_state"), "{text}");
}
}

53
src/bin/rrdp_state_dump.rs
View File

@@ -2,14 +2,12 @@ use std::path::PathBuf;
use rocksdb::{DB, IteratorMode, Options};
use rpki::storage::{
CF_RRDP_SOURCE, CF_RRDP_SOURCE_MEMBER, CF_RRDP_STATE, CF_RRDP_URI_OWNER,
RrdpSourceMemberRecord, RrdpSourceRecord, RrdpUriOwnerRecord, column_family_descriptors,
CF_RRDP_SOURCE, CF_RRDP_SOURCE_MEMBER, CF_RRDP_URI_OWNER, RrdpSourceMemberRecord,
RrdpSourceRecord, RrdpUriOwnerRecord, column_family_descriptors,
};
use rpki::sync::rrdp::RrdpState;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum DumpView {
LegacyState,
Source,
Members,
Owners,
@@ -19,13 +17,12 @@ enum DumpView {
impl DumpView {
fn parse(value: &str) -> Result<Self, String> {
match value {
"legacy-state" => Ok(Self::LegacyState),
"source" => Ok(Self::Source),
"members" => Ok(Self::Members),
"owners" => Ok(Self::Owners),
"all" => Ok(Self::All),
other => Err(format!(
"invalid --view: {other} (expected one of: legacy-state, source, members, owners, all)"
"invalid --view: {other} (expected one of: source, members, owners, all)"
)),
}
}
@@ -42,7 +39,7 @@ fn usage() -> String {
format!(
"\
Usage:
{bin} --db <path> [--view <legacy-state|source|members|owners|all>]
{bin} --db <path> [--view <source|members|owners|all>]
Options:
--db <path> RocksDB directory
@@ -94,22 +91,6 @@ fn open_db(path: &std::path::Path) -> Result<DB, Box<dyn std::error::Error>> {
)?)
}
fn collect_legacy_state(db: &DB) -> Result<Vec<(String, RrdpState)>, Box<dyn std::error::Error>> {
let cf = db
.cf_handle(CF_RRDP_STATE)
.ok_or("missing column family: rrdp_state")?;
let mut out = Vec::new();
for res in db.iterator_cf(cf, IteratorMode::Start) {
let (k, v) = res?;
let notify_uri = String::from_utf8_lossy(&k).to_string();
let state = RrdpState::decode(&v)
.map_err(|e| format!("decode rrdp_state failed for {notify_uri}: {e}"))?;
out.push((notify_uri, state));
}
out.sort_by(|a, b| a.0.cmp(&b.0));
Ok(out)
}
fn collect_source_records(db: &DB) -> Result<Vec<RrdpSourceRecord>, Box<dyn std::error::Error>> {
let cf = db
.cf_handle(CF_RRDP_SOURCE)
@@ -163,14 +144,6 @@ fn collect_uri_owner_records(
Ok(out)
}
fn print_legacy_state(entries: &[(String, RrdpState)]) {
println!("[legacy-state]");
println!("notify_uri\tserial\tsession_id");
for (notify_uri, state) in entries {
println!("{notify_uri}\t{}\t{}", state.serial, state.session_id);
}
}
fn print_source_records(entries: &[RrdpSourceRecord]) {
println!("[source]");
println!("notify_uri\tlast_serial\tlast_session_id\tsync_state\tlast_snapshot_uri\tlast_error");
@@ -228,12 +201,10 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
let db = open_db(&args.db_path)?;
match args.view {
DumpView::LegacyState => print_legacy_state(&collect_legacy_state(&db)?),
DumpView::Source => print_source_records(&collect_source_records(&db)?),
DumpView::Members => print_source_member_records(&collect_source_member_records(&db)?),
DumpView::Owners => print_uri_owner_records(&collect_uri_owner_records(&db)?),
DumpView::All => {
print_legacy_state(&collect_legacy_state(&db)?);
print_source_records(&collect_source_records(&db)?);
print_source_member_records(&collect_source_member_records(&db)?);
print_uri_owner_records(&collect_uri_owner_records(&db)?);
@@ -275,19 +246,9 @@ mod tests {
}
#[test]
fn collect_rrdp_views_reads_legacy_and_current_records() {
fn collect_rrdp_views_reads_current_records() {
let dir = tempfile::tempdir().expect("tempdir");
let store = RocksStore::open(dir.path()).expect("open store");
let legacy_state = RrdpState {
session_id: "session-1".to_string(),
serial: 42,
};
store
.put_rrdp_state(
"https://example.test/notify.xml",
&legacy_state.encode().expect("encode legacy state"),
)
.expect("put legacy state");
store
.put_rrdp_source_record(&RrdpSourceRecord {
notify_uri: "https://example.test/notify.xml".to_string(),
@@ -328,14 +289,10 @@ mod tests {
drop(store);
let db = open_db(dir.path()).expect("open db");
let legacy = collect_legacy_state(&db).expect("legacy dump");
let sources = collect_source_records(&db).expect("source dump");
let members = collect_source_member_records(&db).expect("members dump");
let owners = collect_uri_owner_records(&db).expect("owners dump");
assert_eq!(legacy.len(), 1);
assert_eq!(legacy[0].0, "https://example.test/notify.xml");
assert_eq!(legacy[0].1.serial, 42);
assert_eq!(sources.len(), 1);
assert_eq!(sources[0].sync_state, RrdpSourceSyncState::DeltaReady);
assert_eq!(members.len(), 1);

628
src/storage.rs
View File

@@ -10,9 +10,6 @@ use sha2::Digest;
use crate::data_model::rc::{AsResourceSet, IpResourceSet};
pub const CF_RAW_OBJECTS: &str = "raw_objects";
pub const CF_RRDP_STATE: &str = "rrdp_state";
pub const CF_RRDP_OBJECT_INDEX: &str = "rrdp_object_index";
pub const CF_REPOSITORY_VIEW: &str = "repository_view";
pub const CF_RAW_BY_HASH: &str = "raw_by_hash";
pub const CF_VCIR: &str = "vcir";
@@ -22,9 +19,6 @@ pub const CF_RRDP_SOURCE_MEMBER: &str = "rrdp_source_member";
pub const CF_RRDP_URI_OWNER: &str = "rrdp_uri_owner";
pub const ALL_COLUMN_FAMILY_NAMES: &[&str] = &[
CF_RAW_OBJECTS,
CF_RRDP_STATE,
CF_RRDP_OBJECT_INDEX,
CF_REPOSITORY_VIEW,
CF_RAW_BY_HASH,
CF_VCIR,
@@ -34,7 +28,6 @@ pub const ALL_COLUMN_FAMILY_NAMES: &[&str] = &[
CF_RRDP_URI_OWNER,
];
const RRDP_OBJECT_INDEX_PREFIX: &[u8] = b"rrdp_obj:";
const REPOSITORY_VIEW_KEY_PREFIX: &str = "repo_view:";
const RAW_BY_HASH_KEY_PREFIX: &str = "rawbyhash:";
const VCIR_KEY_PREFIX: &str = "vcir:";
@@ -785,224 +778,6 @@ impl RocksStore {
.ok_or(StorageError::MissingColumnFamily(name))
}
pub fn put_raw(&self, rsync_uri: &str, bytes: &[u8]) -> StorageResult<()> {
let cf = self.cf(CF_RAW_OBJECTS)?;
self.db
.put_cf(cf, rsync_uri.as_bytes(), bytes)
.map_err(|e| StorageError::RocksDb(e.to_string()))?;
Ok(())
}
pub fn put_raw_batch(&self, objects: Vec<(String, Vec<u8>)>) -> StorageResult<usize> {
if objects.is_empty() {
return Ok(0);
}
let cf = self.cf(CF_RAW_OBJECTS)?;
let mut batch = WriteBatch::default();
for (rsync_uri, bytes) in &objects {
batch.put_cf(cf, rsync_uri.as_bytes(), bytes.as_slice());
}
self.write_batch(batch)?;
Ok(objects.len())
}
pub fn get_raw(&self, rsync_uri: &str) -> StorageResult<Option<Vec<u8>>> {
let cf = self.cf(CF_RAW_OBJECTS)?;
let v = self
.db
.get_cf(cf, rsync_uri.as_bytes())
.map_err(|e| StorageError::RocksDb(e.to_string()))?;
Ok(v)
}
pub fn delete_raw(&self, rsync_uri: &str) -> StorageResult<()> {
let cf = self.cf(CF_RAW_OBJECTS)?;
self.db
.delete_cf(cf, rsync_uri.as_bytes())
.map_err(|e| StorageError::RocksDb(e.to_string()))?;
Ok(())
}
pub fn put_rrdp_state(&self, notification_uri: &str, bytes: &[u8]) -> StorageResult<()> {
let cf = self.cf(CF_RRDP_STATE)?;
self.db
.put_cf(cf, notification_uri.as_bytes(), bytes)
.map_err(|e| StorageError::RocksDb(e.to_string()))?;
Ok(())
}
pub fn get_rrdp_state(&self, notification_uri: &str) -> StorageResult<Option<Vec<u8>>> {
let cf = self.cf(CF_RRDP_STATE)?;
let v = self
.db
.get_cf(cf, notification_uri.as_bytes())
.map_err(|e| StorageError::RocksDb(e.to_string()))?;
Ok(v)
}
#[allow(dead_code)]
pub fn delete_rrdp_state(&self, notification_uri: &str) -> StorageResult<()> {
let cf = self.cf(CF_RRDP_STATE)?;
self.db
.delete_cf(cf, notification_uri.as_bytes())
.map_err(|e| StorageError::RocksDb(e.to_string()))?;
Ok(())
}
fn rrdp_object_index_key(notification_uri: &str, rsync_uri: &str) -> Vec<u8> {
let mut out = Vec::with_capacity(
RRDP_OBJECT_INDEX_PREFIX.len() + notification_uri.len() + 1 + rsync_uri.len(),
);
out.extend_from_slice(RRDP_OBJECT_INDEX_PREFIX);
out.extend_from_slice(notification_uri.as_bytes());
out.push(0);
out.extend_from_slice(rsync_uri.as_bytes());
out
}
fn rrdp_object_index_prefix(notification_uri: &str) -> Vec<u8> {
let mut out =
Vec::with_capacity(RRDP_OBJECT_INDEX_PREFIX.len() + notification_uri.len() + 1);
out.extend_from_slice(RRDP_OBJECT_INDEX_PREFIX);
out.extend_from_slice(notification_uri.as_bytes());
out.push(0);
out
}
#[allow(dead_code)]
pub fn rrdp_object_index_contains(
&self,
notification_uri: &str,
rsync_uri: &str,
) -> StorageResult<bool> {
let cf = self.cf(CF_RRDP_OBJECT_INDEX)?;
let k = Self::rrdp_object_index_key(notification_uri, rsync_uri);
Ok(self
.db
.get_cf(cf, k)
.map_err(|e| StorageError::RocksDb(e.to_string()))?
.is_some())
}
#[allow(dead_code)]
pub fn rrdp_object_index_iter(
&self,
notification_uri: &str,
) -> StorageResult<impl Iterator<Item = String> + '_> {
let cf = self.cf(CF_RRDP_OBJECT_INDEX)?;
let prefix = Self::rrdp_object_index_prefix(notification_uri);
let prefix_len = prefix.len();
let mode = IteratorMode::From(prefix.as_slice(), Direction::Forward);
Ok(self
.db
.iterator_cf(cf, mode)
.take_while(move |res| match res {
Ok((k, _v)) => k.starts_with(prefix.as_slice()),
Err(_) => false,
})
.filter_map(move |res| {
let (k, _v) = res.ok()?;
let rsync_part = k.get(prefix_len..)?;
let s = std::str::from_utf8(rsync_part).ok()?;
Some(s.to_string())
}))
}
#[allow(dead_code)]
pub fn rrdp_object_index_clear(&self, notification_uri: &str) -> StorageResult<usize> {
let cf = self.cf(CF_RRDP_OBJECT_INDEX)?;
let prefix = Self::rrdp_object_index_prefix(notification_uri);
let mode = IteratorMode::From(prefix.as_slice(), Direction::Forward);
let keys: Vec<Box<[u8]>> = self
.db
.iterator_cf(cf, mode)
.take_while(|res| match res {
Ok((k, _v)) => k.starts_with(prefix.as_slice()),
Err(_) => false,
})
.filter_map(|res| res.ok().map(|(k, _v)| k))
.collect();
if keys.is_empty() {
return Ok(0);
}
let mut batch = WriteBatch::default();
for k in &keys {
batch.delete_cf(cf, k);
}
self.write_batch(batch)?;
Ok(keys.len())
}
pub fn apply_rrdp_snapshot(
&self,
notification_uri: &str,
published: &[(String, Vec<u8>)],
) -> StorageResult<usize> {
let raw_cf = self.cf(CF_RAW_OBJECTS)?;
let idx_cf = self.cf(CF_RRDP_OBJECT_INDEX)?;
let mut new_set: HashSet<&str> = HashSet::with_capacity(published.len());
for (u, _b) in published {
new_set.insert(u.as_str());
}
let old_uris: Vec<String> = self.rrdp_object_index_iter(notification_uri)?.collect();
let mut batch = WriteBatch::default();
for old in &old_uris {
if !new_set.contains(old.as_str()) {
batch.delete_cf(raw_cf, old.as_bytes());
let k = Self::rrdp_object_index_key(notification_uri, old.as_str());
batch.delete_cf(idx_cf, k);
}
}
for (uri, bytes) in published {
batch.put_cf(raw_cf, uri.as_bytes(), bytes.as_slice());
let k = Self::rrdp_object_index_key(notification_uri, uri.as_str());
batch.put_cf(idx_cf, k, b"");
}
self.write_batch(batch)?;
Ok(published.len())
}
pub fn apply_rrdp_delta(
&self,
notification_uri: &str,
ops: &[RrdpDeltaOp],
) -> StorageResult<usize> {
if ops.is_empty() {
return Ok(0);
}
let raw_cf = self.cf(CF_RAW_OBJECTS)?;
let idx_cf = self.cf(CF_RRDP_OBJECT_INDEX)?;
let mut batch = WriteBatch::default();
for op in ops {
match op {
RrdpDeltaOp::Upsert { rsync_uri, bytes } => {
batch.put_cf(raw_cf, rsync_uri.as_bytes(), bytes.as_slice());
let k = Self::rrdp_object_index_key(notification_uri, rsync_uri.as_str());
batch.put_cf(idx_cf, k, b"");
}
RrdpDeltaOp::Delete { rsync_uri } => {
batch.delete_cf(raw_cf, rsync_uri.as_bytes());
let k = Self::rrdp_object_index_key(notification_uri, rsync_uri.as_str());
batch.delete_cf(idx_cf, k);
}
}
}
self.write_batch(batch)?;
Ok(ops.len())
}
pub fn put_repository_view_entry(&self, entry: &RepositoryViewEntry) -> StorageResult<()> {
entry.validate_internal()?;
let cf = self.cf(CF_REPOSITORY_VIEW)?;
@@ -1145,6 +920,16 @@ impl RocksStore {
self.write_batch(batch)
}
pub fn delete_raw_by_hash_entry(&self, sha256_hex: &str) -> StorageResult<()> {
validate_sha256_hex("raw_by_hash.sha256_hex", sha256_hex)?;
let cf = self.cf(CF_RAW_BY_HASH)?;
let key = raw_by_hash_key(sha256_hex);
self.db
.delete_cf(cf, key.as_bytes())
.map_err(|e| StorageError::RocksDb(e.to_string()))?;
Ok(())
}
pub fn get_raw_by_hash_entry(&self, sha256_hex: &str) -> StorageResult<Option<RawByHashEntry>> {
let cf = self.cf(CF_RAW_BY_HASH)?;
let key = raw_by_hash_key(sha256_hex);
@@ -1411,6 +1196,55 @@ impl RocksStore {
.collect()
}
pub fn list_current_rrdp_source_members(
&self,
notify_uri: &str,
) -> StorageResult<Vec<RrdpSourceMemberRecord>> {
let mut records = self.list_rrdp_source_member_records(notify_uri)?;
records.retain(|record| record.present);
records.sort_by(|a, b| a.rsync_uri.cmp(&b.rsync_uri));
Ok(records)
}
pub fn is_current_rrdp_source_member(
&self,
notify_uri: &str,
rsync_uri: &str,
) -> StorageResult<bool> {
Ok(matches!(
self.get_rrdp_source_member_record(notify_uri, rsync_uri)?,
Some(record) if record.present
))
}
pub fn load_current_object_bytes_by_uri(&self, rsync_uri: &str) -> StorageResult<Option<Vec<u8>>> {
let Some(view) = self.get_repository_view_entry(rsync_uri)? else {
return Ok(None);
};
match view.state {
RepositoryViewState::Withdrawn => Ok(None),
RepositoryViewState::Present | RepositoryViewState::Replaced => {
let hash = view
.current_hash
.as_deref()
.ok_or(StorageError::InvalidData {
entity: "repository_view",
detail: format!(
"current_hash missing for current object URI: {rsync_uri}"
),
})?;
let raw = self.get_raw_by_hash_entry(hash)?.ok_or(StorageError::InvalidData {
entity: "repository_view",
detail: format!(
"raw_by_hash entry missing for current object URI: {rsync_uri} (hash={hash})"
),
})?;
Ok(Some(raw.bytes))
}
}
}
pub fn put_rrdp_uri_owner_record(&self, record: &RrdpUriOwnerRecord) -> StorageResult<()> {
record.validate_internal()?;
let cf = self.cf(CF_RRDP_URI_OWNER)?;
@@ -1449,31 +1283,6 @@ impl RocksStore {
Ok(())
}
#[allow(dead_code)]
pub fn raw_iter_prefix<'a>(
&'a self,
prefix: &'a [u8],
) -> StorageResult<impl Iterator<Item = (Box<[u8]>, Box<[u8]>)> + 'a> {
let cf = self.cf(CF_RAW_OBJECTS)?;
let mode = IteratorMode::From(prefix, Direction::Forward);
Ok(self
.db
.iterator_cf(cf, mode)
.take_while(move |res| match res {
Ok((k, _v)) => k.starts_with(prefix),
Err(_) => false,
})
.filter_map(|res| res.ok()))
}
#[allow(dead_code)]
pub fn raw_iter_all<'a>(
&'a self,
) -> StorageResult<impl Iterator<Item = (Box<[u8]>, Box<[u8]>)> + 'a> {
let cf = self.cf(CF_RAW_OBJECTS)?;
let mode = IteratorMode::Start;
Ok(self.db.iterator_cf(cf, mode).filter_map(|res| res.ok()))
}
#[allow(dead_code)]
@@ -1865,121 +1674,6 @@ mod tests {
}
}
#[test]
fn rrdp_object_index_and_snapshot_delta_helpers_work_end_to_end() {
let td = tempfile::tempdir().expect("tempdir");
let store = RocksStore::open(td.path()).expect("open rocksdb");
let notification_uri = "https://rrdp.example.test/notification.xml";
let u1 = "rsync://rpki.example.test/repo/obj1.cer".to_string();
let u2 = "rsync://rpki.example.test/repo/obj2.mft".to_string();
assert_eq!(
store
.rrdp_object_index_clear(notification_uri)
.expect("clear empty"),
0
);
let published_v1 = vec![(u1.clone(), vec![1u8, 2, 3]), (u2.clone(), vec![9u8, 8, 7])];
let n = store
.apply_rrdp_snapshot(notification_uri, &published_v1)
.expect("apply snapshot v1");
assert_eq!(n, 2);
assert_eq!(
store.get_raw(&u1).expect("get_raw u1"),
Some(vec![1u8, 2, 3])
);
assert_eq!(
store.get_raw(&u2).expect("get_raw u2"),
Some(vec![9u8, 8, 7])
);
assert!(
store
.rrdp_object_index_contains(notification_uri, &u1)
.expect("contains u1")
);
assert!(
store
.rrdp_object_index_contains(notification_uri, &u2)
.expect("contains u2")
);
let mut listed: Vec<String> = store
.rrdp_object_index_iter(notification_uri)
.expect("iter")
.collect();
listed.sort();
assert_eq!(listed, vec![u1.clone(), u2.clone()]);
let published_v2 = vec![(u2.clone(), vec![0u8, 1, 2, 3])];
store
.apply_rrdp_snapshot(notification_uri, &published_v2)
.expect("apply snapshot v2");
assert_eq!(store.get_raw(&u1).expect("get_raw removed"), None);
assert_eq!(
store.get_raw(&u2).expect("get_raw updated"),
Some(vec![0u8, 1, 2, 3])
);
let u3 = "rsync://rpki.example.test/repo/obj3.crl".to_string();
let ops = vec![
RrdpDeltaOp::Upsert {
rsync_uri: u3.clone(),
bytes: vec![4u8, 5, 6],
},
RrdpDeltaOp::Delete {
rsync_uri: u2.clone(),
},
];
let applied = store
.apply_rrdp_delta(notification_uri, &ops)
.expect("apply delta");
assert_eq!(applied, 2);
assert_eq!(store.get_raw(&u2).expect("get_raw deleted"), None);
assert_eq!(
store.get_raw(&u3).expect("get_raw u3"),
Some(vec![4u8, 5, 6])
);
let prefix = b"rsync://rpki.example.test/repo/";
let mut got: Vec<String> = store
.raw_iter_prefix(prefix)
.expect("raw_iter_prefix")
.map(|(k, _v)| String::from_utf8(k.to_vec()).expect("utf8 key"))
.collect();
got.sort();
assert_eq!(got, vec![u3.clone()]);
let all: Vec<String> = store
.raw_iter_all()
.expect("raw_iter_all")
.map(|(k, _v)| String::from_utf8(k.to_vec()).expect("utf8 key"))
.collect();
assert!(all.contains(&u3));
let cleared = store
.rrdp_object_index_clear(notification_uri)
.expect("clear");
assert!(cleared >= 1);
assert!(
!store
.rrdp_object_index_contains(notification_uri, &u3)
.expect("contains after clear")
);
}
#[test]
fn apply_rrdp_delta_empty_ops_is_noop() {
let td = tempfile::tempdir().expect("tempdir");
let store = RocksStore::open(td.path()).expect("open rocksdb");
let n = store
.apply_rrdp_delta("https://rrdp.example.test/notification.xml", &[])
.expect("apply empty delta");
assert_eq!(n, 0);
}
#[test]
fn repository_view_and_raw_by_hash_roundtrip() {
let td = tempfile::tempdir().expect("tempdir");
@@ -2271,47 +1965,6 @@ mod tests {
.validate_internal()
.expect_err("duplicate origin URI must fail");
assert!(err.to_string().contains("duplicate origin URI"));
store
.put_raw_batch(vec![(
"rsync://example.test/repo/raw.cer".to_string(),
vec![1, 2, 3],
)])
.expect("put_raw_batch stores entries");
assert_eq!(
store
.get_raw("rsync://example.test/repo/raw.cer")
.expect("get raw"),
Some(vec![1, 2, 3])
);
store
.delete_raw("rsync://example.test/repo/raw.cer")
.expect("delete raw entry");
assert!(
store
.get_raw("rsync://example.test/repo/raw.cer")
.expect("get deleted raw")
.is_none()
);
store
.put_rrdp_state("https://rrdp.example.test/notification.xml", b"state")
.expect("put rrdp state");
assert_eq!(
store
.get_rrdp_state("https://rrdp.example.test/notification.xml")
.expect("get rrdp state"),
Some(b"state".to_vec())
);
store
.delete_rrdp_state("https://rrdp.example.test/notification.xml")
.expect("delete rrdp state");
assert!(
store
.get_rrdp_state("https://rrdp.example.test/notification.xml")
.expect("get deleted rrdp state")
.is_none()
);
}
#[test]
@@ -2494,4 +2147,167 @@ mod tests {
);
}
#[test]
fn current_rrdp_source_member_helpers_filter_present_records() {
let td = tempfile::tempdir().expect("tempdir");
let store = RocksStore::open(td.path()).expect("open rocksdb");
let notify_uri = "https://rrdp.example.test/notification.xml";
let mut present_a =
sample_rrdp_source_member_record(notify_uri, "rsync://example.test/repo/a.cer", 1);
let mut withdrawn_b =
sample_rrdp_source_member_record(notify_uri, "rsync://example.test/repo/b.roa", 2);
withdrawn_b.present = false;
let present_c =
sample_rrdp_source_member_record(notify_uri, "rsync://example.test/repo/c.crl", 3);
let other_source = sample_rrdp_source_member_record(
"https://other.example.test/notification.xml",
"rsync://other.example.test/repo/x.cer",
4,
);
present_a.last_confirmed_serial = 10;
store
.put_rrdp_source_member_record(&present_a)
.expect("put present a");
store
.put_rrdp_source_member_record(&withdrawn_b)
.expect("put withdrawn b");
store
.put_rrdp_source_member_record(&present_c)
.expect("put present c");
store
.put_rrdp_source_member_record(&other_source)
.expect("put other source");
let members = store
.list_current_rrdp_source_members(notify_uri)
.expect("list current members");
assert_eq!(
members
.iter()
.map(|record| record.rsync_uri.as_str())
.collect::<Vec<_>>(),
vec![
"rsync://example.test/repo/a.cer",
"rsync://example.test/repo/c.crl",
]
);
assert!(
store
.is_current_rrdp_source_member(notify_uri, &present_a.rsync_uri)
.expect("current a")
);
assert!(
!store
.is_current_rrdp_source_member(notify_uri, &withdrawn_b.rsync_uri)
.expect("withdrawn b")
);
assert!(
!store
.is_current_rrdp_source_member(notify_uri, &other_source.rsync_uri)
.expect("other source")
);
}
#[test]
fn load_current_object_bytes_by_uri_uses_repository_view_and_raw_by_hash() {
let td = tempfile::tempdir().expect("tempdir");
let store = RocksStore::open(td.path()).expect("open rocksdb");
let present_bytes = b"present-object".to_vec();
let present_hash = sha256_hex(&present_bytes);
let mut present_raw = RawByHashEntry::from_bytes(present_hash.clone(), present_bytes.clone());
present_raw.origin_uris.push("rsync://example.test/repo/present.roa".to_string());
present_raw.object_type = Some("roa".to_string());
store
.put_raw_by_hash_entry(&present_raw)
.expect("put present raw");
store
.put_repository_view_entry(&RepositoryViewEntry {
rsync_uri: "rsync://example.test/repo/present.roa".to_string(),
current_hash: Some(present_hash),
repository_source: Some("https://rrdp.example.test/notification.xml".to_string()),
object_type: Some("roa".to_string()),
state: RepositoryViewState::Present,
})
.expect("put present view");
let replaced_bytes = b"replaced-object".to_vec();
let replaced_hash = sha256_hex(&replaced_bytes);
let mut replaced_raw =
RawByHashEntry::from_bytes(replaced_hash.clone(), replaced_bytes.clone());
replaced_raw.origin_uris.push("rsync://example.test/repo/replaced.cer".to_string());
replaced_raw.object_type = Some("cer".to_string());
store
.put_raw_by_hash_entry(&replaced_raw)
.expect("put replaced raw");
store
.put_repository_view_entry(&RepositoryViewEntry {
rsync_uri: "rsync://example.test/repo/replaced.cer".to_string(),
current_hash: Some(replaced_hash),
repository_source: Some("https://rrdp.example.test/notification.xml".to_string()),
object_type: Some("cer".to_string()),
state: RepositoryViewState::Replaced,
})
.expect("put replaced view");
store
.put_repository_view_entry(&RepositoryViewEntry {
rsync_uri: "rsync://example.test/repo/withdrawn.crl".to_string(),
current_hash: Some(sha256_hex(b"withdrawn")),
repository_source: Some("https://rrdp.example.test/notification.xml".to_string()),
object_type: Some("crl".to_string()),
state: RepositoryViewState::Withdrawn,
})
.expect("put withdrawn view");
assert_eq!(
store
.load_current_object_bytes_by_uri("rsync://example.test/repo/present.roa")
.expect("load present"),
Some(present_bytes)
);
assert_eq!(
store
.load_current_object_bytes_by_uri("rsync://example.test/repo/replaced.cer")
.expect("load replaced"),
Some(replaced_bytes)
);
assert_eq!(
store
.load_current_object_bytes_by_uri("rsync://example.test/repo/withdrawn.crl")
.expect("load withdrawn"),
None
);
assert_eq!(
store
.load_current_object_bytes_by_uri("rsync://example.test/repo/missing.roa")
.expect("load missing"),
None
);
}
#[test]
fn load_current_object_bytes_by_uri_errors_when_raw_by_hash_is_missing() {
let td = tempfile::tempdir().expect("tempdir");
let store = RocksStore::open(td.path()).expect("open rocksdb");
let rsync_uri = "rsync://example.test/repo/missing.cer";
store
.put_repository_view_entry(&RepositoryViewEntry {
rsync_uri: rsync_uri.to_string(),
current_hash: Some(hex::encode([0x11; 32])),
repository_source: Some("https://rrdp.example.test/notification.xml".to_string()),
object_type: Some("cer".to_string()),
state: RepositoryViewState::Present,
})
.expect("put view");
let err = store
.load_current_object_bytes_by_uri(rsync_uri)
.expect_err("missing raw_by_hash should error");
assert!(matches!(err, StorageError::InvalidData { .. }));
}
}

179
src/sync/repo.rs
View File

@@ -8,7 +8,7 @@ use crate::replay::delta_archive::{ReplayDeltaArchiveIndex, ReplayDeltaRrdpKind}
use crate::report::{RfcRef, Warning};
use crate::storage::RocksStore;
use crate::sync::rrdp::sync_from_notification_with_timing_and_download_log;
use crate::sync::rrdp::{Fetcher as HttpFetcher, RrdpState, RrdpSyncError};
use crate::sync::rrdp::{Fetcher as HttpFetcher, RrdpSyncError, load_rrdp_local_state};
use crate::sync::store_projection::{
build_repository_view_present_entry,
build_repository_view_withdrawn_entry,
@@ -18,6 +18,11 @@ use std::collections::HashSet;
use std::thread;
use std::time::Duration;
#[cfg(test)]
use crate::storage::RrdpSourceSyncState;
#[cfg(test)]
use crate::sync::rrdp::persist_rrdp_local_state;
const RRDP_RETRY_BACKOFFS_PROD: [Duration; 3] = [
Duration::from_millis(200),
Duration::from_millis(500),
@@ -54,7 +59,7 @@ pub enum RepoSyncError {
Storage(String),
}
/// Sync a publication point into `raw_objects`.
/// Sync a publication point into the current repository view.
///
/// v1 behavior:
/// - If `rrdp_notification_uri` is present and `policy.sync_preference` is `rrdp_then_rsync`,
@@ -100,7 +105,7 @@ pub fn sync_publication_point(
.with_rfc_refs(&[RfcRef("RFC 8182 §3.4.5")])
.with_context(notification_uri),
];
let written = rsync_sync_into_raw_objects(
let written = rsync_sync_into_current_store(
store,
rsync_base_uri,
rsync_fetcher,
@@ -120,7 +125,7 @@ pub fn sync_publication_point(
}
}
_ => {
let written = rsync_sync_into_raw_objects(
let written = rsync_sync_into_current_store(
store,
rsync_base_uri,
rsync_fetcher,
@@ -170,7 +175,7 @@ pub fn sync_publication_point_replay(
})
}
ReplayResolvedTransport::Rsync => {
let written = rsync_sync_into_raw_objects(
let written = rsync_sync_into_current_store(
store,
rsync_base_uri,
rsync_fetcher,
@@ -221,7 +226,7 @@ pub fn sync_publication_point_replay_delta(
})
}
ReplayDeltaResolvedTransport::Rsync => {
let written = rsync_sync_into_raw_objects(
let written = rsync_sync_into_current_store(
store,
rsync_base_uri,
rsync_fetcher,
@@ -325,22 +330,17 @@ fn validate_delta_replay_base_state_for_repo(
) -> Result<(), RepoSyncError> {
match base.transport {
ReplayTransport::Rrdp => {
let bytes = store
.get_rrdp_state(notification_uri)
.map_err(|e| RepoSyncError::Storage(e.to_string()))?
let state = load_rrdp_local_state(store, notification_uri)
.map_err(RepoSyncError::Storage)?
.ok_or_else(|| {
RepoSyncError::Replay(format!(
"delta replay base state missing for {notification_uri}: expected RRDP session={} serial={}"
,
"delta replay base state missing for {notification_uri}: expected RRDP session={} serial={}",
base.session.as_deref().unwrap_or("<none>"),
base.serial.map(|v| v.to_string()).unwrap_or_else(|| "<none>".to_string())
base.serial
.map(|v| v.to_string())
.unwrap_or_else(|| "<none>".to_string())
))
})?;
let state = RrdpState::decode(&bytes).map_err(|e| {
RepoSyncError::Replay(format!(
"decode base RRDP state failed for {notification_uri}: {e}"
))
})?;
let expected_session = base.session.as_deref().unwrap_or("");
let expected_serial = base.serial.unwrap_or_default();
if state.session_id != expected_session || state.serial != expected_serial {
@@ -496,7 +496,7 @@ fn try_rrdp_sync_with_retry(
}
}
fn rsync_sync_into_raw_objects(
fn rsync_sync_into_current_store(
store: &RocksStore,
rsync_base_uri: &str,
rsync_fetcher: &dyn RsyncFetcher,
@@ -538,17 +538,9 @@ fn rsync_sync_into_raw_objects(
.map_err(|e| RepoSyncError::Storage(e.to_string()))?;
let new_set: HashSet<&str> = objects.iter().map(|(uri, _)| uri.as_str()).collect();
let _w = timing
.as_ref()
.map(|t| t.span_phase("rsync_write_raw_objects_total"));
store
.put_raw_batch(objects.clone())
.map_err(|e| RepoSyncError::Storage(e.to_string()))?;
drop(_w);
let _proj = timing
.as_ref()
.map(|t| t.span_phase("rsync_write_repository_view_total"));
.map(|t| t.span_phase("rsync_write_current_store_total"));
let prepared_raw = prepare_raw_by_hash_evidence_batch(store, &objects)
.map_err(RepoSyncError::Storage)?;
let mut repository_view_entries = Vec::new();
@@ -635,6 +627,15 @@ mod tests {
}
}
fn assert_current_object(store: &RocksStore, uri: &str, expected: &[u8]) {
assert_eq!(
store
.load_current_object_bytes_by_uri(uri)
.expect("load current object"),
Some(expected.to_vec())
);
}
fn notification_xml(
session_id: &str,
serial: u64,
@@ -952,7 +953,7 @@ mod tests {
}
#[test]
fn rsync_sync_writes_raw_objects_with_batch_and_records_counts() {
fn rsync_sync_writes_current_store_and_records_counts() {
let temp = tempfile::tempdir().expect("tempdir");
let repo_dir = temp.path().join("repo");
@@ -1003,22 +1004,9 @@ mod tests {
assert_eq!(objects.objects_count, 3);
assert_eq!(objects.objects_bytes_total, 9);
assert_eq!(
store.get_raw("rsync://example.test/repo/a.mft").unwrap(),
Some(b"mft".to_vec())
);
assert_eq!(
store
.get_raw("rsync://example.test/repo/sub/b.roa")
.unwrap(),
Some(b"roa".to_vec())
);
assert_eq!(
store
.get_raw("rsync://example.test/repo/sub/c.cer")
.unwrap(),
Some(b"cer".to_vec())
);
assert_current_object(&store, "rsync://example.test/repo/a.mft", b"mft");
assert_current_object(&store, "rsync://example.test/repo/sub/b.roa", b"roa");
assert_current_object(&store, "rsync://example.test/repo/sub/c.cer", b"cer");
let view = store
.get_repository_view_entry("rsync://example.test/repo/a.mft")
@@ -1208,10 +1196,7 @@ mod tests {
.expect("sync ok");
assert_eq!(out.source, RepoSyncSource::Rrdp);
assert_eq!(
store.get_raw(published_uri).unwrap(),
Some(published_bytes.to_vec())
);
assert_current_object(&store, published_uri, published_bytes);
let events = download_log.snapshot_events();
assert_eq!(events.len(), 4, "expected 3x notification + 1x snapshot");
@@ -1400,10 +1385,15 @@ mod tests {
session_id: sid.to_string(),
serial: 1,
};
let state_bytes = state.encode().expect("encode state");
store
.put_rrdp_state(notification_uri, &state_bytes)
.expect("seed state");
persist_rrdp_local_state(
&store,
notification_uri,
&state,
RrdpSourceSyncState::DeltaReady,
Some(snapshot_uri),
None,
)
.expect("seed state");
let delta_2 = format!(
r#"<delta xmlns="http://www.ripe.net/rpki/rrdp" version="1" session_id="{sid}" serial="2"></delta>"#
@@ -1450,10 +1440,7 @@ mod tests {
assert_eq!(out.source, RepoSyncSource::Rrdp);
assert_eq!(out.objects_written, 1);
assert_eq!(
store.get_raw(published_uri).unwrap(),
Some(published_bytes.to_vec())
);
assert_current_object(&store, published_uri, published_bytes);
let events = download_log.snapshot_events();
assert_eq!(events.len(), 4);
@@ -1517,10 +1504,7 @@ mod tests {
assert_eq!(out.source, RepoSyncSource::Rrdp);
assert_eq!(out.objects_written, 1);
assert_eq!(
store.get_raw(&published_uri).unwrap(),
Some(b"mft".to_vec())
);
assert_current_object(&store, &published_uri, b"mft");
}
#[test]
@@ -1560,12 +1544,7 @@ mod tests {
assert_eq!(out.source, RepoSyncSource::Rsync);
assert_eq!(out.objects_written, 1);
assert_eq!(out.warnings.len(), 0);
assert_eq!(
store
.get_raw("rsync://rsync.example.test/repo/sub/fallback.cer")
.unwrap(),
Some(b"cer".to_vec())
);
assert_current_object(&store, "rsync://rsync.example.test/repo/sub/fallback.cer", b"cer");
}
#[test]
@@ -1633,9 +1612,15 @@ mod tests {
session_id: "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa".to_string(),
serial: 10,
};
store
.put_rrdp_state(&notify_uri, &state.encode().unwrap())
.expect("seed base state");
persist_rrdp_local_state(
&store,
&notify_uri,
&state,
RrdpSourceSyncState::DeltaReady,
None,
None,
)
.expect("seed base state");
let out = sync_publication_point_replay_delta(
&store,
@@ -1651,21 +1636,11 @@ mod tests {
assert_eq!(out.source, RepoSyncSource::Rrdp);
assert_eq!(out.objects_written, 2);
assert_eq!(
store.get_raw("rsync://example.test/repo/a.mft").unwrap(),
Some(b"delta-a".to_vec())
);
assert_eq!(
store
.get_raw("rsync://example.test/repo/sub/b.roa")
.unwrap(),
Some(b"delta-b".to_vec())
);
let state_bytes = store
.get_rrdp_state(&notify_uri)
.unwrap()
assert_current_object(&store, "rsync://example.test/repo/a.mft", b"delta-a");
assert_current_object(&store, "rsync://example.test/repo/sub/b.roa", b"delta-b");
let new_state = load_rrdp_local_state(&store, &notify_uri)
.expect("load current state")
.expect("rrdp state present");
let new_state = RrdpState::decode(&state_bytes).expect("decode state");
assert_eq!(new_state.serial, 12);
}
@@ -1727,9 +1702,15 @@ mod tests {
session_id: "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa".to_string(),
serial: 10,
};
store
.put_rrdp_state(&notify_uri, &state.encode().unwrap())
.expect("seed base state");
persist_rrdp_local_state(
&store,
&notify_uri,
&state,
RrdpSourceSyncState::DeltaReady,
None,
None,
)
.expect("seed base state");
let base_locks_body = std::fs::read_to_string(&base_locks).expect("read base locks");
let base_locks_sha = sha256_hex(base_locks_body.as_bytes());
@@ -1808,18 +1789,8 @@ mod tests {
.expect("fallback-rsync delta sync ok");
assert_eq!(out.source, RepoSyncSource::Rsync);
assert_eq!(out.objects_written, 2);
assert_eq!(
store
.get_raw("rsync://rsync.example.test/repo/a.mft")
.unwrap(),
Some(b"base".to_vec())
);
assert_eq!(
store
.get_raw("rsync://rsync.example.test/repo/sub/x.cer")
.unwrap(),
Some(b"overlay-cer".to_vec())
);
assert_current_object(&store, "rsync://rsync.example.test/repo/a.mft", b"base");
assert_current_object(&store, "rsync://rsync.example.test/repo/sub/x.cer", b"overlay-cer");
}
#[test]
@@ -1845,9 +1816,15 @@ mod tests {
session_id: "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa".to_string(),
serial: 10,
};
store
.put_rrdp_state(&notify_uri, &state.encode().unwrap())
.expect("seed base state");
persist_rrdp_local_state(
&store,
&notify_uri,
&state,
RrdpSourceSyncState::DeltaReady,
None,
None,
)
.expect("seed base state");
let locks_body = std::fs::read_to_string(&delta_locks).expect("read delta locks");
let rewritten =

234
src/sync/rrdp.rs
View File

@@ -210,6 +210,41 @@ impl RrdpState {
}
}
pub(crate) fn load_rrdp_local_state(
store: &RocksStore,
notification_uri: &str,
) -> Result<Option<RrdpState>, String> {
if let Some(record) = store
.get_rrdp_source_record(notification_uri)
.map_err(|e| e.to_string())?
{
if let (Some(session_id), Some(serial)) = (record.last_session_id, record.last_serial) {
return Ok(Some(RrdpState { session_id, serial }));
}
}
Ok(None)
}
pub(crate) fn persist_rrdp_local_state(
store: &RocksStore,
notification_uri: &str,
state: &RrdpState,
sync_state: RrdpSourceSyncState,
last_snapshot_uri: Option<&str>,
last_snapshot_hash_hex: Option<&str>,
) -> Result<(), String> {
update_rrdp_source_record_on_success(
store,
notification_uri,
state.session_id.as_str(),
state.serial,
sync_state,
last_snapshot_uri,
last_snapshot_hash_hex,
)
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct NotificationSnapshot {
pub session_id: Uuid,
@@ -582,15 +617,10 @@ fn sync_from_notification_snapshot_inner(
session_id: notif.session_id.to_string(),
serial: notif.serial,
};
let bytes = state.encode().map_err(RrdpSyncError::Storage)?;
store
.put_rrdp_state(notification_uri, &bytes)
.map_err(|e| RrdpSyncError::Storage(e.to_string()))?;
update_rrdp_source_record_on_success(
persist_rrdp_local_state(
store,
notification_uri,
notif.session_id.to_string().as_str(),
notif.serial,
&state,
RrdpSourceSyncState::SnapshotOnly,
Some(&notif.snapshot_uri),
Some(&hex::encode(notif.snapshot_hash_sha256)),
@@ -683,10 +713,7 @@ fn sync_from_notification_inner(
let _read_state_total = timing
.as_ref()
.map(|t| t.span_phase("rrdp_read_state_total"));
let state = store
.get_rrdp_state(notification_uri)
.map_err(|e| RrdpSyncError::Storage(e.to_string()))?
.and_then(|bytes| RrdpState::decode(&bytes).ok());
let state = load_rrdp_local_state(store, notification_uri).map_err(RrdpSyncError::Storage)?;
drop(_read_state_step);
drop(_read_state_total);
@@ -811,15 +838,10 @@ fn sync_from_notification_inner(
session_id: notif.session_id.to_string(),
serial: notif.serial,
};
let bytes = new_state.encode().map_err(RrdpSyncError::Storage)?;
store
.put_rrdp_state(notification_uri, &bytes)
.map_err(|e| RrdpSyncError::Storage(e.to_string()))?;
update_rrdp_source_record_on_success(
persist_rrdp_local_state(
store,
notification_uri,
notif.session_id.to_string().as_str(),
notif.serial,
&new_state,
RrdpSourceSyncState::DeltaReady,
Some(&notif.snapshot_uri),
Some(&hex::encode(notif.snapshot_hash_sha256)),
@@ -913,15 +935,10 @@ fn sync_from_notification_inner(
session_id: notif.session_id.to_string(),
serial: notif.serial,
};
let bytes = new_state.encode().map_err(RrdpSyncError::Storage)?;
store
.put_rrdp_state(notification_uri, &bytes)
.map_err(|e| RrdpSyncError::Storage(e.to_string()))?;
update_rrdp_source_record_on_success(
persist_rrdp_local_state(
store,
notification_uri,
notif.session_id.to_string().as_str(),
notif.serial,
&new_state,
RrdpSourceSyncState::SnapshotOnly,
Some(&notif.snapshot_uri),
Some(&hex::encode(notif.snapshot_hash_sha256)),
@@ -984,7 +1001,7 @@ fn apply_delta(
bytes,
} => {
let is_member = store
.rrdp_object_index_contains(notification_uri, uri.as_str())
.is_current_rrdp_source_member(notification_uri, uri.as_str())
.map_err(|e| RrdpSyncError::Storage(e.to_string()))?;
if !is_member {
return Err(RrdpError::DeltaTargetNotFromRepository { rsync_uri: uri }.into());
@@ -992,7 +1009,7 @@ fn apply_delta(
ensure_rrdp_uri_can_be_owned_by(store, notification_uri, uri.as_str())
.map_err(RrdpSyncError::Storage)?;
let old_bytes = store
.get_raw(uri.as_str())
.load_current_object_bytes_by_uri(uri.as_str())
.map_err(|e| RrdpSyncError::Storage(e.to_string()))?
.ok_or_else(|| RrdpError::DeltaTargetMissing {
rsync_uri: uri.clone(),
@@ -1017,7 +1034,7 @@ fn apply_delta(
bytes,
} => {
let is_member = store
.rrdp_object_index_contains(notification_uri, uri.as_str())
.is_current_rrdp_source_member(notification_uri, uri.as_str())
.map_err(|e| RrdpSyncError::Storage(e.to_string()))?;
if is_member {
return Err(
@@ -1037,7 +1054,7 @@ fn apply_delta(
}
DeltaElement::Withdraw { uri, hash_sha256 } => {
let is_member = store
.rrdp_object_index_contains(notification_uri, uri.as_str())
.is_current_rrdp_source_member(notification_uri, uri.as_str())
.map_err(|e| RrdpSyncError::Storage(e.to_string()))?;
if !is_member {
return Err(RrdpError::DeltaTargetNotFromRepository { rsync_uri: uri }.into());
@@ -1045,7 +1062,7 @@ fn apply_delta(
ensure_rrdp_uri_can_be_owned_by(store, notification_uri, uri.as_str())
.map_err(RrdpSyncError::Storage)?;
let old_bytes = store
.get_raw(uri.as_str())
.load_current_object_bytes_by_uri(uri.as_str())
.map_err(|e| RrdpSyncError::Storage(e.to_string()))?
.ok_or_else(|| RrdpError::DeltaTargetMissing {
rsync_uri: uri.clone(),
@@ -1066,10 +1083,6 @@ fn apply_delta(
}
}
store
.apply_rrdp_delta(notification_uri, ops.as_slice())
.map_err(|e| RrdpSyncError::Storage(e.to_string()))?;
for effect in projection {
match effect {
DeltaProjectionEffect::Upsert { rsync_uri, bytes } => {
@@ -1188,8 +1201,10 @@ fn apply_snapshot(
}
let previous_members: Vec<String> = store
.rrdp_object_index_iter(notification_uri)
.list_current_rrdp_source_members(notification_uri)
.map_err(|e| RrdpSyncError::Storage(e.to_string()))?
.into_iter()
.map(|record| record.rsync_uri)
.collect();
let new_set: std::collections::HashSet<&str> =
published.iter().map(|(uri, _)| uri.as_str()).collect();
@@ -1204,7 +1219,7 @@ fn apply_snapshot(
.and_then(|entry| entry.current_hash)
.or_else(|| {
store
.get_raw(old_uri)
.load_current_object_bytes_by_uri(old_uri)
.ok()
.flatten()
.map(|bytes| compute_sha256_hex(&bytes))
@@ -1212,10 +1227,6 @@ fn apply_snapshot(
withdrawn.push((old_uri.clone(), previous_hash));
}
store
.apply_rrdp_snapshot(notification_uri, published.as_slice())
.map_err(|e| RrdpSyncError::Storage(e.to_string()))?;
let session_id = expected_session_id.to_string();
let prepared_raw = prepare_raw_by_hash_evidence_batch(store, &published)
.map_err(RrdpSyncError::Storage)?;
@@ -1404,6 +1415,15 @@ mod tests {
}
}
fn assert_current_object(store: &RocksStore, uri: &str, expected: &[u8]) {
assert_eq!(
store
.load_current_object_bytes_by_uri(uri)
.expect("load current object"),
Some(expected.to_vec())
);
}
fn notification_xml(
session_id: &str,
serial: u64,
@@ -1682,8 +1702,8 @@ mod tests {
.expect("sync snapshot");
let old_b = store
.get_raw("rsync://example.net/repo/b.roa")
.expect("get_raw")
.load_current_object_bytes_by_uri("rsync://example.net/repo/b.roa")
.expect("load current b")
.expect("b present");
let old_b_hash = hex::encode(sha2::Sha256::digest(old_b.as_slice()));
@@ -1721,34 +1741,34 @@ mod tests {
.expect("apply delta");
assert_eq!(applied, 3);
assert!(
assert_eq!(
store
.get_raw("rsync://example.net/repo/a.mft")
.expect("get_raw")
.is_none(),
.load_current_object_bytes_by_uri("rsync://example.net/repo/a.mft")
.expect("load current a"),
None,
"a withdrawn"
);
let b = store
.get_raw("rsync://example.net/repo/b.roa")
.expect("get_raw")
.load_current_object_bytes_by_uri("rsync://example.net/repo/b.roa")
.expect("load current b")
.expect("b present");
assert_eq!(b, b"b2");
let c = store
.get_raw("rsync://example.net/repo/c.crl")
.expect("get_raw")
.load_current_object_bytes_by_uri("rsync://example.net/repo/c.crl")
.expect("load current c")
.expect("c present");
assert_eq!(c, b"c2");
assert!(
!store
.rrdp_object_index_contains(notif_uri, "rsync://example.net/repo/a.mft")
.expect("contains"),
.is_current_rrdp_source_member(notif_uri, "rsync://example.net/repo/a.mft")
.expect("is member"),
"a removed from rrdp repo index"
);
assert!(
store
.rrdp_object_index_contains(notif_uri, "rsync://example.net/repo/c.crl")
.expect("contains"),
.is_current_rrdp_source_member(notif_uri, "rsync://example.net/repo/c.crl")
.expect("is member"),
"c added to rrdp repo index"
);
@@ -1779,6 +1799,19 @@ mod tests {
.expect("get a member")
.expect("a member exists");
assert!(!a_member.present);
let current_members = store
.list_current_rrdp_source_members(notif_uri)
.expect("list current members");
assert_eq!(
current_members
.iter()
.map(|record| record.rsync_uri.as_str())
.collect::<Vec<_>>(),
vec![
"rsync://example.net/repo/b.roa",
"rsync://example.net/repo/c.crl",
]
);
}
#[test]
@@ -1893,7 +1926,7 @@ mod tests {
sync_from_notification_snapshot(&store, notif_uri, &notif, &fetcher).expect("seed");
let old_bytes = store
.get_raw("rsync://example.net/repo/a.mft")
.load_current_object_bytes_by_uri("rsync://example.net/repo/a.mft")
.expect("get")
.expect("present");
let old_hash = hex::encode(sha2::Sha256::digest(old_bytes.as_slice()));
@@ -1926,8 +1959,8 @@ mod tests {
// Target missing in local cache (index still says it's a member).
store
.delete_raw("rsync://example.net/repo/a.mft")
.expect("delete");
.delete_repository_view_entry("rsync://example.net/repo/a.mft")
.expect("delete current repository view entry");
let delta = delta_xml(
sid,
2,
@@ -2061,23 +2094,12 @@ mod tests {
sync_from_notification_snapshot(&store, notif_uri, &notif, &fetcher).expect("sync");
assert_eq!(published, 2);
let a = store
.get_raw("rsync://example.net/repo/a.mft")
.expect("get_raw")
.expect("a present");
assert_eq!(a, b"mft-bytes");
assert_current_object(&store, "rsync://example.net/repo/a.mft", b"mft-bytes");
assert_current_object(&store, "rsync://example.net/repo/b.roa", b"roa-bytes");
let b = store
.get_raw("rsync://example.net/repo/b.roa")
.expect("get_raw")
.expect("b present");
assert_eq!(b, b"roa-bytes");
let state_bytes = store
.get_rrdp_state(notif_uri)
.expect("get_rrdp_state")
let state = load_rrdp_local_state(&store, notif_uri)
.expect("get rrdp state")
.expect("state present");
let state = RrdpState::decode(&state_bytes).expect("decode state");
assert_eq!(state.session_id, sid);
assert_eq!(state.serial, serial);
@@ -2166,25 +2188,16 @@ mod tests {
};
sync_from_notification_snapshot(&store, notif_uri, &notif_2, &fetcher_2).expect("sync 2");
let a = store
.get_raw("rsync://example.net/repo/a.mft")
.expect("get_raw");
assert!(
a.is_none(),
store
.load_current_object_bytes_by_uri("rsync://example.net/repo/a.mft")
.expect("get current a")
.is_none(),
"a should be deleted by full-state snapshot apply"
);
let b = store
.get_raw("rsync://example.net/repo/b.roa")
.expect("get_raw")
.expect("b present");
assert_eq!(b, b"b2");
let c = store
.get_raw("rsync://example.net/repo/c.crl")
.expect("get_raw")
.expect("c present");
assert_eq!(c, b"c2");
assert_current_object(&store, "rsync://example.net/repo/b.roa", b"b2");
assert_current_object(&store, "rsync://example.net/repo/c.crl", b"c2");
}
#[test]
@@ -2271,20 +2284,51 @@ mod tests {
// Delta 2 publishes c then delta 3 withdraws it => final state should not contain c.
assert!(
store
.get_raw("rsync://example.net/repo/c.crl")
.expect("get_raw")
.load_current_object_bytes_by_uri("rsync://example.net/repo/c.crl")
.expect("get current")
.is_none()
);
let state_bytes = store
.get_rrdp_state(notif_uri)
.expect("get_rrdp_state")
let state = load_rrdp_local_state(&store, notif_uri)
.expect("get rrdp state")
.expect("state present");
let state = RrdpState::decode(&state_bytes).expect("decode");
assert_eq!(state.session_id, Uuid::parse_str(sid).unwrap().to_string());
assert_eq!(state.serial, 3);
}
#[test]
fn load_rrdp_local_state_uses_source_record_only() {
let tmp = tempfile::tempdir().expect("tempdir");
let store = RocksStore::open(tmp.path()).expect("open rocksdb");
let notif_uri = "https://example.net/notification.xml";
assert_eq!(
load_rrdp_local_state(&store, notif_uri).expect("load empty"),
None
);
update_rrdp_source_record_on_success(
&store,
notif_uri,
"source-session",
9,
crate::storage::RrdpSourceSyncState::DeltaReady,
Some("https://example.net/snapshot.xml"),
Some(&hex::encode([0x11; 32])),
)
.expect("write source record");
let got = load_rrdp_local_state(&store, notif_uri)
.expect("load source preferred")
.expect("source present");
assert_eq!(
got,
RrdpState {
session_id: "source-session".to_string(),
serial: 9,
}
);
}
#[test]
fn sync_from_notification_falls_back_to_snapshot_if_missing_required_deltas() {
let tmp = tempfile::tempdir().expect("tempdir");
@@ -2331,8 +2375,8 @@ mod tests {
assert_eq!(published, 1);
assert!(
store
.get_raw("rsync://example.net/repo/z.roa")
.expect("get_raw")
.load_current_object_bytes_by_uri("rsync://example.net/repo/z.roa")
.expect("get current")
.is_some()
);
}

38
src/validation/manifest.rs
View File

@@ -136,7 +136,7 @@ pub enum ManifestFreshError {
RepoSyncFailed { detail: String },
#[error(
"manifest not found in raw_objects: {manifest_rsync_uri} (RFC 9286 §6.2; RFC 9286 §6.6)"
"manifest not found in current repository view: {manifest_rsync_uri} (RFC 9286 §6.2; RFC 9286 §6.6)"
)]
MissingManifest { manifest_rsync_uri: String },
@@ -204,7 +204,7 @@ pub enum ManifestFreshError {
},
#[error(
"manifest referenced file missing in raw_objects: {rsync_uri} (RFC 9286 §6.4; RFC 9286 §6.6)"
"manifest referenced file missing in current repository view: {rsync_uri} (RFC 9286 §6.4; RFC 9286 §6.6)"
)]
MissingFile { rsync_uri: String },
@@ -548,7 +548,7 @@ pub(crate) fn try_build_fresh_publication_point(
}
let manifest_bytes = store
.get_raw(manifest_rsync_uri)
.load_current_object_bytes_by_uri(manifest_rsync_uri)
.map_err(|e| ManifestFreshError::MissingManifest {
manifest_rsync_uri: format!("{manifest_rsync_uri} ({e})"),
})?
@@ -642,7 +642,7 @@ pub(crate) fn try_build_fresh_publication_point(
let rsync_uri =
join_rsync_dir_and_file(publication_point_rsync_uri, entry.file_name.as_str());
let bytes = store
.get_raw(&rsync_uri)
.load_current_object_bytes_by_uri(&rsync_uri)
.map_err(|_e| ManifestFreshError::MissingFile {
rsync_uri: rsync_uri.clone(),
})?
@@ -836,6 +836,27 @@ mod tests {
entry
}
fn put_current_object(
store: &RocksStore,
rsync_uri: &str,
bytes: Vec<u8>,
object_type: &str,
) {
let hash = hex::encode(sha2::Sha256::digest(&bytes));
store
.put_raw_by_hash_entry(&raw_by_hash_entry(rsync_uri, bytes, object_type))
.expect("put raw_by_hash entry");
store
.put_repository_view_entry(&crate::storage::RepositoryViewEntry {
rsync_uri: rsync_uri.to_string(),
current_hash: Some(hash),
repository_source: Some("https://example.test/notification.xml".to_string()),
object_type: Some(object_type.to_string()),
state: crate::storage::RepositoryViewState::Present,
})
.expect("put repository view entry");
}
fn sample_current_instance_vcir(
manifest_rsync_uri: &str,
publication_point_rsync_uri: &str,
@@ -985,9 +1006,7 @@ mod tests {
publication_point_rsync_uri,
validation_time,
) = load_manifest_fixture();
store
.put_raw(&manifest_rsync_uri, &manifest_bytes)
.expect("store manifest");
put_current_object(&store, &manifest_rsync_uri, manifest_bytes, "mft");
let first_non_crl = manifest
.manifest
.parse_files()
@@ -1001,9 +1020,8 @@ mod tests {
.join(first_non_crl.file_name.as_str());
let bytes = std::fs::read(&file_path).expect("read fixture file");
let rsync_uri = format!("{publication_point_rsync_uri}{}", first_non_crl.file_name);
store
.put_raw(&rsync_uri, &bytes)
.expect("store single file");
let object_type = rsync_uri.rsplit('.').next().unwrap_or("bin");
put_current_object(&store, &rsync_uri, bytes, object_type);
let err = try_build_fresh_publication_point(
&store,

2
src/validation/tree_runner.rs
View File

@@ -64,7 +64,7 @@ pub struct Rpkiv1PublicationPointRunner<'a> {
/// In-run RRDP dedup: when RRDP is enabled, only sync each `rrdp_notification_uri` once per run.
///
/// - If RRDP succeeded for a repo, later publication points referencing that same RRDP repo
/// skip network fetches and reuse the already-populated `raw_objects`.
/// skip network fetches and reuse the already-populated current repository view.
/// - If RRDP failed for a repo, later publication points skip RRDP attempts and go straight
/// to rsync for their own `rsync_base_uri` (still per-publication-point).
pub rrdp_dedup: bool,