Abhi/drone connect retry (#466)

* add request with retry to nats wrapper

* use request-with-reply in drone connection request

* also delay for interval

this may in some cases make the thing wait for 2x interval, but shouldn't matter too much

core/src/nats.rs

@@ -4,6 +4,7 @@
 
 use crate::logging::LogError;
 use crate::messages::initialize_jetstreams;
+use crate::retry::do_with_retry;
 use anyhow::{anyhow, Context, Result};
 use async_nats::jetstream;
 use async_nats::jetstream::consumer::push::Messages;
@@ -510,6 +511,31 @@ impl TypedNats {
         Ok(value)
     }
 
+    pub async fn request_with_retries<T: TypedMessage>(
+        &self,
+        value: &T,
+        retries: u16,
+        interval: Duration,
+    ) -> Result<T::Response> {
+        let bytes = &Bytes::from(serde_json::to_vec(value)?);
+
+        let result = do_with_retry(
+            || {
+                let req = async_nats::Request::new()
+                    .payload(bytes.clone())
+                    .timeout(Some(interval));
+                let fut = self.nc.send_request(value.subject(), req);
+                fut
+            },
+            retries,
+            interval,
+        )
+        .await?;
+
+        let value: T::Response = serde_json::from_slice(&result.payload)?;
+        Ok(value)
+    }
+
     pub async fn subscribe<T>(&self, subject: SubscribeSubject<T>) -> Result<TypedSubscription<T>>
     where
         T: TypedMessage,

drone/src/agent/mod.rs

@@ -200,7 +200,8 @@ pub async fn run_agent(agent_opts: AgentOptions) -> Result<AgentSupervisors> {
         git_hash: GIT_HASH.map(|s| s.to_string()),
     };
 
-    nats.request(&request).await?;
+    nats.request_with_retries(&request, 10, Duration::from_secs(5))
+        .await?;
 
     let executor = Executor::new(docker, db.clone(), nats.clone(), ip, cluster.clone());