node: Create / upsert files with retries

Author: MaxFangX

node/src/api/client.rs

@@ -17,15 +17,22 @@ use serde::Serialize;
 use tokio::time;
 use tracing::debug;
 
+use self::ApiVersion::*;
+use self::BaseUrl::*;
 use crate::api::*;
 
 const API_REQUEST_TIMEOUT: Duration = Duration::from_secs(5);
-const MAX_ATTEMPTS: usize = 3;
 /// How long to wait after the second failed API request before trying again.
 /// This is relatively long since if the second try failed, it probably means
 /// that the backend is down, which could be the case for a while.
 const RETRY_INTERVAL: Duration = Duration::from_secs(15);
 
+// Avoid `Method::` prefix. Associated constants can't be imported
+const GET: Method = Method::GET;
+const PUT: Method = Method::PUT;
+const POST: Method = Method::POST;
+const DELETE: Method = Method::DELETE;
+
 /// Enumerates the base urls that can be used in an API call.
 #[derive(Copy, Clone)]
 enum BaseUrl {
@@ -38,9 +45,6 @@ enum ApiVersion {
     V1,
 }
 
-use ApiVersion::*;
-use BaseUrl::*;
-
 impl Display for ApiVersion {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match &self {
@@ -76,8 +80,7 @@ impl ApiClient for LexeApiClient {
         user_pk: UserPk,
     ) -> Result<Option<Node>, ApiError> {
         let data = GetByUserPk { user_pk };
-        self.request(&Method::GET, Backend, V1, "/node", &data)
-            .await
+        self.request(&GET, Backend, V1, "/node", &data).await
     }
 
     async fn get_instance(
@@ -90,8 +93,7 @@ impl ApiClient for LexeApiClient {
             measurement,
         };
         let maybe_instance: Option<Instance> = self
-            .request(&Method::GET, Backend, V1, "/instance", &data)
-            .await?;
+            .request(&GET, Backend, V1, "/instance", &data).await?;
 
         if let Some(instance) = maybe_instance.as_ref() {
             if instance.measurement != measurement {
@@ -111,34 +113,49 @@ impl ApiClient for LexeApiClient {
         &self,
         data: SealedSeedId,
     ) -> Result<Option<SealedSeed>, ApiError> {
-        self.request(&Method::GET, Backend, V1, "/sealed_seed", &data)
-            .await
+        self.request(&GET, Backend, V1, "/sealed_seed", &data).await
     }
 
     async fn create_node_instance_seed(
         &self,
         data: NodeInstanceSeed,
     ) -> Result<NodeInstanceSeed, ApiError> {
         let endpoint = "/acid/node_instance_seed";
-        self.request(&Method::POST, Backend, V1, endpoint, &data)
-            .await
+        self.request(&POST, Backend, V1, endpoint, &data).await
     }
 
     async fn get_file(&self, data: &FileId) -> Result<Option<File>, ApiError> {
         let endpoint = "/file";
-        self.request(&Method::GET, Backend, V1, endpoint, &data)
-            .await
+        self.request(&GET, Backend, V1, endpoint, &data).await
     }
 
     async fn create_file(&self, data: &File) -> Result<File, ApiError> {
         let endpoint = "/file";
-        self.request(&Method::POST, Backend, V1, endpoint, &data)
+        self.request(&POST, Backend, V1, endpoint, &data).await
+    }
+
+    async fn create_file_with_retries(
+        &self,
+        data: &File,
+        retries: usize,
+    ) -> Result<File, ApiError> {
+        let endpoint = "/file";
+        self.request_with_retries(&POST, Backend, V1, endpoint, &data, retries)
             .await
     }
 
     async fn upsert_file(&self, data: &File) -> Result<File, ApiError> {
         let endpoint = "/file";
-        self.request(&Method::PUT, Backend, V1, endpoint, &data)
+        self.request(&PUT, Backend, V1, endpoint, &data).await
+    }
+
+    async fn upsert_file_with_retries(
+        &self,
+        data: &File,
+        retries: usize,
+    ) -> Result<File, ApiError> {
+        let endpoint = "/file";
+        self.request_with_retries(&PUT, Backend, V1, endpoint, &data, retries)
             .await
     }
 
@@ -147,46 +164,47 @@ impl ApiClient for LexeApiClient {
     #[allow(dead_code)]
     async fn delete_file(&self, data: &FileId) -> Result<String, ApiError> {
         let endpoint = "/file";
-        self.request(&Method::DELETE, Backend, V1, endpoint, &data)
-            .await
+        self.request(&DELETE, Backend, V1, endpoint, &data).await
     }
 
     async fn get_directory(
         &self,
         data: &Directory,
     ) -> Result<Vec<File>, ApiError> {
         let endpoint = "/directory";
-        self.request(&Method::GET, Backend, V1, endpoint, &data)
-            .await
+        self.request(&GET, Backend, V1, endpoint, &data).await
     }
 
     async fn notify_runner(
         &self,
         data: UserPorts,
     ) -> Result<UserPorts, ApiError> {
-        self.request(&Method::POST, Runner, V1, "/ready", &data)
-            .await
+        self.request(&POST, Runner, V1, "/ready", &data).await
     }
 }
 
 impl LexeApiClient {
-    /// Tries to complete an API request, making up to `MAX_ATTEMPTS` attempts.
-    async fn request<D: Serialize, T: DeserializeOwned>(
+    /// Tries to complete an API request, retrying up to `retries` times.
+    async fn request_with_retries<D: Serialize, T: DeserializeOwned>(
         &self,
         method: &Method,
         base: BaseUrl,
         ver: ApiVersion,
         endpoint: &str,
         data: &D,
+        retries: usize,
     ) -> Result<T, ApiError> {
         let mut retry_timer = time::interval(RETRY_INTERVAL);
 
-        // Try the first n-1 times, return early if successful
-        for _ in 0..MAX_ATTEMPTS - 1 {
-            let res = self.execute(method, base, ver, endpoint, data).await;
+        // 'Do the retries first' and return early if successful.
+        // If retries == 0 this block is a noop.
+        for _ in 0..retries {
+            let res = self.request(method, base, ver, endpoint, data).await;
             if res.is_ok() {
                 return res;
             } else {
+                // TODO log errors here
+
                 // Since the first tick resolves immediately, and we tick only
                 // on failures, the first failed attempt is immediately followed
                 // up with second attempt (to encode that sometimes messages are
@@ -197,12 +215,12 @@ impl LexeApiClient {
             }
         }
 
-        // Last try
-        self.execute(method, base, ver, endpoint, &data).await
+        // Do the 'main attempt'.
+        self.request(method, base, ver, endpoint, &data).await
     }
 
     /// Executes an API request once.
-    async fn execute<D: Serialize, T: DeserializeOwned>(
+    async fn request<D: Serialize, T: DeserializeOwned>(
         &self,
         method: &Method,
         base: BaseUrl,
@@ -220,7 +238,7 @@ impl LexeApiClient {
         // If GET, serialize the data in a query string
         let method = method.to_owned();
         let query_str = match method {
-            Method::GET => Some(serde_qs::to_string(data)?),
+            GET => Some(serde_qs::to_string(data)?),
             _ => None,
         };
         // Append directly to url since RequestBuilder.param() API is unwieldy
@@ -234,7 +252,7 @@ impl LexeApiClient {
 
         // If PUT or POST, serialize the data in the request body
         let body = match method {
-            Method::PUT | Method::POST => serde_json::to_string(data)?,
+            PUT | POST => serde_json::to_string(data)?,
             _ => String::new(),
         };
         // println!("    Body: {}", body);

node/src/api/mock.rs

@@ -165,11 +165,27 @@ impl ApiClient for MockApiClient {
         Ok(file.clone())
     }
 
+    async fn create_file_with_retries(
+        &self,
+        file: &File,
+        _retries: usize,
+    ) -> Result<File, ApiError> {
+        self.create_file(file).await
+    }
+
     async fn upsert_file(&self, file: &File) -> Result<File, ApiError> {
         self.vfs.lock().unwrap().insert(file.clone());
         Ok(file.clone())
     }
 
+    async fn upsert_file_with_retries(
+        &self,
+        file: &File,
+        _retries: usize,
+    ) -> Result<File, ApiError> {
+        self.upsert_file(file).await
+    }
+
     /// Returns "OK" if exactly one row was deleted.
     async fn delete_file(&self, file_id: &FileId) -> Result<String, ApiError> {
         let file_opt = self.vfs.lock().unwrap().remove(file_id.clone());

node/src/api/mod.rs

@@ -61,8 +61,20 @@ pub trait ApiClient {
 
     async fn create_file(&self, file: &File) -> Result<File, ApiError>;
 
+    async fn create_file_with_retries(
+        &self,
+        file: &File,
+        retries: usize,
+    ) -> Result<File, ApiError>;
+
     async fn upsert_file(&self, file: &File) -> Result<File, ApiError>;
 
+    async fn upsert_file_with_retries(
+        &self,
+        file: &File,
+        retries: usize,
+    ) -> Result<File, ApiError>;
+
     // TODO We want to delete channel peers / monitors when channels close
     /// Returns "OK" if exactly one row was deleted.
     async fn delete_file(&self, file_id: &FileId) -> Result<String, ApiError>;

node/src/lexe/persister.rs

@@ -42,6 +42,9 @@ const SCORER_FILENAME: &str = "scorer";
 pub const CHANNEL_PEERS_DIRECTORY: &str = "channel_peers";
 pub const CHANNEL_MONITORS_DIRECTORY: &str = "channel_monitors";
 
+/// The default number of retries for important persisted state
+const DEFAULT_RETRIES: usize = 3;
+
 /// An Arc is held internally, so it is fine to clone and use directly.
 #[derive(Clone)] // TODO Try removing this
 pub struct LexePersister {
@@ -318,8 +321,9 @@ impl InnerPersister {
             Vec::new(),
         );
 
+        // Retry up to 3 times
         self.api
-            .create_file(&cp_file)
+            .create_file_with_retries(&cp_file, DEFAULT_RETRIES)
             .await
             .map(|_| ())
             .map_err(|e| e.into())
@@ -342,8 +346,9 @@ impl InnerPersister {
             data,
         );
 
+        // Channel manager is more important so let's retry up to three times
         self.api
-            .upsert_file(&cm_file)
+            .upsert_file_with_retries(&cm_file, DEFAULT_RETRIES)
             .await
             .map(|_| ())
             .context("Could not persist channel manager")
@@ -432,21 +437,14 @@ impl Persist<SignerType> for InnerPersister {
             self.channel_monitor_updated_tx.clone();
         tokio::spawn(async move {
             // Retry indefinitely until it succeeds
-            loop {
-                // TODO Also attempt to persist to cloud backup
-                match api_clone.create_file(&cm_file).await {
-                    Ok(_file) => {
-                        if let Err(e) =
-                            channel_monitor_updated_tx.try_send(update)
-                        {
-                            error!("Couldn't notify chain monitor: {:#}", e);
-                        }
-                        return;
-                    }
-                    Err(e) => {
-                        error!("Couldn't persist new channel monitor: {:#}", e)
-                    }
-                }
+            // TODO Also attempt to persist to cloud backup
+            api_clone
+                .create_file_with_retries(&cm_file, usize::MAX)
+                .await
+                .expect("Unlimited retries always return Ok");
+
+            if let Err(e) = channel_monitor_updated_tx.try_send(update) {
+                error!("Couldn't notify chain monitor: {:#}", e);
             }
         });
 
@@ -487,21 +485,14 @@ impl Persist<SignerType> for InnerPersister {
             self.channel_monitor_updated_tx.clone();
         tokio::spawn(async move {
             // Retry indefinitely until it succeeds
-            loop {
-                // TODO Also attempt to persist to cloud backup
-                match api_clone.upsert_file(&cm_file).await {
-                    Ok(_) => {
-                        if let Err(e) =
-                            channel_monitor_updated_tx.try_send(update)
-                        {
-                            error!("Couldn't notify chain monitor: {:#}", e);
-                        }
-                        return;
-                    }
-                    Err(e) => {
-                        error!("Could not update channel monitor: {:#}", e)
-                    }
-                }
+            // TODO Also attempt to persist to cloud backup
+            api_clone
+                .upsert_file_with_retries(&cm_file, usize::MAX)
+                .await
+                .expect("Unlimited retries always return Ok");
+
+            if let Err(e) = channel_monitor_updated_tx.try_send(update) {
+                error!("Couldn't notify chain monitor: {:#}", e);
             }
         });