tantivy-cache 0.7.0

use std::path::PathBuf;

use eyre::{Result, ensure};
use tantivy::{Directory, Index, ReloadPolicy, directory::RamDirectory};

use super::{
    dir::FailingDirectory,
    utils::{create_multi_segment_index, redis_pool},
};
use crate::CachingDirectory;

/// [`Warmer::with_concurrency`] clamps a concurrency of `0` up to `1` (a
/// `Semaphore` with zero permits would otherwise deadlock).
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn warmer_clamps_concurrency_to_at_least_one() -> Result<()> {
    let pool = redis_pool().await?;
    let dir = CachingDirectory::new(RamDirectory::create(), pool);

    assert_eq!(dir.warmer().with_concurrency(0).concurrency(), 1);
    assert_eq!(dir.warmer().with_concurrency(4).concurrency(), 4);

    Ok(())
}

/// Warms an entire index, then proves the cache holds everything needed to open
/// a segment reader by re-opening the index over a [`FailingDirectory`] (which
/// panics on any read) backed by the same Redis pool.
///
/// If warming missed a footer, opening the reader would fall through to the
/// failing directory and panic.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn warm_index_populates_cache() -> Result<()> {
    let pool = redis_pool().await?;
    let (index, ram, dir) = create_multi_segment_index(pool.clone())?;

    // Merging is disabled, so we should have one segment per commit.
    let metas = index.searchable_segment_metas()?;
    ensure!(
        metas.len() >= 4,
        "expected several segments, got {}",
        metas.len(),
    );

    // Warm every file of every segment in parallel.
    dir.warmer().with_concurrency(8).index(&index).await?;

    // Fresh `CachingDirectory` (cold in-memory cache) over a `FailingDirectory`,
    // sharing the same Redis pool. Opening a reader reads every cached footer; a
    // miss would panic in the failing directory.
    let dir2 = CachingDirectory::new(FailingDirectory(ram), pool.clone());
    let index2 = Index::open(dir2)?;

    let _reader: tantivy::IndexReader = index2
        .reader_builder()
        .reload_policy(ReloadPolicy::Manual)
        .try_into()?;

    Ok(())
}

/// After warming an index, every existing segment file must have its footer
/// loaded *in memory* (not merely persisted to Redis), so reads avoid both the
/// directory and Redis.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn warm_index_loads_in_memory_footers() -> Result<()> {
    let pool = redis_pool().await?;
    let (index, _ram, dir) = create_multi_segment_index(pool.clone())?;

    dir.warmer().with_concurrency(4).index(&index).await?;

    let metas = index.searchable_segment_metas()?;
    let mut checked = 0usize;
    for meta in &metas {
        for path in meta.list_files() {
            // Some listed files are optional and may not exist (e.g. `.del`).
            if !dir.exists(&path).unwrap_or(false) {
                continue;
            }

            let cached = dir
                .cache
                .get_async(&path)
                .await
                .ok_or_else(|| eyre::eyre!("{path:?} missing from the in-memory cache"))?;
            ensure!(
                cached.get().data().is_some(),
                "footer data should be loaded in memory for {path:?}",
            );

            checked += 1;
        }
    }

    // At minimum the always-present components (.idx, .term, .fast, .store,
    // .fieldnorm) of each segment should have been checked.
    ensure!(
        checked >= metas.len() * 5,
        "expected to check several files per segment, only checked {checked}",
    );

    Ok(())
}

/// Warming a list of files that do not exist is a no-op rather than an error.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn warm_files_ignores_missing_files() -> Result<()> {
    let pool = redis_pool().await?;
    let (_index, _ram, dir) = create_multi_segment_index(pool.clone())?;

    let missing = [
        PathBuf::from("does-not-exist.idx"),
        PathBuf::from("also-missing.store"),
    ];
    dir.warmer().with_concurrency(2).files(missing).await?;

    Ok(())
}