lruttl: add params to control timeout and retry on sema timeout

These are hooked up only for memoize at this time.  No default
behavior is changed by this commit, but you can optionally
specify these parameters in order to change the behavior.

Author: wez

crates/lruttl/src/lib.rs

@@ -8,7 +8,7 @@ use std::collections::HashSet;
 use std::fmt::Debug;
 use std::future::Future;
 use std::hash::Hash;
-use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::sync::{Arc, LazyLock, Weak};
 use tokio::sync::Semaphore;
 use tokio::time::{timeout, Duration, Instant};
@@ -25,6 +25,8 @@ struct Inner<K: Clone + Hash + Eq + Debug, V: Clone + Send + Sync + Debug> {
     capacity: AtomicUsize,
     cache: DashMap<K, Item<V>>,
     lru_samples: AtomicUsize,
+    sema_timeout_milliseconds: AtomicUsize,
+    retry_on_sema_timeout: AtomicBool,
     lookup_counter: IntCounter,
     evict_counter: IntCounter,
     expire_counter: IntCounter,
@@ -525,6 +527,8 @@ impl<
             tick: AtomicUsize::new(0),
             capacity: AtomicUsize::new(capacity),
             lru_samples: AtomicUsize::new(10),
+            sema_timeout_milliseconds: AtomicUsize::new(120_000),
+            retry_on_sema_timeout: AtomicBool::new(false),
             lookup_counter,
             evict_counter,
             expire_counter,
@@ -552,6 +556,18 @@ impl<
         Self { inner }
     }
 
+    pub fn set_retry_on_sema_timeout(&self, value: bool) {
+        self.inner
+            .retry_on_sema_timeout
+            .store(value, Ordering::Relaxed);
+    }
+
+    pub fn set_sema_timeout(&self, duration: Duration) {
+        self.inner
+            .sema_timeout_milliseconds
+            .store(duration.as_millis() as usize, Ordering::Relaxed);
+    }
+
     pub fn clear(&self) -> usize {
         self.inner.clear()
     }
@@ -722,22 +738,38 @@ impl<
                     }
 
                     let wait_count = DecOnDrop::new(self.inner.wait_gauge.clone());
-                    let wait_result =
-                        match timeout(Duration::from_secs(120), sema.acquire_owned()).await {
-                            Err(_) => {
-                                self.inner.error_counter.inc();
-                                tracing::error!(
-                                    "{} semaphore acquire for {name:?} timed out",
+                    let wait_result = match timeout(
+                        Duration::from_millis(
+                            self.inner.sema_timeout_milliseconds.load(Ordering::Relaxed) as u64,
+                        ),
+                        sema.acquire_owned(),
+                    )
+                    .await
+                    {
+                        Err(_) => {
+                            self.inner.error_counter.inc();
+
+                            if self.inner.retry_on_sema_timeout.load(Ordering::Relaxed) {
+                                tracing::warn!(
+                                    "{} semaphore acquire for {name:?} timed out, \
+                                    will restart cache resolve.",
                                     self.inner.name
                                 );
-                                return Err(Arc::new(anyhow::anyhow!(
-                                    "{} lookup for {name:?} \
-                                            timed out on semaphore acquire",
-                                    self.inner.name
-                                )));
+                                continue 'retry;
                             }
-                            Ok(r) => r,
-                        };
+
+                            tracing::error!(
+                                "{} semaphore acquire for {name:?} timed out",
+                                self.inner.name
+                            );
+                            return Err(Arc::new(anyhow::anyhow!(
+                                "{} lookup for {name:?} \
+                                            timed out on semaphore acquire",
+                                self.inner.name
+                            )));
+                        }
+                        Ok(r) => r,
+                    };
 
                     drop(wait_count);
 

crates/mod-memoize/src/lib.rs

@@ -66,6 +66,10 @@ pub struct MemoizeParams {
     pub name: String,
     #[serde(default)]
     pub invalidate_with_epoch: bool,
+    #[serde(default)]
+    pub retry_on_populate_timeout: bool,
+    #[serde(default, with = "duration_serde")]
+    pub populate_timeout: Option<Duration>,
 }
 
 #[derive(Clone, Hash, Eq, PartialEq)]
@@ -397,12 +401,18 @@ pub fn register(lua: &Lua) -> anyhow::Result<()> {
                 }
                 changed
             });
-            CACHES
-                .entry(cache_name.to_string())
-                .or_insert_with(|| MemoizeCache {
+            CACHES.entry(cache_name.to_string()).or_insert_with(|| {
+                let cache = LruCacheWithTtl::new(cache_name.clone(), params.capacity);
+                cache.set_retry_on_sema_timeout(params.retry_on_populate_timeout);
+                if let Some(duration) = params.populate_timeout {
+                    cache.set_sema_timeout(duration);
+                }
+
+                MemoizeCache {
                     params: params.clone(),
-                    cache: Arc::new(LruCacheWithTtl::new(cache_name.clone(), params.capacity)),
-                });
+                    cache: Arc::new(cache),
+                }
+            });
 
             let lookup_counter = CACHE_LOOKUP
                 .get_metric_with_label_values(&[&cache_name])

docs/reference/kumo/memoize.md

@@ -71,7 +71,7 @@ the *memoized function*.
 The parameters it accepts are:
 
 * *FUNCTION* - the function or lambda which will be called when there is a cache miss.
-  When it is called, it will be passed the parameters that were passed to the *memoized function*.
+  When it is called, it will be passed the parameters that were passed to the *memoized function* in order to populate the cache.
 * *PARAMS* is a required lua table with the following fields, all of which are required:
      * `name` - the name for the cache. You should create one name per function/purpose.
      * `ttl` - the Time To Live for cache entries; how long a previously computed
@@ -83,6 +83,13 @@ The parameters it accepts are:
      * `invalidate_with_epoch` - optional boolean that defaults to `false`.
        If true, anything that bumps the config epoch (eg: config file changes,
        TSA config overrides and so on) will invalidate the cache. {{since('2025.03.19-1d3f1f67', inline=True)}}
+     * `populate_timeout` - optional duration string. The effective default
+       value is `120 seconds`.  Specifies how long to allow the cache population
+       function to run before generating a timeout error. {{since('dev', inline=True)}}
+     * `retry_on_populate_timeout` - optional boolean that defaults to `false`.
+       If true, if the `populate_timeout` is reached, then instead of generating
+       an error, memoize will retry the population attempt.
+       {{since('dev', inline=True)}}
 
 In the example above calling: