feat(backends): Add throttle option to opendal backend

Author: aawsome

crates/backend/Cargo.toml

@@ -82,16 +82,17 @@ rand = { version = "0.8.5", optional = true }
 semver = { version = "1.0.22", optional = true }
 
 # opendal backend
+bytesize = "1.3.0"
 rayon = { version = "1.9.0", optional = true }
 tokio = { version = "1.36.0", optional = true, default-features = false }
 
 [target.'cfg(not(windows))'.dependencies]
 # opendal backend - sftp is not supported on windows, see https://github.com/apache/incubator-opendal/issues/2963
-opendal = { version = "0.45", features = ["services-b2", "services-sftp", "services-swift", "layers-blocking"], optional = true }
+opendal = { version = "0.45", features = ["services-b2", "services-sftp", "services-swift", "layers-blocking", "layers-throttle"], optional = true }
 
 [target.'cfg(windows)'.dependencies]
 # opendal backend
-opendal = { version = "0.45", features = ["services-b2", "services-swift", "layers-blocking"], optional = true }
+opendal = { version = "0.45", features = ["services-b2", "services-swift", "layers-blocking", "layers-throttle"], optional = true }
 
 [dev-dependencies]
 rstest = { workspace = true }

crates/backend/src/opendal.rs

@@ -1,11 +1,12 @@
 /// `OpenDAL` backend for rustic.
 use std::{collections::HashMap, path::PathBuf, str::FromStr, sync::OnceLock};
 
-use anyhow::Result;
+use anyhow::{anyhow, Error, Result};
 use bytes::Bytes;
+use bytesize::ByteSize;
 use log::trace;
 use opendal::{
-    layers::{BlockingLayer, ConcurrentLimitLayer, LoggingLayer, RetryLayer},
+    layers::{BlockingLayer, ConcurrentLimitLayer, LoggingLayer, RetryLayer, ThrottleLayer},
     BlockingOperator, ErrorKind, Metakey, Operator, Scheme,
 };
 use rayon::prelude::{IntoParallelIterator, ParallelIterator};
@@ -34,6 +35,31 @@ fn runtime() -> &'static Runtime {
     })
 }
 
+/// Throttling parameters
+///
+/// Note: Throttle implements FromStr to read it from something like "10kiB,10MB"
+#[derive(Debug, Clone, Copy)]
+pub struct Throttle {
+    bandwidth: u32,
+    burst: u32,
+}
+
+impl FromStr for Throttle {
+    type Err = Error;
+    fn from_str(s: &str) -> Result<Self> {
+        let mut values = s
+            .split(',')
+            .map(|s| ByteSize::from_str(s).map_err(|err| anyhow!("Error: {err}")))
+            .map(|b| -> Result<u32> { Ok(b?.as_u64().try_into()?) });
+        let bandwidth = values
+            .next()
+            .ok_or_else(|| anyhow!("no bandwidth given"))??;
+        let burst = values.next().ok_or_else(|| anyhow!("no burst given"))??;
+        let throttle = Throttle { bandwidth, burst };
+        Ok(throttle)
+    }
+}
+
 impl OpenDALBackend {
     /// Create a new openDAL backend.
     ///
@@ -60,10 +86,19 @@ impl OpenDALBackend {
             .map(|c| usize::from_str(c))
             .transpose()?;
 
+        let throttle = options
+            .get("throttle")
+            .map(|t| Throttle::from_str(t))
+            .transpose()?;
+
         let schema = Scheme::from_str(path.as_ref())?;
         let mut operator = Operator::via_map(schema, options)?
             .layer(RetryLayer::new().with_max_times(max_retries).with_jitter());
 
+        if let Some(Throttle { bandwidth, burst }) = throttle {
+            operator = operator.layer(ThrottleLayer::new(bandwidth, burst));
+        }
+
         if let Some(connections) = connections {
             operator = operator.layer(ConcurrentLimitLayer::new(connections));
         }