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

Control-Plane gateway dispatch, graph traversal (BFS, BSP pagerank/WCC,
cluster resolve, subgraph traverse), DDL neutral graph/tree ops, typed
automation routing, and shuffle fan-out handlers took long positional
argument lists. Bundle them into per-handler params structs and update
call sites, matching the pattern already applied elsewhere in the engine.

Author: farhan-syah

nodedb/src/control/gateway/dispatcher.rs

@@ -87,6 +87,17 @@ pub async fn dispatch_route(
     }
 }
 
+/// Parameters for [`dispatch_route_stream`].
+pub struct DispatchRouteStreamParams<'a> {
+    pub route: TaskRoute,
+    pub shared: &'a Arc<SharedState>,
+    pub tenant_id: TenantId,
+    pub database_id: DatabaseId,
+    pub trace_id: TraceId,
+    pub deadline_ms: u64,
+    pub version_set: &'a GatewayVersionSet,
+}
+
 /// Streaming sibling of [`dispatch_route`]: dispatch a single route and return
 /// a [`ResultStream`] of row batches.
 ///
@@ -96,16 +107,18 @@ pub async fn dispatch_route(
 /// - `Remote` → [`dispatch_remote_stream`] (eager first frame + retry split).
 /// - `Broadcast` → unreachable (router splits broadcasts before dispatch).
 /// - `LeaderUnknown` → `NotLeader` so the gateway retry loop re-resolves.
-#[allow(clippy::too_many_arguments)]
 pub async fn dispatch_route_stream(
-    route: TaskRoute,
-    shared: &Arc<SharedState>,
-    tenant_id: TenantId,
-    database_id: DatabaseId,
-    trace_id: TraceId,
-    deadline_ms: u64,
-    version_set: &GatewayVersionSet,
+    args: DispatchRouteStreamParams<'_>,
 ) -> Result<ResultStream, Error> {
+    let DispatchRouteStreamParams {
+        route,
+        shared,
+        tenant_id,
+        database_id,
+        trace_id,
+        deadline_ms,
+        version_set,
+    } = args;
     match route.decision {
         // Cluster gateway route dispatch: no session-transaction context
         // crosses this boundary yet, so `None`. TRACKED: cross-node

nodedb/src/control/gateway/stream.rs

@@ -14,7 +14,7 @@ use crate::control::server::result_stream::ResultStream;
 use nodedb_physical::physical_plan::PhysicalPlan;
 
 use super::core::{Gateway, QueryContext};
-use super::dispatcher::{default_deadline_ms, dispatch_route_stream};
+use super::dispatcher::{DispatchRouteStreamParams, default_deadline_ms, dispatch_route_stream};
 use super::retry::retry_not_leader;
 use super::route::TaskRoute;
 use super::router::resolve_decision;
@@ -104,15 +104,15 @@ impl Gateway {
                         decision,
                         vshard_id: vshard_id_u32,
                     };
-                    dispatch_route_stream(
+                    dispatch_route_stream(DispatchRouteStreamParams {
                         route,
-                        &shared,
+                        shared: &shared,
                         tenant_id,
                         database_id,
                         trace_id,
                         deadline_ms,
-                        &version_set,
-                    )
+                        version_set: &version_set,
+                    })
                     .await
                 }
             })

nodedb/src/control/server/graph_dispatch/bfs.rs

@@ -21,21 +21,34 @@ use crate::types::{DatabaseId, TenantId};
 use super::helpers::{encode_path, ok_response};
 use super::hop::{NeighborHopParams, execute_neighbor_hop};
 
+/// Parameters for [`cross_core_bfs_with_options`].
+pub struct CrossCoreBfsParams<'a> {
+    pub tenant_id: TenantId,
+    pub database_id: DatabaseId,
+    pub start_nodes: Vec<String>,
+    pub edge_label: Option<String>,
+    pub direction: crate::engine::graph::edge_store::Direction,
+    pub max_depth: usize,
+    pub options: &'a GraphTraversalOptions,
+}
+
 /// Cross-core BFS with explicit traversal options (fan-out limits, partial mode).
 ///
 /// This is the cluster-aware entry point. Callers pass
 /// `&GraphTraversalOptions::default()` for standard traversal.
-#[allow(clippy::too_many_arguments)]
 pub async fn cross_core_bfs_with_options(
     shared: &SharedState,
-    tenant_id: TenantId,
-    database_id: DatabaseId,
-    start_nodes: Vec<String>,
-    edge_label: Option<String>,
-    direction: crate::engine::graph::edge_store::Direction,
-    max_depth: usize,
-    options: &GraphTraversalOptions,
+    params: CrossCoreBfsParams<'_>,
 ) -> crate::Result<Response> {
+    let CrossCoreBfsParams {
+        tenant_id,
+        database_id,
+        start_nodes,
+        edge_label,
+        direction,
+        max_depth,
+        options,
+    } = params;
     let mut visited: HashSet<String> = HashSet::new();
     let mut all_discovered: Vec<String> = Vec::new();
     let mut frontier: Vec<String> = start_nodes;

nodedb/src/control/server/graph_dispatch/bsp_pagerank/coord.rs

@@ -49,7 +49,7 @@ use crate::engine::graph::algo::result::AlgoResultBatch;
 use crate::types::{DatabaseId, TenantId};
 
 use super::enumerate::enumerate_shards;
-use super::scatter::{ShardDispatch, scatter_superstep};
+use super::scatter::{ScatterSuperstepParams, ShardDispatch, scatter_superstep};
 
 /// Default max supersteps when the query carries no explicit `ITERATIONS`.
 /// Mirrors the single-node PageRank default iteration budget.
@@ -135,14 +135,16 @@ pub async fn run_bsp_pagerank(
         .collect();
     let counts = scatter_superstep(
         state,
-        tenant_id,
-        database_id,
-        algorithm,
-        &params,
-        0,
-        0, // count-only sentinel
-        count_dispatches,
-        deadline_ms,
+        ScatterSuperstepParams {
+            tenant_id,
+            database_id,
+            algorithm,
+            params: &params,
+            superstep: 0,
+            global_n: 0, // count-only sentinel
+            dispatches: count_dispatches,
+            deadline_ms,
+        },
     )
     .await?;
 
@@ -233,14 +235,16 @@ pub async fn run_bsp_pagerank(
 
         let results = scatter_superstep(
             state,
-            tenant_id,
-            database_id,
-            algorithm,
-            &params,
-            superstep,
-            global_n,
-            dispatches,
-            deadline_ms,
+            ScatterSuperstepParams {
+                tenant_id,
+                database_id,
+                algorithm,
+                params: &params,
+                superstep,
+                global_n,
+                dispatches,
+                deadline_ms,
+            },
         )
         .await?;
 

nodedb/src/control/server/graph_dispatch/bsp_pagerank/scatter.rs

@@ -17,7 +17,7 @@ use futures::future::join_all;
 use crate::bridge::envelope::{Payload, PhysicalPlan};
 use crate::control::gateway::version_set::GatewayVersionSet;
 use crate::control::server::graph_dispatch::cluster_resolve::{
-    dispatch_superstep_to_node, gateway_shared,
+    DispatchSuperstepParams, dispatch_superstep_to_node, gateway_shared,
 };
 use crate::types::{DatabaseId, TenantId};
 use nodedb_graph::{AlgoParams, GraphAlgorithm};
@@ -56,21 +56,35 @@ pub(super) struct ShardResult {
     pub(super) result: BspSuperstepResult,
 }
 
+/// Parameters for [`scatter_superstep`].
+pub(super) struct ScatterSuperstepParams<'a> {
+    pub(super) tenant_id: TenantId,
+    pub(super) database_id: DatabaseId,
+    pub(super) algorithm: GraphAlgorithm,
+    pub(super) params: &'a AlgoParams,
+    pub(super) superstep: u32,
+    pub(super) global_n: usize,
+    pub(super) dispatches: Vec<ShardDispatch>,
+    pub(super) deadline_ms: u64,
+}
+
 /// Dispatch one `BspSuperstep` to every owner node concurrently and decode each
 /// node's [`BspSuperstepResult`]. `global_n == 0` is the count-only phase
 /// (handler short-circuits after counting owned nodes).
-#[allow(clippy::too_many_arguments)]
 pub(super) async fn scatter_superstep(
     state: &crate::control::state::SharedState,
-    tenant_id: TenantId,
-    database_id: DatabaseId,
-    algorithm: GraphAlgorithm,
-    params: &AlgoParams,
-    superstep: u32,
-    global_n: usize,
-    dispatches: Vec<ShardDispatch>,
-    deadline_ms: u64,
+    args: ScatterSuperstepParams<'_>,
 ) -> crate::Result<Vec<ShardResult>> {
+    let ScatterSuperstepParams {
+        tenant_id,
+        database_id,
+        algorithm,
+        params,
+        superstep,
+        global_n,
+        dispatches,
+        deadline_ms,
+    } = args;
     let shared_arc = gateway_shared(state)?;
     let version_set = GatewayVersionSet::from_pairs(Vec::new());
 
@@ -96,14 +110,16 @@ pub(super) async fn scatter_superstep(
         Box::pin(async move {
             let payload = dispatch_superstep_to_node(
                 shared_arc,
-                tenant_id,
-                database_id,
-                deadline_ms,
-                node_id,
-                is_local,
-                route_vshard,
-                plan,
-                &version_set,
+                DispatchSuperstepParams {
+                    tenant_id,
+                    database_id,
+                    deadline_ms,
+                    node_id,
+                    is_local,
+                    route_vshard,
+                    plan,
+                    version_set: &version_set,
+                },
             )
             .await?;
             let result = decode_single_result_from_payload(node_id, payload)?;

nodedb/src/control/server/graph_dispatch/bsp_wcc/scatter.rs

@@ -18,7 +18,7 @@ use crate::bridge::envelope::{Payload, PhysicalPlan};
 use crate::control::gateway::version_set::GatewayVersionSet;
 use crate::control::server::graph_dispatch::bsp_pagerank::enumerate::ShardTarget;
 use crate::control::server::graph_dispatch::cluster_resolve::{
-    dispatch_superstep_to_node, gateway_shared,
+    DispatchSuperstepParams, dispatch_superstep_to_node, gateway_shared,
 };
 use crate::types::{DatabaseId, TenantId};
 use nodedb_graph::AlgoParams;
@@ -60,14 +60,16 @@ pub(super) async fn scatter_wcc_round(
         Box::pin(async move {
             let payload = dispatch_superstep_to_node(
                 &shared_arc,
-                tenant_id,
-                database_id,
-                deadline_ms,
-                node_id,
-                is_local,
-                route_vshard,
-                plan,
-                &version_set,
+                DispatchSuperstepParams {
+                    tenant_id,
+                    database_id,
+                    deadline_ms,
+                    node_id,
+                    is_local,
+                    route_vshard,
+                    plan,
+                    version_set: &version_set,
+                },
             )
             .await?;
             let result = decode_wcc_from_payload(node_id, payload)?;

nodedb/src/control/server/graph_dispatch/cluster_resolve.rs

@@ -49,25 +49,39 @@ pub(super) fn resolve_for_vshard(state: &SharedState, vshard_id: u32) -> RouteDe
     )
 }
 
+/// Parameters for [`dispatch_superstep_to_node`].
+pub(in crate::control::server::graph_dispatch) struct DispatchSuperstepParams<'a> {
+    pub(in crate::control::server::graph_dispatch) tenant_id: TenantId,
+    pub(in crate::control::server::graph_dispatch) database_id: DatabaseId,
+    pub(in crate::control::server::graph_dispatch) deadline_ms: u64,
+    pub(in crate::control::server::graph_dispatch) node_id: u64,
+    pub(in crate::control::server::graph_dispatch) is_local: bool,
+    pub(in crate::control::server::graph_dispatch) route_vshard: u32,
+    pub(in crate::control::server::graph_dispatch) plan: PhysicalPlan,
+    pub(in crate::control::server::graph_dispatch) version_set: &'a GatewayVersionSet,
+}
+
 /// Dispatch a single already-built graph-superstep `plan` to one owner node and
 /// return its node-level payload. The LOCAL node fans the plan across all its
 /// Data-Plane cores via `execute_plan_all_local_cores` (per-core results merged
 /// into one payload); a REMOTE node gets one `RouteDecision::Remote` dispatch via
 /// `dispatch_route`. An empty payload denotes a zero-vertex shard — the caller's
 /// decoder maps it to its result type's `::default()`. Shared by the PageRank and
 /// WCC per-node scatter paths.
-#[allow(clippy::too_many_arguments)]
 pub(in crate::control::server::graph_dispatch) async fn dispatch_superstep_to_node(
     shared_arc: &Arc<SharedState>,
-    tenant_id: TenantId,
-    database_id: DatabaseId,
-    deadline_ms: u64,
-    node_id: u64,
-    is_local: bool,
-    route_vshard: u32,
-    plan: PhysicalPlan,
-    version_set: &GatewayVersionSet,
+    args: DispatchSuperstepParams<'_>,
 ) -> crate::Result<Payload> {
+    let DispatchSuperstepParams {
+        tenant_id,
+        database_id,
+        deadline_ms,
+        node_id,
+        is_local,
+        route_vshard,
+        plan,
+        version_set,
+    } = args;
     if is_local {
         // Local node: fan across ALL local cores and merge. The per-core
         // owned-node sets are disjoint, so the merged result is correct without

nodedb/src/control/server/graph_dispatch/mod.rs

@@ -29,12 +29,12 @@ pub mod match_scatter;
 pub mod shortest_path;
 pub mod traverse_subgraph;
 
-pub use bfs::cross_core_bfs_with_options;
+pub use bfs::{CrossCoreBfsParams, cross_core_bfs_with_options};
 pub use bsp_pagerank::run_bsp_pagerank;
 pub use bsp_wcc::run_bsp_wcc;
 pub use match_broadcast::{
     MatchBroadcastOutcome, broadcast_match_to_all_cores, unwrap_match_envelope,
 };
 pub use match_scatter::{MatchScatterOutcome, scatter_match};
 pub use shortest_path::cross_core_shortest_path;
-pub use traverse_subgraph::cross_core_traverse_subgraph;
+pub use traverse_subgraph::{CrossCoreTraverseSubgraphParams, cross_core_traverse_subgraph};

nodedb/src/control/server/graph_dispatch/traverse_subgraph.rs

@@ -57,6 +57,17 @@ struct WireSubGraph<'a> {
     edges: Vec<WireEdge<'a>>,
 }
 
+/// Parameters for [`cross_core_traverse_subgraph`].
+pub struct CrossCoreTraverseSubgraphParams<'a> {
+    pub tenant_id: TenantId,
+    pub database_id: DatabaseId,
+    pub start: String,
+    pub edge_label: Option<String>,
+    pub direction: Direction,
+    pub max_depth: usize,
+    pub options: &'a GraphTraversalOptions,
+}
+
 /// BFS that returns a `{nodes,edges}` JSON subgraph for `GRAPH TRAVERSE`.
 ///
 /// Each hop expands every frontier node at the node that owns
@@ -66,17 +77,19 @@ struct WireSubGraph<'a> {
 ///   * each `(src, label, dst)` edge the hop crossed where `src` is in the
 ///     current frontier — fully attributed for BOTH the local-shard and
 ///     remote-shard portions of the frontier.
-#[allow(clippy::too_many_arguments)]
 pub async fn cross_core_traverse_subgraph(
     shared: &SharedState,
-    tenant_id: TenantId,
-    database_id: DatabaseId,
-    start: String,
-    edge_label: Option<String>,
-    direction: Direction,
-    max_depth: usize,
-    options: &GraphTraversalOptions,
+    params: CrossCoreTraverseSubgraphParams<'_>,
 ) -> crate::Result<Response> {
+    let CrossCoreTraverseSubgraphParams {
+        tenant_id,
+        database_id,
+        start,
+        edge_label,
+        direction,
+        max_depth,
+        options,
+    } = params;
     // Per-node depth: the start node is at depth 0; subsequent nodes
     // are tagged with the hop index that first surfaced them.
     let mut depth_of: HashMap<String, u8> = HashMap::new();