[WIP] Refactor gossip protocol

src/ckb/actor.rs

@@ -1,5 +1,5 @@
 use ckb_sdk::{rpc::ResponseFormatGetter, CkbRpcClient, RpcError};
-use ckb_types::{core::TransactionView, packed, prelude::*};
+use ckb_types::{core::TransactionView, packed, prelude::*, H256};
 use ractor::{
     concurrency::{sleep, Duration},
     Actor, ActorProcessingErr, ActorRef, RpcReplyPort,
@@ -25,19 +25,6 @@ pub struct TraceTxRequest {
     pub confirmations: u64,
 }
 
-#[derive(Debug)]
-pub enum CkbChainMessage {
-    Fund(
-        FundingTx,
-        FundingRequest,
-        RpcReplyPort<Result<FundingTx, FundingError>>,
-    ),
-    Sign(FundingTx, RpcReplyPort<Result<FundingTx, FundingError>>),
-    SendTx(TransactionView, RpcReplyPort<Result<(), RpcError>>),
-    TraceTx(TraceTxRequest, RpcReplyPort<TraceTxResponse>),
-    GetCurrentBlockNumber((), RpcReplyPort<Result<u64, RpcError>>),
-}
-
 #[derive(Debug)]
 pub struct TraceTxResponse {
     pub tx: Option<ckb_jsonrpc_types::TransactionView>,
@@ -53,6 +40,39 @@ impl TraceTxResponse {
     }
 }
 
+#[derive(Debug, Clone)]
+pub struct GetBlockTimestampRequest {
+    block_hash: H256,
+}
+
+impl GetBlockTimestampRequest {
+    pub fn from_block_hash(block_hash: H256) -> Self {
+        Self { block_hash }
+    }
+
+    pub fn block_hash(&self) -> H256 {
+        self.block_hash.clone()
+    }
+}
+
+pub type GetBlockTimestampResponse = u64;
+
+#[derive(Debug)]
+pub enum CkbChainMessage {
+    Fund(
+        FundingTx,
+        FundingRequest,
+        RpcReplyPort<Result<FundingTx, FundingError>>,
+    ),
+    Sign(FundingTx, RpcReplyPort<Result<FundingTx, FundingError>>),
+    SendTx(TransactionView, RpcReplyPort<Result<(), RpcError>>),
+    TraceTx(TraceTxRequest, RpcReplyPort<TraceTxResponse>),
+    GetBlockTimestamp(
+        GetBlockTimestampRequest,
+        RpcReplyPort<Result<Option<GetBlockTimestampResponse>, RpcError>>,
+    ),
+}
+
 #[ractor::async_trait]
 impl Actor for CkbChainActor {
     type Msg = CkbChainMessage;
@@ -89,17 +109,8 @@ impl Actor for CkbChainActor {
         message: Self::Msg,
         state: &mut Self::State,
     ) -> Result<(), ActorProcessingErr> {
-        use CkbChainMessage::{Fund, GetCurrentBlockNumber, SendTx, Sign, TraceTx};
+        use CkbChainMessage::{Fund, SendTx, Sign, TraceTx};
         match message {
-            GetCurrentBlockNumber(_, reply) => {
-                // Have to use block_in_place here, see https://github.com/seanmonstar/reqwest/issues/1017.
-                let result = tokio::task::block_in_place(move || {
-                    CkbRpcClient::new(&state.config.rpc_url)
-                        .get_tip_block_number()
-                        .map(|x| x.value())
-                });
-                let _ = reply.send(result);
-            }
             Fund(tx, request, reply_port) => {
                 let context = state.build_funding_context(&request);
                 if !reply_port.is_closed() {
@@ -251,6 +262,20 @@ impl Actor for CkbChainActor {
                     }
                 }
             }
+            CkbChainMessage::GetBlockTimestamp(
+                GetBlockTimestampRequest { block_hash },
+                reply_port,
+            ) => {
+                let rpc_url = state.config.rpc_url.clone();
+                tokio::task::block_in_place(move || {
+                    let ckb_client = CkbRpcClient::new(&rpc_url);
+                    let _ = reply_port.send(
+                        ckb_client
+                            .get_block(block_hash)
+                            .map(|x| x.map(|x| x.header.inner.timestamp.into())),
+                    );
+                });
+            }
         }
         Ok(())
     }

src/ckb/config.rs

@@ -148,7 +148,7 @@ serde_with::serde_conv!(
 );
 
 #[serde_as]
-#[derive(Serialize, Deserialize, Debug, Clone, Default, PartialEq)]
+#[derive(Serialize, Deserialize, Debug, Clone, Default, Eq, PartialEq, Hash)]
 pub struct UdtScript {
     pub code_hash: H256,
     #[serde_as(as = "ScriptHashTypeWrapper")]
@@ -158,23 +158,23 @@ pub struct UdtScript {
 }
 
 #[serde_as]
-#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
+#[derive(Serialize, Deserialize, Clone, Debug, Eq, PartialEq, Hash)]
 pub struct UdtCellDep {
     #[serde_as(as = "DepTypeWrapper")]
     pub dep_type: DepType,
     pub tx_hash: H256,
     pub index: u32,
 }
 
-#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq)]
+#[derive(Serialize, Deserialize, Clone, Debug, Default, Eq, PartialEq, Hash)]
 pub struct UdtArgInfo {
     pub name: String,
     pub script: UdtScript,
     pub auto_accept_amount: Option<u128>,
     pub cell_deps: Vec<UdtCellDep>,
 }
 
-#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq)]
+#[derive(Serialize, Deserialize, Clone, Debug, Default, Eq, PartialEq, Hash)]
 pub struct UdtCfgInfos(pub Vec<UdtArgInfo>);
 
 impl FromStr for UdtCfgInfos {

src/ckb/mod.rs

@@ -2,7 +2,10 @@ mod actor;
 mod error;
 mod funding;
 
-pub use actor::{CkbChainActor, CkbChainMessage, TraceTxRequest, TraceTxResponse};
+pub use actor::{
+    CkbChainActor, CkbChainMessage, GetBlockTimestampRequest, GetBlockTimestampResponse,
+    TraceTxRequest, TraceTxResponse,
+};
 pub use config::{CkbConfig, DEFAULT_CKB_BASE_DIR_NAME};
 pub use error::{CkbChainError, FundingError};
 pub use funding::{FundingRequest, FundingTx};

src/ckb/tests/test_utils.rs

@@ -6,14 +6,19 @@ use ckb_types::{
     core::{DepType, TransactionView},
     packed::{CellDep, CellOutput, OutPoint, Script, Transaction},
     prelude::{Builder, Entity, IntoTransactionView, Pack, PackVec, Unpack},
+    H256,
 };
-use once_cell::sync::Lazy;
+use once_cell::sync::{Lazy, OnceCell};
 use std::{collections::HashMap, sync::Arc, sync::RwLock};
-
-use crate::ckb::{
-    config::UdtCfgInfos,
-    contracts::{Contract, ContractsContext, ContractsInfo},
-    TraceTxRequest, TraceTxResponse,
+use tokio::sync::RwLock as TokioRwLock;
+
+use crate::{
+    ckb::{
+        config::UdtCfgInfos,
+        contracts::{Contract, ContractsContext, ContractsInfo},
+        TraceTxRequest, TraceTxResponse,
+    },
+    now_timestamp_as_millis_u64,
 };
 
 use crate::ckb::CkbChainMessage;
@@ -308,9 +313,6 @@ impl Actor for MockChainActor {
         debug!("MockChainActor received message: {:?}", message);
         use CkbChainMessage::*;
         match message {
-            GetCurrentBlockNumber(_, reply) => {
-                let _ = reply.send(Ok(0));
-            }
             Fund(tx, request, reply_port) => {
                 let mut fulfilled_tx = tx.clone();
                 let outputs = fulfilled_tx
@@ -490,6 +492,33 @@ impl Actor for MockChainActor {
                     }
                 };
             }
+            GetBlockTimestamp(request, rpc_reply_port) => {
+                // The problem of channel announcement is that each nodes will query the block timestamp
+                // and use it as the channel announcement timestamp.
+                // Guaranteeing the block timestamp is the same across all nodes is important
+                // because if a node A has a greater channel announcement timestamp than node B, then when
+                // A tries to get broadcast messages after this channel announcement timestamp, B will return
+                // the channel announcement. But for A, it is not a later broadcast message. This process will
+                // cause an infinite loop.
+                // So here we create an static lock which is shared across all nodes, and we use this lock to
+                // guarantee that the block timestamp is the same across all nodes.
+                static BLOCK_TIMESTAMP: OnceCell<TokioRwLock<HashMap<H256, u64>>> = OnceCell::new();
+                BLOCK_TIMESTAMP.get_or_init(|| TokioRwLock::new(HashMap::new()));
+                let timestamp = *BLOCK_TIMESTAMP
+                    .get()
+                    .unwrap()
+                    .write()
+                    .await
+                    .entry(request.block_hash())
+                    .or_insert(now_timestamp_as_millis_u64());
+
+                debug!(
+                    "Get block timestamp: block_hash: {:?}, timestamp: {}",
+                    request.block_hash(),
+                    timestamp
+                );
+                let _ = rpc_reply_port.send(Ok(Some(timestamp)));
+            }
         }
         Ok(())
     }

src/errors.rs

@@ -1,3 +1,4 @@
+use ckb_sdk::RpcError;
 use ractor::{MessagingErr, SpawnErr};
 use tentacle::{error::SendErrorKind, secio::PeerId};
 use thiserror::Error;
@@ -48,8 +49,14 @@ pub enum Error {
     InvalidPeerMessage(String),
     #[error("Onion packet error: {0}")]
     InvalidOnionPacket(crate::fiber::types::Error),
+    #[error("Ckb Rpc error: {0}")]
+    CkbRpcError(RpcError),
     #[error("Database error: {0}")]
     DBInternalError(String),
+    #[error("Internal error: {0}")]
+    InternalError(anyhow::Error),
+    #[error("Invalid chain hash: {0} (expecting {1})")]
+    InvalidChainHash(Hash256, Hash256),
 }
 
 pub type Result<T> = std::result::Result<T, Error>;

src/fiber/channel.rs

@@ -7,8 +7,10 @@ use crate::{
     fiber::{
         fee::calculate_tlc_forward_fee,
         network::{get_chain_hash, SendOnionPacketCommand},
-        serde_utils::PubNonceAsBytes,
-        types::{ChannelUpdate, PeeledPaymentOnionPacket, TlcErr, TlcErrPacket, TlcErrorCode},
+        types::{
+            BroadcastMessage, ChannelUpdate, PeeledPaymentOnionPacket, TlcErr, TlcErrPacket,
+            TlcErrorCode,
+        },
     },
     invoice::{CkbInvoice, CkbInvoiceStatus, InvoiceStore},
     now_timestamp_as_millis_u64,
@@ -70,7 +72,7 @@ use super::{
     hash_algorithm::HashAlgorithm,
     key::blake2b_hash_with_salt,
     network::FiberMessageWithPeerId,
-    serde_utils::EntityHex,
+    serde_utils::{EntityHex, PubNonceAsBytes},
     types::{
         AcceptChannel, AddTlc, ChannelAnnouncement, ChannelReady, ClosingSigned, CommitmentSigned,
         EcdsaSignature, FiberChannelMessage, FiberMessage, Hash256, OpenChannel,
@@ -90,6 +92,14 @@ pub const FUNDING_CELL_WITNESS_LEN: usize = 16 + 32 + 64;
 // is funded or not.
 pub const INITIAL_COMMITMENT_NUMBER: u64 = 0;
 
+// Whether we are receiving a channel update from node1 or node2.
+// If the flag is set, it means the channel update is from node2, otherwise it is from node1.
+pub const MESSAGE_OF_NODE1_FLAG: u32 = 0;
+
+// Whether we are receiving a channel update from node1 or node2.
+// If the flag is set, it means the channel update is from node2, otherwise it is from node1.
+pub const MESSAGE_OF_NODE2_FLAG: u32 = 1;
+
 // The channel is disabled, and no more tlcs can be added to the channel.
 pub const CHANNEL_DISABLED_FLAG: u32 = 1;
 
@@ -1630,10 +1640,9 @@ where
 
                 self.network
                     .send_message(NetworkActorMessage::new_command(
-                        NetworkActorCommand::ProccessChannelUpdate(
-                            self.get_remote_peer_id(),
-                            update,
-                        ),
+                        NetworkActorCommand::BroadcastMessages(vec![
+                            BroadcastMessage::ChannelUpdate(update),
+                        ]),
                     ))
                     .expect(ASSUME_NETWORK_ACTOR_ALIVE);
 
@@ -3272,6 +3281,11 @@ impl ChannelActorState {
             .get_unsigned_channel_update_message()
             .expect("public channel can generate channel update message");
         f(&mut channel_update);
+        debug!(
+            "Generated channel update message for channel {:?}: {:?}",
+            &self.get_id(),
+            &channel_update
+        );
         let node_signature =
             sign_network_message(network.clone(), channel_update.message_to_sign())
                 .await
@@ -3312,10 +3326,9 @@ impl ChannelActorState {
 
         network
             .send_message(NetworkActorMessage::new_command(
-                NetworkActorCommand::ProccessChannelUpdate(
-                    self.get_remote_peer_id(),
+                NetworkActorCommand::BroadcastMessages(vec![BroadcastMessage::ChannelUpdate(
                     channel_update,
-                ),
+                )]),
             ))
             .expect(ASSUME_NETWORK_ACTOR_ALIVE);
     }
@@ -3360,7 +3373,7 @@ impl ChannelActorState {
                 ) => Some(ChannelUpdate::new_unsigned(
                     Default::default(),
                     self.must_get_funding_transaction_outpoint(),
-                    std::time::UNIX_EPOCH.elapsed().expect("Duration since unix epoch").as_secs(),
+                    now_timestamp_as_millis_u64(),
                     message_flags,
                     0,
                     expiry_delta,
@@ -5241,30 +5254,15 @@ impl ChannelActorState {
             self.on_channel_ready(network).await;
 
             debug!(
-                "Broadcasting channel announcement message {:?}",
-                &channel_announcement,
+                "Broadcasting channel announcement {:?} and channel update {:?}",
+                &channel_announcement, &channel_update
             );
             network
                 .send_message(NetworkActorMessage::new_command(
-                    NetworkActorCommand::ProcessChannelAnnouncement(
-                        self.get_remote_peer_id(),
-                        self.get_funding_transaction_block_number(),
-                        self.get_funding_transaction_index(),
-                        channel_announcement,
-                    ),
-                ))
-                .expect(ASSUME_NETWORK_ACTOR_ALIVE);
-            debug!(
-                "Broadcasting channel update message to peers: {:?}",
-                &channel_update
-            );
-
-            network
-                .send_message(NetworkActorMessage::new_command(
-                    NetworkActorCommand::ProccessChannelUpdate(
-                        self.get_remote_peer_id(),
-                        channel_update,
-                    ),
+                    NetworkActorCommand::BroadcastMessages(vec![
+                        BroadcastMessage::ChannelAnnouncement(channel_announcement),
+                        BroadcastMessage::ChannelUpdate(channel_update),
+                    ]),
                 ))
                 .expect(ASSUME_NETWORK_ACTOR_ALIVE);
         }