plugins(lsps2): route recovered sessions through forward_event path

After restart, recovered session actors were stored in a separate
recovery_handles Vec, unreachable by the forward_event notification
path that routes via the sessions HashMap. This caused intermittent
CI failures where on_payment_settled could not find the session and
the internal forward-monitoring loop failed to detect settlement.

Register recovered sessions in the sessions HashMap keyed by
payment_hash so forward_event notifications reach them directly.
For already-settled forwards, recover into Broadcasting state so
the actor self-drives to completion without needing forward_event
re-delivery. Remove the now-redundant internal polling loop
(get_forward_activity + wait_for_forward_resolution).

Author: nepet

plugins/lsps-plugin/src/cln_adapters/rpc.rs

@@ -26,9 +26,8 @@ use cln_rpc::{
             FundchannelCompleteRequest, FundchannelStartRequest, FundpsbtRequest, GetinfoRequest,
             ListdatastoreRequest, ListforwardsIndex, ListforwardsRequest, ListpeerchannelsRequest,
             SendpsbtRequest, SignpsbtRequest, UnreserveinputsRequest,
-            WaitIndexname, WaitRequest, WaitSubsystem,
         },
-        responses::{ListdatastoreResponse, ListforwardsForwardsStatus, WaitForwardsStatus},
+        responses::{ListdatastoreResponse, ListforwardsForwardsStatus},
     },
     primitives::{Amount, AmountOrAll, ChannelState, Feerate, Sha256},
     ClnRpc,
@@ -845,63 +844,6 @@ impl RecoveryProvider for ClnRecoveryProvider {
         }
     }
 
-    async fn wait_for_forward_resolution(
-        &self,
-        channel_id: &str,
-        from_index: u64,
-    ) -> Result<(ForwardActivity, u64)> {
-        // Get the scid for this channel so we can match wait responses.
-        let scid = self.rpc.get_channel_scid(channel_id).await?;
-
-        let mut next_index = from_index + 1;
-        loop {
-            let mut rpc = self.rpc.create_rpc().await?;
-            let wait_res = rpc
-                .call_typed(&WaitRequest {
-                    subsystem: WaitSubsystem::FORWARDS,
-                    indexname: WaitIndexname::UPDATED,
-                    nextvalue: next_index,
-                })
-                .await
-                .with_context(|| {
-                    format!("calling wait for channel_id={channel_id} at index={next_index}")
-                })?;
-
-            let new_index = wait_res.updated.unwrap_or(next_index);
-
-            // Check if this update is for our channel.
-            let is_our_channel = match (&scid, &wait_res.forwards) {
-                (Some(our_scid), Some(fwd)) => fwd
-                    .out_channel
-                    .as_ref()
-                    .map(|c| c == our_scid)
-                    .unwrap_or(false),
-                _ => false,
-            };
-
-            if is_our_channel {
-                if let Some(fwd) = &wait_res.forwards {
-                    match fwd.status {
-                        Some(WaitForwardsStatus::SETTLED) => {
-                            return Ok((ForwardActivity::Settled, new_index));
-                        }
-                        Some(WaitForwardsStatus::OFFERED) => {
-                            return Ok((ForwardActivity::Offered, new_index));
-                        }
-                        Some(WaitForwardsStatus::FAILED)
-                        | Some(WaitForwardsStatus::LOCAL_FAILED) => {
-                            // Check full history to decide AllFailed vs Active.
-                            let activity = self.get_forward_activity(channel_id).await?;
-                            return Ok((activity, new_index));
-                        }
-                        None => {}
-                    }
-                }
-            }
-
-            next_index = new_index + 1;
-        }
-    }
 }
 
 // ---------------------------------------------------------------------------

plugins/lsps-plugin/src/core/lsps2/actor.rs

@@ -1,7 +1,7 @@
 use crate::{
     core::lsps2::{
         event_sink::{EventSink, SessionEventEnvelope},
-        provider::{DatastoreProvider, ForwardActivity, RecoveryProvider},
+        provider::DatastoreProvider,
         session::{PaymentPart, Session, SessionAction, SessionEvent, SessionInput},
     },
     proto::{
@@ -173,12 +173,9 @@ impl<A: ActionExecutor + Clone + Send + 'static, D: DatastoreProvider + Clone +
         session: Session,
         entry: DatastoreEntry,
         initial_actions: Vec<SessionAction>,
-        channel_id: String,
         executor: A,
         scid: ShortChannelId,
         datastore: D,
-        recovery: Arc<dyn RecoveryProvider>,
-        forwards_updated_index: Option<u64>,
         event_sink: Arc<dyn EventSink>,
     ) -> ActorInboxHandle {
         let (tx, inbox) = mpsc::channel(128);
@@ -200,12 +197,7 @@ impl<A: ActionExecutor + Clone + Send + 'static, D: DatastoreProvider + Clone +
             event_sink,
         };
 
-        tokio::spawn(actor.run_recovered(
-            initial_actions,
-            channel_id,
-            recovery,
-            forwards_updated_index,
-        ));
+        tokio::spawn(actor.run_recovered(initial_actions));
         handle
     }
 
@@ -376,9 +368,6 @@ impl<A: ActionExecutor + Clone + Send + 'static, D: DatastoreProvider + Clone +
     async fn run_recovered(
         mut self,
         initial_actions: Vec<SessionAction>,
-        channel_id: String,
-        recovery: Arc<dyn RecoveryProvider>,
-        forwards_updated_index: Option<u64>,
     ) {
         // Execute initial actions (e.g., BroadcastFundingTx for Broadcasting state)
         for action in initial_actions {
@@ -390,83 +379,11 @@ impl<A: ActionExecutor + Clone + Send + 'static, D: DatastoreProvider + Clone +
             return;
         }
 
-        // Start forward monitoring
-        let from_index = forwards_updated_index.unwrap_or(0);
-        let self_tx = self.self_send.clone();
-        let monitor_handle = {
-            let recovery = recovery.clone();
-            let channel_id = channel_id.clone();
-            let scid = self.scid;
-
-            tokio::spawn(async move {
-                // First: check listforwards for already-settled forwards
-                match recovery.get_forward_activity(&channel_id).await {
-                    Ok(ForwardActivity::Settled) => {
-                        let _ = self_tx
-                            .send(ActorInput::PaymentSettled {
-                                preimage: None,
-                                updated_index: None,
-                            })
-                            .await;
-                        return;
-                    }
-                    Ok(ForwardActivity::AllFailed) => {
-                        let _ = self_tx
-                            .send(ActorInput::PaymentFailed { updated_index: None })
-                            .await;
-                        return;
-                    }
-                    Ok(ForwardActivity::Offered)
-                    | Ok(ForwardActivity::NoForwards)
-                    | Err(_) => {
-                        // Fall through to wait loop
-                    }
-                }
-
-                // Poll using wait subsystem
-                let mut current_index = from_index;
-                loop {
-                    match recovery
-                        .wait_for_forward_resolution(&channel_id, current_index)
-                        .await
-                    {
-                        Ok((ForwardActivity::Settled, new_index)) => {
-                            let _ = self_tx
-                                .send(ActorInput::PaymentSettled {
-                                    preimage: None,
-                                    updated_index: Some(new_index),
-                                })
-                                .await;
-                            return;
-                        }
-                        Ok((ForwardActivity::AllFailed, new_index)) => {
-                            let _ = self_tx
-                                .send(ActorInput::PaymentFailed {
-                                    updated_index: Some(new_index),
-                                })
-                                .await;
-                            return;
-                        }
-                        Ok((ForwardActivity::Offered, new_index))
-                        | Ok((ForwardActivity::NoForwards, new_index)) => {
-                            current_index = new_index;
-                            continue;
-                        }
-                        Err(e) => {
-                            warn!("forward monitoring error for scid={scid}: {e}");
-                            tokio::time::sleep(Duration::from_secs(5)).await;
-                            continue;
-                        }
-                    }
-                }
-            })
-        };
-
-        // Main loop: process inbox events
+        // Main loop: process inbox events from forward_event notifications
         loop {
             match self.inbox.recv().await {
                 Some(actor_input) => {
-                    // Only process the three shared persistence arms
+                    // Only process settlement/failure/broadcast events
                     let session_input = match &actor_input {
                         ActorInput::PaymentSettled { .. }
                         | ActorInput::PaymentFailed { .. }
@@ -486,7 +403,6 @@ impl<A: ActionExecutor + Clone + Send + 'static, D: DatastoreProvider + Clone +
             }
         }
 
-        monitor_handle.abort();
         Self::finalize(&self.session, &self.datastore, self.scid).await;
     }