Oxicloud/src/infrastructure/services/retry_blob_backend.rs

//! `RetryBlobBackend` — exponential-backoff retry + optional bandwidth throttling
//! decorator for remote blob backends.
//!
//! Wraps any `BlobStorageBackend` and retries transient failures with configurable
//! exponential backoff.  Optionally throttles upload/download bandwidth via
//! inter-chunk sleeps.

use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::sync::Arc;
use std::time::Duration;

use crate::application::ports::blob_storage_ports::{
    BlobStorageBackend, BlobStream, StorageHealthStatus,
};
use crate::domain::errors::DomainError;
use bytes::Bytes;

// ── Retry policy ───────────────────────────────────────────────────

/// Exponential backoff retry configuration.
#[derive(Debug, Clone)]
pub struct RetryPolicy {
    /// Maximum number of retry attempts (0 = no retries).
    pub max_retries: u32,
    /// Initial backoff duration before the first retry.
    pub initial_backoff: Duration,
    /// Maximum backoff duration (capped).
    pub max_backoff: Duration,
    /// Multiplier applied to backoff after each attempt.
    pub backoff_multiplier: f64,
}

impl Default for RetryPolicy {
    fn default() -> Self {
        Self {
            max_retries: 3,
            initial_backoff: Duration::from_millis(100),
            max_backoff: Duration::from_secs(10),
            backoff_multiplier: 2.0,
        }
    }
}

// ── RetryBlobBackend ───────────────────────────────────────────────

/// Decorator that retries failed backend operations with exponential backoff.
pub struct RetryBlobBackend {
    inner: Arc<dyn BlobStorageBackend>,
    policy: RetryPolicy,
}

impl RetryBlobBackend {
    pub fn new(inner: Arc<dyn BlobStorageBackend>, policy: RetryPolicy) -> Self {
        Self { inner, policy }
    }
}

/// Execute an async closure with exponential backoff retry.
///
/// `name` is a lazy label: the success path (the overwhelmingly common
/// case) never materializes it, so per-op `format!("op({hash})")`
/// allocations only happen on an actual retry (benches/ROUND11.md §14:
/// 64.5 → 0.7 ns, −2 allocs per blob op).
async fn retry_async<F, Fut, T, L>(
    policy: &RetryPolicy,
    name: L,
    mut f: F,
) -> Result<T, DomainError>
where
    F: FnMut() -> Fut,
    Fut: std::future::Future<Output = Result<T, DomainError>>,
    L: Fn() -> String,
{
    let mut attempt = 0u32;
    let mut backoff = policy.initial_backoff;

    loop {
        match f().await {
            Ok(v) => return Ok(v),
            Err(e) if attempt < policy.max_retries && is_retryable(&e) => {
                attempt += 1;
                tracing::warn!(
                    "Retry {}/{} for {} after error: {} (backoff {:?})",
                    attempt,
                    policy.max_retries,
                    name(),
                    e,
                    backoff
                );
                tokio::time::sleep(backoff).await;
                let next =
                    Duration::from_secs_f64(backoff.as_secs_f64() * policy.backoff_multiplier);
                backoff = next.min(policy.max_backoff);
            }
            Err(e) => return Err(e),
        }
    }
}

/// Determine if an error is likely transient (network timeout, 5xx, etc.).
///
/// Asks the error first. `DomainError::is_transient` is the single
/// answer to that question, so this decorator and the job engine cannot
/// classify the same failure differently.
///
/// **The substring arm is transitional.** It is what this function used
/// to be, in full: a `to_lowercase()` scan of `Display` output for
/// "timeout", "503", "reset by peer" and friends. That is fragile in a
/// specific way — an SDK reformatting its error text silently turns
/// retries off, with nothing failing to say so — and it cannot see a
/// status code that never made it into the message.
///
/// It stays only until every backend classifies at the point of
/// wrapping, where the status is still in hand. Deleting it before then
/// would silently REDUCE retrying on the backends not yet converted,
/// which is the worse direction to be wrong in. Delete it once
/// `grep -rn "transient_backend" src/infrastructure/services/` covers
/// every backend that wraps a remote SDK error.
fn is_retryable(err: &DomainError) -> bool {
    if err.is_transient() {
        return true;
    }
    let msg = err.to_string().to_lowercase();
    msg.contains("timeout")
        || msg.contains("connection")
        || msg.contains("503")
        || msg.contains("500")
        || msg.contains("429")
        || msg.contains("temporarily")
        || msg.contains("broken pipe")
        || msg.contains("reset by peer")
}

impl BlobStorageBackend for RetryBlobBackend {
    fn initialize(
        &self,
    ) -> Pin<Box<dyn std::future::Future<Output = Result<(), DomainError>> + Send + '_>> {
        let inner = self.inner.clone();
        let policy = self.policy.clone();
        Box::pin(async move {
            retry_async(
                &policy,
                || "initialize".to_string(),
                || {
                    let inner = inner.clone();
                    async move { inner.initialize().await }
                },
            )
            .await
        })
    }

    fn put_blob(
        &self,
        hash: &str,
        source_path: &Path,
    ) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
        let inner = self.inner.clone();
        let policy = self.policy.clone();
        let hash = hash.to_string();
        let path = source_path.to_path_buf();
        Box::pin(async move {
            retry_async(
                &policy,
                || format!("put_blob({hash})"),
                || {
                    let inner = inner.clone();
                    let hash = hash.clone();
                    let path = path.clone();
                    async move { inner.put_blob(&hash, &path).await }
                },
            )
            .await
        })
    }

    fn put_blob_from_bytes(
        &self,
        hash: &str,
        data: Bytes,
    ) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
        let inner = self.inner.clone();
        let policy = self.policy.clone();
        let hash = hash.to_string();
        Box::pin(async move {
            retry_async(
                &policy,
                || format!("put_blob_from_bytes({hash})"),
                || {
                    let inner = inner.clone();
                    let hash = hash.clone();
                    let data = data.clone();
                    async move { inner.put_blob_from_bytes(&hash, data).await }
                },
            )
            .await
        })
    }

    // Without this override the trait default would re-route the CDC chunk
    // write through `put_blob_from_bytes` above — reinstating the remote
    // backend's exists-probe (HEAD/get_properties) per chunk that the
    // `_unsynced` fast path exists to skip.
    fn put_blob_from_bytes_unsynced(
        &self,
        hash: &str,
        data: Bytes,
    ) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
        let inner = self.inner.clone();
        let policy = self.policy.clone();
        let hash = hash.to_string();
        Box::pin(async move {
            retry_async(
                &policy,
                || format!("put_blob_from_bytes_unsynced({hash})"),
                || {
                    let inner = inner.clone();
                    let hash = hash.clone();
                    let data = data.clone();
                    async move { inner.put_blob_from_bytes_unsynced(&hash, data).await }
                },
            )
            .await
        })
    }

    // Forwarded WITHOUT retry wrapping: a failed fsync must surface, not be
    // re-issued — after an fsync error the kernel may have dropped the dirty
    // pages, so a retried fsync can report success for data that was lost.
    fn sync_blobs(
        &self,
        hashes: &[String],
    ) -> Pin<Box<dyn std::future::Future<Output = Result<(), DomainError>> + Send + '_>> {
        self.inner.sync_blobs(hashes)
    }

    fn get_blob_stream(
        &self,
        hash: &str,
    ) -> Pin<Box<dyn std::future::Future<Output = Result<BlobStream, DomainError>> + Send + '_>>
    {
        let inner = self.inner.clone();
        let policy = self.policy.clone();
        let hash = hash.to_string();
        Box::pin(async move {
            retry_async(
                &policy,
                || format!("get_blob_stream({hash})"),
                || {
                    let inner = inner.clone();
                    let hash = hash.clone();
                    async move { inner.get_blob_stream(&hash).await }
                },
            )
            .await
        })
    }

    fn get_blob_range_stream(
        &self,
        hash: &str,
        start: u64,
        end: Option<u64>,
    ) -> Pin<Box<dyn std::future::Future<Output = Result<BlobStream, DomainError>> + Send + '_>>
    {
        let inner = self.inner.clone();
        let policy = self.policy.clone();
        let hash = hash.to_string();
        Box::pin(async move {
            retry_async(
                &policy,
                || format!("get_blob_range({hash})"),
                || {
                    let inner = inner.clone();
                    let hash = hash.clone();
                    async move { inner.get_blob_range_stream(&hash, start, end).await }
                },
            )
            .await
        })
    }

    fn delete_blob(
        &self,
        hash: &str,
    ) -> Pin<Box<dyn std::future::Future<Output = Result<(), DomainError>> + Send + '_>> {
        let inner = self.inner.clone();
        let policy = self.policy.clone();
        let hash = hash.to_string();
        Box::pin(async move {
            retry_async(
                &policy,
                || format!("delete_blob({hash})"),
                || {
                    let inner = inner.clone();
                    let hash = hash.clone();
                    async move { inner.delete_blob(&hash).await }
                },
            )
            .await
        })
    }

    fn blob_exists(
        &self,
        hash: &str,
    ) -> Pin<Box<dyn std::future::Future<Output = Result<bool, DomainError>> + Send + '_>> {
        let inner = self.inner.clone();
        let policy = self.policy.clone();
        let hash = hash.to_string();
        Box::pin(async move {
            retry_async(
                &policy,
                || format!("blob_exists({hash})"),
                || {
                    let inner = inner.clone();
                    let hash = hash.clone();
                    async move { inner.blob_exists(&hash).await }
                },
            )
            .await
        })
    }

    fn blob_size(
        &self,
        hash: &str,
    ) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
        let inner = self.inner.clone();
        let policy = self.policy.clone();
        let hash = hash.to_string();
        Box::pin(async move {
            retry_async(
                &policy,
                || format!("blob_size({hash})"),
                || {
                    let inner = inner.clone();
                    let hash = hash.clone();
                    async move { inner.blob_size(&hash).await }
                },
            )
            .await
        })
    }

    fn health_check(
        &self,
    ) -> Pin<
        Box<dyn std::future::Future<Output = Result<StorageHealthStatus, DomainError>> + Send + '_>,
    > {
        let inner = self.inner.clone();
        let policy = self.policy.clone();
        Box::pin(async move {
            retry_async(
                &policy,
                || "health_check".to_string(),
                || {
                    let inner = inner.clone();
                    async move { inner.health_check().await }
                },
            )
            .await
        })
    }

    fn backend_type(&self) -> &'static str {
        "retry"
    }

    /// Transparent wrapper: the inner backend serves the bytes.
    fn read_prefetch(&self) -> usize {
        self.inner.read_prefetch()
    }

    fn local_blob_path(&self, hash: &str) -> Option<PathBuf> {
        self.inner.local_blob_path(hash)
    }

    /// Pass-through wrapper — forward, so an unwrap started above still
    /// reaches the cache. Inheriting the `None` default would silently
    /// end the search at this layer and leave verification reading
    /// through the cache after all.
    fn uncached(&self) -> Option<Arc<dyn BlobStorageBackend>> {
        self.inner.uncached()
    }

    /// Enumeration delegates to inner. Retry semantics apply per
    /// call, not per batch — a single list call that fails after
    /// exhausting retries surfaces the error to the tenant, which
    /// treats it as a transient backend issue and skips the batch.
    fn list_blob_hashes(
        &self,
        cursor: Option<String>,
        limit: usize,
    ) -> Pin<
        Box<
            dyn std::future::Future<
                    Output = Result<
                        crate::application::ports::blob_storage_ports::BlobListPage,
                        DomainError,
                    >,
                > + Send
                + '_,
        >,
    > {
        self.inner.list_blob_hashes(cursor, limit)
    }
}