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.

//! Durable store ownership, open admission and shared checkpoint state.
//!
//! Private child modules own artifact I/O, backup, scrub, WAL, checkpoint
//! publication, manifest validation and physical-generation reclamation.

#[path = "durable/artifacts.rs"]
pub(in crate::store) mod artifacts;
#[path = "durable/backup.rs"]
mod backup;
#[path = "durable/checkpoint.rs"]
mod checkpoint;
#[path = "durable/checkpoint_tail.rs"]
mod checkpoint_tail;
pub use checkpoint_tail::CheckpointWalTail;
#[path = "durable/manifest.rs"]
mod manifest;
#[path = "durable/reclamation.rs"]
mod reclamation;
#[path = "durable/scrub.rs"]
mod scrub;
#[path = "durable/wal.rs"]
mod wal;

use artifacts::{admit_graph_manifest_binding, load_published_canonical_segments};
pub(super) use artifacts::{
    load_published_canonical_adjacency, load_published_property_projection,
};
pub(super) use manifest::DurableManifest;
pub(super) use wal::WalAppendFile;

use super::{
    cleanup_abandoned_checkpoint_preparations, derived_repair, doctor, has_storage_artifacts,
    source_scan, store_id_for_path, CheckpointStatisticsState, CANONICAL_MANIFEST_MAX_BYTES,
    MANIFEST_FILE, PROJECTED_GRAPHS_FILE, PROPERTY_PROJECTION_MANIFEST_MAX_BYTES,
    PROPERTY_SPILL_MANIFEST_MAX_BYTES, STABLE_ID_MAPPING_FILE, WAL_BINARY_FILE_HEADER_BYTES,
};
use crate::error::{HawDBError, Result};
use crate::file_io as fs;
use hawdb_integrity::Sha256Digest;
use hawdb_storage::{
    append_table::{AppendGenerationArtifacts, AppendGenerationReader},
    cache::{ManifestGeneration, SegmentCache, StoreId},
    canonical::CanonicalSegmentReader,
    canonical_adjacency::{CanonicalAdjacencyGenerationArtifacts, CanonicalAdjacencyReader},
    config::{DurabilityPolicy, WalReplayConfig},
    durability::WalSyncGroupState,
    graph_descriptor_tree::GraphDescriptorTreeBuildConfig,
    ownership::DatabaseDirectoryLease,
    property_projection::PersistentPropertyProjectionReader,
    relational::{
        RelationalIndexGenerationArtifacts, RelationalOverflowGenerationArtifacts,
        RelationalRowPageGenerationArtifacts, RelationalState,
    },
    scan::FileSegmentRangeReader,
    stable_identity::{StableIdentityMappingError, StableIdentityMappingReader},
    telemetry::StorageTelemetrySink,
};
use std::path::{Path, PathBuf};
use std::sync::Arc;

const WAL_FREE_SPACE_PROBE_INTERVAL_BYTES: u64 = 64 * 1024 * 1024;

fn stable_identity_error(error: StableIdentityMappingError) -> HawDBError {
    HawDBError::from_storage_error(error)
}

#[derive(Debug, Clone)]
pub(super) struct DurableStore {
    project_files: crate::file_descriptors::ProjectFileDescriptors,
    _directory_lease: Arc<DatabaseDirectoryLease>,
    pub(super) branch_runtime: Option<super::immutable_root::BranchRuntimeBinding>,
    pub(super) allocation_project: Option<crate::branch_project::CapturedProject>,
    // UUID-leased branch admission may reconstruct derived row/index deltas
    // even for read-only queries. This never authorizes WAL or head writes.
    pub(super) recovery_artifact_writes_allowed: bool,
    pub(super) root_path: PathBuf,
    pub(super) checkpoint_path: PathBuf,
    manifest_path: PathBuf,
    projected_graphs_path: PathBuf,
    stable_id_mapping_path: PathBuf,
    pub(super) wal_path: PathBuf,
    pub(super) wal_append_file: Option<Arc<WalAppendFile>>,
    // Current captured generation only. Explicit checkpoint/rotation transitions
    // still use their legacy paths and clear this custody until native adoption.
    pub(super) wal_append_owner: Option<Arc<DatabaseDirectoryLease>>,
    #[cfg(test)]
    pub(super) wal_append_open_count: usize,
    pub(super) checkpoint_encoded_len: Option<u64>,
    checkpoint_encoded_checksum: Option<u64>,
    checkpoint_encoded_sha256: Option<Sha256Digest>,
    pub(super) relational_checkpoint_encoded_len: Option<u64>,
    pub(super) relational_checkpoint_encoded_checksum: Option<u64>,
    pub(super) relational_checkpoint_encoded_sha256: Option<Sha256Digest>,
    canonical_manifest_encoded_len: Option<u64>,
    canonical_manifest_encoded_checksum: Option<u64>,
    canonical_manifest_encoded_sha256: Option<Sha256Digest>,
    canonical_adjacency_generation_artifacts: Option<CanonicalAdjacencyGenerationArtifacts>,
    property_spill_manifest_encoded_len: Option<u64>,
    property_spill_manifest_encoded_checksum: Option<u64>,
    property_spill_manifest_encoded_sha256: Option<Sha256Digest>,
    property_projection_manifest_encoded_len: Option<u64>,
    property_projection_manifest_encoded_checksum: Option<u64>,
    property_projection_manifest_encoded_sha256: Option<Sha256Digest>,
    pub(super) relational_row_generation_artifacts: Option<RelationalRowPageGenerationArtifacts>,
    pub(super) relational_overflow_generation_artifacts:
        Option<RelationalOverflowGenerationArtifacts>,
    pub(super) relational_index_generation_artifacts: Option<RelationalIndexGenerationArtifacts>,
    pub(super) append_generation_artifacts: Option<AppendGenerationArtifacts>,
    pub(super) wal_generation: u64,
    pub(super) checkpoint_epoch: u64,
    pub(super) checkpoint_commit_epoch: u64,
    pub(super) oldest_reader_commit_epoch: Option<u64>,
    pub(super) safe_reclaim_commit_epoch: u64,
    pub(super) wal_replay_start_lsn: u64,
    pub(super) next_lsn: u64,
    pub(super) wal_bytes: u64,
    // Age of checkpoint debt, rather than time since the most recent write.
    // Reopening starts a fresh monotonic observation window; active writes
    // never move its beginning forward.
    pub(super) wal_uncheckpointed_since: Option<std::time::Instant>,
    pub(super) wal_tail_repair: Option<hawdb_storage::doctor::WalTailRepairReport>,
    /// Commit epoch recorded in binary WAL records (spec §3.4.3). Advisory:
    /// replay derives commit epochs from LSN order, exactly as before.
    pub(super) wal_commit_epoch: u64,
    pub(super) max_wal_bytes: Option<u64>,
    source_scan_commit_epoch: Option<u64>,
    source_scan_descriptor_checksum: Option<u64>,
    store_id: StoreId,
    pub(super) segment_cache: Arc<SegmentCache>,
    max_graph_manifest_open_bytes: u64,
    pub(super) canonical_segments: Option<CanonicalSegmentReader>,
    pub(super) canonical_adjacency: Option<CanonicalAdjacencyReader>,
    pub(super) persistent_property_projection: Option<PersistentPropertyProjectionReader>,
    stable_id_mapping_reader: Option<Arc<StableIdentityMappingReader>>,
    pub(super) source_scan_reader: FileSegmentRangeReader,
    durability: DurabilityPolicy,
    pub(super) read_only: bool,
    max_record_bytes: Option<usize>,
    max_batch_operations: Option<usize>,
    pub(super) telemetry: Option<Arc<dyn StorageTelemetrySink>>,
    wal_sync_group: Option<WalSyncGroupState>,
    // Process-local knowledge, initialized only from the durable checkpoint.
    // Reading/replaying a WAL or choosing a synchronous future write policy
    // does not prove that the recovered WAL prefix has completed a barrier.
    known_durable_commit_epoch: u64,
    wal_namespace_known_durable: bool,
    // The selected frontend and off-gate retirement source observe the same
    // maintenance result. A checkpoint-source clone must not hide a failed
    // reclamation from admission, pressure reports, or the next retry.
    generation_reclamation_debt: Arc<std::sync::Mutex<GenerationReclamationDebt>>,
    wal_free_space_probe: WalFreeSpaceProbeState,
}

#[derive(Debug, Clone, Default)]
struct WalFreeSpaceProbeState {
    last_available_bytes: Option<u64>,
    wal_bytes_since_probe: u64,
    #[cfg(test)]
    available_bytes_override: Option<u64>,
    #[cfg(test)]
    probe_count: u64,
}

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub(super) struct GenerationReclamationDebt {
    pub(super) retry_required: bool,
    pub(super) pending_file_count: usize,
    pub(super) pending_bytes: u64,
}

pub(super) use hawdb_storage::checkpoint::CheckpointImage;

#[derive(Debug)]
#[doc(hidden)]
pub struct PreparedCheckpoint {
    pub(super) source_commit_epoch: u64,
    pub(super) source_checkpoint_epoch: u64,
    pub(super) source_next_lsn: u64,
    pub(super) source_wal_generation: u64,
    pub(super) source_wal_bytes: u64,
    pub(super) generation: u64,
    pub(super) checkpoint_out_of_core: bool,
    pub(super) projected_graph_artifacts:
        Option<crate::projection::artifact::CheckpointProjectedGraphRoot>,
    pub(super) publish_projected_graph_artifacts: bool,
    pub(super) source_scan_publication: Option<source_scan::SourceScanPublication>,
    pub(super) checkpoint_statistics: CheckpointStatisticsState,
    pub(super) checkpoint_relational_state: Option<RelationalState>,
    pub(super) checkpoint_append_reader: AppendGenerationReader,
    pub(super) relational_index_candidate:
        Option<super::relational_index_shadow::PreparedRelationalIndexCandidate>,
    pub(super) relational_overflow_compaction_report:
        Option<super::RelationalOverflowCompactionReport>,
    pub(super) relational_row_compaction_report: Option<super::RelationalRowPageCompactionReport>,
    pub(super) manifest_artifacts: CheckpointManifestArtifacts,
    pub(super) staging_path: PathBuf,
}

pub(super) use hawdb_storage::derived_repair::DerivedArtifactBuildConfig;

#[derive(Debug, Clone, Copy)]
struct DurableStoreOpenOptions {
    read_only: bool,
    initialize_if_empty: bool,
    load_rebuildable_artifacts: bool,
    segment_cache_capacity_bytes: u64,
    max_graph_manifest_open_bytes: u64,
    max_wal_bytes: Option<u64>,
    max_record_bytes: Option<usize>,
    max_batch_operations: Option<usize>,
    automatic_tail_repair: Option<WalReplayConfig>,
}

pub(super) use hawdb_storage::artifact_binding::{
    DurableArtifactMetadata, GraphManifestOpenBudget,
};

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) struct CanonicalAdjacencyCheckpointArtifacts {
    pub(super) generation: CanonicalAdjacencyGenerationArtifacts,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) struct CheckpointManifestArtifacts {
    pub(super) checkpoint: DurableArtifactMetadata,
    pub(super) relational_checkpoint: Option<DurableArtifactMetadata>,
    pub(super) canonical_manifest: DurableArtifactMetadata,
    pub(super) canonical_adjacency: CanonicalAdjacencyCheckpointArtifacts,
    pub(super) property_spill_manifest: DurableArtifactMetadata,
    pub(super) property_projection_manifest: DurableArtifactMetadata,
    pub(super) relational_row: RelationalRowPageGenerationArtifacts,
    pub(super) relational_overflow: RelationalOverflowGenerationArtifacts,
    pub(super) relational_index: Option<RelationalIndexGenerationArtifacts>,
    pub(super) append: AppendGenerationArtifacts,
}

#[derive(Debug, Clone, Copy)]
pub(super) struct CheckpointReplayBoundary {
    pub(super) start_lsn: u64,
    pub(super) next_lsn: u64,
    pub(super) commit_epoch: u64,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum DurableOpenMode {
    CreateIfMissing,
    ExistingOnly,
}

impl DurableStore {
    pub(super) fn max_wal_record_bytes(&self) -> Option<usize> {
        self.max_record_bytes
    }

    pub(super) fn project_file_descriptors(
        &self,
    ) -> &crate::file_descriptors::ProjectFileDescriptors {
        &self.project_files
    }

    pub(super) fn reserve_project_bootstrap_identity(
        &self,
        proposed: crate::branch_project::ProjectSelector,
    ) -> Result<crate::branch_project::ProjectSelector> {
        if self.read_only || self.branch_runtime.is_some() {
            return Err(HawDBError::Storage(
                "project bootstrap requires the writable legacy directory lease".into(),
            ));
        }
        // This DurableStore owns the original directory lease. Keep it alive
        // while choosing identities for metadata adoption; no source copy or
        // second branch runtime is needed.
        crate::branch_project::reserve_bootstrap_identity_captured(
            &self.project_files,
            self._directory_lease.captured_directory(),
            proposed,
        )
    }

    pub(super) fn verify_consumer_allocation_domain(
        &self,
        selector: crate::branch_project::ProjectSelector,
    ) -> Result<()> {
        crate::consumer_allocation::bootstrap(
            self._directory_lease.captured_directory(),
            &self.project_files.io_context(),
            selector,
            false,
        )
        .map(|_| ())
    }

    pub(super) fn file_descriptor_context(&self) -> crate::file_descriptors::FileOpenContext {
        self.project_files.io_context()
    }

    pub(super) fn file_descriptor_metrics(&self) -> crate::file_descriptors::FileDescriptorMetrics {
        self.project_files.metrics()
    }
    pub(super) fn open(
        path: &Path,
        durability: DurabilityPolicy,
        replay_config: WalReplayConfig,
    ) -> Result<Self> {
        if replay_config.max_graph_manifest_open_bytes == 0 {
            return Err(HawDBError::Storage(
                "max_graph_manifest_open_bytes must be non-zero".to_string(),
            ));
        }
        fs::create_dir_all(path)?;
        Self::open_existing(
            path,
            durability,
            DurableStoreOpenOptions {
                read_only: false,
                initialize_if_empty: true,
                load_rebuildable_artifacts: true,
                segment_cache_capacity_bytes: replay_config.segment_cache_capacity_bytes,
                max_graph_manifest_open_bytes: replay_config.max_graph_manifest_open_bytes,
                max_wal_bytes: replay_config.max_bytes,
                max_record_bytes: replay_config.max_record_bytes,
                max_batch_operations: replay_config.max_batch_operations,
                automatic_tail_repair: (replay_config.recovery_mode
                    == hawdb_storage::config::RecoveryMode::AutoRepairTornTail)
                    .then_some(replay_config),
            },
        )
    }

    pub(super) fn open_existing_only(
        path: &Path,
        durability: DurabilityPolicy,
        segment_cache_capacity_bytes: u64,
        max_graph_manifest_open_bytes: u64,
        max_wal_bytes: Option<u64>,
        max_record_bytes: Option<usize>,
        max_batch_operations: Option<usize>,
    ) -> Result<Self> {
        Self::validate_read_only_path(path)?;
        Self::open_existing(
            path,
            durability,
            DurableStoreOpenOptions {
                read_only: true,
                initialize_if_empty: false,
                load_rebuildable_artifacts: true,
                automatic_tail_repair: None,
                segment_cache_capacity_bytes,
                max_graph_manifest_open_bytes,
                max_wal_bytes,
                max_record_bytes,
                max_batch_operations,
            },
        )
    }

    pub(super) fn validate_read_only_path(path: &Path) -> Result<()> {
        if !path.exists() {
            return Err(HawDBError::Storage(format!(
                "read-only database path does not exist: {}",
                path.display()
            )));
        }
        if !path.is_dir() {
            return Err(HawDBError::Storage(format!(
                "read-only database path is not a directory: {}",
                path.display()
            )));
        }
        Ok(())
    }

    pub(super) fn open_for_derived_repair(
        path: &Path,
        durability: DurabilityPolicy,
        segment_cache_capacity_bytes: u64,
        max_graph_manifest_open_bytes: u64,
        max_wal_bytes: Option<u64>,
        max_record_bytes: Option<usize>,
        max_batch_operations: Option<usize>,
    ) -> Result<Self> {
        if !path.is_dir() {
            return Err(HawDBError::Storage(format!(
                "derived repair database path is not a directory: {}",
                path.display()
            )));
        }
        Self::open_existing(
            path,
            durability,
            DurableStoreOpenOptions {
                read_only: true,
                initialize_if_empty: false,
                load_rebuildable_artifacts: false,
                automatic_tail_repair: None,
                segment_cache_capacity_bytes,
                max_graph_manifest_open_bytes,
                max_wal_bytes,
                max_record_bytes,
                max_batch_operations,
            },
        )
    }

    fn open_existing(
        path: &Path,
        durability: DurabilityPolicy,
        options: DurableStoreOpenOptions,
    ) -> Result<Self> {
        Self::open_existing_with_artifacts(path, path, durability, options)
    }

    pub(super) fn open_branch_runtime(
        runtime_path: &Path,
        artifact_path: &Path,
        durability: DurabilityPolicy,
        replay_config: WalReplayConfig,
        read_only: bool,
        load_rebuildable_artifacts: bool,
    ) -> Result<Self> {
        let mut store = Self::open_existing_with_artifacts(
            runtime_path,
            artifact_path,
            durability,
            DurableStoreOpenOptions {
                read_only,
                initialize_if_empty: false,
                load_rebuildable_artifacts,
                segment_cache_capacity_bytes: replay_config.segment_cache_capacity_bytes,
                max_graph_manifest_open_bytes: replay_config.max_graph_manifest_open_bytes,
                max_wal_bytes: replay_config.max_bytes,
                max_record_bytes: replay_config.max_record_bytes,
                max_batch_operations: replay_config.max_batch_operations,
                automatic_tail_repair: None,
            },
        )?;
        store.recovery_artifact_writes_allowed = true;
        Ok(store)
    }

    fn open_existing_with_artifacts(
        path: &Path,
        artifact_path: &Path,
        durability: DurabilityPolicy,
        options: DurableStoreOpenOptions,
    ) -> Result<Self> {
        let DurableStoreOpenOptions {
            read_only,
            initialize_if_empty,
            load_rebuildable_artifacts,
            segment_cache_capacity_bytes,
            max_graph_manifest_open_bytes,
            max_wal_bytes,
            max_record_bytes,
            max_batch_operations,
            automatic_tail_repair,
        } = options;
        if max_graph_manifest_open_bytes == 0 {
            return Err(HawDBError::Storage(
                "max_graph_manifest_open_bytes must be non-zero".to_string(),
            ));
        }
        let project_files = crate::file_descriptors::ProjectFileDescriptors::registered(path)?;
        let directory_lease =
            DatabaseDirectoryLease::acquire(path).map_err(HawDBError::from_storage_error)?;
        if load_rebuildable_artifacts {
            derived_repair::reject_pending_derived_artifact_repair(path)?;
        }
        let wal_tail_repair = match automatic_tail_repair {
            Some(config) => doctor::resume_automatic_wal_tail_repair_locked(path, config)?,
            None => {
                doctor::reject_pending_wal_doctor_repair(path)?;
                None
            }
        };
        let manifest_path = path.join(MANIFEST_FILE);
        let manifest = if fs::try_exists(&manifest_path)? {
            DurableManifest::load(&manifest_path)?
        } else if has_storage_artifacts(path)? {
            return Err(HawDBError::Storage(
                "database has storage artifacts but no durable manifest".to_string(),
            ));
        } else if initialize_if_empty {
            let manifest = DurableManifest::initial_generation();
            manifest.write(&manifest_path)?;
            manifest
        } else {
            return Err(HawDBError::Storage(
                "read-only database directory has no durable manifest".to_string(),
            ));
        };
        manifest.validate()?;
        if !read_only {
            cleanup_abandoned_checkpoint_preparations(artifact_path, manifest.checkpoint_epoch)?;
        }
        let checkpoint_path = manifest.checkpoint_path(artifact_path);
        let wal_path = manifest.wal_path(path);
        let wal_bytes = match fs::metadata(&wal_path) {
            Ok(metadata) => metadata.len(),
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => 0,
            Err(error) => return Err(error.into()),
        };
        let segment_cache = Arc::new(SegmentCache::new(segment_cache_capacity_bytes));
        let store_id = store_id_for_path(artifact_path)?;
        let mut graph_manifest_budget = GraphManifestOpenBudget::new(max_graph_manifest_open_bytes);
        let canonical_segments = load_published_canonical_segments(
            artifact_path,
            manifest,
            Arc::clone(&segment_cache),
            store_id,
            &mut graph_manifest_budget,
        )?;
        let canonical_adjacency = load_rebuildable_artifacts
            .then(|| {
                load_published_canonical_adjacency(
                    artifact_path,
                    manifest,
                    Arc::clone(&segment_cache),
                    store_id,
                    &mut graph_manifest_budget,
                )
            })
            .transpose()?
            .flatten();
        let persistent_property_projection = load_rebuildable_artifacts
            .then(|| {
                load_published_property_projection(
                    artifact_path,
                    manifest,
                    Arc::clone(&segment_cache),
                    store_id,
                    &mut graph_manifest_budget,
                )
            })
            .transpose()?
            .flatten();
        if let (Some(canonical), Some(adjacency)) = (&canonical_segments, &canonical_adjacency)
            && canonical.manifest().relationship_count != adjacency.relationship_count()
        {
            return Err(HawDBError::Storage(
                "canonical adjacency relationship count does not match canonical segments"
                    .to_string(),
            ));
        }
        let mut source_scan_reader = FileSegmentRangeReader::new().with_cache(
            Arc::clone(&segment_cache),
            store_id,
            ManifestGeneration(manifest.checkpoint_epoch),
        );
        source_scan_reader.register(
            source_scan::SOURCE_SCAN_ARTIFACT_ID,
            artifact_path.join(source_scan::SOURCE_SCAN_PAYLOAD_FILE),
        );
        Ok(Self {
            project_files,
            _directory_lease: Arc::new(directory_lease),
            branch_runtime: None,
            allocation_project: None,
            recovery_artifact_writes_allowed: !read_only,
            root_path: artifact_path.to_path_buf(),
            checkpoint_path,
            manifest_path,
            projected_graphs_path: artifact_path.join(PROJECTED_GRAPHS_FILE),
            stable_id_mapping_path: artifact_path.join(STABLE_ID_MAPPING_FILE),
            wal_path,
            wal_append_file: None,
            wal_append_owner: None,
            #[cfg(test)]
            wal_append_open_count: 0,
            checkpoint_encoded_len: manifest.checkpoint_encoded_len,
            checkpoint_encoded_checksum: manifest.checkpoint_encoded_checksum,
            checkpoint_encoded_sha256: manifest.checkpoint_encoded_sha256,
            relational_checkpoint_encoded_len: None,
            relational_checkpoint_encoded_checksum: None,
            relational_checkpoint_encoded_sha256: None,
            canonical_manifest_encoded_len: manifest.canonical_manifest_encoded_len,
            canonical_manifest_encoded_checksum: manifest.canonical_manifest_encoded_checksum,
            canonical_manifest_encoded_sha256: manifest.canonical_manifest_encoded_sha256,
            canonical_adjacency_generation_artifacts: manifest
                .canonical_adjacency_generation_artifacts,
            property_spill_manifest_encoded_len: manifest.property_spill_manifest_encoded_len,
            property_spill_manifest_encoded_checksum: manifest
                .property_spill_manifest_encoded_checksum,
            property_spill_manifest_encoded_sha256: manifest.property_spill_manifest_encoded_sha256,
            property_projection_manifest_encoded_len: manifest
                .property_projection_manifest_encoded_len,
            property_projection_manifest_encoded_checksum: manifest
                .property_projection_manifest_encoded_checksum,
            property_projection_manifest_encoded_sha256: manifest
                .property_projection_manifest_encoded_sha256,
            relational_row_generation_artifacts: manifest.relational_row_generation_artifacts,
            relational_overflow_generation_artifacts: manifest
                .relational_overflow_generation_artifacts,
            relational_index_generation_artifacts: manifest.relational_index_generation_artifacts,
            append_generation_artifacts: manifest.append_generation_artifacts,
            wal_generation: manifest.wal_generation,
            checkpoint_epoch: manifest.checkpoint_epoch,
            checkpoint_commit_epoch: manifest.checkpoint_commit_epoch,
            oldest_reader_commit_epoch: manifest.oldest_reader_commit_epoch,
            safe_reclaim_commit_epoch: manifest.safe_reclaim_commit_epoch,
            wal_replay_start_lsn: manifest.wal_replay_start_lsn,
            next_lsn: manifest.next_lsn,
            wal_bytes,
            wal_uncheckpointed_since: (wal_bytes > WAL_BINARY_FILE_HEADER_BYTES as u64)
                .then(std::time::Instant::now),
            wal_tail_repair,
            wal_commit_epoch: manifest.checkpoint_commit_epoch,
            max_wal_bytes,
            source_scan_commit_epoch: manifest.source_scan_commit_epoch,
            source_scan_descriptor_checksum: manifest.source_scan_descriptor_checksum,
            store_id,
            segment_cache,
            max_graph_manifest_open_bytes,
            canonical_segments,
            canonical_adjacency,
            persistent_property_projection,
            stable_id_mapping_reader: None,
            source_scan_reader,
            durability,
            read_only,
            max_record_bytes,
            max_batch_operations,
            telemetry: None,
            wal_sync_group: None,
            known_durable_commit_epoch: manifest.checkpoint_commit_epoch,
            wal_namespace_known_durable: false,
            generation_reclamation_debt: Arc::new(std::sync::Mutex::new(
                GenerationReclamationDebt::default(),
            )),
            wal_free_space_probe: WalFreeSpaceProbeState::default(),
        })
    }

    pub(super) fn root_path(&self) -> &Path {
        &self.root_path
    }

    pub(super) fn reserve_branch_admission_resources(
        &self,
        minimum: usize,
    ) -> Result<crate::file_descriptors::DescriptorReservation> {
        self.project_files
            .reserve_admission(minimum)
            .map_err(HawDBError::from_storage_error)
    }

    pub(super) fn retain_admitted_runtime(
        &self,
    ) -> Arc<crate::file_descriptors::AdmittedRuntimeOwner> {
        self.project_files.retain_admitted_runtime()
    }

    pub(super) fn manifest_path(&self) -> &Path {
        &self.manifest_path
    }

    pub(super) fn next_checkpoint_generation(&self) -> Result<u64> {
        // Sealing can advance the active WAL before branch admission. A
        // checkpoint candidate must never replace that acknowledged prefix.
        let mut generation = self
            .checkpoint_epoch
            .max(self.wal_generation)
            .checked_add(1)
            .ok_or_else(|| HawDBError::Storage("checkpoint generation overflow".to_string()))?;
        let branch_directory = self
            .branch_runtime
            .as_ref()
            .and_then(|branch| branch.head_path().parent());
        // Interrupted candidates are evidence, not reusable WAL names. Bound
        // discovery per attempt; maintenance must resolve a larger backlog.
        for _ in 0..256 {
            let preparation_path = self
                .root_path
                .join(format!(".checkpoint.{generation}.prepare"));
            let preparation_exists = match fs::symlink_metadata(&preparation_path) {
                Ok(metadata) if metadata.is_dir() => true,
                Ok(_) => {
                    // Only a directory can be an owned preparation. An
                    // unexpected file or link is an operational obstruction,
                    // not another candidate to silently allocate around.
                    return Err(HawDBError::Storage(format!(
                        "checkpoint preparation namespace is not a directory: {}",
                        preparation_path.display()
                    )));
                }
                Err(error) if error.kind() == std::io::ErrorKind::NotFound => false,
                Err(error) => return Err(error.into()),
            };
            let wal_exists = match branch_directory {
                Some(directory) => fs::try_exists(
                    directory.join(crate::artifact_files::wal_generation_file(generation)),
                )?,
                None => false,
            };
            if !preparation_exists && !wal_exists {
                return Ok(generation);
            }
            generation = generation
                .checked_add(1)
                .ok_or_else(|| HawDBError::Storage("checkpoint generation overflow".into()))?;
        }
        Err(HawDBError::Storage(
            "checkpoint generation discovery exceeded its bounded orphan probe budget".into(),
        ))
    }

    pub(super) fn defer_generation_reclamation(&self) {
        // Preserve already measured file/byte debt when execution admission
        // postpones the next scan. Unknown work is explicitly retry-required.
        self.generation_reclamation_debt
            .lock()
            .unwrap_or_else(|error| error.into_inner())
            .retry_required = true;
    }

    pub(super) fn generation_reclamation_debt(&self) -> GenerationReclamationDebt {
        *self
            .generation_reclamation_debt
            .lock()
            .unwrap_or_else(|error| error.into_inner())
    }

    pub(super) fn store_id(&self) -> StoreId {
        self.store_id
    }

    pub(super) const fn graph_manifest_open_budget_bytes(&self) -> u64 {
        self.max_graph_manifest_open_bytes
    }

    #[cfg(test)]
    pub(super) fn set_graph_manifest_open_budget_bytes(&mut self, max_bytes: u64) {
        self.max_graph_manifest_open_bytes = max_bytes;
    }

    pub(super) fn graph_manifest_encoded_bytes(&self) -> u64 {
        [
            self.canonical_manifest_encoded_len,
            self.canonical_adjacency_generation_artifacts
                .map(|binding| binding.descriptor_root_artifact.encoded_len),
            self.property_spill_manifest_encoded_len,
            self.property_projection_manifest_encoded_len,
        ]
        .into_iter()
        .flatten()
        .fold(0u64, u64::saturating_add)
    }

    fn admit_graph_manifest_artifacts(
        &self,
        artifacts: &CheckpointManifestArtifacts,
    ) -> Result<()> {
        let mut budget = GraphManifestOpenBudget::new(self.max_graph_manifest_open_bytes);
        let adjacency_root = artifacts
            .canonical_adjacency
            .generation
            .descriptor_root_artifact;
        for (artifact, metadata, format_max_bytes) in [
            (
                "canonical manifest",
                artifacts.canonical_manifest,
                CANONICAL_MANIFEST_MAX_BYTES,
            ),
            (
                "property spill manifest",
                artifacts.property_spill_manifest,
                PROPERTY_SPILL_MANIFEST_MAX_BYTES,
            ),
            (
                "canonical adjacency descriptor root",
                DurableArtifactMetadata {
                    encoded_len: adjacency_root.encoded_len,
                    encoded_checksum: u64::from(adjacency_root.encoded_crc32c),
                    encoded_sha256: adjacency_root.encoded_sha256,
                },
                GraphDescriptorTreeBuildConfig::default()
                    .max_root_bytes
                    .get() as u64,
            ),
            (
                "property projection manifest",
                artifacts.property_projection_manifest,
                PROPERTY_PROJECTION_MANIFEST_MAX_BYTES,
            ),
        ] {
            admit_graph_manifest_binding(
                metadata.encoded_len,
                format_max_bytes,
                artifact,
                &mut budget,
            )?;
        }
        Ok(())
    }
}

#[cfg(test)]
thread_local! {
    pub(in crate::store) static AFTER_CAPTURED_WAL_APPEND_OPEN: std::cell::RefCell<Option<Box<dyn FnOnce()>>> = const {
        std::cell::RefCell::new(None)
    };
}