[ENH][log] Cache the manifest and etag. (#5368)

## Description of changes

Prior to this commit, the code cached just the manifest.  By caching the
etag we can HEAD the manifest if it hasn't changed and get some latency
improvements for scout logs.

## Test plan

CI

## Migration plan

I changed the cache key, so even if we had a disk cache it would be
backwards-compatible.

## Observability plan

N/A

## Documentation Changes

N/A

---------

Co-authored-by: propel-code-bot[bot] <203372662+propel-code-bot[bot]@users.noreply.github.com>

Author: rescrv

rust/log-service/src/lib.rs

@@ -47,7 +47,7 @@ use uuid::Uuid;
 use wal3::{
     Cursor, CursorName, CursorStore, CursorStoreOptions, Fragment, GarbageCollectionOptions,
     Limits, LogPosition, LogReader, LogReaderOptions, LogWriter, LogWriterOptions, Manifest,
-    MarkDirty as MarkDirtyTrait, Witness,
+    ManifestAndETag, MarkDirty as MarkDirtyTrait, Witness,
 };
 
 pub mod state_hash_table;
@@ -289,10 +289,10 @@ async fn get_log_from_handle_with_mutex_held<'a>(
     })
 }
 
-////////////////////////////////////// cache_key_for_manifest //////////////////////////////////////
+////////////////////////////////////////// cache_key_for_* /////////////////////////////////////////
 
-fn cache_key_for_manifest(collection_id: CollectionUuid) -> String {
-    format!("{collection_id}::MANIFEST")
+fn cache_key_for_manifest_and_etag(collection_id: CollectionUuid) -> String {
+    format!("{collection_id}::MANIFEST/ETAG")
 }
 
 fn cache_key_for_cursor(collection_id: CollectionUuid, name: &CursorName) -> String {
@@ -1352,11 +1352,11 @@ impl LogServer {
             Err(err) => return Err(Status::new(err.code().into(), err.to_string())),
         };
         if let Some(cache) = self.cache.as_ref() {
-            let cache_key = cache_key_for_manifest(collection_id);
-            if let Some(manifest) = log.manifest() {
-                if let Ok(manifest_bytes) = serde_json::to_vec(&manifest) {
+            let cache_key = cache_key_for_manifest_and_etag(collection_id);
+            if let Some(manifest_and_etag) = log.manifest_and_etag() {
+                if let Ok(manifest_and_etag_bytes) = serde_json::to_vec(&manifest_and_etag) {
                     let cache_value = CachedBytes {
-                        bytes: manifest_bytes,
+                        bytes: manifest_and_etag_bytes,
                     };
                     cache.insert(cache_key, cache_value).await;
                 }
@@ -1427,16 +1427,17 @@ impl LogServer {
         pull_logs: &PullLogsRequest,
     ) -> Option<Vec<Fragment>> {
         if let Some(cache) = self.cache.as_ref() {
-            let cache_key = cache_key_for_manifest(collection_id);
+            let cache_key = cache_key_for_manifest_and_etag(collection_id);
             let cached_bytes = cache.get(&cache_key).await.ok().flatten()?;
-            let manifest: Manifest = serde_json::from_slice(&cached_bytes.bytes).ok()?;
+            let manifest_and_etag: ManifestAndETag =
+                serde_json::from_slice(&cached_bytes.bytes).ok()?;
             let limits = Limits {
                 max_files: Some(pull_logs.batch_size as u64 + 1),
                 max_bytes: None,
                 max_records: Some(pull_logs.batch_size as u64),
             };
             LogReader::scan_from_manifest(
-                &manifest,
+                &manifest_and_etag.manifest,
                 LogPosition::from_offset(pull_logs.start_from_offset as u64),
                 limits,
             )
@@ -2027,7 +2028,7 @@ impl LogServer {
                 let collection_id = Uuid::parse_str(&x)
                     .map(CollectionUuid)
                     .map_err(|_| Status::invalid_argument("Failed to parse collection id"))?;
-                Some(cache_key_for_manifest(collection_id))
+                Some(cache_key_for_manifest_and_etag(collection_id))
             }
             Some(EntryToEvict::Fragment(f)) => {
                 let collection_id = Uuid::parse_str(&f.collection_id)

rust/wal3/src/lib.rs

@@ -23,7 +23,9 @@ pub use copy::copy;
 pub use cursors::{Cursor, CursorName, CursorStore, Witness};
 pub use destroy::destroy;
 pub use gc::{Garbage, GarbageCollector};
-pub use manifest::{unprefixed_snapshot_path, Manifest, Snapshot, SnapshotPointer};
+pub use manifest::{
+    unprefixed_snapshot_path, Manifest, ManifestAndETag, Snapshot, SnapshotPointer,
+};
 pub use manifest_manager::ManifestManager;
 pub use reader::{Limits, LogReader};
 pub use snapshot_cache::SnapshotCache;

rust/wal3/src/manifest.rs

@@ -925,6 +925,14 @@ impl Manifest {
     }
 }
 
+////////////////////////////////////////// ManifestAndETag /////////////////////////////////////////
+
+#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
+pub struct ManifestAndETag {
+    pub manifest: Manifest,
+    pub e_tag: ETag,
+}
+
 /////////////////////////////////////////////// tests //////////////////////////////////////////////
 
 #[cfg(test)]

rust/wal3/src/manifest_manager.rs

@@ -4,22 +4,14 @@ use std::time::Instant;
 use chroma_storage::{ETag, Storage};
 
 use crate::gc::Garbage;
-use crate::manifest::{Manifest, Snapshot};
+use crate::manifest::{Manifest, ManifestAndETag, Snapshot};
 use crate::reader::read_fragment;
 use crate::writer::MarkDirty;
 use crate::{
     unprefixed_fragment_path, Error, Fragment, FragmentSeqNo, GarbageCollectionOptions,
     LogPosition, SnapshotCache, SnapshotOptions, SnapshotPointerOrFragmentSeqNo, ThrottleOptions,
 };
 
-////////////////////////////////////////// ManifestAndETag /////////////////////////////////////////
-
-#[derive(Debug)]
-struct ManifestAndETag {
-    manifest: Manifest,
-    e_tag: ETag,
-}
-
 ////////////////////////////////////////////// Staging /////////////////////////////////////////////
 
 #[derive(Debug)]
@@ -270,9 +262,9 @@ impl ManifestManager {
     }
 
     /// Return the latest stable manifest
-    pub fn latest(&self) -> Manifest {
+    pub fn latest(&self) -> ManifestAndETag {
         let staging = self.staging.lock().unwrap();
-        staging.stable.manifest.clone()
+        staging.stable.clone()
     }
 
     /// Recover from a fault in writing.  It is possible that fragments have been written that are

rust/wal3/src/writer.rs

@@ -18,7 +18,8 @@ use tracing_opentelemetry::OpenTelemetrySpanExt;
 use crate::{
     unprefixed_fragment_path, BatchManager, CursorStore, CursorStoreOptions, Error,
     ExponentialBackoff, Fragment, FragmentSeqNo, Garbage, GarbageCollectionOptions, LogPosition,
-    LogReader, LogReaderOptions, LogWriterOptions, Manifest, ManifestManager, ThrottleOptions,
+    LogReader, LogReaderOptions, LogWriterOptions, Manifest, ManifestAndETag, ManifestManager,
+    ThrottleOptions,
 };
 
 /// The epoch writer is a counting writer.  Every epoch exists.  An epoch goes
@@ -275,6 +276,10 @@ impl LogWriter {
     }
 
     pub fn manifest(&self) -> Option<Manifest> {
+        self.manifest_and_etag().map(|m| m.manifest)
+    }
+
+    pub fn manifest_and_etag(&self) -> Option<ManifestAndETag> {
         // SAFETY(rescrv):  Mutex poisoning.
         let inner = self.inner.lock().unwrap();
         inner