Run fmt, clippy and tests in CI

.github/workflows/ci.yml

@@ -0,0 +1,42 @@
+name: ci
+on:
+  pull_request:
+  push:
+    branches: [main]
+
+permissions:
+  contents: read
+
+env:
+  CARGO_TERM_COLOR: always
+
+jobs:
+  fmt:
+    name: fmt
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      # Installs the toolchain and components pinned in `rust-toolchain`.
+      - run: rustup toolchain install
+      - run: cargo fmt --all -- --check
+
+  clippy:
+    name: clippy
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - run: rustup toolchain install
+      # sqd-messages compiles its protobuf definitions at build time.
+      - run: sudo apt-get update && sudo apt-get install -y protobuf-compiler
+      - uses: Swatinem/rust-cache@v2
+      - run: cargo clippy --workspace --all-targets --locked -- -D warnings
+
+  test:
+    name: test
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - run: rustup toolchain install
+      - run: sudo apt-get update && sudo apt-get install -y protobuf-compiler
+      - uses: Swatinem/rust-cache@v2
+      - run: cargo test --workspace --locked

crates/collector-utils/src/storage.rs

@@ -7,7 +7,9 @@ use serde::{Deserialize, Serialize};
 use serde_repr::{Deserialize_repr, Serialize_repr};
 
 use sqd_messages::signatures::sha3_256;
-use sqd_messages::{query_error, query_executed, query_finished, Heartbeat, QueryExecuted, QueryFinished};
+use sqd_messages::{
+    query_error, query_executed, query_finished, Heartbeat, QueryExecuted, QueryFinished,
+};
 use sqd_network_transport::{protocol, PeerId};
 
 use crate::cli::ClickhouseArgs;
@@ -378,15 +380,13 @@ pub struct QueryFinishedRow {
 }
 
 impl QueryFinishedRow {
-    pub fn try_from(
-        query_finished: QueryFinished,
-    ) -> Result<Self, &'static str> {
+    pub fn try_from(query_finished: QueryFinished) -> Result<Self, &'static str> {
         let QueryFinished {
             worker_id,
             query_id,
             total_time_micros: total_time,
             worker_signature,
-            result
+            result,
         } = query_finished;
         let result_hash = match result {
             Some(query_finished::Result::Ok(ok)) => ok.data_hash,
@@ -406,7 +406,6 @@ impl QueryFinishedRow {
     }
 }
 
-
 impl ClickhouseStorage {
     pub async fn new(args: ClickhouseArgs) -> anyhow::Result<Self> {
         let client = Client::default()
@@ -416,7 +415,10 @@ impl ClickhouseStorage {
             .with_password(args.clickhouse_password);
         client.query(&LOGS_TABLE_DEFINITION).execute().await?;
         client.query(&PINGS_TABLE_DEFINITION).execute().await?;
-        client.query(&PORTAL_LOGS_TABLE_DEFINITION).execute().await?;
+        client
+            .query(&PORTAL_LOGS_TABLE_DEFINITION)
+            .execute()
+            .await?;
         Ok(Self(client))
     }
 }
@@ -603,7 +605,7 @@ mod tests {
         let last_stored = storage.get_last_stored().await.unwrap();
         assert_eq!(last_stored.get(&worker_id.to_string()), Some(&123456789500));
 
-        let assignment_id = 
+        let assignment_id =
             "2025-10-12T12:00:45_C1A955A7E13FABEC64DCA7965104FA0CBF98C063A6FCB4473E243348CADFAFAE";
 
         // Check pings storing

crates/pings-collector/src/server.rs

@@ -65,12 +65,10 @@ impl Server {
 
         let mut sigint = signal(SignalKind::interrupt())?;
         let mut sigterm = signal(SignalKind::terminate())?;
-        loop {
-            tokio::select! {
-                _ = sigint.recv() => break,
-                _ = sigterm.recv() => break,
-                else => break
-            }
+        tokio::select! {
+            _ = sigint.recv() => {}
+            _ = sigterm.recv() => {}
+            else => {}
         }
         tracing::info!("Server shutting down");
         self.task_manager.await_stop().await;
@@ -162,7 +160,7 @@ impl Server {
                     })
                     .buffered(concurrency_limit)
                     .take_until(cancellation_token.cancelled_owned())
-                    .filter_map(|x| std::future::ready(x))
+                    .filter_map(std::future::ready)
                     .collect()
                     .await;
 

crates/router-controller/src/controller.rs

@@ -10,7 +10,7 @@ use base64::Engine;
 use rand::prelude::SliceRandom;
 use serde::{Deserialize, Serialize};
 
-use super::messages::{Range, RangeSet, WorkerState, DataChunk};
+use super::messages::{DataChunk, Range, RangeSet, WorkerState};
 
 use crate::atom::Atom;
 
@@ -527,7 +527,7 @@ mod tests {
             .set_datasets((0..2).map(|i| (i.to_string(), i.to_string())))
             .build();
 
-        let chunks = vec![
+        let chunks = [
             vec![
                 DataChunk::new(0, 0, 10, "".to_string()),
                 DataChunk::new(0, 11, 200, "".to_string()),

crates/router-controller/src/messages.rs

@@ -3,7 +3,7 @@ use std::{
     ops::{Deref, DerefMut},
 };
 
-pub use sqd_messages::{Range, RangeSet, data_chunk::DataChunk};
+pub use sqd_messages::{data_chunk::DataChunk, Range, RangeSet};
 
 #[derive(serde::Serialize, serde::Deserialize, Default, Debug, Clone, PartialEq, Eq)]
 pub struct WorkerState {