Merge branch 'main' into copilot/sub-pr-72

Signed-off-by: Adam Poulemanos <89049923+bashandbone@users.noreply.github.com>

Author: bashandbone

.github/workflows/gemini-review.yml

@@ -49,6 +49,7 @@ jobs:
       - name: 'Run Gemini pull request review'
         uses: 'google-github-actions/run-gemini-cli@v0' # ratchet:exclude
         id: 'gemini_pr_review'
+        continue-on-error: true
         env:
           GITHUB_TOKEN: '${{ steps.mint_identity_token.outputs.token || secrets.GITHUB_TOKEN || github.token }}'
           ISSUE_TITLE: '${{ github.event.pull_request.title || github.event.issue.title }}'

crates/recoco-core/src/base/value.rs

@@ -1767,4 +1767,51 @@ mod tests {
         let size = value.estimated_byte_size();
         assert_eq!(size, std::mem::size_of::<Value<ScopeValue>>());
     }
+
+    #[test]
+    fn test_range_value_len() {
+        let range1 = RangeValue { start: 0, end: 5 };
+        assert_eq!(range1.len(), 5);
+
+        let range2 = RangeValue { start: 5, end: 5 };
+        assert_eq!(range2.len(), 0);
+
+        let range3 = RangeValue { start: 10, end: 25 };
+        assert_eq!(range3.len(), 15);
+    
+    #[test]
+    fn test_key_part_to_strs() {
+        let bytes_part = KeyPart::from(vec![1u8, 2, 3]);
+        assert_eq!(bytes_part.to_strs(), vec!["AQID"]);
+
+        let str_part = KeyPart::from(String::from("hello"));
+        assert_eq!(str_part.to_strs(), vec!["hello"]);
+
+        let bool_part = KeyPart::from(true);
+        assert_eq!(bool_part.to_strs(), vec!["true"]);
+
+        let int64_part = KeyPart::from(42i64);
+        assert_eq!(int64_part.to_strs(), vec!["42"]);
+
+        let range_part = KeyPart::from(RangeValue::new(10, 20));
+        assert_eq!(range_part.to_strs(), vec!["10", "20"]);
+
+        let uuid_val = uuid::Uuid::nil();
+        let uuid_part = KeyPart::from(uuid_val);
+        assert_eq!(uuid_part.to_strs(), vec![uuid_val.to_string()]);
+
+        let date_val = chrono::NaiveDate::from_ymd_opt(2023, 10, 15)
+            .expect("test date 2023-10-15 should be a valid NaiveDate");
+        let date_part = KeyPart::from(date_val);
+        assert_eq!(date_part.to_strs(), vec!["2023-10-15"]);
+
+        let struct_part = KeyPart::from(vec![
+            KeyPart::from(String::from("world")),
+            KeyPart::from(100i64),
+            KeyPart::from(vec![
+                KeyPart::from(false),
+            ]),
+        ]);
+        assert_eq!(struct_part.to_strs(), vec!["world", "100", "false"]);
+    }
 }

crates/recoco-core/src/ops/sources/local_file.rs

@@ -32,6 +32,7 @@ pub struct Spec {
 
 struct Executor {
     root_path: PathBuf,
+    canonical_root_path: Option<PathBuf>,
     binary: bool,
     pattern_matcher: PatternMatcher,
     max_file_size: Option<i64>,
@@ -134,14 +135,60 @@ impl SourceExecutor for Executor {
         options: &SourceExecutorReadOptions,
     ) -> Result<PartialSourceRowData> {
         let path = key.single_part()?.str_value()?.as_ref();
-        if !self.pattern_matcher.is_file_included(path) {
+        let path_obj = Path::new(path);
+
+        // Prevent path traversal vulnerabilities by verifying the path
+        // doesn't contain parent directory or absolute components.
+        if path_obj.components().any(|c| {
+            matches!(
+                c,
+                std::path::Component::ParentDir
+                    | std::path::Component::RootDir
+                    | std::path::Component::Prefix(_)
+            )
+        }) || !self.pattern_matcher.is_file_included(path)
+        {
             return Ok(PartialSourceRowData {
                 value: Some(SourceValue::NonExistence),
                 ordinal: Some(Ordinal::unavailable()),
                 content_version_fp: None,
             });
         }
+
         let path = self.root_path.join(path);
+
+        // Mitigate symlink-based path traversal by canonicalizing and checking boundaries
+        if let Some(root_canon) = &self.canonical_root_path {
+            let path_canon = match tokio::fs::canonicalize(&path).await {
+                Ok(c) => c,
+                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
+                    // Target file doesn't exist.
+                    return Ok(PartialSourceRowData {
+                        value: Some(SourceValue::NonExistence),
+                        ordinal: Some(Ordinal::unavailable()),
+                        content_version_fp: None,
+                    });
+                }
+                Err(e) => Err(e)?,
+            };
+
+            if !path_canon.starts_with(root_canon) {
+                // Symlink points outside the allowed root directory.
+                return Ok(PartialSourceRowData {
+                    value: Some(SourceValue::NonExistence),
+                    ordinal: Some(Ordinal::unavailable()),
+                    content_version_fp: None,
+                });
+            }
+        } else {
+            // Root doesn't exist (failed to canonicalize during setup), so the file cannot exist.
+            return Ok(PartialSourceRowData {
+                value: Some(SourceValue::NonExistence),
+                ordinal: Some(Ordinal::unavailable()),
+                content_version_fp: None,
+            });
+        }
+
         let mut metadata: Option<Metadata> = None;
         // Check file size limit
         if let Some(max_size) = self.max_file_size
@@ -237,8 +284,12 @@ impl SourceFactoryBase for Factory {
         spec: Spec,
         _context: Arc<FlowInstanceContext>,
     ) -> Result<Box<dyn SourceExecutor>> {
+        let root_path = PathBuf::from(spec.path);
+        let canonical_root_path = tokio::fs::canonicalize(&root_path).await.ok();
+
         Ok(Box::new(Executor {
-            root_path: PathBuf::from(spec.path),
+            root_path,
+            canonical_root_path,
             binary: spec.binary,
             pattern_matcher: PatternMatcher::new(spec.included_patterns, spec.excluded_patterns)?,
             max_file_size: spec.max_file_size,

crates/recoco-core/src/ops/targets/postgres.rs

@@ -287,22 +287,59 @@ impl ExportContext {
         deletions: &[interface::ExportTargetDeleteEntry],
         txn: &mut sqlx::PgTransaction<'_>,
     ) -> Result<()> {
-        // TODO: Find a way to batch delete.
-        for deletion in deletions.iter() {
+        if deletions.is_empty() {
+            return Ok(());
+        }
+
+        let num_parameters = self.key_fields_schema.len();
+        if num_parameters == 0 {
+            return Ok(());
+        }
+
+        for deletion_chunk in deletions.chunks(BIND_LIMIT / num_parameters) {
             let mut query_builder = sqlx::QueryBuilder::new("");
             query_builder.push(&self.delete_sql_prefix);
-            for (i, ((schema, _), value)) in
-                std::iter::zip(&self.key_fields_schema, &deletion.key).enumerate()
-            {
+
+            if num_parameters > 1 {
+                query_builder.push("(");
+            }
+            for (i, (schema, _)) in self.key_fields_schema.iter().enumerate() {
                 if i > 0 {
-                    query_builder.push(" AND ");
+                    query_builder.push(", ");
                 }
                 query_builder.push("\"");
                 query_builder.push(schema.name.as_str());
                 query_builder.push("\"");
-                query_builder.push("=");
-                bind_key_field(&mut query_builder, value)?;
             }
+            if num_parameters > 1 {
+                query_builder.push(")");
+            }
+
+            query_builder.push(" IN (");
+
+            for (i, deletion) in deletion_chunk.iter().enumerate() {
+                if i > 0 {
+                    query_builder.push(", ");
+                }
+                if num_parameters > 1 {
+                    query_builder.push("(");
+                }
+                for (j, (_schema, _)) in self.key_fields_schema.iter().enumerate() {
+                    if j > 0 {
+                        query_builder.push(", ");
+                    }
+                    if let Some(value) = deletion.key.get(j) {
+                        bind_key_field(&mut query_builder, value)?;
+                    } else {
+                        query_builder.push("NULL");
+                    }
+                }
+                if num_parameters > 1 {
+                    query_builder.push(")");
+                }
+            }
+
+            query_builder.push(")");
             query_builder.build().execute(&mut **txn).await?;
         }
         Ok(())
@@ -993,9 +1030,11 @@ impl AttachmentSetupChange for SqlCommandSetupChange {
     }
 
     async fn apply_change(&self) -> Result<()> {
-        for teardown_sql in self.teardown_sql_to_run.iter() {
-            sqlx::raw_sql(teardown_sql).execute(&self.db_pool).await?;
-        }
+        let teardown_futs = self
+            .teardown_sql_to_run
+            .iter()
+            .map(|teardown_sql| sqlx::raw_sql(teardown_sql).execute(&self.db_pool));
+        futures::future::try_join_all(teardown_futs).await?;
         if let Some(setup_sql) = &self.setup_sql_to_run {
             sqlx::raw_sql(setup_sql).execute(&self.db_pool).await?;
         }

crates/recoco-core/src/ops/targets/qdrant.rs

@@ -26,6 +26,9 @@ use qdrant_client::qdrant::{
 const DEFAULT_VECTOR_SIMILARITY_METRIC: spec::VectorSimilarityMetric =
     spec::VectorSimilarityMetric::CosineSimilarity;
 const DEFAULT_URL: &str = "http://localhost:6334/";
+/// Maximum number of setup operations (deletes or creates) to run concurrently.
+/// Bounds the number of simultaneous Qdrant client requests during collection setup.
+const MAX_CONCURRENT_SETUP_OPS: usize = 16;
 
 ////////////////////////////////////////////////////////////
 // Public Types
@@ -577,23 +580,45 @@ impl TargetFactoryBase for Factory {
         setup_change: Vec<TypedResourceSetupChangeItem<'async_trait, Self>>,
         context: Arc<FlowInstanceContext>,
     ) -> Result<()> {
-        for setup_change in setup_change.iter() {
-            let qdrant_client =
-                self.get_qdrant_client(&setup_change.key.connection, &context.auth_registry)?;
-            setup_change
-                .setup_change
-                .apply_delete(&setup_change.key.collection_name, &qdrant_client)
-                .await?;
+        let auth_registry = &context.auth_registry;
+        // Each item in setup_change has a unique CollectionKey (enforced by the setup
+        // orchestration layer), so concurrent operations across items are race-free.
+
+        // Delete phase: collect futures first, then run with bounded concurrency.
+        let mut delete_futures = Vec::with_capacity(setup_change.len());
+        for change in setup_change.iter() {
+            let client_result = self.get_qdrant_client(&change.key.connection, auth_registry);
+            delete_futures.push(async move {
+                let client = client_result?;
+                change
+                    .setup_change
+                    .apply_delete(&change.key.collection_name, &client)
+                    .await
+            });
         }
-        for setup_change in setup_change.iter() {
-            let qdrant_client =
-                self.get_qdrant_client(&setup_change.key.connection, &context.auth_registry)?;
-            setup_change
-                .setup_change
-                .apply_create(&setup_change.key.collection_name, &qdrant_client)
-                .await?;
+        futures::stream::iter(delete_futures)
+            .buffer_unordered(MAX_CONCURRENT_SETUP_OPS)
+            .try_collect::<Vec<_>>()
+            .await
+            .map(|_| ())?;
+
+        // Create phase: collect futures first, then run with bounded concurrency.
+        let mut create_futures = Vec::with_capacity(setup_change.len());
+        for change in setup_change.iter() {
+            let client_result = self.get_qdrant_client(&change.key.connection, auth_registry);
+            create_futures.push(async move {
+                let client = client_result?;
+                change
+                    .setup_change
+                    .apply_create(&change.key.collection_name, &client)
+                    .await
+            });
         }
-        Ok(())
+        futures::stream::iter(create_futures)
+            .buffer_unordered(MAX_CONCURRENT_SETUP_OPS)
+            .try_collect::<Vec<_>>()
+            .await
+            .map(|_| ())
     }
 }
 

crates/recoco-core/src/setup/components.rs

@@ -156,27 +156,35 @@ impl<D: SetupOperator + Send + Sync> ResourceSetupChange for SetupChange<D> {
     }
 }
 
+/// Maximum number of component operations (deletes or upserts) that may run concurrently.
+/// Keeping this bounded prevents overwhelming a database connection pool or
+/// network layer when a large number of components change at once.
+const COMPONENT_CONCURRENCY_LIMIT: usize = 16;
+
 pub async fn apply_component_changes<D: SetupOperator>(
     changes: Vec<&SetupChange<D>>,
     context: &D::Context,
 ) -> Result<()> {
     // First delete components that need to be removed
-    for change in changes.iter() {
-        for key in &change.keys_to_delete {
-            change.desc.delete(key, context).await?;
-        }
-    }
+    let delete_futures = changes.iter().flat_map(|change| {
+        change
+            .keys_to_delete
+            .iter()
+            .map(move |key| change.desc.delete(key, context))
+    });
+    futures::future::try_join_all(delete_futures).await?;
 
     // Then upsert components that need to be updated
-    for change in changes.iter() {
-        for state in &change.states_to_upsert {
+    let upsert_futures = changes.iter().flat_map(|change| {
+        change.states_to_upsert.iter().map(move |state| async move {
             if state.already_exists {
-                change.desc.update(&state.state, context).await?;
+                change.desc.update(&state.state, context).await
             } else {
-                change.desc.create(&state.state, context).await?;
+                change.desc.create(&state.state, context).await
             }
-        }
-    }
+        })
+    });
+    futures::future::try_join_all(upsert_futures).await?;
 
     Ok(())
 }