From 5649b07dc26462b3f3be6bcbc8636c9e5afbac62 Mon Sep 17 00:00:00 2001
From: YI <uuuuuu@protonmail.com>
Date: Sat, 4 Jan 2025 17:21:41 +0800
Subject: [PATCH] Store our own broadcast messages directly

---
 src/ckb/actor.rs            |  29 ++---
 src/ckb/tests/test_utils.rs |   9 +-
 src/fiber/channel.rs        |  32 +++--
 src/fiber/gossip.rs         | 225 +++++++++++++++++-------------------
 src/fiber/network.rs        |  77 +++++++++---
 src/fiber/types.rs          | 126 +++++++-------------
 src/store/tests/store.rs    |   2 +-
 7 files changed, 254 insertions(+), 246 deletions(-)

diff --git a/src/ckb/actor.rs b/src/ckb/actor.rs
index 994748f41..baf7f0ae2 100644
--- a/src/ckb/actor.rs
+++ b/src/ckb/actor.rs
@@ -41,18 +41,19 @@ impl TraceTxResponse {
     }
 }
 
-#[derive(Debug, Clone)]
-pub struct GetBlockTimestampRequest {
-    block_hash: H256,
+#[derive(Debug, Clone, Hash, PartialEq, Eq)]
+pub enum GetBlockTimestampRequest {
+    BlockNumber(u64),
+    BlockHash(H256),
 }
 
 impl GetBlockTimestampRequest {
     pub fn from_block_hash(block_hash: H256) -> Self {
-        Self { block_hash }
+        Self::BlockHash(block_hash)
     }
 
-    pub fn block_hash(&self) -> H256 {
-        self.block_hash.clone()
+    pub fn from_block_number(block_number: u64) -> Self {
+        Self::BlockNumber(block_number)
     }
 }
 
@@ -257,18 +258,20 @@ impl Actor for CkbChainActor {
                     }
                 }
             }
-            CkbChainMessage::GetBlockTimestamp(
-                GetBlockTimestampRequest { block_hash },
-                reply_port,
-            ) => {
+            CkbChainMessage::GetBlockTimestamp(request, 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
+                    let timestamp = match request {
+                        GetBlockTimestampRequest::BlockNumber(block_number) => ckb_client
+                            .get_header_by_number(block_number.into())
+                            .map(|x| x.map(|x| x.inner.timestamp.into())),
+                        GetBlockTimestampRequest::BlockHash(block_hash) => ckb_client
                             .get_header(block_hash)
                             .map(|x| x.map(|x| x.inner.timestamp.into())),
-                    );
+                    };
+                    let _ = reply_port.send(timestamp);
                 });
             }
         }
diff --git a/src/ckb/tests/test_utils.rs b/src/ckb/tests/test_utils.rs
index f47a8166c..37762010a 100644
--- a/src/ckb/tests/test_utils.rs
+++ b/src/ckb/tests/test_utils.rs
@@ -6,7 +6,6 @@ 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, OnceCell};
 use std::{collections::HashMap, sync::Arc, sync::RwLock};
@@ -16,7 +15,7 @@ use crate::{
     ckb::{
         config::UdtCfgInfos,
         contracts::{Contract, ContractsContext, ContractsInfo},
-        TraceTxRequest, TraceTxResponse,
+        GetBlockTimestampRequest, TraceTxRequest, TraceTxResponse,
     },
     now_timestamp_as_millis_u64,
 };
@@ -501,14 +500,16 @@ impl Actor for MockChainActor {
                 // 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();
+                static BLOCK_TIMESTAMP: OnceCell<
+                    TokioRwLock<HashMap<GetBlockTimestampRequest, u64>>,
+                > = OnceCell::new();
                 BLOCK_TIMESTAMP.get_or_init(|| TokioRwLock::new(HashMap::new()));
                 let timestamp = *BLOCK_TIMESTAMP
                     .get()
                     .unwrap()
                     .write()
                     .await
-                    .entry(request.block_hash())
+                    .entry(request.clone())
                     .or_insert(now_timestamp_as_millis_u64());
 
                 let _ = rpc_reply_port.send(Ok(Some(timestamp)));
diff --git a/src/fiber/channel.rs b/src/fiber/channel.rs
index 43e1b75f6..9680cf830 100644
--- a/src/fiber/channel.rs
+++ b/src/fiber/channel.rs
@@ -1,6 +1,9 @@
 #[cfg(debug_assertions)]
 use crate::fiber::network::DebugEvent;
-use crate::{debug_event, fiber::serde_utils::U64Hex};
+use crate::{
+    debug_event,
+    fiber::{serde_utils::U64Hex, types::BroadcastMessageWithTimestamp},
+};
 use bitflags::bitflags;
 use ckb_jsonrpc_types::BlockNumber;
 use futures::future::OptionFuture;
@@ -25,7 +28,7 @@ use crate::{
         },
         serde_utils::{CompactSignatureAsBytes, EntityHex, PubNonceAsBytes},
         types::{
-            AcceptChannel, AddTlc, AnnouncementSignatures, BroadcastMessage, BroadcastMessageQuery,
+            AcceptChannel, AddTlc, AnnouncementSignatures, BroadcastMessageQuery,
             BroadcastMessageQueryFlags, ChannelAnnouncement, ChannelReady, ChannelUpdate,
             ClosingSigned, CommitmentSigned, EcdsaSignature, FiberChannelMessage, FiberMessage,
             Hash256, OpenChannel, PaymentOnionPacket, PeeledPaymentOnionPacket, Privkey, Pubkey,
@@ -1709,7 +1712,7 @@ where
         event: ChannelEvent,
     ) -> Result<(), ProcessingChannelError> {
         match event {
-            ChannelEvent::FundingTransactionConfirmed(block_number, tx_index) => {
+            ChannelEvent::FundingTransactionConfirmed(block_number, tx_index, timestamp) => {
                 debug!("Funding transaction confirmed");
                 let flags = match state.state {
                     ChannelState::AwaitingChannelReady(flags) => flags,
@@ -1723,7 +1726,7 @@ where
                             "Expecting funding transaction confirmed event in state AwaitingChannelReady or after TX_SIGNATURES_SENT, but got state {:?}", &state.state)));
                     }
                 };
-                state.funding_tx_confirmed_at = Some((block_number, tx_index));
+                state.funding_tx_confirmed_at = Some((block_number, tx_index, timestamp));
                 self.network
                     .send_message(NetworkActorMessage::new_command(
                         NetworkActorCommand::SendFiberMessage(FiberMessageWithPeerId::new(
@@ -1766,7 +1769,7 @@ where
                     self.network
                         .send_message(NetworkActorMessage::new_command(
                             NetworkActorCommand::BroadcastMessages(vec![
-                                BroadcastMessage::ChannelUpdate(update),
+                                BroadcastMessageWithTimestamp::ChannelUpdate(update),
                             ]),
                         ))
                         .expect(ASSUME_NETWORK_ACTOR_ALIVE);
@@ -2766,7 +2769,7 @@ pub struct ChannelActorState {
     #[serde_as(as = "Option<EntityHex>")]
     pub funding_tx: Option<Transaction>,
 
-    pub funding_tx_confirmed_at: Option<(BlockNumber, u32)>,
+    pub funding_tx_confirmed_at: Option<(BlockNumber, u32, u64)>,
 
     #[serde_as(as = "Option<EntityHex>")]
     pub funding_udt_type_script: Option<Script>,
@@ -2932,7 +2935,7 @@ pub struct ClosedChannel {}
 #[derive(Debug)]
 pub enum ChannelEvent {
     PeerDisconnected,
-    FundingTransactionConfirmed(BlockNumber, u32),
+    FundingTransactionConfirmed(BlockNumber, u32, u64),
     CommitmentTransactionConfirmed,
     ClosingTransactionConfirmed,
     CheckTlcSetdown,
@@ -3453,9 +3456,9 @@ impl ChannelActorState {
         let channel_update = self.generate_channel_update(network).await;
         network
             .send_message(NetworkActorMessage::new_command(
-                NetworkActorCommand::BroadcastMessages(vec![BroadcastMessage::ChannelUpdate(
-                    channel_update,
-                )]),
+                NetworkActorCommand::BroadcastMessages(vec![
+                    BroadcastMessageWithTimestamp::ChannelUpdate(channel_update),
+                ]),
             ))
             .expect(ASSUME_NETWORK_ACTOR_ALIVE);
     }
@@ -5384,8 +5387,13 @@ impl ChannelActorState {
             network
                 .send_message(NetworkActorMessage::new_command(
                     NetworkActorCommand::BroadcastMessages(vec![
-                        BroadcastMessage::ChannelAnnouncement(channel_announcement),
-                        BroadcastMessage::ChannelUpdate(channel_update),
+                        BroadcastMessageWithTimestamp::ChannelAnnouncement(
+                            self.funding_tx_confirmed_at
+                                .expect("funding tx confirmed")
+                                .2,
+                            channel_announcement,
+                        ),
+                        BroadcastMessageWithTimestamp::ChannelUpdate(channel_update),
                     ]),
                 ))
                 .expect(ASSUME_NETWORK_ACTOR_ALIVE);
diff --git a/src/fiber/gossip.rs b/src/fiber/gossip.rs
index f2f70c17e..c2083ba1b 100644
--- a/src/fiber/gossip.rs
+++ b/src/fiber/gossip.rs
@@ -39,10 +39,10 @@ use super::{
     network::{check_chain_hash, get_chain_hash, GossipMessageWithPeerId, GOSSIP_PROTOCOL_ID},
     types::{
         BroadcastMessage, BroadcastMessageID, BroadcastMessageQuery, BroadcastMessageQueryFlags,
-        BroadcastMessageWithOnChainInfo, BroadcastMessageWithTimestamp, BroadcastMessagesFilter,
-        BroadcastMessagesFilterResult, ChannelAnnouncement, ChannelOnchainInfo, ChannelUpdate,
-        Cursor, GetBroadcastMessages, GetBroadcastMessagesResult, GossipMessage, NodeAnnouncement,
-        Pubkey, QueryBroadcastMessages, QueryBroadcastMessagesResult,
+        BroadcastMessageWithTimestamp, BroadcastMessagesFilter, BroadcastMessagesFilterResult,
+        ChannelAnnouncement, ChannelOnchainInfo, ChannelUpdate, Cursor, GetBroadcastMessages,
+        GetBroadcastMessagesResult, GossipMessage, NodeAnnouncement, Pubkey,
+        QueryBroadcastMessages, QueryBroadcastMessagesResult,
     },
 };
 
@@ -323,7 +323,7 @@ pub enum GossipActorMessage {
     // Try to broadcast BroadcastMessage created by us to the network.
     // We will save and broadcast the messages. Note that we don't check the dependencies of
     // these messages because we assume that the messages created by us are always valid.
-    TryBroadcastMessages(Vec<BroadcastMessage>),
+    TryBroadcastMessages(Vec<BroadcastMessageWithTimestamp>),
     // Broadcast a message to the network. The message here must have all its dependencies met.
     // This is normally the case when we saved a message to the store.
     // This is an internal command used by both TryBroadcastMessages and ProcessBroadcastMessage,
@@ -976,7 +976,7 @@ pub struct ExtendedGossipMessageStoreState<S> {
     chain_actor: ActorRef<CkbChainMessage>,
     next_id: u64,
     output_ports: HashMap<u64, BroadcastMessageOutput>,
-    messages_to_be_saved: Vec<BroadcastMessageWithOnChainInfo>,
+    messages_to_be_saved: HashSet<BroadcastMessage>,
 }
 
 impl<S: GossipMessageStore> ExtendedGossipMessageStoreState<S> {
@@ -1002,12 +1002,12 @@ impl<S: GossipMessageStore> ExtendedGossipMessageStoreState<S> {
     async fn prune_messages_to_be_saved(&mut self) -> Vec<BroadcastMessageWithTimestamp> {
         // Note that we have to call has_dependencies_available before changing messages_to_be_saved,
         // as the function will check the dependencies of the message in the current messages_to_be_saved.
-        let (complete_messages, uncomplete_messages) = self
+        let (complete_messages, incomplete_messages) = self
             .messages_to_be_saved
             .clone()
             .into_iter()
             .partition(|m| self.has_dependencies_available(m));
-        self.messages_to_be_saved = uncomplete_messages;
+        self.messages_to_be_saved = incomplete_messages;
 
         let mut sorted_messages = complete_messages.into_iter().collect::<Vec<_>>();
         sorted_messages.sort_unstable();
@@ -1018,9 +1018,10 @@ impl<S: GossipMessageStore> ExtendedGossipMessageStoreState<S> {
 
         let mut verified_sorted_messages = Vec::with_capacity(sorted_messages.len());
         for message in sorted_messages {
-            match verify_and_save_broadcast_message(&message, &self.store).await {
-                Ok(_) => {
-                    verified_sorted_messages.push(message.into());
+            match verify_and_save_broadcast_message(&message, &self.store, &self.chain_actor).await
+            {
+                Ok(message) => {
+                    verified_sorted_messages.push(message);
                 }
                 Err(error) => {
                     trace!(
@@ -1035,6 +1036,55 @@ impl<S: GossipMessageStore> ExtendedGossipMessageStoreState<S> {
         verified_sorted_messages
     }
 
+    fn store_messages(&mut self, messages: &[BroadcastMessageWithTimestamp]) {
+        for message in messages {
+            match message {
+                BroadcastMessageWithTimestamp::ChannelAnnouncement(
+                    timestamp,
+                    channel_announcement,
+                ) => {
+                    self.store
+                        .save_channel_announcement(*timestamp, channel_announcement.clone());
+                }
+                BroadcastMessageWithTimestamp::ChannelUpdate(channel_update) => {
+                    self.store.save_channel_update(channel_update.clone());
+                }
+                BroadcastMessageWithTimestamp::NodeAnnouncement(node_announcement) => {
+                    self.store.save_node_announcement(node_announcement.clone());
+                }
+            }
+        }
+    }
+
+    fn broadcast_messages(&mut self, messages: &[BroadcastMessageWithTimestamp]) {
+        if messages.is_empty() {
+            return;
+        }
+
+        for (id, subscription) in self.output_ports.iter() {
+            let messages_to_send = messages
+                .iter()
+                .filter(|m| &m.cursor() > &subscription.filter)
+                .cloned()
+                .collect::<Vec<_>>();
+            trace!(
+                "Sending complete messages in memory to subscription #{}: number of messages = {}",
+                id,
+                messages_to_send.len()
+            );
+            for chunk in messages_to_send.chunks(MAX_NUM_OF_BROADCAST_MESSAGES as usize) {
+                subscription
+                    .output_port
+                    .send(GossipMessageUpdates::new(chunk.to_vec()));
+            }
+        }
+    }
+
+    fn store_and_broadcast_messages(&mut self, messages: &[BroadcastMessageWithTimestamp]) {
+        self.store_messages(messages);
+        self.broadcast_messages(messages);
+    }
+
     fn get_channel_annnouncement(&self, outpoint: &OutPoint) -> Option<(u64, ChannelAnnouncement)> {
         self.store
             .get_latest_channel_announcement(outpoint)
@@ -1046,11 +1096,10 @@ impl<S: GossipMessageStore> ExtendedGossipMessageStoreState<S> {
         outpoint: &OutPoint,
     ) -> Option<(u64, ChannelAnnouncement)> {
         self.messages_to_be_saved.iter().find_map(|m| match m {
-            BroadcastMessageWithOnChainInfo::ChannelAnnouncement(
-                on_chain_info,
-                channel_announcement,
-            ) if &channel_announcement.channel_outpoint == outpoint => {
-                Some((on_chain_info.timestamp, channel_announcement.clone()))
+            BroadcastMessage::ChannelAnnouncement(channel_announcement)
+                if &channel_announcement.channel_outpoint == outpoint =>
+            {
+                Some((0, channel_announcement.clone()))
             }
             _ => None,
         })
@@ -1059,38 +1108,33 @@ impl<S: GossipMessageStore> ExtendedGossipMessageStoreState<S> {
     async fn insert_message_to_be_saved_list(
         &mut self,
         message: &BroadcastMessage,
-    ) -> Result<BroadcastMessageWithTimestamp, GossipMessageProcessingError> {
+    ) -> Result<(), GossipMessageProcessingError> {
         if let Some(existing_message) = get_existing_newer_broadcast_message(message, &self.store) {
             if &BroadcastMessage::from(existing_message.clone()) != message {
                 return Err(GossipMessageProcessingError::NewerMessageSaved(
                     existing_message,
                 ));
             } else {
-                return Ok(existing_message);
+                return Ok(());
             }
         }
 
-        match self.messages_to_be_saved.iter().find(|m| message == *m) {
-            Some(existing_message) => {
-                return Ok(existing_message.clone().into());
-            }
-            None => {}
+        if let Some(_) = self.messages_to_be_saved.get(&message) {
+            return Ok(());
         }
 
-        let message = get_broadcast_message_with_on_chain_info(message.clone(), &self.chain_actor)
-            .await
-            .map_err(|error| GossipMessageProcessingError::ProcessingError(error.to_string()))?;
-
-        let max_acceptable_gossip_message_timestamp = max_acceptable_gossip_message_timestamp();
-        if message.timestamp() > max_acceptable_gossip_message_timestamp {
-            return Err(GossipMessageProcessingError::MessageTooNew(
-                message.timestamp(),
-                max_acceptable_gossip_message_timestamp,
-            ));
+        if let Some(timestamp) = message.timestamp() {
+            let max_acceptable_gossip_message_timestamp = max_acceptable_gossip_message_timestamp();
+            if timestamp > max_acceptable_gossip_message_timestamp {
+                return Err(GossipMessageProcessingError::MessageTooNew(
+                    timestamp,
+                    max_acceptable_gossip_message_timestamp,
+                ));
+            }
         }
 
         if !self.announce_private_addr {
-            if let BroadcastMessageWithOnChainInfo::NodeAnnouncement(node_announcement) = &message {
+            if let BroadcastMessage::NodeAnnouncement(node_announcement) = &message {
                 if !node_announcement.addresses.iter().any(|addr| {
                     multiaddr_to_socketaddr(addr)
                         .map(|socket_addr| is_reachable(socket_addr.ip()))
@@ -1104,13 +1148,13 @@ impl<S: GossipMessageStore> ExtendedGossipMessageStoreState<S> {
         }
 
         trace!("New gossip message saved to memory: {:?}", message);
-        self.messages_to_be_saved.push(message.clone());
-        Ok(message.into())
+        self.messages_to_be_saved.insert(message.clone());
+        Ok(())
     }
 
-    fn has_dependencies_available(&self, message: &BroadcastMessageWithOnChainInfo) -> bool {
+    fn has_dependencies_available(&self, message: &BroadcastMessage) -> bool {
         match message {
-            BroadcastMessageWithOnChainInfo::ChannelUpdate(channel_update) => self
+            BroadcastMessage::ChannelUpdate(channel_update) => self
                 .get_channel_annnouncement(&channel_update.channel_outpoint)
                 .is_some(),
             _ => true,
@@ -1279,21 +1323,8 @@ impl<S: GossipMessageStore + Send + Sync + 'static> Actor for ExtendedGossipMess
                 }
             }
 
-            ExtendedGossipMessageStoreMessage::SaveAndBroadcastMessage(message) => {
-                match state.insert_message_to_be_saved_list(&message).await {
-                    Err(error) => error!(
-                        "Failed to save message and broadcast message: {:?}, error: {:?}",
-                        message, error
-                    ),
-                    Ok(message) => {
-                        // Normally we will need to check if the message has any dependencies that are not saved yet.
-                        // But this is not necessary for the messages that we are proactively broadcasting.
-                        // These are our own channel announcements, channel updates and node announcements.
-                        let _ = state
-                            .gossip_actor
-                            .send_message(GossipActorMessage::BroadcastMessageImmediately(message));
-                    }
-                }
+            ExtendedGossipMessageStoreMessage::SaveAndBroadcastMessages(messages) => {
+                state.store_and_broadcast_messages(&messages);
             }
 
             ExtendedGossipMessageStoreMessage::Tick => {
@@ -1305,26 +1336,7 @@ impl<S: GossipMessageStore + Send + Sync + 'static> Actor for ExtendedGossipMess
 
                 // These are the messages that have complete dependencies and can be sent to the subscribers.
                 let complete_messages = state.prune_messages_to_be_saved().await;
-                if complete_messages.is_empty() {
-                    return Ok(());
-                }
-                for (id, subscription) in state.output_ports.iter() {
-                    let messages_to_send = complete_messages
-                        .iter()
-                        .filter(|m| &m.cursor() > &subscription.filter)
-                        .cloned()
-                        .collect::<Vec<_>>();
-                    trace!(
-                        "Sending complete messages in memory to subscription #{}: number of messages = {}",
-                        id,
-                        messages_to_send.len()
-                    );
-                    for chunk in messages_to_send.chunks(MAX_NUM_OF_BROADCAST_MESSAGES as usize) {
-                        subscription
-                            .output_port
-                            .send(GossipMessageUpdates::new(chunk.to_vec()));
-                    }
-                }
+                state.broadcast_messages(&complete_messages);
             }
         }
         Ok(())
@@ -1353,7 +1365,7 @@ pub enum ExtendedGossipMessageStoreMessage {
     SaveMessage(BroadcastMessage),
     // Save a new message to the store. If the message's dependencies are all met, then we will
     // broadcast the message to the network. Otherwise, we will ignore the broadcasting.
-    SaveAndBroadcastMessage(BroadcastMessage),
+    SaveAndBroadcastMessages(Vec<BroadcastMessageWithTimestamp>),
     // Send broadcast messages after the cursor to the subscriber specified in the u64 id.
     // This is normally called immediately after a new subscription is created. This is the time when
     // we need to send existing messages to the subscriber.
@@ -1771,28 +1783,6 @@ fn get_existing_newer_broadcast_message<S: GossipMessageStore>(
     })
 }
 
-async fn get_broadcast_message_with_on_chain_info(
-    message: BroadcastMessage,
-    chain: &ActorRef<CkbChainMessage>,
-) -> Result<BroadcastMessageWithOnChainInfo, Error> {
-    match message {
-        BroadcastMessage::ChannelAnnouncement(channel_announcement) => {
-            let on_chain_info =
-                get_channel_on_chain_info(&channel_announcement.channel_outpoint, chain).await?;
-            Ok(BroadcastMessageWithOnChainInfo::ChannelAnnouncement(
-                on_chain_info,
-                channel_announcement,
-            ))
-        }
-        BroadcastMessage::ChannelUpdate(channel_update) => Ok(
-            BroadcastMessageWithOnChainInfo::ChannelUpdate(channel_update),
-        ),
-        BroadcastMessage::NodeAnnouncement(node_announcement) => Ok(
-            BroadcastMessageWithOnChainInfo::NodeAnnouncement(node_announcement),
-        ),
-    }
-}
-
 // Channel updates depends on channel announcements to obtain the node public keys.
 // If a channel update is saved before the channel announcement, we can't reliably determine if
 // this channel update is valid. So we need to save the channel update to lagged_messages and
@@ -1802,33 +1792,36 @@ async fn get_broadcast_message_with_on_chain_info(
 // announcement is saved before the channel announcement, we need to temporarily save the channel
 // announcement to lagged_messages and wait for the node announcement to be saved.
 async fn verify_and_save_broadcast_message<S: GossipMessageStore>(
-    message: &BroadcastMessageWithOnChainInfo,
+    message: &BroadcastMessage,
     store: &S,
-) -> Result<(), Error> {
-    match message {
-        BroadcastMessageWithOnChainInfo::ChannelAnnouncement(
-            on_chain_info,
-            channel_announcement,
-        ) => {
-            if !verify_channel_announcement(channel_announcement, on_chain_info, store).await? {
+    chain: &ActorRef<CkbChainMessage>,
+) -> Result<BroadcastMessageWithTimestamp, Error> {
+    let timestamp = match message {
+        BroadcastMessage::ChannelAnnouncement(channel_announcement) => {
+            let on_chain_info =
+                get_channel_on_chain_info(channel_announcement.out_point(), chain).await?;
+            if !verify_channel_announcement(channel_announcement, &on_chain_info, store).await? {
                 store.save_channel_announcement(
                     on_chain_info.timestamp,
                     channel_announcement.clone(),
                 );
             }
+            on_chain_info.timestamp
         }
-        BroadcastMessageWithOnChainInfo::ChannelUpdate(channel_update) => {
+        BroadcastMessage::ChannelUpdate(channel_update) => {
             if !verify_channel_update(channel_update, store)? {
                 store.save_channel_update(channel_update.clone());
             }
+            channel_update.timestamp
         }
-        BroadcastMessageWithOnChainInfo::NodeAnnouncement(node_announcement) => {
+        BroadcastMessage::NodeAnnouncement(node_announcement) => {
             if !verify_node_announcement(node_announcement, store)? {
                 store.save_node_announcement(node_announcement.clone());
             }
+            node_announcement.timestamp
         }
-    }
-    Ok(())
+    };
+    Ok((message.clone(), timestamp).into())
 }
 
 async fn get_channel_tx(
@@ -2297,15 +2290,13 @@ where
                     .await?;
             }
             GossipActorMessage::TryBroadcastMessages(messages) => {
-                for message in messages {
-                    state
-                        .store
-                        .actor
-                        .send_message(ExtendedGossipMessageStoreMessage::SaveAndBroadcastMessage(
-                            message.clone(),
-                        ))
-                        .expect("store actor alive");
-                }
+                state
+                    .store
+                    .actor
+                    .send_message(ExtendedGossipMessageStoreMessage::SaveAndBroadcastMessages(
+                        messages,
+                    ))
+                    .expect("store actor alive");
             }
             GossipActorMessage::BroadcastMessageImmediately(message) => {
                 for (peer, peer_state) in &state.peer_states {
diff --git a/src/fiber/network.rs b/src/fiber/network.rs
index d504c05d9..ee61b52dc 100644
--- a/src/fiber/network.rs
+++ b/src/fiber/network.rs
@@ -59,15 +59,18 @@ use super::gossip::{GossipActorMessage, GossipMessageStore, GossipMessageUpdates
 use super::graph::{NetworkGraph, NetworkGraphStateStore, SessionRoute};
 use super::key::blake2b_hash_with_salt;
 use super::types::{
-    BroadcastMessage, BroadcastMessageQuery, EcdsaSignature, FiberMessage, GossipMessage, Hash256,
-    NodeAnnouncement, OpenChannel, PaymentHopData, Privkey, Pubkey, RemoveTlcReason, TlcErr,
-    TlcErrData, TlcErrorCode,
+    BroadcastMessage, BroadcastMessageQuery, BroadcastMessageWithTimestamp, EcdsaSignature,
+    FiberMessage, GossipMessage, Hash256, NodeAnnouncement, OpenChannel, PaymentHopData, Privkey,
+    Pubkey, RemoveTlcReason, TlcErr, TlcErrData, TlcErrorCode,
 };
 use super::{FiberConfig, ASSUME_NETWORK_ACTOR_ALIVE};
 
 use crate::ckb::config::UdtCfgInfos;
 use crate::ckb::contracts::{check_udt_script, get_udt_whitelist, is_udt_type_auto_accept};
-use crate::ckb::{CkbChainMessage, FundingRequest, FundingTx, TraceTxRequest, TraceTxResponse};
+use crate::ckb::{
+    CkbChainMessage, FundingRequest, FundingTx, GetBlockTimestampRequest, TraceTxRequest,
+    TraceTxResponse,
+};
 use crate::fiber::channel::{
     AddTlcCommand, AddTlcResponse, TxCollaborationCommand, TxUpdateCommand,
 };
@@ -227,7 +230,7 @@ pub enum NetworkActorCommand {
     // we use this to query them.
     QueryBroadcastMessages(PeerId, Vec<BroadcastMessageQuery>),
     // Broadcast our BroadcastMessage to the network.
-    BroadcastMessages(Vec<BroadcastMessage>),
+    BroadcastMessages(Vec<BroadcastMessageWithTimestamp>),
     // Broadcast local information to the network.
     BroadcastLocalInfo(LocalInfoKind),
     SignMessage([u8; 32], RpcReplyPort<EcdsaSignature>),
@@ -598,8 +601,8 @@ pub enum NetworkActorEvent {
     FundingTransactionPending(Transaction, OutPoint, Hash256),
 
     /// A funding transaction has been confirmed. The transaction was included in the
-    /// block with the given transaction index.
-    FundingTransactionConfirmed(OutPoint, BlockNumber, u32),
+    /// block with the given transaction index, and the timestamp in the block header.
+    FundingTransactionConfirmed(OutPoint, BlockNumber, u32, u64),
 
     /// A funding transaction has failed.
     FundingTransactionFailed(OutPoint),
@@ -857,9 +860,14 @@ where
                     .on_funding_transaction_pending(transaction, outpoint.clone(), channel_id)
                     .await;
             }
-            NetworkActorEvent::FundingTransactionConfirmed(outpoint, block_number, tx_index) => {
+            NetworkActorEvent::FundingTransactionConfirmed(
+                outpoint,
+                block_number,
+                tx_index,
+                timestamp,
+            ) => {
                 state
-                    .on_funding_transaction_confirmed(outpoint, block_number, tx_index)
+                    .on_funding_transaction_confirmed(outpoint, block_number, tx_index, timestamp)
                     .await;
             }
             NetworkActorEvent::CommitmentTransactionPending(transaction, channel_id) => {
@@ -1305,7 +1313,7 @@ where
                     myself
                         .send_message(NetworkActorMessage::new_command(
                             NetworkActorCommand::BroadcastMessages(vec![
-                                BroadcastMessage::NodeAnnouncement(message),
+                                BroadcastMessageWithTimestamp::NodeAnnouncement(message),
                             ]),
                         ))
                         .expect(ASSUME_NETWORK_MYSELF_ALIVE);
@@ -2441,9 +2449,9 @@ where
                 remote_peer_id, &message
             );
             let _ = self.network.send_message(NetworkActorMessage::new_command(
-                NetworkActorCommand::BroadcastMessages(vec![BroadcastMessage::NodeAnnouncement(
-                    message,
-                )]),
+                NetworkActorCommand::BroadcastMessages(vec![
+                    BroadcastMessageWithTimestamp::NodeAnnouncement(message),
+                ]),
             ));
         } else {
             debug!(
@@ -2662,6 +2670,7 @@ where
             &outpoint, &channel_id, &tx_hash
         );
         let network = self.network.clone();
+        let chain = self.chain_actor.clone();
         self.broadcast_tx_with_callback(transaction, move |result| {
             let message = match result {
                 Ok(TraceTxResponse {
@@ -2673,11 +2682,51 @@ where
                         },
                     ..
                 }) => {
+                    const MAX_GET_BLOCK_TIMESTAMP_RETRY: u8 = 5;
+                    use tokio::runtime::Runtime;
+                    let rt = Runtime::new().unwrap();
+
+
+    let timestamp=                        rt.block_on( async {
+                            for _ in 0..MAX_GET_BLOCK_TIMESTAMP_RETRY {
+                                match call!(
+                                    chain,
+                                    |reply| 
+                                CkbChainMessage::GetBlockTimestamp(
+                                    GetBlockTimestampRequest::from_block_number(block_number.into()), reply
+                                )
+                            ) {
+                                Ok(Ok(Some(timestamp))) => return timestamp,
+                                Ok(Ok(None)) => {
+                                    panic!(
+                                        "Failed to get block timestamp for block number {:?}: block not found",
+                                        &block_number
+                                    );
+                                }
+                                Ok(Err(err)) => {
+                                    error!(
+                                        "Failed to get block timestamp for block number {:?}: {:?}",
+                                        &block_number, &err
+                                    );
+                                }
+                                Err(err) => {
+                                    error!(
+                                        "Failed to get block timestamp for block number {:?}: {:?}",
+                                        &block_number, &err
+                                    );
+                                }
+
+                            } 
+                        }
+                            panic!("Failed to get block timestamp for block number {:?} after {} retries", &block_number, MAX_GET_BLOCK_TIMESTAMP_RETRY);
+                    });
+
                     info!("Funding transaction {:?} confirmed", &tx_hash);
                     NetworkActorEvent::FundingTransactionConfirmed(
                         outpoint,
                         block_number.into(),
                         DUMMY_FUNDING_TX_INDEX,
+                        timestamp,
                     )
                 }
                 Ok(status) => {
@@ -2754,6 +2803,7 @@ where
         outpoint: OutPoint,
         block_number: BlockNumber,
         tx_index: u32,
+        timestamp: u64,
     ) {
         debug!("Funding transaction is confirmed: {:?}", &outpoint);
         let channel_id = match self.pending_channels.remove(&outpoint) {
@@ -2772,6 +2822,7 @@ where
             ChannelActorMessage::Event(ChannelEvent::FundingTransactionConfirmed(
                 block_number,
                 tx_index,
+                timestamp,
             )),
         )
         .await;
diff --git a/src/fiber/types.rs b/src/fiber/types.rs
index 855e0f7a8..b4f1d5c4d 100644
--- a/src/fiber/types.rs
+++ b/src/fiber/types.rs
@@ -2441,87 +2441,40 @@ impl BroadcastMessage {
             }
         }
     }
-}
-
-#[derive(Debug, Clone, PartialEq, Eq, Hash)]
-pub struct ChannelOnchainInfo {
-    pub timestamp: u64,
-    pub first_output: CellOutput,
-}
-
-// Augment the broadcast message with on-chain information so that we can verify the validity of the message.
-#[derive(Debug, Clone, PartialEq, Eq, Hash)]
-pub enum BroadcastMessageWithOnChainInfo {
-    NodeAnnouncement(NodeAnnouncement),
-    ChannelAnnouncement(ChannelOnchainInfo, ChannelAnnouncement),
-    ChannelUpdate(ChannelUpdate),
-}
-
-impl BroadcastMessageWithOnChainInfo {
-    pub fn cursor(&self) -> Cursor {
-        match self {
-            Self::NodeAnnouncement(node_announcement) => Cursor::new(
-                node_announcement.timestamp,
-                BroadcastMessageID::NodeAnnouncement(node_announcement.node_id),
-            ),
-            Self::ChannelAnnouncement(on_chain_info, channel_announcement) => Cursor::new(
-                on_chain_info.timestamp,
-                BroadcastMessageID::ChannelAnnouncement(
-                    channel_announcement.channel_outpoint.clone(),
-                ),
-            ),
-            Self::ChannelUpdate(channel_update) => Cursor::new(
-                channel_update.timestamp,
-                BroadcastMessageID::ChannelUpdate(channel_update.channel_outpoint.clone()),
-            ),
-        }
-    }
 
-    pub fn timestamp(&self) -> u64 {
+    pub(crate) fn message_id(&self) -> BroadcastMessageID {
         match self {
-            Self::NodeAnnouncement(node_announcement) => node_announcement.timestamp,
-            Self::ChannelAnnouncement(on_chain_info, _) => on_chain_info.timestamp,
-            Self::ChannelUpdate(channel_update) => channel_update.timestamp,
-        }
-    }
-
-    pub fn message_id(&self) -> BroadcastMessageID {
-        match self {
-            Self::NodeAnnouncement(node_announcement) => {
+            BroadcastMessage::NodeAnnouncement(node_announcement) => {
                 BroadcastMessageID::NodeAnnouncement(node_announcement.node_id)
             }
-            Self::ChannelAnnouncement(_, channel_announcement) => {
+            BroadcastMessage::ChannelAnnouncement(channel_announcement) => {
                 BroadcastMessageID::ChannelAnnouncement(
                     channel_announcement.channel_outpoint.clone(),
                 )
             }
-            Self::ChannelUpdate(channel_update) => {
+            BroadcastMessage::ChannelUpdate(channel_update) => {
                 BroadcastMessageID::ChannelUpdate(channel_update.channel_outpoint.clone())
             }
         }
     }
-}
 
-impl PartialEq<BroadcastMessageWithOnChainInfo> for BroadcastMessage {
-    fn eq(&self, other: &BroadcastMessageWithOnChainInfo) -> bool {
-        match (self, other) {
-            (
-                BroadcastMessage::NodeAnnouncement(node_announcement),
-                BroadcastMessageWithOnChainInfo::NodeAnnouncement(other_node_announcement),
-            ) => node_announcement == other_node_announcement,
-            (
-                BroadcastMessage::ChannelAnnouncement(channel_announcement),
-                BroadcastMessageWithOnChainInfo::ChannelAnnouncement(_, other_channel_announcement),
-            ) => channel_announcement == other_channel_announcement,
-            (
-                BroadcastMessage::ChannelUpdate(channel_update),
-                BroadcastMessageWithOnChainInfo::ChannelUpdate(other_channel_update),
-            ) => channel_update == other_channel_update,
-            _ => false,
+    pub(crate) fn timestamp(&self) -> Option<u64> {
+        match self {
+            BroadcastMessage::NodeAnnouncement(node_announcement) => {
+                Some(node_announcement.timestamp)
+            }
+            BroadcastMessage::ChannelAnnouncement(_) => None,
+            BroadcastMessage::ChannelUpdate(channel_update) => Some(channel_update.timestamp),
         }
     }
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct ChannelOnchainInfo {
+    pub timestamp: u64,
+    pub first_output: CellOutput,
+}
+
 // Augment the broadcast message with timestamp so that we can easily obtain the cursor of the message.
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub enum BroadcastMessageWithTimestamp {
@@ -2601,7 +2554,7 @@ impl BroadcastMessageWithTimestamp {
     }
 }
 
-impl Ord for BroadcastMessageWithOnChainInfo {
+impl Ord for BroadcastMessage {
     fn cmp(&self, other: &Self) -> Ordering {
         self.message_id()
             .cmp(&other.message_id())
@@ -2609,12 +2562,33 @@ impl Ord for BroadcastMessageWithOnChainInfo {
     }
 }
 
-impl PartialOrd for BroadcastMessageWithOnChainInfo {
+impl PartialOrd for BroadcastMessage {
     fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
         Some(self.cmp(other))
     }
 }
 
+impl From<(BroadcastMessage, Option<ChannelOnchainInfo>)> for BroadcastMessageWithTimestamp {
+    fn from(
+        (broadcast_message, channel_onchain_info): (BroadcastMessage, Option<ChannelOnchainInfo>),
+    ) -> Self {
+        match broadcast_message {
+            BroadcastMessage::NodeAnnouncement(node_announcement) => {
+                BroadcastMessageWithTimestamp::NodeAnnouncement(node_announcement)
+            }
+            BroadcastMessage::ChannelAnnouncement(channel_announcement) => {
+                let timestamp = channel_onchain_info
+                    .expect("Channel onchain info is required for channel announcement")
+                    .timestamp;
+                BroadcastMessageWithTimestamp::ChannelAnnouncement(timestamp, channel_announcement)
+            }
+            BroadcastMessage::ChannelUpdate(channel_update) => {
+                BroadcastMessageWithTimestamp::ChannelUpdate(channel_update)
+            }
+        }
+    }
+}
+
 impl From<BroadcastMessageWithTimestamp> for BroadcastMessage {
     fn from(broadcast_message_with_timestamp: BroadcastMessageWithTimestamp) -> Self {
         match broadcast_message_with_timestamp {
@@ -2649,26 +2623,6 @@ impl From<(BroadcastMessage, u64)> for BroadcastMessageWithTimestamp {
     }
 }
 
-impl From<BroadcastMessageWithOnChainInfo> for BroadcastMessageWithTimestamp {
-    fn from(broadcast_message_with_onchain_info: BroadcastMessageWithOnChainInfo) -> Self {
-        match broadcast_message_with_onchain_info {
-            BroadcastMessageWithOnChainInfo::NodeAnnouncement(node_announcement) => {
-                BroadcastMessageWithTimestamp::NodeAnnouncement(node_announcement)
-            }
-            BroadcastMessageWithOnChainInfo::ChannelAnnouncement(
-                channel_onchain_info,
-                channel_announcement,
-            ) => BroadcastMessageWithTimestamp::ChannelAnnouncement(
-                channel_onchain_info.timestamp,
-                channel_announcement,
-            ),
-            BroadcastMessageWithOnChainInfo::ChannelUpdate(channel_update) => {
-                BroadcastMessageWithTimestamp::ChannelUpdate(channel_update)
-            }
-        }
-    }
-}
-
 impl From<BroadcastMessage> for molecule_gossip::BroadcastMessageUnion {
     fn from(fiber_broadcast_message: BroadcastMessage) -> Self {
         match fiber_broadcast_message {
diff --git a/src/store/tests/store.rs b/src/store/tests/store.rs
index c383cf676..c6bf1a32f 100644
--- a/src/store/tests/store.rs
+++ b/src/store/tests/store.rs
@@ -331,7 +331,7 @@ fn test_channel_actor_state_store() {
         local_pubkey: gen_rand_fiber_public_key(),
         remote_pubkey: gen_rand_fiber_public_key(),
         funding_tx: Some(Transaction::default()),
-        funding_tx_confirmed_at: Some((1.into(), 1)),
+        funding_tx_confirmed_at: Some((1.into(), 1, 1)),
         is_acceptor: true,
         funding_udt_type_script: Some(Script::default()),
         to_local_amount: 100,