[ENH]: don't skip GC for collection in fork tree if using GC v2 (#4681)

## Description of changes

Does what the title says. Missed this in the original PR.

## Test plan

_How are these changes tested?_

Updated the basic GC v2 E2E test to include a forked collection.

## Documentation Changes

_Are all docstrings for user-facing APIs updated if required? Do we need
to make documentation changes in the [docs
section](https://github.com/chroma-core/chroma/tree/main/docs/docs.trychroma.com)?_

n/a

Author: codetheweb

rust/garbage_collector/src/garbage_collector_component.rs

@@ -140,21 +140,20 @@ impl GarbageCollector {
             .as_ref()
             .ok_or(GarbageCollectCollectionError::Uninitialized)?;
 
-        let use_v1_orchestrator = match cleanup_mode {
-            CleanupMode::DryRun | CleanupMode::Rename | CleanupMode::Delete => true,
-            CleanupMode::DryRunV2 | CleanupMode::DeleteV2 => false,
-        };
-
-        if use_v1_orchestrator {
+        if cleanup_mode.is_v2() {
             let orchestrator =
-                crate::garbage_collector_orchestrator::GarbageCollectorOrchestrator::new(
+                crate::garbage_collector_orchestrator_v2::GarbageCollectorOrchestrator::new(
                     collection.id,
                     collection.version_file_path,
+                    collection.lineage_file_path,
                     absolute_cutoff_time,
                     self.sysdb_client.clone(),
                     dispatcher.clone(),
+                    system.clone(),
                     self.storage.clone(),
+                    self.root_manager.clone(),
                     cleanup_mode,
+                    2,
                 );
 
             let started_at = SystemTime::now();
@@ -178,23 +177,19 @@ impl GarbageCollector {
                     format!("{:?}", cleanup_mode),
                 )],
             );
+
             return Ok(result);
         }
 
-        let orchestrator =
-            crate::garbage_collector_orchestrator_v2::GarbageCollectorOrchestrator::new(
-                collection.id,
-                collection.version_file_path,
-                collection.lineage_file_path,
-                absolute_cutoff_time,
-                self.sysdb_client.clone(),
-                dispatcher.clone(),
-                system.clone(),
-                self.storage.clone(),
-                self.root_manager.clone(),
-                cleanup_mode,
-                2,
-            );
+        let orchestrator = crate::garbage_collector_orchestrator::GarbageCollectorOrchestrator::new(
+            collection.id,
+            collection.version_file_path,
+            absolute_cutoff_time,
+            self.sysdb_client.clone(),
+            dispatcher.clone(),
+            self.storage.clone(),
+            cleanup_mode,
+        );
 
         let started_at = SystemTime::now();
         let result = orchestrator.run(system.clone()).await?;
@@ -217,7 +212,6 @@ impl GarbageCollector {
                 format!("{:?}", cleanup_mode),
             )],
         );
-
         Ok(result)
     }
 }
@@ -360,9 +354,18 @@ impl Handler<GarbageCollectMessage> for GarbageCollector {
                 continue;
             }
 
-            if collection.lineage_file_path.is_some() {
+            let cleanup_mode = if let Some(tenant_mode_overrides) = &self.tenant_mode_overrides {
+                tenant_mode_overrides
+                    .get(&collection.tenant)
+                    .cloned()
+                    .unwrap_or(self.default_cleanup_mode)
+            } else {
+                self.default_cleanup_mode
+            };
+
+            if collection.lineage_file_path.is_some() && !cleanup_mode.is_v2() {
                 tracing::debug!(
-                    "Skipping garbage collection for root of fork tree: {}",
+                    "Skipping garbage collection for root of fork tree because GC v1 cannot handle fork trees: {}",
                     collection.id
                 );
                 num_skipped_jobs += 1;
@@ -376,17 +379,6 @@ impl Handler<GarbageCollectMessage> for GarbageCollector {
                 collection.version_file_path
             );
 
-            let cleanup_mode = if let Some(tenant_mode_overrides) = &self.tenant_mode_overrides {
-                tenant_mode_overrides
-                    .get(&collection.tenant)
-                    .cloned()
-                    .unwrap_or(self.default_cleanup_mode)
-            } else {
-                self.default_cleanup_mode
-            };
-
-            tracing::info!("Creating gc orchestrator for collection: {}", collection.id);
-
             let instrumented_span = span!(parent: None, tracing::Level::INFO, "Garbage collection job", collection_id = ?collection.id, tenant_id = %collection.tenant, cleanup_mode = ?cleanup_mode);
             instrumented_span.follows_from(Span::current());
 
@@ -1011,6 +1003,26 @@ mod tests {
         let (collection_in_delete_mode, database_name_in_delete_mode) =
             collection_in_delete_mode_handle.await.unwrap();
 
+        // Fork collection in delete mode to give it a lineage file (only GC v2 can handle fork trees)
+        {
+            let source_collection = sysdb
+                .get_collections(Some(collection_in_delete_mode), None, None, None, None, 0)
+                .await
+                .unwrap();
+            let source_collection = source_collection.first().unwrap();
+
+            sysdb
+                .fork_collection(
+                    collection_in_delete_mode,
+                    source_collection.total_records_post_compaction,
+                    source_collection.total_records_post_compaction,
+                    CollectionUuid::new(),
+                    "test-fork".to_string(),
+                )
+                .await
+                .unwrap();
+        }
+
         // Wait 1 second for cutoff time
         tokio::time::sleep(Duration::from_secs(1)).await;
 

rust/garbage_collector/src/types.rs

@@ -19,6 +19,12 @@ pub enum CleanupMode {
     DeleteV2,
 }
 
+impl CleanupMode {
+    pub fn is_v2(&self) -> bool {
+        matches!(self, CleanupMode::DryRunV2 | CleanupMode::DeleteV2)
+    }
+}
+
 #[derive(Debug, Clone, Copy)]
 pub enum VersionStatus {
     #[allow(dead_code)]