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

KV, sparse B-tree, graph edge/pattern, and timeseries grouped-scan
entry points took eight-plus positional arguments and were annotated
with #[allow(clippy::too_many_arguments)]. Bundle each call's
arguments into a dedicated params struct instead, removing the lint
allowance and updating every executor handler and test call site.

Author: farhan-syah

nodedb/src/data/executor/core_loop/maintenance.rs

@@ -200,12 +200,14 @@ impl CoreLoop {
             for v in new_vals.difference(&old_vals) {
                 if let Err(e) = self.sparse.index_put_in_txn(
                     txn,
-                    database_id,
-                    tid,
-                    collection,
-                    &index_path.path,
-                    v,
-                    doc_id,
+                    crate::engine::sparse::btree_index::IndexEntryTxn {
+                        database_id,
+                        tenant_id: tid,
+                        collection,
+                        field: &index_path.path,
+                        value: v,
+                        document_id: doc_id,
+                    },
                 ) {
                     tracing::warn!(
                         core = self.core_id,

nodedb/src/data/executor/handlers/control/move_tenant.rs

@@ -44,13 +44,14 @@ impl CoreLoop {
         }
 
         // KV engine.
-        self.kv_engine.rename_collection(
-            old_database_id,
-            new_database_id,
-            tenant_id,
-            old_collection,
-            new_collection,
-        );
+        self.kv_engine
+            .rename_collection(crate::engine::kv::RenameCollectionParams {
+                old_database_id,
+                new_database_id,
+                tenant_id,
+                old_collection,
+                new_collection,
+            });
 
         self.response_ok(task)
     }

nodedb/src/data/executor/handlers/control/snapshot.rs

@@ -139,26 +139,29 @@ impl CoreLoop {
         debug!(core = self.core_id, %collection, %field, limit, "range scan");
 
         // Try index-backed range scan first.
-        let results = match self.sparse.range_scan(
-            task.request.database_id.as_u64(),
-            tid,
-            collection,
-            field,
-            lower,
-            upper,
-            limit,
-        ) {
-            Ok(r) => r,
-            Err(e) => {
-                warn!(core = self.core_id, error = %e, "sparse range scan failed");
-                return self.response_error(
-                    task,
-                    ErrorCode::Internal {
-                        detail: e.to_string(),
-                    },
-                );
-            }
-        };
+        let results =
+            match self
+                .sparse
+                .range_scan(crate::engine::sparse::btree_index::RangeScanParams {
+                    database_id: task.request.database_id.as_u64(),
+                    tenant_id: tid,
+                    collection,
+                    field,
+                    lower,
+                    upper,
+                    limit,
+                }) {
+                Ok(r) => r,
+                Err(e) => {
+                    warn!(core = self.core_id, error = %e, "sparse range scan failed");
+                    return self.response_error(
+                        task,
+                        ErrorCode::Internal {
+                            detail: e.to_string(),
+                        },
+                    );
+                }
+            };
 
         // If the index returned nothing, fall back to full scan + sort.
         // This handles collections without a secondary index on `field`.

nodedb/src/data/executor/handlers/document/write.rs

@@ -302,12 +302,14 @@ impl CoreLoop {
                 }
                 if let Err(e) = self.sparse.index_put_in_txn(
                     &txn,
-                    task.request.database_id.as_u64(),
-                    tid,
-                    collection,
-                    path,
-                    &stored,
-                    doc_id,
+                    crate::engine::sparse::btree_index::IndexEntryTxn {
+                        database_id: task.request.database_id.as_u64(),
+                        tenant_id: tid,
+                        collection,
+                        field: path,
+                        value: &stored,
+                        document_id: doc_id,
+                    },
                 ) {
                     return self.response_error(
                         task,

nodedb/src/data/executor/handlers/graph_match.rs

@@ -253,13 +253,15 @@ impl CoreLoop {
         let overlay = self.match_graph_overlay(task, tid);
         match crate::engine::graph::pattern::executor::execute(
             &query,
-            partition,
-            &self.edge_store,
-            frontier_bitmap,
-            is_remote_node,
-            varlen_caps,
-            &props,
-            overlay.as_ref(),
+            crate::engine::graph::pattern::executor::MatchExecCtx {
+                csr: partition,
+                edge_store: &self.edge_store,
+                frontier_bitmap,
+                is_remote_node,
+                varlen_caps,
+                props: &props,
+                overlay: overlay.as_ref(),
+            },
         ) {
             Ok(outcome) => self.match_outcome_response(task, outcome),
             Err(e) => self.response_error(task, ErrorCode::from(e)),
@@ -363,17 +365,19 @@ impl CoreLoop {
         // unit.
         match crate::engine::graph::pattern::executor::execute_continuation(
             &query,
-            partition,
-            &self.edge_store,
-            None, // no anchor prefilter on the resume path
-            is_remote_node,
+            crate::engine::graph::pattern::executor::MatchExecCtx {
+                csr: partition,
+                edge_store: &self.edge_store,
+                frontier_bitmap: None, // no anchor prefilter on the resume path
+                is_remote_node,
+                varlen_caps,
+                props: &props,
+                overlay: None,
+            },
             ContinuationSeed {
                 triple_idx: resume_triple_idx,
                 seed_row,
             },
-            varlen_caps,
-            &props,
-            None,
         ) {
             Ok(outcome) => self.match_outcome_response(task, outcome),
             Err(e) => self.response_error(task, ErrorCode::from(e)),
@@ -459,14 +463,16 @@ impl CoreLoop {
         // built. Cross-shard MATCH read-your-own-writes is a separate unit.
         match crate::engine::graph::pattern::executor::execute_varlen_resume(
             &query,
-            partition,
-            &self.edge_store,
-            None, // no anchor prefilter on the resume path
-            is_remote_node,
+            crate::engine::graph::pattern::executor::MatchExecCtx {
+                csr: partition,
+                edge_store: &self.edge_store,
+                frontier_bitmap: None, // no anchor prefilter on the resume path
+                is_remote_node,
+                varlen_caps,
+                props: &props,
+                overlay: None,
+            },
             resume,
-            varlen_caps,
-            &props,
-            None,
         ) {
             Ok(outcome) => self.match_outcome_response(task, outcome),
             Err(e) => self.response_error(task, ErrorCode::from(e)),

nodedb/src/data/executor/handlers/graph_temporal.rs

@@ -57,45 +57,22 @@ impl CoreLoop {
         );
         let database_id = task.request.database_id.as_u64();
         let tenant = TenantId::new(tid);
+        let as_of_params = crate::engine::graph::edge_store::NeighborsAsOfParams {
+            db: database_id,
+            tid: tenant,
+            collection,
+            node: node_id,
+            label_filter: edge_label.as_deref(),
+            system_as_of_ms,
+            valid_at_ms,
+        };
         let edges_result = match direction {
-            Direction::Out => self.edge_store.neighbors_out_as_of(
-                database_id,
-                tenant,
-                collection,
-                node_id,
-                edge_label.as_deref(),
-                system_as_of_ms,
-                valid_at_ms,
-            ),
-            Direction::In => self.edge_store.neighbors_in_as_of(
-                database_id,
-                tenant,
-                collection,
-                node_id,
-                edge_label.as_deref(),
-                system_as_of_ms,
-                valid_at_ms,
-            ),
+            Direction::Out => self.edge_store.neighbors_out_as_of(as_of_params),
+            Direction::In => self.edge_store.neighbors_in_as_of(as_of_params),
             Direction::Both => {
-                let out = self.edge_store.neighbors_out_as_of(
-                    database_id,
-                    tenant,
-                    collection,
-                    node_id,
-                    edge_label.as_deref(),
-                    system_as_of_ms,
-                    valid_at_ms,
-                );
+                let out = self.edge_store.neighbors_out_as_of(as_of_params);
                 match out {
-                    Ok(mut out) => match self.edge_store.neighbors_in_as_of(
-                        database_id,
-                        tenant,
-                        collection,
-                        node_id,
-                        edge_label.as_deref(),
-                        system_as_of_ms,
-                        valid_at_ms,
-                    ) {
+                    Ok(mut out) => match self.edge_store.neighbors_in_as_of(as_of_params) {
                         Ok(inbound) => {
                             out.extend(inbound);
                             Ok(out)

nodedb/src/data/executor/handlers/kv/crud.rs

@@ -150,9 +150,16 @@ impl CoreLoop {
             .epoch_system_ms
             .map(|ms| ms as u64)
             .unwrap_or_else(current_ms);
-        let old = self
-            .kv_engine
-            .put(did, tid, collection, key, value, ttl_ms, now_ms, surrogate);
+        let old = self.kv_engine.put(crate::engine::kv::KvPutParams {
+            database_id: did,
+            tenant_id: tid,
+            collection,
+            key,
+            value,
+            ttl_ms,
+            now_ms,
+            surrogate,
+        });
         if let Some(ref m) = self.metrics {
             m.record_kv_put();
         }
@@ -222,8 +229,16 @@ impl CoreLoop {
             );
         }
 
-        self.kv_engine
-            .put(did, tid, collection, key, value, ttl_ms, now_ms, surrogate);
+        self.kv_engine.put(crate::engine::kv::KvPutParams {
+            database_id: did,
+            tenant_id: tid,
+            collection,
+            key,
+            value,
+            ttl_ms,
+            now_ms,
+            surrogate,
+        });
         if let Some(ref m) = self.metrics {
             m.record_kv_put();
         }
@@ -276,8 +291,16 @@ impl CoreLoop {
             return self.response_ok(task);
         }
 
-        self.kv_engine
-            .put(did, tid, collection, key, value, ttl_ms, now_ms, surrogate);
+        self.kv_engine.put(crate::engine::kv::KvPutParams {
+            database_id: did,
+            tenant_id: tid,
+            collection,
+            key,
+            value,
+            ttl_ms,
+            now_ms,
+            surrogate,
+        });
         if let Some(ref m) = self.metrics {
             m.record_kv_put();
         }
@@ -378,16 +401,16 @@ impl CoreLoop {
             }
         };
 
-        self.kv_engine.put(
-            did,
-            tid,
+        self.kv_engine.put(crate::engine::kv::KvPutParams {
+            database_id: did,
+            tenant_id: tid,
             collection,
             key,
-            &stored_bytes,
+            value: &stored_bytes,
             ttl_ms,
             now_ms,
             surrogate,
-        );
+        });
         if let Some(ref m) = self.metrics {
             m.record_kv_put();
         }

nodedb/src/data/executor/handlers/kv/field.rs

@@ -92,16 +92,16 @@ impl CoreLoop {
             Err(e) => return self.response_error(task, e),
         };
 
-        self.kv_engine.put(
-            did,
-            tid,
+        self.kv_engine.put(crate::engine::kv::KvPutParams {
+            database_id: did,
+            tenant_id: tid,
             collection,
             key,
-            &computed.new_value,
-            0,
+            value: &computed.new_value,
+            ttl_ms: 0,
             now_ms,
             surrogate,
-        );
+        });
         match response_codec::encode_json(
             &serde_json::json!({ "fields_added": computed.fields_added }),
         ) {

nodedb/src/data/executor/handlers/kv/index.rs

@@ -24,15 +24,17 @@ impl CoreLoop {
     ) -> Response {
         debug!(core = self.core_id, %collection, %field, "kv register index");
         let now_ms = current_ms();
-        let backfilled = self.kv_engine.register_index(
-            did,
-            tid,
-            collection,
-            field,
-            field_position,
-            backfill,
-            now_ms,
-        );
+        let backfilled = self
+            .kv_engine
+            .register_index(crate::engine::kv::RegisterIndexParams {
+                database_id: did,
+                tenant_id: tid,
+                collection,
+                field,
+                field_position,
+                backfill,
+                now_ms,
+            });
         match response_codec::encode_json(&serde_json::json!({
             "index": field,
             "backfilled": backfilled,

nodedb/src/data/executor/handlers/kv/sorted.rs

@@ -205,8 +205,14 @@ impl CoreLoop {
 
         match self
             .kv_engine
-            .sorted_index_range(did, tid, index_name, score_min, score_max, now_ms)
-        {
+            .sorted_index_range(crate::engine::kv::SortedIndexRangeParams {
+                database_id: did,
+                tenant_id: tid,
+                index_name,
+                score_min,
+                score_max,
+                now_ms,
+            }) {
             Some(entries) => {
                 let rows: Vec<serde_json::Value> = entries
                     .into_iter()