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.

//! Reader-safe physical-generation retention and retryable reclamation.

use super::{DurableStore, GenerationReclamationDebt};
use crate::error::{HawDBError, Result};
use crate::file_io as fs;
use crate::store::{
    parse_append_segment_generation_file, parse_relational_overflow_extent_generation_file,
    parse_relational_row_page_artifact_generation_file, remove_generation_reclamation_candidate,
    storage_generation_for_file, sync_parent_dir,
};
use hawdb_storage::append_table::{
    append_generation_manifest_file, AppendGenerationManifest, AppendPublicationConfig,
};
use std::collections::BTreeSet;

impl DurableStore {
    pub(in crate::store) fn obsolete_generation_bytes(
        &self,
        oldest_reader_commit_epoch: Option<u64>,
    ) -> u64 {
        if oldest_reader_commit_epoch.is_none() {
            return 0;
        }
        let retain_from = self.checkpoint_epoch.saturating_sub(1);
        fs::read_dir(&self.root_path)
            .into_iter()
            .flatten()
            .filter_map(std::result::Result::ok)
            .filter_map(|entry| {
                let name = entry.file_name();
                let name = name.to_str()?;
                let generation = storage_generation_for_file(name)?;
                (generation < retain_from)
                    .then(|| entry.metadata().ok().map(|metadata| metadata.len()))
                    .flatten()
            })
            .fold(0u64, u64::saturating_add)
    }

    pub(in crate::store) fn reclaim_old_generations(
        &mut self,
        current_generation: u64,
        previous_checkpoint_generation: u64,
        pinned_reader_generations: Option<&BTreeSet<u64>>,
    ) {
        let debt = self.try_reclaim_old_generations(
            current_generation,
            previous_checkpoint_generation,
            pinned_reader_generations,
        );
        // Filesystem work stays outside this lock. Publish the complete debt
        // receipt atomically to all views of this admitted runtime.
        *self
            .generation_reclamation_debt
            .lock()
            .unwrap_or_else(|error| error.into_inner()) = debt;
    }

    fn try_reclaim_old_generations(
        &self,
        current_generation: u64,
        previous_checkpoint_generation: u64,
        pinned_reader_generations: Option<&BTreeSet<u64>>,
    ) -> GenerationReclamationDebt {
        let mut debt = GenerationReclamationDebt::default();
        let Ok(_descriptor_operation) = self.file_descriptor_context().internal_operation() else {
            debt.retry_required = true;
            return debt;
        };
        if self.oldest_reader_commit_epoch.is_some() && pinned_reader_generations.is_none() {
            return debt;
        }

        let mut retained_generations = pinned_reader_generations.cloned().unwrap_or_default();
        retained_generations.insert(current_generation);
        // WAL rotations share the namespace and may leave gaps between
        // checkpoints. Preserve the actual preceding recovery generation.
        if previous_checkpoint_generation > 0 {
            retained_generations.insert(previous_checkpoint_generation);
        }
        let (retained_row_page_generations, retained_overflow_extent_generations) =
            match self.retained_relational_physical_generations(&retained_generations) {
                Ok(generations) => generations,
                Err(_) => {
                    debt.retry_required = true;
                    return debt;
                }
            };
        let retained_append_segment_generations =
            match self.retained_append_physical_generations(&retained_generations) {
                Ok(generations) => generations,
                Err(_) => {
                    debt.retry_required = true;
                    return debt;
                }
            };
        let entries = match fs::read_dir(&self.root_path) {
            Ok(entries) => entries,
            Err(_) => {
                debt.retry_required = true;
                return debt;
            }
        };
        for entry in entries {
            let entry = match entry {
                Ok(entry) => entry,
                Err(_) => {
                    debt.retry_required = true;
                    continue;
                }
            };
            let name = entry.file_name();
            let Some(name) = name.to_str() else {
                continue;
            };
            let generation = storage_generation_for_file(name);
            if generation.is_some_and(|generation| {
                generation < current_generation && !retained_generations.contains(&generation)
            }) {
                if parse_relational_row_page_artifact_generation_file(name)
                    .is_some_and(|generation| retained_row_page_generations.contains(&generation))
                    || parse_relational_overflow_extent_generation_file(name).is_some_and(
                        |generation| retained_overflow_extent_generations.contains(&generation),
                    )
                    || parse_append_segment_generation_file(name).is_some_and(|generation| {
                        retained_append_segment_generations.contains(&generation)
                    })
                {
                    continue;
                }
                let pending_bytes = entry.metadata().map_or(0, |metadata| metadata.len());
                if remove_generation_reclamation_candidate(&entry.path()).is_err() {
                    debt.retry_required = true;
                    debt.pending_file_count = debt.pending_file_count.saturating_add(1);
                    debt.pending_bytes = debt.pending_bytes.saturating_add(pending_bytes);
                }
            }
        }
        if sync_parent_dir(&self.manifest_path).is_err() {
            debt.retry_required = true;
        }
        debt
    }

    fn retained_relational_physical_generations(
        &self,
        retained_generations: &BTreeSet<u64>,
    ) -> Result<(BTreeSet<u64>, BTreeSet<u64>)> {
        let mut row_page_generations = BTreeSet::new();
        let mut overflow_extent_generations = BTreeSet::new();

        for &generation in retained_generations {
            let overflow_manifest = self.root_path.join(
                hawdb_storage::relational::relational_overflow_manifest_generation_file(generation),
            );
            if overflow_manifest.exists() {
                let overflow =
                    hawdb_storage::relational::RelationalOverflowRootReader::open_generation(
                        &self.root_path,
                        generation,
                        hawdb_storage::relational::RelationalOverflowPublicationConfig::default(),
                    )
                    .map_err(HawDBError::from_storage_error)?;
                overflow
                    .visit_descriptors(|descriptor| {
                        overflow_extent_generations.insert(descriptor.physical_generation);
                        Ok(())
                    })
                    .map_err(HawDBError::from_storage_error)?;
            }

            let row_manifest = self.root_path.join(
                hawdb_storage::relational::relational_row_page_manifest_generation_file(generation),
            );
            if row_manifest.exists() {
                let rows = hawdb_storage::relational::RelationalRowPageRootReader::open_generation(
                    &self.root_path,
                    generation,
                    hawdb_storage::relational::RelationalRowPagePublicationConfig::default(),
                )
                .map_err(HawDBError::from_storage_error)?;
                let tables = rows
                    .manifest()
                    .tables
                    .iter()
                    .map(|table| table.table.clone())
                    .collect::<Vec<_>>();
                for table in tables {
                    rows.visit_table_pages(&table, |descriptor| {
                        row_page_generations.insert(descriptor.physical_generation);
                        Ok(())
                    })
                    .map_err(HawDBError::from_storage_error)?;
                }
            }
        }

        Ok((row_page_generations, overflow_extent_generations))
    }

    fn retained_append_physical_generations(
        &self,
        retained_generations: &BTreeSet<u64>,
    ) -> Result<BTreeSet<u64>> {
        let mut segment_generations = BTreeSet::new();
        for &generation in retained_generations {
            let path = self
                .root_path
                .join(append_generation_manifest_file(generation));
            if !path.exists() {
                continue;
            }
            let manifest = AppendGenerationManifest::read_generation(
                &self.root_path,
                generation,
                AppendPublicationConfig::default(),
            )
            .map_err(HawDBError::from_storage_error)?;
            segment_generations.extend(manifest.segments.iter().map(|segment| segment.generation));
        }
        Ok(segment_generations)
    }
}