Merge pull request #160 from code0-tech/#158-retry-sending-definitons

retry sending definitons

Author: raphael-goetz

Cargo.lock

@@ -8,7 +8,7 @@ edition = "2024"
 
 [workspace.dependencies]
 async-trait = "0.1.89"
-code0-flow = { version = "0.0.30" }
+code0-flow = { version = "0.0.31" }
 tucana = { version = "0.0.68" }
 tokio = { version = "1.44.1", features = ["rt-multi-thread", "signal"] }
 log = "0.4.27"

Cargo.toml

@@ -14,12 +14,13 @@ use futures_lite::StreamExt;
 use log::error;
 use prost::Message;
 use std::collections::HashMap;
-use std::time::Instant;
+use std::time::{Duration, Instant};
 use taurus_core::context::context::Context;
 use taurus_core::context::executor::Executor;
 use taurus_core::context::registry::FunctionStore;
 use taurus_core::context::signal::Signal;
 use tokio::signal;
+use tokio::time::sleep;
 use tonic_health::pb::health_server::HealthServer;
 use tucana::shared::value::Kind;
 use tucana::shared::{
@@ -111,17 +112,30 @@ async fn main() {
         None
     };
 
-    // Optional: dynamic mode sync at startup
     if config.mode == DYNAMIC {
-        FlowUpdateService::from_url(
+        let definition_service = FlowUpdateService::from_url(
             config.aquila_url.clone(),
             config.definitions.clone().as_str(),
             config.aquila_token.clone(),
         )
-        .await
-        .send()
         .await;
 
+        let mut success = false;
+        let mut count = 1;
+        while !success {
+            success = definition_service.send_with_status().await;
+            if success {
+                break;
+            }
+
+            log::warn!(
+                "Updating definitions failed, trying again in 2 secs (retry number {})",
+                count
+            );
+            count += 1;
+            sleep(Duration::from_secs(3)).await;
+        }
+
         let usage_service = TaurusRuntimeUsageService::from_url(
             config.aquila_url.clone(),
             config.aquila_token.clone(),