[ENH][wal3] Make GC be delete-free for log/gc/GARBAGE (#4904)

## Description of changes

AWS doesn't support delete of things with an If-Match header except in
directory buckets and we are not using those.  Instead overwrite using
the same hand-over-hand pattern as the manifest.  Load a GC, decide
what to do while holding on the the ETag, and then overwrite the garbage
with a sentinel rather than deleting it.

## Test plan

CI

- [wal3] Tests pass locally with `pytest` for python, `yarn test` for
js, `cargo test` for rust

## Documentation Changes

N/A

Author: rescrv

rust/wal3/src/gc.rs

@@ -17,7 +17,7 @@ use crate::{
 
 ////////////////////////////////////////////// Garbage /////////////////////////////////////////////
 
-#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
+#[derive(Debug, Clone, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
 pub struct Garbage {
     pub snapshots_to_drop: Vec<SnapshotPointer>,
     pub snapshots_to_make: Vec<Snapshot>,
@@ -33,6 +33,22 @@ pub struct Garbage {
 }
 
 impl Garbage {
+    pub fn empty() -> Self {
+        Garbage {
+            snapshots_to_drop: Vec::new(),
+            snapshots_to_make: Vec::new(),
+            snapshot_for_root: None,
+            fragments_to_drop_start: FragmentSeqNo(0),
+            fragments_to_drop_limit: FragmentSeqNo(0),
+            setsum_to_discard: Setsum::default(),
+            first_to_keep: LogPosition::from_offset(1),
+        }
+    }
+
+    pub fn is_empty(&self) -> bool {
+        *self == Self::empty()
+    }
+
     pub fn path(prefix: &str) -> String {
         format!("{}/gc/GARBAGE", prefix)
     }
@@ -151,32 +167,62 @@ impl Garbage {
         options: &ThrottleOptions,
         storage: &Storage,
         prefix: &str,
+        existing: Option<&ETag>,
+    ) -> Result<Option<ETag>, Error> {
+        self.install_new_snapshots(storage, prefix, options).await?;
+        Self::transition(options, storage, prefix, existing, self).await
+    }
+
+    #[tracing::instrument(skip(self, storage), err(Display))]
+    pub async fn reset(
+        &self,
+        options: &ThrottleOptions,
+        storage: &Storage,
+        prefix: &str,
+        existing: &ETag,
     ) -> Result<Option<ETag>, Error> {
         self.install_new_snapshots(storage, prefix, options).await?;
+        match Self::transition(options, storage, prefix, Some(existing), &Self::empty()).await {
+            Ok(e_tag) => Ok(e_tag),
+            Err(Error::LogContentionFailure) => Ok(None),
+            Err(err) => Err(err),
+        }
+    }
+
+    async fn transition(
+        options: &ThrottleOptions,
+        storage: &Storage,
+        prefix: &str,
+        existing: Option<&ETag>,
+        replacement: &Self,
+    ) -> Result<Option<ETag>, Error> {
         let exp_backoff = crate::backoff::ExponentialBackoff::new(
             options.throughput as f64,
             options.headroom as f64,
         );
         let mut retry_count = 0;
         loop {
             let path = Self::path(prefix);
-            let payload = serde_json::to_string(&self)
+            let payload = serde_json::to_string(replacement)
                 .map_err(|e| {
                     Error::CorruptManifest(format!("could not encode JSON garbage: {e:?}"))
                 })?
                 .into_bytes();
-            let options = PutOptions::if_not_exists(StorageRequestPriority::P0);
+            let options = if let Some(e_tag) = existing {
+                PutOptions::if_matches(e_tag, StorageRequestPriority::P0)
+            } else {
+                PutOptions::if_not_exists(StorageRequestPriority::P0)
+            };
             match storage.put_bytes(&path, payload, options).await {
                 Ok(e_tag) => return Ok(e_tag),
                 Err(StorageError::Precondition { path: _, source: _ }) => {
-                    // NOTE(rescrv):  We know that someone put the file before us, and therefore we
-                    // know this write failed.  Because the garbage file is created and deleted
-                    // we cannot just overwrite, so fail with log contention and let higher level
-                    // protocol decide.
-                    return Err(Error::LogContentionRetry);
+                    // NOTE(rescrv):  We know that we put the file.  The e_tag no longer matches.
+                    // Therefore, we know someone else transitioned the file and our reset should
+                    // be a NOP.
+                    return Err(Error::LogContentionFailure);
                 }
                 Err(e) => {
-                    tracing::error!("error uploading manifest: {e:?}");
+                    tracing::error!("error uploading garbage: {e:?}");
                     let backoff = exp_backoff.next();
                     if backoff > Duration::from_secs(60) || retry_count >= 3 {
                         return Err(Arc::new(e).into());

rust/wal3/src/writer.rs

@@ -5,7 +5,7 @@ use std::time::{Duration, Instant, SystemTime};
 
 use arrow::array::{ArrayRef, BinaryArray, RecordBatch, UInt64Array};
 use chroma_storage::admissioncontrolleds3::StorageRequestPriority;
-use chroma_storage::{DeleteOptions, GetOptions, PutOptions, Storage, StorageError};
+use chroma_storage::{DeleteOptions, ETag, GetOptions, PutOptions, Storage, StorageError};
 use parquet::arrow::ArrowWriter;
 use parquet::basic::Compression;
 use parquet::file::properties::WriterProperties;
@@ -582,6 +582,15 @@ impl OnceLogWriter {
 
     #[tracing::instrument(skip(self))]
     async fn garbage_collect(&self, options: &GarbageCollectionOptions) -> Result<(), Error> {
+        self.garbage_collect_recursive(options, false, None).await
+    }
+
+    async fn garbage_collect_recursive(
+        &self,
+        options: &GarbageCollectionOptions,
+        base: bool,
+        existing: Option<&ETag>,
+    ) -> Result<(), Error> {
         let cutoff = self.garbage_collection_cutoff().await?;
         self.manifest_manager.heartbeat().await?;
         let garbage = self
@@ -590,16 +599,40 @@ impl OnceLogWriter {
             .compute_garbage(options, cutoff, &())
             .await?;
         let (garbage, e_tag) = match garbage
-            .install(&self.options.throttle_manifest, &self.storage, &self.prefix)
+            .install(
+                &self.options.throttle_manifest,
+                &self.storage,
+                &self.prefix,
+                existing,
+            )
             .await
         {
             Ok(e_tag) => (garbage, e_tag),
-            Err(Error::LogContentionRetry) => {
+            Err(Error::LogContentionFailure)
+            | Err(Error::LogContentionRetry)
+            | Err(Error::LogContentionDurable) => {
                 match Garbage::load(&self.options.throttle_manifest, &self.storage, &self.prefix)
                     .await
                 {
-                    Ok(Some((garbage, e_tag))) => (garbage, e_tag),
-                    Ok(None) => return Err(Error::LogContentionFailure),
+                    Ok(Some((garbage, e_tag))) => {
+                        if garbage.is_empty() {
+                            if base {
+                                return Err(Error::LogContentionRetry);
+                            } else {
+                                return Box::pin(self.garbage_collect_recursive(
+                                    options,
+                                    true,
+                                    e_tag.as_ref(),
+                                ))
+                                .await;
+                            }
+                        } else {
+                            (garbage, e_tag)
+                        }
+                    }
+                    Ok(None) => {
+                        return Err(Error::LogContentionFailure);
+                    }
                     Err(err) => {
                         return Err(err);
                     }
@@ -609,10 +642,15 @@ impl OnceLogWriter {
                 return Err(err);
             }
         };
+        let Some(e_tag) = e_tag else {
+            return Err(Error::GarbageCollection(
+                "installed garbage without an ETag".to_string(),
+            ));
+        };
         let paths = garbage
             .prefixed_paths_to_delete(&self.prefix)
             .collect::<Vec<_>>();
-        self.manifest_manager.apply_garbage(garbage).await?;
+        self.manifest_manager.apply_garbage(garbage.clone()).await?;
         let exp_backoff: ExponentialBackoff = options.throttle.into();
         let start = Instant::now();
         for path in paths {
@@ -637,33 +675,14 @@ impl OnceLogWriter {
                 }
             }
         }
-        loop {
-            let options = if let Some(e_tag) = e_tag.as_ref() {
-                DeleteOptions::if_matches(e_tag, StorageRequestPriority::P0)
-            } else {
-                DeleteOptions::default()
-            };
-            match self
-                .storage
-                .delete(&Garbage::path(&self.prefix), options)
-                .await
-            {
-                Ok(()) => break,
-                Err(StorageError::NotFound { .. }) => break,
-                Err(err) => {
-                    tracing::error!("could not cleanup garbage: {err:?}");
-                    if start.elapsed() > Duration::from_secs(600) {
-                        tracing::error!("could not cleanup garbage within 10 minutes, returning");
-                        return Err(Error::StorageError(Arc::new(err)));
-                    }
-                    let mut backoff = exp_backoff.next();
-                    if backoff > Duration::from_secs(600) {
-                        backoff = Duration::from_secs(600);
-                    }
-                    tokio::time::sleep(backoff).await;
-                }
-            }
-        }
+        garbage
+            .reset(
+                &self.options.throttle_manifest,
+                &self.storage,
+                &self.prefix,
+                &e_tag,
+            )
+            .await?;
         Ok(())
     }