sail-rs 0.6.3

//! 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. The cache lives in the core so every language
//! wrapper shares one implementation, keyed by the hash of the spec's
//! canonical JSON. Callers share one in-flight build per spec whatever wait
//! budget each passes. A joiner inherits the running build's deadline; if
//! that deadline lapses first, the joiner leads a fresh build, and its own
//! timeout envelope bounds its total wait either way.
//!
//! A successful build from an immutable spec serves later calls without
//! re-submission. A registry tag is different: another client can force its
//! organization to resolve that tag again, so a settled tag result is not
//! retained. Its next call reaches the backend to observe the durable
//! resolution, while digest-pinned OCI specs and non-OCI specs keep the
//! readiness optimization. Retained 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. A caller that forces a fresh tag lookup skips reuse entirely and
//! leads a build whose result replaces the entry when it is immutable.
//!
//! 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:
/// 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>>>>;

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 {
    /// Keyed by the sha256 of the spec's canonical (key-sorted) JSON.
    entries: HashMap<String, Entry>,
    /// Insertion order for oldest-first eviction past [`MAX_READY_ENTRIES`].
    order: VecDeque<(String, 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 whatever
    /// deadline it runs under, since a joiner that saw an earlier deadline
    /// lapse retries with a fresh entry and its own envelope bounds its wait
    /// (see [`crate::Client::build_spec_ready_cached`]). A miss, a stale
    /// success, or an abandoned in-flight build (every waiter dropped)
    /// invokes `make` with the new entry's id. `force` reuses nothing and
    /// always leads, so its result replaces what the entry held. `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: &str,
        recovery: bool,
        force: bool,
        make: impl FnOnce(u64) -> SharedBuild,
    ) -> Joined {
        let mut state = self.state.lock().unwrap();
        if let Some(entry) = state.entries.get(key).filter(|_| !force) {
            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.to_string(),
            Entry {
                id,
                started_at: Instant::now(),
                recovery,
                state: EntryState::InFlight(weak),
            },
        );
        state.order.push_back((key.to_string(), 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. When
    /// `retain` is true, the value serves later callers until the refresh
    /// window lapses; otherwise the settled entry is dropped after its
    /// existing waiters receive the shared result. A superseded id is ignored.
    pub(crate) fn settle_success(&self, key: &str, id: u64, build: ImageBuild, retain: bool) {
        let mut state = self.state.lock().unwrap();
        if state.entries.get(key).is_some_and(|entry| entry.id == id) {
            if retain {
                let entry = state
                    .entries
                    .get_mut(key)
                    .expect("the matching entry was just observed");
                entry.state = EntryState::Ready {
                    build,
                    resolved_at: Instant::now(),
                };
            } else {
                state.entries.remove(key);
            }
        }
    }

    /// Drop the spec's entry if it is suspect. 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();
        let Some(entry) = state.entries.get(spec_hash) else {
            return;
        };
        let suspect = entry.started_at < cutoff
            && !(entry.recovery && matches!(entry.state, EntryState::InFlight(_)));
        if suspect {
            state.entries.remove(spec_hash);
        }
    }

    /// 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: &str, 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(),
            resolved_oci_ref: String::new(),
        }
    }

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

    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(
            "a",
            /* recovery */ false,
            /* force */ false,
            |_| pending_build("img"),
        );
        assert_led(&first, true);
        let second = cache.join_or_lead(
            "a",
            /* recovery */ false,
            /* force */ 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(
            "a",
            /* recovery */ false,
            /* force */ 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(
            "a",
            /* recovery */ false,
            /* force */ 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(
            "a",
            /* recovery */ false,
            /* force */ false,
            |_| pending_build("img"),
        ));
        cache.settle_success(
            "a",
            /* id */ 0,
            ready_build("img"),
            /* retain */ true,
        );
        match cache.join_or_lead(
            "a",
            /* recovery */ false,
            /* force */ 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 nonretained_success_is_revalidated_by_the_next_caller() {
        let cache = ImageReadyCache::new();
        drop(cache.join_or_lead(
            "a",
            /* recovery */ false,
            /* force */ false,
            |_| pending_build("img"),
        ));
        cache.settle_success(
            "a",
            /* id */ 0,
            ready_build("img"),
            /* retain */ false,
        );
        let next = cache.join_or_lead(
            "a",
            /* recovery */ false,
            /* force */ false,
            |_| pending_build("img2"),
        );
        assert_led(&next, true);
    }

    #[tokio::test]
    async fn expired_success_is_replaced() {
        let cache = ImageReadyCache::new();
        cache.set_refresh_window(Duration::ZERO);
        drop(cache.join_or_lead(
            "a",
            /* recovery */ false,
            /* force */ false,
            |_| pending_build("img"),
        ));
        cache.settle_success(
            "a",
            /* id */ 0,
            ready_build("img"),
            /* retain */ true,
        );
        // A zero refresh window makes the settled success immediately stale.
        let next = cache.join_or_lead(
            "a",
            /* recovery */ false,
            /* force */ 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(
            "a",
            /* recovery */ false,
            /* force */ false,
            |_| pending_build("img"),
        );
        cache.settle_failure("a", /* id */ 0);
        let second = cache.join_or_lead(
            "a",
            /* recovery */ false,
            /* force */ 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("a", /* id */ 0);
        let third = cache.join_or_lead(
            "a",
            /* recovery */ false,
            /* force */ 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(
            "a",
            /* recovery */ true,
            /* force */ false,
            |_| 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("a", /* recovery */ true, /* force */ false, |_| {
                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(
            "a",
            /* id */ 0,
            ready_build("rec"),
            /* retain */ true,
        );
        cache.invalidate_spec_started_before("a", Instant::now());
        let next = cache.join_or_lead(
            "a",
            /* recovery */ true,
            /* force */ false,
            |_| 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(
                "hot",
                /* recovery */ false,
                /* force */ 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(
                "hot",
                /* id */ i as u64,
                ready_build("img"),
                /* retain */ true,
            );
        }
        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();
        // A build started before the cutoff is suspect however it settled, so
        // even a success recorded after the cutoff is dropped.
        drop(cache.join_or_lead(
            "a",
            /* recovery */ false,
            /* force */ false,
            |_| pending_build("pre"),
        ));
        let cutoff = Instant::now();
        cache.settle_success(
            "a",
            /* id */ 0,
            ready_build("settled-late"),
            /* retain */ true,
        );
        cache.invalidate_spec_started_before("a", cutoff);
        let post = cache.join_or_lead(
            "a",
            /* recovery */ false,
            /* force */ false,
            |_| pending_build("post"),
        );
        assert_led(&post, true);
        // Its replacement started after the cutoff: that build is another
        // caller's recovery and survives a repeat invalidation.
        cache.invalidate_spec_started_before("a", cutoff);
        let rejoined = cache.join_or_lead(
            "a",
            /* recovery */ false,
            /* force */ false,
            |_| panic!("post-cutoff entry must survive"),
        );
        assert_led(&rejoined, false);
        drop((post, rejoined));
    }

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

    #[tokio::test]
    async fn a_forced_build_replaces_a_success_the_window_would_still_serve() {
        let cache = ImageReadyCache::new();
        drop(cache.join_or_lead(
            "a",
            /* recovery */ false,
            /* force */ false,
            |_| pending_build("img"),
        ));
        cache.settle_success(
            "a",
            /* id */ 0,
            ready_build("img"),
            /* retain */ true,
        );
        let forced = cache.join_or_lead(
            "a",
            /* recovery */ false,
            /* force */ true,
            |_| pending_build("img2"),
        );
        assert_led(&forced, true);
        cache.settle_success(
            "a",
            /* id */ 1,
            ready_build("img2"),
            /* retain */ true,
        );
        // Later callers see what the forced build produced, not the value it
        // replaced.
        match cache.join_or_lead(
            "a",
            /* recovery */ false,
            /* force */ false,
            |_| panic!("must serve the forced build's result"),
        ) {
            Joined::Ready(build) => assert_eq!(build.image_id, "img2"),
            Joined::Pending { .. } => panic!("expected the forced build's result"),
        }
        drop(forced);
    }

    #[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());
        // NOTE: two HashMaps holding the same keys iterate identically within
        // one process, so this pair alone cannot detect a loss of ordering; it
        // is kept as the API-level statement of intent. The property that
        // actually matters is asserted below, against a key computed from
        // deliberately unsorted JSON.
        assert_eq!(
            crate::imagebuild::canonical_spec_key(&forward).unwrap(),
            crate::imagebuild::canonical_spec_key(&reverse).unwrap()
        );
    }

    #[test]
    fn canonical_spec_key_is_computed_from_sorted_json() {
        // The cache key must not depend on serde_json's map type. `env` is a
        // HashMap whose iteration order is seeded per process, so a build where
        // object order is insertion order would otherwise hash the same spec
        // differently in two CLI invocations and rebuild ready images.
        use crate::image::ImageSpec;
        let mut spec = ImageSpec::default();
        for key in ["Z_LAST", "A_FIRST", "M_MIDDLE"] {
            spec.env.insert(key.to_string(), "v".to_string());
        }
        let key = crate::imagebuild::canonical_spec_key(&spec).unwrap();

        // Recompute the way the function documents: sha256 over key-sorted JSON.
        let value = serde_json::to_value(&spec).unwrap();
        let sorted = serde_json::to_string(&sorted_for_test(&value)).unwrap();
        use sha2::{Digest, Sha256};
        let mut hasher = Sha256::new();
        hasher.update(sorted.as_bytes());
        assert_eq!(key, format!("{:x}", hasher.finalize()));
    }

    fn sorted_for_test(value: &serde_json::Value) -> serde_json::Value {
        match value {
            serde_json::Value::Object(map) => {
                let mut keys: Vec<&String> = map.keys().collect();
                keys.sort();
                let mut out = serde_json::Map::new();
                for key in keys {
                    out.insert(key.clone(), sorted_for_test(&map[key]));
                }
                serde_json::Value::Object(out)
            }
            serde_json::Value::Array(items) => {
                serde_json::Value::Array(items.iter().map(sorted_for_test).collect())
            }
            other => other.clone(),
        }
    }
}