siftdb_core/

tombstone.rs

use crate::types::HandleMetadata;
use anyhow::{Context, Result};
use std::collections::{HashMap, HashSet};
use std::fs;
use std::path::{Path, PathBuf};
use std::time::{SystemTime, UNIX_EPOCH};
use serde::{Deserialize, Serialize};

/// Tombstone entry for tracking deleted files
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Tombstone {
    pub file_handle: u32,
    pub path: PathBuf,
    pub deleted_at: u64,
    pub deleted_epoch: u64,
    /// Original segment and offset for cleanup during compaction
    pub segment_id: u32,
    pub segment_offset: u64,
}

impl Tombstone {
    pub fn new(
        file_handle: u32,
        path: PathBuf,
        deleted_epoch: u64,
        segment_id: u32,
        segment_offset: u64,
    ) -> Self {
        let deleted_at = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_secs();

        Self {
            file_handle,
            path,
            deleted_at,
            deleted_epoch,
            segment_id,
            segment_offset,
        }
    }
}

/// Tombstone registry for managing deleted files
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TombstoneRegistry {
    pub tombstones: HashMap<u32, Tombstone>, // file_handle -> tombstone
    pub path_to_handle: HashMap<PathBuf, u32>, // path -> file_handle for quick lookup
    pub last_compaction_epoch: u64,
}

impl TombstoneRegistry {
    pub fn new() -> Self {
        Self {
            tombstones: HashMap::new(),
            path_to_handle: HashMap::new(),
            last_compaction_epoch: 0,
        }
    }

    /// Load tombstone registry from file, or create new if doesn't exist
    pub fn load_or_create(registry_path: &Path) -> Result<Self> {
        if registry_path.exists() {
            Self::read_from_file(registry_path)
        } else {
            Ok(Self::new())
        }
    }

    pub fn read_from_file(path: &Path) -> Result<Self> {
        let content = fs::read_to_string(path)
            .context("Failed to read tombstone registry")?;
        let registry = serde_json::from_str(&content)
            .context("Failed to parse tombstone registry")?;
        Ok(registry)
    }

    pub fn write_to_file(&self, path: &Path) -> Result<()> {
        let json = serde_json::to_string_pretty(self)
            .context("Failed to serialize tombstone registry")?;
        fs::write(path, json)
            .context("Failed to write tombstone registry")?;
        Ok(())
    }

    /// Add a tombstone for a deleted file
    pub fn add_tombstone(&mut self, tombstone: Tombstone) {
        let file_handle = tombstone.file_handle;
        let path = tombstone.path.clone();
        
        self.tombstones.insert(file_handle, tombstone);
        self.path_to_handle.insert(path, file_handle);
    }

    /// Check if a file handle is tombstoned (deleted)
    pub fn is_tombstoned(&self, file_handle: u32) -> bool {
        self.tombstones.contains_key(&file_handle)
    }

    /// Check if a file path is tombstoned (deleted)
    pub fn is_path_tombstoned(&self, path: &Path) -> bool {
        self.path_to_handle.contains_key(path)
    }

    /// Get tombstone by file handle
    pub fn get_tombstone(&self, file_handle: u32) -> Option<&Tombstone> {
        self.tombstones.get(&file_handle)
    }

    /// Get all tombstones created since a specific epoch
    pub fn get_tombstones_since_epoch(&self, epoch: u64) -> Vec<&Tombstone> {
        self.tombstones
            .values()
            .filter(|t| t.deleted_epoch >= epoch)
            .collect()
    }

    /// Remove tombstones older than specified epoch (for compaction)
    pub fn compact_tombstones_before_epoch(&mut self, epoch: u64) -> Vec<Tombstone> {
        let to_remove: Vec<u32> = self.tombstones
            .iter()
            .filter(|(_, t)| t.deleted_epoch < epoch)
            .map(|(handle, _)| *handle)
            .collect();

        let mut removed = Vec::new();
        for handle in to_remove {
            if let Some(tombstone) = self.tombstones.remove(&handle) {
                self.path_to_handle.remove(&tombstone.path);
                removed.push(tombstone);
            }
        }

        self.last_compaction_epoch = epoch;
        removed
    }

    /// Get statistics about tombstones
    pub fn stats(&self) -> TombstoneStats {
        let total_count = self.tombstones.len();
        let oldest_epoch = self.tombstones
            .values()
            .map(|t| t.deleted_epoch)
            .min()
            .unwrap_or(0);
        let newest_epoch = self.tombstones
            .values()
            .map(|t| t.deleted_epoch)
            .max()
            .unwrap_or(0);

        TombstoneStats {
            total_count,
            oldest_epoch,
            newest_epoch,
            last_compaction_epoch: self.last_compaction_epoch,
        }
    }
}

#[derive(Debug)]
pub struct TombstoneStats {
    pub total_count: usize,
    pub oldest_epoch: u64,
    pub newest_epoch: u64,
    pub last_compaction_epoch: u64,
}

/// Tombstone manager for high-level operations
pub struct TombstoneManager {
    collection_path: PathBuf,
    registry_path: PathBuf,
}

impl TombstoneManager {
    pub fn new(collection_path: &Path) -> Self {
        let registry_path = collection_path.join("index").join("tombstones.json");
        Self {
            collection_path: collection_path.to_path_buf(),
            registry_path,
        }
    }

    /// Load the tombstone registry
    pub fn load_registry(&self) -> Result<TombstoneRegistry> {
        TombstoneRegistry::load_or_create(&self.registry_path)
    }

    /// Save the tombstone registry
    pub fn save_registry(&self, registry: &TombstoneRegistry) -> Result<()> {
        // Ensure index directory exists
        if let Some(parent) = self.registry_path.parent() {
            fs::create_dir_all(parent)?;
        }
        registry.write_to_file(&self.registry_path)
    }

    /// Mark a file as deleted by adding a tombstone
    pub fn mark_file_deleted(
        &self,
        file_handle: u32,
        path: PathBuf,
        current_epoch: u64,
        segment_id: u32,
        segment_offset: u64,
    ) -> Result<()> {
        let mut registry = self.load_registry()?;
        
        let tombstone = Tombstone::new(
            file_handle,
            path,
            current_epoch,
            segment_id,
            segment_offset,
        );
        
        registry.add_tombstone(tombstone);
        self.save_registry(&registry)?;
        
        Ok(())
    }

    /// Check if search results should filter out tombstoned files
    pub fn filter_live_handles(&self, handles: Vec<u32>) -> Result<Vec<u32>> {
        let registry = self.load_registry()?;
        
        let live_handles = handles
            .into_iter()
            .filter(|h| !registry.is_tombstoned(*h))
            .collect();
        
        Ok(live_handles)
    }

    /// Get compaction candidates (old tombstones that can be cleaned up)
    pub fn get_compaction_candidates(&self, before_epoch: u64) -> Result<Vec<Tombstone>> {
        let registry = self.load_registry()?;
        let candidates = registry.get_tombstones_since_epoch(0)
            .into_iter()
            .filter(|t| t.deleted_epoch < before_epoch)
            .cloned()
            .collect();
        
        Ok(candidates)
    }

    /// Perform tombstone compaction (remove old tombstones)
    pub fn compact_tombstones(&self, before_epoch: u64) -> Result<usize> {
        let mut registry = self.load_registry()?;
        let removed = registry.compact_tombstones_before_epoch(before_epoch);
        let count = removed.len();
        
        self.save_registry(&registry)?;
        
        Ok(count)
    }

    /// Get tombstone statistics
    pub fn get_stats(&self) -> Result<TombstoneStats> {
        let registry = self.load_registry()?;
        Ok(registry.stats())
    }
}