hawdb-storage 0.16.0

// Copyright 2026 Nowledge
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! Checkpoint encoding, staged sidecars and manifest selection.

use super::{
    load_published_canonical_adjacency, load_published_canonical_segments,
    load_published_property_projection, CheckpointImage, CheckpointManifestArtifacts,
    CheckpointReplayBoundary, DurableArtifactMetadata, DurableManifest, DurableStore,
    GraphManifestOpenBudget,
};
use crate::error::{HawDBError, Result};
use crate::file_io::{self as fs, File};
use crate::store::{
    canonical_adjacency_artifact_generation_file, canonical_artifact_generation_file,
    canonical_manifest_generation_file, checkpoint_generation_file, checkpoint_publish_failpoint,
    property_projection_artifact_generation_file, property_projection_manifest_generation_file,
    property_spill_artifact_generation_file, property_spill_manifest_generation_file,
    read_durable_text_bytes_with_limit, relational_checkpoint_generation_file,
    remove_source_scan_artifacts, safe_reclaim_commit_epoch, source_scan, sync_parent_dir,
    verify_integrity, wal_generation_file, CheckpointPublishStage, PROJECTED_GRAPHS_FILE,
};
use hawdb_storage::{
    cache::ManifestGeneration,
    config::{DurableCompression, WalReplayConfig},
    durability::durable_replace_file,
    projection::SearchProjectionGraphChange,
    relational::{
        encode_relational_checkpoint_with_work_context, CheckpointOutputIo, RelationalDecodeLimits,
        RelationalState,
    },
    scan::FileSegmentRangeReader,
};
use std::io::Read;
use std::path::{Path, PathBuf};
use std::sync::Arc;

// Same-project path operations use one transient slot. Windows replacement
// additionally holds its write-capable candidate during rename.
const SIDECAR_PUBLICATION_DESCRIPTORS: usize = if cfg!(windows) { 2 } else { 1 };

#[cfg(test)]
#[path = "checkpoint/tests.rs"]
mod tests;

impl DurableStore {
    pub(in crate::store) fn read_checkpoint_text(&self, config: WalReplayConfig) -> Result<String> {
        let metadata = fs::metadata(&self.checkpoint_path)?;
        if config
            .max_checkpoint_encoded_bytes
            .is_some_and(|limit| metadata.len() > limit)
        {
            return Err(HawDBError::Storage(format!(
                "checkpoint encoded byte limit exceeded: max_checkpoint_encoded_bytes={}",
                config.max_checkpoint_encoded_bytes.unwrap_or_default()
            )));
        }
        let bytes = fs::read(&self.checkpoint_path)?;
        let expected_len = self.checkpoint_encoded_len.ok_or_else(|| {
            HawDBError::Storage("checkpoint is missing its encoded length".to_string())
        })?;
        let expected_checksum = self.checkpoint_encoded_checksum.ok_or_else(|| {
            HawDBError::Storage("checkpoint is missing its encoded checksum".to_string())
        })?;
        let expected_sha256 = self.checkpoint_encoded_sha256.ok_or_else(|| {
            HawDBError::Storage("checkpoint is missing its encoded SHA-256".to_string())
        })?;
        verify_integrity(
            &bytes,
            expected_len,
            expected_checksum,
            expected_sha256,
            "checkpoint",
        )?;
        read_durable_text_bytes_with_limit(
            &bytes,
            "checkpoint",
            config.max_checkpoint_decoded_bytes,
        )
    }

    pub(in crate::store) fn write_relational_checkpoint(
        &self,
        state: &RelationalState,
        commit_epoch: u64,
        generation: u64,
        work: &crate::background::CheckpointWorkContext,
    ) -> Result<Option<DurableArtifactMetadata>> {
        let path = self
            .root_path
            .join(relational_checkpoint_generation_file(generation));
        // Canonical metadata-only state deliberately omits database-sized row
        // and overflow residency. Its durable authority is the bound row and
        // overflow roots written by the same checkpoint publication. Encoding
        // it as a legacy full-row checkpoint would either require whole-store
        // hydration or produce an incomplete artifact whose retained overflow
        // segments appear unreachable.
        if state.is_empty() || state.canonical_row_metadata_only() {
            let unit = work.start_unit().map_err(HawDBError::from_storage_error)?;
            let _wave = work.io_wave().map_err(HawDBError::from_storage_error)?;
            match fs::remove_file(&path) {
                Ok(()) => sync_parent_dir(&path)?,
                Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
                Err(error) => return Err(error.into()),
            }
            unit.finish();
            work.checkpoint().map_err(HawDBError::from_storage_error)?;
            return Ok(None);
        }
        let max_bytes = RelationalDecodeLimits::checkpoint().max_record_bytes;
        let tmp_path = path.with_extension("hawdb.tmp");
        let mut temporary = super::artifacts::CheckpointMetadataTemporaryPath(None);
        {
            let mut file = {
                let unit = work.start_unit().map_err(HawDBError::from_storage_error)?;
                let _wave = work.io_wave().map_err(HawDBError::from_storage_error)?;
                let file = File::options()
                    .write(true)
                    .create_new(true)
                    .open(&tmp_path)?;
                temporary.0 = Some(tmp_path.clone());
                unit.finish();
                file
            };
            encode_relational_checkpoint_with_work_context(
                &mut file,
                commit_epoch,
                state,
                max_bytes,
                CheckpointOutputIo::File,
                work,
            )
            .map_err(HawDBError::from_storage_error)?;
            let unit = work.start_unit().map_err(HawDBError::from_storage_error)?;
            let _wave = work.io_wave().map_err(HawDBError::from_storage_error)?;
            file.sync_all()?;
            unit.finish();
        }
        let metadata = checkpoint_file_integrity(&tmp_path, work)?;
        {
            let unit = work.start_unit().map_err(HawDBError::from_storage_error)?;
            let _wave = work.io_wave().map_err(HawDBError::from_storage_error)?;
            durable_replace_file(&tmp_path, &path)?;
            temporary.0 = None;
            unit.finish();
        }
        work.checkpoint().map_err(HawDBError::from_storage_error)?;
        Ok(Some(metadata))
    }

    pub(in crate::store) fn write_checkpoint<'a>(
        &self,
        image: CheckpointImage<'_>,
        generation: u64,
        changes: impl Iterator<Item = &'a SearchProjectionGraphChange> + Clone,
        work: &crate::background::CheckpointWorkContext,
    ) -> Result<DurableArtifactMetadata> {
        let mut data = hawdb_storage::checkpoint::encode_checkpoint_body_with_work_context(
            &image, generation, changes, work,
        )?;
        let checksum = work
            .checksum(data.as_bytes())
            .map_err(HawDBError::from_storage_error)?;
        let unit = work.start_unit().map_err(HawDBError::from_storage_error)?;
        data.push_str(&format!("checksum\t{checksum}\n"));
        unit.finish();
        let checkpoint_path = self.root_path.join(checkpoint_generation_file(generation));
        let encoded = crate::text::envelope::encode_durable_text_with_work_context(
            &data,
            DurableCompression::default(),
            work,
        )?;
        let digest = work
            .integrity(&encoded)
            .map_err(HawDBError::from_storage_error)?;
        let metadata = DurableArtifactMetadata {
            encoded_len: encoded.len() as u64,
            encoded_checksum: digest.crc32c.as_u64(),
            encoded_sha256: digest.sha256,
        };
        super::artifacts::publish_checkpoint_metadata(&checkpoint_path, &encoded, work)?;
        Ok(metadata)
    }

    pub(in crate::store) fn prepare_checkpoint_staging(&self, generation: u64) -> Result<PathBuf> {
        let staging_path = self
            .root_path
            .join(format!(".checkpoint.{generation}.prepare"));
        // Another retained preparation may own this generation. Atomic
        // creation must fail rather than delete its private recovery files.
        fs::create_dir(&staging_path)?;
        Ok(staging_path)
    }

    pub(in crate::store) fn publish_checkpoint_sidecars(
        &self,
        staging_path: &Path,
        publish_projected_graph_artifacts: bool,
        source_scan_publication: Option<source_scan::SourceScanPublication>,
    ) -> Result<()> {
        // Admit the whole IO wave before changing the first cache file. Reuse
        // these slots across replacements even if another owner fills all
        // remaining project capacity. Filesystem failures can still leave a
        // partial cache publication; the checkpoint manifest remains authority.
        let _publication_resources = self
            .project_files
            .reserve_admission(SIDECAR_PUBLICATION_DESCRIPTORS)
            .map_err(HawDBError::from_storage_error)?;
        if publish_projected_graph_artifacts {
            durable_replace_file(
                &staging_path.join(PROJECTED_GRAPHS_FILE),
                &self.projected_graphs_path,
            )?;
        } else {
            self.remove_projected_graph_artifacts()?;
        }
        #[cfg(test)]
        tests::after_first_publication();
        if source_scan_publication.is_some() {
            for file in [
                source_scan::SOURCE_SCAN_PAYLOAD_FILE,
                source_scan::SOURCE_SCAN_DESCRIPTOR_FILE,
            ] {
                durable_replace_file(&staging_path.join(file), &self.root_path.join(file))?;
            }
        } else {
            remove_source_scan_artifacts(&self.root_path)?;
        }
        match fs::remove_dir(staging_path) {
            Ok(()) => Ok(()),
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
            Err(error) => Err(error.into()),
        }
    }

    pub(in crate::store) fn discard_prepared_checkpoint(
        &self,
        generation: u64,
        staging_path: &Path,
    ) -> Result<()> {
        let _descriptor_operation = self.file_descriptor_context().internal_operation()?;
        for file in [
            checkpoint_generation_file(generation),
            relational_checkpoint_generation_file(generation),
            wal_generation_file(generation),
            canonical_artifact_generation_file(generation),
            canonical_manifest_generation_file(generation),
            hawdb_storage::canonical::canonical_segment_descriptor_page_file(generation),
            hawdb_storage::canonical::canonical_segment_descriptor_root_file(generation),
            canonical_adjacency_artifact_generation_file(generation),
            hawdb_storage::canonical_adjacency::canonical_adjacency_descriptor_page_file(
                generation,
            ),
            hawdb_storage::canonical_adjacency::canonical_adjacency_descriptor_root_file(
                generation,
            ),
            property_spill_artifact_generation_file(generation),
            property_spill_manifest_generation_file(generation),
            hawdb_storage::property_spill::property_spill_descriptor_page_file(generation),
            hawdb_storage::property_spill::property_spill_descriptor_root_file(generation),
            property_projection_artifact_generation_file(generation),
            property_projection_manifest_generation_file(generation),
            hawdb_storage::property_projection::property_projection_descriptor_page_file(
                generation,
            ),
            hawdb_storage::property_projection::property_projection_descriptor_root_file(
                generation,
            ),
            hawdb_storage::relational::relational_index_shadow_artifact_file(generation),
            hawdb_storage::relational::relational_index_shadow_manifest_generation_file(generation),
            hawdb_storage::relational::relational_row_page_artifact_file(generation),
            hawdb_storage::relational::relational_row_page_root_descriptor_file(generation),
            hawdb_storage::relational::relational_row_page_root_key_file(generation),
            hawdb_storage::relational::relational_row_page_manifest_generation_file(generation),
            hawdb_storage::relational::relational_overflow_extent_file(generation),
            hawdb_storage::relational::relational_overflow_descriptor_file(generation),
            hawdb_storage::relational::relational_overflow_manifest_generation_file(generation),
        ] {
            match fs::remove_file(self.root_path.join(file)) {
                Ok(()) => {}
                Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
                Err(error) => return Err(error.into()),
            }
        }
        match fs::remove_dir_all(staging_path) {
            Ok(()) => sync_parent_dir(&self.manifest_path),
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
            Err(error) => Err(error.into()),
        }
    }

    pub(in crate::store) fn checkpoint_manifest(
        &self,
        generation: u64,
        artifacts: CheckpointManifestArtifacts,
        checkpoint_commit_epoch: u64,
        oldest_reader_commit_epoch: Option<u64>,
        source_scan_publication: Option<source_scan::SourceScanPublication>,
        replay: CheckpointReplayBoundary,
    ) -> Result<DurableManifest> {
        if replay.next_lsn < replay.start_lsn
            || replay.commit_epoch < checkpoint_commit_epoch
            || replay.next_lsn - replay.start_lsn != replay.commit_epoch - checkpoint_commit_epoch
        {
            return Err(HawDBError::StorageIntegrity(
                "checkpoint replay LSN interval does not cover its commit epochs".into(),
            ));
        }
        self.admit_graph_manifest_artifacts(&artifacts)?;
        let safe_reclaim_commit_epoch =
            safe_reclaim_commit_epoch(checkpoint_commit_epoch, oldest_reader_commit_epoch);
        let source_scan_commit_epoch = source_scan_publication.map(|value| value.graph_epoch());
        let source_scan_descriptor_checksum =
            source_scan_publication.map(|value| value.descriptor_checksum());
        let CheckpointManifestArtifacts {
            checkpoint,
            relational_checkpoint: _,
            canonical_manifest,
            canonical_adjacency,
            property_spill_manifest,
            property_projection_manifest,
            relational_row,
            relational_overflow,
            relational_index,
            append,
        } = artifacts;
        let manifest = DurableManifest {
            checkpoint_generation: Some(generation),
            checkpoint_encoded_len: Some(checkpoint.encoded_len),
            checkpoint_encoded_checksum: Some(checkpoint.encoded_checksum),
            checkpoint_encoded_sha256: Some(checkpoint.encoded_sha256),
            canonical_manifest_encoded_len: Some(canonical_manifest.encoded_len),
            canonical_manifest_encoded_checksum: Some(canonical_manifest.encoded_checksum),
            canonical_manifest_encoded_sha256: Some(canonical_manifest.encoded_sha256),
            canonical_adjacency_generation_artifacts: Some(canonical_adjacency.generation),
            property_spill_manifest_encoded_len: Some(property_spill_manifest.encoded_len),
            property_spill_manifest_encoded_checksum: Some(
                property_spill_manifest.encoded_checksum,
            ),
            property_spill_manifest_encoded_sha256: Some(property_spill_manifest.encoded_sha256),
            property_projection_manifest_encoded_len: Some(
                property_projection_manifest.encoded_len,
            ),
            property_projection_manifest_encoded_checksum: Some(
                property_projection_manifest.encoded_checksum,
            ),
            property_projection_manifest_encoded_sha256: Some(
                property_projection_manifest.encoded_sha256,
            ),
            relational_row_generation_artifacts: Some(relational_row),
            relational_overflow_generation_artifacts: Some(relational_overflow),
            relational_index_generation_artifacts: relational_index,
            append_generation_artifacts: Some(append),
            wal_generation: generation,
            checkpoint_epoch: generation,
            checkpoint_commit_epoch,
            oldest_reader_commit_epoch,
            safe_reclaim_commit_epoch,
            wal_replay_start_lsn: replay.start_lsn,
            next_lsn: replay.next_lsn,
            source_scan_commit_epoch,
            source_scan_descriptor_checksum,
        };
        manifest.validate()?;
        Ok(manifest)
    }

    pub(in crate::store) fn publish_checkpoint_manifest(
        &mut self,
        generation: u64,
        artifacts: CheckpointManifestArtifacts,
        checkpoint_commit_epoch: u64,
        oldest_reader_commit_epoch: Option<u64>,
        source_scan_publication: Option<source_scan::SourceScanPublication>,
    ) -> Result<()> {
        let manifest = self.checkpoint_manifest(
            generation,
            artifacts,
            checkpoint_commit_epoch,
            oldest_reader_commit_epoch,
            source_scan_publication,
            CheckpointReplayBoundary {
                start_lsn: self.next_lsn,
                next_lsn: self.next_lsn,
                commit_epoch: checkpoint_commit_epoch,
            },
        )?;
        let publication = manifest
            .write(&self.manifest_path)
            .and_then(|()| checkpoint_publish_failpoint(CheckpointPublishStage::ManifestPublished));
        if let Err(error) = publication {
            // The disposable manifest may already select the candidate while
            // the authoritative branch head still selects the old closure.
            // Retain both and require head-based recovery before more writes.
            return Err(if self.branch_runtime.is_some() {
                HawDBError::StorageIntegrity(format!(
                    "branch checkpoint publication failed; close and reopen the branch: {error}"
                ))
            } else {
                error
            });
        }

        self.adopt_checkpoint_manifest(
            manifest,
            artifacts.relational_checkpoint,
            checkpoint_commit_epoch,
        )
    }

    /// Mounts validated immutable bindings without writing a durable selector.
    /// A private candidate can use this before replaying its captured suffix.
    pub(in crate::store) fn adopt_checkpoint_manifest(
        &mut self,
        manifest: DurableManifest,
        relational_checkpoint: Option<DurableArtifactMetadata>,
        wal_commit_epoch: u64,
    ) -> Result<()> {
        let generation = manifest.checkpoint_epoch;
        self.wal_append_file = None;
        self.wal_append_owner = None;
        self.checkpoint_path = manifest.checkpoint_path(&self.root_path);
        self.wal_path = manifest.wal_path(&self.root_path);
        self.checkpoint_encoded_len = manifest.checkpoint_encoded_len;
        self.checkpoint_encoded_checksum = manifest.checkpoint_encoded_checksum;
        self.checkpoint_encoded_sha256 = manifest.checkpoint_encoded_sha256;
        self.relational_checkpoint_encoded_len =
            relational_checkpoint.map(|artifact| artifact.encoded_len);
        self.relational_checkpoint_encoded_checksum =
            relational_checkpoint.map(|artifact| artifact.encoded_checksum);
        self.relational_checkpoint_encoded_sha256 =
            relational_checkpoint.map(|artifact| artifact.encoded_sha256);
        self.canonical_manifest_encoded_len = manifest.canonical_manifest_encoded_len;
        self.canonical_manifest_encoded_checksum = manifest.canonical_manifest_encoded_checksum;
        self.canonical_manifest_encoded_sha256 = manifest.canonical_manifest_encoded_sha256;
        self.canonical_adjacency_generation_artifacts =
            manifest.canonical_adjacency_generation_artifacts;
        self.property_spill_manifest_encoded_len = manifest.property_spill_manifest_encoded_len;
        self.property_spill_manifest_encoded_checksum =
            manifest.property_spill_manifest_encoded_checksum;
        self.property_spill_manifest_encoded_sha256 =
            manifest.property_spill_manifest_encoded_sha256;
        self.property_projection_manifest_encoded_len =
            manifest.property_projection_manifest_encoded_len;
        self.property_projection_manifest_encoded_checksum =
            manifest.property_projection_manifest_encoded_checksum;
        self.property_projection_manifest_encoded_sha256 =
            manifest.property_projection_manifest_encoded_sha256;
        self.relational_row_generation_artifacts = manifest.relational_row_generation_artifacts;
        self.relational_overflow_generation_artifacts =
            manifest.relational_overflow_generation_artifacts;
        self.relational_index_generation_artifacts = manifest.relational_index_generation_artifacts;
        self.append_generation_artifacts = manifest.append_generation_artifacts;
        self.wal_generation = manifest.wal_generation;
        self.checkpoint_epoch = manifest.checkpoint_epoch;
        self.checkpoint_commit_epoch = manifest.checkpoint_commit_epoch;
        self.known_durable_commit_epoch = manifest.checkpoint_commit_epoch;
        self.wal_namespace_known_durable = true;
        self.oldest_reader_commit_epoch = manifest.oldest_reader_commit_epoch;
        self.safe_reclaim_commit_epoch = manifest.safe_reclaim_commit_epoch;
        self.wal_replay_start_lsn = manifest.wal_replay_start_lsn;
        self.next_lsn = manifest.next_lsn;
        self.wal_bytes = fs::metadata(&self.wal_path)?.len();
        if manifest.next_lsn == manifest.wal_replay_start_lsn {
            self.wal_uncheckpointed_since = None;
        }
        self.wal_commit_epoch = wal_commit_epoch;
        self.wal_free_space_probe.last_available_bytes = None;
        self.wal_free_space_probe.wal_bytes_since_probe = 0;
        self.source_scan_commit_epoch = manifest.source_scan_commit_epoch;
        self.source_scan_descriptor_checksum = manifest.source_scan_descriptor_checksum;
        let mut graph_manifest_budget =
            GraphManifestOpenBudget::new(self.max_graph_manifest_open_bytes);
        self.canonical_segments = load_published_canonical_segments(
            &self.root_path,
            manifest,
            Arc::clone(&self.segment_cache),
            self.store_id,
            &mut graph_manifest_budget,
        )?;
        self.canonical_adjacency = load_published_canonical_adjacency(
            &self.root_path,
            manifest,
            Arc::clone(&self.segment_cache),
            self.store_id,
            &mut graph_manifest_budget,
        )?;
        self.persistent_property_projection = load_published_property_projection(
            &self.root_path,
            manifest,
            Arc::clone(&self.segment_cache),
            self.store_id,
            &mut graph_manifest_budget,
        )?;
        if let (Some(canonical), Some(adjacency)) =
            (&self.canonical_segments, &self.canonical_adjacency)
            && canonical.manifest().relationship_count != adjacency.relationship_count()
        {
            return Err(HawDBError::Storage(
                "canonical adjacency relationship count does not match canonical segments"
                    .to_string(),
            ));
        }
        self.source_scan_reader = FileSegmentRangeReader::new().with_cache(
            Arc::clone(&self.segment_cache),
            self.store_id,
            ManifestGeneration(generation),
        );
        self.source_scan_reader.register(
            source_scan::SOURCE_SCAN_ARTIFACT_ID,
            self.root_path.join(source_scan::SOURCE_SCAN_PAYLOAD_FILE),
        );
        Ok(())
    }
}

fn checkpoint_file_integrity(
    path: &Path,
    work: &crate::background::CheckpointWorkContext,
) -> Result<DurableArtifactMetadata> {
    let mut file = {
        let unit = work.start_unit().map_err(HawDBError::from_storage_error)?;
        let _wave = work.io_wave().map_err(HawDBError::from_storage_error)?;
        let file = File::open(path)?;
        unit.finish();
        file
    };
    let mut hasher = hawdb_integrity::IntegrityHasher::new();
    let mut encoded_len = 0u64;
    let mut block = {
        let unit = work.start_unit().map_err(HawDBError::from_storage_error)?;
        let block = vec![0; 64 * 1024];
        unit.finish();
        block
    };
    loop {
        let unit = work.start_unit().map_err(HawDBError::from_storage_error)?;
        let _wave = work.io_wave().map_err(HawDBError::from_storage_error)?;
        let read = file.read(&mut block)?;
        if read == 0 {
            unit.finish();
            break;
        }
        encoded_len = encoded_len
            .checked_add(read as u64)
            .ok_or_else(|| HawDBError::Storage("checkpoint file length overflows u64".into()))?;
        hasher.update(&block[..read]);
        unit.finish();
    }
    let unit = work.start_unit().map_err(HawDBError::from_storage_error)?;
    let digest = hasher.finish();
    unit.finish();
    work.checkpoint().map_err(HawDBError::from_storage_error)?;
    Ok(DurableArtifactMetadata {
        encoded_len,
        encoded_checksum: digest.crc32c.as_u64(),
        encoded_sha256: digest.sha256,
    })
}