Merge pull request #24208 from danhhz/persist_blob_cache

persist: reintroduce in-mem blob cache

src/persist-client/proptest-regressions/internal/cache.txt

@@ -0,0 +1,7 @@
+# Seeds for failure cases proptest has generated in the past. It is
+# automatically read and these particular cases re-run before any
+# novel cases are generated.
+#
+# It is recommended to check this file in to source control so that
+# everyone who runs the test benefits from these saved cases.
+cc 520a1ce380cba2b6a303454a884b5feecbf32e3628eae0f2840b793c9a75b78a # shrinks to state = [Insert { key: 235, weight: 0 }, Get { key: 235 }]

src/persist-client/src/cache.rs

@@ -33,6 +33,7 @@ use tracing::{debug, instrument};
 
 use crate::async_runtime::IsolatedRuntime;
 use crate::error::{CodecConcreteType, CodecMismatch};
+use crate::internal::cache::BlobMemCache;
 use crate::internal::machine::retry_external;
 use crate::internal::metrics::{LockMetrics, Metrics, MetricsBlob, MetricsConsensus, ShardMetrics};
 use crate::internal::state::TypedState;
@@ -218,6 +219,9 @@ impl PersistClientCache {
                     Self::PROMETHEUS_SCRAPE_INTERVAL,
                 )
                 .await;
+                // This is intentionally "outside" (wrapping) MetricsBlob so
+                // that we don't include cached responses in blob metrics.
+                let blob = BlobMemCache::new(&self.cfg, Arc::clone(&self.metrics), blob);
                 Arc::clone(&x.insert((RttLatencyTask(task.abort_on_drop()), blob)).1)
             }
         };