release 1.11.1 merge to main

Cargo.toml

@@ -12,7 +12,7 @@ exclude = ["infra/scripts/generate_keys"]
 resolver = "2"
 
 [workspace.package]
-version = "1.11.0"
+version = "1.11.1"
 
 [workspace.dependencies]
 alloy = { version = "=1.0.38", features = ["contract", "json"] }

chain-signatures/node/src/lib.rs

@@ -1,3 +1,5 @@
+pub const PROTOCOL_VERSION: u64 = 1;
+
 pub mod backlog;
 pub mod checkpoint_consensus;
 pub mod cli;

chain-signatures/node/src/mesh/connection.rs

@@ -106,7 +106,7 @@ impl NodeConnection {
                     });
                 }
                 _ = interval.tick() => {
-                    let status = match client.status(&url).await {
+                    let resp = match client.status(&url).await {
                         Ok(status) => status,
                         Err(err) => {
                             tracing::warn!(?node, ?err, "checking /status failed");
@@ -117,10 +117,21 @@ impl NodeConnection {
                         }
                     };
 
-                    // note: borrowing and sending later on `status_tx` can potentially deadlock,
-                    // but since we are copying the status, this is not the case. Change this carefully.
+                    if resp.protocol_version != crate::PROTOCOL_VERSION {
+                        tracing::warn!(
+                            ?node,
+                            our_version = crate::PROTOCOL_VERSION,
+                            peer_version = resp.protocol_version,
+                            "protocol version mismatch"
+                        );
+                        status_tx.send_if_modified(|(status, _)| {
+                            std::mem::replace(status, NodeStatus::Offline) != NodeStatus::Offline
+                        });
+                        continue;
+                    }
+
                     let old_status = status_tx.borrow().0;
-                    let mut new_status = match status {
+                    let mut new_status = match resp.status {
                         OtherNodeStatus::Running { .. } => NodeStatus::Active,
                         OtherNodeStatus::Resharing { .. }
                         | OtherNodeStatus::Generating { .. }

chain-signatures/node/src/mesh/mod.rs

@@ -202,6 +202,23 @@ mod tests {
 
     const PING_INTERVAL: Duration = Duration::from_millis(10);
 
+    async fn expect_status(
+        watcher: &mut connection::ConnectionWatcher,
+        participant: Participant,
+        expected: NodeStatus,
+    ) {
+        for _ in 0..20 {
+            if let Ok((p, status, _info)) =
+                tokio::time::timeout(Duration::from_millis(500), watcher.next()).await
+            {
+                if p == participant && status == expected {
+                    return;
+                }
+            }
+        }
+        panic!("timed out waiting for {participant:?} to become {expected:?}");
+    }
+
     #[test(tokio::test)]
     async fn test_pool_update() {
         let num_nodes = 3;
@@ -440,4 +457,32 @@ mod tests {
 
         mesh_task.abort();
     }
+
+    #[test(tokio::test)]
+    async fn test_protocol_version_mismatch_marks_offline() {
+        let mut servers = MockServers::run(2).await;
+        let participants = servers.participants();
+        let my_id = servers[0].account_id().clone();
+
+        let mut pool = Pool::new(&servers.client(), &my_id, PING_INTERVAL);
+        let mut watcher = pool.watch();
+        pool.connect_nodes(&participants, &mut HashSet::new()).await;
+
+        let remote_id = servers[1].id();
+
+        expect_status(&mut watcher, remote_id, NodeStatus::Syncing).await;
+        pool.report_node_synced(remote_id).await;
+        expect_status(&mut watcher, remote_id, NodeStatus::Active).await;
+
+        servers[1].set_protocol_version(None).await;
+        expect_status(&mut watcher, remote_id, NodeStatus::Offline).await;
+
+        servers[1].make_online().await;
+        expect_status(&mut watcher, remote_id, NodeStatus::Syncing).await;
+        pool.report_node_synced(remote_id).await;
+        expect_status(&mut watcher, remote_id, NodeStatus::Active).await;
+
+        servers[1].set_protocol_version(Some(0)).await;
+        expect_status(&mut watcher, remote_id, NodeStatus::Offline).await;
+    }
 }

chain-signatures/node/src/node_client.rs

@@ -1,9 +1,8 @@
 use crate::backlog::Checkpoint;
 use crate::protocol::message::cbor_to_bytes;
-use crate::protocol::state::NodeStatus;
 use crate::protocol::sync::SyncUpdate;
 use crate::protocol::Chain;
-use crate::web::StateView;
+use crate::web::{StateView, StatusResponse};
 
 use hyper::StatusCode;
 use mpc_keys::hpke::Ciphered;
@@ -152,7 +151,7 @@ impl NodeClient {
         Ok(resp.json::<StateView>().await?)
     }
 
-    pub async fn status(&self, base: impl IntoUrl) -> Result<NodeStatus, RequestError> {
+    pub async fn status(&self, base: impl IntoUrl) -> Result<StatusResponse, RequestError> {
         let mut url = base.into_url()?;
         url.set_path("status");
 

chain-signatures/node/src/protocol/sync/mod.rs

@@ -262,10 +262,18 @@ impl SyncUpdate {
     async fn process(self, triples: TripleStorage, presignatures: PresignatureStorage) {
         let start = Instant::now();
 
-        let outdated_triples = triples.remove_outdated(self.from, &self.triples).await;
-        let outdated_presignatures = presignatures
-            .remove_outdated(self.from, &self.presignatures)
-            .await;
+        let outdated_triples = if !self.triples.is_empty() {
+            triples.remove_outdated(self.from, &self.triples).await
+        } else {
+            Vec::new()
+        };
+        let outdated_presignatures = if !self.presignatures.is_empty() {
+            presignatures
+                .remove_outdated(self.from, &self.presignatures)
+                .await
+        } else {
+            Vec::new()
+        };
 
         if !outdated_triples.is_empty() || !outdated_presignatures.is_empty() {
             tracing::info!(

chain-signatures/node/src/web/mock.rs

@@ -5,6 +5,8 @@ use near_sdk::AccountId;
 use crate::{
     node_client::NodeClient,
     protocol::{contract::primitives::Participants, state::NodeStatus, ParticipantInfo},
+    web::StatusResponse,
+    PROTOCOL_VERSION,
 };
 
 use super::StateView;
@@ -77,11 +79,15 @@ impl MockServer {
     }
 
     pub async fn make_online(&mut self) {
+        self.set_protocol_version(Some(PROTOCOL_VERSION)).await;
+    }
+
+    pub async fn set_protocol_version(&mut self, protocol_version: Option<u64>) {
         self.server
             .mock("GET", "/status")
             .with_status(201)
             .with_header("content-type", "application/json")
-            .with_body(default_status_body(self.id))
+            .with_body(status_body(self.id, protocol_version))
             .create_async()
             .await;
     }
@@ -166,11 +172,23 @@ fn default_state_body() -> Vec<u8> {
 }
 
 fn default_status_body(id: u32) -> Vec<u8> {
-    serde_json::to_vec(&NodeStatus::Running {
+    status_body(id, Some(PROTOCOL_VERSION))
+}
+
+fn status_body(id: u32, version: Option<u64>) -> Vec<u8> {
+    let status = NodeStatus::Running {
         me: Participant::from(id),
         participants: vec![Participant::from(0)],
         ongoing_triple_gen: 0,
         ongoing_presignature_gen: 0,
-    })
-    .unwrap()
+    };
+
+    match version {
+        Some(protocol_version) => serde_json::to_vec(&StatusResponse {
+            protocol_version,
+            status: status.clone(),
+        })
+        .unwrap(),
+        None => serde_json::to_vec(&status).unwrap(),
+    }
 }

chain-signatures/node/src/web/mod.rs

@@ -214,9 +214,19 @@ async fn state(Extension(web): Extension<Arc<AxumState>>) -> Result<Json<StateVi
     result
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct StatusResponse {
+    pub status: NodeStatus,
+    #[serde(default)]
+    pub protocol_version: u64,
+}
+
 #[tracing::instrument(level = "debug", skip_all)]
-async fn status(Extension(web): Extension<Arc<AxumState>>) -> Json<NodeStatus> {
-    Json(web.node.status())
+async fn status(Extension(web): Extension<Arc<AxumState>>) -> Json<StatusResponse> {
+    Json(StatusResponse {
+        status: web.node.status(),
+        protocol_version: crate::PROTOCOL_VERSION,
+    })
 }
 
 #[tracing::instrument(level = "debug", skip_all)]