offers: wait for a ready messenger state before paying offer

src/lib.rs

@@ -25,10 +25,10 @@ use log4rs::append::console::ConsoleAppender;
 use log4rs::append::file::FileAppender;
 use log4rs::config::{Appender, Config as LogConfig, Root};
 use log4rs::encode::pattern::PatternEncoder;
+use std::cell::RefCell;
 use std::collections::HashMap;
 use std::str::FromStr;
 use std::sync::Mutex;
-use tokio::time::{sleep, Duration};
 use tonic_lnd::lnrpc::GetInfoRequest;
 use tonic_lnd::Client;
 use triggered::{Listener, Trigger};
@@ -43,6 +43,14 @@ pub struct Cfg {
     pub listener: Listener,
 }
 
+// MessengerState tells us whether our onion messenger is still starting up is ready to start
+// forwarding messages.
+#[derive(Debug)]
+pub enum MessengerState {
+    Starting,
+    Ready,
+}
+
 #[allow(dead_code)]
 pub enum OfferState {
     OfferAdded,
@@ -55,13 +63,15 @@ pub enum OfferState {
 pub struct OfferHandler {
     active_offers: Mutex<HashMap<String, OfferState>>,
     pending_messages: Mutex<Vec<PendingOnionMessage<OffersMessage>>>,
+    messenger_state: RefCell<MessengerState>,
 }
 
 impl OfferHandler {
     pub fn new() -> Self {
         OfferHandler {
             active_offers: Mutex::new(HashMap::new()),
             pending_messages: Mutex::new(Vec::new()),
+            messenger_state: RefCell::new(MessengerState::Starting),
         }
     }
 
@@ -75,7 +85,7 @@ impl OfferHandler {
         blinded_path: BlindedPath,
         reply_path: Option<BlindedPath>,
     ) -> Result<(), OfferError<Secp256k1Error>> {
-        sleep(Duration::from_secs(5)).await;
+        self.wait_for_ready().await;
 
         validate_amount(&offer, amount).await?;
 

src/onion_messenger.rs

@@ -1,7 +1,7 @@
 use crate::clock::TokioClock;
 use crate::lnd::{features_support_onion_messages, ONION_MESSAGES_OPTIONAL};
 use crate::rate_limit::{RateLimiter, TokenLimiter};
-use crate::OfferHandler;
+use crate::{MessengerState, OfferHandler};
 use async_trait::async_trait;
 use bitcoin::blockdata::constants::ChainHash;
 use bitcoin::network::constants::Network;
@@ -27,7 +27,7 @@ use std::io::Cursor;
 use std::marker::Copy;
 use std::str::FromStr;
 use tokio::sync::mpsc::{channel, Receiver, Sender};
-use tokio::{select, time, time::Duration, time::Interval};
+use tokio::{select, time, time::sleep, time::Duration, time::Interval};
 use tonic_lnd::{
     lnrpc::peer_event::EventType::PeerOffline, lnrpc::peer_event::EventType::PeerOnline,
     lnrpc::CustomMessage, lnrpc::PeerEvent, lnrpc::SendCustomMessageRequest,
@@ -216,6 +216,7 @@ impl OfferHandler {
             }
         });
 
+        self.messenger_state.replace(MessengerState::Ready);
 
         // Consume events is our main controlling loop, so we run it inline here. We use a RefCell in onion_messenger to
         // allow interior mutability (see LndNodeSigner) so this function can't safely be passed off to another thread.
@@ -265,7 +266,17 @@ impl OfferHandler {
         Ok(())
     }
 
-    Ok(())
+    /// wait_for_ready waits for our onion messenger to finish starting up.
+    pub(crate) async fn wait_for_ready(&self) {
+        loop {
+            sleep(Duration::from_secs(2)).await;
+
+            match *self.messenger_state.borrow() {
+                MessengerState::Starting => continue,
+                MessengerState::Ready => break,
+            };
+        }
+    }
 }
 
 /// lookup_onion_support performs a best-effort lookup in the node's list of current peers to determine whether it