reductstore 1.20.8

// Copyright 2021-2026 ReductSoftware UG
// Licensed under the Apache License, Version 2.0

use crate::cfg::InstanceRole;
use crate::core::file_cache::FILE_CACHE;
use crate::storage::block_manager::block_index::BlockIndex;
use crate::storage::block_manager::{
    BlockManager, BLOCK_INDEX_FILE, COMPRESSED_DATA_FILE_EXT, COMPRESSED_DESCRIPTOR_FILE_EXT,
    DATA_FILE_EXT, DESCRIPTOR_FILE_EXT,
};
use crate::storage::proto::block_index::Block as BlockEntry;
use log::warn;
use reduct_base::error::ReductError;
use std::collections::HashMap;
use std::path::PathBuf;
use std::time::Instant;

pub(in crate::storage) struct ReplicaIndexReload {
    entry_path: PathBuf,
    index_path: PathBuf,
    previous_state: HashMap<u64, BlockEntry>,
}

impl ReplicaIndexReload {
    pub(in crate::storage) async fn load_updated_index(&self) -> Result<BlockIndex, ReductError> {
        FILE_CACHE
            .invalidate_local_cache_file(&self.index_path)
            .await?;

        let updated_index = BlockIndex::try_load(self.index_path.clone()).await?;

        let mut first_err = None;

        for (block_id, new_block_info) in updated_index.info().iter() {
            if let Some(previous_block_info) = self.previous_state.get(block_id) {
                if previous_block_info.crc64 != new_block_info.crc64 {
                    let extensions = [
                        DESCRIPTOR_FILE_EXT,
                        DATA_FILE_EXT,
                        COMPRESSED_DESCRIPTOR_FILE_EXT,
                        COMPRESSED_DATA_FILE_EXT,
                    ];

                    for ext in extensions {
                        let path = self.entry_path.join(format!("{}{}", block_id, ext));
                        if let Err(err) = FILE_CACHE.invalidate_local_cache_file(&path).await {
                            warn!(
                                "Failed to invalidate replica cache file {:?}: {}",
                                path, err
                            );
                            if first_err.is_none() {
                                first_err = Some(err);
                            }
                        }
                    }
                }
            }
        }

        if let Some(err) = first_err {
            Err(err)
        } else {
            Ok(updated_index)
        }
    }
}

impl BlockManager {
    #[cfg(test)]
    pub(super) async fn reload_if_readonly(&mut self) -> Result<(), ReductError> {
        // Replica index refresh is driven by the launcher background task.
        Ok(())
    }

    pub(in crate::storage) fn prepare_replica_index_reload(&self) -> Option<ReplicaIndexReload> {
        if self.cfg.role != InstanceRole::Replica {
            return None;
        }

        Some(ReplicaIndexReload {
            entry_path: self.path.clone(),
            index_path: self.path.join(BLOCK_INDEX_FILE),
            previous_state: self.block_index.info().clone(),
        })
    }

    pub(in crate::storage) fn apply_replica_index_reload(&mut self, updated_index: BlockIndex) {
        self.block_index = updated_index;
        self.block_cache.clear();
        self.last_replica_sync = Instant::now();
    }
}

#[cfg(test)]
mod tests {
    // test reloading in read-only mode

    use super::*;
    use crate::cfg::storage_engine::StorageEngineConfig;
    use crate::cfg::Cfg;
    use crate::storage::block_manager::block::Block;
    use crate::storage::block_manager::block_index::BlockIndex;
    use crate::storage::block_manager::{BlockManager, BLOCK_INDEX_FILE};
    use crate::storage::proto::Block as BlockProto;
    use prost::Message;
    use reduct_base::error::ErrorCode;
    use rstest::{fixture, rstest};
    use std::path::PathBuf;
    use std::sync::Arc;
    use std::time::Duration;
    use tempfile::tempdir;

    #[rstest]
    #[tokio::test]
    async fn test_reload_if_readonly_is_noop(#[future] path: PathBuf) {
        let path = path.await;
        let cfg = Cfg {
            role: InstanceRole::Replica,
            data_path: path.clone(),
            engine_config: StorageEngineConfig {
                replica_update_interval: Duration::from_millis(100),
                ..Default::default()
            },
            ..Default::default()
        };

        let index = BlockIndex::new(path.join(BLOCK_INDEX_FILE));
        index.save().await.unwrap();
        let mut block_manager = BlockManager::build(
            path.clone(),
            index,
            "bucket".to_string(),
            "entry".to_string(),
            Arc::new(cfg.clone()),
            Default::default(),
        )
        .await
        .unwrap();

        // change index on disc
        let mut new_index = BlockIndex::try_load(path.join(BLOCK_INDEX_FILE))
            .await
            .unwrap();

        let block = Block::new(1);
        new_index.insert_or_update(block);
        new_index.save().await.unwrap();

        // wait for the replica update interval to pass
        tokio::time::sleep(Duration::from_millis(150)).await;
        block_manager.reload_if_readonly().await.unwrap();

        assert!(block_manager.block_index.info().get(&1).is_none());
    }

    #[rstest]
    #[tokio::test]
    async fn test_background_replica_index_reload(#[future] path: PathBuf) {
        let path = path.await;
        let cfg = Cfg {
            role: InstanceRole::Replica,
            data_path: path.clone(),
            engine_config: StorageEngineConfig {
                replica_update_interval: Duration::from_millis(100),
                ..Default::default()
            },
            ..Default::default()
        };

        let index = BlockIndex::new(path.join(BLOCK_INDEX_FILE));
        index.save().await.unwrap();
        let mut block_manager = BlockManager::build(
            path.clone(),
            index,
            "bucket".to_string(),
            "entry".to_string(),
            Arc::new(cfg.clone()),
            Default::default(),
        )
        .await
        .unwrap();

        let mut new_index = BlockIndex::try_load(path.join(BLOCK_INDEX_FILE))
            .await
            .unwrap();
        new_index.insert_or_update(Block::new(1));
        new_index.save().await.unwrap();

        let reload = block_manager.prepare_replica_index_reload().unwrap();
        let updated_index = reload.load_updated_index().await.unwrap();
        block_manager.apply_replica_index_reload(updated_index);

        assert!(block_manager.block_index.info().get(&1).is_some());
    }

    #[rstest]
    #[tokio::test(flavor = "current_thread")]
    async fn test_reload_if_readonly_discards_cache_on_crc_change(#[future] path: PathBuf) {
        let path = path.await;
        let entry_path = path.join("bucket").join("entry");

        let cfg = Cfg {
            role: InstanceRole::Replica,
            data_path: path.clone(),
            engine_config: StorageEngineConfig {
                replica_update_interval: Duration::from_millis(50),
                ..Default::default()
            },
            ..Default::default()
        };

        let index_path = entry_path.join(BLOCK_INDEX_FILE);
        let mut index = BlockIndex::new(index_path.clone());
        index.insert_or_update_with_crc(Block::new(1), 1);
        index.save().await.unwrap();

        let mut block_manager = BlockManager::build(
            entry_path.clone(),
            index,
            "bucket".to_string(),
            "entry".to_string(),
            Arc::new(cfg.clone()),
            Default::default(),
        )
        .await
        .unwrap();

        let mut updated_index = BlockIndex::new(index_path.clone());
        updated_index.insert_or_update_with_crc(Block::new(1), 2);
        updated_index.save().await.unwrap();

        let reload = block_manager.prepare_replica_index_reload().unwrap();
        let updated_index = reload.load_updated_index().await.unwrap();
        block_manager.apply_replica_index_reload(updated_index);

        assert_eq!(
            block_manager.block_index.info().get(&1).unwrap().crc64,
            Some(2)
        );
    }

    #[rstest]
    #[tokio::test(flavor = "current_thread")]
    async fn test_reload_if_readonly_discards_compressed_cache_on_crc_change(
        #[future] path: PathBuf,
    ) {
        let path = path.await;
        let entry_path = path.join("bucket").join("entry");

        let cfg = Cfg {
            role: InstanceRole::Replica,
            data_path: path.clone(),
            engine_config: StorageEngineConfig {
                replica_update_interval: Duration::from_millis(50),
                ..Default::default()
            },
            ..Default::default()
        };

        let index_path = entry_path.join(BLOCK_INDEX_FILE);
        let mut index = BlockIndex::new(index_path.clone());
        index.insert_or_update_with_crc(Block::new(1), 1);
        index.save().await.unwrap();

        std::fs::write(entry_path.join("1.meta.zst"), b"old-meta").unwrap();
        std::fs::write(entry_path.join("1.blk.zst"), b"old-data").unwrap();

        FILE_CACHE
            .read(&entry_path.join("1.meta.zst"), std::io::SeekFrom::Start(0))
            .await
            .unwrap();
        FILE_CACHE
            .read(&entry_path.join("1.blk.zst"), std::io::SeekFrom::Start(0))
            .await
            .unwrap();

        let mut block_manager = BlockManager::build(
            entry_path.clone(),
            index,
            "bucket".to_string(),
            "entry".to_string(),
            Arc::new(cfg),
            Default::default(),
        )
        .await
        .unwrap();

        let mut updated_index = BlockIndex::new(index_path.clone());
        updated_index.insert_or_update_with_crc(Block::new(1), 2);
        updated_index.save().await.unwrap();

        std::fs::write(entry_path.join("1.meta.zst"), b"new-meta").unwrap();
        std::fs::write(entry_path.join("1.blk.zst"), b"new-data").unwrap();

        let reload = block_manager.prepare_replica_index_reload().unwrap();
        let updated_index = reload.load_updated_index().await.unwrap();
        block_manager.apply_replica_index_reload(updated_index);

        let mut meta = FILE_CACHE
            .read(&entry_path.join("1.meta.zst"), std::io::SeekFrom::Start(0))
            .await
            .unwrap();
        let mut meta_content = vec![];
        use std::io::Read;
        meta.read_to_end(&mut meta_content).unwrap();

        let mut data = FILE_CACHE
            .read(&entry_path.join("1.blk.zst"), std::io::SeekFrom::Start(0))
            .await
            .unwrap();
        let mut data_content = vec![];
        data.read_to_end(&mut data_content).unwrap();

        assert_eq!(meta_content, b"new-meta");
        assert_eq!(data_content, b"new-data");
    }

    #[rstest]
    #[tokio::test(flavor = "current_thread")]
    async fn test_load_block_missing_descriptor_on_replica_returns_too_early(
        #[future] path: PathBuf,
    ) {
        let path = path.await;
        let entry_path = path.join("bucket").join("entry");

        let cfg = Cfg {
            role: InstanceRole::Replica,
            data_path: path.clone(),
            ..Default::default()
        };

        let index_path = entry_path.join(BLOCK_INDEX_FILE);
        let mut index = BlockIndex::new(index_path);
        index.insert_or_update(Block::new(1));
        index.save().await.unwrap();

        let mut block_manager = BlockManager::build(
            entry_path,
            index,
            "bucket".to_string(),
            "entry".to_string(),
            Arc::new(cfg),
            Default::default(),
        )
        .await
        .unwrap();

        let err = block_manager.load_block(1).await.err().unwrap();
        assert_eq!(err.status(), ErrorCode::TooEarly);
        assert!(block_manager.index().get_block(1).is_none());
    }

    #[rstest]
    #[tokio::test(flavor = "current_thread")]
    async fn test_load_block_crc_mismatch_on_replica_returns_too_early(#[future] path: PathBuf) {
        let path = path.await;
        let entry_path = path.join("bucket").join("entry");

        let cfg = Cfg {
            role: InstanceRole::Replica,
            data_path: path.clone(),
            ..Default::default()
        };

        let index_path = entry_path.join(BLOCK_INDEX_FILE);
        let mut index = BlockIndex::new(index_path);
        index.insert_or_update_with_crc(Block::new(1), 1);
        index.save().await.unwrap();

        let descriptor = BlockProto::from(Block::new(1)).encode_to_vec();
        std::fs::write(entry_path.join("1.meta"), descriptor).unwrap();

        let mut block_manager = BlockManager::build(
            entry_path,
            index,
            "bucket".to_string(),
            "entry".to_string(),
            Arc::new(cfg),
            Default::default(),
        )
        .await
        .unwrap();

        let err = block_manager.load_block(1).await.err().unwrap();
        assert_eq!(err.status(), ErrorCode::TooEarly);
        assert!(block_manager.index().get_block(1).is_none());
    }

    #[fixture]
    async fn path() -> PathBuf {
        let dir = tempdir().unwrap().keep();
        tokio::fs::create_dir_all(dir.join("bucket").join("entry"))
            .await
            .unwrap();

        dir
    }
}