Repository: breez/breez-sdk-greenlight (language: python)

Issue:
detect and act on network change
When there is a network change, tonic behaves as follows:
- After the keepalive timeout, reconnect automatically.
- Before the keepalive timeout, any grpc call will time out. After the timeout, the connection is reestablished.

This commit adds a mechanism to reconnect all grpc clients after one of the clients detects a network change. Initially it was attempted to only retry based on a keepalive timeout error, but a network change affects all grpc clients, so subsequent requests to other grpc endpoints would still fail with a timeout. Ofcourse those grpc clients can also add a retry-on-timeout, but since it is known at this point the grpc clients are temporarily dead, reconnect them immediately. This ensures subsequent calls to other endpoints won't add additional time by waiting for a timeout.

Currently this PR wraps the `trampoline_pay`, `pay`, and `invoice` calls in the greenlight client.

tokio was bumped to allow cloning watch::Sender


Discussion / hints:
Updated to wrap most grpc calls in `with_connection_fallback`.
The notification pattern that recreated other grpc clients when one hanged is now gone.
Payment was still getting stuck. Added more wrappers.
@roeierez Found another way to make the compiler happy by passing a single function, what do you think?

```
pub async fn with_connection_retry<C, RQ, RS>(
    client: &mut C, 
    req: RQ,
    f: impl for<'c> Fn(&'c mut C, RQ) -> Pin<Box<dyn Future<Output = Result<tonic::Response<RS>, tonic::Status>> + Send + 'c>>,
) -> Result<tonic::Response<RS>, tonic::Status>
where
    RQ: std::fmt::Debug + Clone,
    RS: std::fmt::Debug,
{
    let res = f(client, req.clone()).await;
    ...
    f(client, req).await
}
```

```
        let chain_api_servers = with_connection_retry(
            &mut client,
            ChainApiServersRequest {},
            |client, req| Box::pin(client.chain_api_servers(req)),
        )
            .await
```



Now the retry logic is wrapped in a macro. Posting the commit message here:

A macro makes for a little less code duplication. The macro returns a
single awaitable future. So the result of the macro can be used in join!
calls. I was unable to do this without consuming the grpc client object.
Therefore the grpc client object has to be cloned if it's used later
again.


Fix patch:
diff --git a/libs/sdk-common/src/breez_server.rs b/libs/sdk-common/src/breez_server.rs
index 0cc668370..c9961ca2b 100644
--- a/libs/sdk-common/src/breez_server.rs
+++ b/libs/sdk-common/src/breez_server.rs
@@ -26,7 +26,6 @@ pub static STAGING_BREEZSERVER_URL: &str = "https://bs1-st.breez.technology:443"
 pub struct BreezServer {
     grpc_channel: Mutex<Channel>,
     api_key: Option<String>,
-    server_url: String,
 }
 
 impl BreezServer {
@@ -34,15 +33,9 @@ impl BreezServer {
         Ok(Self {
             grpc_channel: Mutex::new(Self::create_endpoint(&server_url)?.connect_lazy()),
             api_key,
-            server_url,
         })
     }
 
-    pub async fn reconnect(&self) -> Result<()> {
-        *self.grpc_channel.lock().await = Self::create_endpoint(&self.server_url)?.connect_lazy();
-        Ok(())
-    }
-
     fn create_endpoint(server_url: &str) -> Result<Endpoint> {
         Ok(Endpoint::from_shared(server_url.to_string())?
             .http2_keep_alive_interval(Duration::new(5, 0))
diff --git a/libs/sdk-core/src/breez_services.rs b/libs/sdk-core/src/breez_services.rs
index 6bc1d7772..a5d9ccf5a 100644
--- a/libs/sdk-core/src/breez_services.rs
+++ b/libs/sdk-core/src/breez_services.rs
@@ -48,8 +48,6 @@ use crate::swap_out::boltzswap::BoltzApi;
 use crate::swap_out::reverseswap::BTCSendSwap;
 use crate::*;
 
-const DETECT_HIBERNATE_SLEEP_DURATION: Duration = Duration::from_secs(1);
-const DETECT_HIBERNATE_MAX_OFFSET: Duration = Duration::from_secs(2);
 pub type BreezServicesResult<T, E = ConnectError> = Result<T, E>;
 
 /// Trait that can be used to react to various [BreezEvent]s emitted by the SDK.
@@ -170,8 +168,6 @@ pub struct BreezServices {
     backup_watcher: Arc<BackupWatcher>,
     shutdown_sender: watch::Sender<()>,
     shutdown_receiver: watch::Receiver<()>,
-    reconnect_sender: watch::Sender<()>,
-    reconnect_receiver: watch::Receiver<()>,
 }
 
 impl BreezServices {
@@ -1437,9 +1433,6 @@ impl BreezServices {
     ///
     /// Internal method. Should only be used as part of [BreezServices::start]
     async fn start_background_tasks(self: &Arc<BreezServices>) -> SdkResult<()> {
-        // Detect hibernation
-        self.detect_hibernation();
-
         // start the signer
         let (shutdown_signer_sender, signer_signer_receiver) = watch::channel(());
         self.start_signer(signer_signer_receiver).await;
@@ -1491,45 +1484,7 @@ impl BreezServices {
         Ok(())
     }
 
-    async fn reconnect(self: &Arc<BreezServices>) {
-        // Reconnect node api before notifying anything else, to
-        // ensure there are no races reconnecting dependant
-        // services.
-        debug!("Reconnecting node api.");
-        self.node_api.reconnect().await;
-
-        // Now notify dependant services.
-        debug!("Notifying services about reconnect.");
-        let _ = self.reconnect_sender.send(());
-    }
-
-    fn detect_hibernation(self: &Arc<BreezServices>) {
-        let cloned = Arc::clone(self);
-        tokio::spawn(async move {
-            loop {
-                let now = SystemTime::now();
-                tokio::time::sleep(DETECT_HIBERNATE_SLEEP_DURATION).await;
-                let elapsed = match now.elapsed() {
-                    Ok(elapsed) => elapsed,
-                    Err(e) => {
-                        error!("track_hibernation failed with: {:?}", e);
-                        continue;
-                    }
-                };
-
-                if elapsed
-                    .saturating_sub(DETECT_HIBERNATE_SLEEP_DURATION)
-                    .ge(&DETECT_HIBERNATE_MAX_OFFSET)
-                {
-                    debug!("Hibernation detected, time diff {}s", elapsed.as_secs_f32());
-                    cloned.reconnect().await;
-                }
-            }
-        });
-    }
-
     async fn start_signer(self: &Arc<BreezServices>, mut shutdown_receiver: watch::Receiver<()>) {
-        let mut reconnect_receiver = self.reconnect_receiver.clone();
         let node_api = self.node_api.clone();
 
         tokio::spawn(async move {
@@ -1544,12 +1499,6 @@ impl BreezServices {
                     _ = shutdown_receiver.changed() => {
                         true
                     }
-
-                    _ = reconnect_receiver.changed() => {
-                        // NOTE: The node api is reconnected already inside the
-                        //       reconnect function, to avoid races.
-                        false
-                    }
                 };
 
                 debug!("shutting down signer");
@@ -1650,7 +1599,6 @@ impl BreezServices {
         let cloned = self.clone();
         tokio::spawn(async move {
             let mut shutdown_receiver = cloned.shutdown_receiver.clone();
-            let mut reconnect_receiver = cloned.reconnect_receiver.clone();
             loop {
                 if shutdown_receiver.has_changed().unwrap_or(true) {
                     return;
@@ -1674,11 +1622,6 @@ impl BreezServices {
                             debug!("Invoice tracking task has completed");
                             return;
                         }
-
-                        _ = reconnect_receiver.changed() => {
-                            debug!("Reconnect: track invoices");
-                            break;
-                        }
                     };
 
                     let i = match paid_invoice_res {
@@ -1735,7 +1678,6 @@ impl BreezServices {
         let cloned = self.clone();
         tokio::spawn(async move {
             let mut shutdown_receiver = cloned.shutdown_receiver.clone();
-            let mut reconnect_receiver = cloned.reconnect_receiver.clone();
             loop {
                 if shutdown_receiver.has_changed().unwrap_or(true) {
                     return;
@@ -1759,11 +1701,6 @@ impl BreezServices {
                             debug!("Track logs task has completed");
                             return;
                         }
-
-                        _ = reconnect_receiver.changed() => {
-                            debug!("Reconnect: track logs");
-                            break;
-                        }
                     };
 
                     match log_message_res {
@@ -2396,7 +2333,6 @@ impl BreezServicesBuilder {
             });
         }
 
-        let (reconnect_sender, reconnect_receiver) = watch::channel(());
         // The storage is implemented via sqlite.
         let persister = self
             .persister
@@ -2470,20 +2406,6 @@ impl BreezServicesBuilder {
             }
         });
 
-        // Reconnect breez server on when requested.
-        let cloned_breez_server = breez_server.clone();
-        let mut cloned_reconnect_receiver = reconnect_receiver.clone();
-        tokio::spawn(async move {
-            loop {
-                if cloned_reconnect_receiver.changed().await.is_err() {
-                    return;
-                }
-
-                debug!("Reconnect: reconnecting breez server");
-                let _ = cloned_breez_server.reconnect().await;
-            }
-        });
-
         let current_lsp_id = persister.get_lsp_id()?;
         if current_lsp_id.is_none() && self.config.default_lsp_id.is_some() {
             persister.set_lsp(self.config.default_lsp_id.clone().unwrap(), None)?;
@@ -2574,8 +2496,6 @@ impl BreezServicesBuilder {
             backup_watcher: Arc::new(backup_watcher),
             shutdown_sender,
             shutdown_receiver,
-            reconnect_sender,
-            reconnect_receiver,
         });
 
         Ok(breez_services)
diff --git a/libs/sdk-core/src/greenlight/node_api.rs b/libs/sdk-core/src/greenlight/node_api.rs
index 5a923efab..e2905a864 100644
--- a/libs/sdk-core/src/greenlight/node_api.rs
+++ b/libs/sdk-core/src/greenlight/node_api.rs
@@ -67,7 +67,6 @@ pub(crate) struct Greenlight {
     sdk_config: Config,
     signer: Mutex<Arc<Signer>>,
     device: Device,
-    seed: Vec<u8>,
     gl_client: Mutex<Option<node::Client>>,
     node_client: Mutex<Option<ClnClient>>,
     persister: Arc<SqliteStorage>,
@@ -182,7 +181,6 @@ impl Greenlight {
             sdk_config,
             signer: Mutex::new(Arc::new(signer)),
             device,
-            seed,
             gl_client: Mutex::new(None),
             node_client: Mutex::new(None),
             persister,
@@ -1002,32 +1000,6 @@ struct SyncState {
 
 #[tonic::async_trait]
 impl NodeAPI for Greenlight {
-    async fn reconnect(&self) {
-        debug!("Reconnect: request received");
-
-        // Force refresh existing grpc clients
-        *self.gl_client.lock().await = None;
-        *self.node_client.lock().await = None;
-
-        // Create a new signer
-        debug!("Reconnect: creating new signer");
-        let new_signer = match Signer::new(
-            self.seed.clone(),
-            self.sdk_config.network.into(),
-            self.device.clone(),
-        ) {
-            Ok(new_signer) => new_signer,
-            Err(e) => {
-                error!(
-                    "Reconnect: failed to create new signer after reconnect request: {:?}",
-                    e
-                );
-                return;
-            }
-        };
-        *self.signer.lock().await = Arc::new(new_signer);
-    }
-
     async fn node_credentials(&self) -> NodeResult<Option<NodeCredentials>> {
         Ok(Self::get_node_credentials(
             self.sdk_config.network,
diff --git a/libs/sdk-core/src/node_api.rs b/libs/sdk-core/src/node_api.rs
index f7a3e1e00..05dba8306 100644
--- a/libs/sdk-core/src/node_api.rs
+++ b/libs/sdk-core/src/node_api.rs
@@ -114,7 +114,6 @@ pub struct FetchBolt11Result {
 /// Trait covering functions affecting the LN node
 #[tonic::async_trait]
 pub trait NodeAPI: Send + Sync {
-    async fn reconnect(&self);
     async fn node_credentials(&self) -> NodeResult<Option<NodeCredentials>>;
     async fn configure_node(&self, close_to_address: Option<String>) -> NodeResult<()>;
     async fn create_invoice(&self, request: CreateInvoiceRequest) -> NodeResult<String>;


Test patch:
diff --git a/libs/sdk-core/src/test_utils.rs b/libs/sdk-core/src/test_utils.rs
index 201654529..d237ba7ad 100644
--- a/libs/sdk-core/src/test_utils.rs
+++ b/libs/sdk-core/src/test_utils.rs
@@ -330,7 +330,6 @@ pub struct MockNodeAPI {
 
 #[tonic::async_trait]
 impl NodeAPI for MockNodeAPI {
-    async fn reconnect(&self) {}
     async fn node_credentials(&self) -> NodeResult<Option<NodeCredentials>> {
         Err(NodeError::Generic("Not implemented".to_string()))
     }