Offload watchtower TLC status queries to dedicated actor

- Add WatchtowerQueryActor to run query_tlc_status without blocking Network Actor
- Queue QueryPreimage and QuerySettled jobs from CheckChannels; handle results
  via WatchtowerPreimageResult and WatchtowerSettledResult commands
- Pass tlc_id through preimage flow so result handler can send RemoveTlc
  when preimage is found; only send if channel still has that received TLC

Made-with: Cursor

Author: doitian

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

@@ -19,6 +19,7 @@ mod in_flight_ckb_tx_actor;
 mod key;
 mod path;
 mod settle_tlc_set_command;
+mod watchtower_query_actor;
 
 pub use config::FiberConfig;
 pub use in_flight_ckb_tx_actor::{

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

@@ -69,6 +69,7 @@ use super::types::{
     Hash256, Init, NodeAnnouncement, OpenChannel, Privkey, Pubkey, RemoveTlcFulfill,
     RemoveTlcReason, TlcErr, TlcErrorCode,
 };
+use super::watchtower_query_actor::{WatchtowerQueryActor, WatchtowerQueryActorMessage};
 use super::{
     FiberConfig, InFlightCkbTxActor, InFlightCkbTxActorArguments, InFlightCkbTxKind,
     ASSUME_NETWORK_ACTOR_ALIVE,
@@ -383,6 +384,21 @@ pub enum NetworkActorCommand {
     // Get all inbound channel requests that are waiting for `accept_channel`
     GetPendingAcceptChannels(RpcReplyPort<Result<Vec<PendingAcceptChannel>, String>>),
 
+    // Watchtower query result: preimage found for a TLC, settle it immediately.
+    WatchtowerPreimageResult {
+        channel_id: Hash256,
+        tlc_id: u64,
+        payment_hash: Hash256,
+        preimage: Hash256,
+    },
+    // Watchtower query result: TLC is settled on-chain.
+    // The forwarding TLC should be failed with ExpiryTooSoon.
+    WatchtowerSettledResult {
+        forwarding_channel_id: Hash256,
+        forwarding_tlc_id: u64,
+        shared_secret: [u8; 32],
+    },
+
     #[cfg(any(debug_assertions, feature = "bench"))]
     UpdateFeatures(FeatureVector),
 }
@@ -1349,28 +1365,6 @@ where
                                         continue;
                                     }
                                 }
-
-                                let mut payment_preimage =
-                                    self.store.get_preimage(&tlc.payment_hash);
-                                if payment_preimage.is_none() {
-                                    if let Some(querier) = self.watchtower_querier.as_ref() {
-                                        payment_preimage = querier
-                                            .query_tlc_status(&channel_id, &tlc.payment_hash)
-                                            .await
-                                            .and_then(|s| s.preimage);
-                                        if let Some(preimage) = payment_preimage {
-                                            self.store.insert_preimage(tlc.payment_hash, preimage);
-                                        }
-                                    }
-                                }
-                                let Some(payment_preimage) = payment_preimage else {
-                                    continue;
-                                };
-                                debug!(
-                                    "Found payment preimage for channel {:?} tlc {:?}",
-                                    channel_id,
-                                    tlc.id()
-                                );
                                 if self
                                     .store
                                     .get_invoice_status(&tlc.payment_hash)
@@ -1384,6 +1378,29 @@ where
                                     continue;
                                 }
 
+                                let payment_preimage = self.store.get_preimage(&tlc.payment_hash);
+                                if payment_preimage.is_none() {
+                                    if let Some(wt_actor) = state.watchtower_query_actor.as_ref() {
+                                        let _ = wt_actor.send_message(
+                                            WatchtowerQueryActorMessage::QueryPreimage {
+                                                channel_id,
+                                                tlc_id: tlc.id(),
+                                                payment_hash: tlc.payment_hash,
+                                            },
+                                        );
+                                    }
+                                    continue;
+                                }
+                                let Some(payment_preimage) = payment_preimage else {
+                                    continue;
+                                };
+
+                                debug!(
+                                    "Found payment preimage for channel {:?} tlc {:?}",
+                                    channel_id,
+                                    tlc.id()
+                                );
+
                                 let (send, _recv) = oneshot::channel();
                                 let rpc_reply = RpcReplyPort::from(send);
 
@@ -1521,43 +1538,16 @@ where
                                 if let Some((forwarding_channel_id, forwarding_tlc_id)) =
                                     tlc.forwarding_tlc
                                 {
-                                    let is_settled =
-                                        if let Some(querier) = self.watchtower_querier.as_ref() {
-                                            querier
-                                                .query_tlc_status(&channel_id, &tlc.payment_hash)
-                                                .await
-                                                .is_some_and(|s| s.is_settled)
-                                        } else {
-                                            false
-                                        };
-                                    if is_settled {
-                                        let (send, _recv) = oneshot::channel();
-                                        let rpc_reply = RpcReplyPort::from(send);
-                                        if let Err(err) = state
-                                            .send_command_to_channel(
+                                    if let Some(wt_actor) = state.watchtower_query_actor.as_ref() {
+                                        let _ = wt_actor.send_message(
+                                            WatchtowerQueryActorMessage::QuerySettled {
+                                                channel_id,
+                                                payment_hash: tlc.payment_hash,
                                                 forwarding_channel_id,
-                                                ChannelCommand::RemoveTlc(
-                                                    RemoveTlcCommand {
-                                                        id: forwarding_tlc_id,
-                                                        reason: RemoveTlcReason::RemoveTlcFail(
-                                                            TlcErrPacket::new(
-                                                                TlcErr::new(
-                                                                    TlcErrorCode::ExpiryTooSoon,
-                                                                ),
-                                                                &tlc.shared_secret,
-                                                            ),
-                                                        ),
-                                                    },
-                                                    rpc_reply,
-                                                ),
-                                            )
-                                            .await
-                                        {
-                                            error!(
-                                                "Failed to remove settled tlc {:?} for channel {:?}: {}",
-                                                forwarding_tlc_id, forwarding_channel_id, err
-                                            );
-                                        }
+                                                forwarding_tlc_id,
+                                                shared_secret: tlc.shared_secret,
+                                            },
+                                        );
                                     }
                                 }
                             }
@@ -1622,6 +1612,90 @@ where
                     }
                 }
             }
+            NetworkActorCommand::WatchtowerPreimageResult {
+                channel_id,
+                tlc_id,
+                payment_hash,
+                preimage,
+            } => {
+                self.store.insert_preimage(payment_hash, preimage);
+                debug!(
+                    "Found watchtower preimage for channel {:?} tlc {:?}",
+                    channel_id, tlc_id
+                );
+                if self
+                    .store
+                    .get_invoice_status(&payment_hash)
+                    .is_some_and(|s| {
+                        !matches!(s, CkbInvoiceStatus::Open | CkbInvoiceStatus::Received)
+                    })
+                {
+                    return Ok(());
+                }
+                // Only send RemoveTlc if the channel still has this received TLC (it may have
+                // been removed already, e.g. by a concurrent payment winning the race).
+                let tlc_still_exists =
+                    self.store
+                        .get_channel_actor_state(&channel_id)
+                        .is_some_and(|actor_state| {
+                            actor_state
+                                .get_received_tlc(TLCId::Received(tlc_id))
+                                .is_some()
+                        });
+                if tlc_still_exists {
+                    let (send, _recv) = oneshot::channel();
+                    let rpc_reply = RpcReplyPort::from(send);
+                    if let Err(err) = state
+                        .send_command_to_channel(
+                            channel_id,
+                            ChannelCommand::RemoveTlc(
+                                RemoveTlcCommand {
+                                    id: tlc_id,
+                                    reason: RemoveTlcReason::RemoveTlcFulfill(RemoveTlcFulfill {
+                                        payment_preimage: preimage,
+                                    }),
+                                },
+                                rpc_reply,
+                            ),
+                        )
+                        .await
+                    {
+                        error!(
+                            "Failed to remove tlc {:?} with preimage for channel {:?}: {}",
+                            tlc_id, channel_id, err
+                        );
+                    }
+                }
+            }
+            NetworkActorCommand::WatchtowerSettledResult {
+                forwarding_channel_id,
+                forwarding_tlc_id,
+                shared_secret,
+            } => {
+                let (send, _recv) = oneshot::channel();
+                let rpc_reply = RpcReplyPort::from(send);
+                if let Err(err) = state
+                    .send_command_to_channel(
+                        forwarding_channel_id,
+                        ChannelCommand::RemoveTlc(
+                            RemoveTlcCommand {
+                                id: forwarding_tlc_id,
+                                reason: RemoveTlcReason::RemoveTlcFail(TlcErrPacket::new(
+                                    TlcErr::new(TlcErrorCode::ExpiryTooSoon),
+                                    &shared_secret,
+                                )),
+                            },
+                            rpc_reply,
+                        ),
+                    )
+                    .await
+                {
+                    error!(
+                        "Failed to remove settled tlc {:?} for channel {:?}: {}",
+                        forwarding_tlc_id, forwarding_channel_id, err
+                    );
+                }
+            }
             NetworkActorCommand::SettleHoldTlcSet(payment_hash) => {
                 self.settle_hold_tlc_set(state, payment_hash).await;
             }
@@ -2928,6 +3002,8 @@ pub struct NetworkActorState<S, C> {
 
     // Inflight payment actors
     inflight_payments: HashMap<Hash256, ActorRef<PaymentActorMessage>>,
+    // Actor for querying watchtower TLC status without blocking the network actor.
+    watchtower_query_actor: Option<ActorRef<WatchtowerQueryActorMessage>>,
 }
 
 #[derive(Debug, Clone)]
@@ -4696,6 +4772,20 @@ where
         let chain_actor = self.chain_actor.clone();
         let features = config.gen_node_features();
 
+        let watchtower_query_actor = if let Some(querier) = self.watchtower_querier.clone() {
+            let actor = WatchtowerQueryActor::new(querier, myself.clone());
+            let (actor_ref, _) = Actor::spawn_linked(
+                Some("watchtower_query".to_string()),
+                actor,
+                (),
+                myself.get_cell(),
+            )
+            .await?;
+            Some(actor_ref)
+        } else {
+            None
+        };
+
         let mut state = NetworkActorState {
             store: self.store.clone(),
             state_to_be_persisted,
@@ -4732,6 +4822,7 @@ where
                 funding_timeout_seconds: config.funding_timeout_seconds,
             },
             inflight_payments: Default::default(),
+            watchtower_query_actor,
         };
 
         let node_announcement = state.get_or_create_new_node_announcement_message();