mod config; mod keys; mod pack; mod pp_cache_index; use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; use base64::Engine; use rocksdb::{ColumnFamily, DB, Direction, IteratorMode, Options, WriteBatch}; use serde::{Deserialize, Serialize}; use crate::blob_store::{ExternalRawStoreDb, ExternalRepoBytesDb, RawObjectStore}; use crate::data_model::rc::{AsResourceSet, IpResourceSet}; use config::*; pub use config::{ ALL_COLUMN_FAMILY_NAMES, CF_CHILD_CERTIFICATE_CACHE_PROJECTION, CF_MANIFEST_REPLAY_META, CF_PUBLICATION_POINT_CACHE_PROJECTION, CF_RAW_BY_HASH, CF_REPOSITORY_VIEW, CF_ROA_CACHE_PROJECTION, CF_RRDP_SOURCE, CF_RRDP_SOURCE_MEMBER, CF_RRDP_URI_OWNER, CF_TRANSPORT_PREFETCH, CF_VCIR, column_family_descriptors, }; use keys::*; use pack::compute_sha256_32; pub use pack::{PackBytes, PackFile, PackTime}; pub use pp_cache_index::{PpCacheIndexLoadStats, PpCacheIndexRefreshStats}; use pp_cache_index::{ PpCacheIndexLookup, PpCacheMmapIndexSet, compact_pp_cache_index, default_pp_cache_index_dir, load_pp_cache_mmap_index, load_pp_cache_mmap_index_set, pp_cache_index_directory_stats, write_pp_cache_index_atomic, write_pp_cache_index_segment, }; #[derive(Debug, thiserror::Error)] pub enum StorageError { #[error("rocksdb error: {0}")] RocksDb(String), #[error("missing column family: {0}")] MissingColumnFamily(&'static str), #[error("cbor codec error for {entity}: {detail}")] Codec { entity: &'static str, detail: String, }, #[error("invalid {entity}: {detail}")] InvalidData { entity: &'static str, detail: String, }, } pub type StorageResult = Result; #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)] pub struct ChildCertificateCacheMmapLookup { pub projections: Vec>, pub hits: usize, pub misses: usize, pub file_bytes: u64, } pub struct RocksStore { db: DB, external_raw_store: Option, external_repo_bytes: Option, publication_point_cache_index_dir: PathBuf, child_certificate_cache_index_dir: PathBuf, publication_point_cache_projection_index: Mutex, } enum PublicationPointCacheProjectionIndexState { Uninitialized, Disabled, BuildingFromEmpty { index: HashMap>, bytes: usize, limit: usize, }, Loaded { index: HashMap>, bytes: usize, }, LoadedMmap { mmap: PpCacheMmapIndexSet, dirty: HashMap>, dirty_bytes: usize, load_stats: PpCacheIndexLoadStats, }, } #[derive(Clone, Copy)] pub(crate) enum PublicationPointCacheProjectionWriteAction<'a> { Keep, Write(&'a PublicationPointCacheProjection), Delete { manifest_rsync_uri: &'a str }, } fn process_vm_rss_kb() -> Option { let status = std::fs::read_to_string("/proc/self/status").ok()?; status.lines().find_map(|line| { let rest = line.strip_prefix("VmRSS:")?; rest.split_whitespace().next()?.parse::().ok() }) } fn default_child_certificate_cache_index_dir(db_path: &Path) -> PathBuf { let file_name = db_path .file_name() .and_then(|name| name.to_str()) .unwrap_or("work-db"); db_path.with_file_name(format!("{file_name}.child-cert-cache-index")) } fn child_certificate_cache_segment_file_name(manifest_rsync_uri: &str) -> String { format!( "{}.idx", hex::encode(compute_sha256_32(manifest_rsync_uri.as_bytes())) ) } const ROCKSDB_MEMORY_PROPERTY_NAMES: &[(&str, &str)] = &[ ("cur_size_all_mem_tables", "rocksdb.cur-size-all-mem-tables"), ("size_all_mem_tables", "rocksdb.size-all-mem-tables"), ( "estimate_table_readers_mem", "rocksdb.estimate-table-readers-mem", ), ("block_cache_capacity", "rocksdb.block-cache-capacity"), ("block_cache_usage", "rocksdb.block-cache-usage"), ( "block_cache_pinned_usage", "rocksdb.block-cache-pinned-usage", ), ("num_snapshots", "rocksdb.num-snapshots"), ("background_errors", "rocksdb.background-errors"), ]; const PP_CACHE_RAW_INDEX_ENV: &str = "RPKI_PP_CACHE_RAW_INDEX"; const PP_CACHE_RAW_INDEX_EMPTY_BUILD_LIMIT_ENV: &str = "RPKI_PP_CACHE_RAW_INDEX_EMPTY_BUILD_LIMIT_BYTES"; const DEFAULT_PP_CACHE_RAW_INDEX_EMPTY_BUILD_LIMIT_BYTES: usize = 32 * 1024 * 1024; const PP_CACHE_INDEX_COMPACTION_SEGMENT_THRESHOLD: usize = 16; const PP_CACHE_INDEX_COMPACTION_BYTES_THRESHOLD: u64 = 1_610_612_736; fn pp_cache_raw_index_enabled() -> bool { match std::env::var(PP_CACHE_RAW_INDEX_ENV) { Ok(value) => !matches!( value.trim().to_ascii_lowercase().as_str(), "0" | "false" | "off" | "no" ), Err(_) => true, } } fn pp_cache_raw_index_empty_build_limit_bytes() -> usize { std::env::var(PP_CACHE_RAW_INDEX_EMPTY_BUILD_LIMIT_ENV) .ok() .and_then(|value| value.trim().parse::().ok()) .unwrap_or(DEFAULT_PP_CACHE_RAW_INDEX_EMPTY_BUILD_LIMIT_BYTES) } #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)] pub struct RocksDbMemoryProperties { pub cur_size_all_mem_tables: Option, pub size_all_mem_tables: Option, pub estimate_table_readers_mem: Option, pub block_cache_capacity: Option, pub block_cache_usage: Option, pub block_cache_pinned_usage: Option, pub num_snapshots: Option, pub background_errors: Option, pub errors: Vec, } #[derive(Clone, Debug, PartialEq, Eq, Serialize)] pub struct RocksDbColumnFamilyMemoryProperties { pub name: String, pub properties: RocksDbMemoryProperties, } #[derive(Clone, Debug, PartialEq, Eq, Serialize)] pub struct RocksDbMemoryDbSnapshot { pub label: String, pub properties: RocksDbMemoryProperties, pub column_families: Vec, } #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)] pub struct RocksDbMemoryTotals { pub cur_size_all_mem_tables: u64, pub size_all_mem_tables: u64, pub estimate_table_readers_mem: u64, pub block_cache_capacity: u64, pub block_cache_usage: u64, pub block_cache_pinned_usage: u64, } #[derive(Clone, Debug, PartialEq, Eq, Serialize)] pub struct RocksDbMemorySnapshot { pub databases: Vec, pub totals: RocksDbMemoryTotals, } impl RocksDbMemoryTotals { fn add_properties(&mut self, properties: &RocksDbMemoryProperties) { self.cur_size_all_mem_tables += properties.cur_size_all_mem_tables.unwrap_or(0); self.size_all_mem_tables += properties.size_all_mem_tables.unwrap_or(0); self.estimate_table_readers_mem += properties.estimate_table_readers_mem.unwrap_or(0); self.block_cache_capacity += properties.block_cache_capacity.unwrap_or(0); self.block_cache_usage += properties.block_cache_usage.unwrap_or(0); self.block_cache_pinned_usage += properties.block_cache_pinned_usage.unwrap_or(0); } } fn set_memory_property(properties: &mut RocksDbMemoryProperties, name: &str, value: u64) { match name { "cur_size_all_mem_tables" => properties.cur_size_all_mem_tables = Some(value), "size_all_mem_tables" => properties.size_all_mem_tables = Some(value), "estimate_table_readers_mem" => properties.estimate_table_readers_mem = Some(value), "block_cache_capacity" => properties.block_cache_capacity = Some(value), "block_cache_usage" => properties.block_cache_usage = Some(value), "block_cache_pinned_usage" => properties.block_cache_pinned_usage = Some(value), "num_snapshots" => properties.num_snapshots = Some(value), "background_errors" => properties.background_errors = Some(value), _ => {} } } fn parse_rocksdb_property_int(raw: Option) -> Option { raw.and_then(|value| value.trim().parse::().ok()) } fn memory_properties_for_db(db: &DB) -> RocksDbMemoryProperties { let mut properties = RocksDbMemoryProperties::default(); for (field_name, property_name) in ROCKSDB_MEMORY_PROPERTY_NAMES { match db.property_value(*property_name) { Ok(value) => { if let Some(parsed) = parse_rocksdb_property_int(value) { set_memory_property(&mut properties, field_name, parsed); } } Err(err) => properties.errors.push(format!("{property_name}: {}", err)), } } properties } fn memory_properties_for_cf(db: &DB, cf_name: &'static str) -> RocksDbColumnFamilyMemoryProperties { let mut properties = RocksDbMemoryProperties::default(); let Some(cf) = db.cf_handle(cf_name) else { properties .errors .push(format!("missing column family: {cf_name}")); return RocksDbColumnFamilyMemoryProperties { name: cf_name.to_string(), properties, }; }; for (field_name, property_name) in ROCKSDB_MEMORY_PROPERTY_NAMES { match db.property_value_cf(cf, *property_name) { Ok(value) => { if let Some(parsed) = parse_rocksdb_property_int(value) { set_memory_property(&mut properties, field_name, parsed); } } Err(err) => properties.errors.push(format!("{property_name}: {}", err)), } } RocksDbColumnFamilyMemoryProperties { name: cf_name.to_string(), properties, } } pub(crate) fn memory_db_snapshot_for_column_families( label: impl Into, db: &DB, column_families: Option<&[&'static str]>, ) -> RocksDbMemoryDbSnapshot { RocksDbMemoryDbSnapshot { label: label.into(), properties: memory_properties_for_db(db), column_families: column_families .map(|names| { names .iter() .map(|name| memory_properties_for_cf(db, name)) .collect() }) .unwrap_or_default(), } } #[derive(Clone, Debug, PartialEq, Eq)] pub enum RrdpDeltaOp { Upsert { rsync_uri: String, bytes: Vec }, Delete { rsync_uri: String }, } #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum RepositoryViewState { Present, Withdrawn, Replaced, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct RepositoryViewEntry { pub rsync_uri: String, pub current_hash: Option, pub repository_source: Option, pub object_type: Option, pub state: RepositoryViewState, } impl RepositoryViewEntry { pub fn validate_internal(&self) -> StorageResult<()> { validate_non_empty("repository_view.rsync_uri", &self.rsync_uri)?; if let Some(source) = &self.repository_source { validate_non_empty("repository_view.repository_source", source)?; } match self.state { RepositoryViewState::Present | RepositoryViewState::Replaced => { let hash = self .current_hash .as_deref() .ok_or(StorageError::InvalidData { entity: "repository_view", detail: "current_hash is required when state is present or replaced" .to_string(), })?; validate_sha256_hex("repository_view.current_hash", hash)?; } RepositoryViewState::Withdrawn => { if let Some(hash) = &self.current_hash { validate_sha256_hex("repository_view.current_hash", hash)?; } } } Ok(()) } } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct RawByHashEntry { pub sha256_hex: String, pub bytes: Vec, pub origin_uris: Vec, pub object_type: Option, pub encoding: Option, } impl RawByHashEntry { pub fn from_bytes(sha256_hex: impl Into, bytes: Vec) -> Self { Self { sha256_hex: sha256_hex.into(), bytes, origin_uris: Vec::new(), object_type: None, encoding: None, } } pub fn validate_internal(&self) -> StorageResult<()> { validate_sha256_hex("raw_by_hash.sha256_hex", &self.sha256_hex)?; if self.bytes.is_empty() { return Err(StorageError::InvalidData { entity: "raw_by_hash", detail: "bytes must not be empty".to_string(), }); } let computed = hex::encode(compute_sha256_32(&self.bytes)); if computed != self.sha256_hex.to_ascii_lowercase() { return Err(StorageError::InvalidData { entity: "raw_by_hash", detail: "sha256_hex does not match bytes".to_string(), }); } let mut seen = HashSet::with_capacity(self.origin_uris.len()); for uri in &self.origin_uris { validate_non_empty("raw_by_hash.origin_uris[]", uri)?; if !seen.insert(uri.as_str()) { return Err(StorageError::InvalidData { entity: "raw_by_hash", detail: format!("duplicate origin URI: {uri}"), }); } } Ok(()) } } #[derive(Clone, Debug, PartialEq, Eq)] pub struct CurrentObjectWithHash { pub current_hash_hex: String, pub current_hash: [u8; 32], pub bytes: Vec, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct ValidatedManifestMeta { pub validated_manifest_number: Vec, pub validated_manifest_this_update: PackTime, pub validated_manifest_next_update: PackTime, } impl ValidatedManifestMeta { pub fn validate_internal(&self) -> StorageResult<()> { validate_manifest_number_be( "validated_manifest_meta.validated_manifest_number", &self.validated_manifest_number, )?; let this_update = parse_time( "validated_manifest_meta.validated_manifest_this_update", &self.validated_manifest_this_update, )?; let next_update = parse_time( "validated_manifest_meta.validated_manifest_next_update", &self.validated_manifest_next_update, )?; if next_update < this_update { return Err(StorageError::InvalidData { entity: "validated_manifest_meta", detail: "validated_manifest_next_update must be >= validated_manifest_this_update" .to_string(), }); } Ok(()) } } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct ManifestReplayMeta { pub manifest_rsync_uri: String, pub manifest_number_be: Vec, pub manifest_this_update: PackTime, pub manifest_sha256: Vec, pub updated_at_validation_time: PackTime, } impl ManifestReplayMeta { pub fn from_vcir(vcir: &ValidatedCaInstanceResult) -> Self { Self { manifest_rsync_uri: vcir.manifest_rsync_uri.clone(), manifest_number_be: vcir .validated_manifest_meta .validated_manifest_number .clone(), manifest_this_update: vcir .validated_manifest_meta .validated_manifest_this_update .clone(), manifest_sha256: vcir.ccr_manifest_projection.manifest_sha256.clone(), updated_at_validation_time: vcir.last_successful_validation_time.clone(), } } pub fn validate_internal(&self) -> StorageResult<()> { validate_non_empty( "manifest_replay_meta.manifest_rsync_uri", &self.manifest_rsync_uri, )?; validate_manifest_number_be( "manifest_replay_meta.manifest_number_be", &self.manifest_number_be, )?; parse_time( "manifest_replay_meta.manifest_this_update", &self.manifest_this_update, )?; validate_sha256_digest_bytes( "manifest_replay_meta.manifest_sha256", &self.manifest_sha256, )?; parse_time( "manifest_replay_meta.updated_at_validation_time", &self.updated_at_validation_time, )?; Ok(()) } } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct VcirCcrManifestProjection { pub manifest_rsync_uri: String, pub manifest_sha256: Vec, pub manifest_size: u64, pub manifest_ee_aki: Vec, pub manifest_number_be: Vec, pub manifest_this_update: PackTime, pub manifest_sia_locations_der: Vec>, pub subordinate_skis: Vec>, } impl VcirCcrManifestProjection { pub fn validate_internal(&self) -> StorageResult<()> { validate_non_empty( "vcir.ccr_manifest_projection.manifest_rsync_uri", &self.manifest_rsync_uri, )?; validate_sha256_digest_bytes( "vcir.ccr_manifest_projection.manifest_sha256", &self.manifest_sha256, )?; if self.manifest_size < 1000 { return Err(StorageError::InvalidData { entity: "vcir.ccr_manifest_projection.manifest_size", detail: format!("must be >= 1000, got {}", self.manifest_size), }); } validate_fixed_len_bytes( "vcir.ccr_manifest_projection.manifest_ee_aki", &self.manifest_ee_aki, 20, )?; validate_manifest_number_be( "vcir.ccr_manifest_projection.manifest_number_be", &self.manifest_number_be, )?; parse_time( "vcir.ccr_manifest_projection.manifest_this_update", &self.manifest_this_update, )?; if self.manifest_sia_locations_der.is_empty() { return Err(StorageError::InvalidData { entity: "vcir.ccr_manifest_projection.manifest_sia_locations_der", detail: "must contain at least one AccessDescription".to_string(), }); } for location in &self.manifest_sia_locations_der { validate_full_der_with_tag( "vcir.ccr_manifest_projection.manifest_sia_locations_der[]", location, Some(0x30), )?; } validate_sorted_unique_fixed_len_bytes( "vcir.ccr_manifest_projection.subordinate_skis", &self.subordinate_skis, 20, )?; Ok(()) } } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct VcirInstanceGate { pub manifest_next_update: PackTime, pub current_crl_next_update: PackTime, pub self_ca_not_after: PackTime, pub instance_effective_until: PackTime, } impl VcirInstanceGate { pub fn validate_internal(&self) -> StorageResult<()> { let manifest_next_update = parse_time( "vcir.instance_gate.manifest_next_update", &self.manifest_next_update, )?; let current_crl_next_update = parse_time( "vcir.instance_gate.current_crl_next_update", &self.current_crl_next_update, )?; let self_ca_not_after = parse_time( "vcir.instance_gate.self_ca_not_after", &self.self_ca_not_after, )?; let instance_effective_until = parse_time( "vcir.instance_gate.instance_effective_until", &self.instance_effective_until, )?; let expected = manifest_next_update .min(current_crl_next_update) .min(self_ca_not_after); if instance_effective_until != expected { return Err(StorageError::InvalidData { entity: "vcir.instance_gate", detail: "instance_effective_until must equal min(manifest_next_update, current_crl_next_update, self_ca_not_after)".to_string(), }); } Ok(()) } } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct VcirChildEntry { pub child_manifest_rsync_uri: String, pub child_cert_rsync_uri: String, pub child_cert_hash: String, pub child_ski: String, pub child_rsync_base_uri: String, pub child_publication_point_rsync_uri: String, pub child_rrdp_notification_uri: Option, pub child_effective_ip_resources: Option, pub child_effective_as_resources: Option, pub accepted_at_validation_time: PackTime, } impl VcirChildEntry { pub fn validate_internal(&self) -> StorageResult<()> { validate_non_empty( "vcir.child_entries[].child_manifest_rsync_uri", &self.child_manifest_rsync_uri, )?; validate_non_empty( "vcir.child_entries[].child_cert_rsync_uri", &self.child_cert_rsync_uri, )?; validate_sha256_hex( "vcir.child_entries[].child_cert_hash", &self.child_cert_hash, )?; validate_non_empty("vcir.child_entries[].child_ski", &self.child_ski)?; validate_non_empty( "vcir.child_entries[].child_rsync_base_uri", &self.child_rsync_base_uri, )?; validate_non_empty( "vcir.child_entries[].child_publication_point_rsync_uri", &self.child_publication_point_rsync_uri, )?; if let Some(uri) = &self.child_rrdp_notification_uri { validate_non_empty("vcir.child_entries[].child_rrdp_notification_uri", uri)?; } parse_time( "vcir.child_entries[].accepted_at_validation_time", &self.accepted_at_validation_time, )?; Ok(()) } } #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum VcirOutputType { Vrp, Aspa, RouterKey, } #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum VcirSourceObjectType { Roa, Aspa, RouterKey, Other, } impl VcirSourceObjectType { pub fn as_str(self) -> &'static str { match self { Self::Roa => "roa", Self::Aspa => "aspa", Self::RouterKey => "router_key", Self::Other => "other", } } } struct FixedBytesVisitor; impl<'de, const N: usize> serde::de::Visitor<'de> for FixedBytesVisitor { type Value = [u8; N]; fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!(formatter, "{N} bytes") } fn visit_bytes(self, value: &[u8]) -> Result where E: serde::de::Error, { if value.len() != N { return Err(E::invalid_length(value.len(), &self)); } let mut out = [0u8; N]; out.copy_from_slice(value); Ok(out) } fn visit_byte_buf(self, value: Vec) -> Result where E: serde::de::Error, { self.visit_bytes(&value) } fn visit_seq(self, mut seq: A) -> Result where A: serde::de::SeqAccess<'de>, { let mut out = [0u8; N]; for (idx, slot) in out.iter_mut().enumerate() { *slot = seq .next_element()? .ok_or_else(|| serde::de::Error::invalid_length(idx, &self))?; } Ok(out) } } fn deserialize_fixed_bytes<'de, D, const N: usize>(deserializer: D) -> Result<[u8; N], D::Error> where D: serde::Deserializer<'de>, { deserializer.deserialize_bytes(FixedBytesVisitor::) } mod serde_bytes_16 { pub(super) fn serialize(value: &[u8; 16], serializer: S) -> Result where S: serde::Serializer, { serializer.serialize_bytes(value) } pub(super) fn deserialize<'de, D>(deserializer: D) -> Result<[u8; 16], D::Error> where D: serde::Deserializer<'de>, { super::deserialize_fixed_bytes::(deserializer) } } mod serde_bytes_32 { pub(super) fn serialize(value: &[u8; 32], serializer: S) -> Result where S: serde::Serializer, { serializer.serialize_bytes(value) } pub(super) fn deserialize<'de, D>(deserializer: D) -> Result<[u8; 32], D::Error> where D: serde::Deserializer<'de>, { super::deserialize_fixed_bytes::(deserializer) } } struct ByteVecVisitor; impl<'de> serde::de::Visitor<'de> for ByteVecVisitor { type Value = Vec; fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { formatter.write_str("byte vector") } fn visit_bytes(self, value: &[u8]) -> Result where E: serde::de::Error, { Ok(value.to_vec()) } fn visit_byte_buf(self, value: Vec) -> Result where E: serde::de::Error, { Ok(value) } fn visit_seq(self, mut seq: A) -> Result where A: serde::de::SeqAccess<'de>, { let mut out = Vec::with_capacity(seq.size_hint().unwrap_or(0)); while let Some(byte) = seq.next_element()? { out.push(byte); } Ok(out) } } mod serde_byte_vec { pub(super) fn serialize(value: &[u8], serializer: S) -> Result where S: serde::Serializer, { serializer.serialize_bytes(value) } pub(super) fn deserialize<'de, D>(deserializer: D) -> Result, D::Error> where D: serde::Deserializer<'de>, { deserializer.deserialize_bytes(super::ByteVecVisitor) } } mod serde_optional_byte_vec { pub(super) fn serialize(value: &Option>, serializer: S) -> Result where S: serde::Serializer, { match value { Some(bytes) => serializer.serialize_some(bytes), None => serializer.serialize_none(), } } pub(super) fn deserialize<'de, D>(deserializer: D) -> Result>, D::Error> where D: serde::Deserializer<'de>, { struct OptionalByteVecVisitor; impl<'de> serde::de::Visitor<'de> for OptionalByteVecVisitor { type Value = Option>; fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { formatter.write_str("optional byte vector") } fn visit_none(self) -> Result where E: serde::de::Error, { Ok(None) } fn visit_some(self, deserializer: D) -> Result where D: serde::Deserializer<'de>, { deserializer .deserialize_bytes(super::ByteVecVisitor) .map(Some) } } deserializer.deserialize_option(OptionalByteVecVisitor) } } mod serde_optional_bytes_32 { pub(super) fn serialize(value: &Option<[u8; 32]>, serializer: S) -> Result where S: serde::Serializer, { match value { Some(bytes) => serializer.serialize_some(bytes.as_slice()), None => serializer.serialize_none(), } } pub(super) fn deserialize<'de, D>(deserializer: D) -> Result, D::Error> where D: serde::Deserializer<'de>, { struct OptionalBytes32Visitor; impl<'de> serde::de::Visitor<'de> for OptionalBytes32Visitor { type Value = Option<[u8; 32]>; fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { formatter.write_str("optional 32-byte array") } fn visit_none(self) -> Result where E: serde::de::Error, { Ok(None) } fn visit_some(self, deserializer: D) -> Result where D: serde::Deserializer<'de>, { super::deserialize_fixed_bytes::(deserializer).map(Some) } } deserializer.deserialize_option(OptionalBytes32Visitor) } } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum VcirLocalOutputPayload { Vrp { asn: u32, afi: crate::data_model::roa::RoaAfi, prefix_len: u16, #[serde(with = "serde_bytes_16")] addr: [u8; 16], max_length: u16, }, Aspa { customer_as_id: u32, provider_as_ids: Vec, }, RouterKey { as_id: u32, #[serde(with = "serde_byte_vec")] ski: Vec, #[serde(with = "serde_byte_vec")] spki_der: Vec, }, } impl VcirLocalOutputPayload { pub fn typed_body_bytes(&self) -> u64 { match self { Self::Vrp { .. } => 4 + 1 + 2 + 16 + 2, Self::Aspa { provider_as_ids, .. } => 4 + (provider_as_ids.len() as u64 * 4), Self::RouterKey { ski, spki_der, .. } => 4 + ski.len() as u64 + spki_der.len() as u64, } } } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct VcirLocalOutput { #[serde(rename = "t")] pub output_type: VcirOutputType, #[serde(rename = "e")] pub item_effective_until: PackTime, #[serde(rename = "u")] pub source_object_uri: String, #[serde(rename = "k")] pub source_object_type: VcirSourceObjectType, #[serde(rename = "h")] #[serde(with = "serde_bytes_32")] pub source_object_hash: [u8; 32], #[serde(rename = "c")] #[serde(with = "serde_bytes_32")] pub source_ee_cert_hash: [u8; 32], #[serde(rename = "p")] pub payload: VcirLocalOutputPayload, #[serde(rename = "r")] #[serde(with = "serde_bytes_32")] pub rule_hash: [u8; 32], } impl VcirLocalOutput { pub fn output_id(&self) -> String { self.rule_hash_hex() } pub fn source_object_hash_hex(&self) -> String { hex::encode(self.source_object_hash) } pub fn source_ee_cert_hash_hex(&self) -> String { hex::encode(self.source_ee_cert_hash) } pub fn rule_hash_hex(&self) -> String { hex::encode(self.rule_hash) } pub fn source_object_type_name(&self) -> &'static str { self.source_object_type.as_str() } pub fn payload_json(&self) -> String { match &self.payload { VcirLocalOutputPayload::Vrp { asn, afi, prefix_len, addr, max_length, } => { let prefix = match afi { crate::data_model::roa::RoaAfi::Ipv4 => { let ip = std::net::Ipv4Addr::new(addr[0], addr[1], addr[2], addr[3]); format!("{ip}/{prefix_len}") } crate::data_model::roa::RoaAfi::Ipv6 => { let ip = std::net::Ipv6Addr::from(*addr); format!("{ip}/{prefix_len}") } }; format!(r#"{{"asn":{asn},"max_length":{max_length},"prefix":"{prefix}"}}"#) } VcirLocalOutputPayload::Aspa { customer_as_id, provider_as_ids, } => { let providers = provider_as_ids .iter() .map(u32::to_string) .collect::>() .join(","); format!(r#"{{"customer_as_id":{customer_as_id},"provider_as_ids":[{providers}]}}"#) } VcirLocalOutputPayload::RouterKey { as_id, ski, spki_der, } => { let ski_hex = hex::encode(ski); let spki_der_base64 = base64::engine::general_purpose::STANDARD.encode(spki_der); format!( r#"{{"as_id":{as_id},"ski_hex":"{ski_hex}","spki_der_base64":"{spki_der_base64}"}}"# ) } } } pub fn validate_internal(&self) -> StorageResult<()> { parse_time( "vcir.local_outputs[].item_effective_until", &self.item_effective_until, )?; validate_non_empty( "vcir.local_outputs[].source_object_uri", &self.source_object_uri, )?; validate_local_output_type_matches_payload(self)?; Ok(()) } } fn validate_local_output_type_matches_payload(output: &VcirLocalOutput) -> StorageResult<()> { let matches_payload = matches!( (&output.output_type, &output.payload), (VcirOutputType::Vrp, VcirLocalOutputPayload::Vrp { .. }) | (VcirOutputType::Aspa, VcirLocalOutputPayload::Aspa { .. }) | ( VcirOutputType::RouterKey, VcirLocalOutputPayload::RouterKey { .. } ) ); if !matches_payload { return Err(StorageError::InvalidData { entity: "vcir.local_outputs[]", detail: "output_type must match payload variant".to_string(), }); } Ok(()) } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct RoaCacheCrlProjection { #[serde(rename = "u")] pub uri: String, #[serde(rename = "h")] pub sha256: String, } impl RoaCacheCrlProjection { pub fn validate_internal(&self) -> StorageResult<()> { validate_non_empty("roa_cache_projection.crls[].uri", &self.uri)?; validate_sha256_hex("roa_cache_projection.crls[].sha256", &self.sha256)?; Ok(()) } } #[derive(Clone, Debug, PartialEq, Eq)] pub struct RoaCacheObjectMeta { pub source_object_uri: String, pub source_object_hash: [u8; 32], pub ee_serial: Vec, pub crl_uri: String, } #[derive(Clone, Debug, PartialEq, Eq)] pub struct RoaCacheProjectionContext { pub parent_context_digest: [u8; 32], pub policy_fingerprint: [u8; 32], pub object_meta: Vec, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct RoaCacheLocalOutputProjection { #[serde(rename = "e")] pub item_effective_until: PackTime, #[serde(rename = "c")] #[serde(with = "serde_bytes_32")] pub source_ee_cert_hash: [u8; 32], #[serde(rename = "p")] pub payload: VcirLocalOutputPayload, #[serde(rename = "r")] #[serde(with = "serde_bytes_32")] pub rule_hash: [u8; 32], } impl RoaCacheLocalOutputProjection { fn from_local_output(output: &VcirLocalOutput) -> Option { if output.output_type != VcirOutputType::Vrp || output.source_object_type != VcirSourceObjectType::Roa { return None; } Some(Self { item_effective_until: output.item_effective_until.clone(), source_ee_cert_hash: output.source_ee_cert_hash, payload: output.payload.clone(), rule_hash: output.rule_hash, }) } pub fn validate_internal(&self) -> StorageResult<()> { parse_time( "roa_cache_projection.entries[].outputs[].item_effective_until", &self.item_effective_until, )?; if !matches!(self.payload, VcirLocalOutputPayload::Vrp { .. }) { return Err(StorageError::InvalidData { entity: "roa_cache_projection.entries[].outputs[]", detail: "payload must be VRP".to_string(), }); } Ok(()) } } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct RoaCacheObjectProjection { #[serde(rename = "u")] pub source_object_uri: String, #[serde(rename = "h")] #[serde(with = "serde_bytes_32")] pub source_object_hash: [u8; 32], #[serde(rename = "s", default, skip_serializing_if = "Option::is_none")] #[serde(with = "serde_optional_byte_vec")] pub ee_serial: Option>, #[serde(rename = "c", default, skip_serializing_if = "Option::is_none")] pub crl_uri: Option, #[serde(rename = "x")] pub outputs_effective_until_unix: i64, #[serde(rename = "o")] pub outputs: Vec, } impl RoaCacheObjectProjection { pub fn validate_internal(&self) -> StorageResult<()> { validate_non_empty( "roa_cache_projection.entries[].source_object_uri", &self.source_object_uri, )?; if let Some(serial) = &self.ee_serial { if serial.is_empty() { return Err(StorageError::InvalidData { entity: "roa_cache_projection.entries[].ee_serial", detail: "must not be empty when present".to_string(), }); } } if let Some(crl_uri) = &self.crl_uri { validate_non_empty("roa_cache_projection.entries[].crl_uri", crl_uri)?; } if self.outputs.is_empty() { return Err(StorageError::InvalidData { entity: "roa_cache_projection.entries[]", detail: "outputs must not be empty".to_string(), }); } for output in &self.outputs { output.validate_internal()?; } let expected_effective_until = self .outputs .iter() .map(|output| { output .item_effective_until .parse() .map(|time| time.unix_timestamp()) .map_err(|detail| StorageError::InvalidData { entity: "roa_cache_projection.entries[].outputs_effective_until_unix", detail, }) }) .collect::>>()? .into_iter() .min() .expect("outputs must not be empty"); if self.outputs_effective_until_unix != expected_effective_until { return Err(StorageError::InvalidData { entity: "roa_cache_projection.entries[].outputs_effective_until_unix", detail: "must equal the earliest output item_effective_until".to_string(), }); } Ok(()) } } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct RoaCacheProjection { #[serde(rename = "m")] pub manifest_rsync_uri: String, #[serde(rename = "e")] pub instance_effective_until: PackTime, #[serde(rename = "i")] pub issuer_ca_sha256_hex: Option, #[serde(rename = "p", default, skip_serializing_if = "Option::is_none")] #[serde(with = "serde_optional_bytes_32")] pub parent_context_digest: Option<[u8; 32]>, #[serde(rename = "f", default, skip_serializing_if = "Option::is_none")] #[serde(with = "serde_optional_bytes_32")] pub policy_fingerprint: Option<[u8; 32]>, #[serde(rename = "c")] pub crl_sha256_by_uri: Vec, #[serde(rename = "r")] pub entries: Vec, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct ChildCertificateCacheRouterKeyProjection { #[serde(rename = "a")] pub as_id: u32, #[serde(rename = "s")] #[serde(with = "serde_byte_vec")] pub ski: Vec, #[serde(rename = "p")] #[serde(with = "serde_byte_vec")] pub spki_der: Vec, #[serde(rename = "e")] pub item_effective_until: PackTime, } impl ChildCertificateCacheRouterKeyProjection { pub fn validate_internal(&self) -> StorageResult<()> { if self.ski.is_empty() { return Err(StorageError::InvalidData { entity: "child_certificate_cache_projection.router_keys[].ski", detail: "must not be empty".to_string(), }); } if self.spki_der.is_empty() { return Err(StorageError::InvalidData { entity: "child_certificate_cache_projection.router_keys[].spki_der", detail: "must not be empty".to_string(), }); } parse_time( "child_certificate_cache_projection.router_keys[].item_effective_until", &self.item_effective_until, )?; Ok(()) } } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum ChildCertificateCachePayload { ChildCa { #[serde(rename = "cm")] child_manifest_rsync_uri: String, #[serde(rename = "ski")] child_ski: String, #[serde(rename = "rb")] child_rsync_base_uri: String, #[serde(rename = "pp")] child_publication_point_rsync_uri: String, #[serde(rename = "rn")] child_rrdp_notification_uri: Option, #[serde(rename = "ip")] child_effective_ip_resources: Option, #[serde(rename = "as")] child_effective_as_resources: Option, }, Router { #[serde(rename = "r")] router_keys: Vec, }, } impl ChildCertificateCachePayload { pub fn validate_internal(&self) -> StorageResult<()> { match self { Self::ChildCa { child_manifest_rsync_uri, child_ski, child_rsync_base_uri, child_publication_point_rsync_uri, child_rrdp_notification_uri, .. } => { validate_non_empty( "child_certificate_cache_projection.child_manifest_rsync_uri", child_manifest_rsync_uri, )?; validate_non_empty("child_certificate_cache_projection.child_ski", child_ski)?; validate_non_empty( "child_certificate_cache_projection.child_rsync_base_uri", child_rsync_base_uri, )?; validate_non_empty( "child_certificate_cache_projection.child_publication_point_rsync_uri", child_publication_point_rsync_uri, )?; if let Some(uri) = child_rrdp_notification_uri { validate_non_empty( "child_certificate_cache_projection.child_rrdp_notification_uri", uri, )?; } Ok(()) } Self::Router { router_keys } => { if router_keys.is_empty() { return Err(StorageError::InvalidData { entity: "child_certificate_cache_projection.router_keys", detail: "must not be empty".to_string(), }); } for router_key in router_keys { router_key.validate_internal()?; } Ok(()) } } } } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct ChildCertificateCacheProjection { #[serde(rename = "sv")] pub schema_version: u32, #[serde(rename = "av")] pub algorithm_version: u32, #[serde(rename = "k")] pub cache_key_sha256_hex: String, #[serde(rename = "cu")] pub child_cert_uri: String, #[serde(rename = "ch")] pub child_cert_sha256_hex: String, #[serde(rename = "cs")] #[serde(with = "serde_byte_vec")] pub child_cert_serial: Vec, #[serde(rename = "ih")] pub issuer_ca_sha256_hex: String, #[serde(rename = "cru")] pub issuer_crl_uri: String, #[serde(rename = "crh")] pub issuer_crl_sha256_hex: String, #[serde(rename = "pc")] #[serde(with = "serde_bytes_32")] pub parent_context_digest: [u8; 32], #[serde(rename = "pf")] #[serde(with = "serde_bytes_32")] pub validation_policy_fingerprint: [u8; 32], #[serde(rename = "nb")] pub effective_not_before: PackTime, #[serde(rename = "nu")] pub effective_until: PackTime, #[serde(rename = "p")] pub payload: ChildCertificateCachePayload, } pub const CHILD_CERTIFICATE_CACHE_SCHEMA_VERSION: u32 = 2; pub const CHILD_CERTIFICATE_CACHE_ALGORITHM_VERSION: u32 = 3; impl ChildCertificateCacheProjection { pub fn validate_internal(&self) -> StorageResult<()> { if self.schema_version != CHILD_CERTIFICATE_CACHE_SCHEMA_VERSION { return Err(StorageError::InvalidData { entity: "child_certificate_cache_projection.schema_version", detail: format!("unsupported schema_version {}", self.schema_version), }); } if self.algorithm_version != CHILD_CERTIFICATE_CACHE_ALGORITHM_VERSION { return Err(StorageError::InvalidData { entity: "child_certificate_cache_projection.algorithm_version", detail: format!("unsupported algorithm_version {}", self.algorithm_version), }); } validate_sha256_hex( "child_certificate_cache_projection.cache_key_sha256_hex", &self.cache_key_sha256_hex, )?; validate_non_empty( "child_certificate_cache_projection.child_cert_uri", &self.child_cert_uri, )?; validate_sha256_hex( "child_certificate_cache_projection.child_cert_sha256_hex", &self.child_cert_sha256_hex, )?; if self.child_cert_serial.is_empty() { return Err(StorageError::InvalidData { entity: "child_certificate_cache_projection.child_cert_serial", detail: "must not be empty".to_string(), }); } validate_sha256_hex( "child_certificate_cache_projection.issuer_ca_sha256_hex", &self.issuer_ca_sha256_hex, )?; validate_non_empty( "child_certificate_cache_projection.issuer_crl_uri", &self.issuer_crl_uri, )?; validate_sha256_hex( "child_certificate_cache_projection.issuer_crl_sha256_hex", &self.issuer_crl_sha256_hex, )?; let effective_not_before = parse_time( "child_certificate_cache_projection.effective_not_before", &self.effective_not_before, )?; let effective_until = parse_time( "child_certificate_cache_projection.effective_until", &self.effective_until, )?; if effective_not_before >= effective_until { return Err(StorageError::InvalidData { entity: "child_certificate_cache_projection.effective_window", detail: "effective_not_before must be before effective_until".to_string(), }); } self.payload.validate_internal()?; Ok(()) } } pub const PUBLICATION_POINT_CACHE_SCHEMA_VERSION: u32 = 1; pub const PUBLICATION_POINT_CACHE_ALGORITHM_VERSION: u32 = 1; #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct PublicationPointCacheOutput { #[serde(rename = "t")] pub output_type: VcirOutputType, #[serde(rename = "nb")] pub item_effective_not_before: PackTime, #[serde(rename = "nu")] pub item_effective_until: PackTime, #[serde(rename = "u")] pub source_object_uri: String, #[serde(rename = "k")] pub source_object_type: VcirSourceObjectType, #[serde(rename = "h")] #[serde(with = "serde_bytes_32")] pub source_object_hash: [u8; 32], #[serde(rename = "c")] #[serde(with = "serde_bytes_32")] pub source_ee_cert_hash: [u8; 32], #[serde(rename = "p")] pub payload: VcirLocalOutputPayload, #[serde(rename = "r")] #[serde(with = "serde_bytes_32")] pub rule_hash: [u8; 32], } impl PublicationPointCacheOutput { fn from_local_output(output: &VcirLocalOutput, default_not_before: &PackTime) -> Self { Self { output_type: output.output_type, item_effective_not_before: default_not_before.clone(), item_effective_until: output.item_effective_until.clone(), source_object_uri: output.source_object_uri.clone(), source_object_type: output.source_object_type, source_object_hash: output.source_object_hash, source_ee_cert_hash: output.source_ee_cert_hash, payload: output.payload.clone(), rule_hash: output.rule_hash, } } pub fn to_local_output(&self) -> VcirLocalOutput { VcirLocalOutput { output_type: self.output_type, item_effective_until: self.item_effective_until.clone(), source_object_uri: self.source_object_uri.clone(), source_object_type: self.source_object_type, source_object_hash: self.source_object_hash, source_ee_cert_hash: self.source_ee_cert_hash, payload: self.payload.clone(), rule_hash: self.rule_hash, } } pub fn validate_internal(&self) -> StorageResult<()> { parse_time( "publication_point_cache_projection.outputs[].item_effective_not_before", &self.item_effective_not_before, )?; parse_time( "publication_point_cache_projection.outputs[].item_effective_until", &self.item_effective_until, )?; validate_non_empty( "publication_point_cache_projection.outputs[].source_object_uri", &self.source_object_uri, )?; VcirLocalOutput { output_type: self.output_type, item_effective_until: self.item_effective_until.clone(), source_object_uri: self.source_object_uri.clone(), source_object_type: self.source_object_type, source_object_hash: self.source_object_hash, source_ee_cert_hash: self.source_ee_cert_hash, payload: self.payload.clone(), rule_hash: self.rule_hash, } .validate_internal() } } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct PublicationPointCacheChild { #[serde(rename = "cm")] pub child_manifest_rsync_uri: String, #[serde(rename = "cu")] pub child_cert_rsync_uri: String, #[serde(rename = "ch")] pub child_cert_hash: String, #[serde(rename = "ski")] pub child_ski: String, #[serde(rename = "rb")] pub child_rsync_base_uri: String, #[serde(rename = "pp")] pub child_publication_point_rsync_uri: String, #[serde(rename = "rn")] pub child_rrdp_notification_uri: Option, #[serde(rename = "ip")] pub child_effective_ip_resources: Option, #[serde(rename = "as")] pub child_effective_as_resources: Option, #[serde(rename = "nb")] pub child_effective_not_before: PackTime, #[serde(rename = "nu")] pub child_effective_until: PackTime, } impl PublicationPointCacheChild { fn from_child_entry( entry: &VcirChildEntry, default_not_before: &PackTime, default_until: &PackTime, ) -> Self { Self { child_manifest_rsync_uri: entry.child_manifest_rsync_uri.clone(), child_cert_rsync_uri: entry.child_cert_rsync_uri.clone(), child_cert_hash: entry.child_cert_hash.clone(), child_ski: entry.child_ski.clone(), child_rsync_base_uri: entry.child_rsync_base_uri.clone(), child_publication_point_rsync_uri: entry.child_publication_point_rsync_uri.clone(), child_rrdp_notification_uri: entry.child_rrdp_notification_uri.clone(), child_effective_ip_resources: entry.child_effective_ip_resources.clone(), child_effective_as_resources: entry.child_effective_as_resources.clone(), child_effective_not_before: default_not_before.clone(), child_effective_until: default_until.clone(), } } pub fn to_child_entry(&self, accepted_at_validation_time: PackTime) -> VcirChildEntry { VcirChildEntry { child_manifest_rsync_uri: self.child_manifest_rsync_uri.clone(), child_cert_rsync_uri: self.child_cert_rsync_uri.clone(), child_cert_hash: self.child_cert_hash.clone(), child_ski: self.child_ski.clone(), child_rsync_base_uri: self.child_rsync_base_uri.clone(), child_publication_point_rsync_uri: self.child_publication_point_rsync_uri.clone(), child_rrdp_notification_uri: self.child_rrdp_notification_uri.clone(), child_effective_ip_resources: self.child_effective_ip_resources.clone(), child_effective_as_resources: self.child_effective_as_resources.clone(), accepted_at_validation_time, } } pub fn validate_internal(&self) -> StorageResult<()> { self.to_child_entry(self.child_effective_not_before.clone()) .validate_internal()?; parse_time( "publication_point_cache_projection.children[].child_effective_not_before", &self.child_effective_not_before, )?; parse_time( "publication_point_cache_projection.children[].child_effective_until", &self.child_effective_until, )?; Ok(()) } } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct PublicationPointCacheObject { #[serde(rename = "r")] pub artifact_role: VcirArtifactRole, #[serde(rename = "k")] pub artifact_kind: VcirArtifactKind, #[serde(rename = "u")] pub uri: Option, #[serde(rename = "h")] pub sha256: String, #[serde(rename = "t")] pub object_type: Option, #[serde(rename = "s")] pub validation_status: VcirArtifactValidationStatus, } impl PublicationPointCacheObject { fn from_related_artifact(artifact: &VcirRelatedArtifact) -> Self { Self { artifact_role: artifact.artifact_role, artifact_kind: artifact.artifact_kind, uri: artifact.uri.clone(), sha256: artifact.sha256.clone(), object_type: artifact.object_type.clone(), validation_status: artifact.validation_status, } } pub fn to_related_artifact(&self) -> VcirRelatedArtifact { VcirRelatedArtifact { artifact_role: self.artifact_role, artifact_kind: self.artifact_kind, uri: self.uri.clone(), sha256: self.sha256.clone(), object_type: self.object_type.clone(), validation_status: self.validation_status, } } pub fn validate_internal(&self) -> StorageResult<()> { self.to_related_artifact().validate_internal() } } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct PublicationPointCacheProjection { #[serde(rename = "sv")] pub schema_version: u32, #[serde(rename = "av")] pub algorithm_version: u32, #[serde(rename = "m")] pub manifest_rsync_uri: String, #[serde(rename = "pp")] pub publication_point_rsync_uri: String, #[serde(rename = "cu")] pub ca_cert_uri: Option, #[serde(rename = "ch")] #[serde(with = "serde_bytes_32")] pub ca_cert_sha256: [u8; 32], #[serde(rename = "mh")] #[serde(with = "serde_bytes_32")] pub manifest_sha256: [u8; 32], #[serde(rename = "tal")] pub tal_id: String, #[serde(rename = "ta")] #[serde(with = "serde_bytes_32")] pub ta_context_digest: [u8; 32], #[serde(rename = "pc")] #[serde(with = "serde_bytes_32")] pub parent_context_digest: [u8; 32], #[serde(rename = "pf")] #[serde(with = "serde_bytes_32")] pub validation_policy_fingerprint: [u8; 32], #[serde(rename = "nb")] pub instance_effective_not_before: PackTime, #[serde(rename = "nu")] pub instance_effective_until: PackTime, #[serde(rename = "mt")] pub manifest_this_update: PackTime, #[serde(rename = "mn")] pub manifest_next_update: PackTime, #[serde(rename = "cn")] pub current_crl_next_update: PackTime, #[serde(rename = "ca")] pub self_ca_not_after: PackTime, #[serde(rename = "ccr")] pub ccr_manifest_projection: VcirCcrManifestProjection, #[serde(rename = "o")] pub outputs: Vec, #[serde(rename = "c")] pub children: Vec, #[serde(rename = "ra")] pub related_objects: Vec, #[serde(rename = "s")] pub summary: VcirSummary, } impl PublicationPointCacheProjection { pub fn from_vcir_with_context( vcir: &ValidatedCaInstanceResult, publication_point_rsync_uri: String, ca_cert_uri: Option, ca_cert_sha256: [u8; 32], manifest_sha256: [u8; 32], ta_context_digest: [u8; 32], parent_context_digest: [u8; 32], validation_policy_fingerprint: [u8; 32], ) -> StorageResult { let projection = Self { schema_version: PUBLICATION_POINT_CACHE_SCHEMA_VERSION, algorithm_version: PUBLICATION_POINT_CACHE_ALGORITHM_VERSION, manifest_rsync_uri: vcir.manifest_rsync_uri.clone(), publication_point_rsync_uri, ca_cert_uri, ca_cert_sha256, manifest_sha256, tal_id: vcir.tal_id.clone(), ta_context_digest, parent_context_digest, validation_policy_fingerprint, instance_effective_not_before: vcir.last_successful_validation_time.clone(), instance_effective_until: vcir.instance_gate.instance_effective_until.clone(), manifest_this_update: vcir .validated_manifest_meta .validated_manifest_this_update .clone(), manifest_next_update: vcir.instance_gate.manifest_next_update.clone(), current_crl_next_update: vcir.instance_gate.current_crl_next_update.clone(), self_ca_not_after: vcir.instance_gate.self_ca_not_after.clone(), ccr_manifest_projection: vcir.ccr_manifest_projection.clone(), outputs: vcir .local_outputs .iter() .map(|output| { PublicationPointCacheOutput::from_local_output( output, &vcir.last_successful_validation_time, ) }) .collect(), children: vcir .child_entries .iter() .map(|child| { PublicationPointCacheChild::from_child_entry( child, &vcir.last_successful_validation_time, &vcir.instance_gate.instance_effective_until, ) }) .collect(), related_objects: vcir .related_artifacts .iter() .map(PublicationPointCacheObject::from_related_artifact) .collect(), summary: vcir.summary.clone(), }; projection.validate_internal()?; Ok(projection) } pub fn to_vcir_for_reuse( &self, validation_time: time::OffsetDateTime, ) -> ValidatedCaInstanceResult { let validation_time = PackTime::from_utc_offset_datetime(validation_time); ValidatedCaInstanceResult { manifest_rsync_uri: self.manifest_rsync_uri.clone(), parent_manifest_rsync_uri: None, tal_id: self.tal_id.clone(), ca_subject_name: String::from("publication-point-cache"), ca_ski: hex::encode(self.ca_cert_sha256), issuer_ski: hex::encode(self.parent_context_digest), last_successful_validation_time: validation_time.clone(), current_manifest_rsync_uri: self.manifest_rsync_uri.clone(), current_crl_rsync_uri: self .related_objects .iter() .find(|object| object.artifact_role == VcirArtifactRole::CurrentCrl) .and_then(|object| object.uri.clone()) .unwrap_or_default(), validated_manifest_meta: ValidatedManifestMeta { validated_manifest_number: self.ccr_manifest_projection.manifest_number_be.clone(), validated_manifest_this_update: self.manifest_this_update.clone(), validated_manifest_next_update: self.manifest_next_update.clone(), }, ccr_manifest_projection: self.ccr_manifest_projection.clone(), instance_gate: VcirInstanceGate { manifest_next_update: self.manifest_next_update.clone(), current_crl_next_update: self.current_crl_next_update.clone(), self_ca_not_after: self.self_ca_not_after.clone(), instance_effective_until: self.instance_effective_until.clone(), }, child_entries: self .children .iter() .map(|child| child.to_child_entry(validation_time.clone())) .collect(), local_outputs: self .outputs .iter() .map(PublicationPointCacheOutput::to_local_output) .collect(), related_artifacts: self .related_objects .iter() .map(PublicationPointCacheObject::to_related_artifact) .collect(), summary: self.summary.clone(), audit_summary: VcirAuditSummary { failed_fetch_eligible: false, last_failed_fetch_reason: None, warning_count: 0, audit_flags: vec!["publication_point_cache_projection".to_string()], }, } } pub fn validate_internal(&self) -> StorageResult<()> { if self.schema_version != PUBLICATION_POINT_CACHE_SCHEMA_VERSION { return Err(StorageError::InvalidData { entity: "publication_point_cache_projection.schema_version", detail: format!("unsupported schema_version {}", self.schema_version), }); } if self.algorithm_version != PUBLICATION_POINT_CACHE_ALGORITHM_VERSION { return Err(StorageError::InvalidData { entity: "publication_point_cache_projection.algorithm_version", detail: format!("unsupported algorithm_version {}", self.algorithm_version), }); } validate_non_empty( "publication_point_cache_projection.manifest_rsync_uri", &self.manifest_rsync_uri, )?; validate_non_empty( "publication_point_cache_projection.publication_point_rsync_uri", &self.publication_point_rsync_uri, )?; if let Some(uri) = &self.ca_cert_uri { validate_non_empty("publication_point_cache_projection.ca_cert_uri", uri)?; } validate_non_empty("publication_point_cache_projection.tal_id", &self.tal_id)?; parse_time( "publication_point_cache_projection.instance_effective_not_before", &self.instance_effective_not_before, )?; parse_time( "publication_point_cache_projection.instance_effective_until", &self.instance_effective_until, )?; parse_time( "publication_point_cache_projection.manifest_this_update", &self.manifest_this_update, )?; parse_time( "publication_point_cache_projection.manifest_next_update", &self.manifest_next_update, )?; parse_time( "publication_point_cache_projection.current_crl_next_update", &self.current_crl_next_update, )?; parse_time( "publication_point_cache_projection.self_ca_not_after", &self.self_ca_not_after, )?; self.ccr_manifest_projection.validate_internal()?; for output in &self.outputs { output.validate_internal()?; } for child in &self.children { child.validate_internal()?; } for object in &self.related_objects { object.validate_internal()?; } Ok(()) } } impl RoaCacheProjection { pub fn from_vcir(vcir: &ValidatedCaInstanceResult) -> StorageResult> { Self::from_vcir_with_context(vcir, None) } pub fn from_vcir_with_context( vcir: &ValidatedCaInstanceResult, context: Option<&RoaCacheProjectionContext>, ) -> StorageResult> { let mut issuer_ca_sha256_hex = None; let mut crl_sha256_by_uri = Vec::new(); for artifact in &vcir.related_artifacts { if artifact.validation_status != VcirArtifactValidationStatus::Accepted { continue; } match (artifact.artifact_role, artifact.artifact_kind) { ( VcirArtifactRole::IssuerCert | VcirArtifactRole::TrustAnchorCert, VcirArtifactKind::Cer, ) => { issuer_ca_sha256_hex = Some(artifact.sha256.clone()); } (_, VcirArtifactKind::Crl) => { if let Some(uri) = artifact.uri.as_ref() { crl_sha256_by_uri.push(RoaCacheCrlProjection { uri: uri.clone(), sha256: artifact.sha256.clone(), }); } } _ => {} } } crl_sha256_by_uri.sort_by(|left, right| left.uri.cmp(&right.uri)); let meta_by_uri = context.map(|context| { context .object_meta .iter() .map(|meta| (meta.source_object_uri.as_str(), meta)) .collect::>() }); let mut entries: Vec = Vec::new(); let mut entry_index_by_uri: HashMap = HashMap::new(); for output in &vcir.local_outputs { let Some(projected_output) = RoaCacheLocalOutputProjection::from_local_output(output) else { continue; }; let projected_output_effective_until = projected_output .item_effective_until .parse() .map(|time| time.unix_timestamp()) .map_err(|detail| StorageError::InvalidData { entity: "roa_cache_projection.entries[].outputs_effective_until_unix", detail, })?; let meta = meta_by_uri .as_ref() .and_then(|meta| meta.get(output.source_object_uri.as_str()).copied()); if context.is_some() && meta.is_none() { continue; } if let Some(meta) = meta { if meta.source_object_hash != output.source_object_hash { return Err(StorageError::InvalidData { entity: "roa_cache_projection.entries[]", detail: format!( "metadata source object hash mismatch for {}", output.source_object_uri ), }); } } if let Some(entry_index) = entry_index_by_uri.get(output.source_object_uri.as_str()) { let entry = &mut entries[*entry_index]; if entry.source_object_hash != output.source_object_hash { return Err(StorageError::InvalidData { entity: "roa_cache_projection.entries[]", detail: format!( "source object hash mismatch for {}", output.source_object_uri ), }); } entry.outputs_effective_until_unix = entry .outputs_effective_until_unix .min(projected_output_effective_until); entry.outputs.push(projected_output); } else { entry_index_by_uri.insert(output.source_object_uri.clone(), entries.len()); entries.push(RoaCacheObjectProjection { source_object_uri: output.source_object_uri.clone(), source_object_hash: output.source_object_hash, ee_serial: meta.map(|meta| meta.ee_serial.clone()), crl_uri: meta.map(|meta| meta.crl_uri.clone()), outputs_effective_until_unix: projected_output_effective_until, outputs: vec![projected_output], }); } } if entries.is_empty() { return Ok(None); } entries.sort_by(|left, right| left.source_object_uri.cmp(&right.source_object_uri)); let projection = Self { manifest_rsync_uri: vcir.manifest_rsync_uri.clone(), instance_effective_until: vcir.instance_gate.instance_effective_until.clone(), issuer_ca_sha256_hex, parent_context_digest: context.map(|context| context.parent_context_digest), policy_fingerprint: context.map(|context| context.policy_fingerprint), crl_sha256_by_uri, entries, }; projection.validate_internal()?; Ok(Some(projection)) } pub fn validate_internal(&self) -> StorageResult<()> { validate_non_empty( "roa_cache_projection.manifest_rsync_uri", &self.manifest_rsync_uri, )?; parse_time( "roa_cache_projection.instance_effective_until", &self.instance_effective_until, )?; if let Some(hash) = &self.issuer_ca_sha256_hex { validate_sha256_hex("roa_cache_projection.issuer_ca_sha256_hex", hash)?; } if self.parent_context_digest.is_some() != self.policy_fingerprint.is_some() { return Err(StorageError::InvalidData { entity: "roa_cache_projection.context", detail: "parent_context_digest and policy_fingerprint must be both present or both absent" .to_string(), }); } let mut seen_crls = HashSet::with_capacity(self.crl_sha256_by_uri.len()); for crl in &self.crl_sha256_by_uri { crl.validate_internal()?; if !seen_crls.insert(crl.uri.as_str()) { return Err(StorageError::InvalidData { entity: "roa_cache_projection.crls[]", detail: format!("duplicate CRL URI: {}", crl.uri), }); } } if self.entries.is_empty() { return Err(StorageError::InvalidData { entity: "roa_cache_projection.entries", detail: "must not be empty".to_string(), }); } let mut seen_entries = HashSet::with_capacity(self.entries.len()); for entry in &self.entries { entry.validate_internal()?; if !seen_entries.insert(entry.source_object_uri.as_str()) { return Err(StorageError::InvalidData { entity: "roa_cache_projection.entries[]", detail: format!("duplicate ROA URI: {}", entry.source_object_uri), }); } } Ok(()) } } #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum VcirArtifactRole { Manifest, CurrentCrl, ChildCaCert, SignedObject, EeCert, IssuerCert, Tal, TrustAnchorCert, Other, } #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum VcirArtifactKind { Cer, Crl, Mft, Roa, Aspa, Gbr, Tal, Other, } #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum VcirArtifactValidationStatus { Accepted, Rejected, WarningOnly, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct VcirRelatedArtifact { #[serde(rename = "r")] pub artifact_role: VcirArtifactRole, #[serde(rename = "k")] pub artifact_kind: VcirArtifactKind, #[serde(rename = "u")] pub uri: Option, #[serde(rename = "h")] pub sha256: String, #[serde(rename = "t")] pub object_type: Option, #[serde(rename = "s")] pub validation_status: VcirArtifactValidationStatus, } impl VcirRelatedArtifact { pub fn validate_internal(&self) -> StorageResult<()> { if let Some(uri) = &self.uri { validate_non_empty("vcir.related_artifacts[].uri", uri)?; } validate_sha256_hex("vcir.related_artifacts[].sha256", &self.sha256)?; Ok(()) } } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct VcirSummary { #[serde(rename = "v")] pub local_vrp_count: u32, #[serde(rename = "a")] pub local_aspa_count: u32, #[serde(rename = "r")] pub local_router_key_count: u32, #[serde(rename = "c")] pub child_count: u32, #[serde(rename = "o")] pub accepted_object_count: u32, #[serde(rename = "x")] pub rejected_object_count: u32, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct VcirAuditSummary { #[serde(rename = "f")] pub failed_fetch_eligible: bool, #[serde(rename = "r")] pub last_failed_fetch_reason: Option, #[serde(rename = "w")] pub warning_count: u32, #[serde(rename = "a")] pub audit_flags: Vec, } impl VcirAuditSummary { pub fn validate_internal(&self) -> StorageResult<()> { if let Some(reason) = &self.last_failed_fetch_reason { validate_non_empty("vcir.audit_summary.last_failed_fetch_reason", reason)?; } for flag in &self.audit_flags { validate_non_empty("vcir.audit_summary.audit_flags[]", flag)?; } Ok(()) } } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct ValidatedCaInstanceResult { #[serde(rename = "m")] pub manifest_rsync_uri: String, #[serde(rename = "pm")] pub parent_manifest_rsync_uri: Option, #[serde(rename = "tal")] pub tal_id: String, #[serde(rename = "subj")] pub ca_subject_name: String, #[serde(rename = "ski")] pub ca_ski: String, #[serde(rename = "aki")] pub issuer_ski: String, #[serde(rename = "vt")] pub last_successful_validation_time: PackTime, #[serde(rename = "cm")] pub current_manifest_rsync_uri: String, #[serde(rename = "crl")] pub current_crl_rsync_uri: String, #[serde(rename = "mm")] pub validated_manifest_meta: ValidatedManifestMeta, #[serde(rename = "ccr")] pub ccr_manifest_projection: VcirCcrManifestProjection, #[serde(rename = "g")] pub instance_gate: VcirInstanceGate, #[serde(rename = "ch")] pub child_entries: Vec, #[serde(rename = "lo")] pub local_outputs: Vec, #[serde(rename = "ra")] pub related_artifacts: Vec, #[serde(rename = "s")] pub summary: VcirSummary, #[serde(rename = "as")] pub audit_summary: VcirAuditSummary, } #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)] pub struct VcirFieldSizeBreakdown { pub local_output_count: u64, pub local_output_source_uri_bytes: u64, pub local_output_source_type_bytes: u64, pub local_output_source_hash_hex_bytes: u64, pub local_output_source_ee_hash_hex_bytes: u64, pub local_output_payload_json_bytes: u64, pub local_output_rule_hash_hex_bytes: u64, pub local_output_source_hash_binary_bytes: u64, pub local_output_source_ee_hash_binary_bytes: u64, pub local_output_payload_typed_body_bytes: u64, pub local_output_rule_hash_binary_bytes: u64, pub related_artifact_count: u64, pub related_artifact_uri_bytes: u64, pub related_artifact_hash_hex_bytes: u64, pub related_artifact_type_bytes: u64, pub child_entry_count: u64, pub child_entry_uri_bytes: u64, pub child_entry_hash_hex_bytes: u64, } fn serialized_cbor_len(value: &T) -> u64 { serde_cbor::to_vec(value) .map(|bytes| bytes.len() as u64) .unwrap_or(0) } impl VcirFieldSizeBreakdown { pub fn from_vcir(vcir: &ValidatedCaInstanceResult) -> Self { let mut out = Self::default(); out.local_output_count = vcir.local_outputs.len() as u64; for local in &vcir.local_outputs { out.local_output_source_uri_bytes += local.source_object_uri.len() as u64; out.local_output_source_type_bytes += local.source_object_type_name().len() as u64; out.local_output_source_hash_hex_bytes += 64; out.local_output_source_ee_hash_hex_bytes += 64; out.local_output_payload_json_bytes += local.payload_json().len() as u64; out.local_output_rule_hash_hex_bytes += 64; out.local_output_source_hash_binary_bytes += 32; out.local_output_source_ee_hash_binary_bytes += 32; out.local_output_payload_typed_body_bytes += local.payload.typed_body_bytes(); out.local_output_rule_hash_binary_bytes += 32; } out.related_artifact_count = vcir.related_artifacts.len() as u64; for artifact in &vcir.related_artifacts { if let Some(uri) = &artifact.uri { out.related_artifact_uri_bytes += uri.len() as u64; } out.related_artifact_hash_hex_bytes += artifact.sha256.len() as u64; if let Some(object_type) = &artifact.object_type { out.related_artifact_type_bytes += object_type.len() as u64; } } out.child_entry_count = vcir.child_entries.len() as u64; for child in &vcir.child_entries { out.child_entry_uri_bytes += child.child_manifest_rsync_uri.len() as u64 + child.child_cert_rsync_uri.len() as u64 + child.child_rsync_base_uri.len() as u64 + child.child_publication_point_rsync_uri.len() as u64 + child .child_rrdp_notification_uri .as_ref() .map(|uri| uri.len() as u64) .unwrap_or(0); out.child_entry_hash_hex_bytes += child.child_cert_hash.len() as u64 + child.child_ski.len() as u64; } out } pub fn add_assign(&mut self, other: &Self) { self.local_output_count += other.local_output_count; self.local_output_source_uri_bytes += other.local_output_source_uri_bytes; self.local_output_source_type_bytes += other.local_output_source_type_bytes; self.local_output_source_hash_hex_bytes += other.local_output_source_hash_hex_bytes; self.local_output_source_ee_hash_hex_bytes += other.local_output_source_ee_hash_hex_bytes; self.local_output_payload_json_bytes += other.local_output_payload_json_bytes; self.local_output_rule_hash_hex_bytes += other.local_output_rule_hash_hex_bytes; self.local_output_source_hash_binary_bytes += other.local_output_source_hash_binary_bytes; self.local_output_source_ee_hash_binary_bytes += other.local_output_source_ee_hash_binary_bytes; self.local_output_payload_typed_body_bytes += other.local_output_payload_typed_body_bytes; self.local_output_rule_hash_binary_bytes += other.local_output_rule_hash_binary_bytes; self.related_artifact_count += other.related_artifact_count; self.related_artifact_uri_bytes += other.related_artifact_uri_bytes; self.related_artifact_hash_hex_bytes += other.related_artifact_hash_hex_bytes; self.related_artifact_type_bytes += other.related_artifact_type_bytes; self.child_entry_count += other.child_entry_count; self.child_entry_uri_bytes += other.child_entry_uri_bytes; self.child_entry_hash_hex_bytes += other.child_entry_hash_hex_bytes; } pub fn local_output_old_projection_bytes(&self) -> u64 { self.local_output_source_type_bytes + self.local_output_source_hash_hex_bytes + self.local_output_source_ee_hash_hex_bytes + self.local_output_payload_json_bytes + self.local_output_rule_hash_hex_bytes } pub fn local_output_typed_projection_bytes(&self) -> u64 { self.local_output_count + self.local_output_source_hash_binary_bytes + self.local_output_source_ee_hash_binary_bytes + self.local_output_payload_typed_body_bytes + self.local_output_rule_hash_binary_bytes } pub fn local_output_projection_saved_bytes(&self) -> u64 { self.local_output_old_projection_bytes() .saturating_sub(self.local_output_typed_projection_bytes()) } } #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)] pub struct VcirReplaceTimingBreakdown { pub validate_ms: u64, pub vcir_encode_ms: u64, pub vcir_value_bytes: u64, pub replay_meta_encode_ms: u64, pub replay_meta_value_bytes: u64, pub roa_cache_projection_encode_ms: u64, pub roa_cache_projection_value_bytes: u64, pub publication_point_cache_projection_encode_ms: u64, pub publication_point_cache_projection_value_bytes: u64, pub batch_build_ms: u64, pub write_batch_ms: u64, pub total_encoded_bytes: u64, pub field_sizes: VcirFieldSizeBreakdown, pub rss_before_kb: Option, pub rss_after_validate_kb: Option, pub rss_after_vcir_encode_kb: Option, pub rss_after_replay_meta_encode_kb: Option, pub rss_after_roa_cache_projection_encode_kb: Option, pub rss_after_publication_point_cache_projection_encode_kb: Option, pub rss_after_write_batch_kb: Option, } #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)] pub struct VcirStorageSummary { pub entry_count: u64, pub vcir_value_bytes: u64, pub vcir_value_bytes_max: u64, pub vcir_value_bytes_max_manifest_rsync_uri: Option, pub core_fields: VcirCoreFieldSizeBreakdown, pub ccr_projection: VcirCcrProjectionSizeBreakdown, pub child_resources: VcirChildResourceSizeBreakdown, pub field_sizes: VcirFieldSizeBreakdown, pub local_output_old_projection_bytes: u64, pub local_output_typed_projection_bytes: u64, pub local_output_projection_saved_bytes: u64, pub top_entries_by_vcir_value_bytes: Vec, } #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)] pub struct VcirStorageEntrySummary { pub manifest_rsync_uri: String, pub vcir_value_bytes: u64, pub local_vrp_count: u32, pub local_aspa_count: u32, pub local_router_key_count: u32, pub accepted_object_count: u32, pub rejected_object_count: u32, pub child_count: u32, pub core_fields: VcirCoreFieldSizeBreakdown, pub ccr_projection: VcirCcrProjectionSizeBreakdown, pub child_resources: VcirChildResourceSizeBreakdown, pub field_sizes: VcirFieldSizeBreakdown, pub local_output_old_projection_bytes: u64, pub local_output_typed_projection_bytes: u64, pub local_output_projection_saved_bytes: u64, } impl VcirStorageEntrySummary { fn from_vcir(vcir: &ValidatedCaInstanceResult, vcir_value_bytes: u64) -> Self { let field_sizes = VcirFieldSizeBreakdown::from_vcir(vcir); Self { manifest_rsync_uri: vcir.manifest_rsync_uri.clone(), vcir_value_bytes, local_vrp_count: vcir.summary.local_vrp_count, local_aspa_count: vcir.summary.local_aspa_count, local_router_key_count: vcir.summary.local_router_key_count, accepted_object_count: vcir.summary.accepted_object_count, rejected_object_count: vcir.summary.rejected_object_count, child_count: vcir.summary.child_count, core_fields: VcirCoreFieldSizeBreakdown::from_vcir(vcir), ccr_projection: VcirCcrProjectionSizeBreakdown::from_projection( &vcir.ccr_manifest_projection, ), child_resources: VcirChildResourceSizeBreakdown::from_vcir(vcir), local_output_old_projection_bytes: field_sizes.local_output_old_projection_bytes(), local_output_typed_projection_bytes: field_sizes.local_output_typed_projection_bytes(), local_output_projection_saved_bytes: field_sizes.local_output_projection_saved_bytes(), field_sizes, } } } #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)] pub struct VcirCoreFieldSizeBreakdown { pub manifest_rsync_uri_bytes: u64, pub parent_manifest_rsync_uri_bytes: u64, pub tal_id_bytes: u64, pub ca_subject_name_bytes: u64, pub ca_ski_bytes: u64, pub issuer_ski_bytes: u64, pub current_manifest_rsync_uri_bytes: u64, pub current_crl_rsync_uri_bytes: u64, pub validated_manifest_number_bytes: u64, } impl VcirCoreFieldSizeBreakdown { fn from_vcir(vcir: &ValidatedCaInstanceResult) -> Self { Self { manifest_rsync_uri_bytes: vcir.manifest_rsync_uri.len() as u64, parent_manifest_rsync_uri_bytes: vcir .parent_manifest_rsync_uri .as_ref() .map(|uri| uri.len() as u64) .unwrap_or(0), tal_id_bytes: vcir.tal_id.len() as u64, ca_subject_name_bytes: vcir.ca_subject_name.len() as u64, ca_ski_bytes: vcir.ca_ski.len() as u64, issuer_ski_bytes: vcir.issuer_ski.len() as u64, current_manifest_rsync_uri_bytes: vcir.current_manifest_rsync_uri.len() as u64, current_crl_rsync_uri_bytes: vcir.current_crl_rsync_uri.len() as u64, validated_manifest_number_bytes: vcir .validated_manifest_meta .validated_manifest_number .len() as u64, } } fn add_assign(&mut self, other: &Self) { self.manifest_rsync_uri_bytes += other.manifest_rsync_uri_bytes; self.parent_manifest_rsync_uri_bytes += other.parent_manifest_rsync_uri_bytes; self.tal_id_bytes += other.tal_id_bytes; self.ca_subject_name_bytes += other.ca_subject_name_bytes; self.ca_ski_bytes += other.ca_ski_bytes; self.issuer_ski_bytes += other.issuer_ski_bytes; self.current_manifest_rsync_uri_bytes += other.current_manifest_rsync_uri_bytes; self.current_crl_rsync_uri_bytes += other.current_crl_rsync_uri_bytes; self.validated_manifest_number_bytes += other.validated_manifest_number_bytes; } } #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)] pub struct VcirChildResourceSizeBreakdown { pub effective_ip_resource_cbor_bytes: u64, pub effective_as_resource_cbor_bytes: u64, } impl VcirChildResourceSizeBreakdown { fn from_vcir(vcir: &ValidatedCaInstanceResult) -> Self { let mut out = Self::default(); for child in &vcir.child_entries { out.effective_ip_resource_cbor_bytes += serialized_cbor_len(&child.child_effective_ip_resources); out.effective_as_resource_cbor_bytes += serialized_cbor_len(&child.child_effective_as_resources); } out } fn add_assign(&mut self, other: &Self) { self.effective_ip_resource_cbor_bytes += other.effective_ip_resource_cbor_bytes; self.effective_as_resource_cbor_bytes += other.effective_as_resource_cbor_bytes; } } #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)] pub struct VcirCcrProjectionSizeBreakdown { pub manifest_rsync_uri_bytes: u64, pub manifest_sha256_bytes: u64, pub manifest_ee_aki_bytes: u64, pub manifest_number_bytes: u64, pub manifest_sia_locations_count: u64, pub manifest_sia_locations_der_bytes: u64, pub subordinate_ski_count: u64, pub subordinate_ski_bytes: u64, } impl VcirCcrProjectionSizeBreakdown { fn from_projection(projection: &VcirCcrManifestProjection) -> Self { Self { manifest_rsync_uri_bytes: projection.manifest_rsync_uri.len() as u64, manifest_sha256_bytes: projection.manifest_sha256.len() as u64, manifest_ee_aki_bytes: projection.manifest_ee_aki.len() as u64, manifest_number_bytes: projection.manifest_number_be.len() as u64, manifest_sia_locations_count: projection.manifest_sia_locations_der.len() as u64, manifest_sia_locations_der_bytes: projection .manifest_sia_locations_der .iter() .map(|location| location.len() as u64) .sum(), subordinate_ski_count: projection.subordinate_skis.len() as u64, subordinate_ski_bytes: projection .subordinate_skis .iter() .map(|ski| ski.len() as u64) .sum(), } } fn add_assign(&mut self, other: &Self) { self.manifest_rsync_uri_bytes += other.manifest_rsync_uri_bytes; self.manifest_sha256_bytes += other.manifest_sha256_bytes; self.manifest_ee_aki_bytes += other.manifest_ee_aki_bytes; self.manifest_number_bytes += other.manifest_number_bytes; self.manifest_sia_locations_count += other.manifest_sia_locations_count; self.manifest_sia_locations_der_bytes += other.manifest_sia_locations_der_bytes; self.subordinate_ski_count += other.subordinate_ski_count; self.subordinate_ski_bytes += other.subordinate_ski_bytes; } } fn push_top_vcir_storage_entry( entries: &mut Vec, entry: VcirStorageEntrySummary, ) { const TOP_N: usize = 20; entries.push(entry); entries.sort_by(|left, right| { right .vcir_value_bytes .cmp(&left.vcir_value_bytes) .then_with(|| left.manifest_rsync_uri.cmp(&right.manifest_rsync_uri)) }); entries.truncate(TOP_N); } impl ValidatedCaInstanceResult { pub fn validate_internal(&self) -> StorageResult<()> { validate_non_empty("vcir.manifest_rsync_uri", &self.manifest_rsync_uri)?; if let Some(parent_manifest_rsync_uri) = &self.parent_manifest_rsync_uri { validate_non_empty("vcir.parent_manifest_rsync_uri", parent_manifest_rsync_uri)?; } validate_non_empty("vcir.tal_id", &self.tal_id)?; validate_non_empty("vcir.ca_subject_name", &self.ca_subject_name)?; validate_non_empty("vcir.ca_ski", &self.ca_ski)?; validate_non_empty("vcir.issuer_ski", &self.issuer_ski)?; parse_time( "vcir.last_successful_validation_time", &self.last_successful_validation_time, )?; validate_non_empty( "vcir.current_manifest_rsync_uri", &self.current_manifest_rsync_uri, )?; validate_non_empty("vcir.current_crl_rsync_uri", &self.current_crl_rsync_uri)?; self.validated_manifest_meta.validate_internal()?; self.ccr_manifest_projection.validate_internal()?; self.instance_gate.validate_internal()?; let expected_manifest_next = self .validated_manifest_meta .validated_manifest_next_update .parse() .map_err(|detail| StorageError::InvalidData { entity: "vcir", detail: format!( "validated_manifest_meta.validated_manifest_next_update invalid: {detail}" ), })?; let instance_manifest_next = self.instance_gate .manifest_next_update .parse() .map_err(|detail| StorageError::InvalidData { entity: "vcir", detail: format!("instance_gate.manifest_next_update invalid: {detail}"), })?; if expected_manifest_next != instance_manifest_next { return Err(StorageError::InvalidData { entity: "vcir", detail: "instance_gate.manifest_next_update must equal validated_manifest_meta.validated_manifest_next_update".to_string(), }); } let mut child_manifests = HashSet::with_capacity(self.child_entries.len()); for child in &self.child_entries { child.validate_internal()?; if !child_manifests.insert(child.child_manifest_rsync_uri.as_str()) { return Err(StorageError::InvalidData { entity: "vcir", detail: format!( "duplicate child_manifest_rsync_uri: {}", child.child_manifest_rsync_uri ), }); } } let mut vrp_count = 0u32; let mut aspa_count = 0u32; let mut router_key_count = 0u32; for output in &self.local_outputs { output.validate_internal()?; match output.output_type { VcirOutputType::Vrp => vrp_count += 1, VcirOutputType::Aspa => aspa_count += 1, VcirOutputType::RouterKey => router_key_count += 1, } } let mut output_ids = self .local_outputs .iter() .map(VcirLocalOutput::output_id) .collect::>(); output_ids.sort_unstable(); if let Some(duplicate) = output_ids .windows(2) .find_map(|pair| (pair[0] == pair[1]).then(|| pair[0].clone())) { return Err(StorageError::InvalidData { entity: "vcir", detail: format!("duplicate output_id: {duplicate}"), }); } if self.summary.local_vrp_count != vrp_count { return Err(StorageError::InvalidData { entity: "vcir.summary", detail: format!( "local_vrp_count={} does not match local_outputs count {}", self.summary.local_vrp_count, vrp_count ), }); } if self.summary.local_aspa_count != aspa_count { return Err(StorageError::InvalidData { entity: "vcir.summary", detail: format!( "local_aspa_count={} does not match local_outputs count {}", self.summary.local_aspa_count, aspa_count ), }); } if self.summary.local_router_key_count != router_key_count { return Err(StorageError::InvalidData { entity: "vcir.summary", detail: format!( "local_router_key_count={} does not match local_outputs count {}", self.summary.local_router_key_count, router_key_count ), }); } if self.summary.child_count != self.child_entries.len() as u32 { return Err(StorageError::InvalidData { entity: "vcir.summary", detail: format!( "child_count={} does not match child_entries length {}", self.summary.child_count, self.child_entries.len() ), }); } for artifact in &self.related_artifacts { artifact.validate_internal()?; } self.audit_summary.validate_internal()?; Ok(()) } } #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum RrdpSourceSyncState { Empty, SnapshotOnly, DeltaReady, Error, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct RrdpSourceRecord { pub notify_uri: String, pub last_session_id: Option, pub last_serial: Option, pub first_seen_at: PackTime, pub last_seen_at: PackTime, pub last_sync_at: Option, pub sync_state: RrdpSourceSyncState, pub last_snapshot_uri: Option, pub last_snapshot_hash: Option, pub last_error: Option, } impl RrdpSourceRecord { pub fn validate_internal(&self) -> StorageResult<()> { validate_non_empty("rrdp_source.notify_uri", &self.notify_uri)?; if let Some(session_id) = &self.last_session_id { validate_non_empty("rrdp_source.last_session_id", session_id)?; } parse_time("rrdp_source.first_seen_at", &self.first_seen_at)?; parse_time("rrdp_source.last_seen_at", &self.last_seen_at)?; if let Some(last_sync_at) = &self.last_sync_at { parse_time("rrdp_source.last_sync_at", last_sync_at)?; } if let Some(last_snapshot_uri) = &self.last_snapshot_uri { validate_non_empty("rrdp_source.last_snapshot_uri", last_snapshot_uri)?; } if let Some(last_snapshot_hash) = &self.last_snapshot_hash { validate_sha256_hex("rrdp_source.last_snapshot_hash", last_snapshot_hash)?; } if let Some(last_error) = &self.last_error { validate_non_empty("rrdp_source.last_error", last_error)?; } Ok(()) } } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct RrdpSourceMemberRecord { pub notify_uri: String, pub rsync_uri: String, pub current_hash: Option, pub object_type: Option, pub present: bool, pub last_confirmed_session_id: String, pub last_confirmed_serial: u64, pub last_changed_at: PackTime, } impl RrdpSourceMemberRecord { pub fn validate_internal(&self) -> StorageResult<()> { validate_non_empty("rrdp_source_member.notify_uri", &self.notify_uri)?; validate_non_empty("rrdp_source_member.rsync_uri", &self.rsync_uri)?; validate_non_empty( "rrdp_source_member.last_confirmed_session_id", &self.last_confirmed_session_id, )?; if self.present { let hash = self .current_hash .as_deref() .ok_or(StorageError::InvalidData { entity: "rrdp_source_member", detail: "current_hash is required when present=true".to_string(), })?; validate_sha256_hex("rrdp_source_member.current_hash", hash)?; } else if let Some(hash) = &self.current_hash { validate_sha256_hex("rrdp_source_member.current_hash", hash)?; } parse_time("rrdp_source_member.last_changed_at", &self.last_changed_at)?; Ok(()) } } #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum RrdpUriOwnerState { Active, Conflict, Withdrawn, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct RrdpUriOwnerRecord { pub rsync_uri: String, pub notify_uri: String, pub current_hash: Option, pub last_confirmed_session_id: String, pub last_confirmed_serial: u64, pub last_changed_at: PackTime, pub owner_state: RrdpUriOwnerState, } impl RrdpUriOwnerRecord { pub fn validate_internal(&self) -> StorageResult<()> { validate_non_empty("rrdp_uri_owner.rsync_uri", &self.rsync_uri)?; validate_non_empty("rrdp_uri_owner.notify_uri", &self.notify_uri)?; validate_non_empty( "rrdp_uri_owner.last_confirmed_session_id", &self.last_confirmed_session_id, )?; if let Some(hash) = &self.current_hash { validate_sha256_hex("rrdp_uri_owner.current_hash", hash)?; } parse_time("rrdp_uri_owner.last_changed_at", &self.last_changed_at)?; Ok(()) } } fn write_roa_cache_projection_to_batch( projection_cf: &ColumnFamily, batch: &mut WriteBatch, vcir: &ValidatedCaInstanceResult, context: Option<&RoaCacheProjectionContext>, timing: Option<&mut VcirReplaceTimingBreakdown>, ) -> StorageResult<()> { let projection_key = roa_cache_projection_key(&vcir.manifest_rsync_uri); let projection = RoaCacheProjection::from_vcir_with_context(vcir, context)?; match projection { Some(projection) => { let projection_value = encode_cbor(&projection, "roa_cache_projection")?; if let Some(timing) = timing { timing.roa_cache_projection_value_bytes = projection_value.len() as u64; } batch.put_cf(projection_cf, projection_key.as_bytes(), projection_value); } None => { batch.delete_cf(projection_cf, projection_key.as_bytes()); } } Ok(()) } fn write_publication_point_cache_projection_to_batch( projection_cf: &ColumnFamily, batch: &mut WriteBatch, action: PublicationPointCacheProjectionWriteAction<'_>, timing: Option<&mut VcirReplaceTimingBreakdown>, ) -> StorageResult<()> { match action { PublicationPointCacheProjectionWriteAction::Keep => Ok(()), PublicationPointCacheProjectionWriteAction::Write(projection) => { projection.validate_internal()?; let key = publication_point_cache_projection_key(&projection.manifest_rsync_uri); let value = encode_cbor(projection, "publication_point_cache_projection")?; if let Some(timing) = timing { timing.publication_point_cache_projection_value_bytes = value.len() as u64; } batch.put_cf(projection_cf, key.as_bytes(), value); Ok(()) } PublicationPointCacheProjectionWriteAction::Delete { manifest_rsync_uri } => { let key = publication_point_cache_projection_key(manifest_rsync_uri); batch.delete_cf(projection_cf, key.as_bytes()); Ok(()) } } } impl RocksStore { pub fn open(path: &Path) -> StorageResult { let mut base_opts = Options::default(); base_opts.create_if_missing(true); base_opts.create_missing_column_families(true); let blob_mode = work_db_blob_mode_from_env(); let memory_profile = work_db_memory_profile_from_env(); configure_work_db_options(&mut base_opts, blob_mode, memory_profile); let db = DB::open_cf_descriptors( &base_opts, path, column_family_descriptors_for_blob_mode(blob_mode), ) .map_err(|e| StorageError::RocksDb(e.to_string()))?; Ok(Self { db, external_raw_store: None, external_repo_bytes: None, publication_point_cache_index_dir: default_pp_cache_index_dir(path), child_certificate_cache_index_dir: default_child_certificate_cache_index_dir(path), publication_point_cache_projection_index: Mutex::new( PublicationPointCacheProjectionIndexState::Uninitialized, ), }) } pub fn open_with_external_raw_store(path: &Path, raw_store_path: &Path) -> StorageResult { Self::open_with_external_stores(path, Some(raw_store_path), None) } pub fn open_with_external_repo_bytes( path: &Path, repo_bytes_path: &Path, ) -> StorageResult { Self::open_with_external_stores(path, None, Some(repo_bytes_path)) } pub fn open_with_external_stores( path: &Path, raw_store_path: Option<&Path>, repo_bytes_path: Option<&Path>, ) -> StorageResult { let mut store = Self::open(path)?; if let Some(raw_store_path) = raw_store_path { store.external_raw_store = Some(ExternalRawStoreDb::open(raw_store_path)?); } if let Some(repo_bytes_path) = repo_bytes_path { store.external_repo_bytes = Some(ExternalRepoBytesDb::open(repo_bytes_path)?); } Ok(store) } pub(crate) fn external_raw_store_ref(&self) -> Option<&ExternalRawStoreDb> { self.external_raw_store.as_ref() } pub(crate) fn external_repo_bytes_ref(&self) -> Option<&ExternalRepoBytesDb> { self.external_repo_bytes.as_ref() } pub fn memory_snapshot(&self) -> RocksDbMemorySnapshot { let mut databases = Vec::new(); databases.push(memory_db_snapshot_for_column_families( "work-db", &self.db, Some(ALL_COLUMN_FAMILY_NAMES), )); if let Some(raw_store) = self.external_raw_store.as_ref() { databases.push(raw_store.memory_snapshot("raw-store.db")); } if let Some(repo_bytes) = self.external_repo_bytes.as_ref() { databases.push(repo_bytes.memory_snapshot("repo-bytes.db")); } let mut totals = RocksDbMemoryTotals::default(); for db in &databases { totals.add_properties(&db.properties); } RocksDbMemorySnapshot { databases, totals } } pub fn publication_point_cache_mmap_index_load_stats(&self) -> Option { let guard = self.publication_point_cache_projection_index.lock().ok()?; match &*guard { PublicationPointCacheProjectionIndexState::LoadedMmap { load_stats, .. } => { Some(load_stats.clone()) } PublicationPointCacheProjectionIndexState::Uninitialized | PublicationPointCacheProjectionIndexState::Disabled | PublicationPointCacheProjectionIndexState::BuildingFromEmpty { .. } | PublicationPointCacheProjectionIndexState::Loaded { .. } => None, } } fn try_load_publication_point_cache_mmap_index_for_update( &self, reason: &'static str, ) -> StorageResult<()> { if !pp_cache_raw_index_enabled() { return Ok(()); } let mut guard = self .publication_point_cache_projection_index .lock() .map_err(|e| { StorageError::RocksDb(format!("publication point cache index lock poisoned: {e}")) })?; if !matches!( *guard, PublicationPointCacheProjectionIndexState::Uninitialized ) { return Ok(()); } match load_pp_cache_mmap_index_set(&self.publication_point_cache_index_dir) { Ok((mmap, stats)) => { crate::progress_log::emit( "publication_point_cache_mmap_index_load", serde_json::json!({ "state": "loaded", "reason": reason, "entries": stats.entries, "bytes": stats.bytes, "file_bytes": stats.file_bytes, "load_ms": stats.load_ms, }), ); *guard = PublicationPointCacheProjectionIndexState::LoadedMmap { mmap, dirty: HashMap::new(), dirty_bytes: 0, load_stats: stats, }; } Err(e) => { crate::progress_log::emit( "publication_point_cache_mmap_index_load", serde_json::json!({ "state": "deferred_fallback_scan", "reason": reason, "error": e.to_string(), }), ); } } Ok(()) } fn cf(&self, name: &'static str) -> StorageResult<&ColumnFamily> { self.db .cf_handle(name) .ok_or(StorageError::MissingColumnFamily(name)) } pub fn put_repository_view_entry(&self, entry: &RepositoryViewEntry) -> StorageResult<()> { entry.validate_internal()?; let cf = self.cf(CF_REPOSITORY_VIEW)?; let key = repository_view_key(&entry.rsync_uri); let value = encode_cbor(entry, "repository_view")?; self.db .put_cf(cf, key.as_bytes(), value) .map_err(|e| StorageError::RocksDb(e.to_string()))?; Ok(()) } pub fn get_repository_view_entry( &self, rsync_uri: &str, ) -> StorageResult> { let cf = self.cf(CF_REPOSITORY_VIEW)?; let key = repository_view_key(rsync_uri); let Some(bytes) = self .db .get_cf(cf, key.as_bytes()) .map_err(|e| StorageError::RocksDb(e.to_string()))? else { return Ok(None); }; let entry = decode_cbor::(&bytes, "repository_view")?; entry.validate_internal()?; Ok(Some(entry)) } pub fn delete_repository_view_entry(&self, rsync_uri: &str) -> StorageResult<()> { let cf = self.cf(CF_REPOSITORY_VIEW)?; let key = repository_view_key(rsync_uri); self.db .delete_cf(cf, key.as_bytes()) .map_err(|e| StorageError::RocksDb(e.to_string()))?; Ok(()) } pub fn put_projection_batch( &self, repository_view_entries: &[RepositoryViewEntry], member_records: &[RrdpSourceMemberRecord], owner_records: &[RrdpUriOwnerRecord], ) -> StorageResult<()> { if repository_view_entries.is_empty() && member_records.is_empty() && owner_records.is_empty() { return Ok(()); } let repo_cf = self.cf(CF_REPOSITORY_VIEW)?; let member_cf = self.cf(CF_RRDP_SOURCE_MEMBER)?; let owner_cf = self.cf(CF_RRDP_URI_OWNER)?; let mut batch = WriteBatch::default(); for entry in repository_view_entries { entry.validate_internal()?; let key = repository_view_key(&entry.rsync_uri); let value = encode_cbor(entry, "repository_view")?; batch.put_cf(repo_cf, key.as_bytes(), value); } for record in member_records { record.validate_internal()?; let key = rrdp_source_member_key(&record.notify_uri, &record.rsync_uri); let value = encode_cbor(record, "rrdp_source_member")?; batch.put_cf(member_cf, key.as_bytes(), value); } for record in owner_records { record.validate_internal()?; let key = rrdp_uri_owner_key(&record.rsync_uri); let value = encode_cbor(record, "rrdp_uri_owner")?; batch.put_cf(owner_cf, key.as_bytes(), value); } self.write_batch(batch) } pub fn list_repository_view_entries_with_prefix( &self, rsync_uri_prefix: &str, ) -> StorageResult> { let cf = self.cf(CF_REPOSITORY_VIEW)?; let prefix = repository_view_prefix(rsync_uri_prefix); let mode = IteratorMode::From(prefix.as_bytes(), Direction::Forward); self.db .iterator_cf(cf, mode) .take_while(|res| match res { Ok((key, _)) => key.starts_with(prefix.as_bytes()), Err(_) => false, }) .map(|res| { let (_key, value) = res.map_err(|e| StorageError::RocksDb(e.to_string()))?; let entry = decode_cbor::(&value, "repository_view")?; entry.validate_internal()?; Ok(entry) }) .collect() } pub fn put_raw_by_hash_entry(&self, entry: &RawByHashEntry) -> StorageResult<()> { entry.validate_internal()?; if let Some(raw_store) = self.external_raw_store.as_ref() { return raw_store.put_raw_entry(entry); } let cf = self.cf(CF_RAW_BY_HASH)?; let key = raw_by_hash_key(&entry.sha256_hex); let value = encode_cbor(entry, "raw_by_hash")?; self.db .put_cf(cf, key.as_bytes(), value) .map_err(|e| StorageError::RocksDb(e.to_string()))?; Ok(()) } pub fn put_raw_by_hash_entries_batch(&self, entries: &[RawByHashEntry]) -> StorageResult<()> { if entries.is_empty() { return Ok(()); } if let Some(raw_store) = self.external_raw_store.as_ref() { return raw_store.put_raw_entries_batch(entries); } let cf = self.cf(CF_RAW_BY_HASH)?; let mut batch = WriteBatch::default(); for entry in entries { entry.validate_internal()?; let key = raw_by_hash_key(&entry.sha256_hex); let value = encode_cbor(entry, "raw_by_hash")?; batch.put_cf(cf, key.as_bytes(), value); } self.write_batch(batch) } pub fn put_raw_by_hash_entries_batch_unchecked( &self, entries: &[RawByHashEntry], ) -> StorageResult<()> { if entries.is_empty() { return Ok(()); } if let Some(raw_store) = self.external_raw_store.as_ref() { return raw_store.put_raw_entries_batch(entries); } let cf = self.cf(CF_RAW_BY_HASH)?; let mut batch = WriteBatch::default(); for entry in entries { let key = raw_by_hash_key(&entry.sha256_hex); let value = encode_cbor(entry, "raw_by_hash")?; batch.put_cf(cf, key.as_bytes(), value); } self.write_batch(batch) } pub fn put_blob_bytes_batch(&self, blobs: &[(String, Vec)]) -> StorageResult<()> { if blobs.is_empty() { return Ok(()); } if let Some(repo_bytes) = self.external_repo_bytes.as_ref() { return repo_bytes.put_blob_bytes_batch(blobs); } if let Some(raw_store) = self.external_raw_store.as_ref() { return raw_store.put_blob_bytes_batch(blobs); } let cf = self.cf(CF_RAW_BLOB)?; let mut batch = WriteBatch::default(); for (sha256_hex, bytes) in blobs { validate_sha256_hex("raw_blob.sha256_hex", sha256_hex)?; if bytes.is_empty() { return Err(StorageError::InvalidData { entity: "raw_blob", detail: "bytes must not be empty".to_string(), }); } let key = raw_blob_key(sha256_hex); batch.put_cf(cf, key.as_bytes(), bytes.as_slice()); } 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)?; if let Some(raw_store) = self.external_raw_store.as_ref() { return raw_store.delete_raw_entry(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> { if let Some(raw_store) = self.external_raw_store.as_ref() { return raw_store.get_raw_entry(sha256_hex); } let cf = self.cf(CF_RAW_BY_HASH)?; let key = raw_by_hash_key(sha256_hex); let Some(bytes) = self .db .get_cf(cf, key.as_bytes()) .map_err(|e| StorageError::RocksDb(e.to_string()))? else { return Ok(None); }; let entry = decode_cbor::(&bytes, "raw_by_hash")?; entry.validate_internal()?; Ok(Some(entry)) } pub fn get_raw_by_hash_entries_batch( &self, sha256_hexes: &[String], ) -> StorageResult>> { if sha256_hexes.is_empty() { return Ok(Vec::new()); } if let Some(raw_store) = self.external_raw_store.as_ref() { return raw_store.get_raw_entries_batch(sha256_hexes); } let cf = self.cf(CF_RAW_BY_HASH)?; let keys: Vec = sha256_hexes .iter() .map(|hash| raw_by_hash_key(hash)) .collect(); self.db .multi_get_cf(keys.iter().map(|key| (cf, key.as_bytes()))) .into_iter() .map(|res| { let maybe = res.map_err(|e| StorageError::RocksDb(e.to_string()))?; match maybe { Some(bytes) => { let entry = decode_cbor::(&bytes, "raw_by_hash")?; entry.validate_internal()?; Ok(Some(entry)) } None => Ok(None), } }) .collect() } pub fn get_blob_bytes(&self, sha256_hex: &str) -> StorageResult>> { if let Some(repo_bytes) = self.external_repo_bytes.as_ref() { return repo_bytes.get_blob_bytes(sha256_hex); } if let Some(raw_store) = self.external_raw_store.as_ref() { return raw_store.get_blob_bytes(sha256_hex); } validate_sha256_hex("raw_blob.sha256_hex", sha256_hex)?; let cf = self.cf(CF_RAW_BLOB)?; let key = raw_blob_key(sha256_hex); if let Some(bytes) = self .db .get_cf(cf, key.as_bytes()) .map_err(|e| StorageError::RocksDb(e.to_string()))? { return Ok(Some(bytes)); } self.get_raw_by_hash_entry(sha256_hex) .map(|entry| entry.map(|entry| entry.bytes)) } pub fn get_blob_bytes_batch( &self, sha256_hexes: &[String], ) -> StorageResult>>> { if sha256_hexes.is_empty() { return Ok(Vec::new()); } if let Some(repo_bytes) = self.external_repo_bytes.as_ref() { return repo_bytes.get_blob_bytes_batch(sha256_hexes); } if let Some(raw_store) = self.external_raw_store.as_ref() { return raw_store.get_blob_bytes_batch(sha256_hexes); } let cf = self.cf(CF_RAW_BLOB)?; let keys: Vec = sha256_hexes .iter() .map(|hash| { validate_sha256_hex("raw_blob.sha256_hex", hash)?; Ok::(raw_blob_key(hash)) }) .collect::>()?; let blob_results: Vec>> = self .db .multi_get_cf(keys.iter().map(|key| (cf, key.as_bytes()))) .into_iter() .map(|res| res.map_err(|e| StorageError::RocksDb(e.to_string()))) .collect::>()?; let mut out = Vec::with_capacity(sha256_hexes.len()); for (sha256_hex, maybe_blob) in sha256_hexes.iter().zip(blob_results.into_iter()) { if maybe_blob.is_some() { out.push(maybe_blob); } else { out.push( self.get_raw_by_hash_entry(sha256_hex)? .map(|entry| entry.bytes), ); } } Ok(out) } pub fn put_vcir(&self, vcir: &ValidatedCaInstanceResult) -> StorageResult<()> { self.put_vcir_with_publication_point_cache_projection(vcir, None) } pub fn put_vcir_with_publication_point_cache_projection( &self, vcir: &ValidatedCaInstanceResult, publication_point_projection: Option<&PublicationPointCacheProjection>, ) -> StorageResult<()> { self.put_vcir_with_projections(vcir, None, publication_point_projection) } pub fn put_vcir_with_projections( &self, vcir: &ValidatedCaInstanceResult, roa_cache_context: Option<&RoaCacheProjectionContext>, publication_point_projection: Option<&PublicationPointCacheProjection>, ) -> StorageResult<()> { let publication_point_projection_action = publication_point_projection .map(PublicationPointCacheProjectionWriteAction::Write) .unwrap_or(PublicationPointCacheProjectionWriteAction::Keep); self.put_vcir_with_projection_action( vcir, roa_cache_context, publication_point_projection_action, ) } fn put_vcir_with_projection_action( &self, vcir: &ValidatedCaInstanceResult, roa_cache_context: Option<&RoaCacheProjectionContext>, publication_point_projection_action: PublicationPointCacheProjectionWriteAction<'_>, ) -> StorageResult<()> { vcir.validate_internal()?; let vcir_cf = self.cf(CF_VCIR)?; let replay_cf = self.cf(CF_MANIFEST_REPLAY_META)?; let projection_cf = self.cf(CF_ROA_CACHE_PROJECTION)?; let pp_projection_cf = self.cf(CF_PUBLICATION_POINT_CACHE_PROJECTION)?; let replay_meta = ManifestReplayMeta::from_vcir(vcir); replay_meta.validate_internal()?; let mut batch = WriteBatch::default(); let key = vcir_key(&vcir.manifest_rsync_uri); let value = encode_cbor(vcir, "vcir")?; batch.put_cf(vcir_cf, key.as_bytes(), value); let replay_key = manifest_replay_meta_key(&replay_meta.manifest_rsync_uri); let replay_value = encode_cbor(&replay_meta, "manifest_replay_meta")?; batch.put_cf(replay_cf, replay_key.as_bytes(), replay_value); write_roa_cache_projection_to_batch( projection_cf, &mut batch, vcir, roa_cache_context, None, )?; write_publication_point_cache_projection_to_batch( pp_projection_cf, &mut batch, publication_point_projection_action, None, )?; self.write_batch(batch)?; self.apply_publication_point_cache_projection_index_action( publication_point_projection_action, )?; Ok(()) } pub fn replace_vcir_and_manifest_replay_meta( &self, vcir: &ValidatedCaInstanceResult, ) -> StorageResult { self.replace_vcir_manifest_replay_meta_and_publication_point_cache_projection(vcir, None) } pub fn replace_vcir_manifest_replay_meta_and_publication_point_cache_projection( &self, vcir: &ValidatedCaInstanceResult, publication_point_projection: Option<&PublicationPointCacheProjection>, ) -> StorageResult { self.replace_vcir_manifest_replay_meta_and_projections( vcir, None, publication_point_projection, ) } pub fn replace_vcir_manifest_replay_meta_and_projections( &self, vcir: &ValidatedCaInstanceResult, roa_cache_context: Option<&RoaCacheProjectionContext>, publication_point_projection: Option<&PublicationPointCacheProjection>, ) -> StorageResult { let publication_point_projection_action = publication_point_projection .map(PublicationPointCacheProjectionWriteAction::Write) .unwrap_or(PublicationPointCacheProjectionWriteAction::Keep); self.replace_vcir_manifest_replay_meta_and_projection_action( vcir, roa_cache_context, publication_point_projection_action, ) } pub(crate) fn replace_vcir_manifest_replay_meta_and_projection_action( &self, vcir: &ValidatedCaInstanceResult, roa_cache_context: Option<&RoaCacheProjectionContext>, publication_point_projection_action: PublicationPointCacheProjectionWriteAction<'_>, ) -> StorageResult { let mut timing = VcirReplaceTimingBreakdown { rss_before_kb: process_vm_rss_kb(), ..VcirReplaceTimingBreakdown::default() }; let validate_started = std::time::Instant::now(); vcir.validate_internal()?; timing.validate_ms = validate_started.elapsed().as_millis() as u64; timing.rss_after_validate_kb = process_vm_rss_kb(); timing.field_sizes = VcirFieldSizeBreakdown::from_vcir(vcir); let batch_build_started = std::time::Instant::now(); let vcir_cf = self.cf(CF_VCIR)?; let replay_cf = self.cf(CF_MANIFEST_REPLAY_META)?; let projection_cf = self.cf(CF_ROA_CACHE_PROJECTION)?; let pp_projection_cf = self.cf(CF_PUBLICATION_POINT_CACHE_PROJECTION)?; let mut batch = WriteBatch::default(); let vcir_key = vcir_key(&vcir.manifest_rsync_uri); let vcir_encode_started = std::time::Instant::now(); let vcir_value = encode_cbor(vcir, "vcir")?; timing.vcir_encode_ms = vcir_encode_started.elapsed().as_millis() as u64; timing.vcir_value_bytes = vcir_value.len() as u64; batch.put_cf(vcir_cf, vcir_key.as_bytes(), vcir_value); timing.rss_after_vcir_encode_kb = process_vm_rss_kb(); let replay_meta_encode_started = std::time::Instant::now(); let replay_meta = ManifestReplayMeta::from_vcir(vcir); replay_meta.validate_internal()?; let replay_key = manifest_replay_meta_key(&replay_meta.manifest_rsync_uri); let replay_value = encode_cbor(&replay_meta, "manifest_replay_meta")?; timing.replay_meta_encode_ms = replay_meta_encode_started.elapsed().as_millis() as u64; timing.replay_meta_value_bytes = replay_value.len() as u64; batch.put_cf(replay_cf, replay_key.as_bytes(), replay_value); timing.rss_after_replay_meta_encode_kb = process_vm_rss_kb(); let projection_encode_started = std::time::Instant::now(); write_roa_cache_projection_to_batch( projection_cf, &mut batch, vcir, roa_cache_context, Some(&mut timing), )?; timing.roa_cache_projection_encode_ms = projection_encode_started.elapsed().as_millis() as u64; timing.rss_after_roa_cache_projection_encode_kb = process_vm_rss_kb(); let pp_projection_encode_started = std::time::Instant::now(); write_publication_point_cache_projection_to_batch( pp_projection_cf, &mut batch, publication_point_projection_action, Some(&mut timing), )?; timing.publication_point_cache_projection_encode_ms = pp_projection_encode_started.elapsed().as_millis() as u64; timing.rss_after_publication_point_cache_projection_encode_kb = process_vm_rss_kb(); timing.total_encoded_bytes = timing.vcir_value_bytes + timing.replay_meta_value_bytes + timing.roa_cache_projection_value_bytes + timing.publication_point_cache_projection_value_bytes; timing.batch_build_ms = batch_build_started.elapsed().as_millis() as u64; let write_batch_started = std::time::Instant::now(); self.write_batch(batch)?; self.apply_publication_point_cache_projection_index_action( publication_point_projection_action, )?; timing.write_batch_ms = write_batch_started.elapsed().as_millis() as u64; timing.rss_after_write_batch_kb = process_vm_rss_kb(); Ok(timing) } pub fn get_vcir( &self, manifest_rsync_uri: &str, ) -> StorageResult> { let cf = self.cf(CF_VCIR)?; let key = vcir_key(manifest_rsync_uri); let Some(bytes) = self .db .get_cf(cf, key.as_bytes()) .map_err(|e| StorageError::RocksDb(e.to_string()))? else { return Ok(None); }; let vcir = decode_cbor::(&bytes, "vcir")?; vcir.validate_internal()?; Ok(Some(vcir)) } pub fn get_manifest_replay_meta( &self, manifest_rsync_uri: &str, ) -> StorageResult> { let cf = self.cf(CF_MANIFEST_REPLAY_META)?; let key = manifest_replay_meta_key(manifest_rsync_uri); let Some(bytes) = self .db .get_cf(cf, key.as_bytes()) .map_err(|e| StorageError::RocksDb(e.to_string()))? else { return Ok(None); }; let meta = decode_cbor::(&bytes, "manifest_replay_meta")?; meta.validate_internal()?; Ok(Some(meta)) } pub fn get_roa_cache_projection( &self, manifest_rsync_uri: &str, ) -> StorageResult> { let cf = self.cf(CF_ROA_CACHE_PROJECTION)?; let key = roa_cache_projection_key(manifest_rsync_uri); let Some(bytes) = self .db .get_cf(cf, key.as_bytes()) .map_err(|e| StorageError::RocksDb(e.to_string()))? else { return Ok(None); }; let projection = decode_cbor::(&bytes, "roa_cache_projection")?; projection.validate_internal()?; Ok(Some(projection)) } pub fn put_child_certificate_cache_projection( &self, projection: &ChildCertificateCacheProjection, ) -> StorageResult<()> { projection.validate_internal()?; let cf = self.cf(CF_CHILD_CERTIFICATE_CACHE_PROJECTION)?; let key = child_certificate_cache_projection_key(&projection.cache_key_sha256_hex); let value = encode_cbor(projection, "child_certificate_cache_projection")?; self.db .put_cf(cf, key.as_bytes(), value) .map_err(|e| StorageError::RocksDb(e.to_string()))?; Ok(()) } pub fn get_child_certificate_cache_projection( &self, cache_key_sha256_hex: &str, ) -> StorageResult> { validate_sha256_hex( "child_certificate_cache_projection.cache_key_sha256_hex", cache_key_sha256_hex, )?; let cf = self.cf(CF_CHILD_CERTIFICATE_CACHE_PROJECTION)?; let key = child_certificate_cache_projection_key(cache_key_sha256_hex); let Some(bytes) = self .db .get_cf(cf, key.as_bytes()) .map_err(|e| StorageError::RocksDb(e.to_string()))? else { return Ok(None); }; let projection = decode_cbor::( &bytes, "child_certificate_cache_projection", )?; projection.validate_internal()?; Ok(Some(projection)) } pub fn get_child_certificate_cache_projections_batch( &self, cache_key_sha256_hexes: &[String], ) -> StorageResult>> { if cache_key_sha256_hexes.is_empty() { return Ok(Vec::new()); } for cache_key_sha256_hex in cache_key_sha256_hexes { validate_sha256_hex( "child_certificate_cache_projection.cache_key_sha256_hex", cache_key_sha256_hex, )?; } let cf = self.cf(CF_CHILD_CERTIFICATE_CACHE_PROJECTION)?; let keys: Vec = cache_key_sha256_hexes .iter() .map(|key| child_certificate_cache_projection_key(key)) .collect(); self.db .multi_get_cf(keys.iter().map(|key| (cf, key.as_bytes()))) .into_iter() .map(|res| { let Some(bytes) = res.map_err(|e| StorageError::RocksDb(e.to_string()))? else { return Ok(None); }; let projection = decode_cbor::( &bytes, "child_certificate_cache_projection", )?; projection.validate_internal()?; Ok(Some(projection)) }) .collect() } fn child_certificate_cache_segment_path(&self, manifest_rsync_uri: &str) -> PathBuf { self.child_certificate_cache_index_dir .join(child_certificate_cache_segment_file_name( manifest_rsync_uri, )) } pub fn get_child_certificate_cache_projections_mmap_segment( &self, manifest_rsync_uri: &str, cache_key_sha256_hexes: &[String], ) -> StorageResult> { if cache_key_sha256_hexes.is_empty() { return Ok(Some(ChildCertificateCacheMmapLookup::default())); } validate_non_empty( "child_certificate_cache_mmap_segment.manifest_rsync_uri", manifest_rsync_uri, )?; for cache_key_sha256_hex in cache_key_sha256_hexes { validate_sha256_hex( "child_certificate_cache_projection.cache_key_sha256_hex", cache_key_sha256_hex, )?; } let path = self.child_certificate_cache_segment_path(manifest_rsync_uri); if !path.exists() { return Ok(None); } let (index, _) = load_pp_cache_mmap_index(&path)?; let file_bytes = index.file_bytes(); let mut hits = 0usize; let mut misses = 0usize; let mut projections = Vec::with_capacity(cache_key_sha256_hexes.len()); for cache_key_sha256_hex in cache_key_sha256_hexes { match index.lookup(cache_key_sha256_hex) { Some(PpCacheIndexLookup::Hit(bytes)) => { let projection = decode_cbor::( bytes, "child_certificate_cache_projection_mmap_segment", )?; projection.validate_internal()?; hits = hits.saturating_add(1); projections.push(Some(projection)); } Some(PpCacheIndexLookup::Deleted) | None => { misses = misses.saturating_add(1); projections.push(None); } } } Ok(Some(ChildCertificateCacheMmapLookup { projections, hits, misses, file_bytes, })) } pub fn write_child_certificate_cache_mmap_segment( &self, manifest_rsync_uri: &str, projections: &[ChildCertificateCacheProjection], ) -> StorageResult { validate_non_empty( "child_certificate_cache_mmap_segment.manifest_rsync_uri", manifest_rsync_uri, )?; let mut entries = Vec::with_capacity(projections.len()); for projection in projections { projection.validate_internal()?; entries.push(( projection.cache_key_sha256_hex.clone(), encode_cbor( projection, "child_certificate_cache_projection_mmap_segment", )?, )); } let path = self.child_certificate_cache_segment_path(manifest_rsync_uri); write_pp_cache_index_atomic(&path, entries) } pub fn write_child_certificate_cache_mmap_segment_overlay( &self, manifest_rsync_uri: &str, cache_key_sha256_hexes: &[String], projections: &[ChildCertificateCacheProjection], ) -> StorageResult { validate_non_empty( "child_certificate_cache_mmap_segment.manifest_rsync_uri", manifest_rsync_uri, )?; let mut dirty = HashMap::>::with_capacity(projections.len()); for projection in projections { projection.validate_internal()?; dirty.insert( projection.cache_key_sha256_hex.clone(), encode_cbor( projection, "child_certificate_cache_projection_mmap_segment", )?, ); } let path = self.child_certificate_cache_segment_path(manifest_rsync_uri); let existing = if path.exists() { Some(load_pp_cache_mmap_index(&path)?.0) } else { None }; let mut entries = Vec::with_capacity(cache_key_sha256_hexes.len()); let mut emitted = HashSet::::new(); for cache_key_sha256_hex in cache_key_sha256_hexes { validate_sha256_hex( "child_certificate_cache_projection.cache_key_sha256_hex", cache_key_sha256_hex, )?; if !emitted.insert(cache_key_sha256_hex.clone()) { continue; } if let Some(value) = dirty.remove(cache_key_sha256_hex) { entries.push((cache_key_sha256_hex.clone(), value)); continue; } if let Some(existing) = existing.as_ref() { if let Some(PpCacheIndexLookup::Hit(bytes)) = existing.lookup(cache_key_sha256_hex) { entries.push((cache_key_sha256_hex.clone(), bytes.to_vec())); } } } for (key, value) in dirty { if emitted.insert(key.clone()) { entries.push((key, value)); } } write_pp_cache_index_atomic(&path, entries) } pub fn get_publication_point_cache_projection( &self, manifest_rsync_uri: &str, ) -> StorageResult> { self.get_publication_point_cache_projection_from_db(manifest_rsync_uri) } pub fn get_publication_point_cache_projection_cached( &self, manifest_rsync_uri: &str, ) -> StorageResult> { let mut owned_bytes: Option> = None; { let mut guard = self .publication_point_cache_projection_index .lock() .map_err(|e| { StorageError::RocksDb(format!( "publication point cache index lock poisoned: {e}" )) })?; if matches!( *guard, PublicationPointCacheProjectionIndexState::Uninitialized ) { let load_started = std::time::Instant::now(); let raw_index_enabled = pp_cache_raw_index_enabled(); *guard = if !raw_index_enabled { crate::progress_log::emit( "publication_point_cache_raw_index", serde_json::json!({ "state": "disabled_by_env", "entries": 0, "bytes": 0, "load_ms": load_started.elapsed().as_millis() as u64, }), ); PublicationPointCacheProjectionIndexState::Disabled } else { match load_pp_cache_mmap_index_set(&self.publication_point_cache_index_dir) { Ok((mmap, stats)) => { crate::progress_log::emit( "publication_point_cache_mmap_index_load", serde_json::json!({ "state": "loaded", "entries": stats.entries, "bytes": stats.bytes, "file_bytes": stats.file_bytes, "load_ms": stats.load_ms, }), ); PublicationPointCacheProjectionIndexState::LoadedMmap { mmap, dirty: HashMap::new(), dirty_bytes: 0, load_stats: stats, } } Err(e) => { crate::progress_log::emit( "publication_point_cache_mmap_index_load", serde_json::json!({ "state": "fallback_scan", "error": e.to_string(), "load_ms": load_started.elapsed().as_millis() as u64, }), ); let scan_started = std::time::Instant::now(); let (index, bytes) = self.load_publication_point_cache_projection_index()?; if index.is_empty() { let limit = pp_cache_raw_index_empty_build_limit_bytes(); crate::progress_log::emit( "publication_point_cache_raw_index", serde_json::json!({ "state": "empty_building_bounded", "entries": 0, "bytes": 0, "empty_build_limit_bytes": limit, "load_ms": scan_started.elapsed().as_millis() as u64, }), ); PublicationPointCacheProjectionIndexState::BuildingFromEmpty { index, bytes: 0, limit, } } else { crate::progress_log::emit( "publication_point_cache_raw_index", serde_json::json!({ "state": "loaded", "entries": index.len(), "bytes": bytes, "load_ms": scan_started.elapsed().as_millis() as u64, }), ); PublicationPointCacheProjectionIndexState::Loaded { index, bytes } } } } }; } match &*guard { PublicationPointCacheProjectionIndexState::Loaded { index, .. } => { owned_bytes = index.get(manifest_rsync_uri).cloned(); } PublicationPointCacheProjectionIndexState::BuildingFromEmpty { index, .. } => { owned_bytes = index.get(manifest_rsync_uri).cloned(); } PublicationPointCacheProjectionIndexState::LoadedMmap { mmap, dirty, .. } => { if let Some(bytes) = dirty.get(manifest_rsync_uri).cloned() { if bytes.is_empty() { return Ok(None); } owned_bytes = Some(bytes); } else if let Some(lookup) = mmap.lookup(manifest_rsync_uri) { match lookup { PpCacheIndexLookup::Hit(bytes) => { let projection = decode_cbor::( bytes, "publication_point_cache_projection", )?; projection.validate_internal()?; return Ok(Some(projection)); } PpCacheIndexLookup::Deleted => return Ok(None), } } } PublicationPointCacheProjectionIndexState::Disabled | PublicationPointCacheProjectionIndexState::Uninitialized => {} } } let bytes = owned_bytes; let Some(bytes) = bytes else { return Ok(None); }; let projection = decode_cbor::( bytes.as_ref(), "publication_point_cache_projection", )?; projection.validate_internal()?; Ok(Some(projection)) } fn get_publication_point_cache_projection_from_db( &self, manifest_rsync_uri: &str, ) -> StorageResult> { let cf = self.cf(CF_PUBLICATION_POINT_CACHE_PROJECTION)?; let key = publication_point_cache_projection_key(manifest_rsync_uri); let Some(bytes) = self .db .get_cf(cf, key.as_bytes()) .map_err(|e| StorageError::RocksDb(e.to_string()))? else { return Ok(None); }; let projection = decode_cbor::( &bytes, "publication_point_cache_projection", )?; projection.validate_internal()?; Ok(Some(projection)) } fn load_publication_point_cache_projection_index( &self, ) -> StorageResult<(HashMap>, usize)> { let cf = self.cf(CF_PUBLICATION_POINT_CACHE_PROJECTION)?; let mode = IteratorMode::Start; let mut index = HashMap::new(); let mut bytes_total = 0usize; for res in self.db.iterator_cf(cf, mode) { let (key, value) = res.map_err(|e| StorageError::RocksDb(e.to_string()))?; let Some(manifest_rsync_uri) = publication_point_cache_projection_key_manifest_uri(&key) else { continue; }; bytes_total = bytes_total.saturating_add(value.len()); index.insert(manifest_rsync_uri, Arc::<[u8]>::from(value.to_vec())); } Ok((index, bytes_total)) } fn load_publication_point_cache_projection_entries( &self, ) -> StorageResult)>> { let cf = self.cf(CF_PUBLICATION_POINT_CACHE_PROJECTION)?; let mode = IteratorMode::Start; let mut entries = Vec::new(); for res in self.db.iterator_cf(cf, mode) { let (key, value) = res.map_err(|e| StorageError::RocksDb(e.to_string()))?; let Some(manifest_rsync_uri) = publication_point_cache_projection_key_manifest_uri(&key) else { continue; }; entries.push((manifest_rsync_uri, value.to_vec())); } Ok(entries) } fn apply_publication_point_cache_projection_index_action( &self, action: PublicationPointCacheProjectionWriteAction<'_>, ) -> StorageResult<()> { match action { PublicationPointCacheProjectionWriteAction::Keep => Ok(()), PublicationPointCacheProjectionWriteAction::Write(projection) => { self.update_publication_point_cache_projection_index(projection) } PublicationPointCacheProjectionWriteAction::Delete { manifest_rsync_uri } => { self.delete_publication_point_cache_projection_index_entry(manifest_rsync_uri) } } } fn update_publication_point_cache_projection_index( &self, projection: &PublicationPointCacheProjection, ) -> StorageResult<()> { self.try_load_publication_point_cache_mmap_index_for_update("write")?; let mut guard = self .publication_point_cache_projection_index .lock() .map_err(|e| { StorageError::RocksDb(format!("publication point cache index lock poisoned: {e}")) })?; let bytes = encode_cbor(projection, "publication_point_cache_projection")?; match &mut *guard { PublicationPointCacheProjectionIndexState::Loaded { index, bytes: total_bytes, } => { if let Some(previous) = index.insert( projection.manifest_rsync_uri.clone(), Arc::<[u8]>::from(bytes.clone()), ) { *total_bytes = total_bytes.saturating_sub(previous.len()); } *total_bytes = total_bytes.saturating_add(bytes.len()); } PublicationPointCacheProjectionIndexState::BuildingFromEmpty { index, bytes: total_bytes, limit, } => { if let Some(previous) = index.remove(&projection.manifest_rsync_uri) { *total_bytes = total_bytes.saturating_sub(previous.len()); } if total_bytes.saturating_add(bytes.len()) <= *limit { *total_bytes += bytes.len(); index.insert( projection.manifest_rsync_uri.clone(), Arc::<[u8]>::from(bytes), ); } else { *guard = PublicationPointCacheProjectionIndexState::Disabled; } } PublicationPointCacheProjectionIndexState::LoadedMmap { dirty, dirty_bytes, .. } => { if let Some(previous) = dirty.insert( projection.manifest_rsync_uri.clone(), Arc::<[u8]>::from(bytes.clone()), ) { *dirty_bytes = dirty_bytes.saturating_sub(previous.len()); } *dirty_bytes = dirty_bytes.saturating_add(bytes.len()); } PublicationPointCacheProjectionIndexState::Uninitialized | PublicationPointCacheProjectionIndexState::Disabled => {} } Ok(()) } fn delete_publication_point_cache_projection_index_entry( &self, manifest_rsync_uri: &str, ) -> StorageResult<()> { self.try_load_publication_point_cache_mmap_index_for_update("delete")?; let mut guard = self .publication_point_cache_projection_index .lock() .map_err(|e| { StorageError::RocksDb(format!("publication point cache index lock poisoned: {e}")) })?; match &mut *guard { PublicationPointCacheProjectionIndexState::Loaded { index, bytes: total_bytes, } => { if let Some(previous) = index.remove(manifest_rsync_uri) { *total_bytes = total_bytes.saturating_sub(previous.len()); } } PublicationPointCacheProjectionIndexState::BuildingFromEmpty { index, bytes: total_bytes, .. } => { if let Some(previous) = index.remove(manifest_rsync_uri) { *total_bytes = total_bytes.saturating_sub(previous.len()); } } PublicationPointCacheProjectionIndexState::LoadedMmap { dirty, dirty_bytes, .. } => { if let Some(previous) = dirty.insert(manifest_rsync_uri.to_string(), Arc::<[u8]>::from([])) { *dirty_bytes = dirty_bytes.saturating_sub(previous.len()); } } PublicationPointCacheProjectionIndexState::Uninitialized | PublicationPointCacheProjectionIndexState::Disabled => {} } Ok(()) } pub fn refresh_publication_point_cache_mmap_index( &self, ) -> StorageResult> { if !pp_cache_raw_index_enabled() { return Ok(None); } self.try_load_publication_point_cache_mmap_index_for_update("refresh")?; enum RefreshAction { Entries { entries: Vec<(String, Vec)>, write_segment: bool, old_entries: usize, dirty_entries: usize, }, ScanDb, } let action = { let guard = self .publication_point_cache_projection_index .lock() .map_err(|e| { StorageError::RocksDb(format!( "publication point cache index lock poisoned: {e}" )) })?; match &*guard { PublicationPointCacheProjectionIndexState::LoadedMmap { mmap, dirty, .. } => { RefreshAction::Entries { entries: dirty .iter() .map(|(key, value)| (key.clone(), value.as_ref().to_vec())) .collect::>(), write_segment: true, old_entries: mmap.entries(), dirty_entries: dirty.len(), } } PublicationPointCacheProjectionIndexState::Loaded { index, .. } | PublicationPointCacheProjectionIndexState::BuildingFromEmpty { index, .. } => { RefreshAction::Entries { entries: index .iter() .map(|(key, value)| (key.clone(), value.as_ref().to_vec())) .collect::>(), write_segment: false, old_entries: 0, dirty_entries: 0, } } PublicationPointCacheProjectionIndexState::Disabled | PublicationPointCacheProjectionIndexState::Uninitialized => RefreshAction::ScanDb, } }; let (entries, write_segment, old_entries, dirty_entries) = match action { RefreshAction::Entries { entries, write_segment, old_entries, dirty_entries, } => (entries, write_segment, old_entries, dirty_entries), RefreshAction::ScanDb => ( self.load_publication_point_cache_projection_entries()?, false, 0, 0, ), }; if entries.is_empty() { return Ok(None); } let current_path = self.publication_point_cache_index_dir.join("current.idx"); let mut stats = if write_segment && current_path.exists() { write_pp_cache_index_segment(&self.publication_point_cache_index_dir, entries)? } else { write_pp_cache_index_atomic(¤t_path, entries)? }; stats.old_entries = old_entries; stats.dirty_entries = dirty_entries; crate::progress_log::emit( "publication_point_cache_mmap_index_refresh", serde_json::json!({ "state": stats.state, "old_entries": stats.old_entries, "dirty_entries": stats.dirty_entries, "new_entries": stats.new_entries, "file_bytes": stats.file_bytes, "write_ms": stats.write_ms, }), ); let directory_stats = pp_cache_index_directory_stats(&self.publication_point_cache_index_dir)?; let compaction_reason = if directory_stats.segment_count >= PP_CACHE_INDEX_COMPACTION_SEGMENT_THRESHOLD { Some(format!( "segment_count>={PP_CACHE_INDEX_COMPACTION_SEGMENT_THRESHOLD}" )) } else if directory_stats.total_file_bytes >= PP_CACHE_INDEX_COMPACTION_BYTES_THRESHOLD { Some(format!( "total_file_bytes>={PP_CACHE_INDEX_COMPACTION_BYTES_THRESHOLD}" )) } else { None }; if let Some(reason) = compaction_reason { crate::progress_log::emit( "publication_point_cache_mmap_index_compaction", serde_json::json!({ "state": "started", "reason": reason, "segment_count": directory_stats.segment_count, "total_file_bytes": directory_stats.total_file_bytes, }), ); match compact_pp_cache_index(&self.publication_point_cache_index_dir) { Ok(mut compact_stats) => { compact_stats.compaction_reason = Some(reason); compact_stats.old_entries = stats.old_entries; compact_stats.dirty_entries = stats.dirty_entries; crate::progress_log::emit( "publication_point_cache_mmap_index_compaction", serde_json::json!({ "state": "completed", "reason": compact_stats.compaction_reason, "segments_before": compact_stats.compaction_segments_before, "total_file_bytes_before": compact_stats.compaction_total_file_bytes_before, "live_entries": compact_stats.compaction_live_entries, "file_bytes": compact_stats.compaction_file_bytes, "reclaimed_bytes": compact_stats.compaction_reclaimed_bytes, "deleted_segments": compact_stats.compaction_deleted_segments, "compaction_ms": compact_stats.compaction_ms, }), ); stats.compaction_triggered = compact_stats.compaction_triggered; stats.compaction_reason = compact_stats.compaction_reason; stats.compaction_segments_before = compact_stats.compaction_segments_before; stats.compaction_total_file_bytes_before = compact_stats.compaction_total_file_bytes_before; stats.compaction_live_entries = compact_stats.compaction_live_entries; stats.compaction_file_bytes = compact_stats.compaction_file_bytes; stats.compaction_reclaimed_bytes = compact_stats.compaction_reclaimed_bytes; stats.compaction_ms = compact_stats.compaction_ms; stats.compaction_deleted_segments = compact_stats.compaction_deleted_segments; } Err(e) => { let error = e.to_string(); crate::progress_log::emit( "publication_point_cache_mmap_index_compaction", serde_json::json!({ "state": "failed", "reason": reason, "segment_count": directory_stats.segment_count, "total_file_bytes": directory_stats.total_file_bytes, "error": error, }), ); stats.compaction_triggered = true; stats.compaction_reason = Some(reason); stats.compaction_segments_before = directory_stats.segment_count; stats.compaction_total_file_bytes_before = directory_stats.total_file_bytes; stats.compaction_error = Some(error); } } } let mut guard = self .publication_point_cache_projection_index .lock() .map_err(|e| { StorageError::RocksDb(format!("publication point cache index lock poisoned: {e}")) })?; if let PublicationPointCacheProjectionIndexState::LoadedMmap { dirty, dirty_bytes, .. } = &mut *guard { dirty.clear(); *dirty_bytes = 0; } Ok(Some(stats)) } pub fn put_transport_prefetch_snapshot( &self, snapshot: &crate::parallel::transport_prefetch::TransportPrefetchSnapshot, ) -> StorageResult<()> { let cf = self.cf(CF_TRANSPORT_PREFETCH)?; let value = encode_cbor(snapshot, "transport_prefetch_snapshot")?; self.db .put_cf(cf, TRANSPORT_PREFETCH_LAST_SNAPSHOT_KEY.as_bytes(), value) .map_err(|e| StorageError::RocksDb(e.to_string())) } pub fn get_transport_prefetch_snapshot( &self, ) -> StorageResult> { let cf = self.cf(CF_TRANSPORT_PREFETCH)?; let Some(bytes) = self .db .get_cf(cf, TRANSPORT_PREFETCH_LAST_SNAPSHOT_KEY.as_bytes()) .map_err(|e| StorageError::RocksDb(e.to_string()))? else { return Ok(None); }; decode_cbor::( &bytes, "transport_prefetch_snapshot", ) .map(Some) } pub fn list_vcirs(&self) -> StorageResult> { let cf = self.cf(CF_VCIR)?; let mode = IteratorMode::Start; let mut out = Vec::new(); for res in self.db.iterator_cf(cf, mode) { let (_key, bytes) = res.map_err(|e| StorageError::RocksDb(e.to_string()))?; let vcir = decode_cbor::(&bytes, "vcir")?; vcir.validate_internal()?; out.push(vcir); } Ok(out) } pub fn summarize_vcir_storage(&self) -> StorageResult { let cf = self.cf(CF_VCIR)?; let mode = IteratorMode::Start; let mut summary = VcirStorageSummary::default(); for res in self.db.iterator_cf(cf, mode) { let (_key, bytes) = res.map_err(|e| StorageError::RocksDb(e.to_string()))?; let vcir = decode_cbor::(&bytes, "vcir")?; vcir.validate_internal()?; summary.entry_count += 1; let value_bytes = bytes.len() as u64; summary.vcir_value_bytes += value_bytes; if value_bytes > summary.vcir_value_bytes_max { summary.vcir_value_bytes_max = value_bytes; summary.vcir_value_bytes_max_manifest_rsync_uri = Some(vcir.manifest_rsync_uri.clone()); } let entry_summary = VcirStorageEntrySummary::from_vcir(&vcir, value_bytes); summary.core_fields.add_assign(&entry_summary.core_fields); summary .ccr_projection .add_assign(&entry_summary.ccr_projection); summary .child_resources .add_assign(&entry_summary.child_resources); summary.field_sizes.add_assign(&entry_summary.field_sizes); push_top_vcir_storage_entry( &mut summary.top_entries_by_vcir_value_bytes, entry_summary, ); } summary.local_output_old_projection_bytes = summary.field_sizes.local_output_old_projection_bytes(); summary.local_output_typed_projection_bytes = summary.field_sizes.local_output_typed_projection_bytes(); summary.local_output_projection_saved_bytes = summary.field_sizes.local_output_projection_saved_bytes(); Ok(summary) } pub fn delete_vcir(&self, manifest_rsync_uri: &str) -> StorageResult<()> { let vcir_cf = self.cf(CF_VCIR)?; let replay_cf = self.cf(CF_MANIFEST_REPLAY_META)?; let projection_cf = self.cf(CF_ROA_CACHE_PROJECTION)?; let mut batch = WriteBatch::default(); let key = vcir_key(manifest_rsync_uri); batch.delete_cf(vcir_cf, key.as_bytes()); let replay_key = manifest_replay_meta_key(manifest_rsync_uri); batch.delete_cf(replay_cf, replay_key.as_bytes()); let projection_key = roa_cache_projection_key(manifest_rsync_uri); batch.delete_cf(projection_cf, projection_key.as_bytes()); self.write_batch(batch) } pub fn put_rrdp_source_record(&self, record: &RrdpSourceRecord) -> StorageResult<()> { record.validate_internal()?; let cf = self.cf(CF_RRDP_SOURCE)?; let key = rrdp_source_key(&record.notify_uri); let value = encode_cbor(record, "rrdp_source")?; self.db .put_cf(cf, key.as_bytes(), value) .map_err(|e| StorageError::RocksDb(e.to_string()))?; Ok(()) } pub fn get_rrdp_source_record( &self, notify_uri: &str, ) -> StorageResult> { let cf = self.cf(CF_RRDP_SOURCE)?; let key = rrdp_source_key(notify_uri); let Some(bytes) = self .db .get_cf(cf, key.as_bytes()) .map_err(|e| StorageError::RocksDb(e.to_string()))? else { return Ok(None); }; let record = decode_cbor::(&bytes, "rrdp_source")?; record.validate_internal()?; Ok(Some(record)) } pub fn put_rrdp_source_member_record( &self, record: &RrdpSourceMemberRecord, ) -> StorageResult<()> { record.validate_internal()?; let cf = self.cf(CF_RRDP_SOURCE_MEMBER)?; let key = rrdp_source_member_key(&record.notify_uri, &record.rsync_uri); let value = encode_cbor(record, "rrdp_source_member")?; self.db .put_cf(cf, key.as_bytes(), value) .map_err(|e| StorageError::RocksDb(e.to_string()))?; Ok(()) } pub fn get_rrdp_source_member_record( &self, notify_uri: &str, rsync_uri: &str, ) -> StorageResult> { let cf = self.cf(CF_RRDP_SOURCE_MEMBER)?; let key = rrdp_source_member_key(notify_uri, rsync_uri); let Some(bytes) = self .db .get_cf(cf, key.as_bytes()) .map_err(|e| StorageError::RocksDb(e.to_string()))? else { return Ok(None); }; let record = decode_cbor::(&bytes, "rrdp_source_member")?; record.validate_internal()?; Ok(Some(record)) } pub fn list_rrdp_source_member_records( &self, notify_uri: &str, ) -> StorageResult> { let cf = self.cf(CF_RRDP_SOURCE_MEMBER)?; let prefix = rrdp_source_member_prefix(notify_uri); let mode = IteratorMode::From(prefix.as_bytes(), Direction::Forward); self.db .iterator_cf(cf, mode) .take_while(|res| match res { Ok((key, _)) => key.starts_with(prefix.as_bytes()), Err(_) => false, }) .map(|res| { let (_key, value) = res.map_err(|e| StorageError::RocksDb(e.to_string()))?; let record = decode_cbor::(&value, "rrdp_source_member")?; record.validate_internal()?; Ok(record) }) .collect() } pub fn list_current_rrdp_source_members( &self, notify_uri: &str, ) -> StorageResult> { 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 { 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>> { Ok(self .load_current_object_with_hash_by_uri(rsync_uri)? .map(|obj| obj.bytes)) } pub fn load_current_object_with_hash_by_uri( &self, rsync_uri: &str, ) -> StorageResult> { 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 bytes = self .get_blob_bytes(hash)? .ok_or(StorageError::InvalidData { entity: "repository_view", detail: format!( "blob bytes missing for current object URI: {rsync_uri} (hash={hash})" ), })?; let current_hash = decode_sha256_hex_32("repository_view.current_hash", hash)?; Ok(Some(CurrentObjectWithHash { current_hash_hex: hash.to_ascii_lowercase(), current_hash, bytes, })) } } } pub fn put_rrdp_uri_owner_record(&self, record: &RrdpUriOwnerRecord) -> StorageResult<()> { record.validate_internal()?; let cf = self.cf(CF_RRDP_URI_OWNER)?; let key = rrdp_uri_owner_key(&record.rsync_uri); let value = encode_cbor(record, "rrdp_uri_owner")?; self.db .put_cf(cf, key.as_bytes(), value) .map_err(|e| StorageError::RocksDb(e.to_string()))?; Ok(()) } pub fn get_rrdp_uri_owner_record( &self, rsync_uri: &str, ) -> StorageResult> { let cf = self.cf(CF_RRDP_URI_OWNER)?; let key = rrdp_uri_owner_key(rsync_uri); let Some(bytes) = self .db .get_cf(cf, key.as_bytes()) .map_err(|e| StorageError::RocksDb(e.to_string()))? else { return Ok(None); }; let record = decode_cbor::(&bytes, "rrdp_uri_owner")?; record.validate_internal()?; Ok(Some(record)) } pub fn delete_rrdp_uri_owner_record(&self, rsync_uri: &str) -> StorageResult<()> { let cf = self.cf(CF_RRDP_URI_OWNER)?; let key = rrdp_uri_owner_key(rsync_uri); self.db .delete_cf(cf, key.as_bytes()) .map_err(|e| StorageError::RocksDb(e.to_string()))?; Ok(()) } #[allow(dead_code)] pub fn write_batch(&self, batch: WriteBatch) -> StorageResult<()> { self.db .write(batch) .map_err(|e| StorageError::RocksDb(e.to_string()))?; Ok(()) } } #[cfg(test)] #[path = "storage/tests.rs"] mod tests;