Implement auto retries under a config feature flag

Signed-off-by: Mateusz Szczygieł <mateusz.szczygiel@nordsec.com>

changelog.md

@@ -5,6 +5,7 @@
 * Update moose tracker to v13.1.0
     * Adds the `transfer_intent_received` event
 * Added private key retrieval on demand instead of caching it
+* Add internal retries and put under a config feature `auto_retry_interval_ms`
 
 ---
 <br>

drop-config/src/lib.rs

@@ -17,6 +17,7 @@ pub struct DropConfig {
     // Default value is 256KB.
     pub checksum_events_granularity: u64,
     pub connection_retries: u32,
+    pub auto_retry_interval: Option<Duration>,
 }
 
 impl Default for DropConfig {
@@ -28,6 +29,7 @@ impl Default for DropConfig {
             checksum_events_size_threshold: None,
             checksum_events_granularity: 256 * 1024,
             connection_retries: 5,
+            auto_retry_interval: None,
         }
     }
 }

drop-transfer/src/service.rs

@@ -3,23 +3,26 @@ use std::{
     net::IpAddr,
     path::{Component, Path},
     sync::Arc,
-    time::{Instant, SystemTime},
+    time::{Duration, Instant, SystemTime},
 };
 
 use drop_analytics::{InitEventData, Moose, TransferStateEventData};
 use drop_config::DropConfig;
 use drop_core::Status;
 use drop_storage::Storage;
-use slog::{debug, trace, Logger};
-use tokio::sync::{mpsc, Semaphore};
+use slog::{debug, info, trace, Logger};
+use tokio::{
+    sync::{mpsc, Semaphore},
+    time::MissedTickBehavior,
+};
 use tokio_util::sync::CancellationToken;
 use uuid::Uuid;
 
 use crate::{
     auth,
     error::ResultExt,
     manager::{self},
-    tasks::AliveWaiter,
+    tasks::{AliveGuard, AliveWaiter},
     transfer::Transfer,
     ws::{self, EventTxFactory},
     Error, Event, FileId, TransferManager,
@@ -106,12 +109,22 @@ impl Service {
 
             manager::resume(&refresh_trigger.subscribe(), &state, &logger, &guard, &stop).await;
 
+            if let Some(interval) = state.config.auto_retry_interval {
+                spawn_auto_retry_loop(
+                    refresh_trigger.clone(),
+                    interval,
+                    logger.clone(),
+                    guard.clone(),
+                    stop.clone(),
+                );
+            }
+
             Ok(Self {
                 refresh_trigger,
                 state,
                 stop,
                 waiter,
-                logger: logger.clone(),
+                logger,
             })
         };
 
@@ -335,3 +348,40 @@ fn validate_dest_path(parent_dir: &Path) -> crate::Result<()> {
 
     Ok(())
 }
+
+fn spawn_auto_retry_loop(
+    trigger: tokio::sync::watch::Sender<()>,
+    interval: Duration,
+    logger: Logger,
+    guard: AliveGuard,
+    stop: CancellationToken,
+) {
+    info!(
+        logger,
+        "Starting auto retry loop with interval: {}ms",
+        interval.as_millis()
+    );
+
+    let mut interval = tokio::time::interval(interval);
+    interval.set_missed_tick_behavior(MissedTickBehavior::Delay);
+
+    tokio::spawn(async move {
+        let _guard = guard;
+
+        let task = async {
+            loop {
+                interval.tick().await;
+                let _ = trigger.send(());
+            }
+        };
+
+        tokio::select! {
+            biased;
+
+            _ = stop.cancelled() => {
+                debug!(logger, "Stopping auto retry loop");
+            },
+            _ = task => (),
+        }
+    });
+}

norddrop/src/config.rs

@@ -1,3 +1,5 @@
+use std::time::Duration;
+
 #[derive(Debug)]
 pub struct Config {
     pub dir_depth_limit: u64,
@@ -8,6 +10,7 @@ pub struct Config {
     pub checksum_events_size_threshold: Option<u64>,
     pub checksum_events_granularity: Option<u64>,
     pub connection_retries: Option<u32>,
+    pub auto_retry_interval_ms: Option<u32>,
 }
 
 impl Config {
@@ -31,6 +34,7 @@ impl From<Config> for drop_config::Config {
             checksum_events_size_threshold,
             checksum_events_granularity,
             connection_retries,
+            auto_retry_interval_ms,
         } = val;
 
         drop_config::Config {
@@ -43,6 +47,8 @@ impl From<Config> for drop_config::Config {
                     .unwrap_or(Config::default_checksum_granularity() as _),
                 connection_retries: connection_retries
                     .unwrap_or(Config::default_connection_retries()),
+                auto_retry_interval: auto_retry_interval_ms
+                    .map(|ms| Duration::from_millis(ms as _)),
             },
             moose: drop_config::MooseConfig {
                 event_path: moose_event_path,

norddrop/src/norddrop.udl

@@ -60,6 +60,16 @@ dictionary Config {
 
     /// Limits the number of connection retries afer the `network_refresh()` call.
     u32? connection_retries;
+
+    /// Enable auto retry loop inside libdrop. Conceptually this means that
+    /// libdrop is calling `network_refresh()` automatically with the given
+    /// period in milliseconds. When set to `null` the feature is disabled
+    /// and the application needs to call  `network_refresh()` manually.
+    /// Note the setting `connection_retries` still applies, meaning the retry
+    /// is executed in burst with this number of count.
+    /// For example for a single retry every 5 seconds the application needs to
+    /// set `connection_retries = 1` and `auto_retry_interval_ms = 5000`.
+    u32? auto_retry_interval_ms;
 };
 
 /// Posible log levels.

test/drop_test/action.py

@@ -974,18 +974,21 @@ def __init__(
         dbpath: str = ":memory:",
         checksum_events_size_threshold=2**32,  # don't emit events for existing tests
         checksum_events_granularity=None,
+        auto_retry_interval_ms=None,
     ):
         self._addr = addr
         self._dbpath = dbpath
         self._checksum_events_size_threshold = checksum_events_size_threshold
         self._checksum_events_granularity = checksum_events_granularity
+        self._auto_retry_interval_ms = auto_retry_interval_ms
 
     async def run(self, drop: ffi.Drop):
         drop.start(
             peer_resolver.resolve(self._addr),
             self._dbpath,
             self._checksum_events_size_threshold,
             self._checksum_events_granularity,
+            self._auto_retry_interval_ms,
         )
 
     def __str__(self):

test/drop_test/ffi.py

@@ -322,6 +322,7 @@ def start(
         dbpath: str,
         checksum_events_size_threshold=None,
         checksum_events_granularity=None,
+        auto_retry_interval_ms=None,
     ):
         cfg = norddrop.Config(
             dir_depth_limit=5,
@@ -332,6 +333,7 @@ def start(
             checksum_events_size_threshold=checksum_events_size_threshold,
             checksum_events_granularity=checksum_events_granularity,
             connection_retries=1,
+            auto_retry_interval_ms=auto_retry_interval_ms,
         )
 
         self._instance.start(addr, cfg)