transient in memory cache

apollo-router/src/cache/mod.rs

@@ -21,6 +21,10 @@ pub(crate) const DEFAULT_CACHE_CAPACITY: NonZeroUsize = match NonZeroUsize::new(
     Some(v) => v,
     None => unreachable!(),
 };
+pub(crate) const DEFAULT_TRANSIENT_CACHE_CAPACITY: NonZeroUsize = match NonZeroUsize::new(50) {
+    Some(v) => v,
+    None => unreachable!(),
+};
 
 /// Cache implementation with query deduplication
 #[derive(Clone)]
@@ -36,20 +40,27 @@ where
 {
     pub(crate) async fn with_capacity(
         capacity: NonZeroUsize,
+        transient_capacity: NonZeroUsize,
         redis: Option<RedisCache>,
         caller: &str,
     ) -> Result<Self, BoxError> {
         Ok(Self {
             wait_map: Arc::new(Mutex::new(HashMap::new())),
-            storage: CacheStorage::new(capacity, redis, caller).await?,
+            storage: CacheStorage::new(capacity, transient_capacity, redis, caller).await?,
         })
     }
 
     pub(crate) async fn from_configuration(
         config: &crate::configuration::Cache,
         caller: &str,
     ) -> Result<Self, BoxError> {
-        Self::with_capacity(config.in_memory.limit, config.redis.clone(), caller).await
+        Self::with_capacity(
+            config.in_memory.limit,
+            config.in_memory.transient_limit,
+            config.redis.clone(),
+            caller,
+        )
+        .await
     }
 
     pub(crate) async fn get(&self, key: &K) -> Entry<K, V> {
@@ -212,9 +223,14 @@ mod tests {
     #[tokio::test]
     async fn example_cache_usage() {
         let k = "key".to_string();
-        let cache = DeduplicatingCache::with_capacity(NonZeroUsize::new(1).unwrap(), None, "test")
-            .await
-            .unwrap();
+        let cache = DeduplicatingCache::with_capacity(
+            NonZeroUsize::new(1).unwrap(),
+            NonZeroUsize::new(1).unwrap(),
+            None,
+            "test",
+        )
+        .await
+        .unwrap();
 
         let entry = cache.get(&k).await;
 
@@ -230,14 +246,20 @@ mod tests {
 
     #[test(tokio::test)]
     async fn it_should_enforce_cache_limits() {
-        let cache: DeduplicatingCache<usize, usize> =
-            DeduplicatingCache::with_capacity(NonZeroUsize::new(13).unwrap(), None, "test")
-                .await
-                .unwrap();
+        let cache: DeduplicatingCache<usize, usize> = DeduplicatingCache::with_capacity(
+            NonZeroUsize::new(13).unwrap(),
+            NonZeroUsize::new(1).unwrap(),
+            None,
+            "test",
+        )
+        .await
+        .unwrap();
 
         for i in 0..14 {
             let entry = cache.get(&i).await;
             entry.insert(i).await;
+            let _ = cache.get(&i).await.get().await;
+            tokio::task::yield_now().await;
         }
 
         assert_eq!(cache.storage.len().await, 13);
@@ -255,10 +277,14 @@ mod tests {
 
         mock.expect_retrieve().times(1).return_const(1usize);
 
-        let cache: DeduplicatingCache<usize, usize> =
-            DeduplicatingCache::with_capacity(NonZeroUsize::new(10).unwrap(), None, "test")
-                .await
-                .unwrap();
+        let cache: DeduplicatingCache<usize, usize> = DeduplicatingCache::with_capacity(
+            NonZeroUsize::new(10).unwrap(),
+            NonZeroUsize::new(1).unwrap(),
+            None,
+            "test",
+        )
+        .await
+        .unwrap();
 
         // Let's trigger 100 concurrent gets of the same value and ensure only
         // one delegated retrieve is made
@@ -280,4 +306,29 @@ mod tests {
         // To be really sure, check there is only one value in the cache
         assert_eq!(cache.storage.len().await, 1);
     }
+
+    #[test(tokio::test)]
+    async fn transient_cache() {
+        let cache: DeduplicatingCache<usize, usize> = DeduplicatingCache::with_capacity(
+            NonZeroUsize::new(2).unwrap(),
+            NonZeroUsize::new(5).unwrap(),
+            None,
+            "test",
+        )
+        .await
+        .unwrap();
+
+        let entry = cache.get(&0).await;
+        entry.insert(0).await;
+        let _ = cache.get(&0).await.get().await;
+        tokio::task::yield_now().await;
+
+        for i in 1..14 {
+            let entry = cache.get(&i).await;
+            entry.insert(i).await;
+        }
+
+        assert_eq!(cache.get(&0).await.get().await.unwrap(), 0);
+        assert_eq!(cache.storage.len().await, 6);
+    }
 }

apollo-router/src/cache/storage.rs

@@ -51,22 +51,25 @@ pub(crate) type InMemoryCache<K, V> = Arc<Mutex<LruCache<K, V>>>;
 pub(crate) struct CacheStorage<K: KeyType, V: ValueType> {
     caller: String,
     inner: Arc<Mutex<LruCache<K, V>>>,
+    transient: Arc<Mutex<LruCache<K, V>>>,
     redis: Option<RedisCacheStorage>,
 }
 
 impl<K, V> CacheStorage<K, V>
 where
-    K: KeyType,
-    V: ValueType,
+    K: KeyType + 'static,
+    V: ValueType + 'static,
 {
     pub(crate) async fn new(
         max_capacity: NonZeroUsize,
+        transient_capacity: NonZeroUsize,
         config: Option<RedisCache>,
         caller: &str,
     ) -> Result<Self, BoxError> {
         Ok(Self {
             caller: caller.to_string(),
             inner: Arc::new(Mutex::new(LruCache::new(max_capacity))),
+            transient: Arc::new(Mutex::new(LruCache::new(transient_capacity))),
             redis: if let Some(config) = config {
                 let required_to_start = config.required_to_start;
                 match RedisCacheStorage::new(config).await {
@@ -91,6 +94,33 @@ where
 
     pub(crate) async fn get(&self, key: &K) -> Option<V> {
         let instant_memory = Instant::now();
+
+        let transient_res = self.transient.lock().await.get(key).cloned();
+
+        if let Some(v) = transient_res {
+            let cache = self.clone();
+            let key = key.clone();
+            let value = v.clone();
+
+            tokio::task::spawn(async move {
+                cache.insert_inner(key.clone(), value).await;
+                let _ = cache.transient.lock().await.pop_entry(&key);
+            });
+
+            tracing::info!(
+                monotonic_counter.apollo_router_cache_hit_count = 1u64,
+                kind = %self.caller,
+                storage = &tracing::field::display(CacheStorageName::Memory),
+            );
+            let duration = instant_memory.elapsed().as_secs_f64();
+            tracing::info!(
+                histogram.apollo_router_cache_hit_time = duration,
+                kind = %self.caller,
+                storage = &tracing::field::display(CacheStorageName::Memory),
+            );
+            return Some(v);
+        }
+
         let res = self.inner.lock().await.get(key).cloned();
 
         match res {
@@ -163,7 +193,19 @@ where
         }
     }
 
+    // only insert in the transient cache. We will insert in the long term in memory cache on the next access
     pub(crate) async fn insert(&self, key: K, value: V) {
+        let mut in_memory = self.transient.lock().await;
+        in_memory.put(key, value);
+        let size = in_memory.len() as u64;
+        tracing::info!(
+            value.apollo_router_cache_size = size,
+            kind = %self.caller,
+            storage = &tracing::field::display(CacheStorageName::Memory),
+        );
+    }
+
+    async fn insert_inner(&self, key: K, value: V) {
         if let Some(redis) = self.redis.as_ref() {
             redis
                 .insert(RedisKey(key.clone()), RedisValue(value.clone()), None)
@@ -197,7 +239,7 @@ where
 
     #[cfg(test)]
     pub(crate) async fn len(&self) -> usize {
-        self.inner.lock().await.len()
+        self.inner.lock().await.len() + self.transient.lock().await.len()
     }
 }
 

apollo-router/src/configuration/mod.rs

@@ -43,6 +43,7 @@ pub(crate) use self::schema::generate_config_schema;
 pub(crate) use self::schema::generate_upgrade;
 use self::subgraph::SubgraphConfiguration;
 use crate::cache::DEFAULT_CACHE_CAPACITY;
+use crate::cache::DEFAULT_TRANSIENT_CACHE_CAPACITY;
 use crate::configuration::schema::Mode;
 use crate::graphql;
 use crate::notification::Notify;
@@ -1109,17 +1110,30 @@ impl From<QueryPlanCache> for Cache {
 /// In memory cache configuration
 pub(crate) struct InMemoryCache {
     /// Number of entries in the Least Recently Used cache
+    #[serde(default = "in_memory_limit_default")]
     pub(crate) limit: NonZeroUsize,
+    /// Number of entries in the Least Recently Used transient cache
+    #[serde(default = "transient_in_memory_limit_default")]
+    pub(crate) transient_limit: NonZeroUsize,
 }
 
-impl Default for InMemoryCache {
+impl std::default::Default for InMemoryCache {
     fn default() -> Self {
         Self {
             limit: DEFAULT_CACHE_CAPACITY,
+            transient_limit: DEFAULT_TRANSIENT_CACHE_CAPACITY,
         }
     }
 }
 
+fn in_memory_limit_default() -> NonZeroUsize {
+    DEFAULT_CACHE_CAPACITY
+}
+
+fn transient_in_memory_limit_default() -> NonZeroUsize {
+    DEFAULT_TRANSIENT_CACHE_CAPACITY
+}
+
 #[derive(Debug, Clone, Deserialize, Serialize, JsonSchema)]
 #[serde(deny_unknown_fields)]
 /// Redis cache configuration

apollo-router/src/configuration/snapshots/apollo_router__configuration__tests__schema_generation.snap

@@ -15,7 +15,8 @@ expression: "&schema"
         "router": {
           "cache": {
             "in_memory": {
-              "limit": 512
+              "limit": 512,
+              "transient_limit": 50
             },
             "redis": null
           }
@@ -39,7 +40,8 @@ expression: "&schema"
           "default": {
             "cache": {
               "in_memory": {
-                "limit": 512
+                "limit": 512,
+                "transient_limit": 50
               },
               "redis": null
             }
@@ -50,7 +52,8 @@ expression: "&schema"
               "description": "Cache configuration",
               "default": {
                 "in_memory": {
-                  "limit": 512
+                  "limit": 512,
+                  "transient_limit": 50
                 },
                 "redis": null
               },
@@ -59,15 +62,21 @@ expression: "&schema"
                 "in_memory": {
                   "description": "Configures the in memory cache (always active)",
                   "default": {
-                    "limit": 512
+                    "limit": 512,
+                    "transient_limit": 50
                   },
                   "type": "object",
-                  "required": [
-                    "limit"
-                  ],
                   "properties": {
                     "limit": {
                       "description": "Number of entries in the Least Recently Used cache",
+                      "default": 512,
+                      "type": "integer",
+                      "format": "uint",
+                      "minimum": 1.0
+                    },
+                    "transient_limit": {
+                      "description": "Number of entries in the Least Recently Used transient cache",
+                      "default": 50,
                       "type": "integer",
                       "format": "uint",
                       "minimum": 1.0
@@ -2893,7 +2902,8 @@ expression: "&schema"
         "query_planning": {
           "cache": {
             "in_memory": {
-              "limit": 512
+              "limit": 512,
+              "transient_limit": 50
             },
             "redis": null
           },
@@ -2963,7 +2973,8 @@ expression: "&schema"
           "default": {
             "cache": {
               "in_memory": {
-                "limit": 512
+                "limit": 512,
+                "transient_limit": 50
               },
               "redis": null
             },
@@ -2979,7 +2990,8 @@ expression: "&schema"
               "description": "Cache configuration",
               "default": {
                 "in_memory": {
-                  "limit": 512
+                  "limit": 512,
+                  "transient_limit": 50
                 },
                 "redis": null
               },
@@ -2988,15 +3000,21 @@ expression: "&schema"
                 "in_memory": {
                   "description": "Configures the in memory cache (always active)",
                   "default": {
-                    "limit": 512
+                    "limit": 512,
+                    "transient_limit": 50
                   },
                   "type": "object",
-                  "required": [
-                    "limit"
-                  ],
                   "properties": {
                     "limit": {
                       "description": "Number of entries in the Least Recently Used cache",
+                      "default": 512,
+                      "type": "integer",
+                      "format": "uint",
+                      "minimum": 1.0
+                    },
+                    "transient_limit": {
+                      "description": "Number of entries in the Least Recently Used transient cache",
+                      "default": 50,
                       "type": "integer",
                       "format": "uint",
                       "minimum": 1.0