lruttl: tidy up api

Fewer weirdly named functions and awkward parameter types.

crates/dns-resolver/src/lib.rs

@@ -516,8 +516,8 @@ async fn lookup_mx_record(domain_name: &Name) -> anyhow::Result<(Vec<ByPreferenc
 
 pub async fn ip_lookup(key: &str) -> anyhow::Result<(Arc<Vec<IpAddr>>, Instant)> {
     let key_fq = fully_qualify(key)?;
-    if let Some(value) = IP_CACHE.get_with_expiry(&key_fq).await {
-        return Ok(value);
+    if let Some(lookup) = IP_CACHE.lookup(&key_fq).await {
+        return Ok((lookup.item, lookup.expiration));
     }
 
     let (v4, v6) = tokio::join!(ipv4_lookup(key), ipv6_lookup(key));
@@ -564,8 +564,8 @@ pub async fn ip_lookup(key: &str) -> anyhow::Result<(Arc<Vec<IpAddr>>, Instant)>
 
 pub async fn ipv4_lookup(key: &str) -> anyhow::Result<(Arc<Vec<IpAddr>>, Instant)> {
     let key_fq = fully_qualify(key)?;
-    if let Some(value) = IPV4_CACHE.get_with_expiry(&key_fq).await {
-        return Ok(value);
+    if let Some(lookup) = IPV4_CACHE.lookup(&key_fq).await {
+        return Ok((lookup.item, lookup.expiration));
     }
 
     let answer = RESOLVER
@@ -582,8 +582,8 @@ pub async fn ipv4_lookup(key: &str) -> anyhow::Result<(Arc<Vec<IpAddr>>, Instant
 
 pub async fn ipv6_lookup(key: &str) -> anyhow::Result<(Arc<Vec<IpAddr>>, Instant)> {
     let key_fq = fully_qualify(key)?;
-    if let Some(value) = IPV6_CACHE.get_with_expiry(&key_fq).await {
-        return Ok(value);
+    if let Some(lookup) = IPV6_CACHE.lookup(&key_fq).await {
+        return Ok((lookup.item, lookup.expiration));
     }
 
     let answer = RESOLVER

crates/kumod/src/egress_source.rs

@@ -7,7 +7,7 @@ use data_loader::KeySource;
 use gcd::Gcd;
 use kumo_log_types::MaybeProxiedSourceAddress;
 use kumo_server_common::config_handle::ConfigHandle;
-use lruttl::{ItemTtl, LruCacheWithTtl};
+use lruttl::LruCacheWithTtl;
 use mlua::prelude::LuaUserData;
 use parking_lot::FairMutex as Mutex;
 use prometheus::IntCounter;
@@ -69,16 +69,10 @@ pub struct EgressSource {
 
 impl LuaUserData for EgressSource {}
 
-impl ItemTtl for EgressSource {
-    fn get_ttl(&self) -> Duration {
-        self.ttl
-    }
-}
-
 impl EgressSource {
     pub async fn resolve(name: &str, config: &mut LuaConfig) -> anyhow::Result<Self> {
         SOURCES
-            .get_or_try_insert_embedded_ttl(name.to_string(), async {
+            .get_or_try_insert(&name.to_string(), |source| source.ttl, async {
                 if name == "unspecified" {
                     Ok(EgressSource {
                         name: name.to_string(),
@@ -103,6 +97,7 @@ impl EgressSource {
             })
             .await
             .map_err(|err| anyhow::anyhow!("{err:#}"))
+            .map(|lookup| lookup.item)
     }
 
     fn resolve_proxy_protocol(&self, address: SocketAddr) -> anyhow::Result<ProxyProto> {
@@ -303,18 +298,12 @@ pub struct EgressPool {
 
 impl LuaUserData for EgressPool {}
 
-impl ItemTtl for EgressPool {
-    fn get_ttl(&self) -> Duration {
-        self.ttl
-    }
-}
-
 impl EgressPool {
     pub async fn resolve(name: Option<&str>, config: &mut LuaConfig) -> anyhow::Result<Self> {
         let name = name.unwrap_or("unspecified");
 
         let pool = POOLS
-            .get_or_try_insert_embedded_ttl(name.to_string(), async {
+            .get_or_try_insert(&name.to_string(), |pool| pool.ttl, async {
                 let pool = if name == "unspecified" {
                     EgressPool {
                         name: "unspecified".to_string(),
@@ -336,7 +325,8 @@ impl EgressPool {
                 Ok(pool)
             })
             .await
-            .map_err(|err: Arc<anyhow::Error>| anyhow::anyhow!("{err:#}"))?;
+            .map_err(|err: Arc<anyhow::Error>| anyhow::anyhow!("{err:#}"))?
+            .item;
 
         // Validate each of the sources
         for entry in &pool.entries {

crates/lruttl/src/lib.rs

@@ -119,8 +119,14 @@ impl<K: Send, V: Send + Sync> Expiry<K, Item<V>> for PerItemExpiry<K, V> {
     }
 }
 
-pub trait ItemTtl {
-    fn get_ttl(&self) -> Duration;
+pub struct ItemLookup<V> {
+    /// A copy of the item
+    pub item: V,
+    /// If true, the get_or_try_insert operation populated the entry;
+    /// the operation was a cache miss
+    pub is_fresh: bool,
+    /// The instant at which this entry will expire
+    pub expiration: Instant,
 }
 
 pub struct LruCacheWithTtl<K: Clone + Hash + Eq, V: Clone + Send + Sync> {
@@ -178,13 +184,17 @@ impl<
         num_entries as usize
     }
 
-    pub async fn get_with_expiry<Q: ?Sized>(&self, name: &Q) -> Option<(V, Instant)>
+    pub async fn lookup<Q: ?Sized>(&self, name: &Q) -> Option<ItemLookup<V>>
     where
         K: Borrow<Q>,
         Q: Hash + Eq,
     {
         let entry = self.inner.cache.get(name).await?;
-        Some((entry.item.clone(), entry.expiration))
+        Some(ItemLookup {
+            item: entry.item.clone(),
+            expiration: entry.expiration,
+            is_fresh: false,
+        })
     }
 
     pub async fn get<Q: ?Sized>(&self, name: &Q) -> Option<V>
@@ -192,8 +202,7 @@ impl<
         K: Borrow<Q>,
         Q: Hash + Eq,
     {
-        let entry = self.inner.cache.get(name).await?;
-        entry.item.clone().into()
+        self.lookup(name).await.map(|lookup| lookup.item)
     }
 
     pub async fn insert(&self, name: K, item: V, expiration: Instant) -> V {
@@ -213,63 +222,21 @@ impl<
     /// Get an existing item, but if that item doesn't already exist,
     /// execute the future `fut` to provide a value that will be inserted and then
     /// returned.  This is done atomically wrt. other callers.
-    pub async fn get_or_insert(&self, name: K, ttl: Duration, fut: impl Future<Output = V>) -> V {
-        let item = self
-            .inner
-            .cache
-            .get_with(name, async move {
-                let item = fut.await;
-                Item {
-                    item,
-                    expiration: Instant::now() + ttl,
-                }
-            })
-            .await;
-        item.item
-    }
-
-    /// Get an existing item, but if that item doesn't already exist,
-    /// execute the future `fut` to provide a value that will be inserted and then
-    /// returned.  This is done atomically wrt. other callers.
-    /// This variant allows the future to be fallible.
-    pub async fn get_or_try_insert<E: Send + Sync + 'static>(
-        &self,
-        name: K,
-        ttl: Duration,
-        fut: impl Future<Output = Result<V, E>>,
-    ) -> Result<V, Arc<E>> {
-        let item = self
-            .inner
-            .cache
-            .try_get_with(name, async move {
-                let item = fut.await?;
-                Ok(Item {
-                    item,
-                    expiration: Instant::now() + ttl,
-                })
-            })
-            .await?;
-        Ok(item.item)
-    }
-
-    /// Get an existing item, but if that item doesn't already exist,
-    /// execute the future `fut` to provide a value that will be inserted and then
-    /// returned.  This is done atomically wrt. other callers.
-    /// This variant allows the future to be fallible, as well as returns a freshness
-    /// flag that can be used to update cache metrics.
-    /// if the freshness flag is true, it was a cache miss.
-    pub async fn get_or_try_insert_with_freshness_status<E: Send + Sync + 'static>(
+    /// The TTL parameter is a function that can extract the TTL from the value type,
+    /// or just return a constant TTL.
+    pub async fn get_or_try_insert<E: Send + Sync + 'static, TTL: FnOnce(&V) -> Duration>(
         &self,
         name: &K,
-        ttl: Duration,
+        ttl: TTL,
         fut: impl Future<Output = Result<V, E>>,
-    ) -> Result<(bool, V), Arc<E>> {
+    ) -> Result<ItemLookup<V>, Arc<E>> {
         let entry = self
             .inner
             .cache
             .entry_by_ref(name)
             .or_try_insert_with(async move {
                 let item = fut.await?;
+                let ttl = (ttl)(&item);
                 Ok(Item {
                     item,
                     expiration: Instant::now() + ttl,
@@ -277,32 +244,12 @@ impl<
             })
             .await?;
         let is_fresh = entry.is_fresh();
-        Ok((is_fresh, entry.value().item.clone()))
-    }
-}
+        let item = entry.value();
 
-impl<
-        K: Clone + Hash + Eq + Send + Sync + std::fmt::Debug + 'static,
-        V: Clone + Send + Sync + ItemTtl + 'static,
-    > LruCacheWithTtl<K, V>
-{
-    pub async fn get_or_try_insert_embedded_ttl<E: Send + Sync + 'static>(
-        &self,
-        name: K,
-        fut: impl Future<Output = Result<V, E>>,
-    ) -> Result<V, Arc<E>> {
-        let item = self
-            .inner
-            .cache
-            .try_get_with(name, async move {
-                let item = fut.await?;
-                let ttl = item.get_ttl();
-                Ok(Item {
-                    item,
-                    expiration: Instant::now() + ttl,
-                })
-            })
-            .await?;
-        Ok(item.item)
+        Ok(ItemLookup {
+            is_fresh,
+            item: item.item.clone(),
+            expiration: item.expiration,
+        })
     }
 }

crates/mod-memoize/src/lib.rs

@@ -443,7 +443,7 @@ pub fn register(lua: &Lua) -> anyhow::Result<()> {
                     let key = (epoch_key, key);
 
                     let value_result = cache
-                        .get_or_try_insert_with_freshness_status(&key, ttl, async {
+                        .get_or_try_insert(&key, |_| ttl, async {
                             tracing::trace!("populate {key:?}");
                             populate_counter.inc();
                             let result: MultiValue = (func?).call_async(params).await?;
@@ -452,13 +452,13 @@ pub fn register(lua: &Lua) -> anyhow::Result<()> {
                         .await;
 
                     match value_result {
-                        Ok((is_fresh, value)) => {
-                            if is_fresh {
+                        Ok(lookup) => {
+                            if lookup.is_fresh {
                                 miss_counter.inc();
                             } else {
                                 hit_counter.inc();
                             }
-                            value.to_value(&lua)
+                            lookup.item.to_value(&lua)
                         }
                         Err(err) => {
                             tracing::error!("{cache_name} {key:?} failed: {err:#}");