use super::*;
pub(super) struct Catalog {
pub header: f::NativePagedMetadataHeader,
pub roots: Vec<f::NativePagedCollectionRoot>,
references: Vec<f::NativePagedCrossRef>,
epoch: Option<u64>,
retired_roots: Vec<u32>,
deferred_pages: Vec<u32>,
pub resume: Option<f::native_resume::CatalogResume>,
proofs: Option<Arc<f::page_proof::CatalogProofs>>,
}
impl Catalog {
pub fn has_cross_references(&self) -> bool {
!self.references.is_empty()
}
pub fn read(source: &Source) -> Result<Self> {
let reference = source.manifest_reference;
let root = if reference.page_id == 0 {
1
} else {
reference.page_id
};
if root >= FIRST_ALLOCATABLE_PAGE {
source.claim(root)?;
}
let page = source.raw_page(root, true)?;
let mut digest = f::page_proof::CatalogDigest::default();
digest.add_page(page.as_bytes());
ensure(
page.page_type()? == PageType::Header,
"invalid catalog page",
)?;
let mut data = page.content().to_vec();
if let Some(header) = f::decode_native_metadata_overflow_header(&data)? {
let total = header.total_payload_bytes as usize;
ensure(
total > 0 && total <= 64 * 1024 * 1024,
"catalog exceeds compaction memory budget",
)?;
let first = total.min(data.len() - 16);
let mut payload = data[16..16 + first].to_vec();
let mut next = header.next_overflow_page_id;
while next != 0 {
source.claim(next)?;
let page = source.raw_page(next, true)?;
digest.add_page(page.as_bytes());
ensure(
page.page_type()? == PageType::Overflow,
"invalid catalog continuation",
)?;
let continuation =
f::decode_native_metadata_overflow_continuation_header(page.content())?;
let length = continuation.chunk_bytes as usize;
ensure(
length > 0
&& length <= PAGE_SIZE - HEADER_SIZE - 8
&& payload.len() + length <= total,
"invalid catalog continuation length",
)?;
payload.extend_from_slice(&page.content()[8..8 + length]);
next = continuation.next_overflow_page_id;
}
ensure(payload.len() == total, "incomplete catalog")?;
data = payload;
}
ensure(
reference.page_id == 0 || digest.finish() == reference.digest,
"catalog generation proof does not match the superblock",
)?;
let header = f::decode_native_paged_metadata_header(&data)?
.ok_or("source is not a native paged catalog")?;
ensure(
matches!(header.format_version, 11 | 12),
"unsupported native record format",
)?;
let mut position = f::METADATA_HEADER_BYTES;
ensure(
header.collection_count as usize <= data.len() / 8,
"invalid catalog count",
)?;
let mut roots = Vec::new();
let mut names = std::collections::BTreeSet::new();
for _ in 0..header.collection_count {
let root = f::decode_native_paged_collection_root(&data, &mut position)?;
ensure(
names.insert(root.collection.clone()),
"duplicate collection",
)?;
roots.push(root);
}
let count = number(&data, position)?;
position += 4;
ensure(
count as usize <= data.len().saturating_sub(position) / 21,
"invalid reference count",
)?;
let mut references = Vec::new();
for _ in 0..count {
references.push(f::decode_native_paged_cross_ref(&data, &mut position)?);
}
let retired_roots = f::decode_native_paged_retired_trees(&data, &mut position)?;
let deferred_pages = f::decode_native_paged_deferred_pages(&data, &mut position)?;
let epoch = f::native_store::decode_native_paged_ownership_epoch(&data, &mut position)?;
let proofs = f::page_proof::decode_catalog(&data, &mut position)?.map(Arc::new);
ensure(
proofs.is_none()
|| (source.header.required_capabilities
& f::file_format::FILE_CAPABILITY_TREE_PROOFS
!= 0
&& reference.page_id != 0),
"tree proofs require independent catalog authority",
)?;
let resume = f::native_resume::decode_catalog_resume(&data)?;
if let Some(resume) = &resume {
let mut canonical = Vec::new();
f::native_resume::encode_catalog_resume(&mut canonical, resume)?;
ensure(
data.get(position..position + canonical.len()) == Some(canonical.as_slice()),
"unsupported/noncanonical resume trailer",
)?;
position += canonical.len();
}
ensure(
data[position..].iter().all(|byte| *byte == 0),
"unknown catalog extension",
)?;
if let Some(proofs) = &proofs {
let live = roots
.iter()
.map(|root| root.root_page)
.chain(
resume
.iter()
.flat_map(|resume| resume.collections.values())
.flat_map(|state| {
state.key_index_root.into_iter().chain(
state
.secondary_indices
.values()
.map(|index| index.root_page),
)
}),
)
.filter(|root| *root != 0)
.collect::<std::collections::BTreeSet<_>>();
ensure(
live.iter()
.chain(retired_roots.iter())
.all(|root| proofs.contains_key(root)),
"published tree has no generation proof",
)?;
ensure(
proofs.iter().all(|(root, proof)| {
(live.contains(root) || retired_roots.contains(root))
&& *root < source.header.page_count
&& proof.reference.page_id < source.header.page_count
}),
"unowned catalog tree proof",
)?;
}
source.install_tree_proofs(proofs.clone())?;
Ok(Self {
header,
roots,
references,
epoch,
retired_roots,
deferred_pages,
resume,
proofs,
})
}
pub fn claim_retired(&self, source: &Arc<Source>) -> Result<()> {
for root in &self.retired_roots {
let tree = source.tree(*root)?;
for page in tree.retired_pages()? {
source.claim(page)?;
}
}
for page in &self.deferred_pages {
source.claim(*page)?;
}
Ok(())
}
pub(super) fn verify_alias_has_no_references(&self, legacy: &str) -> Result<()> {
ensure(
self.references
.iter()
.all(|r| r.target_collection != legacy),
"legacy runtime metadata has incoming references",
)
}
pub fn retain_canonical_system_config(&mut self) -> Result<()> {
let legacy_roots = self
.roots
.iter()
.filter(|root| root.collection == "red_config")
.count();
let canonical_roots = self
.roots
.iter()
.filter(|root| root.collection == "red.config")
.count();
ensure(
legacy_roots == 1 && canonical_roots == 1,
"canonical config repair requires exactly one red_config and one red.config root",
)?;
ensure(
self.references
.iter()
.all(|reference| reference.target_collection != "red_config"),
"legacy red_config has cross-references and cannot be retired",
)?;
let resume = self
.resume
.as_mut()
.ok_or("canonical config repair requires a resume catalog")?;
ensure(
resume.collections.contains_key("red_config")
&& resume.collections.contains_key("red.config"),
"canonical config repair requires both resume identities",
)?;
resume.collections.remove("red_config");
self.roots.retain(|root| root.collection != "red_config");
self.header.collection_count = u32::try_from(self.roots.len())?;
Ok(())
}
pub fn migrate_legacy_system_config(&mut self) -> Result<usize> {
let legacy_roots = self
.roots
.iter()
.filter(|root| root.collection == "red_config")
.count();
let canonical_roots = self
.roots
.iter()
.filter(|root| root.collection == "red.config")
.count();
ensure(
legacy_roots == 1 && canonical_roots == 1,
"canonical config migration requires exactly one red_config and one red.config root",
)?;
ensure(
self.references
.iter()
.all(|reference| reference.target_collection != "red_config"),
"legacy red_config has cross-references and cannot be retired",
)?;
let resume = self
.resume
.as_mut()
.ok_or("canonical config migration requires a resume catalog")?;
let legacy = resume
.collections
.get("red_config")
.ok_or("canonical config migration requires the red_config resume identity")?;
let canonical = resume
.collections
.get("red.config")
.ok_or("canonical config migration requires the red.config resume identity")?;
let legacy_records = usize::try_from(legacy.entities)?;
match (legacy.entities > 0, canonical.entities > 0) {
(true, true) => {
return Err(
"both red_config and red.config contain records; refusing lossy migration"
.into(),
)
}
(true, false) => {
let legacy = resume
.collections
.remove("red_config")
.ok_or("legacy config resume identity disappeared")?;
resume.collections.remove("red.config");
resume.collections.insert("red.config".into(), legacy);
self.roots.retain(|root| root.collection != "red.config");
let root = self
.roots
.iter_mut()
.find(|root| root.collection == "red_config")
.ok_or("legacy config root disappeared")?;
root.collection = "red.config".into();
self.header.collection_count = u32::try_from(self.roots.len())?;
Ok(legacy_records)
}
(false, _) => {
resume.collections.remove("red_config");
self.roots.retain(|root| root.collection != "red_config");
self.header.collection_count = u32::try_from(self.roots.len())?;
Ok(0)
}
}
}
pub fn encode(&self, roots: &BTreeMap<u32, u32>) -> Result<Vec<u8>> {
self.encode_with_proofs(roots, self.proofs.as_deref())
}
pub fn encode_with_proofs(
&self,
roots: &BTreeMap<u32, u32>,
proofs: Option<&f::page_proof::CatalogProofs>,
) -> Result<Vec<u8>> {
let remap = |root: u32| -> Result<u32> {
if root == 0 {
Ok(0)
} else {
roots
.get(&root)
.copied()
.ok_or_else(|| "missing rewritten tree".into())
}
};
let mut data = Vec::new();
f::encode_native_paged_metadata_header(&mut data, self.header);
for root in &self.roots {
f::encode_native_paged_collection_root(
&mut data,
&root.collection,
remap(root.root_page)?,
);
}
data.extend_from_slice(&u32::try_from(self.references.len())?.to_le_bytes());
for reference in &self.references {
f::encode_native_paged_cross_ref(
&mut data,
reference.source_id,
reference.target_id,
reference.ref_type,
&reference.target_collection,
);
}
f::encode_native_paged_retired_trees(&mut data, &[])?;
f::encode_native_paged_deferred_pages(&mut data, &[])?;
if let Some(epoch) = self.epoch {
f::native_store::encode_native_paged_ownership_epoch(&mut data, epoch)?;
}
if let Some(proofs) = proofs {
f::page_proof::encode_catalog(&mut data, proofs)?;
}
if let Some(mut resume) = self.resume.clone() {
for state in resume.collections.values_mut() {
state.root_page = remap(state.root_page)?;
state.key_index_root = state.key_index_root.map(remap).transpose()?;
for index in state.secondary_indices.values_mut() {
index.root_page = remap(index.root_page)?;
}
}
f::native_resume::encode_catalog_resume(&mut data, &resume)?;
}
Ok(data)
}
}
pub(super) fn write_catalog(target: &Pager, data: &[u8]) -> Result<()> {
crate::storage::unified::UnifiedStore::publish_native_catalog(target, data)?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
fn resume(root_page: u32) -> f::native_resume::CollectionResume {
f::native_resume::CollectionResume {
root_page,
entities: 1,
entity_id_next: 2,
row_id_next: 2,
sequence_id_next: 2,
key_index_root: None,
secondary_indices: BTreeMap::new(),
metadata_expiries: Vec::new(),
}
}
fn duplicate_config_catalog() -> Catalog {
Catalog {
proofs: None,
header: f::NativePagedMetadataHeader {
format_version: 11,
collection_count: 3,
},
roots: vec![
f::NativePagedCollectionRoot {
collection: "users".into(),
root_page: 10,
},
f::NativePagedCollectionRoot {
collection: "red_config".into(),
root_page: 20,
},
f::NativePagedCollectionRoot {
collection: "red.config".into(),
root_page: 30,
},
],
references: Vec::new(),
epoch: Some(7),
retired_roots: Vec::new(),
deferred_pages: Vec::new(),
resume: Some(f::native_resume::CatalogResume {
entity_id_next: 100,
xid_high_water_mark: Some(8),
metadata_expiry_prepared: false,
collections: [
("users".into(), resume(10)),
("red_config".into(), resume(20)),
("red.config".into(), resume(30)),
]
.into_iter()
.collect(),
}),
}
}
#[test]
fn explicit_repair_retires_only_legacy_config_identity() {
let mut catalog = duplicate_config_catalog();
catalog.retain_canonical_system_config().unwrap();
assert_eq!(catalog.header.collection_count, 2);
assert_eq!(
catalog
.roots
.iter()
.map(|root| root.collection.as_str())
.collect::<Vec<_>>(),
["users", "red.config"]
);
let resume = catalog.resume.unwrap();
assert!(!resume.collections.contains_key("red_config"));
assert_eq!(resume.collections["red.config"].root_page, 30);
assert_eq!(resume.collections["users"].root_page, 10);
}
#[test]
fn explicit_repair_rejects_legacy_cross_references() {
let mut catalog = duplicate_config_catalog();
catalog.references.push(f::NativePagedCrossRef {
source_id: 1,
target_id: 2,
ref_type: 0,
target_collection: "red_config".into(),
});
assert!(catalog
.retain_canonical_system_config()
.unwrap_err()
.to_string()
.contains("cross-references"));
}
#[test]
fn migration_promotes_populated_legacy_config_and_retires_empty_canonical() {
let mut catalog = duplicate_config_catalog();
catalog
.resume
.as_mut()
.unwrap()
.collections
.get_mut("red.config")
.unwrap()
.entities = 0;
assert_eq!(catalog.migrate_legacy_system_config().unwrap(), 1);
assert_eq!(catalog.header.collection_count, 2);
assert_eq!(
catalog
.roots
.iter()
.map(|root| (root.collection.as_str(), root.root_page))
.collect::<Vec<_>>(),
[("users", 10), ("red.config", 20)]
);
let resume = catalog.resume.unwrap();
assert!(!resume.collections.contains_key("red_config"));
assert_eq!(resume.collections["red.config"].root_page, 20);
}
#[test]
fn migration_refuses_two_populated_config_trees() {
let mut catalog = duplicate_config_catalog();
assert!(catalog
.migrate_legacy_system_config()
.unwrap_err()
.to_string()
.contains("refusing lossy migration"));
}
}