[ENH] Do not do any mutations of the manifest from within GC. (#5050)

## Description of changes

3-phase gc still used the open and recover path.  This causes contention
between GC and the writer, especially for hotly written logs.

## Test plan

CI

## Documentation Changes

N/A

Author: rescrv

rust/garbage_collector/src/operators/delete_unused_logs.rs

@@ -11,7 +11,7 @@ use chroma_system::{Operator, OperatorType};
 use chroma_types::CollectionUuid;
 use futures::future::try_join_all;
 use thiserror::Error;
-use wal3::{GarbageCollectionOptions, LogPosition, LogWriter, LogWriterOptions};
+use wal3::{GarbageCollectionOptions, GarbageCollector, LogPosition, LogWriterOptions};
 
 #[derive(Clone, Debug)]
 pub struct DeleteUnusedLogsOperator {
@@ -67,12 +67,11 @@ impl Operator<DeleteUnusedLogsInput, DeleteUnusedLogsOutput> for DeleteUnusedLog
                 let storage_clone = storage_arc.clone();
                 let mut logs = self.logs.clone();
                 log_gc_futures.push(async move {
-                    let writer = match LogWriter::open(
+                    let writer = match GarbageCollector::open(
                         LogWriterOptions::default(),
                         storage_clone,
                         &collection_id.storage_prefix_for_log(),
                         "garbage collection service",
-                        (),
                     )
                     .await
                     {

rust/garbage_collector/src/operators/truncate_dirty_log.rs

@@ -7,7 +7,7 @@ use chroma_storage::Storage;
 use chroma_system::{Operator, OperatorType};
 use futures::future::try_join_all;
 use thiserror::Error;
-use wal3::{GarbageCollectionOptions, LogWriter, LogWriterOptions};
+use wal3::{GarbageCollectionOptions, GarbageCollector, LogWriterOptions};
 
 #[derive(Clone, Debug)]
 pub struct TruncateDirtyLogOperator {
@@ -56,12 +56,11 @@ impl Operator<TruncateDirtyLogInput, TruncateDirtyLogOutput> for TruncateDirtyLo
         let mut replica_id = 0u64;
         loop {
             let dirty_log_prefix = format!("dirty-rust-log-service-{replica_id}");
-            match LogWriter::open(
+            match GarbageCollector::open(
                 LogWriterOptions::default(),
                 storage_arc.clone(),
                 &dirty_log_prefix,
                 "garbage collection service",
-                (),
             )
             .await
             {

rust/wal3/src/gc.rs

@@ -10,9 +10,11 @@ use chroma_storage::{
 };
 
 use crate::manifest::unprefixed_snapshot_path;
+use crate::writer::OnceLogWriter;
 use crate::{
     deserialize_setsum, prefixed_fragment_path, serialize_setsum, Error, Fragment, FragmentSeqNo,
-    LogPosition, Manifest, ScrubError, Snapshot, SnapshotCache, SnapshotPointer, ThrottleOptions,
+    GarbageCollectionOptions, LogPosition, LogWriterOptions, Manifest, ScrubError, Snapshot,
+    SnapshotCache, SnapshotPointer, ThrottleOptions,
 };
 
 ////////////////////////////////////////////// Garbage /////////////////////////////////////////////
@@ -467,3 +469,48 @@ impl Garbage {
         Ok(drop_acc)
     }
 }
+
+///////////////////////////////////////// GarbageCollector /////////////////////////////////////////
+
+pub struct GarbageCollector {
+    log: Arc<OnceLogWriter>,
+}
+
+impl GarbageCollector {
+    /// Open the log into a state where it can be garbage collected.
+    pub async fn open(
+        options: LogWriterOptions,
+        storage: Arc<Storage>,
+        prefix: &str,
+        writer: &str,
+    ) -> Result<Self, Error> {
+        let log = OnceLogWriter::open_for_read_only_and_stale_ops(
+            options.clone(),
+            Arc::clone(&storage),
+            prefix.to_string(),
+            writer.to_string(),
+            Arc::new(()),
+        )
+        .await?;
+        Ok(Self { log })
+    }
+
+    pub async fn garbage_collect_phase1_compute_garbage(
+        &self,
+        options: &GarbageCollectionOptions,
+        keep_at_least: Option<LogPosition>,
+    ) -> Result<bool, Error> {
+        self.log
+            .garbage_collect_phase1_compute_garbage(options, keep_at_least)
+            .await
+    }
+
+    pub async fn garbage_collect_phase3_delete_garbage(
+        &self,
+        options: &GarbageCollectionOptions,
+    ) -> Result<(), Error> {
+        self.log
+            .garbage_collect_phase3_delete_garbage(options)
+            .await
+    }
+}

rust/wal3/src/lib.rs

@@ -22,7 +22,7 @@ pub use batch_manager::BatchManager;
 pub use copy::copy;
 pub use cursors::{Cursor, CursorName, CursorStore, Witness};
 pub use destroy::destroy;
-pub use gc::Garbage;
+pub use gc::{Garbage, GarbageCollector};
 pub use manifest::{unprefixed_snapshot_path, Manifest, Snapshot, SnapshotPointer};
 pub use manifest_manager::ManifestManager;
 pub use reader::{Limits, LogReader};

rust/wal3/src/writer.rs

@@ -491,7 +491,7 @@ impl Drop for LogWriter {
 /// structure as the recovery procedure does, this allows us to re-use exactly one code path for
 /// both.  That code path can then be well-tested because any contention state gets exercised from
 /// the perspective of initialization.
-struct OnceLogWriter {
+pub(crate) struct OnceLogWriter {
     /// LogWriterOptions are fixed at log creation time.
     /// LogWriter is intentionally cheap to construct and destroy.
     /// Reopen the log to change the options.
@@ -573,6 +573,35 @@ impl OnceLogWriter {
         Ok(this)
     }
 
+    pub(crate) async fn open_for_read_only_and_stale_ops(
+        options: LogWriterOptions,
+        storage: Arc<Storage>,
+        prefix: String,
+        writer: String,
+        mark_dirty: Arc<dyn MarkDirty>,
+    ) -> Result<Arc<Self>, Error> {
+        let done = AtomicBool::new(false);
+        let batch_manager =
+            BatchManager::new(options.throttle_fragment).ok_or_else(|| Error::Internal)?;
+        let manifest_manager = ManifestManager::new(
+            options.throttle_manifest,
+            options.snapshot_manifest,
+            Arc::clone(&storage),
+            prefix.clone(),
+            writer,
+        )
+        .await?;
+        Ok(Arc::new(Self {
+            options,
+            storage,
+            prefix,
+            done,
+            mark_dirty,
+            manifest_manager,
+            batch_manager,
+        }))
+    }
+
     fn shutdown(&self) {
         self.batch_manager.shutdown();
         self.manifest_manager.shutdown();
@@ -727,7 +756,7 @@ impl OnceLogWriter {
     /// Returns Ok(false) if there is no garbage to act upon (e.g., it's already been collected).
     /// Returns Ok(true) if there is garbage to act upon.
     #[tracing::instrument(skip(self, options))]
-    async fn garbage_collect_phase1_compute_garbage(
+    pub(crate) async fn garbage_collect_phase1_compute_garbage(
         &self,
         options: &GarbageCollectionOptions,
         keep_at_least: Option<LogPosition>,
@@ -799,7 +828,7 @@ impl OnceLogWriter {
     /// Post-condition:
     /// - contents of gc/GARBAGE are removed from manifest/MANIFEST.
     #[tracing::instrument(skip(self, _options))]
-    async fn garbage_collect_phase2_update_manifest(
+    pub(crate) async fn garbage_collect_phase2_update_manifest(
         &self,
         _options: &GarbageCollectionOptions,
     ) -> Result<(), Error> {
@@ -828,7 +857,7 @@ impl OnceLogWriter {
     /// Post-condition:
     /// - gc/GARBAGE and the files it references get deleted.
     #[tracing::instrument(skip(self, options))]
-    async fn garbage_collect_phase3_delete_garbage(
+    pub(crate) async fn garbage_collect_phase3_delete_garbage(
         &self,
         options: &GarbageCollectionOptions,
     ) -> Result<(), Error> {
@@ -907,7 +936,7 @@ impl OnceLogWriter {
     }
 
     #[tracing::instrument(skip(self))]
-    async fn garbage_collect(
+    pub(crate) async fn garbage_collect(
         &self,
         options: &GarbageCollectionOptions,
         keep_at_least: Option<LogPosition>,