tellaro-query-language 3.0.1

//! Cache for field mappings resolved from a live index pattern.
//!
//! Resolving mappings is a network round trip, and the search-proxy role builds
//! a fresh [`TqlExecutor`](super::executor::TqlExecutor) per request — so
//! without a cache that lives ABOVE the executor, every proxied TQL search pays
//! a full `_field_caps` call. The cache is therefore owned by the long-lived
//! caller (the role), not by the executor.
//!
//! Three properties matter more than the hit rate:
//!
//! 1. **Failures are cached separately, and briefly.** A failed fetch must not be
//!    retried on every query of a burst, and it must never be confused with a
//!    successful fetch that returned nothing. Those are different facts:
//!    "the cluster is unreachable" and "the index declares no such field"
//!    collapse into the same permissive answer if stored in one map. This
//!    mirrors `IndexMappingResolver` in the Python implementation
//!    (`src/tql/opensearch_mappings.py`), which keeps `_raw_by_index` and
//!    `_failure_by_index` as two dicts for exactly this reason.
//! 2. **Single-flight.** A cold key plus a burst of concurrent searches is
//!    precisely when a fan-out to the cluster is least wanted. `moka`'s
//!    `try_get_with` dedupes concurrent initializers per key.
//! 3. **Short TTL, because wildcards move.** `logs-*` resolves to a different
//!    set of indices tomorrow, and a field added by dynamic mapping is invisible
//!    until the entry expires. There is no invalidation signal from OpenSearch,
//!    so the TTL is the only mechanism — see also
//!    [`MappingCache::invalidate`] for the miss-triggered refresh.

use super::error::{OpenSearchError, Result};
use super::field_mappings::FieldMappings;
use moka::future::Cache;
use std::future::Future;
use std::sync::Arc;
use std::time::Duration;

/// How long a successfully resolved mapping set is reused.
///
/// Minutes, not hours. The failure this bounds is a field arriving via dynamic
/// mapping: a user queries it, the stale entry does not have it, and
/// `get_query_field` passes the bare name through unmapped. A long TTL feels
/// efficient right up until someone spends an afternoon on a field the cache
/// did not know about.
pub const DEFAULT_TTL: Duration = Duration::from_secs(300);

/// How long a failed resolution is remembered.
///
/// Short enough to recover quickly once a cluster comes back, long enough that a
/// broken cluster is not hammered once per query.
pub const DEFAULT_NEGATIVE_TTL: Duration = Duration::from_secs(10);

/// Upper bound on distinct index patterns held.
///
/// The key space is operator-controlled — a custom data source is whatever the
/// user typed — so it has to be bounded.
pub const DEFAULT_MAX_ENTRIES: u64 = 512;

/// Cache of field mappings, keyed by index pattern.
///
/// Cheap to clone; clones share the same underlying store.
#[derive(Clone)]
pub struct MappingCache {
    resolved: Cache<String, Arc<FieldMappings>>,
    failures: Cache<String, Arc<str>>,
}

impl MappingCache {
    /// Build a cache with the default TTLs and bound.
    pub fn new() -> Self {
        Self::with_config(DEFAULT_TTL, DEFAULT_NEGATIVE_TTL, DEFAULT_MAX_ENTRIES)
    }

    /// Build a cache with explicit TTLs and bound.
    ///
    /// Note that staleness COMPOUNDS across tiers: when a backend also caches
    /// these, the worst case a user sees is this TTL plus that one. Budget the
    /// total once and split it, rather than choosing each independently.
    pub fn with_config(ttl: Duration, negative_ttl: Duration, max_entries: u64) -> Self {
        Self {
            resolved: Cache::builder()
                .max_capacity(max_entries)
                .time_to_live(ttl)
                .build(),
            failures: Cache::builder()
                .max_capacity(max_entries)
                .time_to_live(negative_ttl)
                .build(),
        }
    }

    /// Resolve `index_pattern`, fetching via `fetch` on a miss.
    ///
    /// A cached failure short-circuits without calling `fetch`. Concurrent
    /// callers for the same key share one `fetch`.
    pub async fn get_or_fetch<F, Fut>(
        &self,
        index_pattern: &str,
        fetch: F,
    ) -> Result<Arc<FieldMappings>>
    where
        F: FnOnce() -> Fut,
        Fut: Future<Output = Result<FieldMappings>>,
    {
        if let Some(reason) = self.failures.get(index_pattern).await {
            return Err(OpenSearchError::MappingError(reason.to_string()));
        }

        let result = self
            .resolved
            .try_get_with_by_ref(index_pattern, async move { fetch().await.map(Arc::new) })
            .await;

        match result {
            Ok(mappings) => Ok(mappings),
            Err(arc_err) => {
                // moka hands back an Arc of the initializer's error, shared by
                // every caller that waited on this single flight.
                let reason = arc_err.to_string();
                self.failures
                    .insert(index_pattern.to_string(), Arc::from(reason.as_str()))
                    .await;
                Err(OpenSearchError::MappingError(reason))
            }
        }
    }

    /// Drop any entry for `index_pattern`, positive or negative.
    ///
    /// This is the miss-triggered refresh: when a query names a field the cached
    /// mappings do not contain, that is a strong, cheap signal the entry is
    /// stale — a field added by dynamic mapping is the ordinary way to get
    /// there. Refreshing once beats waiting out the TTL, and it costs nothing on
    /// the common path where every field is already known.
    pub async fn invalidate(&self, index_pattern: &str) {
        self.resolved.invalidate(index_pattern).await;
        self.failures.invalidate(index_pattern).await;
    }

    /// Number of resolved entries currently held. Test/diagnostic use.
    pub async fn resolved_len(&self) -> u64 {
        self.resolved.run_pending_tasks().await;
        self.resolved.entry_count()
    }

    /// Number of remembered failures. Test/diagnostic use.
    pub async fn failure_len(&self) -> u64 {
        self.failures.run_pending_tasks().await;
        self.failures.entry_count()
    }
}

impl Default for MappingCache {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::opensearch::field_mappings::{FieldMapping, FieldType};
    use std::collections::HashMap;
    use std::sync::atomic::{AtomicUsize, Ordering};

    fn mappings_with(field: &str) -> FieldMappings {
        let mut m = FieldMappings::new();
        m.add_mapping(
            field.to_string(),
            FieldMapping {
                field_type: FieldType::Keyword,
                subfields: HashMap::new(),
            },
        );
        m
    }

    #[tokio::test]
    async fn a_hit_does_not_refetch() {
        let cache = MappingCache::new();
        let calls = Arc::new(AtomicUsize::new(0));

        for _ in 0..3 {
            let calls = Arc::clone(&calls);
            let got = cache
                .get_or_fetch("logs-*", || async move {
                    calls.fetch_add(1, Ordering::SeqCst);
                    Ok(mappings_with("host.name"))
                })
                .await
                .unwrap();
            assert_eq!(got.len(), 1);
        }
        assert_eq!(
            calls.load(Ordering::SeqCst),
            1,
            "cached entry must not refetch"
        );
    }

    #[tokio::test]
    async fn distinct_patterns_are_distinct_entries() {
        let cache = MappingCache::new();
        let calls = Arc::new(AtomicUsize::new(0));
        for pattern in ["logs-a", "logs-b"] {
            let calls = Arc::clone(&calls);
            cache
                .get_or_fetch(pattern, || async move {
                    calls.fetch_add(1, Ordering::SeqCst);
                    Ok(mappings_with("f"))
                })
                .await
                .unwrap();
        }
        assert_eq!(calls.load(Ordering::SeqCst), 2);
    }

    /// The point of the separate failure map: a failure must not be retried per
    /// query, and must not be stored as a successful empty result.
    #[tokio::test]
    async fn a_failure_is_remembered_and_not_retried() {
        let cache = MappingCache::new();
        let calls = Arc::new(AtomicUsize::new(0));

        for _ in 0..3 {
            let calls = Arc::clone(&calls);
            let err = cache
                .get_or_fetch("logs-*", || async move {
                    calls.fetch_add(1, Ordering::SeqCst);
                    Err(OpenSearchError::MappingError("cluster unreachable".into()))
                })
                .await
                .unwrap_err();
            assert!(err.to_string().contains("cluster unreachable"));
        }
        assert_eq!(
            calls.load(Ordering::SeqCst),
            1,
            "a cached failure must not refetch"
        );
        assert_eq!(cache.failure_len().await, 1);
        assert_eq!(
            cache.resolved_len().await,
            0,
            "a failure must never land in the resolved map, where it would read as empty mappings"
        );
    }

    #[tokio::test]
    async fn a_failure_expires_and_then_recovers() {
        let cache = MappingCache::with_config(
            Duration::from_secs(300),
            Duration::from_millis(50),
            DEFAULT_MAX_ENTRIES,
        );
        let err = cache
            .get_or_fetch("logs-*", || async {
                Err(OpenSearchError::MappingError("down".into()))
            })
            .await;
        assert!(err.is_err());

        tokio::time::sleep(Duration::from_millis(120)).await;

        let ok = cache
            .get_or_fetch("logs-*", || async { Ok(mappings_with("host.name")) })
            .await;
        assert!(
            ok.is_ok(),
            "the negative entry must expire so the cluster is retried"
        );
    }

    #[tokio::test]
    async fn a_resolved_entry_expires() {
        let cache = MappingCache::with_config(
            Duration::from_millis(50),
            DEFAULT_NEGATIVE_TTL,
            DEFAULT_MAX_ENTRIES,
        );
        let calls = Arc::new(AtomicUsize::new(0));
        for _ in 0..2 {
            let calls = Arc::clone(&calls);
            cache
                .get_or_fetch("logs-*", || async move {
                    calls.fetch_add(1, Ordering::SeqCst);
                    Ok(mappings_with("f"))
                })
                .await
                .unwrap();
            tokio::time::sleep(Duration::from_millis(80)).await;
        }
        assert_eq!(
            calls.load(Ordering::SeqCst),
            2,
            "a wildcard's index set moves; the entry has to expire"
        );
    }

    /// Single-flight: a burst on a cold key must produce ONE fetch.
    #[tokio::test]
    async fn concurrent_callers_share_one_fetch() {
        let cache = MappingCache::new();
        let calls = Arc::new(AtomicUsize::new(0));

        let mut handles = Vec::new();
        for _ in 0..16 {
            let cache = cache.clone();
            let calls = Arc::clone(&calls);
            handles.push(tokio::spawn(async move {
                cache
                    .get_or_fetch("logs-*", || async move {
                        calls.fetch_add(1, Ordering::SeqCst);
                        tokio::time::sleep(Duration::from_millis(30)).await;
                        Ok(mappings_with("host.name"))
                    })
                    .await
                    .map(|m| m.len())
            }));
        }
        for h in handles {
            assert_eq!(h.await.unwrap().unwrap(), 1);
        }
        assert_eq!(
            calls.load(Ordering::SeqCst),
            1,
            "16 concurrent searches on one pattern must cost one _field_caps call"
        );
    }

    #[tokio::test]
    async fn invalidate_forces_a_refetch() {
        let cache = MappingCache::new();
        let calls = Arc::new(AtomicUsize::new(0));
        for _ in 0..2 {
            let calls = Arc::clone(&calls);
            cache
                .get_or_fetch("logs-*", || async move {
                    calls.fetch_add(1, Ordering::SeqCst);
                    Ok(mappings_with("f"))
                })
                .await
                .unwrap();
            cache.invalidate("logs-*").await;
        }
        assert_eq!(calls.load(Ordering::SeqCst), 2);
    }

    #[tokio::test]
    async fn invalidate_clears_a_remembered_failure() {
        let cache = MappingCache::new();
        let _ = cache
            .get_or_fetch("logs-*", || async {
                Err(OpenSearchError::MappingError("down".into()))
            })
            .await;
        cache.invalidate("logs-*").await;
        let ok = cache
            .get_or_fetch("logs-*", || async { Ok(mappings_with("f")) })
            .await;
        assert!(
            ok.is_ok(),
            "the miss-triggered refresh has to pierce the negative entry too, \
             or the refresh is wasted"
        );
    }
}