[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

@@ -53,7 +53,7 @@ impl InMemoryLog {
                 next_offset, log.log_offset
             );
         }
-        logs.push(log);
+        logs.push(log.clone());
     }
 }
 

rust/log/src/in_memory_log.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

@@ -79,8 +79,10 @@ shuttle = { workspace = true }
 rand = { workspace = true }
 rand_xorshift = { workspace = true }
 tempfile = { workspace = true }
+reqwest = { workspace = true, features = ["json"] }
 
 chroma-benchmark = { workspace = true }
+chroma-frontend = { workspace = true }
 
 [[bench]]
 name = "filter"