sail-rs 0.4.0

//! Client-scoped memoization of image readiness.
//!
//! Every [`crate::Client::build_spec_with_timeout`] and
//! [`crate::Client::build_image_definition`] call resolves the same
//! content-addressed build, so one successful readiness result can serve
//! every later call with the same spec instead of re-submitting the build
//! and re-polling its status. The cache lives in the core so every language
//! wrapper shares one implementation, keyed by the hash of the spec's
//! canonical JSON plus the caller's timeout (a caller only ever joins a
//! build started with its own bound). Successes are re-verified after a
//! refresh window because the backend garbage-collects idle unreferenced
//! images (`SAILBOX_IMAGE_GC_TTL`) and a create against a collected image
//! fails without rebuilding it; failures are evicted so the next call
//! retries.
//!
//! In-flight builds are held weakly: the build future captures a clone of
//! the client whose cache stores it, so a strong reference here would form
//! a cycle that leaks the client once every caller abandons the build.
//! Waiters hold the strong handles; when the last one drops, the build
//! drops with it and the next caller simply starts a fresh one. Settled
//! successes are stored as plain values, so they survive without retaining
//! any future.

use std::collections::{HashMap, VecDeque};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use futures::future::{BoxFuture, Shared, WeakShared};

use crate::error::SailError;
use crate::imagebuild::ImageBuild;

/// How long a cached successful build is served before the server is
/// re-verified.
pub(crate) const IMAGE_READY_REFRESH: Duration = Duration::from_hours(1);

/// Leak guard for long-lived processes streaming many distinct specs or
/// timeouts: oldest entries are dropped first and simply re-verify if used
/// again. In-flight waiters hold their shared future directly, so eviction
/// never interrupts them.
pub(crate) const MAX_READY_ENTRIES: usize = 64;

/// One build shared by every concurrent caller with the same key. The error
/// arm is `Arc` because [`SailError`] is not `Clone`.
pub(crate) type SharedBuild = Shared<BoxFuture<'static, Result<ImageBuild, Arc<SailError>>>>;

type WeakBuild = WeakShared<BoxFuture<'static, Result<ImageBuild, Arc<SailError>>>>;

/// Cache key: the caller's timeout plus the sha256 of the spec's canonical
/// (key-sorted) JSON.
pub(crate) type CacheKey = (Duration, String);

struct Entry {
    /// Distinguishes this entry from a later one under the same key, so a
    /// settle or eviction for a superseded entry never touches its
    /// replacement.
    id: u64,
    /// When this entry's build was started. Invalidation scopes by it: a
    /// build started before a failed create began may predate the server
    /// change that failed the create, however recently it settled.
    started_at: Instant,
    /// Whether this build was started by stale-create recovery. A recovery
    /// build begins after a conflict was discovered, and therefore after the
    /// server change behind it, so while it is in flight it stays joinable
    /// through later invalidations.
    recovery: bool,
    state: EntryState,
}

enum EntryState {
    /// A build some caller is currently awaiting. Weak: see the module docs.
    InFlight(WeakBuild),
    /// A completed build, served until the refresh window lapses.
    Ready {
        build: ImageBuild,
        resolved_at: Instant,
    },
}

/// What [`ImageReadyCache::join_or_lead`] hands the caller. A `Pending`
/// caller elected to lead must drive the build, so dropping it loses work.
#[must_use]
pub(crate) enum Joined {
    /// A completed build fresh enough to serve directly.
    Ready(ImageBuild),
    /// A build to await; `led` reports whether this caller started it.
    Pending { build: SharedBuild, led: bool },
}

struct State {
    entries: HashMap<CacheKey, Entry>,
    /// Insertion order for oldest-first eviction past [`MAX_READY_ENTRIES`].
    order: VecDeque<(CacheKey, u64)>,
    next_id: u64,
    refresh_window: Duration,
}

pub(crate) struct ImageReadyCache {
    state: Mutex<State>,
}

impl ImageReadyCache {
    pub(crate) fn new() -> ImageReadyCache {
        ImageReadyCache {
            state: Mutex::new(State {
                entries: HashMap::new(),
                order: VecDeque::new(),
                next_id: 0,
                refresh_window: IMAGE_READY_REFRESH,
            }),
        }
    }

    #[cfg(any(test, feature = "test-fakes"))]
    pub(crate) fn set_refresh_window(&self, window: Duration) {
        self.state.lock().unwrap().refresh_window = window;
    }

    /// Serve the cached build for `key`, or start one. A fresh completed
    /// build is returned directly; a live in-flight build is joined; a miss,
    /// a stale success, or an abandoned in-flight build (every waiter
    /// dropped) invokes `make` with the new entry's id. `recovery` marks a
    /// build led here as stale-create recovery (see [`Entry::recovery`]);
    /// joining an existing entry leaves its marking unchanged.
    pub(crate) fn join_or_lead(
        &self,
        key: &CacheKey,
        recovery: bool,
        make: impl FnOnce(u64) -> SharedBuild,
    ) -> Joined {
        let mut state = self.state.lock().unwrap();
        if let Some(entry) = state.entries.get(key) {
            match &entry.state {
                EntryState::Ready { build, resolved_at }
                    if resolved_at.elapsed() <= state.refresh_window =>
                {
                    return Joined::Ready(build.clone());
                }
                EntryState::InFlight(weak) => {
                    if let Some(build) = weak.upgrade() {
                        return Joined::Pending { build, led: false };
                    }
                }
                EntryState::Ready { .. } => {}
            }
        }
        let id = state.next_id;
        state.next_id += 1;
        let build = make(id);
        let weak = build
            .downgrade()
            .expect("a build future cannot complete before it is first polled");
        state.entries.insert(
            key.clone(),
            Entry {
                id,
                started_at: Instant::now(),
                recovery,
                state: EntryState::InFlight(weak),
            },
        );
        state.order.push_back((key.clone(), id));
        while state.entries.len() > MAX_READY_ENTRIES {
            let Some((oldest_key, oldest_id)) = state.order.pop_front() else {
                break;
            };
            if state
                .entries
                .get(&oldest_key)
                .is_some_and(|entry| entry.id == oldest_id)
            {
                state.entries.remove(&oldest_key);
            }
        }
        // Replacements and failure evictions leave superseded records in the
        // queue; compact occasionally so repeated retries or refreshes of a
        // few keys cannot grow it without bound. `retain` keeps relative
        // order, so oldest-first eviction is unaffected.
        if state.order.len() > MAX_READY_ENTRIES * 2 {
            let State { entries, order, .. } = &mut *state;
            order.retain(|(key, id)| entries.get(key).is_some_and(|entry| entry.id == *id));
        }
        Joined::Pending { build, led: true }
    }

    /// Record that entry `id` under `key` completed successfully; the value
    /// serves later callers until the refresh window lapses. A superseded id
    /// is ignored.
    pub(crate) fn settle_success(&self, key: &CacheKey, id: u64, build: ImageBuild) {
        let mut state = self.state.lock().unwrap();
        if let Some(entry) = state.entries.get_mut(key) {
            if entry.id == id {
                entry.state = EntryState::Ready {
                    build,
                    resolved_at: Instant::now(),
                };
            }
        }
    }

    /// Drop the spec's suspect entries, across timeouts. Used when a create
    /// fails in a way that suggests the server-side image identity for the
    /// spec changed (e.g. a backend deploy bumped the canonical image schema
    /// version). Suspect means the build started before the failed create
    /// began (it may carry the old identity, even if it settled afterward),
    /// with one exception: an in-flight recovery build began after a conflict
    /// was discovered, and therefore after the change behind it, so it stays
    /// joinable and staggered stale creates converge on one shared rebuild.
    /// A settled recovery is not exempt: had its
    /// result been good, this create would not have failed. In-flight
    /// waiters hold their shared future directly, so dropping an entry never
    /// interrupts them.
    pub(crate) fn invalidate_spec_started_before(&self, spec_hash: &str, cutoff: Instant) {
        let mut state = self.state.lock().unwrap();
        state.entries.retain(|key, entry| {
            if key.1 != spec_hash || entry.started_at >= cutoff {
                return true;
            }
            entry.recovery && matches!(entry.state, EntryState::InFlight(_))
        });
    }

    /// Drop entry `id` under `key` after a failed build so the next caller
    /// retries. A superseded id is ignored.
    pub(crate) fn settle_failure(&self, key: &CacheKey, id: u64) {
        let mut state = self.state.lock().unwrap();
        if state.entries.get(key).is_some_and(|entry| entry.id == id) {
            state.entries.remove(key);
        }
    }

    #[cfg(test)]
    fn order_len(&self) -> usize {
        self.state.lock().unwrap().order.len()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use futures::FutureExt;

    fn ready_build(id: &str) -> ImageBuild {
        ImageBuild {
            image_id: id.to_string(),
            status: crate::imagebuild::ImageBuildStatus::Ready,
            error_message: String::new(),
        }
    }

    fn pending_build(id: &str) -> SharedBuild {
        let build = ready_build(id);
        async move { Ok(build) }.boxed().shared()
    }

    fn key(name: &str) -> CacheKey {
        (Duration::from_mins(1), name.to_string())
    }

    fn assert_led(joined: &Joined, want: bool) {
        match joined {
            Joined::Pending { led, .. } => assert_eq!(*led, want),
            Joined::Ready(_) => panic!("expected a pending build"),
        }
    }

    #[tokio::test]
    async fn joins_in_flight_entry_while_a_waiter_holds_it() {
        let cache = ImageReadyCache::new();
        let first = cache.join_or_lead(
            &key("a"),
            /* recovery */ false,
            |_| pending_build("img"),
        );
        assert_led(&first, true);
        let second = cache.join_or_lead(
            &key("a"),
            /* recovery */ false,
            |_| panic!("must join"),
        );
        assert_led(&second, false);
    }

    #[tokio::test]
    async fn abandoned_in_flight_entry_is_replaced() {
        let cache = ImageReadyCache::new();
        drop(cache.join_or_lead(
            &key("a"),
            /* recovery */ false,
            |_| pending_build("img"),
        ));
        // Every waiter dropped, so the weak handle is dead and the next
        // caller leads a fresh build instead of leaking the abandoned one.
        let next = cache.join_or_lead(
            &key("a"),
            /* recovery */ false,
            |_| pending_build("img2"),
        );
        assert_led(&next, true);
    }

    #[tokio::test]
    async fn settled_success_serves_without_any_waiter() {
        let cache = ImageReadyCache::new();
        drop(cache.join_or_lead(
            &key("a"),
            /* recovery */ false,
            |_| pending_build("img"),
        ));
        cache.settle_success(&key("a"), /* id */ 0, ready_build("img"));
        match cache.join_or_lead(
            &key("a"),
            /* recovery */ false,
            |_| panic!("must serve the value"),
        ) {
            Joined::Ready(build) => assert_eq!(build.image_id, "img"),
            Joined::Pending { .. } => panic!("expected the settled value"),
        }
    }

    #[tokio::test]
    async fn expired_success_is_replaced() {
        let cache = ImageReadyCache::new();
        cache.set_refresh_window(Duration::ZERO);
        drop(cache.join_or_lead(
            &key("a"),
            /* recovery */ false,
            |_| pending_build("img"),
        ));
        cache.settle_success(&key("a"), /* id */ 0, ready_build("img"));
        // A zero refresh window makes the settled success immediately stale.
        let next = cache.join_or_lead(
            &key("a"),
            /* recovery */ false,
            |_| pending_build("img2"),
        );
        assert_led(&next, true);
    }

    #[tokio::test]
    async fn failed_entry_is_removed_and_superseded_settle_ignored() {
        let cache = ImageReadyCache::new();
        let first = cache.join_or_lead(
            &key("a"),
            /* recovery */ false,
            |_| pending_build("img"),
        );
        cache.settle_failure(&key("a"), /* id */ 0);
        let second = cache.join_or_lead(
            &key("a"),
            /* recovery */ false,
            |_| pending_build("img2"),
        );
        assert_led(&second, true);
        // The replacement entry has id 1; a stale settle for id 0 is a no-op.
        cache.settle_failure(&key("a"), /* id */ 0);
        let third = cache.join_or_lead(
            &key("a"),
            /* recovery */ false,
            |_| panic!("must join"),
        );
        assert_led(&third, false);
        drop((first, second, third));
    }

    #[tokio::test]
    async fn in_flight_recovery_survives_a_later_stale_creates_invalidation() {
        let cache = ImageReadyCache::new();
        // A recovery rebuild led after some conflict discovery.
        let held = cache.join_or_lead(
            &key("a"),
            /* recovery */ true,
            |_| pending_build("rec"),
        );
        // A stale create that began after the recovery started invalidates
        // with a later cutoff; the in-flight recovery must stay joinable.
        cache.invalidate_spec_started_before("a", Instant::now());
        let joined = cache.join_or_lead(&key("a"), /* recovery */ true, |_| {
            panic!("in-flight recovery must survive")
        });
        assert_led(&joined, false);
        // Once settled, a recovery is no longer exempt: a conflict after its
        // result was available means that result is suspect too.
        cache.settle_success(&key("a"), /* id */ 0, ready_build("rec"));
        cache.invalidate_spec_started_before("a", Instant::now());
        let next = cache.join_or_lead(
            &key("a"),
            /* recovery */ true,
            |_| pending_build("rec2"),
        );
        assert_led(&next, true);
        drop((held, joined, next));
    }

    #[tokio::test]
    async fn repeated_replacement_of_one_key_keeps_the_order_queue_bounded() {
        let cache = ImageReadyCache::new();
        cache.set_refresh_window(Duration::ZERO);
        for i in 0..(MAX_READY_ENTRIES * 10) {
            let joined = cache.join_or_lead(
                &key("hot"),
                /* recovery */ false,
                |_| pending_build("img"),
            );
            assert_led(&joined, true);
            // Settle as success; the zero refresh window makes the entry
            // immediately stale, so every iteration replaces it.
            cache.settle_success(&key("hot"), /* id */ i as u64, ready_build("img"));
        }
        assert!(
            cache.order_len() <= MAX_READY_ENTRIES * 2,
            "order queue grew to {}",
            cache.order_len()
        );
    }

    #[tokio::test]
    async fn invalidation_scopes_by_build_start_time() {
        let cache = ImageReadyCache::new();
        // Two builds started before the cutoff: one settles after the cutoff,
        // one stays in flight. Both are suspect and must be dropped.
        drop(cache.join_or_lead(
            &key("a"),
            /* recovery */ false,
            |_| pending_build("pre"),
        ));
        let in_flight_key = (Duration::from_secs(90), "a".to_string());
        let held_pre = cache.join_or_lead(
            &in_flight_key,
            /* recovery */ false,
            |_| pending_build("pre2"),
        );
        let cutoff = Instant::now();
        cache.settle_success(&key("a"), /* id */ 0, ready_build("settled-late"));
        cache.invalidate_spec_started_before("a", cutoff);
        match cache.join_or_lead(
            &key("a"),
            /* recovery */ false,
            |_| pending_build("post"),
        ) {
            Joined::Pending { led, .. } => {
                assert!(led, "a pre-cutoff build must be dropped however it settled");
            }
            Joined::Ready(_) => panic!("a pre-cutoff build must be dropped however it settled"),
        }
        let post = cache.join_or_lead(
            &in_flight_key,
            /* recovery */ false,
            |_| pending_build("post2"),
        );
        assert_led(&post, true);
        // A build started after the cutoff is another caller's recovery and
        // survives a repeat invalidation with the same cutoff.
        cache.invalidate_spec_started_before("a", cutoff);
        let rejoined = cache.join_or_lead(&in_flight_key, /* recovery */ false, |_| {
            panic!("post-cutoff entry must survive")
        });
        assert_led(&rejoined, false);
        drop((held_pre, post, rejoined));
    }

    #[tokio::test]
    async fn evicts_oldest_past_the_cap() {
        let cache = ImageReadyCache::new();
        drop(cache.join_or_lead(
            &key("first"),
            /* recovery */ false,
            |_| pending_build("img"),
        ));
        cache.settle_success(&key("first"), /* id */ 0, ready_build("img"));
        for i in 0..MAX_READY_ENTRIES {
            drop(cache.join_or_lead(
                &key(&format!("filler-{i}")),
                /* recovery */ false,
                |_| pending_build("img"),
            ));
        }
        let first_again = cache.join_or_lead(
            &key("first"),
            /* recovery */ false,
            |_| pending_build("img"),
        );
        assert_led(&first_again, true);
    }

    #[test]
    fn canonical_spec_key_ignores_env_insertion_order() {
        use crate::image::ImageSpec;
        let mut forward = ImageSpec::default();
        forward.env.insert("A_FIRST".to_string(), "1".to_string());
        forward.env.insert("B_SECOND".to_string(), "2".to_string());
        let mut reverse = ImageSpec::default();
        reverse.env.insert("B_SECOND".to_string(), "2".to_string());
        reverse.env.insert("A_FIRST".to_string(), "1".to_string());
        // serde_json's Map is sorted (no preserve_order feature); this guards
        // against a future feature unification silently breaking key
        // determinism.
        assert_eq!(
            crate::imagebuild::canonical_spec_key(&forward).unwrap(),
            crate::imagebuild::canonical_spec_key(&reverse).unwrap()
        );
    }
}