Skip to content

Commit

Permalink
Handle cli interrupt gracefully
Browse files Browse the repository at this point in the history
Print an informative message on how to proceed when any `send`,
`receive`, or `resume` command is interrupted.

Use `tokio::signal::ctrl_c` to do this without introducing any new
dependency.
  • Loading branch information
DanGould committed Jun 27, 2024
1 parent 1656f90 commit b7c367f
Show file tree
Hide file tree
Showing 2 changed files with 116 additions and 52 deletions.
88 changes: 57 additions & 31 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

80 changes: 59 additions & 21 deletions payjoin-cli/src/app/v2.rs
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@ use payjoin::bitcoin::Amount;
use payjoin::receive::v2::ActiveSession;
use payjoin::send::RequestContext;
use payjoin::{base64, bitcoin, Error, Uri};
use tokio::signal;
use tokio::sync::watch;

use super::config::AppConfig;
use super::App as AppTrait;
Expand All @@ -19,13 +21,16 @@ use crate::db::Database;
pub(crate) struct App {
config: AppConfig,
db: Arc<Database>,
interrupt: watch::Receiver<()>,
}

#[async_trait::async_trait]
impl AppTrait for App {
fn new(config: AppConfig) -> Result<Self> {
let db = Arc::new(Database::create(&config.db_path)?);
let app = Self { config, db };
let (interrupt_tx, interrupt_rx) = watch::channel(());
tokio::spawn(handle_interrupt(interrupt_tx));
let app = Self { config, db, interrupt: interrupt_rx };
app.bitcoind()?
.get_blockchain_info()
.context("Failed to connect to bitcoind. Check config RPC connection.")?;
Expand Down Expand Up @@ -102,9 +107,16 @@ impl AppTrait for App {

impl App {
async fn spawn_payjoin_sender(&self, mut req_ctx: RequestContext) -> Result<()> {
let res = self.long_poll_post(&mut req_ctx).await?;
self.process_pj_response(res)?;
self.db.clear_send_session(req_ctx.endpoint())?;
let mut interrupt = self.interrupt.clone();
tokio::select! {
res = self.long_poll_post(&mut req_ctx) => {
self.process_pj_response(res?)?;
self.db.clear_send_session(req_ctx.endpoint())?;
}
_ = interrupt.changed() => {
println!("Interrupted. Call `send` with the same arguments to resume this session or `resume` to resume all sessions.");
}
}
Ok(())
}

Expand All @@ -123,7 +135,15 @@ impl App {
println!("Request Payjoin by sharing this Payjoin Uri:");
println!("{}", pj_uri);

let res = self.long_poll_fallback(&mut session).await?;
let mut interrupt = self.interrupt.clone();
let res = tokio::select! {
res = self.long_poll_fallback(&mut session) => res,
_ = interrupt.changed() => {
println!("Interrupted. Call the `resume` command to resume all sessions.");
return Ok(());
}
}?;

println!("Fallback transaction received. Consider broadcasting this to get paid if the Payjoin fails:");
println!("{}", serialize_hex(&res.extract_tx_to_schedule_broadcast()));
let mut payjoin_proposal = self
Expand Down Expand Up @@ -154,29 +174,40 @@ impl App {
}

pub async fn resume_payjoins(&self) -> Result<()> {
let recv_sessions = self.db.get_recv_sessions()?;
let send_sessions = self.db.get_send_sessions()?;

if recv_sessions.is_empty() && send_sessions.is_empty() {
println!("No sessions to resume.");
return Ok(());
}

let mut tasks = Vec::new();

let recv_sessions = self.db.get_recv_sessions()?;
for recv_session in recv_sessions {
for session in recv_sessions {
let self_clone = self.clone();
tasks.push(tokio::task::spawn(async move {
self_clone.spawn_payjoin_receiver(recv_session, None).await
tasks.push(tokio::spawn(async move {
self_clone.spawn_payjoin_receiver(session, None).await
}));
}
let send_sessions = self.db.get_send_sessions()?;
for send_session in send_sessions {

for session in send_sessions {
let self_clone = self.clone();
tasks.push(tokio::task::spawn(async move {
self_clone.spawn_payjoin_sender(send_session).await
}));
tasks.push(tokio::spawn(async move { self_clone.spawn_payjoin_sender(session).await }));
}
if tasks.is_empty() {
println!("No sessions to resume.");
} else {
for task in tasks {
let _ = task.await?;

let mut interrupt = self.interrupt.clone();
tokio::select! {
_ = async {
for task in tasks {
let _ = task.await;
}
} => {
println!("All resumed sessions completed.");
}
_ = interrupt.changed() => {
println!("Resumed sessions were interrupted.");
}
println!("All resumed sessions completed.");
}
Ok(())
}
Expand All @@ -199,7 +230,7 @@ impl App {
Ok(Some(psbt)) => return Ok(psbt),
Ok(None) => {
println!("No response yet.");
std::thread::sleep(std::time::Duration::from_secs(5))
tokio::time::sleep(std::time::Duration::from_secs(5)).await;
}
Err(re) => {
println!("{}", re);
Expand Down Expand Up @@ -351,6 +382,13 @@ async fn unwrap_ohttp_keys_or_else_fetch(config: &AppConfig) -> Result<payjoin::
}
}

async fn handle_interrupt(tx: watch::Sender<()>) {
if let Err(e) = signal::ctrl_c().await {
eprintln!("Error setting up Ctrl-C handler: {}", e);
}
let _ = tx.send(());
}

fn map_reqwest_err(e: reqwest::Error) -> anyhow::Error {
match e.status() {
Some(status_code) => anyhow!("HTTP request failed: {} {}", status_code, e),
Expand Down

0 comments on commit b7c367f

Please sign in to comment.