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

Convert cluster (Calvin scheduler driver, raft startup), security scope
grant/expiry, DDL scope-grant, trace export, gateway dispatch, and
Data-Plane core spawn to take dedicated params structs instead of long
positional argument lists, updating all call sites accordingly. Also
split the core health watchdog out of data/runtime.rs into its own
data/core_health.rs module.

Author: farhan-syah

nodedb/src/bootstrap/data_plane.rs

@@ -12,7 +12,7 @@ use crate::bridge::quiesce::CollectionQuiesce;
 use crate::control::array_catalog::ArrayCatalog;
 use crate::control::metrics::SystemMetrics;
 use crate::data::eventfd::EventFdNotifier;
-use crate::data::runtime::{CoreCompactionConfig, spawn_core};
+use crate::data::runtime::{CoreCompactionConfig, SpawnCoreParams, spawn_core};
 use crate::event::EventProducer;
 use crate::storage::quarantine::QuarantineRegistry;
 
@@ -91,24 +91,24 @@ pub fn spawn_data_plane_cores(
     for (core_id, (data_side, event_producer)) in
         data_sides.into_iter().zip(event_producers).enumerate()
     {
-        let (handle, notifier) = spawn_core(
+        let (handle, notifier) = spawn_core(SpawnCoreParams {
             core_id,
-            data_side.request_rx,
-            data_side.response_tx,
-            &config.server.data_dir,
-            Arc::clone(&wal_records),
-            replay_tombstones.clone(),
+            request_rx: data_side.request_rx,
+            response_tx: data_side.response_tx,
+            data_dir: &config.server.data_dir,
+            wal_records: Arc::clone(&wal_records),
+            tombstones: replay_tombstones.clone(),
             num_cores,
-            compaction_cfg.clone(),
-            Some(Arc::clone(&system_metrics)),
-            Some(event_producer),
-            Arc::clone(&governor),
-            Some(Arc::clone(&quiesce)),
-            Arc::clone(&hlc),
-            Arc::clone(&array_catalog),
-            Arc::clone(&quarantine_registry),
-            Arc::clone(&maintenance_budget),
-        )?;
+            compaction_config: compaction_cfg.clone(),
+            system_metrics: Some(Arc::clone(&system_metrics)),
+            event_producer: Some(event_producer),
+            governor: Arc::clone(&governor),
+            quiesce: Some(Arc::clone(&quiesce)),
+            hlc: Arc::clone(&hlc),
+            array_catalog: Arc::clone(&array_catalog),
+            quarantine_registry: Arc::clone(&quarantine_registry),
+            maintenance_budget: Arc::clone(&maintenance_budget),
+        })?;
         core_handles.push(handle);
         notifiers.push((core_id, notifier));
     }

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

@@ -5,6 +5,6 @@ pub mod scheduler;
 
 pub use executor::{OllpConfig, OllpError, OllpOrchestrator};
 pub use scheduler::{
-    CalvinReadResultProposal, ReadResultEvent, Scheduler, SchedulerConfig,
+    CalvinReadResultProposal, ReadResultEvent, Scheduler, SchedulerConfig, SchedulerParams,
     propose_calvin_read_result,
 };

nodedb/src/control/cluster/calvin/scheduler/driver/core/mod.rs

@@ -39,4 +39,4 @@ pub mod scheduler;
 mod tests;
 
 pub use propose::{CalvinReadResultProposal, propose_calvin_read_result};
-pub use scheduler::Scheduler;
+pub use scheduler::{Scheduler, SchedulerParams};

nodedb/src/control/cluster/calvin/scheduler/driver/core/scheduler.rs

@@ -89,20 +89,34 @@ pub struct Scheduler {
         mpsc::Sender<CompletionItem>,
 }
 
+/// Parameters for [`Scheduler::new`].
+pub struct SchedulerParams {
+    pub vshard_id: u32,
+    pub receiver: mpsc::Receiver<SequencedTxn>,
+    pub shared: Arc<SharedState>,
+    pub multi_raft: Arc<Mutex<MultiRaft>>,
+    pub last_applied_epoch: u64,
+    pub rebuild_target_epoch: u64,
+    pub config: SchedulerConfig,
+    pub metrics: Arc<SchedulerMetrics>,
+    pub read_result_rx: mpsc::Receiver<ReadResultEvent>,
+}
+
 impl Scheduler {
     /// Construct a scheduler.
-    #[allow(clippy::too_many_arguments)]
-    pub fn new(
-        vshard_id: u32,
-        receiver: mpsc::Receiver<SequencedTxn>,
-        shared: Arc<SharedState>,
-        multi_raft: Arc<Mutex<MultiRaft>>,
-        last_applied_epoch: u64,
-        rebuild_target_epoch: u64,
-        config: SchedulerConfig,
-        metrics: Arc<SchedulerMetrics>,
-        read_result_rx: mpsc::Receiver<ReadResultEvent>,
-    ) -> Self {
+    pub fn new(params: SchedulerParams) -> Self {
+        let SchedulerParams {
+            vshard_id,
+            receiver,
+            shared,
+            multi_raft,
+            last_applied_epoch,
+            rebuild_target_epoch,
+            config,
+            metrics,
+            read_result_rx,
+        } = params;
+
         // Capacity: at most one completion per inflight txn. Use the incoming
         // channel capacity as a proxy for the max concurrent pending count.
         let completion_cap = config.channel_capacity;

nodedb/src/control/cluster/calvin/scheduler/driver/mod.rs

@@ -8,4 +8,4 @@ pub mod types;
 
 pub use barrier::ReadResultEvent;
 pub use config::SchedulerConfig;
-pub use core::{CalvinReadResultProposal, Scheduler, propose_calvin_read_result};
+pub use core::{CalvinReadResultProposal, Scheduler, SchedulerParams, propose_calvin_read_result};

nodedb/src/control/cluster/calvin/scheduler/mod.rs

@@ -6,7 +6,7 @@ pub mod metrics;
 pub mod recovery;
 
 pub use driver::{
-    CalvinReadResultProposal, ReadResultEvent, Scheduler, SchedulerConfig,
+    CalvinReadResultProposal, ReadResultEvent, Scheduler, SchedulerConfig, SchedulerParams,
     propose_calvin_read_result,
 };
 pub use lock_manager::{AcquireOutcome, LockKey, LockManager, TxnId};

nodedb/src/control/cluster/start_raft.rs

@@ -22,7 +22,9 @@ use crate::control::cluster::handle::ClusterHandle;
 use crate::control::cluster::metadata_applier::MetadataCommitApplier;
 use crate::control::cluster::snapshot_hook::RaftSnapshotQuarantineHook;
 use crate::control::cluster::spsc_applier::SpscCommitApplier;
-use crate::control::cluster::start_raft_helpers::{build_vshard_handler, spawn_vshard_schedulers};
+use crate::control::cluster::start_raft_helpers::{
+    SpawnVshardSchedulersParams, build_vshard_handler, spawn_vshard_schedulers,
+};
 use crate::control::distributed_applier::{
     ProposeTracker, create_distributed_applier, run_apply_loop,
 };
@@ -325,14 +327,14 @@ pub fn start_raft(
     let sequencer_metrics = Arc::clone(&sequencer_service.metrics);
 
     let scheduler_config = SchedulerConfig::default();
-    spawn_vshard_schedulers(
+    spawn_vshard_schedulers(SpawnVshardSchedulersParams {
         handle,
-        &shared,
-        raft_loop_handle.clone(),
-        &sequencer_state_machine,
-        &calvin_read_result_senders,
-        &scheduler_config,
-    )?;
+        shared: &shared,
+        raft_loop_handle: raft_loop_handle.clone(),
+        sequencer_state_machine: &sequencer_state_machine,
+        calvin_read_result_senders: &calvin_read_result_senders,
+        scheduler_config: &scheduler_config,
+    })?;
 
     let running = tokio::task::block_in_place(|| {
         tokio::runtime::Handle::current().block_on(nodedb_cluster::start_cluster_subsystems(

nodedb/src/control/cluster/start_raft_helpers.rs

@@ -10,7 +10,9 @@ use nodedb_cluster::wire::VShardEnvelope;
 
 use crate::control::cluster::calvin::scheduler::metrics::SchedulerMetrics;
 use crate::control::cluster::calvin::scheduler::read_last_applied_epoch;
-use crate::control::cluster::calvin::{ReadResultEvent, Scheduler, SchedulerConfig};
+use crate::control::cluster::calvin::{
+    ReadResultEvent, Scheduler, SchedulerConfig, SchedulerParams,
+};
 use crate::control::cluster::handle::ClusterHandle;
 use crate::control::state::SharedState;
 
@@ -91,6 +93,17 @@ fn hosted_vshards(routing: &RwLock<nodedb_cluster::RoutingTable>, node_id: u64)
     vshards
 }
 
+/// Parameters for [`reconcile_vshard_schedulers`].
+struct ReconcileSchedulersParams<'a> {
+    node_id: u64,
+    routing: &'a Arc<RwLock<nodedb_cluster::RoutingTable>>,
+    shared: &'a Arc<SharedState>,
+    raft_loop_handle: &'a Arc<Mutex<nodedb_cluster::multi_raft::MultiRaft>>,
+    sequencer_state_machine: &'a Arc<Mutex<SequencerStateMachine>>,
+    calvin_read_result_senders: &'a ReadResultSenders,
+    scheduler_config: &'a SchedulerConfig,
+}
+
 /// Idempotently ensure a Calvin `Scheduler` is running for every vShard this
 /// node currently hosts.
 ///
@@ -103,16 +116,17 @@ fn hosted_vshards(routing: &RwLock<nodedb_cluster::RoutingTable>, node_id: u64)
 /// left this node. vShard removal happens via migration / decommission, which
 /// own their own teardown path; wiring scheduler removal into that lifecycle is
 /// tracked as a separate follow-up.
-#[allow(clippy::too_many_arguments)]
-fn reconcile_vshard_schedulers(
-    node_id: u64,
-    routing: &Arc<RwLock<nodedb_cluster::RoutingTable>>,
-    shared: &Arc<SharedState>,
-    raft_loop_handle: &Arc<Mutex<nodedb_cluster::multi_raft::MultiRaft>>,
-    sequencer_state_machine: &Arc<Mutex<SequencerStateMachine>>,
-    calvin_read_result_senders: &ReadResultSenders,
-    scheduler_config: &SchedulerConfig,
-) -> crate::Result<usize> {
+fn reconcile_vshard_schedulers(params: ReconcileSchedulersParams<'_>) -> crate::Result<usize> {
+    let ReconcileSchedulersParams {
+        node_id,
+        routing,
+        shared,
+        raft_loop_handle,
+        sequencer_state_machine,
+        calvin_read_result_senders,
+        scheduler_config,
+    } = params;
+
     let mut spawned = 0usize;
     for vshard_id in hosted_vshards(routing, node_id) {
         // Already-served vShards keep their running scheduler untouched.
@@ -139,17 +153,17 @@ fn reconcile_vshard_schedulers(
             .unwrap_or_else(|p| p.into_inner())
             .insert(vshard_id, read_result_tx);
 
-        let scheduler = Scheduler::new(
+        let scheduler = Scheduler::new(SchedulerParams {
             vshard_id,
-            sequenced_rx,
-            Arc::clone(shared),
-            raft_loop_handle.clone(),
+            receiver: sequenced_rx,
+            shared: Arc::clone(shared),
+            multi_raft: raft_loop_handle.clone(),
             last_applied_epoch,
-            last_applied_epoch,
-            scheduler_config.clone(),
-            SchedulerMetrics::new(),
+            rebuild_target_epoch: last_applied_epoch,
+            config: scheduler_config.clone(),
+            metrics: SchedulerMetrics::new(),
             read_result_rx,
-        );
+        });
         let shutdown = shared.shutdown.subscribe();
         tokio::spawn(async move {
             scheduler.run(shutdown).await;
@@ -159,6 +173,16 @@ fn reconcile_vshard_schedulers(
     Ok(spawned)
 }
 
+/// Parameters for [`spawn_vshard_schedulers`].
+pub(super) struct SpawnVshardSchedulersParams<'a> {
+    pub(super) handle: &'a ClusterHandle,
+    pub(super) shared: &'a Arc<SharedState>,
+    pub(super) raft_loop_handle: Arc<Mutex<nodedb_cluster::multi_raft::MultiRaft>>,
+    pub(super) sequencer_state_machine: &'a Arc<Mutex<SequencerStateMachine>>,
+    pub(super) calvin_read_result_senders: &'a ReadResultSenders,
+    pub(super) scheduler_config: &'a SchedulerConfig,
+}
+
 /// Spawn Calvin `Scheduler` tasks for this node's vShards and keep the set in
 /// sync with cluster membership.
 ///
@@ -174,28 +198,31 @@ fn reconcile_vshard_schedulers(
 /// (covers the bootstrap node, which already sees its membership) and then
 /// spawns a background task that re-reconciles on a short interval until
 /// shutdown. Reconcile is idempotent and add-only.
-#[allow(clippy::too_many_arguments)]
 pub(super) fn spawn_vshard_schedulers(
-    handle: &ClusterHandle,
-    shared: &Arc<SharedState>,
-    raft_loop_handle: Arc<Mutex<nodedb_cluster::multi_raft::MultiRaft>>,
-    sequencer_state_machine: &Arc<Mutex<SequencerStateMachine>>,
-    calvin_read_result_senders: &ReadResultSenders,
-    scheduler_config: &SchedulerConfig,
+    params: SpawnVshardSchedulersParams<'_>,
 ) -> crate::Result<()> {
+    let SpawnVshardSchedulersParams {
+        handle,
+        shared,
+        raft_loop_handle,
+        sequencer_state_machine,
+        calvin_read_result_senders,
+        scheduler_config,
+    } = params;
+
     let node_id = handle.node_id;
     let routing = Arc::clone(&handle.routing);
 
     // Initial reconcile: schedulers for vShards this node already knows it hosts.
-    reconcile_vshard_schedulers(
+    reconcile_vshard_schedulers(ReconcileSchedulersParams {
         node_id,
-        &routing,
+        routing: &routing,
         shared,
-        &raft_loop_handle,
+        raft_loop_handle: &raft_loop_handle,
         sequencer_state_machine,
         calvin_read_result_senders,
         scheduler_config,
-    )?;
+    })?;
 
     // Background reconcile: pick up vShards whose membership lands after startup
     // (joiner admission) or shifts later (rebalancing). The routing table has no
@@ -215,15 +242,15 @@ pub(super) fn spawn_vshard_schedulers(
                 biased;
                 _ = shutdown.wait_cancelled() => break,
                 _ = tick.tick() => {
-                    if let Err(e) = reconcile_vshard_schedulers(
+                    if let Err(e) = reconcile_vshard_schedulers(ReconcileSchedulersParams {
                         node_id,
-                        &routing,
-                        &shared_task,
-                        &raft_loop_handle,
-                        &sm_task,
-                        &rr_task,
-                        &cfg_task,
-                    ) {
+                        routing: &routing,
+                        shared: &shared_task,
+                        raft_loop_handle: &raft_loop_handle,
+                        sequencer_state_machine: &sm_task,
+                        calvin_read_result_senders: &rr_task,
+                        scheduler_config: &cfg_task,
+                    }) {
                         tracing::warn!(node_id, error = %e, "calvin scheduler reconcile pass failed");
                     }
                 }

nodedb/src/control/exec_receiver/executor.rs

@@ -23,6 +23,7 @@ use nodedb_cluster::rpc_codec::{ExecuteRequest, ExecuteResponse, TypedClusterErr
 
 use crate::control::server::exchange::execute_plan_all_local_cores;
 use crate::control::state::SharedState;
+use crate::control::trace_export::EmitSpanParams;
 use crate::types::DatabaseId;
 use nodedb_physical::physical_plan::wire as plan_wire;
 
@@ -52,15 +53,15 @@ impl PlanExecutor for LocalPlanExecutor {
         // Emit one OTLP executor span per leaseholder so the gateway's
         // upstream span joins the N leaseholder spans into a single
         // distributed trace via the shared `trace_id`.
-        exporter.emit(
-            "executor.execute_plan",
+        exporter.emit(EmitSpanParams {
+            span_name: "executor.execute_plan",
             trace_id,
             start,
-            SystemTime::now(),
+            end: SystemTime::now(),
             tenant_id,
-            0,
-            resp.success,
-        );
+            vshard_id: 0,
+            status_ok: resp.success,
+        });
         resp
     }
 
@@ -78,15 +79,15 @@ impl PlanExecutor for LocalPlanExecutor {
             .execute_plan_streaming_inner(req, sink)
             .instrument(span)
             .await;
-        exporter.emit(
-            "executor.execute_plan_streaming",
+        exporter.emit(EmitSpanParams {
+            span_name: "executor.execute_plan_streaming",
             trace_id,
             start,
-            SystemTime::now(),
+            end: SystemTime::now(),
             tenant_id,
-            0,
-            outcome.is_none(),
-        );
+            vshard_id: 0,
+            status_ok: outcome.is_none(),
+        });
         outcome
     }
 }