Handle interrupts across the whole receive session

Ctrl-C was only honored while the receiver polled for the original
proposal, because the interrupt select lived inside
read_from_directory. Since tokio::signal::ctrl_c replaces the default
terminate handler for the life of the process, an interrupt during
any other receiver phase (for example a transient retry against an
unresponsive relay) was swallowed entirely and the process could only
be stopped by an external kill.

Wrap process_receiver_session in a select against the interrupt
channel in receive_payjoin, mirroring how send_payjoin guards the
sender driver. This is safe at any phase: transitions persist to the
event log with no await between transition and save, so an
interrupted session resumes from its last saved state via the resume
command. The interrupt message now includes the session id, and
read_from_directory becomes a plain single-step function like the
other state handlers.

Author: xstoicunicornx

payjoin-cli/src/app/v2/mod.rs

@@ -293,9 +293,17 @@ impl AppTrait for App {
         println!("Request Payjoin by sharing this Payjoin Uri:");
         println!("{pj_uri}");
 
-        self.process_receiver_session(ReceiveSession::Initialized(session.clone()), &persister)
-            .await?;
-        Ok(())
+        let mut interrupt = self.interrupt.clone();
+        tokio::select! {
+            res = self.process_receiver_session(ReceiveSession::Initialized(session.clone()), &persister) => res,
+            _ = interrupt.changed() => {
+                let id = persister.session_id();
+                println!(
+                    "Session {id} interrupted. Call the `resume` command to resume all sessions, or `payjoin-cli cancel {id}` to cancel the session."
+                );
+                Err(anyhow!("Interrupted"))
+            }
+        }
     }
 
     async fn resume_payjoins(&self, session_id: Option<SessionId>) -> Result<()> {
@@ -806,43 +814,31 @@ impl App {
         }
     }
 
-    #[allow(clippy::incompatible_msrv)]
     async fn read_from_directory(
         &self,
         session: Receiver<Initialized>,
         persister: &ReceiverPersister,
     ) -> Result<ReceiveSession> {
-        let poll_once = async {
-            println!("Polling receive request...");
-            let (ohttp_response, context) =
-                self.post_via_relay(|relay| session.create_poll_request(relay)).await?;
-            let state_transition = session
-                .process_response(ohttp_response.bytes().await?.to_vec().as_slice(), context)
-                .save(persister);
-            match state_transition {
-                Ok(OptionalTransitionOutcome::Progress(next_state)) => {
-                    println!("Got a request from the sender. Responding with a Payjoin proposal.");
-                    Ok(ReceiveSession::UncheckedOriginalPayload(next_state))
-                }
-                Ok(OptionalTransitionOutcome::Stasis(current_state)) =>
-                    Ok(ReceiveSession::Initialized(current_state)),
-                Err(e) if e.is_transient() => {
-                    tracing::debug!("Transient error polling for request, retrying: {e:?}");
-                    let session =
-                        e.transient_state().expect("transient error carries current state");
-                    tokio::time::sleep(TRANSIENT_RETRY_DELAY).await;
-                    Ok(ReceiveSession::Initialized(session))
-                }
-                Err(e) => Err(e.into()),
+        println!("Polling receive request...");
+        let (ohttp_response, context) =
+            self.post_via_relay(|relay| session.create_poll_request(relay)).await?;
+        let state_transition = session
+            .process_response(ohttp_response.bytes().await?.to_vec().as_slice(), context)
+            .save(persister);
+        match state_transition {
+            Ok(OptionalTransitionOutcome::Progress(next_state)) => {
+                println!("Got a request from the sender. Responding with a Payjoin proposal.");
+                Ok(ReceiveSession::UncheckedOriginalPayload(next_state))
             }
-        };
-        let mut interrupt = self.interrupt.clone();
-        tokio::select! {
-            res = poll_once => res,
-            _ = interrupt.changed() => {
-                println!("Interrupted. Call the `resume` command to resume all sessions.");
-                Err(anyhow!("Interrupted"))
+            Ok(OptionalTransitionOutcome::Stasis(current_state)) =>
+                Ok(ReceiveSession::Initialized(current_state)),
+            Err(e) if e.is_transient() => {
+                tracing::debug!("Transient error polling for request, retrying: {e:?}");
+                let session = e.transient_state().expect("transient error carries current state");
+                tokio::time::sleep(TRANSIENT_RETRY_DELAY).await;
+                Ok(ReceiveSession::Initialized(session))
             }
+            Err(e) => Err(e.into()),
         }
     }