Merge branch 'rescrv/cache-manifest-etag' into rescrv/scout-logs-uses-head

Author: rescrv

rust/log-service/src/lib.rs

@@ -46,7 +46,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;
@@ -288,10 +288,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 {
@@ -1104,11 +1104,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;
                 }
@@ -1181,16 +1181,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,
             )
@@ -1715,7 +1716,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,15 @@ impl LogWriter {
     }
 
     pub fn manifest(&self) -> Option<Manifest> {
+        // SAFETY(rescrv):  Mutex poisoning.
+        let inner = self.inner.lock().unwrap();
+        inner
+            .writer
+            .as_ref()
+            .map(|writer| writer.manifest_manager.latest().manifest)
+    }
+
+    pub fn manifest_and_etag(&self) -> Option<ManifestAndETag> {
         // SAFETY(rescrv):  Mutex poisoning.
         let inner = self.inner.lock().unwrap();
         inner