refactor(engine): replace long argument lists with params structs

Author: farhan-syah

nodedb/src/control/planner/calvin/dependent_recon.rs

@@ -23,12 +23,12 @@ use crate::control::cluster::calvin::executor::ollp::error::OllpError;
 use crate::control::planner::calvin::dispatch::collection_name_from_plan;
 use crate::control::planner::calvin::preexec::{PreexecScan, run_preexec_scan};
 use crate::control::planner::calvin::{
-    build_dependent_tx_class, is_dependent_predicate, predicate_class_for_filters,
-    run_dependent_with_retry, submit_calvin_routed_assign,
+    DependentRetryArgs, build_dependent_tx_class, is_dependent_predicate,
+    predicate_class_for_filters, run_dependent_with_retry, submit_calvin_routed_assign,
 };
 use crate::control::planner::implicit_edges::{
-    EdgeFieldOverrides, append_implicit_edge_delete_tasks, append_implicit_edge_update_tasks,
-    parse_edge_field_overrides,
+    EdgeFieldOverrides, EdgeUpdateCtx, append_implicit_edge_delete_tasks,
+    append_implicit_edge_update_tasks, parse_edge_field_overrides,
 };
 use crate::control::state::SharedState;
 use crate::types::{DatabaseId, TenantId, TraceId};
@@ -371,12 +371,14 @@ pub async fn dispatch_dependent_edge_recon(
                 }
                 EdgeLifecycle::Update(overrides) => {
                     append_implicit_edge_update_tasks(
-                        state,
+                        EdgeUpdateCtx {
+                            state,
+                            tenant_id,
+                            database_id,
+                            trace_id: TraceId::ZERO,
+                            collection: dep_collection,
+                        },
                         &mut edge_tasks,
-                        tenant_id,
-                        database_id,
-                        TraceId::ZERO,
-                        dep_collection,
                         &edges,
                         &surrogates,
                         overrides,
@@ -440,16 +442,16 @@ pub async fn dispatch_dependent_edge_recon(
         )
     };
 
-    run_dependent_with_retry(
+    run_dependent_with_retry(DependentRetryArgs {
         registry,
-        orc,
-        pred_class,
+        orchestrator: orc,
+        predicate_class_hash: pred_class,
         timeout,
         ollp_max_retries,
         initial_predicted,
         submit,
         rescan,
-    )
+    })
     .await?;
 
     Ok(DependentReconOutcome {

nodedb/src/control/planner/calvin/mod.rs

@@ -22,7 +22,7 @@ pub use dispatch::{
 pub use dispatch_multi::dispatch_tasks_to_calvin;
 pub use explain::calvin_explain_preamble;
 pub use predicate::predicate_class_for_filters;
-pub use retry_loop::run_dependent_with_retry;
+pub use retry_loop::{DependentRetryArgs, run_dependent_with_retry};
 pub use submit::{
     RoutedAssignment, submit_and_await_calvin, submit_and_await_calvin_with_timeout,
     submit_calvin_routed, submit_calvin_routed_assign,

nodedb/src/control/planner/calvin/retry_loop.rs

@@ -33,23 +33,36 @@ use crate::control::planner::calvin::submit::RoutedAssignment;
 /// `submit` and `rescan` are injected so this loop is unit-testable WITHOUT a
 /// live server/executor: a fake scheduler driving the real
 /// [`CalvinCompletionRegistry`] suffices.
-#[allow(clippy::too_many_arguments)]
+pub struct DependentRetryArgs<'a, P, SF, RF> {
+    pub registry: &'a CalvinCompletionRegistry,
+    pub orchestrator: &'a OllpOrchestrator,
+    pub predicate_class_hash: u64,
+    pub timeout: std::time::Duration,
+    pub ollp_max_retries: u32,
+    pub initial_predicted: P,
+    pub submit: SF,
+    pub rescan: RF,
+}
+
 pub async fn run_dependent_with_retry<P, SF, SFut, RF, RFut>(
-    registry: &CalvinCompletionRegistry,
-    orchestrator: &OllpOrchestrator,
-    predicate_class_hash: u64,
-    timeout: std::time::Duration,
-    ollp_max_retries: u32,
-    initial_predicted: P,
-    mut submit: SF,
-    mut rescan: RF,
+    args: DependentRetryArgs<'_, P, SF, RF>,
 ) -> crate::Result<()>
 where
     SF: FnMut(&P) -> SFut,
     SFut: std::future::Future<Output = Result<RoutedAssignment, OllpError>>,
     RF: FnMut() -> RFut,
     RFut: std::future::Future<Output = crate::Result<P>>,
 {
+    let DependentRetryArgs {
+        registry,
+        orchestrator,
+        predicate_class_hash,
+        timeout,
+        ollp_max_retries,
+        initial_predicted,
+        mut submit,
+        mut rescan,
+    } = args;
     let mut predicted = initial_predicted;
     let mut retry: u32 = 0;
     loop {

nodedb/src/control/planner/calvin/retry_loop_tests.rs

@@ -100,14 +100,14 @@ fn converges_after_two_mismatches() {
             let submit_calls = Arc::clone(&submit_calls);
             let rescan_calls = Arc::clone(&rescan_calls);
             let tx = tx.clone();
-            run_dependent_with_retry(
-                &registry,
-                &orchestrator,
-                0xABCD,
-                std::time::Duration::from_secs(5),
-                5,
-                vec![1, 2, 3],
-                move |_predicted: &Vec<u32>| {
+            run_dependent_with_retry(DependentRetryArgs {
+                registry: &registry,
+                orchestrator: &orchestrator,
+                predicate_class_hash: 0xABCD,
+                timeout: std::time::Duration::from_secs(5),
+                ollp_max_retries: 5,
+                initial_predicted: vec![1, 2, 3],
+                submit: move |_predicted: &Vec<u32>| {
                     let seq = Arc::clone(&seq);
                     let submit_calls = Arc::clone(&submit_calls);
                     let tx = tx.clone();
@@ -120,14 +120,14 @@ fn converges_after_two_mismatches() {
                         Ok::<RoutedAssignment, OllpError>(assignment)
                     }
                 },
-                move || {
+                rescan: move || {
                     let rescan_calls = Arc::clone(&rescan_calls);
                     async move {
                         let n = rescan_calls.fetch_add(1, Ordering::SeqCst);
                         Ok(vec![100 + n])
                     }
                 },
-            )
+            })
             .await
         };
 
@@ -168,14 +168,14 @@ fn exhausts_on_persistent_mismatch() {
             let seq = Arc::clone(&seq);
             let submit_calls = Arc::clone(&submit_calls);
             let tx = tx.clone();
-            run_dependent_with_retry(
-                &registry,
-                &orchestrator,
-                0xABCD,
-                std::time::Duration::from_secs(5),
-                3,
-                vec![1],
-                move |_predicted: &Vec<u32>| {
+            run_dependent_with_retry(DependentRetryArgs {
+                registry: &registry,
+                orchestrator: &orchestrator,
+                predicate_class_hash: 0xABCD,
+                timeout: std::time::Duration::from_secs(5),
+                ollp_max_retries: 3,
+                initial_predicted: vec![1],
+                submit: move |_predicted: &Vec<u32>| {
                     let seq = Arc::clone(&seq);
                     let submit_calls = Arc::clone(&submit_calls);
                     let tx = tx.clone();
@@ -188,8 +188,8 @@ fn exhausts_on_persistent_mismatch() {
                         Ok::<RoutedAssignment, OllpError>(assignment)
                     }
                 },
-                move || async move { Ok(vec![1]) },
-            )
+                rescan: move || async move { Ok(vec![1]) },
+            })
             .await
         };
 
@@ -231,14 +231,14 @@ fn pre_admission_retry_does_not_rescan() {
             let submit_calls = Arc::clone(&submit_calls);
             let rescan_calls = Arc::clone(&rescan_calls);
             let tx = tx.clone();
-            run_dependent_with_retry(
-                &registry,
-                &orchestrator,
-                0xABCD,
-                std::time::Duration::from_secs(5),
-                5,
-                vec![1],
-                move |_predicted: &Vec<u32>| {
+            run_dependent_with_retry(DependentRetryArgs {
+                registry: &registry,
+                orchestrator: &orchestrator,
+                predicate_class_hash: 0xABCD,
+                timeout: std::time::Duration::from_secs(5),
+                ollp_max_retries: 5,
+                initial_predicted: vec![1],
+                submit: move |_predicted: &Vec<u32>| {
                     let seq = Arc::clone(&seq);
                     let submit_calls = Arc::clone(&submit_calls);
                     let tx = tx.clone();
@@ -255,14 +255,14 @@ fn pre_admission_retry_does_not_rescan() {
                         Ok::<RoutedAssignment, OllpError>(assignment)
                     }
                 },
-                move || {
+                rescan: move || {
                     let rescan_calls = Arc::clone(&rescan_calls);
                     async move {
                         rescan_calls.fetch_add(1, Ordering::SeqCst);
                         Ok(vec![1])
                     }
                 },
-            )
+            })
             .await
         };
 

nodedb/src/control/planner/context/mod.rs

@@ -10,5 +10,5 @@ mod catalog_inputs;
 pub mod query;
 pub mod security;
 
-pub use query::{QueryContext, SYSTEM_FUNCTION_NAMES};
+pub use query::{PlanSqlWithRlsParams, QueryContext, SYSTEM_FUNCTION_NAMES};
 pub use security::PlanSecurityContext;

nodedb/src/control/planner/context/query/mod.rs

@@ -8,6 +8,8 @@ use super::catalog_inputs::CatalogInputs;
 
 mod planning;
 
+pub use planning::PlanSqlWithRlsParams;
+
 /// Query context for the Control Plane.
 ///
 /// SQL queries are parsed and planned via nodedb-sql, then converted

nodedb/src/control/planner/context/query/planning.rs

@@ -13,6 +13,14 @@ use super::QueryContext;
 use crate::control::planner::context::security::PlanSecurityContext;
 use crate::control::server::response_shape::schema::OutputSchema;
 
+/// Bundled arguments for [`QueryContext::plan_sql_with_rls`].
+pub struct PlanSqlWithRlsParams<'a> {
+    pub sql: &'a str,
+    pub tenant_id: crate::types::TenantId,
+    pub database_id: crate::types::DatabaseId,
+    pub sec: &'a PlanSecurityContext<'a>,
+}
+
 impl QueryContext {
     /// Parse SQL and convert to NodeDB physical plan(s).
     ///
@@ -145,17 +153,19 @@ impl QueryContext {
     /// Parse SQL, inject RLS predicates, convert to physical plan.
     ///
     /// This is the primary query entry point.
-    #[allow(clippy::too_many_arguments)]
     pub async fn plan_sql_with_rls(
         &self,
-        sql: &str,
-        tenant_id: crate::types::TenantId,
-        database_id: crate::types::DatabaseId,
-        sec: &PlanSecurityContext<'_>,
+        params: PlanSqlWithRlsParams<'_>,
     ) -> crate::Result<(
         Vec<nodedb_physical::physical_task::PhysicalTask>,
         OutputSchema,
     )> {
+        let PlanSqlWithRlsParams {
+            sql,
+            tenant_id,
+            database_id,
+            sec,
+        } = params;
         self.plan_sql_with_rls_returning(sql, tenant_id, database_id, sec, false)
             .await
     }

nodedb/src/control/planner/implicit_edges/delete.rs

@@ -7,7 +7,7 @@
 use nodedb_physical::physical_task::PhysicalTask;
 
 use super::extract::resolve_edge_label;
-use super::routed::push_edge_delete;
+use super::routed::{EdgeRouteCtx, push_edge_delete};
 use crate::control::state::SharedState;
 use crate::types::{DatabaseId, TenantId, TraceId};
 
@@ -40,14 +40,16 @@ pub async fn append_implicit_edge_delete_tasks(
     for edge in edges {
         let label = resolve_edge_label(edge.label.as_deref());
         push_edge_delete(
-            state,
+            EdgeRouteCtx {
+                state,
+                tenant_id,
+                database_id,
+                trace_id,
+                collection,
+                src: &edge.from,
+                dst: &edge.to,
+            },
             out,
-            tenant_id,
-            database_id,
-            trace_id,
-            collection,
-            &edge.from,
-            &edge.to,
             label,
         )
         .await?;

nodedb/src/control/planner/implicit_edges/mod.rs

@@ -32,6 +32,6 @@ pub use catalog::mark_collection_edge_bearing;
 pub use delete::append_implicit_edge_delete_tasks;
 pub use insert::append_implicit_edge_tasks;
 pub use update::{
-    EdgeFieldOverrides, FieldUpdate, WeightUpdate, append_implicit_edge_update_tasks,
-    parse_edge_field_overrides,
+    EdgeFieldOverrides, EdgeUpdateCtx, FieldUpdate, WeightUpdate,
+    append_implicit_edge_update_tasks, parse_edge_field_overrides,
 };