diff --git a/src/common/config.rs b/src/common/config.rs index c240fba1..f006fb99 100644 --- a/src/common/config.rs +++ b/src/common/config.rs @@ -2,6 +2,8 @@ use std::env; use std::path::PathBuf; use std::time::Duration; +use crate::infrastructure::services::timeout_blob_backend::TimeoutPolicy; + /// Cache configuration #[derive(Debug, Clone)] pub struct CacheConfig { @@ -268,6 +270,9 @@ pub struct StorageConfig { pub encryption: EncryptionConfig, /// Retry policy for remote backends. pub retry: RetryConfig, + /// Wall-clock bounds on backend calls, so a stalled endpoint + /// surfaces as a transient error instead of hanging indefinitely. + pub timeout: TimeoutPolicy, } /// Which blob storage backend to use. @@ -1295,6 +1300,7 @@ impl Default for StorageConfig { cache: BlobCacheConfig::default(), encryption: EncryptionConfig::default(), retry: RetryConfig::default(), + timeout: TimeoutPolicy::default(), } } } @@ -3722,6 +3728,29 @@ impl AppConfig { config.storage.retry.backoff_multiplier = n; } + // Backend call timeouts. `0` means "unbounded" for that class, + // which is the default for writes — see `TimeoutPolicy`. + for (var, slot) in [ + ( + "OXICLOUD_STORAGE_TIMEOUT_METADATA_MS", + &mut config.storage.timeout.metadata, + ), + ( + "OXICLOUD_STORAGE_TIMEOUT_OPEN_MS", + &mut config.storage.timeout.open, + ), + ( + "OXICLOUD_STORAGE_TIMEOUT_WRITE_MS", + &mut config.storage.timeout.write, + ), + ] { + if let Ok(v) = env::var(var) + && let Ok(n) = v.parse::() + { + *slot = (n > 0).then(|| std::time::Duration::from_millis(n)); + } + } + // OIDC configuration if let Ok(v) = env::var("OXICLOUD_OIDC_ENABLED") { config.oidc.enabled = v.parse::().unwrap_or(false); diff --git a/src/common/di.rs b/src/common/di.rs index fe88a3fa..d65377de 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -329,7 +329,8 @@ impl AppServiceFactory { // just before the struct init. let active_backend_name = Arc::new(std::sync::RwLock::new(active_backend_name)); - // Stack decorators: retry → encryption → cache (inner-to-outer). + // Stack decorators: timeout → retry → encryption → cache + // (inner-to-outer). // // Encryption is applied INSIDE build_entry_backend (per-entry // key), so it's already on the base returned above when the @@ -343,6 +344,32 @@ impl AppServiceFactory { // gates the "remote-only" decorators the same as before. let mut blob_backend: Arc = base_backend; + // Timeout decorator — INNERMOST, and applied to every backend + // kind including Local. + // + // It has to sit below retry: a call that never returns produces + // no error, so retry has nothing to react to and the job never + // pauses. Converting the hang into a transient error first is + // what gives every layer above it something to act on. + // + // Unconditional by design. Retry is gated on "not Local" because + // the kernel already retries local I/O, but a bound that never + // fires is free, and keeping the chain uniform avoids a class of + // backend-specific surprise. + { + use crate::infrastructure::services::timeout_blob_backend::TimeoutBlobBackend; + let policy = self.config.storage.timeout.clone(); + if policy.is_enabled() { + blob_backend = Arc::new(TimeoutBlobBackend::new(blob_backend, policy.clone())); + tracing::info!( + metadata_ms = policy.metadata.map(|d| d.as_millis() as u64), + open_ms = policy.open.map(|d| d.as_millis() as u64), + write_ms = policy.write.map(|d| d.as_millis() as u64), + "Blob storage timeout decorator enabled" + ); + } + } + // Retry decorator (for remote backends) if self.config.storage.retry.enabled && active_backend_kind != StorageBackendType::Local { use crate::infrastructure::services::retry_blob_backend::{ diff --git a/src/infrastructure/services/mod.rs b/src/infrastructure/services/mod.rs index b872a0bb..1522429e 100644 --- a/src/infrastructure/services/mod.rs +++ b/src/infrastructure/services/mod.rs @@ -64,6 +64,7 @@ pub mod thumb_derived_import_service; pub mod thumbnail_service; #[cfg(test)] mod thumbnail_service_test; +pub mod timeout_blob_backend; pub mod transcode_import_service; pub mod trash_cleanup_service; pub mod tree_etag_flush_service; diff --git a/src/infrastructure/services/s3_blob_backend.rs b/src/infrastructure/services/s3_blob_backend.rs index ed7df48c..4bee0f2d 100644 --- a/src/infrastructure/services/s3_blob_backend.rs +++ b/src/infrastructure/services/s3_blob_backend.rs @@ -7,6 +7,7 @@ use aws_sdk_s3::primitives::ByteStream; use bytes::Bytes; use std::path::{Path, PathBuf}; use std::pin::Pin; +use std::time::Duration; use tokio::fs; use tokio_util::io::ReaderStream; @@ -39,9 +40,36 @@ impl S3BlobBackend { "oxicloud", ); + // `Builder::new()` starts from nothing — in particular with no + // `TimeoutConfig` at all, which meant a lost network on an + // established connection produced no error until the OS gave up + // on TCP retransmission (~15 minutes). For that whole window a + // migration looked merely slow: no error, so no retry, no log + // and no pause. It also made the `SdkError::TimeoutError` arm of + // `s3_domain_error` unreachable. + // + // These bounds are deliberately not the ones in `TimeoutPolicy`: + // that decorator provides the configurable outer bound for every + // backend, while these are the SDK's finer, per-attempt + // instruments underneath it. + let timeouts = aws_sdk_s3::config::timeout::TimeoutConfig::builder() + .connect_timeout(Duration::from_secs(10)) + // Time to first byte, not transfer duration — a large object + // is never punished for being large. + .read_timeout(Duration::from_secs(30)) + .build(); + let mut builder = aws_sdk_s3::config::Builder::new() .region(aws_sdk_s3::config::Region::new(config.region.clone())) .credentials_provider(credentials) + .timeout_config(timeouts) + // The right tool for a network pulled mid-transfer: it + // measures throughput rather than elapsed time, so it can + // bound a streaming upload without capping how long a + // legitimately large one may take. + .stalled_stream_protection( + aws_sdk_s3::config::StalledStreamProtectionConfig::enabled().build(), + ) .behavior_version_latest(); if let Some(ref endpoint) = config.endpoint_url { diff --git a/src/infrastructure/services/timeout_blob_backend.rs b/src/infrastructure/services/timeout_blob_backend.rs new file mode 100644 index 00000000..3bdd9402 --- /dev/null +++ b/src/infrastructure/services/timeout_blob_backend.rs @@ -0,0 +1,632 @@ +//! `TimeoutBlobBackend` — bounds every backend call in wall-clock time. +//! +//! ## Why this exists +//! +//! A backend call that *fails* is handled: it is classified, retried if +//! transient, and pauses the job at its cursor if it stays transient. A +//! call that never returns is handled by nothing at all. +//! +//! That is not hypothetical. Pull the network on an established TCP +//! connection and there is no RST and no ICMP — the peer simply stops +//! answering, and a socket read blocks until the OS gives up on +//! retransmission, on the order of fifteen minutes. For that whole +//! window the job is neither running nor failed: no error, so no retry, +//! no log line, no pause, nothing on the admin page. It looks exactly +//! like a very slow migration. +//! +//! Refusing a connection is instant and *does* surface (that is what +//! makes a `127.0.0.1` test look reassuring); losing a network mid-flight +//! is the silent case, and it is also the realistic one. +//! +//! ## Why a decorator rather than per-SDK configuration +//! +//! The S3 SDK can express this natively, and does — see the +//! `TimeoutConfig` in `s3_blob_backend.rs`, which is throughput-aware and +//! therefore strictly better for streams. But it only covers S3, and +//! Azure's 0.21 client has no equivalent knob short of supplying a custom +//! transport. That asymmetry is the reason this lives in the chain +//! instead of being configured twice. +//! +//! The local backend is a different story and deliberately not the +//! justification for this decorator: a local path is reached through the +//! kernel, and the kernel already owns that timeout. iSCSI gives up after +//! `replacement_timeout` (120s by default) and returns an I/O error; +//! NVMe-oF and soft-mounted NFS behave the same way. Those surface as an +//! `io::Error` and are classified by `local_io_error`, which is where +//! they belong. Local passes through this decorator only because a +//! uniform chain is simpler than a conditional one, and a bound that +//! never fires costs nothing. +//! +//! ## Operation classes +//! +//! A single timeout cannot fit both a `HEAD` and a multi-gigabyte upload, +//! so calls are bounded by what they do: +//! +//! * **Metadata** — `blob_exists`, `blob_size`, `delete_blob`, +//! `initialize`, `health_check`, `list_blob_hashes`. Bounded tightly. +//! These are the calls a migration makes per blob, and `blob_exists` in +//! particular is its very first probe of the source. +//! * **Open** — `get_blob_stream`, `get_blob_range_stream`. The future +//! resolves once the response *starts*; the body streams afterwards. So +//! this bounds time-to-first-byte, not transfer duration, and a slow +//! large read is never punished for being large. +//! * **Write** — the `put_*` family and `sync_blobs`. The entire transfer +//! happens inside the future, so any wall-clock bound here is also a +//! maximum upload duration. Unbounded by default for that reason: +//! getting it wrong truncates legitimate uploads, which is a worse +//! failure than the hang it would prevent. S3 covers this properly +//! through stalled-stream protection, which measures throughput instead +//! of elapsed time. +//! +//! A timeout is reported as [`DomainError::transient_backend`], because +//! that is what it is: no information was obtained about the blob. The +//! job engine pauses at its cursor and the work resumes when the network +//! does. + +use std::path::{Path, PathBuf}; +use std::pin::Pin; +use std::sync::Arc; +use std::time::Duration; + +use crate::application::ports::blob_storage_ports::{ + BlobListPage, BlobStorageBackend, BlobStream, StorageHealthStatus, +}; +use crate::domain::errors::DomainError; +use bytes::Bytes; + +// ── Timeout policy ───────────────────────────────────────────────── + +/// Per-operation-class wall-clock bounds. +#[derive(Debug, Clone)] +pub struct TimeoutPolicy { + /// Bound for metadata calls (exists, size, delete, init, health, list). + pub metadata: Option, + /// Bound for time-to-first-byte on reads. + pub open: Option, + /// Bound for the whole of a write. `None` (the default) leaves large + /// uploads unbounded — see the module docs. + pub write: Option, +} + +impl Default for TimeoutPolicy { + fn default() -> Self { + Self { + metadata: Some(Duration::from_secs(30)), + open: Some(Duration::from_secs(60)), + write: None, + } + } +} + +impl TimeoutPolicy { + /// A policy that bounds nothing — the pre-decorator behaviour. + pub fn disabled() -> Self { + Self { + metadata: None, + open: None, + write: None, + } + } + + /// True when at least one class is bounded, i.e. wrapping is worth it. + pub fn is_enabled(&self) -> bool { + self.metadata.is_some() || self.open.is_some() || self.write.is_some() + } +} + +// ── TimeoutBlobBackend ───────────────────────────────────────────── + +/// Decorator that fails a call the backend never answers. +pub struct TimeoutBlobBackend { + inner: Arc, + policy: TimeoutPolicy, +} + +impl TimeoutBlobBackend { + pub fn new(inner: Arc, policy: TimeoutPolicy) -> Self { + Self { inner, policy } + } +} + +/// Await `fut`, giving up after `limit`. +/// +/// `name` is lazy for the same reason as the retry decorator's: the +/// success path must not pay for a `format!` it will never print. +async fn with_timeout( + limit: Option, + backend: &'static str, + name: L, + fut: impl std::future::Future>, +) -> Result +where + L: Fn() -> String, +{ + let Some(limit) = limit else { + return fut.await; + }; + match tokio::time::timeout(limit, fut).await { + Ok(result) => result, + Err(_) => { + let op = name(); + // The one log line that distinguishes "hung" from "slow". + // Without it a stalled backend is invisible until the job + // pauses, and the pause reason alone does not say which + // layer noticed. + tracing::warn!( + target: "oxicloud::storage", + wrapper = "timeout", + backend = backend, + operation = %op, + timeout_ms = limit.as_millis() as u64, + "⏱️ Backend call timed out — treating as transient" + ); + Err(DomainError::transient_backend( + "Blob", + format!( + "{op} on {backend} backend timed out after {:?} (no response)", + limit + ), + )) + } + } +} + +impl BlobStorageBackend for TimeoutBlobBackend { + fn initialize( + &self, + ) -> Pin> + Send + '_>> { + let inner = self.inner.clone(); + let limit = self.policy.metadata; + Box::pin(async move { + let backend = inner.backend_type(); + with_timeout( + limit, + backend, + || "initialize".to_string(), + inner.initialize(), + ) + .await + }) + } + + fn put_blob( + &self, + hash: &str, + source_path: &Path, + ) -> Pin> + Send + '_>> { + let inner = self.inner.clone(); + let limit = self.policy.write; + let hash = hash.to_string(); + let path = source_path.to_path_buf(); + Box::pin(async move { + let backend = inner.backend_type(); + let label = hash.clone(); + with_timeout( + limit, + backend, + || format!("put_blob({label})"), + inner.put_blob(&hash, &path), + ) + .await + }) + } + + fn put_blob_from_bytes( + &self, + hash: &str, + data: Bytes, + ) -> Pin> + Send + '_>> { + let inner = self.inner.clone(); + let limit = self.policy.write; + let hash = hash.to_string(); + Box::pin(async move { + let backend = inner.backend_type(); + let label = hash.clone(); + with_timeout( + limit, + backend, + || format!("put_blob_from_bytes({label})"), + inner.put_blob_from_bytes(&hash, data), + ) + .await + }) + } + + fn put_blob_from_bytes_unsynced( + &self, + hash: &str, + data: Bytes, + ) -> Pin> + Send + '_>> { + let inner = self.inner.clone(); + let limit = self.policy.write; + let hash = hash.to_string(); + Box::pin(async move { + let backend = inner.backend_type(); + let label = hash.clone(); + with_timeout( + limit, + backend, + || format!("put_blob_from_bytes_unsynced({label})"), + inner.put_blob_from_bytes_unsynced(&hash, data), + ) + .await + }) + } + + fn put_blob_from_bytes_replace( + &self, + hash: &str, + data: Bytes, + ) -> Pin> + Send + '_>> { + let inner = self.inner.clone(); + let limit = self.policy.write; + let hash = hash.to_string(); + Box::pin(async move { + let backend = inner.backend_type(); + let label = hash.clone(); + with_timeout( + limit, + backend, + || format!("put_blob_from_bytes_replace({label})"), + inner.put_blob_from_bytes_replace(&hash, data), + ) + .await + }) + } + + fn sync_blobs( + &self, + hashes: &[String], + ) -> Pin> + Send + '_>> { + let inner = self.inner.clone(); + let limit = self.policy.write; + let hashes = hashes.to_vec(); + Box::pin(async move { + let backend = inner.backend_type(); + let count = hashes.len(); + with_timeout( + limit, + backend, + || format!("sync_blobs({count} hashes)"), + inner.sync_blobs(&hashes), + ) + .await + }) + } + + fn get_blob_stream( + &self, + hash: &str, + ) -> Pin> + Send + '_>> + { + let inner = self.inner.clone(); + let limit = self.policy.open; + let hash = hash.to_string(); + Box::pin(async move { + let backend = inner.backend_type(); + let label = hash.clone(); + with_timeout( + limit, + backend, + || format!("get_blob_stream({label})"), + inner.get_blob_stream(&hash), + ) + .await + }) + } + + fn get_blob_range_stream( + &self, + hash: &str, + start: u64, + end: Option, + ) -> Pin> + Send + '_>> + { + let inner = self.inner.clone(); + let limit = self.policy.open; + let hash = hash.to_string(); + Box::pin(async move { + let backend = inner.backend_type(); + let label = hash.clone(); + with_timeout( + limit, + backend, + || format!("get_blob_range_stream({label}, {start}..{end:?})"), + inner.get_blob_range_stream(&hash, start, end), + ) + .await + }) + } + + fn delete_blob( + &self, + hash: &str, + ) -> Pin> + Send + '_>> { + let inner = self.inner.clone(); + let limit = self.policy.metadata; + let hash = hash.to_string(); + Box::pin(async move { + let backend = inner.backend_type(); + let label = hash.clone(); + with_timeout( + limit, + backend, + || format!("delete_blob({label})"), + inner.delete_blob(&hash), + ) + .await + }) + } + + fn blob_exists( + &self, + hash: &str, + ) -> Pin> + Send + '_>> { + let inner = self.inner.clone(); + let limit = self.policy.metadata; + let hash = hash.to_string(); + Box::pin(async move { + let backend = inner.backend_type(); + let label = hash.clone(); + with_timeout( + limit, + backend, + || format!("blob_exists({label})"), + inner.blob_exists(&hash), + ) + .await + }) + } + + fn blob_size( + &self, + hash: &str, + ) -> Pin> + Send + '_>> { + let inner = self.inner.clone(); + let limit = self.policy.metadata; + let hash = hash.to_string(); + Box::pin(async move { + let backend = inner.backend_type(); + let label = hash.clone(); + with_timeout( + limit, + backend, + || format!("blob_size({label})"), + inner.blob_size(&hash), + ) + .await + }) + } + + fn health_check( + &self, + ) -> Pin< + Box> + Send + '_>, + > { + let inner = self.inner.clone(); + let limit = self.policy.metadata; + Box::pin(async move { + let backend = inner.backend_type(); + with_timeout( + limit, + backend, + || "health_check".to_string(), + inner.health_check(), + ) + .await + }) + } + + fn list_blob_hashes( + &self, + cursor: Option, + limit_n: usize, + ) -> Pin> + Send + '_>> + { + let inner = self.inner.clone(); + let limit = self.policy.metadata; + Box::pin(async move { + let backend = inner.backend_type(); + with_timeout( + limit, + backend, + || format!("list_blob_hashes(limit {limit_n})"), + inner.list_blob_hashes(cursor, limit_n), + ) + .await + }) + } + + fn backend_type(&self) -> &'static str { + self.inner.backend_type() + } + + fn local_blob_path(&self, hash: &str) -> Option { + self.inner.local_blob_path(hash) + } + + /// Forwarded, and then re-wrapped. + /// + /// `uncached()` exists so a verification pass can read past the + /// cache; an inner backend reached that way is no less able to hang + /// than the cached one, so it keeps the same bound. + fn uncached(&self) -> Option> { + self.inner.uncached().map(|inner| { + Arc::new(TimeoutBlobBackend::new(inner, self.policy.clone())) + as Arc + }) + } + + fn read_prefetch(&self) -> usize { + self.inner.read_prefetch() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::domain::errors::ErrorKind; + + /// A backend whose every call parks forever — the network-pulled case. + struct HangingBackend; + + impl BlobStorageBackend for HangingBackend { + fn initialize( + &self, + ) -> Pin> + Send + '_>> + { + Box::pin(async { std::future::pending().await }) + } + fn put_blob( + &self, + _hash: &str, + _source_path: &Path, + ) -> Pin> + Send + '_>> + { + Box::pin(async { std::future::pending().await }) + } + fn put_blob_from_bytes( + &self, + _hash: &str, + _data: Bytes, + ) -> Pin> + Send + '_>> + { + Box::pin(async { std::future::pending().await }) + } + fn get_blob_stream( + &self, + _hash: &str, + ) -> Pin> + Send + '_>> + { + Box::pin(async { std::future::pending().await }) + } + fn get_blob_range_stream( + &self, + _hash: &str, + _start: u64, + _end: Option, + ) -> Pin> + Send + '_>> + { + Box::pin(async { std::future::pending().await }) + } + fn delete_blob( + &self, + _hash: &str, + ) -> Pin> + Send + '_>> + { + Box::pin(async { std::future::pending().await }) + } + fn blob_exists( + &self, + _hash: &str, + ) -> Pin> + Send + '_>> + { + Box::pin(async { std::future::pending().await }) + } + fn blob_size( + &self, + _hash: &str, + ) -> Pin> + Send + '_>> + { + Box::pin(async { std::future::pending().await }) + } + fn health_check( + &self, + ) -> Pin< + Box< + dyn std::future::Future> + + Send + + '_, + >, + > { + Box::pin(async { std::future::pending().await }) + } + fn backend_type(&self) -> &'static str { + "hanging" + } + fn local_blob_path(&self, _hash: &str) -> Option { + None + } + } + + fn wrapped() -> TimeoutBlobBackend { + TimeoutBlobBackend::new( + Arc::new(HangingBackend), + TimeoutPolicy { + metadata: Some(Duration::from_millis(50)), + open: Some(Duration::from_millis(50)), + write: Some(Duration::from_millis(50)), + }, + ) + } + + /// The whole point: a hang must become a *transient* error, not a + /// hang and not a permanent one. `NotFound` here would tell a + /// migration the blob is absent; a permanent error would fail the + /// run instead of pausing it. + #[tokio::test] + async fn a_hanging_backend_yields_a_transient_error_not_a_hang() { + let backend = wrapped(); + + let err = backend.blob_exists("abc").await.unwrap_err(); + assert!( + err.is_transient(), + "a stalled probe must be transient so the job pauses and resumes: {err}" + ); + assert_ne!( + err.kind, + ErrorKind::NotFound, + "a hang says nothing about whether the blob exists" + ); + + assert!(backend.blob_size("abc").await.unwrap_err().is_transient()); + assert!(backend.delete_blob("abc").await.unwrap_err().is_transient()); + assert!(backend.initialize().await.unwrap_err().is_transient()); + // `BlobStream` is not `Debug`, so go through `.err()` rather than + // `unwrap_err()`. + assert!( + backend + .get_blob_stream("abc") + .await + .err() + .expect("a hanging read must not succeed") + .is_transient() + ); + } + + /// `write: None` is the default, and it must genuinely mean + /// "unbounded" — a large upload cannot be truncated by this + /// decorator. + #[tokio::test] + async fn an_unbounded_class_is_not_bounded() { + let backend = TimeoutBlobBackend::new( + Arc::new(HangingBackend), + TimeoutPolicy { + metadata: Some(Duration::from_millis(50)), + open: None, + write: None, + }, + ); + + // Bounded class still fires... + assert!(backend.blob_exists("abc").await.unwrap_err().is_transient()); + + // ...while an unbounded one is still pending long after the + // bounded one would have given up. `write: None` is the default, + // and a truncated multi-gigabyte upload is a worse outcome than + // the hang the bound would have caught. + let open = backend.get_blob_stream("abc"); + assert!( + tokio::time::timeout(Duration::from_millis(250), open) + .await + .is_err(), + "an unbounded class must never be cut short by the decorator" + ); + } + + #[test] + fn disabled_is_disabled() { + assert!(!TimeoutPolicy::disabled().is_enabled()); + assert!(TimeoutPolicy::default().is_enabled()); + } +}