diff --git a/Cargo.lock b/Cargo.lock index cd43eb83..e80588bf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2882,6 +2882,7 @@ dependencies = [ "tokio-util", "tonic", "tracing", + "url", "uuid", "zstd", ] diff --git a/objectstore-server/src/endpoints/common.rs b/objectstore-server/src/endpoints/common.rs index 97468764..94771df2 100644 --- a/objectstore-server/src/endpoints/common.rs +++ b/objectstore-server/src/endpoints/common.rs @@ -7,6 +7,7 @@ use axum::http::StatusCode; use axum::response::{IntoResponse, Response}; use http::HeaderValue; use objectstore_service::error::Error as ServiceError; +use objectstore_service::error::ErrorKind as ServiceErrorKind; use serde::{Deserialize, Serialize}; use thiserror::Error; @@ -90,15 +91,18 @@ impl ApiError { ApiError::Auth(AuthError::NotPermitted) => StatusCode::FORBIDDEN, ApiError::Auth(AuthError::InternalError(_)) => StatusCode::INTERNAL_SERVER_ERROR, - ApiError::Service(ServiceError::Client(_)) => StatusCode::BAD_REQUEST, - ApiError::Service(ServiceError::Metadata(_)) => StatusCode::BAD_REQUEST, - ApiError::Service(ServiceError::RangeNotSatisfiable { .. }) => { - StatusCode::RANGE_NOT_SATISFIABLE - } - ApiError::Service(ServiceError::InvalidUploadId(_)) => StatusCode::BAD_REQUEST, - ApiError::Service(ServiceError::AtCapacity) => StatusCode::TOO_MANY_REQUESTS, - ApiError::Service(ServiceError::NotImplemented) => StatusCode::NOT_IMPLEMENTED, - ApiError::Service(_) => StatusCode::INTERNAL_SERVER_ERROR, + ApiError::Service(error) => match error.kind { + ServiceErrorKind::ClientStream | ServiceErrorKind::InvalidInput => { + StatusCode::BAD_REQUEST + } + ServiceErrorKind::RangeNotSatisfiable => StatusCode::RANGE_NOT_SATISFIABLE, + ServiceErrorKind::BackendRateLimited => StatusCode::TOO_MANY_REQUESTS, + ServiceErrorKind::AtCapacity + | ServiceErrorKind::BackendTimeout + | ServiceErrorKind::BackendUnavailable => StatusCode::SERVICE_UNAVAILABLE, + ServiceErrorKind::NotImplemented => StatusCode::NOT_IMPLEMENTED, + ServiceErrorKind::Internal => StatusCode::INTERNAL_SERVER_ERROR, + }, ApiError::Internal(_) => StatusCode::INTERNAL_SERVER_ERROR, } diff --git a/objectstore-server/src/endpoints/objects.rs b/objectstore-server/src/endpoints/objects.rs index a08b6620..bae6ea75 100644 --- a/objectstore-server/src/endpoints/objects.rs +++ b/objectstore-server/src/endpoints/objects.rs @@ -4,7 +4,7 @@ use axum::http::{HeaderMap, StatusCode}; use axum::response::{IntoResponse, Response}; use axum::routing; use axum::{Json, Router}; -use objectstore_service::error::Error as ServiceError; +use objectstore_service::error::{Error as ServiceError, ErrorKind as ServiceErrorKind}; use objectstore_service::id::{ObjectContext, ObjectId}; use objectstore_types::metadata::Metadata; use objectstore_types::range::ContentRange; @@ -72,7 +72,13 @@ async fn object_get( let (metadata, content_range, stream) = match result { Ok(Some(result)) => result, Ok(None) => return Ok(StatusCode::NOT_FOUND.into_response()), - Err(ApiError::Service(ServiceError::RangeNotSatisfiable { total })) => { + Err(ApiError::Service(e)) if e.kind == ServiceErrorKind::RangeNotSatisfiable => { + let Some(total) = e.range_total() else { + return Err(ApiError::Service( + e.kind(ServiceErrorKind::Internal) + .context("range error missing object size"), + )); + }; let mut response = ( StatusCode::RANGE_NOT_SATISFIABLE, [( diff --git a/objectstore-server/tests/limits.rs b/objectstore-server/tests/limits.rs index d2ad5a3a..94eb6690 100644 --- a/objectstore-server/tests/limits.rs +++ b/objectstore-server/tests/limits.rs @@ -625,9 +625,10 @@ async fn test_bandwidth_scope_pct_limit() -> Result<()> { } #[tokio::test] -async fn test_batch_at_capacity_returns_429() -> Result<()> { +async fn test_batch_at_capacity_returns_503() -> Result<()> { // With max_concurrency=0 the service has no permits available, so - // BatchExecutor::new() returns AtCapacity and the endpoint responds 429. + // BatchExecutor::new() returns AtCapacity and the endpoint responds 503: + // local load-shedding is surfaced as a temporary Service Unavailable. let server = TestServer::with_config(Config { service: Service { max_concurrency: 0, @@ -660,8 +661,8 @@ async fn test_batch_at_capacity_returns_429() -> Result<()> { assert_eq!( response.status(), - reqwest::StatusCode::TOO_MANY_REQUESTS, - "expected 429 when service has no available permits" + reqwest::StatusCode::SERVICE_UNAVAILABLE, + "expected 503 when service has no available permits" ); Ok(()) diff --git a/objectstore-service/Cargo.toml b/objectstore-service/Cargo.toml index df6b771a..fe13b1ca 100644 --- a/objectstore-service/Cargo.toml +++ b/objectstore-service/Cargo.toml @@ -34,6 +34,7 @@ tokio = { workspace = true } tokio-util = { workspace = true, features = ["io", "rt"] } tonic = { workspace = true } tracing = { workspace = true } +url = { workspace = true } uuid = { workspace = true, features = ["v7"] } [dev-dependencies] diff --git a/objectstore-service/docs/architecture.md b/objectstore-service/docs/architecture.md index 96e51980..745587c8 100644 --- a/objectstore-service/docs/architecture.md +++ b/objectstore-service/docs/architecture.md @@ -180,11 +180,11 @@ prevent exhaustion of internal resources such as memory. ## Concurrency Limit -A concurrency limiter caps in-flight -backend operations. When all execution permits are held, new operations are -queued — adding latency instead of rejecting immediately. The queue itself is -bounded in both depth and time: operations that cannot be served within those -limits fail with [`Error::AtCapacity`](error::Error::AtCapacity). +A concurrency limiter caps in-flight backend operations. When all execution +permits are held, new operations are queued — adding latency instead of +rejecting immediately. The queue itself is bounded in both depth and time: +operations that cannot be served within those limits fail with an +[`ErrorKind::AtCapacity`](error::ErrorKind::AtCapacity) error. The default execution limit is [`DEFAULT_CONCURRENCY_LIMIT`](service::DEFAULT_CONCURRENCY_LIMIT). See @@ -211,3 +211,31 @@ reservation, lazy pulling, memory bounds, and concurrency model. More backpressure mechanisms (e.g. per-backend limits, adaptive throttling) may be added here in the future. + +# Error Model + +Service and backend failures use a single [`Error`](error::Error) containing an +[`ErrorKind`](error::ErrorKind) classification and an `anyhow::Error` +diagnostic. The classification is deliberately independent of both the +concrete error type and HTTP semantics: it can be changed without altering the +diagnostic chain, while the server maps it onto a status code and +[`Error::level`](error::Error::level) maps it onto a log level. + +Construction follows two paths: + +- **Default (`?`)**: a foreign error converts through one of the `From` impls + into an [`ErrorKind::Internal`](error::ErrorKind::Internal) error while the + original error and its source chain become the diagnostic. +- **Override**: the [`ResultExt`](error::ResultExt) extension trait's `.kind(…)` + and `.context(…)` methods reclassify and annotate without a `map_err`. + Repeated `.context(…)` calls use anyhow semantics and retain every context + frame; `.kind(…)` changes only the independent classification. + +`Error` delegates display, debug formatting, chain traversal, and downcasting +to anyhow. Backends classify transport and HTTP failures into the `Backend*` +kinds in [`check_error`](backend), so retryability is a simple check on +[`kind`](error::Error::kind). Unsatisfiable ranges are constructed with +[`Error::range_not_satisfiable`](error::Error::range_not_satisfiable) and expose +their required object size through +[`Error::range_total`](error::Error::range_total), allowing the server to emit a +valid `Content-Range` header without relying on an unchecked source convention. diff --git a/objectstore-service/src/backend/bigtable.rs b/objectstore-service/src/backend/bigtable.rs index fef619b3..3f9d1a18 100644 --- a/objectstore-service/src/backend/bigtable.rs +++ b/objectstore-service/src/backend/bigtable.rs @@ -47,7 +47,7 @@ use crate::backend::common::{ Backend, DeleteResponse, GetResponse, HighVolumeBackend, MetadataResponse, PutResponse, TieredGet, TieredMetadata, TieredWrite, Tombstone, }; -use crate::error::{Error, Result}; +use crate::error::{Error, ErrorKind, Result, ResultExt}; use crate::gcp_auth::PrefetchingTokenProvider; use crate::id::ObjectId; use crate::stream::{ChunkedBytes, ClientStream}; @@ -466,8 +466,7 @@ fn object_mutations(mut metadata: Metadata, payload: Vec) -> Result<[v2::Mut // Record the payload size in the metadata before persisting it. metadata.size = Some(payload.len()); - let metadata_bytes = serde_json::to_vec(&metadata) - .map_err(|cause| Error::serde("failed to serialize metadata", cause))?; + let metadata_bytes = serde_json::to_vec(&metadata).context("failed to serialize metadata")?; Ok([ // NB: We explicitly delete the row to clear metadata on overwrite. @@ -528,8 +527,7 @@ fn tombstone_mutations(tombstone: &Tombstone, now: SystemTime) -> Result<[v2::Mu family_name: family.to_owned(), column_qualifier: COLUMN_TOMBSTONE_META.to_owned(), timestamp_micros, - value: serde_json::to_vec(&tombstone_meta) - .map_err(|cause| Error::serde("failed to serialize tombstone", cause))?, + value: serde_json::to_vec(&tombstone_meta).context("failed to serialize tombstone")?, })), ]) } @@ -601,10 +599,10 @@ impl RowData { payload = cell.value; } COLUMN_TOMBSTONE_META => { - tombstone_meta_opt = - Some(serde_json::from_slice(&cell.value).map_err(|cause| { - Error::serde("failed to deserialize tombstone meta", cause) - })?); + tombstone_meta_opt = Some( + serde_json::from_slice(&cell.value) + .context("failed to deserialize tombstone meta")?, + ); } COLUMN_METADATA => { if let Ok(legacy_meta) = @@ -617,10 +615,10 @@ impl RowData { expiration_policy: legacy_meta.expiration_policy, }); } else { - metadata_opt = - Some(serde_json::from_slice(&cell.value).map_err(|cause| { - Error::serde("failed to deserialize metadata", cause) - })?); + metadata_opt = Some( + serde_json::from_slice(&cell.value) + .context("failed to deserialize metadata")?, + ); } } _ => {} @@ -683,10 +681,11 @@ fn parse_redirect_target(redirect_path: &[u8], tombstone_id: &ObjectId) -> Resul objectstore_metrics::count!("bigtable.empty_redirect_read"); Ok(tombstone_id.clone()) } else { - let redirect_str = std::str::from_utf8(redirect_path) - .map_err(|_| Error::generic("invalid UTF-8 in redirect path"))?; + let redirect_str = std::str::from_utf8(redirect_path).map_err(|_| { + Error::new(ErrorKind::Internal).context("invalid UTF-8 in redirect path") + })?; ObjectId::from_storage_path(redirect_str) - .ok_or_else(|| Error::generic("corrupt redirect path")) + .ok_or_else(|| Error::new(ErrorKind::Internal).context("corrupt redirect path")) } } @@ -930,7 +929,9 @@ impl Backend for BigTableBackend { TieredGet::Object(metadata, content_range, payload) => { Ok(Some((metadata, content_range, payload))) } - TieredGet::Tombstone(_) => Err(Error::UnexpectedTombstone), + TieredGet::Tombstone(_) => { + Err(Error::new(ErrorKind::Internal).context("unexpected tombstone")) + } TieredGet::NotFound => Ok(None), } } @@ -939,7 +940,9 @@ impl Backend for BigTableBackend { async fn get_metadata(&self, id: &ObjectId) -> Result { match self.get_tiered_metadata(id).await? { TieredMetadata::Object(metadata) => Ok(Some(metadata)), - TieredMetadata::Tombstone(_) => Err(Error::UnexpectedTombstone), + TieredMetadata::Tombstone(_) => { + Err(Error::new(ErrorKind::Internal).context("unexpected tombstone")) + } TieredMetadata::NotFound => Ok(None), } } @@ -1002,7 +1005,7 @@ impl HighVolumeBackend for BigTableBackend { } } - Err(Error::generic("BigTable: race loop in put_non_tombstone")) + Err(Error::new(ErrorKind::Internal).context("BigTable: race loop in put_non_tombstone")) } #[tracing::instrument(level = "debug", skip(self))] @@ -1110,9 +1113,7 @@ impl HighVolumeBackend for BigTableBackend { } } - Err(Error::generic( - "BigTable: race loop in delete_non_tombstone", - )) + Err(Error::new(ErrorKind::Internal).context("BigTable: race loop in delete_non_tombstone")) } #[tracing::instrument(level = "debug", skip(self, write))] @@ -1151,13 +1152,12 @@ impl HighVolumeBackend for BigTableBackend { /// required by BigTable, the resulting timestamp has millisecond precision, with the last digits at /// 0. fn ttl_to_micros(ttl: Duration, from: SystemTime) -> Result { - let deadline = from.checked_add(ttl).ok_or_else(|| Error::Generic { - context: format!( + let deadline = from.checked_add(ttl).ok_or_else(|| { + Error::new(ErrorKind::Internal).context(format!( "TTL duration overflow: {} plus {}s cannot be represented as SystemTime", humantime::format_rfc3339_seconds(from), ttl.as_secs() - ), - cause: None, + )) })?; system_time_to_micros(deadline) @@ -1170,19 +1170,15 @@ fn ttl_to_micros(ttl: Duration, from: SystemTime) -> Result { fn system_time_to_micros(deadline: SystemTime) -> Result { let millis = deadline .duration_since(SystemTime::UNIX_EPOCH) - .map_err(|e| Error::Generic { - context: format!( - "unable to get duration since UNIX_EPOCH for SystemTime {}", - humantime::format_rfc3339_seconds(deadline) - ), - cause: Some(Box::new(e)), - })? + .context(format!( + "unable to get duration since UNIX_EPOCH for SystemTime {}", + humantime::format_rfc3339_seconds(deadline) + ))? .as_millis(); - (millis * 1000).try_into().map_err(|e| Error::Generic { - context: format!("failed to convert {millis}ms to i64 microseconds"), - cause: Some(Box::new(e)), - }) + (millis * 1000) + .try_into() + .context(format!("failed to convert {millis}ms to i64 microseconds")) } /// Converts a wall-clock time to Bigtable's microsecond timestamp, saturating at `i64::MAX` @@ -1233,10 +1229,7 @@ where Ok(res) => return Ok(res), Err(e) if retry_count >= REQUEST_RETRY_COUNT || !is_retryable(&e) => { objectstore_metrics::count!("bigtable.failures", action = context); - return Err(Error::Generic { - context: format!("Bigtable: `{context}` failed"), - cause: Some(Box::new(e)), - }); + return Err(Error::from(e).context(format!("Bigtable: `{context}` failed"))); } Err(e) => { retry_count += 1; @@ -1290,7 +1283,7 @@ fn apply_range(payload: Bytes, range: Option) -> Result<(Option assert_eq!(total, 22), + Err(err) if err.kind == ErrorKind::RangeNotSatisfiable => { + let total = err.range_total(); + assert_eq!(total, Some(22)); + } Ok(_) => panic!("expected RangeNotSatisfiable, got Ok"), Err(e) => panic!("expected RangeNotSatisfiable, got {e:?}"), } diff --git a/objectstore-service/src/backend/common.rs b/objectstore-service/src/backend/common.rs index 40d08330..b3aab561 100644 --- a/objectstore-service/src/backend/common.rs +++ b/objectstore-service/src/backend/common.rs @@ -7,7 +7,7 @@ use objectstore_types::range::{ByteRange, ContentRange}; use bytes::Bytes; -use crate::error::{Error, Result}; +use crate::error::{Error, ErrorKind, Result}; use crate::id::ObjectId; use crate::multipart::{ AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse, @@ -67,10 +67,10 @@ pub trait Backend: fmt::Debug + Send + Sync + 'static { /// Borrows this backend as a [`MultipartUploadBackend`] if supported. /// - /// The default returns [`Error::NotImplemented`]. Backends that implement - /// [`MultipartUploadBackend`] should override this to return `Ok(self)`. + /// The default returns an [`ErrorKind::NotImplemented`] error. Backends that + /// implement [`MultipartUploadBackend`] should override this to return `Ok(self)`. fn as_multipart_upload_backend(&self) -> Result<&dyn MultipartUploadBackend> { - Err(Error::NotImplemented) + Err(Error::new(ErrorKind::NotImplemented)) } } diff --git a/objectstore-service/src/backend/extensions.rs b/objectstore-service/src/backend/extensions.rs index 31503126..7ad44ebf 100644 --- a/objectstore-service/src/backend/extensions.rs +++ b/objectstore-service/src/backend/extensions.rs @@ -6,13 +6,25 @@ //! structured error code and message from it (JSON for GCS JSON API, XML for GCS //! XML API and S3). -use reqwest::{Response, header}; +use reqwest::{Response, StatusCode, header}; use serde::Deserialize; use tracing::Instrument; -use crate::error::{BackendDetail, Error, Result}; +use crate::error::{BackendDetail, Error, ErrorKind, Result}; use crate::stream; +/// Classifies a backend HTTP error status into an [`ErrorKind`]. +fn status_to_kind(status: StatusCode) -> ErrorKind { + match status { + StatusCode::TOO_MANY_REQUESTS => ErrorKind::BackendRateLimited, + StatusCode::REQUEST_TIMEOUT | StatusCode::GATEWAY_TIMEOUT => ErrorKind::BackendTimeout, + StatusCode::INTERNAL_SERVER_ERROR + | StatusCode::BAD_GATEWAY + | StatusCode::SERVICE_UNAVAILABLE => ErrorKind::BackendUnavailable, + _ => ErrorKind::Internal, + } +} + /// Extension trait that sends a request inside a tracing span. pub trait SendTraced { /// Sends the request, wrapping it in a span that covers the full request @@ -41,6 +53,28 @@ impl SendTraced for reqwest::RequestBuilder { } } +/// Classifies a request transport result while leaving successful responses +/// available for caller-specific status handling. +pub(crate) fn classify_transport( + result: reqwest::Result, + context: &'static str, +) -> Result { + result.map_err(|error| { + if let Some(client_error) = stream::unpack_client_error(&error) { + return Error::from(client_error).context(context); + } + + let kind = if error.is_timeout() { + ErrorKind::BackendTimeout + } else if error.is_connect() || error.is_request() { + ErrorKind::BackendUnavailable + } else { + ErrorKind::Internal + }; + Error::from(error).kind(kind).context(context) + }) +} + /// GCS JSON API error envelope (`{"error": {"message": "...", ...}}`). #[derive(Deserialize)] struct JsonApiError { @@ -78,7 +112,7 @@ struct XmlApiError { /// Use [`check_error`](Self::check_error) instead of /// [`error_for_status`](reqwest::Response::error_for_status) to avoid losing the response body on /// 4xx/5xx errors. The method parses the structured error body (JSON or XML) and returns an -/// [`Error::BackendResponse`] with the extracted error code and message. +/// error classified by status ([`status_to_kind`]) with the extracted code and message as context. /// /// Implemented for both [`reqwest::Response`] and `Result` so it can be /// chained directly. @@ -91,7 +125,7 @@ pub trait ResponseExt { /// [`reqwest::Response::error_for_status`]. /// /// When called on `Result`, transport errors are - /// wrapped as [`Error::Reqwest`] with the same context string. + /// classified with [`classify_transport`] and given the same context string. async fn check_error(self, context: &'static str) -> Result; /// Drains the response body of a response we are otherwise done with. @@ -124,14 +158,16 @@ impl ResponseExt for Response { return Ok(self); }; self.drain_body().await; - return Err(Error::reqwest(context, e)); + return Err(Error::from(e).kind(status_to_kind(status)).context(context)); }; - Err(Error::BackendResponse { - context, - status, - detail, - }) + let detail = detail.to_string(); + let message = if detail.is_empty() { + format!("{context} ({status})") + } else { + format!("{context} ({status}). {detail}") + }; + Err(Error::new(status_to_kind(status)).context(message)) } async fn drain_body(mut self) { @@ -141,13 +177,9 @@ impl ResponseExt for Response { impl ResponseExt for Result { async fn check_error(self, context: &'static str) -> Result { - match self { - Ok(resp) => resp.check_error(context).await, - Err(e) => Err(match stream::unpack_client_error(&e) { - Some(ce) => Error::Client(ce), - None => Error::reqwest(context, e), - }), - } + classify_transport(self, context)? + .check_error(context) + .await } async fn drain_body(self) { @@ -185,3 +217,21 @@ async fn parse_xml_error(resp: Response) -> BackendDetail { BackendDetail::none() } } + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn classifies_connection_failures_as_backend_unavailable() { + let result = reqwest::Client::new() + .get("http://127.0.0.1:1") + .send() + .await; + let error = classify_transport(result, "connecting to backend").unwrap_err(); + + assert_eq!(error.kind, ErrorKind::BackendUnavailable); + assert_eq!(error.to_string(), "connecting to backend"); + assert!(error.downcast_ref::().is_some()); + } +} diff --git a/objectstore-service/src/backend/gcs.rs b/objectstore-service/src/backend/gcs.rs index 924efae5..5eb25a3e 100644 --- a/objectstore-service/src/backend/gcs.rs +++ b/objectstore-service/src/backend/gcs.rs @@ -6,7 +6,6 @@ use std::future::Future; use std::time::{Duration, SystemTime}; use std::{fmt, io}; -use anyhow::Context; use futures_util::{StreamExt, TryStreamExt}; use gcp_auth::TokenProvider; use objectstore_types::metadata::{ExpirationPolicy, Metadata}; @@ -15,12 +14,12 @@ use reqwest::header::HeaderName; use reqwest::{Body, IntoUrl, Method, RequestBuilder, StatusCode, Url, header, multipart}; use serde::{Deserialize, Serialize}; -use super::extensions::{ResponseExt, SendTraced}; +use super::extensions::{ResponseExt, SendTraced, classify_transport}; use crate::backend::common::{ self, Backend, DeleteResponse, GetResponse, MetadataResponse, MultipartUploadBackend, PutResponse, }; -use crate::error::{Error, Result}; +use crate::error::{Error, ErrorKind, Result, ResultExt}; use crate::gcp_auth::PrefetchingTokenProvider; use crate::id::ObjectId; use crate::multipart::{ @@ -201,10 +200,7 @@ impl GcsObject { .size .map(|size| size.parse()) .transpose() - .map_err(|e| Error::Generic { - context: "GCS: failed to parse size from object metadata".to_string(), - cause: Some(Box::new(e)), - })?; + .context("GCS: failed to parse size from object metadata")?; let time_created = self.time_created; // At this point, all built-in metadata should have been removed from self.metadata. @@ -213,12 +209,9 @@ impl GcsObject { if let GcsMetaKey::Custom(custom_key) = key { custom.insert(custom_key, value); } else { - return Err(Error::Generic { - context: format!( - "GCS: unexpected built-in metadata key in object metadata: {key}" - ), - cause: None, - }); + return Err(Error::new(ErrorKind::Internal).context(format!( + "GCS: unexpected built-in metadata key in object metadata: {key}" + ))); } } @@ -311,10 +304,10 @@ fn metadata_to_gcs_headers(metadata: &Metadata) -> Result { let formatted = humantime::format_rfc3339_seconds(custom_time); headers.insert( HeaderName::from_static("x-goog-custom-time"), - formatted.to_string().parse().map_err(|e| Error::Generic { - context: "GCS: invalid custom-time header value".into(), - cause: Some(Box::new(e)), - })?, + formatted + .to_string() + .parse() + .context("GCS: invalid custom-time header value")?, ); } @@ -324,10 +317,7 @@ fn metadata_to_gcs_headers(metadata: &Metadata) -> Result { compression .to_string() .parse() - .map_err(|e| Error::Generic { - context: "GCS: invalid content-encoding header value".into(), - cause: Some(Box::new(e)), - })?, + .context("GCS: invalid content-encoding header value")?, ); } @@ -361,42 +351,24 @@ fn insert_gcs_meta_header( ) -> Result<()> { let header_name = format!("x-goog-meta-{key}"); headers.insert( - HeaderName::try_from(&header_name).map_err(|e| Error::Generic { - context: format!("GCS: invalid header name: {header_name}"), - cause: Some(Box::new(e)), - })?, - value.parse().map_err(|e| Error::Generic { - context: format!("GCS: invalid header value for {header_name}"), - cause: Some(Box::new(e)), - })?, + HeaderName::try_from(&header_name) + .context(format!("GCS: invalid header name: {header_name}"))?, + value + .parse() + .context(format!("GCS: invalid header value for {header_name}"))?, ); Ok(()) } -/// Returns `true` if the error is a transient reqwest failure worth retrying. +/// Returns `true` if the error is a transient backend failure worth retrying. +/// +/// [`check_error`](super::extensions::ResponseExt::check_error) classifies both +/// transport failures and retryable HTTP statuses (408/429/500/502/503/504) +/// into the backend error kinds below, so retryability is a simple kind check. fn error_is_retryable(error: &Error) -> bool { - match error { - Error::Reqwest { cause, .. } => { - cause.is_timeout() - || cause.is_connect() - || cause.is_request() - || cause.status().is_some_and(status_is_retryable) - } - Error::BackendResponse { status, .. } => status_is_retryable(*status), - _ => false, - } -} - -fn status_is_retryable(status: StatusCode) -> bool { - // https://docs.cloud.google.com/storage/docs/json_api/v1/status-codes matches!( - status, - StatusCode::REQUEST_TIMEOUT - | StatusCode::TOO_MANY_REQUESTS - | StatusCode::INTERNAL_SERVER_ERROR - | StatusCode::BAD_GATEWAY - | StatusCode::SERVICE_UNAVAILABLE - | StatusCode::GATEWAY_TIMEOUT + error.kind, + ErrorKind::BackendTimeout | ErrorKind::BackendRateLimited | ErrorKind::BackendUnavailable ) } @@ -435,12 +407,11 @@ impl GcsBackend { let path = id.as_storage_path().to_string(); url.path_segments_mut() - .map_err(|()| Error::Generic { - context: format!( + .map_err(|()| { + Error::new(ErrorKind::Internal).context(format!( "GCS: invalid endpoint URL, {} cannot be a base", self.endpoint - ), - cause: None, + )) })? .extend(&["storage", "v1", "b", &self.bucket, "o", &path]); @@ -452,12 +423,11 @@ impl GcsBackend { let mut url = self.endpoint.clone(); url.path_segments_mut() - .map_err(|()| Error::Generic { - context: format!( + .map_err(|()| { + Error::new(ErrorKind::Internal).context(format!( "GCS: invalid endpoint URL, {} cannot be a base", self.endpoint - ), - cause: None, + )) })? .extend(&["upload", "storage", "v1", "b", &self.bucket, "o"]); @@ -477,12 +447,11 @@ impl GcsBackend { fn xml_object_url(&self, id: &ObjectId) -> Result { let mut url = self.endpoint.clone(); { - let mut segments = url.path_segments_mut().map_err(|()| Error::Generic { - context: format!( + let mut segments = url.path_segments_mut().map_err(|()| { + Error::new(ErrorKind::Internal).context(format!( "GCS: invalid endpoint URL, {} cannot be a base", self.endpoint - ), - cause: None, + )) })?; segments.push(&self.bucket); for part in id.as_storage_path().to_string().split('/') { @@ -530,12 +499,9 @@ impl GcsBackend { async fn fetch_gcs_metadata(&self, object_url: &Url) -> Result> { let metadata_opt = self .with_retry("get_metadata", || async { - let resp = self - .request(Method::GET, object_url.clone()) - .await? - .send_traced() - .await - .map_err(|e| Error::reqwest("GCS: get metadata request", e))?; + let request = self.request(Method::GET, object_url.clone()).await?; + let resp = + classify_transport(request.send_traced().await, "GCS: get metadata request")?; if resp.status() == StatusCode::NOT_FOUND { resp.drain_body().await; @@ -547,7 +513,7 @@ impl GcsBackend { .await? .json() .await - .map_err(|e| Error::reqwest("GCS: get metadata parse", e))?; + .context("GCS: get metadata parse")?; Ok(Some(metadata)) }) @@ -638,10 +604,8 @@ impl Backend for GcsBackend { // NB: Ensure the order of these fields and that a content-type is attached to them. Both // are required by the GCS API. - let metadata_json = serde_json::to_string(&gcs_metadata).map_err(|cause| Error::Serde { - context: "failed to serialize metadata for GCS upload".to_string(), - cause, - })?; + let metadata_json = serde_json::to_string(&gcs_metadata) + .context("failed to serialize metadata for GCS upload")?; let multipart = multipart::Form::new() .part( @@ -654,10 +618,7 @@ impl Backend for GcsBackend { "media", multipart::Part::stream(Body::wrap_stream(stream)) .mime_str(&metadata.content_type) - .map_err(|e| Error::Generic { - context: format!("invalid mime type: {}", metadata.content_type), - cause: Some(Box::new(e)), - })?, + .context(format!("invalid mime type: {}", metadata.content_type))?, ); // GCS requires a multipart/related request. Its body looks identical to @@ -697,10 +658,7 @@ impl Backend for GcsBackend { if let Some(r) = range { req = req.header(header::RANGE, r.to_header_value()); } - let resp = req - .send_traced() - .await - .map_err(|e| Error::reqwest("GCS: get payload", e))?; + let resp = classify_transport(req.send_traced().await, "GCS: get payload")?; if resp.status() == StatusCode::RANGE_NOT_SATISFIABLE { let raw = resp @@ -709,8 +667,8 @@ impl Backend for GcsBackend { .and_then(|v| v.to_str().ok()); let total = raw.and_then(ContentRange::parse_unsatisfiable_total); let err = match total { - Some(total) => Error::RangeNotSatisfiable { total }, - None => Error::generic(format!( + Some(total) => Error::range_not_satisfiable(total), + None => Error::new(ErrorKind::Internal).context(format!( "GCS: 416 response with invalid Content-Range: {raw:?}" )), }; @@ -729,9 +687,9 @@ impl Backend for GcsBackend { .get(header::CONTENT_RANGE) .and_then(|v| v.to_str().ok()) .and_then(|s| s.parse::().ok()) - .ok_or_else(|| Error::Generic { - context: "GCS: 206 response missing valid Content-Range header".to_owned(), - cause: None, + .ok_or_else(|| { + Error::new(ErrorKind::Internal) + .context("GCS: 206 response missing valid Content-Range header") })?, ) } else { @@ -759,12 +717,8 @@ impl Backend for GcsBackend { let object_url = self.object_url(id)?; self.with_retry("delete", || async { - let resp = self - .request(Method::DELETE, object_url.clone()) - .await? - .send_traced() - .await - .map_err(|e| Error::reqwest("GCS: delete object", e))?; + let request = self.request(Method::DELETE, object_url.clone()).await?; + let resp = classify_transport(request.send_traced().await, "GCS: delete object")?; // Do not error for objects that do not exist if resp.status() == StatusCode::NOT_FOUND { @@ -922,13 +876,10 @@ impl MultipartUploadBackend for GcsBackend { let body = resp .bytes() .await - .map_err(|e| Error::reqwest("GCS: read initiate multipart body", e))?; + .context("GCS: read initiate multipart body")?; let xml: XmlInitiateMultipartUploadResponse = quick_xml::de::from_reader(body.as_ref()) - .map_err(|e| Error::Generic { - context: "GCS: failed to parse initiate multipart response".to_owned(), - cause: Some(Box::new(e)), - })?; + .context("GCS: failed to parse initiate multipart response")?; Ok(xml.try_into()?) } @@ -970,7 +921,10 @@ impl MultipartUploadBackend for GcsBackend { .get(header::ETAG) .and_then(|v| v.to_str().ok()) .map(|s| s.to_owned()) - .ok_or_else(|| Error::generic("GCS: upload part response missing ETag header"))?; + .ok_or_else(|| { + Error::new(ErrorKind::Internal) + .context("GCS: upload part response missing ETag header") + })?; resp.drain_body().await; @@ -1006,16 +960,10 @@ impl MultipartUploadBackend for GcsBackend { .check_error("GCS: list parts") .await?; - let body = resp - .bytes() - .await - .map_err(|e| Error::reqwest("GCS: read list parts body", e))?; + let body = resp.bytes().await.context("GCS: read list parts body")?; - let xml: XmlListPartsResponse = - quick_xml::de::from_reader(body.as_ref()).map_err(|e| Error::Generic { - context: "GCS: failed to parse list parts response".to_owned(), - cause: Some(Box::new(e)), - })?; + let xml: XmlListPartsResponse = quick_xml::de::from_reader(body.as_ref()) + .context("GCS: failed to parse list parts response")?; Ok(xml.into()) } @@ -1054,10 +1002,8 @@ impl MultipartUploadBackend for GcsBackend { url.query_pairs_mut().append_pair("uploadId", upload_id); let body = XmlCompleteMultipartUpload::from(parts); - let xml = quick_xml::se::to_string(&body).map_err(|e| Error::Generic { - context: "GCS: failed to serialize complete multipart request".into(), - cause: Some(Box::new(e)), - })?; + let xml = quick_xml::se::to_string(&body) + .context("GCS: failed to serialize complete multipart request")?; let resp = self .request(Method::POST, url) @@ -1072,7 +1018,7 @@ impl MultipartUploadBackend for GcsBackend { let body = resp .bytes() .await - .map_err(|e| Error::reqwest("GCS: read complete multipart body", e))?; + .context("GCS: read complete multipart body")?; let error = quick_xml::de::from_reader::<_, XmlError>(body.as_ref()) .ok() diff --git a/objectstore-service/src/backend/in_memory.rs b/objectstore-service/src/backend/in_memory.rs index 93e74cfc..f5e30e5c 100644 --- a/objectstore-service/src/backend/in_memory.rs +++ b/objectstore-service/src/backend/in_memory.rs @@ -19,7 +19,7 @@ use super::common::{ DeleteResponse, GetResponse, HighVolumeBackend, MultipartUploadBackend, PutResponse, TieredGet, TieredMetadata, TieredWrite, Tombstone, }; -use crate::error::{Error, Result}; +use crate::error::{Error, ErrorKind, Result}; use crate::id::ObjectId; use crate::multipart::{ AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse, @@ -128,7 +128,9 @@ impl super::common::Backend for InMemoryBackend { let entry = self.store.lock().unwrap().get(id).cloned(); match entry { None => Ok(None), - Some(StoreEntry::Tombstone(_)) => Err(Error::UnexpectedTombstone), + Some(StoreEntry::Tombstone(_)) => { + Err(Error::new(ErrorKind::Internal).context("unexpected tombstone")) + } Some(StoreEntry::Object(mut metadata, bytes)) => { let total = bytes.len() as u64; metadata.size = Some(bytes.len()); @@ -136,7 +138,7 @@ impl super::common::Backend for InMemoryBackend { Some(range) => { let content_range = range .resolve(total) - .ok_or(Error::RangeNotSatisfiable { total })?; + .ok_or_else(|| Error::range_not_satisfiable(total))?; let sliced = bytes.slice(content_range.start as usize..=content_range.end as usize); (Some(content_range), sliced) @@ -193,7 +195,7 @@ impl HighVolumeBackend for InMemoryBackend { Some(range) => { let content_range = range .resolve(total) - .ok_or(Error::RangeNotSatisfiable { total })?; + .ok_or_else(|| Error::range_not_satisfiable(total))?; let sliced = bytes.slice(content_range.start as usize..=content_range.end as usize); (Some(content_range), sliced) @@ -289,7 +291,7 @@ impl MultipartUploadBackend for InMemoryBackend { let mut store = self.multipart_store.lock().unwrap(); let upload = store .get_mut(&(id.clone(), upload_id.clone())) - .ok_or_else(|| Error::generic("multipart upload not found"))?; + .ok_or_else(|| Error::new(ErrorKind::Internal).context("multipart upload not found"))?; upload.parts.insert( part_number, @@ -313,7 +315,7 @@ impl MultipartUploadBackend for InMemoryBackend { let store = self.multipart_store.lock().unwrap(); let upload = store .get(&(id.clone(), upload_id.clone())) - .ok_or_else(|| Error::generic("multipart upload not found"))?; + .ok_or_else(|| Error::new(ErrorKind::Internal).context("multipart upload not found"))?; let iter = upload .parts @@ -376,9 +378,9 @@ impl MultipartUploadBackend for InMemoryBackend { // the client can retry. let assembled = { let store = self.multipart_store.lock().unwrap(); - let upload = store - .get(&key) - .ok_or_else(|| Error::generic("multipart upload not found"))?; + let upload = store.get(&key).ok_or_else(|| { + Error::new(ErrorKind::Internal).context("multipart upload not found") + })?; for completed in &parts { match upload.parts.get(&completed.part_number) { diff --git a/objectstore-service/src/backend/local_fs.rs b/objectstore-service/src/backend/local_fs.rs index af0b233d..6d447a8b 100644 --- a/objectstore-service/src/backend/local_fs.rs +++ b/objectstore-service/src/backend/local_fs.rs @@ -1,6 +1,5 @@ //! Local filesystem backend for development and testing. -use std::io::ErrorKind; use std::path::PathBuf; use std::pin::pin; use std::time::SystemTime; @@ -15,7 +14,7 @@ use tokio_util::io::{ReaderStream, StreamReader}; use crate::backend::common::{ Backend, DeleteResponse, GetResponse, MultipartUploadBackend, PutResponse, }; -use crate::error::{Error, Result}; +use crate::error::{Error, ErrorKind, Result, ResultExt}; use crate::id::ObjectId; use crate::multipart::{ AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse, @@ -96,18 +95,16 @@ impl Backend for LocalFsBackend { let mut reader = pin!(StreamReader::new(stream)); let mut writer = BufWriter::new(file); - let metadata_json = serde_json::to_string(metadata).map_err(|cause| Error::Serde { - context: "failed to serialize metadata".to_string(), - cause, - })?; + let metadata_json = + serde_json::to_string(metadata).context("failed to serialize metadata")?; writer.write_all(metadata_json.as_bytes()).await?; writer.write_all(b"\n").await?; tokio::io::copy(&mut reader, &mut writer) .await .map_err(|e| match stream::unpack_client_error(&e) { - Some(ce) => Error::Client(ce), - None => e.into(), + Some(ce) => Error::from(ce), + None => Error::from(e), })?; writer.flush().await?; @@ -125,7 +122,7 @@ impl Backend for LocalFsBackend { let path = self.path.join(id.as_storage_path().to_string()); let file = match OpenOptions::new().read(true).open(path).await { Ok(file) => file, - Err(err) if err.kind() == ErrorKind::NotFound => { + Err(err) if err.kind() == std::io::ErrorKind::NotFound => { objectstore_log::debug!("Object not found"); return Ok(None); } @@ -136,24 +133,21 @@ impl Backend for LocalFsBackend { let mut metadata_line = String::new(); reader.read_line(&mut metadata_line).await?; let file_len = reader.get_ref().metadata().await?.len(); - let mut metadata: Metadata = - serde_json::from_str(metadata_line.trim_end()).map_err(|cause| Error::Serde { - context: "failed to deserialize metadata".to_string(), - cause, - })?; + let mut metadata: Metadata = serde_json::from_str(metadata_line.trim_end()) + .context("failed to deserialize metadata")?; let payload_size = file_len .checked_sub(metadata_line.len() as u64) - .ok_or_else(|| Error::generic("local-fs file corrupted: shorter than header"))?; + .ok_or_else(|| { + Error::new(ErrorKind::Internal) + .context("local-fs file corrupted: shorter than header") + })?; metadata.size = Some(payload_size as usize); let (content_range, stream) = match range { Some(byte_range) => { - let content_range = - byte_range - .resolve(payload_size) - .ok_or(Error::RangeNotSatisfiable { - total: payload_size, - })?; + let content_range = byte_range + .resolve(payload_size) + .ok_or_else(|| Error::range_not_satisfiable(payload_size))?; let payload_start = metadata_line.len() as u64 + content_range.start; reader.seek(std::io::SeekFrom::Start(payload_start)).await?; let limited = reader.take(content_range.len()); @@ -170,7 +164,7 @@ impl Backend for LocalFsBackend { let path = self.path.join(id.as_storage_path().to_string()); let result = tokio::fs::remove_file(path).await; if let Err(e) = &result - && e.kind() == ErrorKind::NotFound + && e.kind() == std::io::ErrorKind::NotFound { objectstore_log::debug!("Object not found"); } @@ -199,10 +193,8 @@ impl MultipartUploadBackend for LocalFsBackend { tokio::fs::create_dir_all(&dir).await?; let meta_path = dir.join("metadata.json"); - let metadata_json = serde_json::to_string(metadata).map_err(|cause| Error::Serde { - context: "failed to serialize multipart metadata".to_string(), - cause, - })?; + let metadata_json = + serde_json::to_string(metadata).context("failed to serialize multipart metadata")?; tokio::fs::write(meta_path, metadata_json).await?; Ok(upload_id) @@ -219,7 +211,7 @@ impl MultipartUploadBackend for LocalFsBackend { ) -> Result { let dir = self.multipart_dir(id, upload_id); if !tokio::fs::try_exists(&dir).await? { - return Err(Error::generic("multipart upload not found")); + return Err(Error::new(ErrorKind::Internal).context("multipart upload not found")); } let etag = format!("\"etag-{part_number}-{content_length}\""); @@ -229,10 +221,8 @@ impl MultipartUploadBackend for LocalFsBackend { "uploaded_at": SystemTime::now(), "size": content_length, }); - let header_line = serde_json::to_string(&header).map_err(|cause| Error::Serde { - context: "failed to serialize part header".to_string(), - cause, - })?; + let header_line = + serde_json::to_string(&header).context("failed to serialize part header")?; let part_path = dir.join(format!("{part_number}.part")); let file = OpenOptions::new() @@ -250,8 +240,8 @@ impl MultipartUploadBackend for LocalFsBackend { let _bytes_copied = tokio::io::copy(&mut reader, &mut writer) .await .map_err(|e| match stream::unpack_client_error(&e) { - Some(ce) => Error::Client(ce), - None => e.into(), + Some(ce) => Error::from(ce), + None => Error::from(e), })?; // TODO: validate bytes_copied against content_length and return a BadRequest-style @@ -275,7 +265,7 @@ impl MultipartUploadBackend for LocalFsBackend { ) -> Result { let dir = self.multipart_dir(id, upload_id); if !tokio::fs::try_exists(&dir).await? { - return Err(Error::generic("multipart upload not found")); + return Err(Error::new(ErrorKind::Internal).context("multipart upload not found")); } let mut entries = tokio::fs::read_dir(&dir).await?; @@ -299,11 +289,8 @@ impl MultipartUploadBackend for LocalFsBackend { let mut reader = BufReader::new(file); let mut header_line = String::new(); reader.read_line(&mut header_line).await?; - let header: serde_json::Value = - serde_json::from_str(header_line.trim_end()).map_err(|cause| Error::Serde { - context: "failed to deserialize part header".to_string(), - cause, - })?; + let header: serde_json::Value = serde_json::from_str(header_line.trim_end()) + .context("failed to deserialize part header")?; parts.push(Part { part_number: pn, @@ -353,17 +340,14 @@ impl MultipartUploadBackend for LocalFsBackend { ) -> Result { let dir = self.multipart_dir(id, upload_id); if !tokio::fs::try_exists(&dir).await? { - return Err(Error::generic("multipart upload not found")); + return Err(Error::new(ErrorKind::Internal).context("multipart upload not found")); } // Read metadata let meta_path = dir.join("metadata.json"); let meta_bytes = tokio::fs::read(&meta_path).await?; - let metadata: Metadata = - serde_json::from_slice(&meta_bytes).map_err(|cause| Error::Serde { - context: "failed to deserialize multipart metadata".to_string(), - cause, - })?; + let metadata: Metadata = serde_json::from_slice(&meta_bytes) + .context("failed to deserialize multipart metadata")?; // TODO: validate that parts are in ascending part_number order and reject with // InvalidPartOrder if not (matches S3/GCS behavior). Needs a proper client error variant. @@ -382,11 +366,8 @@ impl MultipartUploadBackend for LocalFsBackend { let mut reader = BufReader::new(file); let mut header_line = String::new(); reader.read_line(&mut header_line).await?; - let header: serde_json::Value = - serde_json::from_str(header_line.trim_end()).map_err(|cause| Error::Serde { - context: "failed to deserialize part header".to_string(), - cause, - })?; + let header: serde_json::Value = serde_json::from_str(header_line.trim_end()) + .context("failed to deserialize part header")?; let stored_etag = header["etag"].as_str().unwrap_or(""); if stored_etag != completed.etag { @@ -411,10 +392,8 @@ impl MultipartUploadBackend for LocalFsBackend { .await?; let mut writer = BufWriter::new(file); - let metadata_json = serde_json::to_string(&metadata).map_err(|cause| Error::Serde { - context: "failed to serialize metadata".to_string(), - cause, - })?; + let metadata_json = + serde_json::to_string(&metadata).context("failed to serialize metadata")?; writer.write_all(metadata_json.as_bytes()).await?; writer.write_all(b"\n").await?; @@ -841,7 +820,10 @@ mod tests { .unwrap(); match backend.get_object(&id, Some(ByteRange::From(100))).await { - Err(Error::RangeNotSatisfiable { total: 5 }) => {} + Err(err) if err.kind == ErrorKind::RangeNotSatisfiable => { + let total = err.range_total(); + assert_eq!(total, Some(5)); + } Err(other) => panic!("expected RangeNotSatisfiable, got: {other:?}"), Ok(_) => panic!("expected RangeNotSatisfiable, got Ok"), } diff --git a/objectstore-service/src/backend/s3_compatible.rs b/objectstore-service/src/backend/s3_compatible.rs index 12c1f665..3d849d76 100644 --- a/objectstore-service/src/backend/s3_compatible.rs +++ b/objectstore-service/src/backend/s3_compatible.rs @@ -9,11 +9,11 @@ use objectstore_types::range::{ByteRange, ContentRange}; use reqwest::header::HeaderMap; use reqwest::{Body, IntoUrl, Method, RequestBuilder, Response, StatusCode}; -use super::extensions::{ResponseExt, SendTraced}; +use super::extensions::{ResponseExt, SendTraced, classify_transport}; use crate::backend::common::{ self, Backend, DeleteResponse, GetResponse, MetadataResponse, PutResponse, }; -use crate::error::{Error, Result}; +use crate::error::{Error, ErrorKind, Result, ResultExt}; use crate::id::ObjectId; use crate::stream::ClientStream; @@ -147,10 +147,7 @@ where provider .get_token() .await - .map_err(|err| Error::Generic { - context: "S3: failed to get authentication token".to_owned(), - cause: Some(err.into()), - })? + .context("S3: failed to get authentication token")? .as_str(), ); } @@ -172,13 +169,8 @@ where if let Some(r) = range { builder = builder.header(reqwest::header::RANGE, r.to_header_value()); } - let response = builder - .send_traced() - .await - .map_err(|cause| Error::Reqwest { - context: "S3: failed to send request".to_string(), - cause, - })?; + let response = + classify_transport(builder.send_traced().await, "S3: failed to send request")?; if response.status() == StatusCode::NOT_FOUND { objectstore_log::debug!("Object not found"); @@ -193,8 +185,8 @@ where .and_then(|v| v.to_str().ok()); let total = raw.and_then(ContentRange::parse_unsatisfiable_total); let err = match total { - Some(total) => Error::RangeNotSatisfiable { total }, - None => Error::generic(format!( + Some(total) => Error::range_not_satisfiable(total), + None => Error::new(ErrorKind::Internal).context(format!( "S3: 416 response with invalid Content-Range: {raw:?}" )), }; @@ -212,9 +204,9 @@ where .get(reqwest::header::CONTENT_RANGE) .and_then(|v| v.to_str().ok()) .and_then(|s| s.parse::().ok()) - .ok_or_else(|| Error::Generic { - context: "S3: 206 response missing valid Content-Range header".to_owned(), - cause: None, + .ok_or_else(|| { + Error::new(ErrorKind::Internal) + .context("S3: 206 response missing valid Content-Range header") })?; metadata.size = Some(range.total as usize); Some(range) @@ -348,15 +340,11 @@ impl Backend for S3CompatibleBackend { #[tracing::instrument(level = "debug", skip(self))] async fn delete_object(&self, id: &ObjectId) -> Result { objectstore_log::debug!("Deleting from s3_compatible backend"); - let response = self - .request(Method::DELETE, self.object_url(id)) - .await? - .send_traced() - .await - .map_err(|cause| Error::Reqwest { - context: "S3: failed to send delete request".to_string(), - cause, - })?; + let request = self.request(Method::DELETE, self.object_url(id)).await?; + let response = classify_transport( + request.send_traced().await, + "S3: failed to send delete request", + )?; // Do not error for objects that do not exist. if response.status() == StatusCode::NOT_FOUND { diff --git a/objectstore-service/src/backend/testing.rs b/objectstore-service/src/backend/testing.rs index 24034f24..1845e5fd 100644 --- a/objectstore-service/src/backend/testing.rs +++ b/objectstore-service/src/backend/testing.rs @@ -24,10 +24,11 @@ //! _inner: &InMemoryBackend, //! _id: &ObjectId, //! ) -> Result { -//! Err(crate::error::Error::Io(std::io::Error::new( +//! Err(std::io::Error::new( //! std::io::ErrorKind::ConnectionRefused, //! "simulated delete failure", -//! ))) +//! ) +//! .into()) //! } //! } //! diff --git a/objectstore-service/src/backend/tiered.rs b/objectstore-service/src/backend/tiered.rs index a6af1342..71c0d9da 100644 --- a/objectstore-service/src/backend/tiered.rs +++ b/objectstore-service/src/backend/tiered.rs @@ -114,7 +114,7 @@ use crate::backend::common::{ MultipartUploadBackend, PutResponse, TieredGet, TieredMetadata, TieredWrite, Tombstone, }; use crate::backend::{HighVolumeStorageConfig, MultipartUploadStorageConfig}; -use crate::error::{Error, Result}; +use crate::error::{Error, ErrorKind, Result, ResultExt}; use crate::id::ObjectId; use crate::multipart::{ AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse, @@ -585,8 +585,7 @@ impl TryInto for TieredUploadId { type Error = Error; fn try_into(self) -> Result { - let json = - serde_json::to_vec(&self).map_err(|e| Error::serde("encoding multipart token", e))?; + let json = serde_json::to_vec(&self).context("encoding multipart token")?; Ok(UploadId::new( base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(json), )?) @@ -599,8 +598,11 @@ impl TryFrom<&UploadId> for TieredUploadId { fn try_from(value: &UploadId) -> Result { let json = base64::engine::general_purpose::URL_SAFE_NO_PAD .decode(value.as_bytes()) - .map_err(|e| Error::generic(format!("invalid multipart upload ID: {e}")))?; - serde_json::from_slice(&json).map_err(|e| Error::serde("decoding multipart token", e)) + .map_err(|e| { + Error::new(ErrorKind::InvalidInput) + .context(format!("invalid multipart upload ID: {e}")) + })?; + serde_json::from_slice(&json).kind(ErrorKind::InvalidInput) } } @@ -819,9 +821,8 @@ impl MultipartUploadBackend for TieredStorage { physical = ?physical, "complete_multipart call succeeded on long_term backend, but subsequent get_metadata found no object" ); - return Err(Error::generic( - "completed multipart object not found in long-term storage", - )); + return Err(Error::new(ErrorKind::Internal) + .context("completed multipart object not found in long-term storage")); } // Failed to `get_metadata`, cannot proceed. (Err(get_metadata_err), maybe_complete_multipart_err) => { @@ -877,7 +878,6 @@ mod tests { use crate::backend::changelog::{InMemoryChangeLog, NoopChangeLog}; use crate::backend::in_memory::InMemoryBackend; use crate::backend::testing::{Hooks, TestBackend}; - use crate::error::Error; use crate::id::ObjectContext; use crate::stream::{self, ClientStream}; @@ -1149,10 +1149,11 @@ mod tests { _inner: &InMemoryBackend, _id: &ObjectId, ) -> Result { - Err(Error::Io(std::io::Error::new( + Err(std::io::Error::new( std::io::ErrorKind::ConnectionRefused, "simulated long-term delete failure", - ))) + ) + .into()) } } @@ -1286,10 +1287,11 @@ mod tests { // simulate a network error _after_ commit went through inner.compare_and_write(id, current, write).await?; } - Err(Error::Io(std::io::Error::new( + Err(std::io::Error::new( std::io::ErrorKind::TimedOut, "simulated compare_and_write failure", - ))) + ) + .into()) } } @@ -1937,10 +1939,11 @@ mod tests { .complete_multipart(id, upload_id, parts) .await .unwrap(); - Err(Error::Io(std::io::Error::new( + Err(std::io::Error::new( std::io::ErrorKind::TimedOut, "simulated network error on complete_multipart", - ))) + ) + .into()) } async fn get_metadata( @@ -1948,10 +1951,11 @@ mod tests { _inner: &InMemoryBackend, _id: &ObjectId, ) -> Result { - Err(Error::Io(std::io::Error::new( + Err(std::io::Error::new( std::io::ErrorKind::TimedOut, "simulated network error on get_metadata", - ))) + ) + .into()) } } @@ -2056,10 +2060,10 @@ mod tests { let mut attempt = self.attempt.lock().await; *attempt += 1; if *attempt == 1 { - Err(Error::Io(std::io::Error::new( - std::io::ErrorKind::TimedOut, - "simulated network error", - ))) + Err( + std::io::Error::new(std::io::ErrorKind::TimedOut, "simulated network error") + .into(), + ) } else { Ok(inner .complete_multipart(id, upload_id, parts) @@ -2192,10 +2196,10 @@ mod tests { let mut attempt = self.attempt.lock().await; *attempt += 1; if *attempt == 1 { - Err(Error::Io(std::io::Error::new( - std::io::ErrorKind::TimedOut, - "simulated network error", - ))) + Err( + std::io::Error::new(std::io::ErrorKind::TimedOut, "simulated network error") + .into(), + ) } else { inner.get_metadata(id).await } diff --git a/objectstore-service/src/concurrency.rs b/objectstore-service/src/concurrency.rs index d3e75eaf..2ce914eb 100644 --- a/objectstore-service/src/concurrency.rs +++ b/objectstore-service/src/concurrency.rs @@ -16,7 +16,7 @@ use futures_util::FutureExt; use sentry::{Hub, SentryFutureExt, TransactionContext}; use tokio::sync::{Notify, OwnedSemaphorePermit, Semaphore}; -use crate::error::{Error, Result}; +use crate::error::{Error, ErrorKind, Result}; /// Interval for the periodic metrics emitter. const EMITTER_INTERVAL: Duration = Duration::from_secs(1); @@ -80,26 +80,26 @@ impl ConcurrencyLimiter { /// Acquires a single concurrency permit, waiting if necessary. /// - /// If all `max + queue` slots are occupied, returns - /// [`Error::AtCapacity`] immediately. Otherwise, waits up to the - /// configured queue timeout for an execution permit to become - /// available. Returns [`Error::AtCapacity`] on timeout. + /// If all `max + queue` slots are occupied, returns an + /// [`ErrorKind::AtCapacity`] error immediately. Otherwise, waits up to the + /// configured queue timeout for an execution permit to become available. + /// Returns an [`ErrorKind::AtCapacity`] error on timeout. pub async fn acquire(&self) -> Result { if self.tasks_total == 0 { - return Err(Error::AtCapacity); + return Err(Error::new(ErrorKind::AtCapacity)); } let queue_permit = self .queue .clone() .try_acquire_owned() - .map_err(|_| Error::AtCapacity)?; + .map_err(|_| Error::new(ErrorKind::AtCapacity))?; let acquire = self.tasks.clone().acquire_owned(); let task_permit = tokio::time::timeout(self.timeout, acquire) .await - .map_err(|_| Error::AtCapacity)? - .map_err(|_| Error::AtCapacity)?; + .map_err(|_| Error::new(ErrorKind::AtCapacity))? + .map_err(|_| Error::new(ErrorKind::AtCapacity))?; Ok(ConcurrencyPermit { task_permit: Some(task_permit), @@ -115,19 +115,19 @@ impl ConcurrencyLimiter { /// the execution semaphore and the queue semaphore are checked, so the /// total-capacity invariant (`max + queue`) is maintained. /// - /// Returns [`Error::AtCapacity`] when fewer than `count` permits are available. + /// Returns an [`ErrorKind::AtCapacity`] error when fewer than `count` permits are available. pub fn try_acquire_many(&self, count: u32) -> Result { let queue_permit = self .queue .clone() .try_acquire_many_owned(count) - .map_err(|_| Error::AtCapacity)?; + .map_err(|_| Error::new(ErrorKind::AtCapacity))?; let task_permit = self .tasks .clone() .try_acquire_many_owned(count) - .map_err(|_| Error::AtCapacity)?; + .map_err(|_| Error::new(ErrorKind::AtCapacity))?; Ok(ConcurrencyPermit { task_permit: Some(task_permit), @@ -285,10 +285,10 @@ where } .bind_hub(new_hub), ); - rx.await.map_err(|_| { - objectstore_log::error!(!!&Error::Dropped, operation, "Task failed"); - Error::Dropped + let err = Error::new(ErrorKind::Internal).context("task dropped"); + objectstore_log::error!(!!&err, operation, "Task failed"); + err })? } @@ -297,7 +297,7 @@ mod tests { use std::sync::atomic::{AtomicU32, Ordering}; use super::*; - use crate::error::Error; + use crate::error::ErrorKind; #[test] fn available_permits_tracks_held() { @@ -347,7 +347,7 @@ mod tests { let _permit = limiter.try_acquire_many(1).unwrap(); let result = limiter.try_acquire_many(1); - assert!(matches!(result, Err(Error::AtCapacity))); + assert_eq!(result.unwrap_err().kind, ErrorKind::AtCapacity); } #[test] @@ -422,10 +422,10 @@ mod tests { let p1 = limiter.try_acquire_many(1).unwrap(); let p2 = limiter.try_acquire_many(1).unwrap(); - assert!(matches!( - limiter.try_acquire_many(1), - Err(Error::AtCapacity) - )); + assert_eq!( + limiter.try_acquire_many(1).unwrap_err().kind, + ErrorKind::AtCapacity + ); drop(p1); assert!(limiter.try_acquire_many(1).is_ok()); @@ -475,7 +475,7 @@ mod tests { tokio::time::sleep(Duration::from_secs(2)).await; let result = waiter.await.unwrap(); - assert!(matches!(result, Err(Error::AtCapacity))); + assert_eq!(result.unwrap_err().kind, ErrorKind::AtCapacity); assert_eq!(limiter.queued_permits(), 0); } @@ -491,7 +491,7 @@ mod tests { assert_eq!(limiter.queued_permits(), 1); let result = limiter.acquire().await; - assert!(matches!(result, Err(Error::AtCapacity))); + assert_eq!(result.unwrap_err().kind, ErrorKind::AtCapacity); } #[test] @@ -499,10 +499,10 @@ mod tests { let limiter = ConcurrencyLimiter::new(1).with_queue(2, Duration::from_secs(5)); let _p = limiter.try_acquire_many(1).unwrap(); - assert!(matches!( - limiter.try_acquire_many(1), - Err(Error::AtCapacity) - )); + assert_eq!( + limiter.try_acquire_many(1).unwrap_err().kind, + ErrorKind::AtCapacity + ); } #[test] @@ -512,10 +512,10 @@ mod tests { let _bulk = limiter.try_acquire_many(3).unwrap(); assert_eq!(limiter.used_permits(), 3); - assert!(matches!( - limiter.try_acquire_many(2), - Err(Error::AtCapacity) - )); + assert_eq!( + limiter.try_acquire_many(2).unwrap_err().kind, + ErrorKind::AtCapacity + ); let _single = limiter.try_acquire_many(1).unwrap(); assert_eq!(limiter.used_permits(), 4); @@ -562,7 +562,7 @@ mod tests { let start = tokio::time::Instant::now(); let result = limiter.acquire().await; - assert!(matches!(result, Err(Error::AtCapacity))); + assert_eq!(result.unwrap_err().kind, ErrorKind::AtCapacity); assert_eq!(start.elapsed(), Duration::ZERO); } diff --git a/objectstore-service/src/error.rs b/objectstore-service/src/error.rs index 9643dd43..0754f8b8 100644 --- a/objectstore-service/src/error.rs +++ b/objectstore-service/src/error.rs @@ -1,14 +1,21 @@ //! Error types for service and backend operations. //! -//! [`Error`] covers I/O, serialization, HTTP, metadata, authentication, -//! and backend-specific failures. [`Result`] is the corresponding alias. +//! [`Error`] combines an [`ErrorKind`] (the *classification* of the failure, +//! independent of any HTTP semantics) with an [`anyhow::Error`] diagnostic. +//! The diagnostic preserves the original error and every context frame, while +//! the kind can be changed independently. [`Result`] is the corresponding alias. +//! +//! The default path is `?`: a foreign error converts via one of the [`From`] +//! impls into an [`ErrorKind::Internal`] error carrying the original diagnostic. +//! To override the classification or attach context without a `map_err`, use the +//! [`ResultExt`] extension trait's [`kind`](ResultExt::kind) and +//! [`context`](ResultExt::context) methods. use std::any::Any; +use std::borrow::Cow; use std::fmt; use objectstore_log::Level; -use reqwest::StatusCode; -use thiserror::Error as ThisError; use crate::stream::ClientError; @@ -44,117 +51,122 @@ impl fmt::Display for BackendDetail { } } +/// The category of a service error. +/// +/// These kinds describe the *cause* of a failure, independent of any HTTP +/// semantics. It is up to the API layer to decide how to map each kind onto a +/// status code, and which kinds to group together. +/// +/// Users should rely on the kind for classification and handling, rather than matching on +/// specific variants. +#[derive(Copy, Clone, Debug, Eq, PartialEq)] +pub enum ErrorKind { + /// Error originating from a client-supplied request body stream. + ClientStream, + /// Malformed client input (bad metadata, JSON, upload id, …). + InvalidInput, + /// The requested byte range is not satisfiable for the object's size. + RangeNotSatisfiable, + /// This service instance has reached its own concurrency limit and is + /// shedding load. + AtCapacity, + /// A storage backend rejected the operation with a rate-limit response. + BackendRateLimited, + /// A storage backend timed out. + BackendTimeout, + /// A storage backend is temporarily unavailable. + BackendUnavailable, + /// Operation unsupported by this service instance. + NotImplemented, + /// Internal failure. + Internal, +} + +impl ErrorKind { + /// Returns a short human-readable description of this kind. + fn as_str(self) -> &'static str { + match self { + ErrorKind::ClientStream => "client stream error", + ErrorKind::InvalidInput => "invalid input", + ErrorKind::RangeNotSatisfiable => "range not satisfiable", + ErrorKind::AtCapacity => "at capacity", + ErrorKind::BackendRateLimited => "backend rate limited", + ErrorKind::BackendTimeout => "backend timeout", + ErrorKind::BackendUnavailable => "backend unavailable", + ErrorKind::NotImplemented => "not implemented", + ErrorKind::Internal => "internal error", + } + } +} + +impl fmt::Display for ErrorKind { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + /// Error type for service operations. -#[derive(Debug, ThisError)] -pub enum Error { - /// IO errors related to payload streaming or file operations. - #[error("i/o error: {0}")] - Io(#[from] std::io::Error), +/// +/// See the [module docs](self) for the overall design. Construct one either by +/// converting a foreign error (via `?` or [`ResultExt`]) or with [`Error::new`] +/// for failures without an underlying error. +pub struct Error { + /// The classification of this error. + pub kind: ErrorKind, + /// The diagnostic error and its context chain. + pub inner: anyhow::Error, +} - /// Error originating from a client-supplied input stream. - /// - /// Indicates the client is at fault (e.g. dropped connection mid-upload) and should - /// map to a 4xx response rather than a 5xx. - #[error("error reading client stream: {0}")] - Client(#[from] ClientError), - - /// Errors related to de/serialization. - #[error("serde error: {context}")] - Serde { - /// Context describing what was being serialized/deserialized. - context: String, - /// The underlying serde error. - #[source] - cause: serde_json::Error, - }, - - /// All errors stemming from the reqwest client, used in multiple backends to send requests to - /// e.g. GCP APIs. - /// These can be network errors encountered when sending the requests, but can also indicate - /// errors returned by the API itself. - #[error("reqwest error: {context}")] - Reqwest { - /// Context describing the request that failed. - context: String, - /// The underlying reqwest error. - #[source] - cause: reqwest::Error, - }, - - /// An HTTP error response from a storage backend (e.g., GCS, S3). - /// - /// Unlike [`Reqwest`](Self::Reqwest), which covers transport-level failures, this variant - /// captures application-level error responses where the server returned a 4xx/5xx status code - /// along with a structured error body. - #[error("{context} ({status}). {detail}")] - BackendResponse { - /// Context describing the request that failed. - context: &'static str, - /// The HTTP status code returned by the backend. - status: StatusCode, - /// Parsed error code and message from the response body. - detail: BackendDetail, - }, - - /// Errors related to de/serialization and parsing of object metadata. - #[error("metadata error: {0}")] - Metadata(#[from] objectstore_types::metadata::Error), - - /// Errors encountered when attempting to authenticate with GCP. - #[error("GCP authentication error: {0}")] - GcpAuth(#[from] gcp_auth::Error), - - /// A spawned service task panicked. - #[error("service task failed: {0}")] - Panic(String), - - /// A spawned service task was dropped before it could deliver its result. +impl Error { + /// Creates an error of the given kind using the kind's default message. /// - /// This is an unexpected condition that can occur when the runtime drops the task for unknown - /// reasons. - #[error("task dropped")] - Dropped, + /// Attach a more specific message with [`context`](Self::context). + pub fn new(kind: ErrorKind) -> Self { + Self { + kind, + inner: anyhow::Error::msg(kind.as_str()), + } + } - /// A redirect tombstone was encountered at a place where it is not supported. - /// - /// This indicates a caller bug — tombstone-aware reads must go through the - /// [`HighVolumeBackend`](crate::backend::common::HighVolumeBackend) methods. - #[error("unexpected tombstone")] - UnexpectedTombstone, + fn from_source(kind: ErrorKind, source: E) -> Self + where + E: std::error::Error + Send + Sync + 'static, + { + Self { + kind, + inner: anyhow::Error::new(source), + } + } - /// The requested byte range is not satisfiable for the object's size. - #[error("range not satisfiable (object size: {total} bytes)")] - RangeNotSatisfiable { - /// Total size of the object in bytes. - total: u64, - }, - - /// The service has reached its concurrency limit and cannot accept more operations. - #[error("concurrency limit reached")] - AtCapacity, + /// Creates a range error carrying the object's total size. + pub fn range_not_satisfiable(total: u64) -> Self { + Self::from_source( + ErrorKind::RangeNotSatisfiable, + RangeNotSatisfiableError { total }, + ) + } - /// Any other error stemming from one of the storage backends, which might be specific to that - /// backend or to a certain operation. - #[error("storage backend error: {context}")] - Generic { - /// Context describing the operation that failed. - context: String, - /// The underlying error, if available. - #[source] - cause: Option>, - }, - - /// The functionality is not implemented by this instance of the service. - #[error("not implemented")] - NotImplemented, + /// Returns the object size carried by a range error, if present. + pub fn range_total(&self) -> Option { + self.downcast_ref::() + .map(|error| error.total) + } - /// Invalid upload ID (e.g. path traversal attempt). - #[error(transparent)] - InvalidUploadId(#[from] objectstore_types::multipart::InvalidUploadId), -} + /// Attaches another context frame to this error. + #[must_use] + pub fn context(mut self, context: impl Into>) -> Self { + self.inner = self.inner.context(context.into()); + self + } -impl Error { - /// Creates an [`Error::Panic`] from a panic payload, extracting the message. + /// Changes this error's classification without changing its diagnostic. + #[must_use] + pub fn kind(mut self, kind: ErrorKind) -> Self { + self.kind = kind; + self + } + + /// Creates an [`ErrorKind::Internal`] error from a panic payload, extracting the message. pub fn panic(payload: Box) -> Self { let msg = if let Some(s) = payload.downcast_ref::<&str>() { (*s).to_owned() @@ -163,57 +175,200 @@ impl Error { } else { "unknown panic".to_owned() }; - Self::Panic(msg) + Self::new(ErrorKind::Internal).context(format!("service task panicked: {msg}")) } - /// Creates an [`Error::Reqwest`] from a reqwest error with context. - pub fn reqwest(context: impl Into, cause: reqwest::Error) -> Self { - Self::Reqwest { - context: context.into(), - cause, - } + /// Iterates over the complete diagnostic chain, outermost context first. + pub fn chain(&self) -> anyhow::Chain<'_> { + self.inner.chain() } - /// Creates an [`Error::Serde`] from a serde error with context. - pub fn serde(context: impl Into, cause: serde_json::Error) -> Self { - Self::Serde { - context: context.into(), - cause, - } + /// Downcasts any error or context value in the diagnostic chain. + pub fn downcast_ref(&self) -> Option<&E> + where + E: fmt::Display + fmt::Debug + Send + Sync + 'static, + { + self.inner.downcast_ref() } - /// Creates an [`Error::Generic`] with a context string and no cause. - pub fn generic(context: impl Into) -> Self { - Self::Generic { - context: context.into(), - cause: None, + /// Returns the appropriate log level for this error. + pub fn level(&self) -> Level { + match self.kind { + ErrorKind::ClientStream | ErrorKind::InvalidInput | ErrorKind::RangeNotSatisfiable => { + Level::DEBUG + } + ErrorKind::AtCapacity + | ErrorKind::BackendRateLimited + | ErrorKind::BackendTimeout + | ErrorKind::BackendUnavailable => Level::WARN, + ErrorKind::NotImplemented | ErrorKind::Internal => Level::ERROR, } } +} - /// Returns the appropriate log level for this error. - pub fn level(&self) -> Level { - match self { - // Malformed client input at DEBUG level - Self::Client(_) => Level::DEBUG, - Self::Metadata(_) => Level::DEBUG, - Self::RangeNotSatisfiable { .. } => Level::DEBUG, - // Like rate limits, we treat capacity errors as warnings - Self::AtCapacity => Level::WARN, - // All other errors are service or backend failures - Self::Io(_) => Level::ERROR, - Self::Serde { .. } => Level::ERROR, - Self::Reqwest { .. } => Level::ERROR, - Self::BackendResponse { .. } => Level::ERROR, - Self::GcpAuth(_) => Level::ERROR, - Self::Panic(_) => Level::ERROR, - Self::Dropped => Level::ERROR, - Self::UnexpectedTombstone => Level::ERROR, - Self::NotImplemented => Level::ERROR, - Self::InvalidUploadId(_) => Level::DEBUG, - Self::Generic { .. } => Level::ERROR, +impl fmt::Display for Error { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt::Display::fmt(&self.inner, f) + } +} + +impl fmt::Debug for Error { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt::Debug::fmt(&self.inner, f) + } +} + +impl std::error::Error for Error { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + let inner: &(dyn std::error::Error + Send + Sync + 'static) = self.inner.as_ref(); + inner.source() + } +} + +/// The object's total size, carried in the diagnostic chain of an +/// [`ErrorKind::RangeNotSatisfiable`] error. +#[derive(Debug)] +struct RangeNotSatisfiableError { + total: u64, +} + +impl fmt::Display for RangeNotSatisfiableError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + f, + "range not satisfiable (object size: {} bytes)", + self.total + ) + } +} + +impl std::error::Error for RangeNotSatisfiableError {} + +/// Generates `From for Error` impls that classify `T` as [`ErrorKind::Internal`] +/// and preserve it in the diagnostic chain. +macro_rules! impl_internal_from { + ($($ty:ty),* $(,)?) => { + $( + impl From<$ty> for Error { + fn from(source: $ty) -> Self { + Self::from_source(ErrorKind::Internal, source) + } + } + )* + }; +} + +impl_internal_from!( + std::io::Error, + std::num::ParseIntError, + std::num::TryFromIntError, + std::str::Utf8Error, + std::string::FromUtf8Error, + std::time::SystemTimeError, + serde_json::Error, + reqwest::Error, + reqwest::header::InvalidHeaderName, + reqwest::header::InvalidHeaderValue, + gcp_auth::Error, + url::ParseError, + quick_xml::DeError, + quick_xml::SeError, + bigtable_rs::bigtable::Error, +); + +/// Converts an [`anyhow::Error`] into an [`ErrorKind::Internal`] error without +/// altering its diagnostic chain. +impl From for Error { + fn from(inner: anyhow::Error) -> Self { + Self { + kind: ErrorKind::Internal, + inner, } } } +/// Converts a client-stream fault into an [`ErrorKind::ClientStream`] error. +impl From for Error { + fn from(source: ClientError) -> Self { + Self::from_source(ErrorKind::ClientStream, source) + } +} + +/// Converts a metadata parse error into an [`ErrorKind::InvalidInput`] error. +impl From for Error { + fn from(source: objectstore_types::metadata::Error) -> Self { + Self::from_source(ErrorKind::InvalidInput, source) + } +} + +/// Converts an invalid upload id into an [`ErrorKind::InvalidInput`] error. +impl From for Error { + fn from(source: objectstore_types::multipart::InvalidUploadId) -> Self { + Self::from_source(ErrorKind::InvalidInput, source) + } +} + +/// Extension trait for reclassifying and annotating a [`Result`]'s error as an +/// [`Error`], without a `map_err`. +pub trait ResultExt { + /// Sets the [`ErrorKind`] of the error without changing its diagnostic. + fn kind(self, kind: ErrorKind) -> Result; + + /// Attaches another context frame to the error. + fn context(self, context: impl Into>) -> Result; +} + +impl> ResultExt for Result { + fn kind(self, kind: ErrorKind) -> Result { + self.map_err(|error| error.into().kind(kind)) + } + + fn context(self, context: impl Into>) -> Result { + self.map_err(|error| error.into().context(context)) + } +} + /// Result type for service operations. pub type Result = std::result::Result; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn context_accumulates_without_changing_kind() { + let error = Err::<(), _>(std::io::Error::other("root cause")) + .context("reading configuration") + .kind(ErrorKind::BackendUnavailable) + .context("starting backend") + .unwrap_err(); + + assert_eq!(error.kind, ErrorKind::BackendUnavailable); + assert_eq!(error.to_string(), "starting backend"); + assert!(error.downcast_ref::().is_some()); + assert_eq!( + error.chain().map(ToString::to_string).collect::>(), + ["starting backend", "reading configuration", "root cause",] + ); + assert!(format!("{error:?}").contains("Caused by:")); + } + + #[test] + fn range_total_survives_context() { + let error = Error::range_not_satisfiable(42).context("reading object range"); + + assert_eq!(error.kind, ErrorKind::RangeNotSatisfiable); + assert_eq!(error.range_total(), Some(42)); + } + + #[test] + fn source_skips_displayed_outer_message() { + let error = Error::from(std::io::Error::other("root")).context("outer"); + + assert_eq!(error.to_string(), "outer"); + assert_eq!( + std::error::Error::source(&error).unwrap().to_string(), + "root" + ); + } +} diff --git a/objectstore-service/src/service.rs b/objectstore-service/src/service.rs index 1e70f2bd..de347338 100644 --- a/objectstore-service/src/service.rs +++ b/objectstore-service/src/service.rs @@ -14,7 +14,7 @@ use objectstore_types::range::{ByteRange, ContentRange}; use crate::backend::common::Backend; use crate::backend::counting::CountingBackend; use crate::concurrency::ConcurrencyLimiter; -use crate::error::{Error, Result}; +use crate::error::{Error, ErrorKind, Result}; use crate::id::{ObjectContext, ObjectId}; use crate::multipart::{ AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse, @@ -62,14 +62,16 @@ pub const DEFAULT_CONCURRENCY_LIMIT: u32 = 500; /// blocked. Call [`join`](StorageService::join) during shutdown to wait for /// outstanding cleanup. Operations are also isolated from panics in backend /// code — a failure in one operation does not bring down other in-flight work. -/// See [`Error::Panic`]. +/// See an [`ErrorKind::Internal`] error. /// /// # Concurrency Limit /// /// A [`ConcurrencyLimiter`] caps the number of in-flight backend operations. /// Pass a custom limiter via /// [`with_concurrency`](StorageService::with_concurrency); without one the -/// default is [`DEFAULT_CONCURRENCY_LIMIT`] permits with no queue. +/// default is [`DEFAULT_CONCURRENCY_LIMIT`] permits with no queue. Operations +/// that cannot be served within the configured queue depth and timeout fail +/// with an [`ErrorKind::AtCapacity`] error. #[derive(Clone, Debug)] pub struct StorageService { inner: Arc, @@ -120,13 +122,13 @@ impl StorageService { /// Operations are executed concurrently up to a window derived from the /// service's current capacity. The permits for that window are reserved /// upfront — if the service is at capacity, this returns - /// [`Error::AtCapacity`] immediately before any operations are read. + /// [`ErrorKind::AtCapacity`] immediately before any operations are read. pub fn stream(&self) -> Result { let available = self.tasks_available(); let window = available.div_ceil(10); let acquire_result = match window { - 0 => Err(Error::AtCapacity), + 0 => Err(Error::new(ErrorKind::AtCapacity)), _ => self.concurrency.try_acquire_many(window), }; let reservation = acquire_result.inspect_err(|_| { @@ -163,9 +165,9 @@ impl StorageService { /// Spawns a future in a separate task and awaits its result. /// - /// Returns [`Error::AtCapacity`] if the concurrency limit is reached, - /// [`Error::Panic`] if the spawned task panics (the panic message - /// is captured for diagnostics), or [`Error::Dropped`] if the task is + /// Returns [`ErrorKind::AtCapacity`] if the concurrency limit is reached, + /// an [`ErrorKind::Internal`] error if the spawned task panics (the panic message + /// is captured for diagnostics), or an [`ErrorKind::Internal`] error if the task is /// dropped before sending its result. /// /// Emits `service.task.start` (counter) after acquiring a permit and @@ -380,7 +382,7 @@ mod tests { use crate::backend::in_memory::InMemoryBackend; use crate::backend::testing::{Hooks, TestBackend}; use crate::backend::tiered::TieredStorage; - use crate::error::Error; + use crate::error::ErrorKind; use crate::stream::{self, ClientStream}; fn make_context() -> ObjectContext { @@ -576,10 +578,15 @@ mod tests { let id = ObjectId::new(make_context(), "panic-test".into()); let result = service.get_object(id, None).await; - let Err(Error::Panic(msg)) = result else { - panic!("expected Panic error"); + let Err(err) = result else { + panic!("expected panic error"); }; - assert!(msg.contains("intentional panic in get_object"), "{msg}"); + assert_eq!(err.kind, ErrorKind::Internal); + let message = format!("{err:#}"); + assert!( + message.contains("intentional panic in get_object"), + "{message}" + ); } /// In-memory backend with optional synchronization for `put_object`. @@ -718,7 +725,7 @@ mod tests { .await; assert!( - matches!(result, Err(Error::AtCapacity)), + matches!(&result, Err(e) if e.kind == ErrorKind::AtCapacity), "expected AtCapacity, got {result:?}" ); @@ -771,13 +778,16 @@ mod tests { // First operation panics — the permit must still be released. let id = ObjectId::new(make_context(), "panic-permit".into()); - let result = service.get_object(id.clone(), None).await; - assert!(matches!(result, Err(Error::Panic(_)))); + let Err(err) = service.get_object(id.clone(), None).await else { + panic!("expected panic error"); + }; + assert_eq!(err.kind, ErrorKind::Internal); + assert!(format!("{err:#}").contains("panicked")); // Second operation should succeed in acquiring the permit (not AtCapacity). let result = service.get_object(id, None).await; assert!( - !matches!(result, Err(Error::AtCapacity)), + !matches!(&result, Err(e) if e.kind == ErrorKind::AtCapacity), "permit was not released after panic" ); } diff --git a/objectstore-service/src/stream.rs b/objectstore-service/src/stream.rs index efaf9c43..e992c563 100644 --- a/objectstore-service/src/stream.rs +++ b/objectstore-service/src/stream.rs @@ -66,7 +66,7 @@ impl From for io::Error { /// Uses [`ClientError`] as the error type so that a dropped or interrupted /// client connection is distinguishable from a backend I/O failure. Backends /// that detect a [`ClientError`] (via [`unpack_client_error`]) can surface it -/// as [`crate::error::Error::Client`], which the server maps to HTTP 400 rather +/// as [`crate::error::ErrorKind::ClientStream`], which the server maps to HTTP 400 rather /// than 500. /// /// Use [`single`] to construct a single-chunk `ClientStream` from an owned value. @@ -81,7 +81,7 @@ pub type ClientStream = BoxStream<'static, Result>; /// value is a `ClientError`. /// /// Use this in `put_object` implementations to reclassify body-stream errors -/// as [`crate::error::Error::Client`] instead of an opaque server error. +/// as [`crate::error::ErrorKind::ClientStream`] instead of an opaque server error. pub fn unpack_client_error(err: &E) -> Option where E: Error + 'static, diff --git a/objectstore-service/src/streaming.rs b/objectstore-service/src/streaming.rs index e7811885..828b50aa 100644 --- a/objectstore-service/src/streaming.rs +++ b/objectstore-service/src/streaming.rs @@ -16,7 +16,7 @@ //! //! This means: //! - If the service is at capacity, [`StorageService::stream`](crate::service::StorageService::stream) fails immediately with -//! [`Error::AtCapacity`] before any operations are read. +//! [`ErrorKind::AtCapacity`](crate::error::ErrorKind::AtCapacity) before any operations are read. //! - During execution, operations call the storage backend directly without acquiring //! additional per-operation permits. //! @@ -28,8 +28,8 @@ //! `window × max_operation_size`. Results are yielded in completion order. //! //! Each operation is wrapped in a [`tokio::spawn`] for panic isolation: a panic in -//! one operation surfaces as [`Error::Panic`] for that item and does not affect the -//! others. +//! one operation surfaces as an [`ErrorKind::Internal`](crate::error::ErrorKind::Internal) error +//! for that item and does not affect the others. //! //! ## Future Scope //! @@ -332,7 +332,7 @@ mod tests { use crate::backend::in_memory::InMemoryBackend; use crate::backend::testing::{Hooks, TestBackend}; use crate::concurrency::ConcurrencyLimiter; - use crate::error::Error; + use crate::error::{Error, ErrorKind}; use crate::service::StorageService; use crate::stream::{self, ClientStream}; @@ -362,7 +362,7 @@ mod tests { #[test] fn at_capacity_when_no_permits() { let service = make_service_with_limit(0); - assert!(matches!(service.stream(), Err(Error::AtCapacity))); + assert!(matches!(service.stream(), Err(e) if e.kind == ErrorKind::AtCapacity)); } #[test] @@ -614,7 +614,7 @@ mod tests { // Permit is held — stream() must fail immediately with AtCapacity. assert!( - matches!(service.stream(), Err(Error::AtCapacity)), + matches!(service.stream(), Err(e) if e.kind == ErrorKind::AtCapacity), "expected AtCapacity when all permits are held" );