Cleanup (#1588)

* Remove unused remote communication field from dataflow config

* Resolve various clippy warnings

Author: phil-opp

apis/rust/node/src/event_stream/thread.rs

@@ -23,7 +23,7 @@ pub fn init(
 ) -> eyre::Result<EventStreamThreadHandle> {
     let node_id_cloned = node_id.clone();
     let join_handle = std::thread::spawn(|| event_stream_loop(node_id_cloned, tx, channel, clock));
-    Ok(EventStreamThreadHandle::new(node_id, join_handle))
+    Ok(EventStreamThreadHandle::new(join_handle))
 }
 
 #[derive(Debug)]
@@ -38,20 +38,16 @@ pub enum EventItem {
 }
 
 pub struct EventStreamThreadHandle {
-    node_id: NodeId,
     handle: flume::Receiver<std::thread::Result<()>>,
 }
 
 impl EventStreamThreadHandle {
-    fn new(node_id: NodeId, join_handle: std::thread::JoinHandle<()>) -> Self {
+    fn new(join_handle: std::thread::JoinHandle<()>) -> Self {
         let (tx, rx) = flume::bounded(1);
         std::thread::spawn(move || {
             let _ = tx.send(join_handle.join());
         });
-        Self {
-            node_id,
-            handle: rx,
-        }
+        Self { handle: rx }
     }
 }
 

apis/rust/node/src/node/arrow_utils.rs

@@ -145,7 +145,7 @@ mod tests {
             child_data: vec![],
         };
 
-        let raw_buffer = arrow::buffer::Buffer::from_slice_ref(&[1, 2, 3]);
+        let raw_buffer = arrow::buffer::Buffer::from_slice_ref([1, 2, 3]);
 
         let result = buffer_into_arrow_array(&raw_buffer, &type_info);
         assert!(result.is_err());

binaries/cli/src/command/inspect/top.rs

@@ -257,20 +257,12 @@ async fn run_app<B: Backend>(
 ) -> eyre::Result<()> {
     let mut app = App::new();
     let mut last_update = Instant::now();
-    let mut node_infos: Vec<NodeInfo> = Vec::new();
 
     // Reuse coordinator connection
     let client = connect_and_check_version(coordinator_addr, coordinator_port)
         .await
         .wrap_err("Failed to connect to coordinator")?;
 
-    // Query node info once initially
-    node_infos = rpc(
-        "get node info",
-        client.get_node_info(tarpc::context::current()),
-    )
-    .await?;
-
     loop {
         terminal.draw(|f| ui(f, &mut app, refresh_duration))?;
 
@@ -323,14 +315,14 @@ async fn run_app<B: Backend>(
         // Update data if refresh interval has passed
         if last_update.elapsed() >= refresh_duration {
             // Query node info every refresh interval to get updated metrics
-            node_infos = rpc(
+            let node_infos = rpc(
                 "refresh node info",
                 client.get_node_info(tarpc::context::current()),
             )
             .await?;
 
             // Update stats with current node info
-            app.update_stats(node_infos.clone());
+            app.update_stats(node_infos);
             last_update = Instant::now();
         }
     }
@@ -349,7 +341,7 @@ fn ui(f: &mut Frame, app: &mut App, refresh_duration: Duration) {
         }
     };
 
-    let header_strings = vec![
+    let header_strings = [
         format!("NODE{}", sort_indicator(SortColumn::Node)),
         "DATAFLOW".to_string(),
         "PID".to_string(),

binaries/cli/src/command/list.rs

@@ -11,7 +11,7 @@ use dora_core::topics::DORA_COORDINATOR_PORT_CONTROL_DEFAULT;
 use dora_message::{
     cli_to_coordinator::CoordinatorControlClient, coordinator_to_cli::DataflowStatus, tarpc,
 };
-use eyre::{Context, eyre};
+use eyre::Context;
 use serde::Serialize;
 use tabwriter::TabWriter;
 use uuid::Uuid;
@@ -89,9 +89,7 @@ async fn list(
         std::collections::BTreeMap::new();
 
     for node_info in node_infos {
-        let metrics = dataflow_metrics
-            .entry(node_info.dataflow_id)
-            .or_insert_with(DataflowMetrics::default);
+        let metrics = dataflow_metrics.entry(node_info.dataflow_id).or_default();
         metrics.node_count += 1;
 
         if let Some(node_metrics) = node_info.metrics {
@@ -182,7 +180,11 @@ async fn list(
         OutputFormat::Table => {
             let mut tw = TabWriter::new(std::io::stdout().lock());
             // Header
-            tw.write_all(format!("UUID\tName\tStatus\tNodes\tCPU\tMemory\n").as_bytes())?;
+            tw.write_all(
+                "UUID\tName\tStatus\tNodes\tCPU\tMemory\n"
+                    .to_string()
+                    .as_bytes(),
+            )?;
             for entry in entries {
                 let status = match entry.status {
                     DataflowStatus::Running => "Running",
@@ -192,13 +194,8 @@ async fn list(
 
                 tw.write_all(
                     format!(
-                        "{}\t{}\t{}\t{}\t{}\t{}\n",
-                        entry.uuid,
-                        entry.name,
-                        status,
-                        entry.nodes,
-                        format!("{:.1}%", entry.cpu),
-                        format!("{:.1} GB", entry.memory)
+                        "{}\t{}\t{}\t{}\t{:.1}%\t{:.1} GB\n",
+                        entry.uuid, entry.name, status, entry.nodes, entry.cpu, entry.memory
                     )
                     .as_bytes(),
                 )?;

binaries/cli/src/command/start/mod.rs

@@ -23,7 +23,7 @@ use dora_message::{
 use eyre::Context;
 use std::{
     net::{IpAddr, SocketAddr},
-    path::PathBuf,
+    path::Path,
 };
 use tokio::task::JoinHandle;
 use uuid::{NoContext, Timestamp, Uuid};
@@ -152,7 +152,7 @@ impl Executable for Start {
 /// Send the start RPC to the coordinator.  The caller should have already
 /// subscribed to zenoh log topics before calling this.
 async fn send_start_rpc(
-    dataflow: &PathBuf,
+    dataflow: &Path,
     dataflow_descriptor: &Descriptor,
     dataflow_session: &DataflowSession,
     client: &CoordinatorControlClient,

binaries/cli/src/command/topic/hz.rs

@@ -347,16 +347,6 @@ fn ui(
     start: Instant,
     window_dur: Duration,
 ) {
-    // Layout: table | charts | footer
-    let chunks = Layout::default()
-        .direction(Direction::Vertical)
-        .constraints([
-            Constraint::Percentage(55),
-            Constraint::Percentage(44),
-            Constraint::Length(1),
-        ])
-        .split(f.area());
-
     // Table header: interval stats in ms + derived avg Hz
     let header = Row::new([
         "Output", "Avg (ms)", "Avg (Hz)", "Min (ms)", "Max (ms)", "Std (ms)",

binaries/cli/src/output.rs

@@ -185,7 +185,7 @@ pub fn print_log_message(
             let node_id = node_id
                 .to_string()
                 .bold()
-                .color(word_to_color(&node_id.to_string()));
+                .color(word_to_color(node_id.as_ref()));
             let padding = if daemon.is_empty() { "" } else { " " };
             format!("{node_id}{padding}{daemon}{colon} ")
         }

binaries/coordinator/src/server.rs

@@ -382,7 +382,7 @@ impl CoordinatorControl for CoordinatorControlServer {
         let mut node_infos = Vec::new();
         for r in self.state.running_dataflows.iter() {
             let dataflow = r.value();
-            for (node_id, _node) in &dataflow.nodes {
+            for node_id in dataflow.nodes.keys() {
                 // Get the specific daemon this node is running on
                 if let Some(daemon_id) = dataflow.node_to_daemon.get(node_id) {
                     // Get metrics if available

binaries/daemon/src/coordinator.rs

@@ -173,10 +173,6 @@ impl DaemonControl for DaemonControlServer {
             uv,
         } = request;
 
-        match dataflow_descriptor.communication.remote {
-            dora_core::config::RemoteCommunicationConfig::Tcp => {}
-        }
-
         let base_working_dir =
             crate::Daemon::base_working_dir_static(local_working_dir, session_id)
                 .map_err(|err| format!("{err:?}"))?;
@@ -247,10 +243,6 @@ impl DaemonControl for DaemonControlServer {
             write_events_to,
         } = request;
 
-        match dataflow_descriptor.communication.remote {
-            dora_core::config::RemoteCommunicationConfig::Tcp => {}
-        }
-
         // For spawn, we still route through the event loop because spawn_dataflow
         // needs mutable access to the logger and complex event loop integration.
         // TODO: Move spawn logic here once logger is refactored.

binaries/daemon/src/lib.rs

@@ -1,7 +1,7 @@
 use aligned_vec::{AVec, ConstAlign};
 use crossbeam::queue::ArrayQueue;
 use dora_core::{
-    build::{self, BuildInfo, PrevGitSource, TracingBuildLogger},
+    build::{self, BuildInfo, PrevGitSource},
     config::{DataId, Input, InputMapping, NodeId, NodeRunConfig},
     descriptor::{
         CoreNodeKind, DYNAMIC_SOURCE, Descriptor, DescriptorExt, ResolvedNode, RuntimeNode,
@@ -56,11 +56,10 @@ use std::{
 use tokio::{
     fs::File,
     io::{AsyncReadExt, AsyncSeekExt},
-    net::TcpStream,
     sync::{
         broadcast,
         mpsc::{self, UnboundedSender},
-        oneshot::{self, Sender},
+        oneshot::{self},
     },
 };
 use tokio_stream::{Stream, StreamExt, wrappers::ReceiverStream};
@@ -222,8 +221,7 @@ impl Daemon {
         let (events_tx, events_rx) = flume::bounded(10);
         if nodes
             .iter()
-            .find(|(_n, resolved_nodes)| resolved_nodes.kind.dynamic())
-            .is_some()
+            .any(|(_n, resolved_nodes)| resolved_nodes.kind.dynamic())
         {
             // Spawn local listener for dynamic nodes
             let _listen_port = local_listener::spawn_listener_loop(
@@ -709,10 +707,6 @@ impl Daemon {
         uv: bool,
         write_events_to: Option<PathBuf>,
     ) -> Result<(), String> {
-        match dataflow_descriptor.communication.remote {
-            dora_core::config::RemoteCommunicationConfig::Tcp => {}
-        }
-
         // Resolve base working dir — for spawn we use the daemon's working dir
         let base_working_dir = match local_working_dir {
             Some(working_dir) => {
@@ -879,7 +873,7 @@ impl Daemon {
                     send_output_to_local_receivers(
                         node_id.clone(),
                         output_id.clone(),
-                        &mut *dataflow,
+                        &mut dataflow,
                         &metadata,
                         data.map(DataMessage::Vec),
                         &self.state.clock,
@@ -921,7 +915,7 @@ impl Daemon {
                         if let Some(inputs) = dataflow.mappings.get(&output_id).cloned() {
                             for (receiver_id, input_id) in &inputs {
                                 close_input(
-                                    &mut *dataflow,
+                                    &mut dataflow,
                                     receiver_id,
                                     input_id,
                                     &self.state.clock,
@@ -1461,7 +1455,7 @@ impl Daemon {
                     }
                     Ok(mut dataflow) => {
                         Self::subscribe(
-                            &mut *dataflow,
+                            &mut dataflow,
                             node_id.clone(),
                             event_sender,
                             &self.state.clock,
@@ -1615,7 +1609,7 @@ impl Daemon {
         let data_bytes = send_output_to_local_receivers(
             node_id.clone(),
             output_id.clone(),
-            &mut *dataflow,
+            &mut dataflow,
             &metadata,
             data,
             &self.state.clock,
@@ -1846,7 +1840,7 @@ impl Daemon {
                 .cloned()
                 .collect();
             for (receiver_id, input_id) in &local_node_inputs {
-                close_input(&mut *dataflow, receiver_id, input_id, &self.state.clock);
+                close_input(&mut dataflow, receiver_id, input_id, &self.state.clock);
             }
 
             let mut closed = Vec::new();
@@ -2027,7 +2021,7 @@ impl Daemon {
 
             let df = &mut *dataflow;
             df.pending_nodes
-                .handle_node_stop_sync(node_id, &mut df.cascading_error_causes)
+                .handle_node_stop(node_id, &mut df.cascading_error_causes)
         };
         // DashMap guard is dropped — safe to do async I/O.
         if exited_before_subscribe {
@@ -2573,7 +2567,7 @@ async fn read_last_n_lines(file: &mut File, mut tail: usize) -> io::Result<Vec<u
         file.read_exact(&mut buffer[..read_len]).await?;
         let read_buf = if at_end {
             at_end = false;
-            &buffer[..read_len].trim_ascii_end()
+            buffer[..read_len].trim_ascii_end()
         } else {
             &buffer[..read_len]
         };
@@ -2986,7 +2980,7 @@ impl RunningDataflow {
         clock: &Arc<HLC>,
     ) -> eyre::Result<()> {
         for interval in self.timers.keys().copied() {
-            if self._timer_handles.get(&interval).is_some() {
+            if self._timer_handles.contains_key(&interval) {
                 continue;
             }
             let events_tx = events_tx.clone();