From 0c46f473aacc7d41dabd0f612be314f15df3cb08 Mon Sep 17 00:00:00 2001 From: Robert Escriva Date: Fri, 24 Apr 2026 13:31:06 -0700 Subject: [PATCH] [CHORE](garbage_collector): remove log-only orchestrator Excise the HardDeleteLogOnlyGarbageCollectorOrchestrator, which was a separate code path for hard-deleting logs of destroyed collections. Manual GC requests now look up collection info via sysdb and feed them into the regular GC pipeline. Collections that no longer exist are skipped with a log message instead of being routed to a separate orchestrator. - Replace HashMap with HashSet for manual_collections since the database name is now looked up on demand - Add lookup_manual_collection_to_gc to query sysdb directly - Remove second job stream and its duplicated result processing loop - Add test for lookup returning None after collection removal Co-authored-by: AI --- .../src/garbage_collector_component.rs | 182 ++++++------ rust/garbage_collector/src/lib.rs | 1 - .../src/log_only_orchestrator.rs | 266 ------------------ 3 files changed, 101 insertions(+), 348 deletions(-) delete mode 100644 rust/garbage_collector/src/log_only_orchestrator.rs diff --git a/rust/garbage_collector/src/garbage_collector_component.rs b/rust/garbage_collector/src/garbage_collector_component.rs index 1d4a5fb8836..666305bc428 100644 --- a/rust/garbage_collector/src/garbage_collector_component.rs +++ b/rust/garbage_collector/src/garbage_collector_component.rs @@ -1,4 +1,4 @@ -use std::collections::HashMap; +use std::collections::HashSet; use std::sync::Arc; use crate::operators::truncate_dirty_log::{ @@ -15,9 +15,7 @@ use chroma_error::ChromaError; use chroma_log::Log; use chroma_memberlist::memberlist_provider::Memberlist; use chroma_storage::Storage; -use chroma_sysdb::{ - CollectionToGcInfo, GetCollectionsOptions, GetCollectionsToGcError, SysDb, SysDbConfig, -}; +use chroma_sysdb::{CollectionToGcInfo, GetCollectionsOptions, SysDb, SysDbConfig}; use chroma_system::{ wrap, Component, ComponentContext, ComponentHandle, Dispatcher, Handler, Orchestrator, System, TaskResult, @@ -54,7 +52,7 @@ pub(crate) struct GarbageCollector { job_duration_ms_metric: Histogram, total_files_deleted_metric: Counter, total_versions_deleted_metric: Counter, - manual_collections: Mutex>, + manual_collections: Mutex>, } impl Debug for GarbageCollector { @@ -116,7 +114,7 @@ impl GarbageCollector { .u64_counter("garbage_collector.total_versions_deleted") .with_description("Total number of versions deleted during garbage collection") .build(), - manual_collections: Mutex::new(HashMap::default()), + manual_collections: Mutex::new(HashSet::default()), } } @@ -128,40 +126,6 @@ impl GarbageCollector { self.system = Some(system); } - async fn garbage_collect_hard_delete_log( - &self, - collection_id: CollectionUuid, - database_name: Option, - ) -> Result { - let dispatcher = self - .dispatcher - .as_ref() - .ok_or(GarbageCollectCollectionError::Uninitialized)?; - let system = self - .system - .as_ref() - .ok_or(GarbageCollectCollectionError::Uninitialized)?; - - let orchestrator = - crate::log_only_orchestrator::HardDeleteLogOnlyGarbageCollectorOrchestrator::new( - dispatcher.clone(), - self.storage.clone(), - self.logs.clone(), - self.regions_and_topologies.clone(), - collection_id, - database_name, - ); - - let result = match orchestrator.run(system.clone()).await { - Ok(res) => res, - Err(e) => { - tracing::error!("Failed to run garbage collection orchestrator v2: {:?}", e); - return Err(GarbageCollectCollectionError::OrchestratorV2Error(e)); - } - }; - Ok(result) - } - async fn garbage_collect_attached_functions( &mut self, attached_function_gc_absolute_cutoff_time: SystemTime, @@ -315,6 +279,49 @@ impl GarbageCollector { Ok(result) } + async fn lookup_manual_collection_to_gc( + &mut self, + collection_id: CollectionUuid, + ) -> Result, GarbageCollectCollectionError> { + let mut collections = self + .sysdb_client + .get_collections(GetCollectionsOptions { + collection_id: Some(collection_id), + ..Default::default() + }) + .await?; + + if collections.is_empty() { + return Ok(None); + } + if collections.len() > 1 { + tracing::error!( + "Multiple collections returned when querying for manual GC ID: {}", + collection_id + ); + return Ok(None); + } + + let collection = collections.remove(0); + let Some(database) = DatabaseName::new(&collection.database) else { + tracing::error!( + "Collection {} has invalid database name for manual GC: {}", + collection.collection_id, + collection.database + ); + return Ok(None); + }; + + Ok(Some(CollectionToGcInfo { + id: collection.collection_id, + tenant: collection.tenant, + database, + name: collection.name, + version_file_path: collection.version_file_path.unwrap_or_default(), + lineage_file_path: collection.lineage_file_path, + })) + } + async fn truncate_dirty_log(&self, ctx: &ComponentContext) { let Some(mut dispatcher) = self.dispatcher.as_ref().cloned() else { tracing::error!("Uninitialized dispatcher for garbage collector"); @@ -420,11 +427,10 @@ impl GarbageCollector { return Err(GarbageCollectCollectionError::NoSuchCollection); } let mut manual_collections = self.manual_collections.lock(); - if let Some(database_name) = DatabaseName::new(&collection_info[0].database) { - manual_collections.insert(collection_id, database_name); - } else { + if DatabaseName::new(&collection_info[0].database).is_none() { return Err(GarbageCollectCollectionError::NoSuchCollection); } + manual_collections.insert(collection_id); Ok(()) } } @@ -548,21 +554,20 @@ impl Handler for GarbageCollector { while collections_to_gc.len() + manual.len() < self.config.max_collections_to_gc as usize { - if let Some(&c) = manual_collections.keys().next() { - let db_name = manual_collections.remove(&c); - manual.push((c, db_name)); + if let Some(&c) = manual_collections.iter().next() { + manual_collections.remove(&c); + manual.push(c); } else { break; } } } - let mut collections_to_hard_delete_log = vec![]; - for (collection_id, db_name) in manual { + for collection_id in manual { if collections_to_gc.iter().any(|c| c.id == collection_id) { continue; } - match self.sysdb_client.get_collection_to_gc(collection_id).await { - Ok(collection_info) => { + match self.lookup_manual_collection_to_gc(collection_id).await { + Ok(Some(collection_info)) => { tracing::event!( Level::INFO, name = "manually collecting", @@ -570,8 +575,11 @@ impl Handler for GarbageCollector { ); collections_to_gc.push(collection_info); } - Err(GetCollectionsToGcError::NoSuchCollection) => { - collections_to_hard_delete_log.push((collection_id, db_name)); + Ok(None) => { + tracing::info!( + collection_id = %collection_id, + "Skipping manual GC request because the collection no longer exists" + ); } Err(err) => { tracing::event!( @@ -632,14 +640,7 @@ impl Handler for GarbageCollector { ) .instrument(instrumented_span)) as std::pin::Pin> + Send + '_>> }); - let jobs_iter2 = collections_to_hard_delete_log.into_iter().map(|(collection_id, db_name)| { - tracing::event!(Level::INFO, "hard delete log-only"); - let instrumented_span = span!(parent: None, tracing::Level::INFO, "Garbage collection job (hard delete log)", collection_id =? collection_id); - Span::current().add_link(instrumented_span.context().span().span_context().clone()); - Box::pin(self.garbage_collect_hard_delete_log(collection_id, db_name).instrument(instrumented_span)) as std::pin::Pin> + Send + '_>> - }); let mut jobs_stream1 = futures::stream::iter(jobs_iter1).buffer_unordered(100); - let mut jobs_stream2 = futures::stream::iter(jobs_iter2).buffer_unordered(100); let mut num_completed_jobs = 0; let mut num_failed_jobs = 0; @@ -659,25 +660,6 @@ impl Handler for GarbageCollector { } } } - // NOTE(rescrv): I'm not proud of this duplication, but I cannot coerce the - // futures::stream::iter above to take a chain of two different futures. It just won't - // compile. - while let Some(job_result) = jobs_stream2.next().await { - match job_result { - Ok(result) => { - { - let mut manual_collections = self.manual_collections.lock(); - manual_collections.remove(&result.collection_id); - } - tracing::info!("Garbage collection hard delete completed. Deleted all log files collection {}.", result.collection_id); - num_completed_jobs += 1; - } - Err(e) => { - tracing::error!("Garbage collection failed: {:?}", e); - num_failed_jobs += 1; - } - } - } tracing::info!( "Completed {} jobs, failed {} jobs", num_completed_jobs, @@ -968,12 +950,11 @@ mod tests { let (_storage_dir, storage) = test_storage(); let mut sysdb = SysDb::Test(TestSysDb::new()); let collection_id = CollectionUuid::new(); - let database_name = DatabaseName::new("test_db").expect("valid database name"); sysdb .create_collection( "test-tenant".to_string(), - database_name.clone(), + DatabaseName::new("test_db").expect("valid database name"), collection_id, "test-collection".to_string(), vec![], @@ -1000,7 +981,7 @@ mod tests { assert_eq!( garbage_collector.manual_collections.lock().clone(), - HashMap::from([(collection_id, database_name)]) + HashSet::from([collection_id]) ); } @@ -1025,6 +1006,45 @@ mod tests { )); } + #[tokio::test] + async fn test_lookup_manual_collection_to_gc_returns_none_after_collection_is_removed() { + let (_storage_dir, storage) = test_storage(); + let mut test_sysdb = TestSysDb::new(); + let collection_id = CollectionUuid::new(); + let database_name = DatabaseName::new("test_db").expect("valid database name"); + + SysDb::Test(test_sysdb.clone()) + .create_collection( + "test-tenant".to_string(), + database_name, + collection_id, + "test-collection".to_string(), + vec![], + None, + None, + None, + None, + false, + ) + .await + .expect("collection should be created"); + test_sysdb.remove_collection(collection_id); + + let mut garbage_collector = new_test_garbage_collector( + test_gc_config("gc-a"), + SysDb::Test(test_sysdb), + storage, + Box::new(TestAssignmentPolicy::default()), + ); + + let collection_to_gc = garbage_collector + .lookup_manual_collection_to_gc(collection_id) + .await + .expect("manual collection lookup should succeed"); + + assert!(collection_to_gc.is_none()); + } + async fn wait_for_new_version( clients: &mut ChromaGrpcClients, collection_id: String, diff --git a/rust/garbage_collector/src/lib.rs b/rust/garbage_collector/src/lib.rs index 5305f5d3eff..88157fe0d88 100644 --- a/rust/garbage_collector/src/lib.rs +++ b/rust/garbage_collector/src/lib.rs @@ -25,7 +25,6 @@ pub mod config; mod construct_version_graph_orchestrator; mod garbage_collector_component; pub mod garbage_collector_orchestrator_v2; -mod log_only_orchestrator; pub mod mcmr; #[cfg(test)] diff --git a/rust/garbage_collector/src/log_only_orchestrator.rs b/rust/garbage_collector/src/log_only_orchestrator.rs deleted file mode 100644 index a04a08c5f78..00000000000 --- a/rust/garbage_collector/src/log_only_orchestrator.rs +++ /dev/null @@ -1,266 +0,0 @@ -use crate::garbage_collector_orchestrator_v2::GarbageCollectorError; -use crate::mcmr::RegionsAndTopologies; -use crate::operators::delete_unused_logs::{ - DeleteUnusedLogsError, DeleteUnusedLogsInput, DeleteUnusedLogsOperator, DeleteUnusedLogsOutput, -}; -use crate::types::{CleanupMode, GarbageCollectorResponse}; -use async_trait::async_trait; -use chroma_log::Log; -use chroma_storage::Storage; -use chroma_system::{ - wrap, ComponentContext, ComponentHandle, Dispatcher, Handler, Orchestrator, - OrchestratorContext, TaskResult, -}; -use chroma_types::{CollectionUuid, DatabaseName}; -use std::collections::{HashMap, HashSet}; -use std::sync::Arc; -use tokio::sync::oneshot::Sender; -use tracing::{Level, Span}; - -#[derive(Debug)] -pub struct HardDeleteLogOnlyGarbageCollectorOrchestrator { - context: OrchestratorContext, - storage: Storage, - logs: Log, - regions_and_topologies: Option>, - result_channel: Option>>, - collection_to_destroy: CollectionUuid, - database_name: Option, -} - -#[allow(clippy::too_many_arguments)] -impl HardDeleteLogOnlyGarbageCollectorOrchestrator { - pub fn new( - dispatcher: ComponentHandle, - storage: Storage, - logs: Log, - regions_and_topologies: Option>, - collection_to_destroy: CollectionUuid, - database_name: Option, - ) -> Self { - Self { - context: OrchestratorContext::new(dispatcher), - storage, - logs, - regions_and_topologies, - result_channel: None, - collection_to_destroy, - database_name, - } - } -} - -#[async_trait] -impl Orchestrator for HardDeleteLogOnlyGarbageCollectorOrchestrator { - type Output = GarbageCollectorResponse; - type Error = GarbageCollectorError; - - fn dispatcher(&self) -> ComponentHandle { - self.context.dispatcher.clone() - } - - fn context(&self) -> &OrchestratorContext { - &self.context - } - - async fn on_start(&mut self, ctx: &ComponentContext) { - let _ = self - .try_start_delete_unused_logs_operator(ctx) - .await - .inspect_err(|_| { - tracing::event!( - Level::ERROR, - "could not start job to hard delete unused logs", - ) - }); - } - - fn set_result_channel( - &mut self, - sender: Sender>, - ) { - self.result_channel = Some(sender); - } - - fn take_result_channel( - &mut self, - ) -> Option>> { - self.result_channel.take() - } -} - -impl HardDeleteLogOnlyGarbageCollectorOrchestrator { - async fn try_start_delete_unused_logs_operator( - &mut self, - ctx: &ComponentContext, - ) -> Result<(), GarbageCollectorError> { - let collections_to_destroy = - HashSet::from_iter(vec![self.collection_to_destroy].into_iter()); - let collections_to_garbage_collect = HashMap::new(); - let task = wrap( - Box::new(DeleteUnusedLogsOperator { - enabled: true, - mode: CleanupMode::DeleteV2, - storage: self.storage.clone(), - logs: self.logs.clone(), - regions_and_topologies: self.regions_and_topologies.clone(), - enable_dangerous_option_to_ignore_min_versions_for_wal3: false, - }), - DeleteUnusedLogsInput { - collections_to_destroy, - collections_to_garbage_collect, - database_name: self.database_name.clone(), - }, - ctx.receiver(), - self.context.task_cancellation_token.clone(), - ); - self.dispatcher() - .send(task, Some(Span::current())) - .await - .map_err(GarbageCollectorError::Channel)?; - Ok(()) - } -} - -#[async_trait] -impl Handler> - for HardDeleteLogOnlyGarbageCollectorOrchestrator -{ - type Result = (); - - async fn handle( - &mut self, - message: TaskResult, - ctx: &ComponentContext, - ) { - let _output = match self.ok_or_terminate(message.into_inner(), ctx).await { - Some(output) => output, - None => return, - }; - self.terminate_with_result( - Ok(GarbageCollectorResponse { - collection_id: self.collection_to_destroy, - num_versions_deleted: 0, - num_files_deleted: 0, - ..Default::default() - }), - ctx, - ) - .await; - } -} - -#[cfg(test)] -mod tests { - //! Test suite for the `HardDeleteLogOnlyGarbageCollectorOrchestrator`. - //! - //! This module verifies the core functionality of the hard delete orchestrator, - //! which is responsible for permanently removing log data for destroyed collections. - //! The tests ensure proper initialization, configuration, and trait implementation - //! of the orchestrator component. - //! - //! # Test Coverage - //! - //! The test suite validates: - //! - Correct initialization with required dependencies - //! - Proper storage of collection UUID for destruction - //! - Result channel lifecycle management - //! - Orchestrator trait contract fulfillment - //! - //! # Testing Approach - //! - //! Tests use mock components (test storage, dispatcher, logs) to isolate - //! orchestrator behavior without requiring actual I/O operations. - //! Each test is self-contained and can run in parallel using tokio's - //! multi-threaded runtime. - use super::*; - use chroma_config::registry::Registry; - use chroma_config::Configurable; - use chroma_log::config::{GrpcLogConfig, LogConfig}; - use chroma_storage::test_storage; - use chroma_system::{Dispatcher, System}; - - /// Verifies that the orchestrator correctly initializes with all required components. - /// - /// This test ensures that when creating a new `HardDeleteLogOnlyGarbageCollectorOrchestrator`, - /// all provided dependencies (dispatcher, storage, logs, collection UUID) are properly - /// stored and the result channel starts in an uninitialized state. - /// - /// # Test Invariants - /// - /// - Collection UUID must match the one provided during construction - /// - Result channel must be `None` initially (set later by the system) - #[tokio::test(flavor = "multi_thread")] - async fn test_k8s_integration_orchestrator_initialization() { - let (_storage_dir, storage) = test_storage(); - let system = System::new(); - let dispatcher = Dispatcher::new(Default::default()); - let dispatcher_handle = system.start_component(dispatcher); - let registry = Registry::new(); - let log_config = LogConfig::Grpc(GrpcLogConfig::default()); - let logs = Log::try_from_config(&(log_config, system.clone()), ®istry) - .await - .unwrap(); - let collection_to_destroy = CollectionUuid::new(); - - // Create orchestrator with test dependencies - let orchestrator = HardDeleteLogOnlyGarbageCollectorOrchestrator::new( - dispatcher_handle.clone(), - storage.clone(), - logs.clone(), - None, - collection_to_destroy, - None, - ); - - // Verify the orchestrator is properly initialized - assert_eq!(orchestrator.collection_to_destroy, collection_to_destroy); - assert!(orchestrator.result_channel.is_none()); - } - - /// Validates that the orchestrator correctly stores the collection UUID for hard deletion. - /// - /// This test verifies that the orchestrator preserves the collection UUID that will be - /// passed to the `DeleteUnusedLogsOperator` when `on_start` is called. It also documents - /// the hardcoded configuration that will be used for the delete operation. - /// - /// # Implementation Details - /// - /// When the orchestrator starts the delete operator (in `try_start_delete_unused_logs_operator`), - /// it uses the following hardcoded configuration: - /// - `enabled`: true (operator is active) - /// - `mode`: `CleanupMode::DeleteV2` (performs hard deletion) - /// - `enable_dangerous_option_to_ignore_min_versions_for_wal3`: false (safety check enabled) - /// - /// The collection UUID stored in `collection_to_destroy` is placed in the - /// `collections_to_destroy` set, while `collections_to_garbage_collect` remains empty - /// since this orchestrator only handles hard deletion, not soft garbage collection. - #[tokio::test(flavor = "multi_thread")] - async fn test_k8s_integration_delete_operator_params() { - let (_storage_dir, storage) = test_storage(); - let system = System::new(); - let dispatcher = Dispatcher::new(Default::default()); - let dispatcher_handle = system.start_component(dispatcher); - let registry = Registry::new(); - let log_config = LogConfig::Grpc(GrpcLogConfig::default()); - let logs = Log::try_from_config(&(log_config, system.clone()), ®istry) - .await - .unwrap(); - let collection_to_destroy = CollectionUuid::new(); - - let orchestrator = HardDeleteLogOnlyGarbageCollectorOrchestrator::new( - dispatcher_handle, - storage.clone(), - logs.clone(), - None, - collection_to_destroy, - None, - ); - - // Verify the orchestrator stores correct collection UUID for destruction - assert_eq!(orchestrator.collection_to_destroy, collection_to_destroy); - - // Note: The delete operator configuration is hardcoded in try_start_delete_unused_logs_operator - // and cannot be modified externally. This ensures consistent deletion behavior. - } -}