Ignore spent funding cell(s) in channel gossip message

crates/fiber-lib/src/ckb/client.rs

@@ -1,6 +1,6 @@
 use crate::ckb::config::new_ckb_rpc_async_client;
 use crate::ckb::CkbConfig;
-use ckb_jsonrpc_types::JsonBytes;
+use ckb_jsonrpc_types::{CellWithStatus, JsonBytes, OutPoint};
 use ckb_sdk::rpc::ckb_indexer::{Cell, CellType, Order, Pagination, ScriptType, SearchKey, Tx};
 use ckb_types::H256;
 
@@ -112,6 +112,11 @@ fn first_input_tx_hash(txs: &[Tx]) -> Option<H256> {
 #[async_trait::async_trait]
 pub trait CkbChainClient: Send + Sync {
     async fn get_transaction(&self, hash: H256) -> Result<GetTxResponse, anyhow::Error>;
+    async fn get_live_cell(
+        &self,
+        out_point: OutPoint,
+        with_data: bool,
+    ) -> Result<CellWithStatus, anyhow::Error>;
     async fn get_cells(
         &self,
         search_key: SearchKey,
@@ -226,6 +231,18 @@ impl CkbChainClient for CkbRpcClient {
             .map_err(Into::into)
     }
 
+    async fn get_live_cell(
+        &self,
+        out_point: OutPoint,
+        with_data: bool,
+    ) -> Result<CellWithStatus, anyhow::Error> {
+        let client = self.config.ckb_rpc_client();
+        client
+            .get_live_cell(out_point, with_data)
+            .await
+            .map_err(Into::into)
+    }
+
     async fn get_cells(
         &self,
         search_key: SearchKey,

crates/fiber-lib/src/ckb/tests/test_utils.rs

@@ -861,6 +861,31 @@ async fn test_set_and_get_block_timestamp() {
     assert_eq!(timestamp, now);
 }
 
+#[tokio::test]
+async fn test_mock_ckb_chain_client_reports_dead_live_cell() {
+    let shared_state = Arc::new(RwLock::new(MockChainState::new()));
+    let client = MockCkbChainClient::new(shared_state.clone());
+    let out_point = OutPoint::new_builder()
+        .tx_hash(H256::default())
+        .index(0u32)
+        .build();
+
+    {
+        let mut state = shared_state.write().unwrap();
+        state
+            .cell_status
+            .insert(out_point.clone(), CellStatus::Consumed);
+    }
+
+    let result = client
+        .get_live_cell(out_point.clone().into(), false)
+        .await
+        .expect("get live cell");
+
+    assert_eq!(result.status, "dead");
+    assert!(result.cell.is_none());
+}
+
 #[derive(Clone, Debug)]
 pub struct MockCkbChainClient {
     pub state: Arc<RwLock<MockChainState>>,
@@ -912,6 +937,55 @@ impl CkbChainClient for MockCkbChainClient {
             .into())
     }
 
+    async fn get_live_cell(
+        &self,
+        out_point: ckb_jsonrpc_types::OutPoint,
+        _with_data: bool,
+    ) -> Result<ckb_jsonrpc_types::CellWithStatus, anyhow::Error> {
+        let state = self.state.read().unwrap();
+        let packed_out_point: OutPoint = out_point.clone().into();
+
+        let is_consumed = matches!(
+            state.cell_status.get(&packed_out_point),
+            Some(CellStatus::Consumed)
+        );
+
+        if is_consumed {
+            return Ok(ckb_jsonrpc_types::CellWithStatus {
+                cell: None,
+                status: "dead".to_string(),
+            });
+        }
+
+        for response in state.txs.values() {
+            let Some(tx) = &response.transaction else {
+                continue;
+            };
+            for (index, output) in tx.outputs().into_iter().enumerate() {
+                let candidate = ckb_jsonrpc_types::OutPoint::from(
+                    ckb_types::packed::OutPoint::new_builder()
+                        .tx_hash(tx.hash())
+                        .index(ckb_types::packed::Uint32::from(index as u32))
+                        .build(),
+                );
+                if candidate == out_point {
+                    return Ok(ckb_jsonrpc_types::CellWithStatus {
+                        cell: Some(ckb_jsonrpc_types::CellInfo {
+                            output: output.into(),
+                            data: None,
+                        }),
+                        status: "live".to_string(),
+                    });
+                }
+            }
+        }
+
+        Ok(ckb_jsonrpc_types::CellWithStatus {
+            cell: None,
+            status: "unknown".to_string(),
+        })
+    }
+
     async fn get_cells(
         &self,
         _search_key: ckb_sdk::rpc::ckb_indexer::SearchKey,

crates/fiber-lib/src/fiber/gossip.rs

@@ -3171,7 +3171,7 @@ fn get_existing_newer_broadcast_message<S: GossipMessageStore>(
 }
 
 // Verify and save broadcast messages to the store.
-// Note that we can't relialy verify a message until we have all the messages that it depends on.
+// Note that we can't reliably verify a message until we have all the messages that it depends on.
 // So this function should be called by the dependency order of the messages.
 // E.g. channel updates depends on channel announcements to obtain the node public keys,
 // so we should call this method to save and verify channel announcements before channel updates.
@@ -3234,6 +3234,13 @@ async fn verify_and_save_broadcast_message<S: GossipMessageStore>(
     Ok(((message.clone(), timestamp).into(), is_newly_applied))
 }
 
+async fn is_funding_outpoint_live(outpoint: &OutPoint, client: &impl CkbChainClient) -> bool {
+    match client.get_live_cell(outpoint.clone().into(), false).await {
+        Ok(cell) => cell.cell.is_some() && cell.status == "live",
+        _ => false,
+    }
+}
+
 async fn get_channel_tx(
     outpoint: &OutPoint,
     chain: &ActorRef<CkbChainMessage>,
@@ -3259,6 +3266,14 @@ async fn get_channel_tx(
     #[cfg(not(any(test, feature = "bench")))]
     let _ = chain;
 
+    let is_live = is_funding_outpoint_live(outpoint, client).await;
+    if !is_live {
+        return Err(VerifyBroadcastMessageError::InvalidParameter(format!(
+            "Channel announcement funding outpoint {:?} is not live",
+            outpoint
+        )));
+    }
+
     match client.get_transaction(outpoint.tx_hash().unpack()).await {
         Ok(GetTxResponse {
             transaction: Some(tx),

crates/fiber-lib/src/fiber/tests/gossip.rs

@@ -13,7 +13,7 @@ use crate::tests::test_utils::{
     establish_channel_between_nodes, ChannelParameters, NetworkNode, NetworkNodeConfigBuilder,
 };
 use crate::{
-    ckb::tests::test_utils::{MockChainActor, MockChainState},
+    ckb::tests::test_utils::{CellStatus, MockChainActor, MockChainState},
     fiber::types::{ChannelUpdateChannelFlags, NodeAnnouncement},
 };
 use crate::{
@@ -47,6 +47,7 @@ struct GossipTestingContext {
     chain_actor: ActorRef<CkbChainMessage>,
     gossip_actor: ActorRef<GossipActorMessage>,
     gossip_service: GossipService<Store, MockCkbChainClient>,
+    shared_state: Arc<std::sync::RwLock<MockChainState>>,
 }
 
 impl GossipTestingContext {
@@ -79,6 +80,7 @@ impl GossipTestingContext {
             chain_actor,
             gossip_actor: gossip_protocol_handle.actor().clone(),
             gossip_service,
+            shared_state,
         }
     }
 }
@@ -96,6 +98,13 @@ impl GossipTestingContext {
         self.gossip_service.get_store()
     }
 
+    fn mark_funding_outpoint_dead(&self, outpoint: &OutPoint) {
+        let mut state = self.shared_state.write().unwrap();
+        state
+            .cell_status
+            .insert(outpoint.clone(), CellStatus::Consumed);
+    }
+
     fn get_extended_actor(&self) -> &ActorRef<ExtendedGossipMessageStoreMessage> {
         self.gossip_service.get_extended_actor()
     }
@@ -242,6 +251,27 @@ async fn test_saving_confirmed_channel_announcement() {
     assert_ne!(new_announcement, None);
 }
 
+#[tokio::test]
+async fn test_reject_channel_announcement_when_funding_outpoint_is_dead() {
+    let context = GossipTestingContext::new().await;
+    let channel_context = ChannelTestContext::gen().await;
+
+    let status = context.submit_tx(channel_context.funding_tx.clone()).await;
+    assert!(matches!(status, TxStatus::Committed(..)));
+
+    context.mark_funding_outpoint_dead(&channel_context.channel_outpoint());
+    context.save_message(BroadcastMessage::ChannelAnnouncement(
+        channel_context.channel_announcement.clone(),
+    ));
+
+    tokio::time::sleep(Duration::from_millis(200)).await;
+
+    let new_announcement = context
+        .get_store()
+        .get_latest_channel_announcement(channel_context.channel_outpoint());
+    assert_eq!(new_announcement, None);
+}
+
 #[tokio::test]
 // Not supported on wasm: requires filesystem access
 async fn test_saving_invalid_channel_announcement() {

crates/fiber-lib/src/fiber/tests/in_flight_ckb_tx_actor_tests.rs

@@ -31,6 +31,7 @@ fn permanent_send_tx_error() -> RpcError {
 
 struct MockChainClient {
     tx_status: TxStatus,
+    is_live_cell: bool,
 }
 
 #[async_trait::async_trait]
@@ -42,6 +43,32 @@ impl CkbChainClient for MockChainClient {
         })
     }
 
+    async fn get_live_cell(
+        &self,
+        _out_point: ckb_jsonrpc_types::OutPoint,
+        _with_data: bool,
+    ) -> Result<ckb_jsonrpc_types::CellWithStatus, anyhow::Error> {
+        Ok(ckb_jsonrpc_types::CellWithStatus {
+            cell: self.is_live_cell.then(|| ckb_jsonrpc_types::CellInfo {
+                output: ckb_jsonrpc_types::CellOutput {
+                    capacity: 0u64.into(),
+                    lock: ckb_jsonrpc_types::Script {
+                        code_hash: Default::default(),
+                        hash_type: ckb_jsonrpc_types::ScriptHashType::Data,
+                        args: JsonBytes::default(),
+                    },
+                    type_: None,
+                },
+                data: None,
+            }),
+            status: if self.is_live_cell {
+                "live".to_owned()
+            } else {
+                "dead".to_owned()
+            },
+        })
+    }
+
     async fn get_cells(
         &self,
         _search_key: SearchKey,
@@ -209,6 +236,7 @@ async fn spawn_test_actors(
             chain_actor,
             chain_client: MockChainClient {
                 tx_status: tx_status.clone(),
+                is_live_cell: true,
             },
             network_actor,
             tx_hash,