reddb_server/storage/unified/devx/reddb/

impl_metadata.rs

use super::*;
use crate::storage::unified::metadata::MetadataValue;

const PHYSICAL_METADATA_KEY: &str = "physical";

#[cfg(test)]
fn metadata_qualification_phase(started: &std::time::Instant, name: &str) {
    if std::env::var_os("REDDB_QUALIFICATION_CHECKPOINT_PHASES").is_some() {
        eprintln!(
            "runtime qualification phase=checkpoint_metadata name={name} elapsed_us={}",
            started.elapsed().as_micros()
        );
    }
}

impl RedDB {
    /// Physical metadata follows the same WAL and checkpoint boundary as data.
    /// Legacy sidecars are import sources only, never a publication target.
    pub(crate) fn persist_physical_metadata_record(
        &self,
        metadata: &PhysicalMetadataFile,
    ) -> Result<(), Box<dyn std::error::Error>> {
        if self.options.read_only {
            return Err("physical metadata operations are not allowed in read-only mode".into());
        }
        let value = crate::serde_json::Value::String(metadata.to_document_json(false)?);
        self.store
            .try_set_storage_metadata(PHYSICAL_METADATA_KEY, &value)?;
        Ok(())
    }

    fn load_internal_physical_metadata(
        &self,
    ) -> Result<Option<PhysicalMetadataFile>, Box<dyn std::error::Error>> {
        let Some(value) = self.store.try_get_storage_metadata(PHYSICAL_METADATA_KEY)? else {
            return Ok(None);
        };
        let Value::Text(document) = value else {
            return Err(std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "invalid internal physical metadata record",
            )
            .into());
        };
        Ok(Some(PhysicalMetadataFile::from_document_json(
            &document,
            "invalid internal physical metadata record",
        )?))
    }

    fn has_legacy_metadata_sidecars(&self) -> std::io::Result<bool> {
        let Some(path) = self.path() else {
            return Ok(false);
        };
        Ok(
            PhysicalMetadataFile::metadata_binary_path_for(path).try_exists()?
                || PhysicalMetadataFile::metadata_path_for(path).try_exists()?
                || !PhysicalMetadataFile::journal_paths_for_data_path(path)?.is_empty(),
        )
    }

    /// Called only after the internal record has reached a successful checkpoint.
    /// Failed imports and read-only opens must leave the legacy source intact.
    pub(crate) fn remove_imported_metadata_sidecars(
        &self,
        path: &Path,
    ) -> Result<(), Box<dyn std::error::Error>> {
        if self.options.read_only
            || !self.has_legacy_metadata_sidecars()?
            || self.load_internal_physical_metadata()?.is_none()
        {
            return Ok(());
        }
        reddb_file::physical_metadata::remove_imported_physical_metadata_sidecars(path)?;
        Ok(())
    }

    pub fn is_replica_role(&self) -> bool {
        matches!(
            self.options.replication.role,
            crate::replication::ReplicationRole::Replica { .. }
        )
    }

    pub fn enforce_retention_policy(&self) -> Result<(), Box<dyn std::error::Error>> {
        if self.options.read_only || self.is_replica_role() {
            return Ok(());
        }

        // Export pruning is only meaningful for persistent mode where we
        // track file-backed export artifacts in the internal metadata record.
        if self.options.mode == StorageMode::Persistent {
            let Some(_) = self.path() else {
                return Ok(());
            };

            let _publication = self.store.vault_publication_lock.lock();
            let Ok(mut metadata) = self.load_or_bootstrap_physical_metadata(true) else {
                return Ok(());
            };

            self.prune_export_registry(&mut metadata.exports);
            self.persist_physical_metadata_record(&metadata)?;
        }

        Ok(())
    }

    pub fn entity_visible_at_read(
        &self,
        collection: &str,
        entity: &crate::storage::UnifiedEntity,
    ) -> bool {
        if self
            .store
            .get_collection(collection)
            .is_some_and(|manager| {
                manager.metadata_expiry_projection_ready()
                    && manager.scheduled_metadata_expiry_count() == 0
            })
        {
            return true;
        }
        !self.entity_expired_at(collection, entity, current_unix_ms())
    }

    pub(crate) fn entity_expired_at(
        &self,
        collection: &str,
        entity: &crate::storage::UnifiedEntity,
        now_ms: u64,
    ) -> bool {
        match self.try_entity_expired_at(collection, entity, now_ms) {
            Ok(expired) => expired,
            Err(error) => {
                tracing::warn!(%error, collection, entity_id = entity.id.raw(), "TTL metadata read failed closed");
                true
            }
        }
    }

    pub(crate) fn try_entity_expired_at(
        &self,
        collection: &str,
        entity: &crate::storage::UnifiedEntity,
        now_ms: u64,
    ) -> Result<bool, crate::storage::StoreError> {
        let Some(metadata) = self
            .store
            .try_get_metadata(collection, entity.id, usize::MAX)?
        else {
            return Ok(false);
        };

        Ok(Self::entity_metadata_expired_at(entity, &metadata, now_ms))
    }

    pub(crate) fn entity_metadata_expired_at(
        entity: &crate::storage::UnifiedEntity,
        metadata: &crate::storage::unified::metadata::Metadata,
        now_ms: u64,
    ) -> bool {
        if metadata
            .get("_expires_at")
            .and_then(Self::metadata_u64)
            .is_some_and(|expires_at_ms| expires_at_ms <= now_ms)
        {
            return true;
        }

        let ttl_ms = metadata.get("_ttl_ms").and_then(Self::metadata_u64);
        let ttl_secs = if ttl_ms.is_none() {
            metadata.get("_ttl").and_then(|value| {
                Self::metadata_u64(value).and_then(|value_secs| value_secs.checked_mul(1000))
            })
        } else {
            None
        };

        let Some(ttl_ms) = ttl_ms.or(ttl_secs) else {
            return false;
        };
        entity
            .created_at
            .saturating_mul(1000)
            .saturating_add(ttl_ms)
            <= now_ms
    }

    fn metadata_u64(value: &MetadataValue) -> Option<u64> {
        match value {
            MetadataValue::Int(v) if *v >= 0 => Some(*v as u64),
            MetadataValue::Timestamp(v) => Some(*v),
            MetadataValue::Float(v) => {
                if !v.is_finite() || !v.is_sign_positive() || v.fract().abs() >= f64::EPSILON {
                    return None;
                }
                if *v > u64::MAX as f64 {
                    return None;
                }
                Some(v.trunc() as u64)
            }
            MetadataValue::String(v) => v.parse::<u64>().ok(),
            _ => None,
        }
    }

    // ========================================================================
    // Builder Methods - Create Entities
    // ========================================================================

    /// Start building a graph node
    ///
    /// # Example
    /// ```ignore
    /// let host = db.node("hosts", "Host")
    ///     .property("ip", "192.168.1.1")
    ///     .save()?;
    /// ```
    pub fn node(&self, collection: impl Into<String>, label: impl Into<String>) -> NodeBuilder {
        NodeBuilder::new(
            self.store.clone(),
            self.preprocessors.clone(),
            collection,
            label,
        )
    }

    /// Start building a graph edge
    ///
    /// # Example
    /// ```ignore
    /// let edge = db.edge("connections", "CONNECTS_TO")
    ///     .from(host_a)
    ///     .to(host_b)
    ///     .weight(0.95)
    ///     .property("protocol", "TCP")
    ///     .save()?;
    /// ```
    pub fn edge(&self, collection: impl Into<String>, label: impl Into<String>) -> EdgeBuilder {
        EdgeBuilder::new(
            self.store.clone(),
            self.preprocessors.clone(),
            collection,
            label,
        )
    }

    /// Start building a vector entry
    ///
    /// # Example
    /// ```ignore
    /// let vec = db.vector("embeddings")
    ///     .dense(embedding)
    ///     .content("Original text content")
    ///     .metadata("source", "document.pdf")
    ///     .save()?;
    /// ```
    pub fn vector(&self, collection: impl Into<String>) -> VectorBuilder {
        VectorBuilder::new(self.store.clone(), self.preprocessors.clone(), collection)
    }

    /// Start building a table row
    ///
    /// # Example
    /// ```ignore
    /// let row = db.row("scans", vec![
    ///     ("timestamp", Value::Timestamp(now)),
    ///     ("target", Value::text("192.168.1.0/24")),
    ///     ("findings", Value::Integer(42)),
    /// ]).save()?;
    /// ```
    pub fn row(&self, table: impl Into<String>, columns: Vec<(&str, Value)>) -> RowBuilder {
        RowBuilder::new(
            self.store.clone(),
            self.preprocessors.clone(),
            table,
            columns,
        )
    }

    /// Start building a document
    ///
    /// Documents are stored as enriched table rows with a full JSON body
    /// field and flattened top-level keys for filtering.
    ///
    /// # Example
    /// ```ignore
    /// let doc = db.doc("articles")
    ///     .field("title", "Hello World")
    ///     .field("views", 42)
    ///     .metadata("source", "web")
    ///     .save()?;
    /// ```
    pub fn doc(&self, collection: impl Into<String>) -> DocumentBuilder {
        DocumentBuilder::new(self.store.clone(), self.preprocessors.clone(), collection)
    }

    /// Start building a key-value pair
    ///
    /// KV pairs are stored as table rows with named fields `key` and `value`.
    ///
    /// # Example
    /// ```ignore
    /// let id = db.kv("config", "theme", Value::text("dark"))
    ///     .metadata("updated_by", "admin")
    ///     .save()?;
    /// ```
    pub fn kv(
        &self,
        collection: impl Into<String>,
        key: impl Into<String>,
        value: Value,
    ) -> KvBuilder {
        KvBuilder::new(
            self.store.clone(),
            self.preprocessors.clone(),
            collection,
            key,
            value,
        )
    }

    /// Get a key-value pair by key, returning the value and entity id
    ///
    /// Clone only the selected value. Legacy positional rows without an
    /// indexed key use their collection schema in the borrowed fallback scan.
    pub fn get_kv(&self, collection: &str, key: &str) -> Option<(Value, EntityId)> {
        let manager = self.store.get_collection(collection)?;
        if let Some(value) = manager.project_key_entity(key, None, false, |entity| {
            (
                entity
                    .data
                    .as_row()
                    .and_then(|row| row.get_field("value"))
                    .cloned()
                    .unwrap_or(Value::Null),
                entity.id,
            )
        }) {
            return Some(value);
        }
        let columns = manager.column_schema();
        let key_index = columns
            .as_ref()
            .and_then(|names| names.iter().position(|name| name == "key"));
        let value_index = columns
            .as_ref()
            .and_then(|names| names.iter().position(|name| name == "value"));
        let mut found = None;
        manager.for_each_entity(|entity| {
            let EntityData::Row(row) = &entity.data else {
                return true;
            };
            let legacy = row.named.is_none() && row.schema.is_none();
            let stored_key = row.get_field("key").or_else(|| {
                if legacy {
                    key_index.and_then(|index| row.columns.get(index))
                } else {
                    None
                }
            });
            if matches!(stored_key, Some(Value::Text(stored)) if stored.as_ref() == key) {
                let value = row.get_field("value").or_else(|| {
                    if legacy {
                        value_index.and_then(|index| row.columns.get(index))
                    } else {
                        None
                    }
                });
                found = Some((value.cloned().unwrap_or(Value::Null), entity.id));
                return false;
            }
            true
        });
        found
    }

    /// Delete a key-value pair by key, returning whether it was found and removed
    pub fn delete_kv(
        &self,
        collection: &str,
        key: &str,
    ) -> Result<bool, super::super::error::DevXError> {
        let Some((_, id)) = self.get_kv(collection, key) else {
            return Ok(false);
        };
        self.store
            .delete(collection, id)
            .map_err(|err| super::super::error::DevXError::Storage(format!("{err:?}")))?;
        Ok(true)
    }

    pub(crate) fn with_initialized_metadata(self) -> Result<Self, Box<dyn std::error::Error>> {
        if self.options.mode == StorageMode::Persistent {
            // A damaged authoritative record must never fall back to stale sidecars.
            let has_internal_metadata = self.load_internal_physical_metadata()?.is_some();
            // Load metadata without persisting (avoids blocking catalog snapshot on boot)
            match self.load_or_bootstrap_physical_metadata(false) {
                Ok(metadata) => {
                    crate::reserved_fields::validate_physical_metadata_contracts(&metadata)
                        .map_err(|err| err.to_string())?;
                    if !has_internal_metadata && !self.options.read_only {
                        self.persist_physical_metadata_record(&metadata)?;
                    }
                }
                Err(error)
                    if error
                        .downcast_ref::<std::io::Error>()
                        .is_some_and(crate::physical::is_collection_contract_decode_error) =>
                {
                    return Err(error);
                }
                Err(error) => {
                    if self.has_legacy_metadata_sidecars()? {
                        return Err(error);
                    }
                }
            }
            // Skip repair on boot — deferred to first explicit persist_metadata() call.
            // This avoids the recursive catalog_model_snapshot → physical_metadata loop
            // that caused stack overflow / 12-second hang on startup.
        }
        self.load_collection_ttl_defaults_from_metadata();
        // Single-file artifact: collection contracts ride the store's binary
        // dump, not the metadata sidecar (which is never written in this
        // profile). Seed the contract cache from that blob so each
        // collection's declared model survives a restart — otherwise recovery
        // re-infers the model from the stored entities and a KV collection
        // comes back as a table.
        if self.options.storage_profile.deploy_profile == crate::storage::DeployProfile::Embedded
            && self.options.storage_profile.packaging
                == crate::storage::StoragePackaging::SingleFile
            && !self.paged_mode
        {
            self.seed_contract_cache_from_store_aux()?;
        }
        // Issue #866 — rehydrate the hypertable chunk spine before the
        // API opens so chunk routing / pruning / TTL work immediately
        // after a restart.
        self.load_hypertables_from_metadata();
        // Pending deliveries retain their persisted lock deadlines on recovery
        // (ADR 0020). The lifecycle lazily reclaims expired locks on delivery;
        // deleting every pending row here would redeliver still-owned work.
        Ok(self)
    }

    pub(crate) fn persist_metadata(&self) -> Result<(), Box<dyn std::error::Error>> {
        if self.options.mode != StorageMode::Persistent || self.options.read_only {
            return Ok(());
        }
        // Default embedded options can reopen an existing paged file. Its
        // native headers and index artifacts still require publication.
        if !self.paged_mode
            && self.options.storage_profile.deploy_profile
                == crate::storage::DeployProfile::Embedded
            && self.options.storage_profile.packaging
                == crate::storage::StoragePackaging::SingleFile
        {
            return Ok(());
        }
        let Some(_) = self.path() else {
            return Ok(());
        };

        #[cfg(test)]
        let qualification_started = std::time::Instant::now();
        let _publication = self.store.vault_publication_lock.lock();
        #[cfg(test)]
        metadata_qualification_phase(&qualification_started, "publication-lock");
        let previous = self.load_or_bootstrap_physical_metadata(false).ok();
        #[cfg(test)]
        metadata_qualification_phase(&qualification_started, "previous-metadata");
        let collection_roots = self.physical_collection_roots();
        #[cfg(test)]
        metadata_qualification_phase(&qualification_started, "collection-roots");
        let native_state = self.native_physical_state();
        #[cfg(test)]
        metadata_qualification_phase(&qualification_started, "native-state");
        let mut indexes = native_state
            .map(|state| self.physical_index_state_from_native_state(&state, previous.as_ref()))
            .unwrap_or_else(|| self.physical_index_state());
        if let Some(previous) = previous.as_ref() {
            // Paged contract edits publish the internal record first. Do not
            // resurrect a dropped collection from the older native registry.
            // Preserve unrelated orphan descriptors without a removal proof.
            let absent = |name: &str| {
                !collection_roots.contains_key(name)
                    && !previous
                        .collection_contracts
                        .iter()
                        .any(|contract| contract.name == name)
            };
            let mut removed: BTreeSet<&str> = previous
                .superblock
                .collection_roots
                .keys()
                .map(String::as_str)
                .filter(|name| absent(name))
                .collect();
            let mut seen = BTreeSet::new();
            for event in previous.manifest_events.iter().rev() {
                if seen.insert(event.collection.as_str())
                    && matches!(&event.kind, crate::physical::ManifestEventKind::Remove)
                    && absent(&event.collection)
                {
                    removed.insert(event.collection.as_str());
                }
            }
            indexes.retain(|index| {
                !index
                    .collection
                    .as_deref()
                    .is_some_and(|collection| removed.contains(collection))
            });
        }
        #[cfg(test)]
        metadata_qualification_phase(&qualification_started, "index-state");
        let mut metadata = PhysicalMetadataFile::from_state(
            self.options.clone(),
            self.catalog_snapshot(),
            collection_roots,
            indexes,
            previous.as_ref(),
        );
        metadata.collection_ttl_defaults_ms = self.collection_ttl_defaults_snapshot();
        // Issue #866 — persist the hypertable chunk spine (specs +
        // chunk bounds / routing / TTL) onto the same metadata path as
        // collection contracts so a restart recovers it identically.
        metadata.hypertables = self.hypertable_registry_snapshot();
        #[cfg(test)]
        metadata_qualification_phase(&qualification_started, "metadata-assembled");
        self.persist_native_physical_header(&metadata)?;
        #[cfg(test)]
        metadata_qualification_phase(&qualification_started, "native-header");
        Ok(())
    }

    fn bootstrap_metadata_from_native_state(&self) -> Result<bool, Box<dyn std::error::Error>> {
        if self.options.mode != StorageMode::Persistent || self.options.read_only {
            return Ok(false);
        }
        let Some(_) = self.path() else {
            return Ok(false);
        };
        let Some(native_state) = self.native_physical_state() else {
            return Ok(false);
        };
        if !Self::native_state_is_bootstrap_complete(&native_state) {
            return Ok(false);
        }

        let _publication = self.store.vault_publication_lock.lock();
        let previous = self.load_or_bootstrap_physical_metadata(false).ok();
        let metadata = self.metadata_from_native_state(&native_state, previous.as_ref());
        self.persist_native_physical_header(&metadata)?;
        Ok(true)
    }

    /// Rebuild the external physical metadata view from the native state published in the
    /// paged database file.
    pub fn rebuild_physical_metadata_from_native_state(
        &self,
    ) -> Result<bool, Box<dyn std::error::Error>> {
        self.bootstrap_metadata_from_native_state()
    }

    pub(crate) fn native_state_is_bootstrap_complete(native_state: &NativePhysicalState) -> bool {
        let registry_complete = native_state.registry.as_ref().map(|registry| {
            registry.collections_complete
                && registry.indexes_complete
                && registry.graph_projections_complete
                && registry.analytics_jobs_complete
                && registry.vector_artifacts_complete
        });
        let recovery_complete = native_state
            .recovery
            .as_ref()
            .map(|recovery| recovery.snapshots_complete && recovery.exports_complete);
        let catalog_complete = native_state
            .catalog
            .as_ref()
            .map(|catalog| catalog.collections_complete);

        registry_complete == Some(true)
            && recovery_complete == Some(true)
            && catalog_complete == Some(true)
    }

    fn native_state_is_summary_free_legacy(native_state: &NativePhysicalState) -> bool {
        let header = native_state.header;
        header.manifest_page == 0
            && header.manifest_checksum == 0
            && header.registry_page == 0
            && header.registry_checksum == 0
            && header.recovery_page == 0
            && header.recovery_checksum == 0
            && header.catalog_page == 0
            && header.catalog_checksum == 0
            && header.metadata_state_page == 0
            && header.metadata_state_checksum == 0
            && header.vector_artifact_page == 0
            && header.vector_artifact_checksum == 0
            && native_state.manifest.is_none()
            && native_state.registry.is_none()
            && native_state.recovery.is_none()
            && native_state.catalog.is_none()
            && native_state.metadata_state.is_none()
            && native_state.vector_artifact_pages.is_none()
    }

    pub(crate) fn load_or_bootstrap_physical_metadata(
        &self,
        persist_bootstrapped: bool,
    ) -> Result<PhysicalMetadataFile, Box<dyn std::error::Error>> {
        if self.options.mode != StorageMode::Persistent {
            return Err("physical metadata requires persistent mode".into());
        }
        let Some(path) = self.path() else {
            return Err("database path is not available".into());
        };
        let _publication = self.store.vault_publication_lock.lock();
        let internal_metadata = match self.load_internal_physical_metadata()? {
            Some(metadata) => {
                // The internal document is authoritative. Only load all native
                // summaries when their header proves a newer publication needs
                // reconciliation; ordinary catalog reads need no artifact I/O.
                let native_ahead = self.store.physical_file_header().is_some_and(|header| {
                    Self::repair_policy_for_inspection(
                        &Self::inspect_native_header_against_metadata(header, &metadata),
                    ) == NativeHeaderRepairPolicy::NativeAheadOfMetadata
                });
                if !native_ahead {
                    return Ok(metadata);
                }
                Some(metadata)
            }
            None => None,
        };
        let native_state = self.native_physical_state();
        let metadata = match internal_metadata {
            Some(metadata) => Ok(metadata),
            None => PhysicalMetadataFile::inspect_for_data_path(path),
        };
        match metadata {
            Ok(metadata) => {
                if let Some(native_state) = native_state.as_ref() {
                    let inspection = Self::inspect_native_header_against_metadata(
                        native_state.header,
                        &metadata,
                    );
                    if Self::repair_policy_for_inspection(&inspection)
                        == NativeHeaderRepairPolicy::NativeAheadOfMetadata
                    {
                        let bootstrapped =
                            self.metadata_from_native_state(native_state, Some(&metadata));
                        if persist_bootstrapped && !self.options.read_only {
                            self.persist_native_physical_header(&bootstrapped)?;
                        }
                        return Ok(bootstrapped);
                    }
                }
                Ok(metadata)
            }
            Err(err) => {
                if crate::physical::is_collection_contract_decode_error(&err)
                    || self.has_legacy_metadata_sidecars()?
                {
                    return Err(err.into());
                }
                let Some(native_state) = native_state else {
                    return Err(err.into());
                };
                // Summary-free paged files predate native registry/catalog/recovery
                // publication. Their catalog sequence can legitimately advance during
                // an in-place identity migration, so sequence zero is not a format
                // discriminator. Admit only the exact legacy shape: every summary
                // pointer and checksum is zero. A non-zero unreadable pointer remains a
                // partial/corrupt publication and must fail closed.
                if !Self::native_state_is_summary_free_legacy(&native_state)
                    && !Self::native_state_is_bootstrap_complete(&native_state)
                {
                    return Err(err.into());
                }
                let metadata = self.metadata_from_native_state(&native_state, None);
                if persist_bootstrapped && !self.options.read_only {
                    self.persist_native_physical_header(&metadata)?;
                }
                Ok(metadata)
            }
        }
    }

    pub(crate) fn physical_metadata_preference(&self) -> Option<&'static str> {
        if self
            .load_internal_physical_metadata()
            .ok()
            .flatten()
            .is_some()
        {
            return Some("internal_record");
        }
        let path = self.path()?;
        let native_state = self.native_physical_state();
        let metadata = PhysicalMetadataFile::inspect_for_data_path(path).ok();

        match (metadata, native_state) {
            (Some(metadata), Some(native_state)) => {
                let inspection =
                    Self::inspect_native_header_against_metadata(native_state.header, &metadata);
                match Self::repair_policy_for_inspection(&inspection) {
                    NativeHeaderRepairPolicy::InSync => Some("sidecar_current"),
                    NativeHeaderRepairPolicy::RepairNativeFromMetadata => Some("sidecar_current"),
                    NativeHeaderRepairPolicy::NativeAheadOfMetadata => Some("native_ahead"),
                }
            }
            (Some(_), None) => Some("sidecar_only"),
            (None, Some(_)) => Some("sidecar_missing_native_available"),
            (None, None) => Some("sidecar_missing_no_native"),
        }
    }

    fn metadata_from_native_state(
        &self,
        native_state: &NativePhysicalState,
        previous: Option<&PhysicalMetadataFile>,
    ) -> PhysicalMetadataFile {
        let now = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .as_millis();
        let catalog = self.catalog_snapshot();
        let catalog_name = catalog.name.clone();
        let catalog_total_entities = catalog.total_entities;
        let catalog_total_collections = catalog.total_collections;
        let indexes = self.physical_index_state_from_native_state(native_state, previous);

        let mut manifest =
            crate::api::SchemaManifest::now(self.options.clone(), catalog.total_collections);
        manifest.updated_at_unix_ms = now;

        let manifest_events = native_state
            .manifest
            .as_ref()
            .map(|summary| {
                summary
                    .recent_events
                    .iter()
                    .map(|event| crate::physical::ManifestEvent {
                        collection: event.collection.clone(),
                        object_key: event.object_key.clone(),
                        kind: match event.kind.as_str() {
                            "insert" => crate::physical::ManifestEventKind::Insert,
                            "update" => crate::physical::ManifestEventKind::Update,
                            "remove" => crate::physical::ManifestEventKind::Remove,
                            _ => crate::physical::ManifestEventKind::Checkpoint,
                        },
                        block: crate::physical::BlockReference {
                            index: event.block_index,
                            checksum: event.block_checksum,
                        },
                        snapshot_min: event.snapshot_min,
                        snapshot_max: event.snapshot_max,
                    })
                    .collect()
            })
            .unwrap_or_default();

        let graph_projections = native_state
            .registry
            .as_ref()
            .and_then(|registry| {
                registry.graph_projections_complete.then(|| {
                    registry
                        .graph_projections
                        .iter()
                        .map(|projection| crate::physical::PhysicalGraphProjection {
                            name: projection.name.clone(),
                            created_at_unix_ms: projection.created_at_unix_ms,
                            updated_at_unix_ms: projection.updated_at_unix_ms,
                            state: "materialized".to_string(),
                            source: projection.source.clone(),
                            node_labels: projection.node_labels.clone(),
                            node_types: projection.node_types.clone(),
                            edge_labels: projection.edge_labels.clone(),
                            last_materialized_sequence: projection.last_materialized_sequence,
                        })
                        .collect()
                })
            })
            .or_else(|| previous.map(|metadata| metadata.graph_projections.clone()))
            .unwrap_or_default();

        let analytics_jobs = native_state
            .registry
            .as_ref()
            .and_then(|registry| {
                registry.analytics_jobs_complete.then(|| {
                    registry
                        .analytics_jobs
                        .iter()
                        .map(|job| crate::physical::PhysicalAnalyticsJob {
                            id: job.id.clone(),
                            kind: job.kind.clone(),
                            state: job.state.clone(),
                            projection: job.projection.clone(),
                            created_at_unix_ms: job.created_at_unix_ms,
                            updated_at_unix_ms: job.updated_at_unix_ms,
                            last_run_sequence: job.last_run_sequence,
                            metadata: job.metadata.clone(),
                        })
                        .collect()
                })
            })
            .or_else(|| previous.map(|metadata| metadata.analytics_jobs.clone()))
            .unwrap_or_default();

        let exports = native_state
            .recovery
            .as_ref()
            .and_then(|recovery| {
                recovery.exports_complete.then(|| {
                    recovery
                        .exports
                        .iter()
                        .map(|export| crate::physical::ExportDescriptor {
                            name: export.name.clone(),
                            created_at_unix_ms: export.created_at_unix_ms,
                            snapshot_id: export.snapshot_id,
                            superblock_sequence: export.superblock_sequence,
                            data_path: self
                                .path()
                                .map(|path| {
                                    crate::physical::PhysicalMetadataFile::export_data_path_for(
                                        path,
                                        &export.name,
                                    )
                                    .display()
                                    .to_string()
                                })
                                .unwrap_or_default(),
                            metadata_path: self
                                .path()
                                .map(|path| {
                                    let export_data_path =
                                        crate::physical::PhysicalMetadataFile::export_data_path_for(
                                            path,
                                            &export.name,
                                        );
                                    crate::physical::PhysicalMetadataFile::metadata_path_for(
                                        &export_data_path,
                                    )
                                    .display()
                                    .to_string()
                                })
                                .unwrap_or_default(),
                            collection_count: export.collection_count as usize,
                            total_entities: export.total_entities as usize,
                        })
                        .collect()
                })
            })
            .or_else(|| previous.map(|metadata| metadata.exports.clone()))
            .unwrap_or_default();

        let snapshots = native_state
            .recovery
            .as_ref()
            .and_then(|recovery| {
                recovery.snapshots_complete.then(|| {
                    recovery
                        .snapshots
                        .iter()
                        .map(|snapshot| crate::physical::SnapshotDescriptor {
                            snapshot_id: snapshot.snapshot_id,
                            created_at_unix_ms: snapshot.created_at_unix_ms,
                            superblock_sequence: snapshot.superblock_sequence,
                            collection_count: snapshot.collection_count as usize,
                            total_entities: snapshot.total_entities as usize,
                        })
                        .collect()
                })
            })
            .or_else(|| previous.map(|metadata| metadata.snapshots.clone()))
            .unwrap_or_else(|| {
                vec![crate::physical::SnapshotDescriptor {
                    snapshot_id: native_state.header.sequence,
                    created_at_unix_ms: now,
                    superblock_sequence: native_state.header.sequence,
                    collection_count: catalog_total_collections,
                    total_entities: catalog_total_entities,
                }]
            });

        let catalog_stats = native_state
            .catalog
            .as_ref()
            .and_then(|native_catalog| {
                native_catalog.collections_complete.then(|| {
                    native_catalog
                        .collections
                        .iter()
                        .map(|collection| {
                            (
                                collection.name.clone(),
                                crate::api::CollectionStats {
                                    entities: collection.entities as usize,
                                    cross_refs: collection.cross_refs as usize,
                                    segments: collection.segments as usize,
                                },
                            )
                        })
                        .collect::<BTreeMap<_, _>>()
                })
            })
            .or_else(|| previous.map(|metadata| metadata.catalog.stats_by_collection.clone()))
            .unwrap_or_else(|| catalog.stats_by_collection.clone());

        PhysicalMetadataFile {
            protocol_version: crate::physical::PHYSICAL_METADATA_PROTOCOL_VERSION.to_string(),
            generated_at_unix_ms: now,
            last_loaded_from: Some("native_bootstrap".to_string()),
            last_healed_at_unix_ms: Some(now),
            manifest,
            catalog: crate::api::CatalogSnapshot {
                name: catalog_name,
                total_entities: native_state
                    .catalog
                    .as_ref()
                    .map(|summary| summary.total_entities as usize)
                    .unwrap_or(catalog_total_entities),
                total_collections: native_state
                    .catalog
                    .as_ref()
                    .map(|summary| summary.collection_count as usize)
                    .unwrap_or(catalog_total_collections),
                stats_by_collection: catalog_stats,
                updated_at: SystemTime::now(),
            },
            manifest_events,
            collection_ttl_defaults_ms: previous
                .map(|metadata| metadata.collection_ttl_defaults_ms.clone())
                .unwrap_or_default(),
            collection_contracts: previous
                .map(|metadata| metadata.collection_contracts.clone())
                .unwrap_or_default(),
            hypertables: previous
                .map(|metadata| metadata.hypertables.clone())
                .unwrap_or_default(),
            tree_definitions: previous
                .map(|metadata| metadata.tree_definitions.clone())
                .unwrap_or_default(),
            indexes,
            graph_projections,
            analytics_jobs,
            exports,
            superblock: crate::physical::SuperblockHeader {
                format_version: native_state.header.format_version,
                sequence: native_state.header.sequence,
                copies: crate::physical::DEFAULT_SUPERBLOCK_COPIES,
                manifest: crate::physical::ManifestPointers {
                    oldest: crate::physical::BlockReference {
                        index: native_state.header.manifest_oldest_root,
                        checksum: 0,
                    },
                    newest: crate::physical::BlockReference {
                        index: native_state.header.manifest_root,
                        checksum: 0,
                    },
                },
                free_set: crate::physical::BlockReference {
                    index: native_state.header.free_set_root,
                    checksum: 0,
                },
                collection_roots: native_state.collection_roots.clone(),
            },
            snapshots,
        }
    }

    pub(crate) fn reconcile_index_states_with_native_artifacts(
        &self,
        mut indexes: Vec<PhysicalIndexState>,
    ) -> Vec<PhysicalIndexState> {
        let native_state = self.native_physical_state();
        let native_artifacts = native_state
            .as_ref()
            .and_then(|state| state.registry.as_ref())
            .map(|registry| registry.vector_artifacts.as_slice())
            .unwrap_or_default();
        let artifact_pages = native_state
            .as_ref()
            .and_then(|state| state.vector_artifact_pages.as_ref());
        for index in &mut indexes {
            let Some(collection) = index.collection.as_deref() else {
                continue;
            };
            let Some(artifact_kind) = Self::native_artifact_kind_for_index(index.kind) else {
                continue;
            };
            let Some(artifact) = native_artifacts.iter().find(|artifact| {
                artifact.collection == collection && artifact.artifact_kind == artifact_kind
            }) else {
                index.build_state = if index.backend == "memory-budget-deferred" {
                    "deferred-memory-budget"
                } else {
                    "metadata-only"
                }
                .to_string();
                continue;
            };
            index.entries = artifact.vector_count as usize;
            index.estimated_memory_bytes = artifact.serialized_bytes;
            index.backend = format!("{}+native-artifact", index_backend_name(index.kind));
            index.artifact_kind = Some(artifact.artifact_kind.clone());
            index.artifact_checksum = Some(artifact.checksum);
            index.build_state = "artifact-published".to_string();
            if let Some(pages) = artifact_pages {
                index.artifact_root_page = pages
                    .iter()
                    .find(|page| {
                        page.collection == artifact.collection
                            && page.artifact_kind == artifact.artifact_kind
                    })
                    .map(|page| page.root_page);
            }
        }
        indexes
    }

    pub(crate) fn warmup_native_vector_artifact_for_index(
        &self,
        index: &PhysicalIndexState,
    ) -> Result<(), String> {
        let Some(collection) = index.collection.as_deref() else {
            return Ok(());
        };
        let Some(artifact_kind) = Self::native_artifact_kind_for_index(index.kind) else {
            return Ok(());
        };
        self.warmup_native_vector_artifact(collection, Some(artifact_kind))?;
        Ok(())
    }

    pub(crate) fn apply_runtime_native_artifact_to_index_state(
        &self,
        index: &mut PhysicalIndexState,
    ) -> Result<(), String> {
        let Some(collection) = index.collection.as_deref() else {
            return Ok(());
        };
        let Some(artifact_kind) = Self::native_artifact_kind_for_index(index.kind) else {
            return Ok(());
        };
        let artifact = self.inspect_native_vector_artifact(collection, Some(artifact_kind))?;
        index.entries = artifact
            .graph_edge_count
            .or(artifact.text_posting_count)
            .unwrap_or(artifact.node_count) as usize;
        index.estimated_memory_bytes = artifact.byte_len;
        index.backend = format!("{}+native-artifact", index_backend_name(index.kind));
        index.artifact_kind = Some(artifact.artifact_kind.clone());
        index.artifact_checksum = Some(artifact.checksum);
        index.build_state = "ready".to_string();
        index.artifact_root_page = self
            .native_vector_artifact_pages()
            .and_then(|pages| {
                pages.into_iter().find(|page| {
                    page.collection == artifact.collection
                        && page.artifact_kind == artifact.artifact_kind
                })
            })
            .map(|page| page.root_page);
        Ok(())
    }

    pub(crate) fn physical_index_state_from_native_state(
        &self,
        native_state: &NativePhysicalState,
        previous: Option<&PhysicalMetadataFile>,
    ) -> Vec<PhysicalIndexState> {
        let mut fresh = self.physical_index_state();
        let Some(registry) = native_state.registry.as_ref() else {
            if let Some(previous) = previous {
                for index in &previous.indexes {
                    if !fresh.iter().any(|candidate| candidate.name == index.name) {
                        fresh.push(index.clone());
                    }
                }
            }
            return fresh;
        };

        for index in &mut fresh {
            if let Some(native) = registry
                .indexes
                .iter()
                .find(|candidate| candidate.name == index.name)
            {
                index.enabled = native.enabled;
                index.last_refresh_ms = native.last_refresh_ms;
                index.backend = native.backend.clone();
                index.entries = native.entries as usize;
                index.estimated_memory_bytes = native.estimated_memory_bytes;
                if index.artifact_kind.is_none() {
                    index.artifact_kind = Self::native_artifact_kind_for_index(index.kind)
                        .map(|value| value.to_string());
                }
                if index.build_state == "catalog-derived" {
                    index.build_state = "registry-loaded".to_string();
                }
            }
            if index.backend == "memory-budget-deferred"
                || previous.is_some_and(|metadata| {
                    metadata.indexes.iter().any(|previous| {
                        previous.name == index.name && previous.backend == "memory-budget-deferred"
                    })
                })
            {
                index.backend = "memory-budget-deferred".to_string();
                index.build_state = "deferred-memory-budget".to_string();
                index.artifact_root_page = None;
                index.artifact_checksum = None;
            }
        }

        for native in &registry.indexes {
            if fresh.iter().any(|index| index.name == native.name) {
                continue;
            }
            let Some(kind) = Self::index_kind_from_str(&native.kind) else {
                continue;
            };
            fresh.push(PhysicalIndexState {
                name: native.name.clone(),
                kind,
                collection: native.collection.clone(),
                enabled: native.enabled,
                entries: native.entries as usize,
                estimated_memory_bytes: native.estimated_memory_bytes,
                last_refresh_ms: native.last_refresh_ms,
                backend: native.backend.clone(),
                artifact_kind: Self::native_artifact_kind_for_index(kind)
                    .map(|value| value.to_string()),
                artifact_root_page: None,
                artifact_checksum: None,
                build_state: "registry-loaded".to_string(),
            });
        }

        if !registry.indexes_complete {
            if let Some(previous) = previous {
                for index in &previous.indexes {
                    if !fresh.iter().any(|candidate| candidate.name == index.name) {
                        fresh.push(index.clone());
                    }
                }
            }
        }

        fresh
    }

    pub(crate) fn graph_projections_from_native_state(
        &self,
        native_state: &NativePhysicalState,
    ) -> Vec<PhysicalGraphProjection> {
        native_state
            .registry
            .as_ref()
            .map(|registry| {
                registry
                    .graph_projections
                    .iter()
                    .map(|projection| PhysicalGraphProjection {
                        name: projection.name.clone(),
                        created_at_unix_ms: projection.created_at_unix_ms,
                        updated_at_unix_ms: projection.updated_at_unix_ms,
                        state: "materialized".to_string(),
                        source: projection.source.clone(),
                        node_labels: projection.node_labels.clone(),
                        node_types: projection.node_types.clone(),
                        edge_labels: projection.edge_labels.clone(),
                        last_materialized_sequence: projection.last_materialized_sequence,
                    })
                    .collect()
            })
            .unwrap_or_default()
    }

    pub(crate) fn analytics_jobs_from_native_state(
        &self,
        native_state: &NativePhysicalState,
    ) -> Vec<PhysicalAnalyticsJob> {
        native_state
            .registry
            .as_ref()
            .map(|registry| {
                registry
                    .analytics_jobs
                    .iter()
                    .map(|job| PhysicalAnalyticsJob {
                        id: job.id.clone(),
                        kind: job.kind.clone(),
                        state: job.state.clone(),
                        projection: job.projection.clone(),
                        created_at_unix_ms: job.created_at_unix_ms,
                        updated_at_unix_ms: job.updated_at_unix_ms,
                        last_run_sequence: job.last_run_sequence,
                        metadata: job.metadata.clone(),
                    })
                    .collect()
            })
            .unwrap_or_default()
    }

    pub(crate) fn exports_from_native_state(
        &self,
        native_state: &NativePhysicalState,
    ) -> Vec<ExportDescriptor> {
        native_state
            .recovery
            .as_ref()
            .map(|recovery| {
                recovery
                    .exports
                    .iter()
                    .map(|export| ExportDescriptor {
                        name: export.name.clone(),
                        created_at_unix_ms: export.created_at_unix_ms,
                        snapshot_id: export.snapshot_id,
                        superblock_sequence: export.superblock_sequence,
                        data_path: self
                            .path()
                            .map(|path| {
                                crate::physical::PhysicalMetadataFile::export_data_path_for(
                                    path,
                                    &export.name,
                                )
                                .display()
                                .to_string()
                            })
                            .unwrap_or_default(),
                        metadata_path: self
                            .path()
                            .map(|path| {
                                let export_data_path =
                                    crate::physical::PhysicalMetadataFile::export_data_path_for(
                                        path,
                                        &export.name,
                                    );
                                crate::physical::PhysicalMetadataFile::metadata_path_for(
                                    &export_data_path,
                                )
                                .display()
                                .to_string()
                            })
                            .unwrap_or_default(),
                        collection_count: export.collection_count as usize,
                        total_entities: export.total_entities as usize,
                    })
                    .collect()
            })
            .unwrap_or_default()
    }

    pub(crate) fn snapshots_from_native_state(
        &self,
        native_state: &NativePhysicalState,
    ) -> Vec<crate::physical::SnapshotDescriptor> {
        let snapshots: Vec<_> = native_state
            .recovery
            .as_ref()
            .map(|recovery| {
                recovery
                    .snapshots
                    .iter()
                    .map(|snapshot| crate::physical::SnapshotDescriptor {
                        snapshot_id: snapshot.snapshot_id,
                        created_at_unix_ms: snapshot.created_at_unix_ms,
                        superblock_sequence: snapshot.superblock_sequence,
                        collection_count: snapshot.collection_count as usize,
                        total_entities: snapshot.total_entities as usize,
                    })
                    .collect()
            })
            .unwrap_or_default();
        if !snapshots.is_empty() {
            return snapshots;
        }

        let now = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .as_millis();
        let (collection_count, total_entities) = native_state
            .catalog
            .as_ref()
            .map(|catalog| {
                (
                    catalog.collection_count as usize,
                    catalog.total_entities as usize,
                )
            })
            .unwrap_or_else(|| {
                let catalog = self.catalog_snapshot();
                (catalog.total_collections, catalog.total_entities)
            });

        vec![crate::physical::SnapshotDescriptor {
            snapshot_id: native_state.header.sequence,
            created_at_unix_ms: now,
            superblock_sequence: native_state.header.sequence,
            collection_count,
            total_entities,
        }]
    }

    fn index_kind_from_str(value: &str) -> Option<crate::index::IndexKind> {
        match value {
            "btree" => Some(crate::index::IndexKind::BTree),
            "vector.hnsw" => Some(crate::index::IndexKind::VectorHnsw),
            "vector.inverted" => Some(crate::index::IndexKind::VectorInverted),
            "vector.turbo" => Some(crate::index::IndexKind::VectorTurbo),
            "graph.adjacency" => Some(crate::index::IndexKind::GraphAdjacency),
            "text.fulltext" => Some(crate::index::IndexKind::FullText),
            "document.pathvalue" => Some(crate::index::IndexKind::DocumentPathValue),
            "search.hybrid" => Some(crate::index::IndexKind::HybridSearch),
            _ => None,
        }
    }

    pub(crate) fn native_artifact_kind_for_index(kind: IndexKind) -> Option<&'static str> {
        match kind {
            IndexKind::VectorHnsw => Some("hnsw"),
            IndexKind::VectorInverted => Some("ivf"),
            IndexKind::VectorTurbo => Some("turboquant"),
            IndexKind::GraphAdjacency => Some("graph.adjacency"),
            IndexKind::FullText => Some("text.fulltext"),
            IndexKind::DocumentPathValue => Some("document.pathvalue"),
            _ => None,
        }
    }

    fn index_is_declared(&self, name: &str) -> bool {
        self.physical_metadata()
            .map(|metadata| metadata.indexes.iter().any(|index| index.name == name))
            .unwrap_or(false)
    }

    pub(crate) fn graph_projection_is_declared(&self, name: &str) -> bool {
        self.physical_metadata()
            .map(|metadata| {
                metadata
                    .graph_projections
                    .iter()
                    .any(|projection| projection.name == name)
            })
            .unwrap_or(false)
    }

    pub(crate) fn graph_projection_is_operational(&self, name: &str) -> bool {
        self.operational_graph_projections()
            .into_iter()
            .any(|projection| projection.name == name && projection.state == "materialized")
    }

    pub(crate) fn analytics_job_id(kind: &str, projection: Option<&str>) -> String {
        match projection {
            Some(projection) => format!("{kind}::{projection}"),
            None => format!("{kind}::global"),
        }
    }

    pub(crate) fn update_physical_metadata<T, F>(
        &self,
        mutator: F,
    ) -> Result<T, Box<dyn std::error::Error>>
    where
        F: FnOnce(&mut PhysicalMetadataFile) -> T,
    {
        if self.options.mode != StorageMode::Persistent {
            return Err("physical metadata operations require persistent mode".into());
        }
        if self.options.read_only {
            return Err("physical metadata operations are not allowed in read-only mode".into());
        }
        let Some(_) = self.path() else {
            return Err("database path is not available".into());
        };

        // #2376: the metadata sidecar and the native header pages are
        // published read-modify-write. Two unserialised publishers lost each
        // other's collection contracts and double-freed the header pages
        // (a freed page came back as a B-tree page). Every publisher takes
        // the store's reentrant publication lock, which `persist()` already
        // takes, so this adds no lock-order edge.
        let _publication = self.store.vault_publication_lock.lock();
        let mut metadata = self.load_or_bootstrap_physical_metadata(true)?;

        if metadata.indexes.is_empty() {
            metadata.indexes = self
                .native_physical_state()
                .map(|state| self.physical_index_state_from_native_state(&state, Some(&metadata)))
                .unwrap_or_else(|| self.physical_index_state());
        }
        metadata.superblock.collection_roots = self.physical_collection_roots();

        let result = mutator(&mut metadata);
        self.persist_native_physical_header(&metadata)?;
        Ok(result)
    }

    pub(crate) fn persist_native_physical_header(
        &self,
        metadata: &PhysicalMetadataFile,
    ) -> Result<(), Box<dyn std::error::Error>> {
        if !self.paged_mode {
            return self.persist_physical_metadata_record(metadata);
        }

        let _publication = self.store.vault_publication_lock.lock();
        let existing_page = self
            .store
            .physical_file_header()
            .map(|header| header.collection_roots_page)
            .filter(|page| *page != 0);
        let existing_registry_page = self
            .store
            .physical_file_header()
            .map(|header| header.registry_page)
            .filter(|page| *page != 0);
        let existing_recovery_page = self
            .store
            .physical_file_header()
            .map(|header| header.recovery_page)
            .filter(|page| *page != 0);
        let existing_catalog_page = self
            .store
            .physical_file_header()
            .map(|header| header.catalog_page)
            .filter(|page| *page != 0);
        let existing_metadata_state_page = self
            .store
            .physical_file_header()
            .map(|header| header.metadata_state_page)
            .filter(|page| *page != 0);
        let existing_vector_artifact_page = self
            .store
            .physical_file_header()
            .map(|header| header.vector_artifact_page)
            .filter(|page| *page != 0);
        let existing_manifest_page = self
            .store
            .physical_file_header()
            .map(|header| header.manifest_page)
            .filter(|page| *page != 0);
        let (manifest_page, manifest_checksum) = self.store.write_native_manifest_summary(
            metadata.superblock.sequence,
            &metadata.manifest_events,
            existing_manifest_page,
        )?;
        let (collection_roots_page, collection_roots_checksum) = self
            .store
            .write_native_collection_roots(&metadata.superblock.collection_roots, existing_page)?;
        // Build payload artifacts once, on the publication path. Observers use
        // the persisted summaries instead of rebuilding search indexes.
        let build = self.native_vector_artifact_records()?;
        let vector_artifact_records = build.records;
        let mut metadata = metadata.clone();
        for index in &mut metadata.indexes {
            let Some(kind) = Self::native_artifact_kind_for_index(index.kind) else {
                continue;
            };
            let Some(collection) = &index.collection else {
                continue;
            };
            if build.deferred_collections.contains(collection) {
                index.backend = "memory-budget-deferred".to_string();
                index.build_state = "deferred-memory-budget".to_string();
                index.artifact_kind = Some(kind.to_string());
                index.artifact_root_page = None;
                index.artifact_checksum = None;
            } else if let Some((summary, _)) =
                vector_artifact_records.iter().find(|(summary, _)| {
                    &summary.collection == collection && summary.artifact_kind == kind
                })
            {
                index.backend = format!("{}+native-artifact", index_backend_name(index.kind));
                index.build_state = "artifact-published".to_string();
                index.artifact_kind = Some(kind.to_string());
                index.artifact_checksum = Some(summary.checksum);
            }
        }
        let metadata = &metadata;
        let artifact_summaries = vector_artifact_records
            .iter()
            .map(|(summary, _)| summary.clone())
            .collect::<Vec<_>>();
        let registry_summary = self.native_registry_summary_from_metadata(
            metadata,
            &artifact_summaries,
            artifact_summaries.len(),
        );
        let (registry_page, registry_checksum) = self
            .store
            .write_native_registry_summary(&registry_summary, existing_registry_page)?;
        let recovery_summary = Self::native_recovery_summary_from_metadata(metadata);
        let (recovery_page, recovery_checksum) = self
            .store
            .write_native_recovery_summary(&recovery_summary, existing_recovery_page)?;
        let catalog_summary = Self::native_catalog_summary_from_metadata(metadata);
        let (catalog_page, catalog_checksum) = self
            .store
            .write_native_catalog_summary(&catalog_summary, existing_catalog_page)?;
        let metadata_state_summary = Self::native_metadata_state_summary_from_metadata(metadata);
        let (metadata_state_page, metadata_state_checksum) =
            self.store.write_native_metadata_state_summary(
                &metadata_state_summary,
                existing_metadata_state_page,
            )?;
        let vector_artifact_payloads = vector_artifact_records
            .iter()
            .map(|(summary, bytes)| {
                (
                    summary.collection.clone(),
                    summary.artifact_kind.clone(),
                    bytes.clone(),
                )
            })
            .collect::<Vec<_>>();
        let (vector_artifact_page, vector_artifact_checksum, _vector_artifact_pages) =
            self.store.write_native_vector_artifact_store(
                &vector_artifact_payloads,
                existing_vector_artifact_page,
            )?;
        let mut header = Self::native_header_from_metadata(metadata);
        header.manifest_page = manifest_page;
        header.manifest_checksum = manifest_checksum;
        header.collection_roots_page = collection_roots_page;
        header.collection_roots_checksum = collection_roots_checksum;
        header.registry_page = registry_page;
        header.registry_checksum = registry_checksum;
        header.recovery_page = recovery_page;
        header.recovery_checksum = recovery_checksum;
        header.catalog_page = catalog_page;
        header.catalog_checksum = catalog_checksum;
        header.metadata_state_page = metadata_state_page;
        header.metadata_state_checksum = metadata_state_checksum;
        header.vector_artifact_page = vector_artifact_page;
        header.vector_artifact_checksum = vector_artifact_checksum;
        self.persist_physical_metadata_record(metadata)?;
        self.store.update_physical_file_header(header)?;
        self.store.persist()?;
        self.store.publish_detached_native_blob_pages()?;
        Ok(())
    }

    pub(crate) fn native_header_from_metadata(
        metadata: &PhysicalMetadataFile,
    ) -> PhysicalFileHeader {
        PhysicalFileHeader {
            format_version: metadata.superblock.format_version,
            sequence: metadata.superblock.sequence,
            manifest_oldest_root: metadata.superblock.manifest.oldest.index,
            manifest_root: metadata.superblock.manifest.newest.index,
            free_set_root: metadata.superblock.free_set.index,
            manifest_page: 0,
            manifest_checksum: 0,
            collection_roots_page: 0,
            collection_roots_checksum: 0,
            collection_root_count: metadata.superblock.collection_roots.len() as u32,
            snapshot_count: metadata.snapshots.len() as u32,
            index_count: metadata.indexes.len() as u32,
            catalog_collection_count: metadata.catalog.total_collections as u32,
            catalog_total_entities: metadata.catalog.total_entities as u64,
            export_count: metadata.exports.len() as u32,
            graph_projection_count: metadata.graph_projections.len() as u32,
            analytics_job_count: metadata.analytics_jobs.len() as u32,
            manifest_event_count: metadata.manifest_events.len() as u32,
            registry_page: 0,
            registry_checksum: 0,
            recovery_page: 0,
            recovery_checksum: 0,
            catalog_page: 0,
            catalog_checksum: 0,
            metadata_state_page: 0,
            metadata_state_checksum: 0,
            vector_artifact_page: 0,
            vector_artifact_checksum: 0,
        }
    }
}

fn current_unix_ms() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis()
        .min(u128::from(u64::MAX)) as u64
}

#[cfg(test)]
mod catalog_observation_tests {
    use super::*;

    #[test]
    fn summary_free_legacy_native_state_rebuilds_a_missing_sidecar_after_catalog_publication() {
        let directory = tempfile::tempdir().expect("temporary database");
        let path = directory.path().join("legacy-summary-free.rdb");
        let options = RedDBOptions::persistent(&path)
            .with_memory_budget(64 * 1024 * 1024)
            .with_storage_profile(crate::StorageDeployPreset::PrimaryReplicaDev.selection())
            .expect("paged profile");
        let runtime = crate::RedDBRuntime::with_options(options).expect("paged runtime");
        runtime
            .execute_query("CREATE TABLE retained (value INTEGER)")
            .expect("retained table");
        runtime
            .execute_query("INSERT INTO retained (value) VALUES (17)")
            .expect("retained row");
        runtime.checkpoint().expect("publish retained row");

        let db = runtime.db();
        let store = db.store();
        let mut header = store.physical_file_header().expect("physical header");
        assert_ne!(
            header.collection_roots_page, 0,
            "fixture retains native roots"
        );
        header.sequence = header
            .sequence
            .checked_add(1)
            .expect("next catalog sequence");
        header.manifest_page = 0;
        header.manifest_checksum = 0;
        header.registry_page = 0;
        header.registry_checksum = 0;
        header.recovery_page = 0;
        header.recovery_checksum = 0;
        header.catalog_page = 0;
        header.catalog_checksum = 0;
        header.metadata_state_page = 0;
        header.metadata_state_checksum = 0;
        header.vector_artifact_page = 0;
        header.vector_artifact_checksum = 0;
        store
            .update_physical_file_header(header)
            .expect("publish legacy catalog sequence");

        let legacy = db.native_physical_state().expect("legacy native state");
        assert!(RedDB::native_state_is_summary_free_legacy(&legacy));
        let mut unreadable_publication = legacy.clone();
        unreadable_publication.header.registry_page = u32::MAX;
        assert!(
            !RedDB::native_state_is_summary_free_legacy(&unreadable_publication),
            "a non-zero unreadable summary pointer must still fail closed"
        );

        let binary = PhysicalMetadataFile::metadata_binary_path_for(&path);
        let json = PhysicalMetadataFile::metadata_path_for(&path);
        db.delete_kv("red.storage.metadata", PHYSICAL_METADATA_KEY)
            .expect("legacy fixture");
        if binary.exists() {
            std::fs::remove_file(binary).expect("remove binary sidecar");
        }
        if json.exists() {
            std::fs::remove_file(json).expect("remove JSON sidecar");
        }
        for journal in PhysicalMetadataFile::journal_paths_for_data_path(&path)
            .expect("list metadata journals")
        {
            std::fs::remove_file(journal).expect("remove metadata journal");
        }

        let metadata = db
            .load_or_bootstrap_physical_metadata(true)
            .expect("summary-free legacy state must bootstrap internal metadata");
        assert_eq!(
            metadata.superblock.collection_roots.get("retained"),
            db.physical_collection_roots().get("retained")
        );
        assert!(!PhysicalMetadataFile::metadata_binary_path_for(&path).exists());
        assert!(db.load_internal_physical_metadata().unwrap().is_some());
        let rows = runtime
            .execute_query("SELECT value FROM retained")
            .expect("retained row remains queryable");
        assert_eq!(rows.result.records.len(), 1);
    }

    #[test]
    fn legacy_paged_drop_does_not_resurrect_derived_index_descriptors() {
        let directory = tempfile::tempdir().expect("temporary database");
        let path = directory.path().join("drop-descriptors.rdb");
        let options = RedDBOptions::persistent(&path)
            .with_memory_budget(64 * 1024 * 1024)
            .with_storage_profile(crate::StorageDeployPreset::PrimaryReplicaDev.selection())
            .expect("paged profile");
        {
            let runtime = crate::RedDBRuntime::with_options(options).expect("paged runtime");
            runtime
                .execute_query("CREATE TABLE retained_source (value INTEGER)")
                .expect("retained table");
            runtime
                .execute_query("INSERT INTO retained_source (value) VALUES (17)")
                .expect("retained row");
            runtime.checkpoint().expect("initial checkpoint");
        }
        // Managed legacy files also reopen with the default embedded options.
        // That path edits contracts without rebuilding every native artifact.
        let options = RedDBOptions::persistent(&path).with_memory_budget(64 * 1024 * 1024);
        {
            let runtime =
                crate::RedDBRuntime::with_options(options.clone()).expect("legacy reopen");
            assert!(runtime.db().is_paged());
            assert_eq!(
                runtime.db().options().storage_profile.deploy_profile,
                crate::storage::DeployProfile::Embedded
            );
            // Exercise both distinct names and recreation after a removal proof.
            for cycle in 0..6 {
                let table = format!("discarded_source_{}", cycle % 4);
                runtime
                    .execute_query(&format!("CREATE TABLE {table} (value INTEGER)"))
                    .expect("temporary table");
                runtime.checkpoint().expect("publish derived descriptors");
                let before = runtime.db().physical_metadata().expect("physical metadata");
                assert!(before
                    .indexes
                    .iter()
                    .any(|index| index.collection.as_deref() == Some(table.as_str())));
                runtime
                    .execute_query(&format!("DROP TABLE {table}"))
                    .expect("drop temporary table");
                for _ in 0..2 {
                    runtime.checkpoint().expect("checkpoint after drop");
                    let metadata = runtime.db().physical_metadata().expect("physical metadata");
                    assert!(
                        !metadata.indexes.iter().any(|index| index
                            .collection
                            .as_deref()
                            .is_some_and(|name| name.starts_with("discarded_source_"))),
                        "a dropped collection's derived descriptors were resurrected"
                    );
                    let native = runtime
                        .db()
                        .native_registry_summary()
                        .expect("native registry");
                    assert!(!native.indexes.iter().any(|index| {
                        index
                            .collection
                            .as_deref()
                            .is_some_and(|name| name.starts_with("discarded_source_"))
                    }));
                }
                if cycle == 0 {
                    // Model the consistent sidecar/native pair left by an older
                    // writer. Its final Remove event is a durable removal proof.
                    let ghost = before
                        .indexes
                        .iter()
                        .find(|index| index.collection.as_deref() == Some(table.as_str()))
                        .expect("old derived descriptor")
                        .clone();
                    runtime
                        .db()
                        .update_physical_metadata(|metadata| {
                            assert!(metadata.manifest_events.iter().any(|event| {
                                event.collection == table
                                    && matches!(
                                        &event.kind,
                                        crate::physical::ManifestEventKind::Remove
                                    )
                            }));
                            metadata.indexes.push(ghost.clone());
                            let mut orphan = ghost.clone();
                            orphan.name = "audited_legacy_orphan".into();
                            orphan.collection = Some("audited_missing_source".into());
                            orphan.artifact_root_page = None;
                            orphan.artifact_checksum = None;
                            metadata.indexes.push(orphan.clone());
                            orphan.name = "global_descriptor".into();
                            orphan.collection = None;
                            metadata.indexes.push(orphan);
                        })
                        .expect("simulate old metadata without deleting any rows");
                    runtime
                        .checkpoint()
                        .expect("repair a confirmed old removal");
                }
                let metadata = runtime.db().physical_metadata().expect("physical metadata");
                assert!(metadata
                    .indexes
                    .iter()
                    .any(|index| index.name == "audited_legacy_orphan"));
                assert!(metadata
                    .indexes
                    .iter()
                    .any(|index| index.name == "global_descriptor"));
            }
        }
        let runtime = crate::RedDBRuntime::with_options(options).expect("second legacy reopen");
        assert_eq!(
            runtime
                .db()
                .store()
                .get_collection("retained_source")
                .expect("retained source")
                .count(),
            1
        );
        let metadata = runtime.db().physical_metadata().expect("physical metadata");
        assert!(!metadata.indexes.iter().any(|index| {
            index
                .collection
                .as_deref()
                .is_some_and(|name| name.starts_with("discarded_source_"))
        }));
        assert!(metadata
            .indexes
            .iter()
            .any(|index| index.collection.as_deref() == Some("retained_source")));
        assert!(metadata
            .indexes
            .iter()
            .any(|index| index.name == "audited_legacy_orphan"));
        assert!(metadata
            .indexes
            .iter()
            .any(|index| index.name == "global_descriptor"));
    }

    #[test]
    fn index_reconciliation_reads_one_native_observation() {
        let directory = tempfile::tempdir().expect("temporary database");
        let runtime = crate::RedDBRuntime::with_options(
            RedDBOptions::persistent(directory.path().join("catalog.rdb"))
                .with_storage_profile(crate::StorageDeployPreset::Serverless.selection())
                .expect("paged profile")
                .with_capability(Capability::FullText),
        )
        .expect("runtime");
        for number in 0..4 {
            runtime
                .execute_query(&format!("CREATE TABLE source_{number} (payload TEXT)"))
                .expect("table");
            runtime
                .execute_query(&format!(
                    "INSERT INTO source_{number} (payload) VALUES ('searchable payload')"
                ))
                .expect("row");
        }
        runtime.checkpoint().expect("published artifacts");
        let db = runtime.db();
        let pager = db.store.pager().expect("paged store");
        let reads = || {
            let stats = pager.cache_stats();
            stats.hits + stats.misses
        };
        let before = reads();
        let native = db.native_physical_state().expect("native state");
        let observation_reads = reads() - before;
        let artifacts = native.registry.expect("registry").vector_artifacts;
        let pages = native.vector_artifact_pages.expect("artifact pages");
        let indexes: Vec<_> = artifacts
            .iter()
            .filter(|artifact| artifact.artifact_kind == "text.fulltext")
            .map(|artifact| PhysicalIndexState {
                name: format!("{}::fulltext", artifact.collection),
                kind: IndexKind::FullText,
                collection: Some(artifact.collection.clone()),
                enabled: true,
                entries: 0,
                estimated_memory_bytes: 0,
                last_refresh_ms: None,
                backend: "catalog-derived".into(),
                artifact_kind: None,
                artifact_root_page: None,
                artifact_checksum: None,
                build_state: "catalog-derived".into(),
            })
            .collect();
        assert!(indexes.len() >= 4, "fixture must publish all four indexes");
        let declarations = indexes.clone();
        let before = reads();
        let indexes = db.reconcile_index_states_with_native_artifacts(indexes);
        let reconciliation_reads = reads() - before;
        for index in indexes {
            let artifact = artifacts
                .iter()
                .find(|artifact| {
                    Some(&artifact.collection) == index.collection.as_ref()
                        && artifact.artifact_kind == "text.fulltext"
                })
                .expect("published artifact");
            let page = pages
                .iter()
                .find(|page| {
                    page.collection == artifact.collection
                        && page.artifact_kind == artifact.artifact_kind
                })
                .expect("artifact root");
            assert_eq!(index.artifact_root_page, Some(page.root_page));
            assert_eq!(index.artifact_checksum, Some(artifact.checksum));
            assert_eq!(index.entries, artifact.vector_count as usize);
            assert_eq!(index.build_state, "artifact-published");
        }
        assert_eq!(
            reconciliation_reads, observation_reads,
            "catalog reconciliation must not reread artifact pages per index"
        );

        // An undecodable artifact directory must retain the existing fallback,
        // not reuse the successful observation from the preceding call.
        let page_id = native.header.vector_artifact_page;
        let mut page = pager.read_page(page_id).expect("artifact directory");
        page.as_bytes_mut()[crate::storage::engine::HEADER_SIZE..].fill(0);
        pager
            .write_page(page_id, page)
            .expect("corrupt artifact directory");
        assert!(db.native_vector_artifact_pages().is_none());
        let indexes = db.reconcile_index_states_with_native_artifacts(declarations);
        for index in indexes {
            assert_eq!(index.artifact_root_page, None);
            assert!(
                index.artifact_checksum.is_some(),
                "registry remains available"
            );
            assert_eq!(index.build_state, "artifact-published");
        }
    }
}