reddb_server/physical/

metadata_file.rs

use super::*;
use crate::api::REDDB_FORMAT_VERSION;

impl PhysicalMetadataFile {
    pub fn from_state(
        options: RedDBOptions,
        catalog: CatalogSnapshot,
        collection_roots: BTreeMap<String, u64>,
        indexes: Vec<PhysicalIndexState>,
        previous: Option<&PhysicalMetadataFile>,
    ) -> Self {
        let now = unix_ms_now();
        let mut manifest = SchemaManifest::now(options, catalog.total_collections);
        if let Some(previous) = previous {
            manifest.created_at_unix_ms = previous.manifest.created_at_unix_ms;
        }
        manifest.updated_at_unix_ms = now;

        let sequence = previous
            .map(|previous| previous.superblock.sequence.saturating_add(1))
            .unwrap_or(1);
        let mut manifest_events = previous
            .map(|previous| previous.manifest_events.clone())
            .unwrap_or_default();
        manifest_events.extend(build_manifest_events(
            previous.map(|previous| &previous.superblock.collection_roots),
            &collection_roots,
            sequence,
        ));
        trim_manifest_history(&mut manifest_events);

        let superblock = SuperblockHeader {
            format_version: REDDB_FORMAT_VERSION,
            sequence,
            copies: DEFAULT_SUPERBLOCK_COPIES,
            manifest: previous
                .map(|previous| previous.superblock.manifest.clone())
                .unwrap_or_default(),
            free_set: previous
                .map(|previous| previous.superblock.free_set)
                .unwrap_or_default(),
            collection_roots,
        };

        let mut snapshots = previous
            .map(|previous| previous.snapshots.clone())
            .unwrap_or_default();
        snapshots.push(SnapshotDescriptor {
            snapshot_id: sequence,
            created_at_unix_ms: now,
            superblock_sequence: sequence,
            collection_count: catalog.total_collections,
            total_entities: catalog.total_entities,
        });
        trim_snapshot_history(&mut snapshots, manifest.options.snapshot_retention);

        Self {
            protocol_version: PHYSICAL_METADATA_PROTOCOL_VERSION.to_string(),
            generated_at_unix_ms: now,
            last_loaded_from: previous.and_then(|previous| previous.last_loaded_from.clone()),
            last_healed_at_unix_ms: previous.and_then(|previous| previous.last_healed_at_unix_ms),
            manifest,
            catalog,
            manifest_events,
            indexes,
            graph_projections: previous
                .map(|previous| previous.graph_projections.clone())
                .unwrap_or_default(),
            analytics_jobs: previous
                .map(|previous| previous.analytics_jobs.clone())
                .unwrap_or_default(),
            tree_definitions: previous
                .map(|previous| previous.tree_definitions.clone())
                .unwrap_or_default(),
            collection_ttl_defaults_ms: previous
                .map(|previous| previous.collection_ttl_defaults_ms.clone())
                .unwrap_or_default(),
            collection_contracts: previous
                .map(|previous| previous.collection_contracts.clone())
                .unwrap_or_default(),
            hypertables: previous
                .map(|previous| previous.hypertables.clone())
                .unwrap_or_default(),
            exports: previous
                .map(|previous| previous.exports.clone())
                .unwrap_or_default(),
            superblock,
            snapshots,
        }
    }

    pub fn metadata_path_for(data_path: &Path) -> PathBuf {
        reddb_file::layout::physical_metadata_json_path(data_path)
    }

    pub fn metadata_binary_path_for(data_path: &Path) -> PathBuf {
        reddb_file::layout::physical_metadata_binary_path(data_path)
    }

    pub fn metadata_journal_path_for(data_path: &Path, sequence: u64) -> PathBuf {
        reddb_file::layout::physical_metadata_journal_path(data_path, sequence)
    }

    pub fn export_data_path_for(data_path: &Path, name: &str) -> PathBuf {
        reddb_file::layout::physical_export_data_path(data_path, name)
    }

    pub fn load_for_data_path(data_path: &Path) -> io::Result<Self> {
        Self::load_for_data_path_with_source(data_path).map(|(metadata, _)| metadata)
    }

    pub fn load_for_data_path_with_source(
        data_path: &Path,
    ) -> io::Result<(Self, PhysicalMetadataSource)> {
        Self::load_for_data_path_with_repair(data_path, true)
    }

    /// Inspect available metadata without repairing or publishing any file.
    pub fn inspect_for_data_path(data_path: &Path) -> io::Result<Self> {
        Self::inspect_for_data_path_with_source(data_path).map(|(metadata, _)| metadata)
    }

    pub fn inspect_for_data_path_with_source(
        data_path: &Path,
    ) -> io::Result<(Self, PhysicalMetadataSource)> {
        Self::load_for_data_path_with_repair(data_path, false)
    }

    fn load_for_data_path_with_repair(
        data_path: &Path,
        repair_primary: bool,
    ) -> io::Result<(Self, PhysicalMetadataSource)> {
        let binary_path = Self::metadata_binary_path_for(data_path);
        if binary_path.exists() {
            match Self::load_from_binary_path(&binary_path) {
                Ok(metadata) => {
                    return Ok((metadata, PhysicalMetadataSource::Binary));
                }
                Err(error) if is_collection_contract_decode_error(&error) => return Err(error),
                Err(_) => {
                    let mut journal_paths = Self::journal_paths_for_data_path(data_path)?;
                    journal_paths.reverse();
                    for journal_path in journal_paths {
                        match Self::load_from_binary_path(&journal_path) {
                            Ok(metadata) => {
                                if !repair_primary {
                                    return Ok((metadata, PhysicalMetadataSource::BinaryJournal));
                                }
                                let healed =
                                    metadata.mark_recovery(PhysicalMetadataSource::BinaryJournal);
                                let _ = healed.heal_primary_metadata_for_data_path(data_path);
                                return Ok((healed, PhysicalMetadataSource::BinaryJournal));
                            }
                            Err(error) if is_collection_contract_decode_error(&error) => {
                                return Err(error)
                            }
                            Err(_) => {}
                        }
                    }
                }
            }
        }
        Self::load_from_path(&Self::metadata_path_for(data_path)).map(|metadata| {
            if !repair_primary {
                return (metadata, PhysicalMetadataSource::Json);
            }
            let healed = metadata.mark_recovery(PhysicalMetadataSource::Json);
            let _ = healed.heal_primary_metadata_for_data_path(data_path);
            (healed, PhysicalMetadataSource::Json)
        })
    }

    pub fn save_for_data_path(&self, data_path: &Path) -> io::Result<PathBuf> {
        let binary_path = Self::metadata_binary_path_for(data_path);
        if binary_path.exists() && super::seqn_journal_enabled() {
            let sequence = Self::load_from_binary_path(&binary_path)
                .map(|metadata| metadata.superblock.sequence)
                .unwrap_or(self.superblock.sequence);
            let _ = reddb_file::copy_physical_metadata_binary_to_journal(
                data_path,
                &binary_path,
                sequence,
            );
        }
        self.save_to_binary_path(&binary_path)?;
        self.prune_journal_for_data_path(data_path)?;
        if super::meta_json_sidecar_enabled() {
            let json_path = Self::metadata_path_for(data_path);
            self.save_to_path(&json_path)?;
        }
        Ok(binary_path)
    }

    pub fn load_from_path(path: &Path) -> io::Result<Self> {
        let text = reddb_file::read_physical_metadata_document(path).map_err(|err| {
            io::Error::new(
                io::ErrorKind::InvalidData,
                format!("invalid physical metadata JSON: {err}"),
            )
        })?;
        Self::from_document_json(&text, "invalid physical metadata JSON")
    }

    pub fn save_to_path(&self, path: &Path) -> io::Result<()> {
        let text = self.to_document_json(true)?;
        reddb_file::write_physical_metadata_json_document(path, &text)
            .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err.to_string()))
    }

    pub fn load_from_binary_path(path: &Path) -> io::Result<Self> {
        let text = reddb_file::read_physical_metadata_document(path).map_err(|err| {
            io::Error::new(
                io::ErrorKind::InvalidData,
                format!("invalid physical metadata binary: {err}"),
            )
        })?;
        Self::from_document_json(&text, "invalid physical metadata binary")
    }

    pub fn save_to_binary_path(&self, path: &Path) -> io::Result<()> {
        let text = self.to_document_json(false)?;
        reddb_file::write_physical_metadata_binary_document(path, &text)
            .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err.to_string()))
    }

    pub fn journal_paths_for_data_path(data_path: &Path) -> io::Result<Vec<PathBuf>> {
        reddb_file::list_physical_metadata_journal_paths(data_path)
            .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err.to_string()))
    }

    pub fn prune_journal_for_data_path(&self, data_path: &Path) -> io::Result<()> {
        let retention = super::seqn_journal_retention();
        reddb_file::prune_physical_metadata_journal_paths(data_path, retention)
            .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err.to_string()))
    }

    pub fn heal_primary_metadata_for_data_path(&self, data_path: &Path) -> io::Result<()> {
        let binary_path = Self::metadata_binary_path_for(data_path);
        self.save_to_binary_path(&binary_path)?;
        if super::meta_json_sidecar_enabled() {
            let json_path = Self::metadata_path_for(data_path);
            self.save_to_path(&json_path)?;
        }
        Ok(())
    }

    pub fn mark_recovery(&self, source: PhysicalMetadataSource) -> Self {
        let mut metadata = self.clone();
        metadata.last_loaded_from = Some(source.as_str().to_string());
        metadata.last_healed_at_unix_ms = Some(unix_ms_now());
        metadata
    }

    pub fn to_json_value(&self) -> JsonValue {
        let json = self
            .to_document_json(false)
            .expect("physical metadata must encode as JSON");
        crate::json::from_str::<JsonValue>(&json)
            .expect("reddb-file emitted JSON the server can parse")
    }

    pub fn from_json_value(value: &JsonValue) -> io::Result<Self> {
        Self::from_document_json(&value.to_string_compact(), "invalid physical metadata JSON")
    }

    fn to_document_json(&self, pretty: bool) -> io::Result<String> {
        reddb_file::encode_physical_metadata_document_root_json(
            &self.to_document_envelope(),
            pretty,
        )
        .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err.to_string()))
    }

    fn from_document_json(text: &str, context: &'static str) -> io::Result<Self> {
        let envelope =
            reddb_file::decode_physical_metadata_document_root_json(text).map_err(|err| {
                io::Error::new(io::ErrorKind::InvalidData, format!("{context}: {err}"))
            })?;
        Self::from_document_envelope(envelope)
    }

    fn to_document_envelope(&self) -> reddb_file::PhysicalMetadataDocumentEnvelope {
        reddb_file::PhysicalMetadataDocumentEnvelope {
            protocol_version: self.protocol_version.clone(),
            generated_at_unix_ms: self.generated_at_unix_ms,
            last_loaded_from: self.last_loaded_from.clone(),
            last_healed_at_unix_ms: self.last_healed_at_unix_ms,
            manifest_json: manifest_to_json(&self.manifest).to_string_compact(),
            catalog_json: catalog_to_json(&self.catalog).to_string_compact(),
            manifest_events_json: self
                .manifest_events
                .iter()
                .map(|event| manifest_event_to_json(event).to_string_compact())
                .collect(),
            indexes_json: self
                .indexes
                .iter()
                .map(|index| index_state_to_json(index).to_string_compact())
                .collect(),
            graph_projections_json: self
                .graph_projections
                .iter()
                .map(|projection| graph_projection_to_json(projection).to_string_compact())
                .collect(),
            analytics_jobs_json: self
                .analytics_jobs
                .iter()
                .map(|job| analytics_job_to_json(job).to_string_compact())
                .collect(),
            tree_definitions_json: self
                .tree_definitions
                .iter()
                .map(|definition| tree_definition_to_json(definition).to_string_compact())
                .collect(),
            collection_ttl_defaults_ms: self.collection_ttl_defaults_ms.clone(),
            collection_contracts_json: self
                .collection_contracts
                .iter()
                .map(|contract| collection_contract_to_json(contract).to_string_compact())
                .collect(),
            hypertables_json: self
                .hypertables
                .iter()
                .map(|hypertable| hypertable_to_json(hypertable).to_string_compact())
                .collect(),
            exports_json: self
                .exports
                .iter()
                .map(|export| export_descriptor_to_json(export).to_string_compact())
                .collect(),
            superblock_json: superblock_to_json(&self.superblock).to_string_compact(),
            snapshots_json: self
                .snapshots
                .iter()
                .map(|snapshot| snapshot_descriptor_to_json(snapshot).to_string_compact())
                .collect(),
        }
    }

    fn from_document_envelope(
        envelope: reddb_file::PhysicalMetadataDocumentEnvelope,
    ) -> io::Result<Self> {
        Ok(Self {
            protocol_version: envelope.protocol_version,
            generated_at_unix_ms: envelope.generated_at_unix_ms,
            last_loaded_from: envelope.last_loaded_from,
            last_healed_at_unix_ms: envelope.last_healed_at_unix_ms,
            manifest: manifest_from_json(&parse_document_fragment(
                &envelope.manifest_json,
                "manifest",
            )?)?,
            catalog: catalog_from_json(&parse_document_fragment(
                &envelope.catalog_json,
                "catalog",
            )?)?,
            manifest_events: parse_document_fragments(
                &envelope.manifest_events_json,
                "manifest_events",
                manifest_event_from_json,
            )?,
            indexes: parse_document_fragments(
                &envelope.indexes_json,
                "indexes",
                index_state_from_json,
            )?,
            graph_projections: parse_document_fragments(
                &envelope.graph_projections_json,
                "graph_projections",
                graph_projection_from_json,
            )?,
            analytics_jobs: parse_document_fragments(
                &envelope.analytics_jobs_json,
                "analytics_jobs",
                analytics_job_from_json,
            )?,
            tree_definitions: parse_document_fragments(
                &envelope.tree_definitions_json,
                "tree_definitions",
                tree_definition_from_json,
            )?,
            collection_ttl_defaults_ms: envelope.collection_ttl_defaults_ms,
            collection_contracts: parse_document_fragments(
                &envelope.collection_contracts_json,
                "collection_contracts",
                collection_contract_from_json,
            )?,
            hypertables: parse_document_fragments(
                &envelope.hypertables_json,
                "hypertables",
                hypertable_from_json,
            )?,
            exports: parse_document_fragments(
                &envelope.exports_json,
                "exports",
                export_descriptor_from_json,
            )?,
            superblock: superblock_from_json(&parse_document_fragment(
                &envelope.superblock_json,
                "superblock",
            )?)?,
            snapshots: parse_document_fragments(
                &envelope.snapshots_json,
                "snapshots",
                snapshot_descriptor_from_json,
            )?,
        })
    }
}

fn parse_document_fragment(json: &str, field: &'static str) -> io::Result<JsonValue> {
    crate::json::from_str::<JsonValue>(json)
        .map_err(|err| invalid_data(format!("invalid physical metadata field '{field}': {err}")))
}

fn parse_document_fragments<T>(
    fragments: &[String],
    field: &'static str,
    decode: fn(&JsonValue) -> io::Result<T>,
) -> io::Result<Vec<T>> {
    fragments
        .iter()
        .map(|fragment| decode(&parse_document_fragment(fragment, field)?))
        .collect()
}