hawdb-storage 0.13.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.

//! Bounded streaming construction for immutable graph descriptor trees.

use super::{
    admission, corrupt, remove_if_exists, GraphDescriptorTreeBuildConfig,
    GraphDescriptorTreeBuildReport, GraphDescriptorTreeError, GraphDescriptorTreePaths,
    GraphDescriptorTreeRoot, PreparedGraphDescriptorTree, ROOT_HEADER_BYTES,
};
use crate::background::{CheckpointAllocationOwner, CheckpointWorkContext, CheckpointWorkError};
use crate::file_io::{self as fs, File};
use crate::graph_descriptor_page::{
    decode_page_ref, encode_page_ref, GraphDescriptorInteriorEntry, GraphDescriptorKind,
    GraphDescriptorLeafEntry, GraphDescriptorPageId, GraphDescriptorPageLimits,
    GraphDescriptorPageRef, ImmutableGraphDescriptorPage, ImmutableGraphDescriptorPageBody,
    GRAPH_DESCRIPTOR_FIELD_HEADER_BYTES, GRAPH_DESCRIPTOR_PAGE_HEADER_BYTES,
};
use hawdb_integrity::IntegrityHasher;
use std::collections::BTreeSet;
use std::io::{BufReader, BufWriter, Read, Write};
use std::num::NonZeroU64;
use std::path::{Path, PathBuf};

const REF_RUN_MAGIC: &[u8; 8] = b"SKGDRF01";
const TREE_IO_BUFFER_BYTES: usize = 8 * 1024;

mod finish;
mod writer;
use writer::BufferedWriter;

struct WriterWork {
    context: CheckpointWorkContext,
    checkpoint_buffers: bool,
}

struct PendingLeafPage {
    page: ImmutableGraphDescriptorPage,
    reference: Option<GraphDescriptorPageRef>,
}

pub struct GraphDescriptorTreeBuilder {
    work: CheckpointWorkContext,
    paths: GraphDescriptorTreePaths,
    kind: GraphDescriptorKind,
    generation: u64,
    source_commit_epoch: u64,
    artifact_id: u64,
    config: GraphDescriptorTreeBuildConfig,
    page_writer: BufferedWriter,
    page_hasher: IntegrityHasher,
    page_artifact_bytes: u64,
    next_page_id: u64,
    descriptor_count: u64,
    page_count: u64,
    leaf_page_count: u64,
    leaf_entries: Vec<GraphDescriptorLeafEntry>,
    pending_leaf_page: Option<PendingLeafPage>,
    finish_progress: Option<finish::Progress>,
    finished: bool,
    leaf_payload_bytes: usize,
    last_key: Option<Vec<u8>>,
    level_zero: Option<RefRunWriter>,
    total_intermediate_bytes: u64,
    peak_intermediate_level_bytes: u64,
    peak_resident_bytes: u64,
    temporary_files: TemporaryFiles,
    checkpoint_buffers: bool,
    // Buffered data must be destroyed before its retained allocation inventory.
    _page_buffer_memory: CheckpointAllocationOwner,
}

impl GraphDescriptorTreeBuilder {
    // The enclosing preparation retains the original memory admission and calls
    // this only between attempts, with no work unit or I/O wave held.
    pub(crate) fn bind_checkpoint_attempt(
        &mut self,
        work: &crate::background::CheckpointWorkContext,
    ) {
        self.work = work.clone();
    }

    pub fn create(
        paths: GraphDescriptorTreePaths,
        kind: GraphDescriptorKind,
        generation: u64,
        source_commit_epoch: u64,
        artifact_id: u64,
        config: GraphDescriptorTreeBuildConfig,
    ) -> Result<Self, GraphDescriptorTreeError> {
        Self::create_with_work_context(
            paths,
            kind,
            generation,
            source_commit_epoch,
            artifact_id,
            config,
            CheckpointWorkContext::default(),
        )
    }

    #[doc(hidden)]
    pub fn create_with_work_context(
        paths: GraphDescriptorTreePaths,
        kind: GraphDescriptorKind,
        generation: u64,
        source_commit_epoch: u64,
        artifact_id: u64,
        config: GraphDescriptorTreeBuildConfig,
        work: CheckpointWorkContext,
    ) -> Result<Self, GraphDescriptorTreeError> {
        Self::create_inner(
            paths,
            kind,
            generation,
            source_commit_epoch,
            artifact_id,
            config,
            WriterWork {
                context: work,
                checkpoint_buffers: false,
            },
        )
    }

    pub(crate) fn create_checkpoint(
        paths: GraphDescriptorTreePaths,
        kind: GraphDescriptorKind,
        generation: u64,
        source_commit_epoch: u64,
        artifact_id: u64,
        config: GraphDescriptorTreeBuildConfig,
        work: CheckpointWorkContext,
    ) -> Result<Self, GraphDescriptorTreeError> {
        Self::create_inner(
            paths,
            kind,
            generation,
            source_commit_epoch,
            artifact_id,
            config,
            WriterWork {
                context: work,
                checkpoint_buffers: true,
            },
        )
    }

    fn create_inner(
        paths: GraphDescriptorTreePaths,
        kind: GraphDescriptorKind,
        generation: u64,
        source_commit_epoch: u64,
        artifact_id: u64,
        config: GraphDescriptorTreeBuildConfig,
        writer_work: WriterWork,
    ) -> Result<Self, GraphDescriptorTreeError> {
        let WriterWork {
            context: work,
            checkpoint_buffers,
        } = writer_work;
        let unit = work.start_unit()?;
        let _wave = work.io_wave()?;
        if generation == 0 || artifact_id == 0 {
            return Err(admission(
                "graph descriptor tree generation and artifact id must be non-zero",
            ));
        }
        if config.page_limits.max_payload_bytes().is_none() {
            return Err(admission(
                "graph descriptor page limit is smaller than its fixed header",
            ));
        }
        if config.page_limits.max_entries.get() < 2 {
            return Err(admission(
                "graph descriptor tree requires an interior fanout of at least two",
            ));
        }
        if config.max_root_bytes.get() < ROOT_HEADER_BYTES {
            return Err(admission(format!(
                "graph descriptor root limit {} is smaller than its fixed header {ROOT_HEADER_BYTES}",
                config.max_root_bytes
            )));
        }
        if config.max_intermediate_bytes.get() < REF_RUN_MAGIC.len() as u64 {
            return Err(admission(
                "graph descriptor intermediate budget is smaller than a run header",
            ));
        }
        let page_tmp = paths.page_tmp();
        let root_tmp = paths.root_tmp();
        let level_zero_path = paths.ref_run(0);
        let owned_paths = [
            &paths.page_artifact,
            &paths.root_manifest,
            &page_tmp,
            &root_tmp,
            &level_zero_path,
        ];
        for (index, path) in owned_paths.iter().enumerate() {
            if owned_paths[index + 1..].contains(path) {
                return Err(admission(format!(
                    "graph descriptor tree path {} aliases another owned path",
                    path.display()
                )));
            }
        }
        for destination in [&paths.page_artifact, &paths.root_manifest] {
            if destination.exists() {
                return Err(admission(format!(
                    "immutable graph descriptor artifact {} already exists",
                    destination.display()
                )));
            }
        }
        remove_if_exists(&page_tmp)?;
        remove_if_exists(&root_tmp)?;
        remove_if_exists(&level_zero_path)?;
        let mut page_buffer_memory = CheckpointAllocationOwner::default();
        let page_token = if checkpoint_buffers {
            Some(page_buffer_memory.reserve(TREE_IO_BUFFER_BYTES, &work)?)
        } else {
            None
        };
        let mut temporary_files = TemporaryFiles::default();
        temporary_files.track(page_tmp.clone());
        temporary_files.track(level_zero_path.clone());
        let page_writer = BufferedWriter::new(
            TREE_IO_BUFFER_BYTES,
            File::create(&page_tmp)?,
            checkpoint_buffers,
        );
        if let Some(token) = page_token {
            check_buffer_capacity(page_writer.capacity(), &work)?;
            token.address(page_writer.buffer().as_ptr() as usize);
        }
        let level_zero = if checkpoint_buffers {
            RefRunWriter::create_checkpoint(level_zero_path, &work)?
        } else {
            RefRunWriter::create(level_zero_path)?
        };
        unit.finish();
        Ok(Self {
            work: work.clone(),
            paths,
            kind,
            generation,
            source_commit_epoch,
            artifact_id,
            config,
            page_writer,
            page_hasher: IntegrityHasher::new(),
            page_artifact_bytes: 0,
            next_page_id: 1,
            descriptor_count: 0,
            page_count: 0,
            leaf_page_count: 0,
            leaf_entries: Vec::new(),
            pending_leaf_page: None,
            finish_progress: None,
            finished: false,
            leaf_payload_bytes: 0,
            last_key: None,
            level_zero: Some(level_zero),
            total_intermediate_bytes: REF_RUN_MAGIC.len() as u64,
            peak_intermediate_level_bytes: 0,
            peak_resident_bytes: (2 * TREE_IO_BUFFER_BYTES) as u64,
            temporary_files,
            checkpoint_buffers,
            _page_buffer_memory: page_buffer_memory,
        })
    }

    pub fn push(&mut self, key: Vec<u8>, value: Vec<u8>) -> Result<(), GraphDescriptorTreeError> {
        if self.finish_progress.is_some() || self.finished {
            return Err(admission("cannot append to a finishing descriptor tree"));
        }
        let unit = self.work.start_unit()?;
        validate_leaf_input(&key, &value, self.config.page_limits)?;
        if self
            .last_key
            .as_ref()
            .is_some_and(|previous| previous.as_slice() >= key.as_slice())
        {
            return Err(admission(
                "graph descriptor keys must be supplied in strictly increasing order",
            ));
        }
        let entry_payload_bytes = leaf_entry_payload_bytes(key.len(), value.len())?;
        let max_payload_bytes = self
            .config
            .page_limits
            .max_payload_bytes()
            .expect("page limit was validated at builder creation");
        if entry_payload_bytes > max_payload_bytes {
            return Err(admission(format!(
                "graph descriptor leaf entry requires {entry_payload_bytes} payload bytes, exceeding page payload limit {max_payload_bytes}"
            )));
        }
        let page_full = !self.leaf_entries.is_empty()
            && (self.leaf_entries.len() == self.config.page_limits.max_entries.get()
                || self
                    .leaf_payload_bytes
                    .checked_add(entry_payload_bytes)
                    .is_none_or(|bytes| bytes > max_payload_bytes));
        if page_full || self.pending_leaf_page.is_some() {
            self.flush_leaf_page()?;
        }
        self.leaf_payload_bytes = self
            .leaf_payload_bytes
            .checked_add(entry_payload_bytes)
            .ok_or_else(|| admission("graph descriptor leaf payload length overflow"))?;
        self.last_key = Some(key.clone());
        self.leaf_entries
            .push(GraphDescriptorLeafEntry { key, value });
        self.descriptor_count = self
            .descriptor_count
            .checked_add(1)
            .ok_or_else(|| admission("graph descriptor count overflow"))?;
        self.observe_resident_leaf();
        unit.finish();
        Ok(())
    }

    pub fn finish(mut self) -> Result<PreparedGraphDescriptorTree, GraphDescriptorTreeError> {
        let work = self.work.clone();
        self.finish_with_work_context(&work)
    }

    pub(crate) fn finish_with_work_context(
        &mut self,
        work: &CheckpointWorkContext,
    ) -> Result<PreparedGraphDescriptorTree, GraphDescriptorTreeError> {
        if self.finished {
            return Err(admission("descriptor tree already finished"));
        }
        self.bind_checkpoint_attempt(work);
        let mut progress = self.finish_progress.take().unwrap_or_default();
        let result = finish::resume(self, &mut progress);
        match result {
            Ok(prepared) => {
                self.finished = true;
                Ok(prepared)
            }
            Err(error) => {
                self.finish_progress = Some(progress);
                Err(error)
            }
        }
    }

    fn flush_leaf_page(&mut self) -> Result<(), GraphDescriptorTreeError> {
        if self.pending_leaf_page.is_none() {
            if self.leaf_entries.is_empty() {
                return Ok(());
            }
            let page_id = self.allocate_page_id()?;
            self.pending_leaf_page = Some(PendingLeafPage {
                page: ImmutableGraphDescriptorPage {
                    kind: self.kind,
                    physical_generation: self.generation,
                    source_commit_epoch: self.source_commit_epoch,
                    page_id,
                    body: ImmutableGraphDescriptorPageBody::Leaf(std::mem::take(
                        &mut self.leaf_entries,
                    )),
                },
                reference: None,
            });
            self.leaf_payload_bytes = 0;
        }
        // Only the completed reference append retires the captured leaf. A
        // denied work unit or wave must preserve either the original page or
        // its already-written reference, without writing the data page twice.
        let mut pending = self.pending_leaf_page.take().expect("captured leaf page");
        let result = (|| {
            if pending.reference.is_none() {
                pending.reference = Some(self.write_page(&pending.page)?);
            }
            self.write_level_zero_ref(pending.reference.as_ref().expect("written leaf page"))?;
            self.leaf_page_count = self
                .leaf_page_count
                .checked_add(1)
                .ok_or_else(|| admission("graph descriptor leaf page count overflow"))?;
            Ok(())
        })();
        if result.is_err() {
            self.pending_leaf_page = Some(pending);
        }
        result
    }

    fn allocate_page_id(&mut self) -> Result<GraphDescriptorPageId, GraphDescriptorTreeError> {
        let page_id = NonZeroU64::new(self.next_page_id)
            .map(GraphDescriptorPageId::new)
            .ok_or_else(|| admission("graph descriptor page id overflow"))?;
        self.next_page_id = self
            .next_page_id
            .checked_add(1)
            .ok_or_else(|| admission("graph descriptor page id overflow"))?;
        Ok(page_id)
    }

    fn write_page(
        &mut self,
        page: &ImmutableGraphDescriptorPage,
    ) -> Result<GraphDescriptorPageRef, GraphDescriptorTreeError> {
        let unit = self.work.start_unit()?;
        let required_pages = self
            .page_count
            .checked_add(1)
            .ok_or_else(|| admission("graph descriptor page count overflow"))?;
        if required_pages > self.config.max_page_count.get() {
            return Err(admission(format!(
                "graph descriptor tree requires {required_pages} pages, exceeding limit {}",
                self.config.max_page_count
            )));
        }
        let (encoded, reference) = page.encode_with_ref(
            self.artifact_id,
            self.page_artifact_bytes,
            self.config.page_limits,
        )?;
        let required_artifact_bytes = self
            .page_artifact_bytes
            .checked_add(encoded.len() as u64)
            .ok_or_else(|| admission("graph descriptor page artifact length overflow"))?;
        if required_artifact_bytes > self.config.max_page_artifact_bytes.get() {
            return Err(admission(format!(
                "graph descriptor page artifact requires {required_artifact_bytes} bytes, exceeding limit {}",
                self.config.max_page_artifact_bytes
            )));
        }
        let page_resident_bytes = page_body_resident_bytes(&page.body)
            .saturating_add(encoded.len())
            .saturating_add(3 * TREE_IO_BUFFER_BYTES) as u64;
        self.peak_resident_bytes = self.peak_resident_bytes.max(page_resident_bytes);
        {
            let _wave = self.work.io_wave()?;
            self.page_writer.write_all(&encoded)?;
        }
        self.page_hasher.update(&encoded);
        self.page_artifact_bytes = required_artifact_bytes;
        self.page_count = required_pages;
        self.peak_resident_bytes = self.peak_resident_bytes.max(encoded.len() as u64);
        unit.finish();
        Ok(reference)
    }

    fn write_level_zero_ref(
        &mut self,
        reference: &GraphDescriptorPageRef,
    ) -> Result<(), GraphDescriptorTreeError> {
        let mut writer = self
            .level_zero
            .take()
            .expect("graph descriptor level-zero writer is present");
        let result = self.write_ref(&mut writer, reference);
        self.level_zero = Some(writer);
        result
    }

    fn write_ref(
        &mut self,
        writer: &mut RefRunWriter,
        reference: &GraphDescriptorPageRef,
    ) -> Result<(), GraphDescriptorTreeError> {
        self.work.checkpoint()?;
        let encoded = encode_page_ref(reference)?;
        let _ = u32::try_from(encoded.len())
            .map_err(|_| admission("graph descriptor page reference length exceeds u32"))?;
        let encoded_len = 4u64.saturating_add(encoded.len() as u64);
        let required = self
            .total_intermediate_bytes
            .checked_add(encoded_len)
            .ok_or_else(|| admission("graph descriptor intermediate byte count overflow"))?;
        if required > self.config.max_intermediate_bytes.get() {
            return Err(admission(format!(
                "graph descriptor build requires {required} intermediate bytes, exceeding limit {}",
                self.config.max_intermediate_bytes
            )));
        }
        {
            let _wave = self.work.io_wave()?;
            writer.write_encoded_ref(&encoded)?;
        }
        self.total_intermediate_bytes = required;
        Ok(())
    }

    fn record_finished_run(&mut self, run: &RefRun) -> Result<(), GraphDescriptorTreeError> {
        if run.bytes > self.config.max_intermediate_bytes.get() {
            return Err(admission(format!(
                "graph descriptor reference level contains {} bytes, exceeding limit {}",
                run.bytes, self.config.max_intermediate_bytes
            )));
        }
        self.peak_intermediate_level_bytes = self.peak_intermediate_level_bytes.max(run.bytes);
        Ok(())
    }

    fn observe_resident_leaf(&mut self) {
        let structural = self
            .leaf_entries
            .len()
            .saturating_mul(std::mem::size_of::<GraphDescriptorLeafEntry>());
        let bytes = GRAPH_DESCRIPTOR_PAGE_HEADER_BYTES
            .saturating_add(self.leaf_payload_bytes)
            .saturating_add(structural) as u64;
        self.peak_resident_bytes = self.peak_resident_bytes.max(
            bytes
                .saturating_add((2 * TREE_IO_BUFFER_BYTES) as u64)
                .saturating_add(self.last_key.as_ref().map_or(0, Vec::len) as u64),
        );
    }

    fn observe_resident_interior(
        &mut self,
        payload_bytes: usize,
        children: &[GraphDescriptorInteriorEntry],
    ) {
        let structural = children
            .len()
            .saturating_mul(std::mem::size_of::<GraphDescriptorInteriorEntry>());
        let bytes = GRAPH_DESCRIPTOR_PAGE_HEADER_BYTES
            .saturating_add(payload_bytes)
            .saturating_add(structural) as u64;
        self.peak_resident_bytes = self
            .peak_resident_bytes
            .max(bytes.saturating_add((3 * TREE_IO_BUFFER_BYTES) as u64));
    }

    fn remove_temporary(&mut self, path: &Path) -> Result<(), GraphDescriptorTreeError> {
        remove_if_exists(path)?;
        self.temporary_files.disarm(path);
        Ok(())
    }
}

fn validate_leaf_input(
    key: &[u8],
    value: &[u8],
    limits: GraphDescriptorPageLimits,
) -> Result<(), GraphDescriptorTreeError> {
    if key.is_empty() || value.is_empty() {
        return Err(admission(
            "graph descriptor leaf key and value must be non-empty",
        ));
    }
    if key.len() > limits.max_key_bytes.get() || value.len() > limits.max_value_bytes.get() {
        return Err(admission(format!(
            "graph descriptor leaf key/value bytes {}/{} exceed limits {}/{}",
            key.len(),
            value.len(),
            limits.max_key_bytes,
            limits.max_value_bytes
        )));
    }
    Ok(())
}

fn leaf_entry_payload_bytes(
    key_len: usize,
    value_len: usize,
) -> Result<usize, GraphDescriptorTreeError> {
    GRAPH_DESCRIPTOR_FIELD_HEADER_BYTES
        .checked_add(4)
        .and_then(|bytes| bytes.checked_add(key_len))
        .and_then(|bytes| bytes.checked_add(4))
        .and_then(|bytes| bytes.checked_add(value_len))
        .ok_or_else(|| admission("graph descriptor leaf entry length overflow"))
}

fn interior_entry_payload_bytes(
    reference: &GraphDescriptorPageRef,
) -> Result<usize, GraphDescriptorTreeError> {
    let encoded = encode_page_ref(reference)?;
    GRAPH_DESCRIPTOR_FIELD_HEADER_BYTES
        .checked_add(encoded.len())
        .ok_or_else(|| admission("graph descriptor interior entry length overflow"))
}

fn page_ref_resident_bytes(reference: &GraphDescriptorPageRef) -> usize {
    std::mem::size_of::<GraphDescriptorPageRef>()
        .saturating_add(reference.lower_bound.len())
        .saturating_add(reference.upper_bound.len())
}

fn page_body_resident_bytes(body: &ImmutableGraphDescriptorPageBody) -> usize {
    match body {
        ImmutableGraphDescriptorPageBody::Leaf(entries) => {
            entries
                .iter()
                .fold(std::mem::size_of_val(entries.as_slice()), |bytes, entry| {
                    bytes
                        .saturating_add(entry.key.len())
                        .saturating_add(entry.value.len())
                })
        }
        ImmutableGraphDescriptorPageBody::Interior(entries) => entries
            .iter()
            .fold(std::mem::size_of_val(entries.as_slice()), |bytes, entry| {
                bytes.saturating_add(page_ref_resident_bytes(&entry.child))
            }),
    }
}

struct RefRun {
    path: PathBuf,
    count: u64,
    bytes: u64,
}

struct RefRunWriter {
    path: PathBuf,
    writer: BufferedWriter,
    count: u64,
    bytes: u64,
    _buffer_memory: CheckpointAllocationOwner,
}

impl RefRunWriter {
    fn create(path: PathBuf) -> Result<Self, GraphDescriptorTreeError> {
        Self::create_inner(path, None)
    }

    fn create_checkpoint(
        path: PathBuf,
        work: &CheckpointWorkContext,
    ) -> Result<Self, GraphDescriptorTreeError> {
        Self::create_inner(path, Some(work))
    }

    fn create_inner(
        path: PathBuf,
        work: Option<&CheckpointWorkContext>,
    ) -> Result<Self, GraphDescriptorTreeError> {
        let mut memory = CheckpointAllocationOwner::default();
        let token = work
            .map(|work| memory.reserve(TREE_IO_BUFFER_BYTES, work))
            .transpose()?;
        let mut writer =
            BufferedWriter::new(TREE_IO_BUFFER_BYTES, File::create(&path)?, work.is_some());
        if let (Some(token), Some(work)) = (token, work) {
            check_buffer_capacity(writer.capacity(), work)?;
            token.address(writer.buffer().as_ptr() as usize);
        }
        writer.write_all(REF_RUN_MAGIC)?;
        Ok(Self {
            path,
            writer,
            count: 0,
            bytes: REF_RUN_MAGIC.len() as u64,
            _buffer_memory: memory,
        })
    }

    fn write_encoded_ref(&mut self, encoded: &[u8]) -> Result<(), GraphDescriptorTreeError> {
        let len = u32::try_from(encoded.len())
            .map_err(|_| admission("graph descriptor page reference length exceeds u32"))?;
        self.writer.write_all(&len.to_le_bytes())?;
        self.writer.write_all(encoded)?;
        self.count = self
            .count
            .checked_add(1)
            .ok_or_else(|| admission("graph descriptor reference run count overflow"))?;
        self.bytes = self
            .bytes
            .checked_add(4 + encoded.len() as u64)
            .ok_or_else(|| admission("graph descriptor reference run length overflow"))?;
        Ok(())
    }

    fn finish_retained(&mut self) -> Result<RefRun, GraphDescriptorTreeError> {
        self.writer.flush()?;
        Ok(RefRun {
            path: self.path.clone(),
            count: self.count,
            bytes: self.bytes,
        })
    }
}

fn check_buffer_capacity(
    capacity: usize,
    work: &CheckpointWorkContext,
) -> Result<(), GraphDescriptorTreeError> {
    if capacity != TREE_IO_BUFFER_BYTES {
        return Err(GraphDescriptorTreeError::Work(work.record_failure(
            CheckpointWorkError::Allocation {
                bytes: TREE_IO_BUFFER_BYTES as u64,
                reason: "descriptor-tree writer capacity differs from admitted capacity".into(),
            },
        )));
    }
    Ok(())
}

struct RefRunReader {
    reader: BufReader<File>,
    expected_count: u64,
    decoded_count: u64,
    config: GraphDescriptorTreeBuildConfig,
}

impl RefRunReader {
    fn open(
        path: &Path,
        expected_count: u64,
        config: GraphDescriptorTreeBuildConfig,
    ) -> Result<Self, GraphDescriptorTreeError> {
        let mut reader = BufReader::with_capacity(TREE_IO_BUFFER_BYTES, File::open(path)?);
        let mut magic = [0u8; 8];
        reader.read_exact(&mut magic)?;
        if &magic != REF_RUN_MAGIC {
            return Err(corrupt("invalid graph descriptor reference run header"));
        }
        Ok(Self {
            reader,
            expected_count,
            decoded_count: 0,
            config,
        })
    }

    fn next_ref(&mut self) -> Result<Option<GraphDescriptorPageRef>, GraphDescriptorTreeError> {
        let mut first = [0u8; 1];
        match self.reader.read(&mut first)? {
            0 => return Ok(None),
            1 => {}
            _ => unreachable!("one-byte graph descriptor reference read"),
        }
        let mut remaining_len = [0u8; 3];
        self.reader.read_exact(&mut remaining_len)?;
        let len = u32::from_le_bytes([
            first[0],
            remaining_len[0],
            remaining_len[1],
            remaining_len[2],
        ]) as usize;
        let max_ref_bytes = self
            .config
            .max_root_bytes
            .get()
            .min(self.config.page_limits.max_page_bytes.get());
        if len == 0 || len > max_ref_bytes {
            return Err(corrupt(format!(
                "graph descriptor reference run record length {len} exceeds limit {max_ref_bytes}"
            )));
        }
        let mut encoded = vec![0u8; len];
        self.reader.read_exact(&mut encoded)?;
        self.decoded_count = self
            .decoded_count
            .checked_add(1)
            .ok_or_else(|| corrupt("graph descriptor reference run count overflow"))?;
        if self.decoded_count > self.expected_count {
            return Err(corrupt(
                "graph descriptor reference run contains more records than declared",
            ));
        }
        decode_page_ref(&encoded, self.config.page_limits)
            .map(Some)
            .map_err(GraphDescriptorTreeError::Page)
    }

    fn finish(&mut self) -> Result<(), GraphDescriptorTreeError> {
        if self.decoded_count != self.expected_count {
            return Err(corrupt(format!(
                "graph descriptor reference run count mismatch: expected {}, got {}",
                self.expected_count, self.decoded_count
            )));
        }
        let mut trailing = [0u8; 1];
        if self.reader.read(&mut trailing)? != 0 {
            return Err(corrupt(
                "graph descriptor reference run contains trailing records",
            ));
        }
        Ok(())
    }
}

#[derive(Default)]
struct TemporaryFiles {
    paths: BTreeSet<PathBuf>,
}

impl TemporaryFiles {
    fn track(&mut self, path: PathBuf) {
        self.paths.insert(path);
    }

    fn disarm(&mut self, path: &Path) {
        self.paths.remove(path);
    }
}

impl Drop for TemporaryFiles {
    fn drop(&mut self) {
        for path in &self.paths {
            let _ = fs::remove_file(path);
        }
    }
}

#[cfg(test)]
mod checkpoint_buffer_memory_tests;

#[cfg(all(test, unix))]
mod checkpoint_abort_io_tests;