[ENH] Disable quota and add limit for collection fork (chroma-core#4359)

Author: warpbuild-benchmark-bot[bot]

go/pkg/common/errors.go

@@ -26,6 +26,7 @@ var (
 	ErrCollectionVersionInvalid              = errors.New("collection version invalid")
 	ErrCollectionVersionFileNameStale        = errors.New("collection version file name stale")
 	ErrCollectionEntryIsStale                = errors.New("collection entry is stale - one of version, version_file_name, or log_position is stale")
+	ErrCollectionTooManyFork                 = errors.New("collection entry has too many forks")
 
 	// Collection metadata errors
 	ErrUnknownCollectionMetadataType = errors.New("collection metadata value type not supported")

go/pkg/sysdb/coordinator/table_catalog.go

@@ -894,7 +894,7 @@ func (tc *Catalog) ForkCollection(ctx context.Context, forkCollection *model.For
 		// This scenario could occur during fork because we will reach out to log service first to fork logs. For exampls:
 		// t0: Fork source collection in log with offset [200, 300] (i.e. compaction offset 200, enumeration offset 300)
 		// t1: User writes to source collection, compaction takes place, source collection log offset become [400, 500]
-		// t2: Fork source collection in sysdb, the latest source collection compaction offset is 400. We could not fork this version unless we update the log offsets of the forking collection
+		// t2: Fork source collection in sysdb, the latest source collection compaction offset is 400. If we add new logs, it will start after offset 300, and the data is lost after compaction.
 		latestSourceCompactionOffset := uint64(sourceCollection.LogPosition)
 		if forkCollection.SourceCollectionLogEnumerationOffset < latestSourceCompactionOffset || latestSourceCompactionOffset < forkCollection.SourceCollectionLogCompactionOffset {
 			return common.ErrCollectionLogPositionStale
@@ -953,6 +953,11 @@ func (tc *Catalog) ForkCollection(ctx context.Context, forkCollection *model.For
 		if err != nil {
 			return err
 		}
+		// NOTE: This is a temporary hardcoded limit for the size of the lineage file
+		// TODO: Load the limit value from quota / scorecard, and/or improve the lineage file design to avoid large lineage file
+		if len(lineageFile.Dependencies) > 1000000 {
+			return common.ErrCollectionTooManyFork
+		}
 		lineageFile.Dependencies = append(lineageFile.Dependencies, &coordinatorpb.CollectionVersionDependency{
 			SourceCollectionId:      sourceCollectionIDStr,
 			SourceCollectionVersion: uint64(sourceCollection.Version),

rust/frontend/src/server.rs

@@ -1135,7 +1135,7 @@ pub struct ForkCollectionPayload {
     )
 )]
 async fn fork_collection(
-    headers: HeaderMap,
+    _headers: HeaderMap,
     Path((tenant, database, collection_id)): Path<(String, String, String)>,
     State(mut server): State<FrontendServer>,
     Json(payload): Json<ForkCollectionPayload>,
@@ -1150,17 +1150,19 @@ async fn fork_collection(
     tracing::info!(
         "Forking collection [{collection_id}] in database [{database}] for tenant [{tenant}]"
     );
-    server
-        .authenticate_and_authorize(
-            &headers,
-            AuthzAction::ForkCollection,
-            AuthzResource {
-                tenant: Some(tenant.clone()),
-                database: Some(database.clone()),
-                collection: Some(collection_id.clone()),
-            },
-        )
-        .await?;
+    // NOTE: The quota check if skipped for fork collection for now, and we rely on the scorecard to limit access to certain tenents
+    // TODO: Unify the quota and scorecard
+    // server
+    //     .authenticate_and_authorize(
+    //         &headers,
+    //         AuthzAction::ForkCollection,
+    //         AuthzResource {
+    //             tenant: Some(tenant.clone()),
+    //             database: Some(database.clone()),
+    //             collection: Some(collection_id.clone()),
+    //         },
+    //     )
+    //     .await?;
     let _guard =
         server.scorecard_request(&["op:fork_collection", format!("tenant:{}", tenant).as_str()]);
     let collection_id =