[ENH]: Integrate task operators into compaction

- Update compaction orchestration to use task-based execution
- Add task scheduling and execution flow
- Update configs and dependencies
- Integration with heap tender service

Author: tanujnay112

Cargo.lock

@@ -237,7 +237,7 @@ k8s_resource(
 k8s_resource('postgres', resource_deps=['k8s_setup'], labels=["infrastructure"], port_forwards='5432:5432')
 # Jobs are suffixed with the image tag to ensure they are unique. In this context, the image tag is defined in k8s/distributed-chroma/values.yaml.
 k8s_resource('sysdb-migration-latest', resource_deps=['postgres'], labels=["infrastructure"])
-k8s_resource('rust-log-service', labels=["chroma"], port_forwards=['50054:50051', '50052:50052'], resource_deps=['minio-deployment'])
+k8s_resource('rust-log-service', labels=["chroma"], port_forwards=['50054:50051'], resource_deps=['minio-deployment'])
 k8s_resource('sysdb', resource_deps=['sysdb-migration-latest'], labels=["chroma"], port_forwards='50051:50051')
 k8s_resource('rust-frontend-service', resource_deps=['sysdb', 'rust-log-service'], labels=["chroma"], port_forwards='8000:8000')
 k8s_resource('query-service', resource_deps=['sysdb'], labels=["chroma"], port_forwards='50053:50051')

Tiltfile

@@ -7,6 +7,7 @@
 """
 
 import chromadb
+import time
 
 # Connect to Chroma server
 client = chromadb.HttpClient(host="localhost", port=8000)
@@ -60,6 +61,8 @@
 print("Task is now registered and will run on new data!")
 print("=" * 60)
 
+time.sleep(10)
+
 # Add more documents to trigger task execution
 print("\nAdding more documents...")
 collection.add(

examples/task_api_example.py

@@ -1388,7 +1388,10 @@ impl LogServer {
         if records.len() != pull_logs.batch_size as usize
             || (!records.is_empty() && records[0].log_offset != pull_logs.start_from_offset)
         {
-            return Err(Status::not_found("Some entries have been purged"));
+            return Err(Status::not_found(format!(
+                "Some entries have been purged {} versus {}",
+                records[0].log_offset, pull_logs.start_from_offset
+            )));
         }
         Ok(Response::new(PullLogsResponse { records }))
     }

rust/log-service/src/lib.rs

@@ -14,6 +14,7 @@ serde_json = { workspace = true }
 tokio = { workspace = true }
 tonic = { workspace = true }
 tonic-health = { workspace = true }
+tower = { workspace = true }
 tracing = { workspace = true }
 
 chroma-config = { workspace = true }

rust/s3heap-service/Cargo.toml

@@ -6,6 +6,26 @@ use uuid::Uuid;
 
 use crate::CollectionUuid;
 
+define_uuid_newtype!(
+    /// JobId is a wrapper around Uuid to provide a unified type for job identifiers.
+    /// Jobs can be either collection compaction jobs or task execution jobs.
+    JobId,
+    new_v4
+);
+
+// Custom From implementations for JobId
+impl From<CollectionUuid> for JobId {
+    fn from(collection_uuid: CollectionUuid) -> Self {
+        JobId(collection_uuid.0)
+    }
+}
+
+impl From<TaskUuid> for JobId {
+    fn from(task_uuid: TaskUuid) -> Self {
+        JobId(task_uuid.0)
+    }
+}
+
 define_uuid_newtype!(
     /// TaskUuid is a wrapper around Uuid to provide a type for task identifiers.
     #[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]

rust/types/src/task.rs

@@ -79,6 +79,7 @@ shuttle = { workspace = true }
 rand = { workspace = true }
 rand_xorshift = { workspace = true }
 tempfile = { workspace = true }
+reqwest = { workspace = true, features = ["json"] }
 
 chroma-benchmark = { workspace = true }