fix: operator now releases graph lock if execution fails & use simplified graph to reduce message size

node/src/action.rs

@@ -2,12 +2,13 @@ use crate::env;
 use crate::middleware::AllBehaviours;
 use crate::relayer_action::do_tick_action;
 use crate::utils::{statics::*, *};
+use crate::{defer, dismiss_defer};
 use anyhow::Result;
 use bitcoin::PublicKey;
 use bitcoin::{Amount, Network, Txid};
 use bitvm2_lib::actors::Actor;
 use bitvm2_lib::keys::*;
-use bitvm2_lib::types::{Bitvm2Graph, Bitvm2Parameters, CustomInputs};
+use bitvm2_lib::types::{Bitvm2Graph, Bitvm2Parameters, CustomInputs, SimplifiedBitvm2Graph};
 use bitvm2_lib::verifier::export_challenge_tx;
 use bitvm2_lib::{committee::*, operator::*, verifier::*};
 use client::client::BitVM2Client;
@@ -70,7 +71,7 @@ pub struct CreateGraphPrepare {
 pub struct CreateGraph {
     pub instance_id: Uuid,
     pub graph_id: Uuid,
-    pub graph: Bitvm2Graph,
+    pub graph: SimplifiedBitvm2Graph,
 }
 
 #[derive(Serialize, Deserialize)]
@@ -96,7 +97,7 @@ pub struct CommitteePresign {
 pub struct GraphFinalize {
     pub instance_id: Uuid,
     pub graph_id: Uuid,
-    pub graph: Bitvm2Graph,
+    pub graph: SimplifiedBitvm2Graph,
     pub graph_ipfs_cid: String,
 }
 
@@ -265,6 +266,9 @@ pub async fn recv_and_dispatch(
             {
                 let graph_id = Uuid::new_v4();
                 if try_start_new_graph(receive_data.instance_id, graph_id) {
+                    defer!(on_err, {
+                        force_stop_current_graph();
+                    });
                     tracing::info!("generating new graph: {graph_id}");
                     let master_key = OperatorMasterKey::new(env::get_bitvm_key()?);
                     let keypair = master_key.keypair_for_graph(graph_id);
@@ -306,23 +310,25 @@ pub async fn recv_and_dispatch(
                     let message_content = GOATMessageContent::CreateGraph(CreateGraph {
                         instance_id: receive_data.instance_id,
                         graph_id,
-                        graph,
+                        graph: graph.to_simplified(),
                     });
                     // TODO: compress huge message
                     send_to_peer(
                         swarm,
                         GOATMessage::from_typed(Actor::Committee, &message_content)?,
                     )?;
+                    dismiss_defer!(on_err);
                 };
             };
         }
         (GOATMessageContent::CreateGraph(receive_data), Actor::Committee) => {
             tracing::info!("Handle CreateGraph");
+            let graph = Bitvm2Graph::from_simplified(receive_data.graph)?;
             store_graph(
                 client,
                 receive_data.instance_id,
                 receive_data.graph_id,
-                &receive_data.graph,
+                &graph,
                 Some(GraphStatus::OperatorPresigned.to_string()),
             )
             .await?;
@@ -340,7 +346,7 @@ pub async fn recv_and_dispatch(
                 pub_nonces.clone(),
             )
             .await?;
-            let committee_members_num = receive_data.graph.parameters.committee_pubkeys.len();
+            let committee_members_num = graph.parameters.committee_pubkeys.len();
             let message_content = GOATMessageContent::NonceGeneration(NonceGeneration {
                 instance_id: receive_data.instance_id,
                 graph_id: receive_data.graph_id,
@@ -428,6 +434,9 @@ pub async fn recv_and_dispatch(
             if Some((receive_data.instance_id, receive_data.graph_id))
                 == statics::current_processing_graph()
             {
+                defer!(on_err, {
+                    force_stop_current_graph();
+                });
                 store_committee_partial_sigs(
                     client,
                     receive_data.instance_id,
@@ -495,22 +504,24 @@ pub async fn recv_and_dispatch(
                     let message_content = GOATMessageContent::GraphFinalize(GraphFinalize {
                         instance_id: receive_data.instance_id,
                         graph_id: receive_data.graph_id,
-                        graph,
+                        graph: graph.to_simplified(),
                         graph_ipfs_cid,
                     });
                     send_to_peer(swarm, GOATMessage::from_typed(Actor::All, &message_content)?)?;
                     force_stop_current_graph();
-                }
+                };
+                dismiss_defer!(on_err);
             };
         }
         (GOATMessageContent::GraphFinalize(receive_data), _) => {
             tracing::info!("Handle GraphFinalize");
             // TODO: validate graph & ipfs
+            let graph = Bitvm2Graph::from_simplified(receive_data.graph)?;
             store_graph(
                 client,
                 receive_data.instance_id,
                 receive_data.graph_id,
-                &receive_data.graph,
+                &graph,
                 Some(GraphStatus::CommitteePresigned.to_string()),
             )
             .await?;

node/src/middleware/behaviour.rs

@@ -23,7 +23,7 @@ impl AllBehaviours {
             .unwrap();
 
         let gossipsub_config = gossipsub::ConfigBuilder::default()
-            .max_transmit_size(12582912) // 12 MB
+            .max_transmit_size(4194304) // 4 MB
             .build()
             .map_err(io::Error::other)
             .unwrap();

node/src/utils.rs

@@ -494,7 +494,7 @@ pub fn node_sign(
     sighash_type: EcdsaSighashType,
     node_keypair: &Keypair,
 ) -> Result<(), Box<dyn std::error::Error>> {
-    let node_pubkey = get_node_pubkey()?;
+    let node_pubkey: PublicKey = node_keypair.public_key().into();
     populate_p2wsh_witness(
         tx,
         input_index,
@@ -974,6 +974,62 @@ pub async fn publish_graph_to_ipfs(
     Ok(dir_cid)
 }
 
+pub mod defer {
+    pub struct Defer<F: FnOnce()> {
+        cleanup: Option<F>,
+    }
+    impl<F: FnOnce()> Defer<F> {
+        pub fn new(f: F) -> Self {
+            Self { cleanup: Some(f) }
+        }
+        pub fn dismiss(&mut self) {
+            self.cleanup = None;
+        }
+    }
+    impl<F: FnOnce()> Drop for Defer<F> {
+        fn drop(&mut self) {
+            if let Some(cleanup) = self.cleanup.take() {
+                cleanup();
+            }
+        }
+    }
+    #[macro_export]
+    macro_rules! defer {
+        ($name:ident, $cleanup:block) => {
+            let mut $name = $crate::utils::defer::Defer::new(|| $cleanup);
+        };
+    }
+    #[macro_export]
+    macro_rules! dismiss_defer {
+        ($name:ident) => {
+            $name.dismiss();
+        };
+    }
+
+    #[test]
+    fn test_defer() {
+        use super::statics::*;
+        use uuid::Uuid;
+        fn inner_func(should_success: bool) -> Result<(), Box<dyn std::error::Error>> {
+            if should_success { Ok(()) } else { Err("inner functions not success".into()) }
+        }
+        fn guarded_operation(should_success: bool) -> Result<(), Box<dyn std::error::Error>> {
+            defer!(on_err, {
+                force_stop_current_graph();
+            });
+            inner_func(should_success)?;
+            dismiss_defer!(on_err);
+            Ok(())
+        }
+        try_start_new_graph(Uuid::new_v4(), Uuid::new_v4());
+        assert!(is_processing_graph());
+        let _ = guarded_operation(true);
+        assert!(is_processing_graph());
+        let _ = guarded_operation(false);
+        assert!(!is_processing_graph());
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use crate::action::{CreateInstance, GOATMessageContent, KickoffReady};