scopedb · tisonkun · May 26, 2025 · May 26, 2025 · May 26, 2025
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -53,7 +53,7 @@ backon = { version = "1.5.0" }
 build-data = { version = "0.3.0" }
 clap = { version = "4.5.35", features = ["derive"] }
 const_format = { version = "0.2.34" }
-criterion = { version = "0.5.1", features = ["async_tokio"] }
+criterion = { version = "0.6.0", features = ["async_tokio"] }
 ctrlc = { version = "3.4", features = ["termination"] }
 error-stack = { version = "0.5", default-features = false, features = [
   "std",
@@ -97,7 +97,7 @@ pretty-hex = { version = "0.4.1" }
 pretty_assertions = { version = "1.4.1" }
 regex = { version = "1.11.1" }
 reqwest = { version = "0.12.15", default-features = false, features = ["json"] }
-schemars = { version = "1.0.0-alpha.17" }
+schemars = { version = "1.0.0-alpha.18", features = ["jiff02"] }
 scopeguard = { version = "1.2.0" }
 sealed_test = { version = "1.1" }
 serde = { version = "1.0", features = ["derive"] }

diff --git a/crates/cluster/src/gossip.rs b/crates/cluster/src/gossip.rs
@@ -150,33 +150,31 @@ impl GossipState {
     }
 
     fn handle_message(&self, message: Message) -> Option<Message> {
-        log::debug!("received message: {:?}", message);
+        log::debug!("received message: {message:?}");
         let result = match message {
             Message::Ping(info) => {
                 if let Some(current) = self.membership.read().unwrap().members().get(&info.node_id)
+                    && current.info.incarnation < info.incarnation
                 {
-                    if current.info.incarnation < info.incarnation {
-                        self.membership.write().unwrap().update_member(MemberState {
-                            info: info.clone(),
-                            status: MemberStatus::Alive,
-                            heartbeat: Timestamp::now(),
-                        });
-                    }
+                    self.membership.write().unwrap().update_member(MemberState {
+                        info: info.clone(),
+                        status: MemberStatus::Alive,
+                        heartbeat: Timestamp::now(),
+                    });
                 }
 
                 // Respond with an ack
                 Some(Message::Ack(self.current()))
             }
             Message::Ack(info) => {
                 if let Some(current) = self.membership.read().unwrap().members().get(&info.node_id)
+                    && current.info.incarnation < info.incarnation
                 {
-                    if current.info.incarnation < info.incarnation {
-                        self.membership.write().unwrap().update_member(MemberState {
-                            info: info.clone(),
-                            status: MemberStatus::Alive,
-                            heartbeat: Timestamp::now(),
-                        });
-                    }
+                    self.membership.write().unwrap().update_member(MemberState {
+                        info: info.clone(),
+                        status: MemberStatus::Alive,
+                        heartbeat: Timestamp::now(),
+                    });
                 }
 
                 None
@@ -263,7 +261,7 @@ impl GossipState {
             self.transport
                 .send(&peer.advertise_peer_addr, &message)
                 .await
-                .inspect_err(|e| log::error!("failed to send ping message: {:?}", e))
+                .inspect_err(|e| log::error!("failed to send ping message: {e:?}"))
         };
         let with_retry = do_send.retry(
             ConstantBuilder::new()
@@ -290,7 +288,7 @@ impl GossipState {
             self.transport
                 .send(&peer.advertise_peer_addr, &message)
                 .await
-                .inspect_err(|e| log::error!("failed to send sync message: {:?}", e))
+                .inspect_err(|e| log::error!("failed to send sync message: {e:?}"))
         };
         let with_retry = do_send.retry(
             ConstantBuilder::new()
@@ -315,7 +313,7 @@ impl GossipState {
                 self.transport
                     .send(peer, &message)
                     .await
-                    .inspect_err(|e| log::error!("failed to send ping message: {:?}", e))
+                    .inspect_err(|e| log::error!("failed to send ping message: {e:?}"))
             };
             let with_retry = do_send.retry(
                 ConstantBuilder::new()
@@ -335,7 +333,7 @@ impl GossipState {
                 self.transport
                     .send(peer, &message)
                     .await
-                    .inspect_err(|e| log::error!("failed to send sync message: {:?}", e))
+                    .inspect_err(|e| log::error!("failed to send sync message: {e:?}"))
             };
             let with_retry = do_send.retry(
                 ConstantBuilder::new()
@@ -560,7 +558,7 @@ async fn drive_gossip(
                 ticker.tick().await;
                 let dead_members = state.remove_dead_members();
                 if !dead_members.is_empty() {
-                    log::info!("removed dead members: {:?}", dead_members);
+                    log::info!("removed dead members: {dead_members:?}");
                     state.rebuild_ring();
                 }
             }
@@ -581,7 +579,7 @@ async fn drive_gossip(
 
 #[handler]
 async fn gossip(Json(msg): Json<Message>, Data(state): Data<&Arc<GossipState>>) -> Response {
-    log::debug!("received message: {:?}", msg);
+    log::debug!("received message: {msg:?}");
 
     if let Some(response) = state.handle_message(msg) {
         Json(response).into_response()

diff --git a/crates/cluster/src/proxy.rs b/crates/cluster/src/proxy.rs
@@ -40,10 +40,10 @@ impl Proxy {
         let members = membership.members();
 
         if let Some(id) = ring.lookup_until(key, |id| {
-            if let Some(member) = members.get(id) {
-                if member.status == MemberStatus::Alive {
-                    return true;
-                }
+            if let Some(member) = members.get(id)
+                && member.status == MemberStatus::Alive
+            {
+                return true;
             }
             false
         }) {
@@ -57,10 +57,7 @@ impl Proxy {
                 RouteDest::Local
             }
         } else {
-            log::debug!(
-                "no target found for key: [{key}] , current ring: {:#?}",
-                ring
-            );
+            log::debug!("no target found for key: [{key}] , current ring: {ring:#?}");
             RouteDest::Local
         }
     }

diff --git a/crates/core/benches/benchmark.rs b/crates/core/benches/benchmark.rs
@@ -16,7 +16,6 @@
 
 use criterion::BenchmarkId;
 use criterion::Criterion;
-use criterion::black_box;
 use criterion::criterion_group;
 use criterion::criterion_main;
 use percas_core::FoyerEngine;
@@ -74,7 +73,7 @@ fn foyer_engine(c: &mut Criterion) {
                     |b, s| {
                         b.to_async(&runtime).iter(|| async {
                             let key = &s[std::random::random::<usize>() % keys.len()];
-                            black_box(engine.get(key).await);
+                            std::hint::black_box(engine.get(key).await);
                         })
                     },
                 );

diff --git a/crates/core/src/config.rs b/crates/core/src/config.rs
@@ -19,7 +19,6 @@ use serde::Deserialize;
 use serde::Serialize;
 
 use crate::newtype::DiskThrottle;
-use crate::newtype::SignedDuration;
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(test, derive(schemars::JsonSchema))]
@@ -188,11 +187,11 @@ pub struct MetricsConfig {
 pub struct OpentelemetryMetricsConfig {
     pub otlp_endpoint: String,
     #[serde(default = "default_metrics_push_interval")]
-    pub push_interval: SignedDuration,
+    pub push_interval: jiff::SignedDuration,
 }
 
-fn default_metrics_push_interval() -> SignedDuration {
-    jiff::SignedDuration::from_secs(30).into()
+const fn default_metrics_push_interval() -> jiff::SignedDuration {
+    jiff::SignedDuration::from_secs(30)
 }
 
 impl Default for Config {
@@ -233,7 +232,7 @@ impl Default for Config {
                 metrics: Some(MetricsConfig {
                     opentelemetry: Some(OpentelemetryMetricsConfig {
                         otlp_endpoint: "http://127.0.0.1:4317".to_string(),
-                        push_interval: jiff::SignedDuration::from_secs(30).into(),
+                        push_interval: default_metrics_push_interval(),
                     }),
                 }),
             },

diff --git a/crates/core/src/newtype.rs b/crates/core/src/newtype.rs
@@ -17,35 +17,6 @@ use std::num::NonZeroUsize;
 use serde::Deserialize;
 use serde::Serialize;
 
-#[derive(Default, Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
-pub struct SignedDuration(jiff::SignedDuration);
-
-impl From<SignedDuration> for jiff::SignedDuration {
-    fn from(value: SignedDuration) -> Self {
-        value.0
-    }
-}
-
-impl From<jiff::SignedDuration> for SignedDuration {
-    fn from(value: jiff::SignedDuration) -> Self {
-        Self(value)
-    }
-}
-
-impl std::ops::Deref for SignedDuration {
-    type Target = jiff::SignedDuration;
-
-    fn deref(&self) -> &Self::Target {
-        &self.0
-    }
-}
-
-impl std::ops::DerefMut for SignedDuration {
-    fn deref_mut(&mut self) -> &mut Self::Target {
-        &mut self.0
-    }
-}
-
 /// Throttle config for the device.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(test, derive(schemars::JsonSchema))]
@@ -121,35 +92,3 @@ impl From<foyer::IopsCounter> for IopsCounter {
         }
     }
 }
-
-#[cfg(test)]
-mod json_schema {
-    use std::borrow::Cow;
-
-    use schemars::Schema;
-    use schemars::SchemaGenerator;
-    use schemars::json_schema;
-
-    use super::*;
-
-    impl schemars::JsonSchema for SignedDuration {
-        fn always_inline_schema() -> bool {
-            true
-        }
-
-        fn schema_name() -> Cow<'static, str> {
-            Cow::Borrowed("SignedDuration")
-        }
-
-        fn schema_id() -> Cow<'static, str> {
-            Cow::Borrowed("jiff::SignedDuration")
-        }
-
-        fn json_schema(_: &mut SchemaGenerator) -> Schema {
-            json_schema!({
-                "type": "string",
-                "format": "duration",
-            })
-        }
-    }
-}
diff --git a/crates/server/src/server.rs b/crates/server/src/server.rs
@@ -94,7 +94,7 @@ pub async fn make_acceptor_and_advertise_addr(
     listen_addr: &str,
     advertise_addr: Option<&str>,
 ) -> Result<(TcpAcceptor, SocketAddr), io::Error> {
-    log::info!("listening on {}", listen_addr);
+    log::info!("listening on {listen_addr}");
 
     let acceptor = TcpListener::bind(&listen_addr).into_acceptor().await?;
     let listen_addr = acceptor.local_addr()[0]

diff --git a/rust-toolchain.toml b/rust-toolchain.toml
@@ -13,5 +13,5 @@
 # limitations under the License.
 
 [toolchain]
-channel = "nightly-2025-04-14"
+channel = "nightly-2025-05-26"
 components = ["cargo", "rustfmt", "clippy", "rust-analyzer"]
diff --git a/xtask/src/main.rs b/xtask/src/main.rs
@@ -123,7 +123,7 @@ fn find_command(cmd: &str) -> StdCommand {
     } else {
         let stdout = String::from_utf8_lossy(&output.stdout);
         let stderr = String::from_utf8_lossy(&output.stderr);
-        panic!("{cmd} not found.\nstdout: {}\nstderr: {}", stdout, stderr);
+        panic!("{cmd} not found.\nstdout: {stdout}\nstderr: {stderr}");
     }
 }