[leader] drive guardian committee handoff (#568)

Author: 0xsiddharthks

crates/hashi-types/proto/sui/hashi/v1alpha/bridge_service.proto

@@ -25,6 +25,8 @@ service BridgeService {
   rpc SignWithdrawalTxSigning(SignWithdrawalTxSigningRequest) returns (SignWithdrawalTxSigningResponse);
   // Step 4: Sign committee approval to confirm a processed withdrawal on-chain.
   rpc SignWithdrawalConfirmation(SignWithdrawalConfirmationRequest) returns (SignWithdrawalConfirmationResponse);
+  // Sign a committee transition for the guardian's `UpdateCommittee`.
+  rpc SignCommitteeTransition(SignCommitteeTransitionRequest) returns (SignCommitteeTransitionResponse);
 }
 
 message GetServiceInfoRequest {}
@@ -186,3 +188,15 @@ message SignGuardianWithdrawalRequestRequest {
 message SignGuardianWithdrawalRequestResponse {
   MemberSignature member_signature = 1;
 }
+
+// Peers rebuild the transition from on-chain state; no committee bytes on the wire.
+message SignCommitteeTransitionRequest {
+  // Outgoing committee epoch. The new committee is the next entry in the
+  // on-chain `committees` map after `from_epoch` — committee epochs are
+  // sparse, so it is not generally `from_epoch + 1`.
+  uint64 from_epoch = 1;
+}
+
+message SignCommitteeTransitionResponse {
+  MemberSignature member_signature = 1;
+}

crates/hashi-types/src/proto/generated/sui.hashi.v1alpha.fds.bin

@@ -175,6 +175,20 @@ pub struct SignGuardianWithdrawalRequestResponse {
     #[prost(message, optional, tag = "1")]
     pub member_signature: ::core::option::Option<MemberSignature>,
 }
+/// Peers rebuild the transition from on-chain state; no committee bytes on the wire.
+#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct SignCommitteeTransitionRequest {
+    /// Outgoing committee epoch. The new committee is the next entry in the
+    /// on-chain `committees` map after `from_epoch` — committee epochs are
+    /// sparse, so it is not generally `from_epoch + 1`.
+    #[prost(uint64, tag = "1")]
+    pub from_epoch: u64,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct SignCommitteeTransitionResponse {
+    #[prost(message, optional, tag = "1")]
+    pub member_signature: ::core::option::Option<MemberSignature>,
+}
 /// Generated client implementations.
 pub mod bridge_service_client {
     #![allow(
@@ -505,6 +519,36 @@ pub mod bridge_service_client {
                 );
             self.inner.unary(req, path, codec).await
         }
+        /// Sign a committee transition for the guardian's `UpdateCommittee`.
+        pub async fn sign_committee_transition(
+            &mut self,
+            request: impl tonic::IntoRequest<super::SignCommitteeTransitionRequest>,
+        ) -> std::result::Result<
+            tonic::Response<super::SignCommitteeTransitionResponse>,
+            tonic::Status,
+        > {
+            self.inner
+                .ready()
+                .await
+                .map_err(|e| {
+                    tonic::Status::unknown(
+                        format!("Service was not ready: {}", e.into()),
+                    )
+                })?;
+            let codec = tonic_prost::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static(
+                "/sui.hashi.v1alpha.BridgeService/SignCommitteeTransition",
+            );
+            let mut req = request.into_request();
+            req.extensions_mut()
+                .insert(
+                    GrpcMethod::new(
+                        "sui.hashi.v1alpha.BridgeService",
+                        "SignCommitteeTransition",
+                    ),
+                );
+            self.inner.unary(req, path, codec).await
+        }
     }
 }
 /// Generated server implementations.
@@ -586,6 +630,14 @@ pub mod bridge_service_server {
             tonic::Response<super::SignWithdrawalConfirmationResponse>,
             tonic::Status,
         >;
+        /// Sign a committee transition for the guardian's `UpdateCommittee`.
+        async fn sign_committee_transition(
+            &self,
+            request: tonic::Request<super::SignCommitteeTransitionRequest>,
+        ) -> std::result::Result<
+            tonic::Response<super::SignCommitteeTransitionResponse>,
+            tonic::Status,
+        >;
     }
     #[derive(Debug)]
     pub struct BridgeServiceServer<T> {
@@ -1075,6 +1127,57 @@ pub mod bridge_service_server {
                     };
                     Box::pin(fut)
                 }
+                "/sui.hashi.v1alpha.BridgeService/SignCommitteeTransition" => {
+                    #[allow(non_camel_case_types)]
+                    struct SignCommitteeTransitionSvc<T: BridgeService>(pub Arc<T>);
+                    impl<
+                        T: BridgeService,
+                    > tonic::server::UnaryService<super::SignCommitteeTransitionRequest>
+                    for SignCommitteeTransitionSvc<T> {
+                        type Response = super::SignCommitteeTransitionResponse;
+                        type Future = BoxFuture<
+                            tonic::Response<Self::Response>,
+                            tonic::Status,
+                        >;
+                        fn call(
+                            &mut self,
+                            request: tonic::Request<
+                                super::SignCommitteeTransitionRequest,
+                            >,
+                        ) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move {
+                                <T as BridgeService>::sign_committee_transition(
+                                        &inner,
+                                        request,
+                                    )
+                                    .await
+                            };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = SignCommitteeTransitionSvc(inner);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(
+                                accept_compression_encodings,
+                                send_compression_encodings,
+                            )
+                            .apply_max_message_size_config(
+                                max_decoding_message_size,
+                                max_encoding_message_size,
+                            );
+                        let res = grpc.unary(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
                 _ => {
                     Box::pin(async move {
                         let mut response = http::Response::new(

crates/hashi-types/src/proto/generated/sui.hashi.v1alpha.rs

@@ -17,6 +17,8 @@ use crate::withdrawals::WithdrawalTxSigning;
 use hashi_types::bitcoin_txid::BitcoinTxid;
 use hashi_types::proto::GetServiceInfoRequest;
 use hashi_types::proto::GetServiceInfoResponse;
+use hashi_types::proto::SignCommitteeTransitionRequest;
+use hashi_types::proto::SignCommitteeTransitionResponse;
 use hashi_types::proto::SignDepositConfirmationRequest;
 use hashi_types::proto::SignDepositConfirmationResponse;
 use hashi_types::proto::SignGuardianWithdrawalRequestRequest;
@@ -133,6 +135,30 @@ impl BridgeService for HttpService {
         }))
     }
 
+    /// Validate and BLS-sign a `CommitteeTransition` for the guardian.
+    #[tracing::instrument(
+        level = "info",
+        skip_all,
+        fields(from_epoch = tracing::field::Empty, caller = tracing::field::Empty),
+    )]
+    async fn sign_committee_transition(
+        &self,
+        request: Request<SignCommitteeTransitionRequest>,
+    ) -> Result<Response<SignCommitteeTransitionResponse>, Status> {
+        let caller = authenticate_caller(&request)?;
+        tracing::Span::current().record("caller", tracing::field::display(&caller));
+        let from_epoch = request.get_ref().from_epoch;
+        tracing::Span::current().record("from_epoch", from_epoch);
+        let member_signature = self
+            .inner
+            .validate_and_sign_committee_transition(from_epoch)
+            .map_err(|e| Status::failed_precondition(e.to_string()))?;
+        tracing::info!(from_epoch, "Signed committee transition");
+        Ok(Response::new(SignCommitteeTransitionResponse {
+            member_signature: Some(member_signature),
+        }))
+    }
+
     /// Validate and BLS-sign a `StandardWithdrawalRequest` for the guardian.
     #[tracing::instrument(
         level = "info",

crates/hashi/src/grpc/bridge_service.rs

@@ -1,19 +1,44 @@
 // Copyright (c) Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
+use std::sync::Arc;
 use std::time::Duration;
 
-use hashi_types::proto::guardian_service_client::GuardianServiceClient;
+use axum::http;
+use sui_http::middleware::callback::CallbackLayer;
+use tonic::body::Body;
 use tonic::transport::Channel;
 use tonic::transport::Endpoint;
+use tower::ServiceBuilder;
+use tower::util::BoxCloneService;
+
+use crate::grpc::metrics_layer::RpcMetricsMakeCallbackHandler;
+use crate::metrics::Metrics;
+use hashi_types::proto::guardian_service_client::GuardianServiceClient;
 
 type BoxError = Box<dyn std::error::Error + Send + Sync + 'static>;
 
+/// Boxed transport handed to the tonic-generated `GuardianServiceClient`.
+/// Same shape as `crate::grpc::Client::BoxedChannel`, so the metrics
+/// callback layer wraps validator-validator and validator-guardian RPCs
+/// identically.
+pub type BoxedChannel = BoxCloneService<http::Request<Body>, http::Response<Body>, tonic::Status>;
+
 /// Lazy gRPC channel to a `hashi-guardian`.
-#[derive(Clone, Debug)]
+#[derive(Clone)]
 pub struct GuardianClient {
     endpoint: String,
     channel: Channel,
+    metrics: Option<Arc<Metrics>>,
+}
+
+impl std::fmt::Debug for GuardianClient {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("GuardianClient")
+            .field("endpoint", &self.endpoint)
+            .field("metrics_enabled", &self.metrics.is_some())
+            .finish()
+    }
 }
 
 impl GuardianClient {
@@ -27,15 +52,53 @@ impl GuardianClient {
         Ok(Self {
             endpoint: endpoint.to_string(),
             channel,
+            metrics: None,
         })
     }
 
+    /// Attach the metrics registry so outbound guardian RPCs are observed
+    /// by [`RpcMetricsMakeCallbackHandler`] via `sui_http`'s callback
+    /// layer. Without this, the client emits no RPC traffic metrics.
+    pub fn with_metrics(mut self, metrics: Arc<Metrics>) -> Self {
+        self.metrics = Some(metrics);
+        self
+    }
+
     pub fn endpoint(&self) -> &str {
         &self.endpoint
     }
 
-    pub fn guardian_service_client(&self) -> GuardianServiceClient<Channel> {
-        GuardianServiceClient::new(self.channel.clone())
+    /// Build a boxed transport, applying the metrics callback layer when
+    /// a registry is configured. Mirrors `crate::grpc::client::Client::boxed_channel`
+    /// so guardian RPCs surface under the same `hashi_requests_total` /
+    /// `hashi_request_latency_seconds` metrics as validator-validator
+    /// traffic.
+    fn boxed_channel(&self) -> BoxedChannel {
+        let channel = self.channel.clone();
+        match &self.metrics {
+            Some(metrics) => {
+                let svc = ServiceBuilder::new()
+                    .map_err(tonic::Status::from_error)
+                    .map_response(|resp: http::Response<_>| resp.map(Body::new))
+                    .layer(CallbackLayer::new(RpcMetricsMakeCallbackHandler::client(
+                        metrics.clone(),
+                    )))
+                    .map_request(|req: http::Request<_>| req.map(Body::new))
+                    .map_err(|e: tonic::transport::Error| -> BoxError { Box::new(e) })
+                    .service(channel);
+                BoxCloneService::new(svc)
+            }
+            None => {
+                let svc = ServiceBuilder::new()
+                    .map_err(|e: tonic::transport::Error| tonic::Status::from_error(Box::new(e)))
+                    .service(channel);
+                BoxCloneService::new(svc)
+            }
+        }
+    }
+
+    pub fn guardian_service_client(&self) -> GuardianServiceClient<BoxedChannel> {
+        GuardianServiceClient::new(self.boxed_channel())
     }
 
     pub async fn get_guardian_info(
@@ -58,4 +121,15 @@ impl GuardianClient {
             .await?;
         Ok(response.into_inner())
     }
+
+    pub async fn update_committee(
+        &self,
+        request: hashi_types::proto::SignedCommitteeTransition,
+    ) -> Result<hashi_types::proto::UpdateCommitteeResponse, tonic::Status> {
+        let response = self
+            .guardian_service_client()
+            .update_committee(request)
+            .await?;
+        Ok(response.into_inner())
+    }
 }