From ef9dc89339ea3cfd042467beabaf30410cdf7fe1 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Mon, 31 Aug 2026 15:46:28 +0200 Subject: [PATCH 1/4] ref(service): Simplify semantic service errors --- objectstore-server/src/endpoints/common.rs | 25 +- objectstore-server/src/endpoints/multipart.rs | 4 +- objectstore-server/src/endpoints/objects.rs | 39 +- objectstore-service/docs/architecture.md | 2 +- objectstore-service/src/backend/bigtable.rs | 95 ++-- objectstore-service/src/backend/common.rs | 6 +- objectstore-service/src/backend/extensions.rs | 130 +++++- objectstore-service/src/backend/gcs.rs | 216 ++++----- objectstore-service/src/backend/in_memory.rs | 20 +- objectstore-service/src/backend/local_fs.rs | 82 ++-- .../src/backend/s3_compatible.rs | 55 +-- objectstore-service/src/backend/testing.rs | 2 +- objectstore-service/src/backend/tiered.rs | 72 +-- objectstore-service/src/concurrency.rs | 56 +-- objectstore-service/src/error.rs | 417 ++++++++++-------- objectstore-service/src/service.rs | 24 +- objectstore-service/src/stream.rs | 7 +- objectstore-service/src/streaming.rs | 16 +- 18 files changed, 724 insertions(+), 544 deletions(-) diff --git a/objectstore-server/src/endpoints/common.rs b/objectstore-server/src/endpoints/common.rs index 97468764..8d8b3e7c 100644 --- a/objectstore-server/src/endpoints/common.rs +++ b/objectstore-server/src/endpoints/common.rs @@ -6,7 +6,7 @@ use axum::Json; use axum::http::StatusCode; use axum::response::{IntoResponse, Response}; use http::HeaderValue; -use objectstore_service::error::Error as ServiceError; +use objectstore_service::error::{Error as ServiceError, ErrorKind as ServiceErrorKind}; use serde::{Deserialize, Serialize}; use thiserror::Error; @@ -90,15 +90,20 @@ 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::InvalidMetadata + | ServiceErrorKind::InvalidUploadId + | ServiceErrorKind::ClientStream => StatusCode::BAD_REQUEST, + ServiceErrorKind::RangeNotSatisfiable { .. } => StatusCode::RANGE_NOT_SATISFIABLE, + ServiceErrorKind::AtCapacity => StatusCode::TOO_MANY_REQUESTS, + ServiceErrorKind::Unsupported => StatusCode::NOT_IMPLEMENTED, + ServiceErrorKind::BackendFailure + | ServiceErrorKind::BackendResponse(_) + | ServiceErrorKind::CorruptData + | ServiceErrorKind::Panic + | ServiceErrorKind::Internal => StatusCode::INTERNAL_SERVER_ERROR, + _ => StatusCode::INTERNAL_SERVER_ERROR, + }, ApiError::Internal(_) => StatusCode::INTERNAL_SERVER_ERROR, } diff --git a/objectstore-server/src/endpoints/multipart.rs b/objectstore-server/src/endpoints/multipart.rs index f1e231d2..0452be13 100644 --- a/objectstore-server/src/endpoints/multipart.rs +++ b/objectstore-server/src/endpoints/multipart.rs @@ -16,7 +16,6 @@ use bytes::Bytes; use futures::StreamExt; use http::HeaderValue; use http::header; -use objectstore_service::error::Error as ServiceError; use objectstore_service::id::{ObjectContext, ObjectId}; use objectstore_service::multipart::{CompletedPart, PartNumber, UploadId}; use objectstore_types::metadata::Metadata; @@ -96,7 +95,8 @@ async fn initiate_inner( headers: HeaderMap, ) -> ApiResult { // TODO: Update time_created in `complete`, when we have a Service API to mutate metadata. - let metadata = Metadata::from_insert_headers(&headers, "").map_err(ServiceError::from)?; + let metadata = Metadata::from_insert_headers(&headers, "") + .map_err(|error| ApiError::Client(error.to_string()))?; state .config diff --git a/objectstore-server/src/endpoints/objects.rs b/objectstore-server/src/endpoints/objects.rs index 1aafcbac..87dc36ac 100644 --- a/objectstore-server/src/endpoints/objects.rs +++ b/objectstore-server/src/endpoints/objects.rs @@ -6,7 +6,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::{ErrorKind, ResultExt as _}; use objectstore_service::id::{ObjectContext, ObjectId}; use objectstore_types::headers::ExtValue; use objectstore_types::metadata::Metadata; @@ -46,7 +46,8 @@ async fn objects_post( headers: HeaderMap, MeteredBody(body): MeteredBody, ) -> ApiResult { - let metadata = Metadata::from_insert_headers(&headers, "").map_err(ServiceError::from)?; + let metadata = Metadata::from_insert_headers(&headers, "") + .map_err(|error| ApiError::Client(error.to_string()))?; state .config @@ -75,23 +76,26 @@ 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 })) => { - let mut response = ( - StatusCode::RANGE_NOT_SATISFIABLE, - [( - http::header::CONTENT_RANGE, - ContentRange::unsatisfiable_total_to_header_value(total), - )], - ) - .into_response(); - insert_accept_ranges(&mut response); - return Ok(response); - } + Err(ApiError::Service(e)) => match e.kind() { + ErrorKind::RangeNotSatisfiable { total } => { + let mut response = ( + StatusCode::RANGE_NOT_SATISFIABLE, + [( + http::header::CONTENT_RANGE, + ContentRange::unsatisfiable_total_to_header_value(total), + )], + ) + .into_response(); + insert_accept_ranges(&mut response); + return Ok(response); + } + _ => return Err(e.into()), + }, Err(e) => return Err(e), }; let stream = state.meter_stream(stream, &context); - let mut metadata_headers = metadata.to_headers("").map_err(ServiceError::from)?; + let mut metadata_headers = metadata.to_headers("").context(ErrorKind::Internal)?; let mut response = match content_range { Some(ref content_range) => { @@ -125,7 +129,7 @@ async fn object_head(service: AuthAwareService, Xt(id): Xt) -> ApiResu return Ok(StatusCode::NOT_FOUND.into_response()); }; - let mut headers = metadata.to_headers("").map_err(ServiceError::from)?; + let mut headers = metadata.to_headers("").context(ErrorKind::Internal)?; insert_content_length(&mut headers, &metadata); let mut response = (StatusCode::OK, headers).into_response(); @@ -202,7 +206,8 @@ async fn object_put( headers: HeaderMap, MeteredBody(body): MeteredBody, ) -> ApiResult { - let metadata = Metadata::from_insert_headers(&headers, "").map_err(ServiceError::from)?; + let metadata = Metadata::from_insert_headers(&headers, "") + .map_err(|error| ApiError::Client(error.to_string()))?; let ObjectId { context, key } = id; diff --git a/objectstore-service/docs/architecture.md b/objectstore-service/docs/architecture.md index 36564890..f1b939e2 100644 --- a/objectstore-service/docs/architecture.md +++ b/objectstore-service/docs/architecture.md @@ -237,7 +237,7 @@ 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). +limits fail with [`ErrorKind::AtCapacity`](error::ErrorKind::AtCapacity). The default execution limit is [`DEFAULT_CONCURRENCY_LIMIT`](service::DEFAULT_CONCURRENCY_LIMIT). See diff --git a/objectstore-service/src/backend/bigtable.rs b/objectstore-service/src/backend/bigtable.rs index c6f58a9a..dbd44d8d 100644 --- a/objectstore-service/src/backend/bigtable.rs +++ b/objectstore-service/src/backend/bigtable.rs @@ -50,7 +50,7 @@ use crate::backend::common::{ use crate::change_stream::{ ChangeStream, ChangeStreamFactory, CostTrackerStreamConfig, flush_change_stream, }; -use crate::error::{Error, Result}; +use crate::error::{Error, ErrorKind, Result, ResultExt as _}; use crate::gcp_auth::PrefetchingTokenProvider; use crate::id::ObjectId; use crate::stream::{ChunkedBytes, ClientStream}; @@ -486,8 +486,7 @@ fn object_mutations( // 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(ErrorKind::Internal)?; let mutations = [ // NB: We explicitly delete the row to clear metadata on overwrite. @@ -574,8 +573,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(ErrorKind::Internal)?, })), ]) } @@ -648,9 +646,7 @@ impl RowData { } COLUMN_TOMBSTONE_META => { tombstone_meta_opt = - Some(serde_json::from_slice(&cell.value).map_err(|cause| { - Error::serde("failed to deserialize tombstone meta", cause) - })?); + Some(serde_json::from_slice(&cell.value).context(ErrorKind::CorruptData)?); } COLUMN_METADATA => { if let Ok(legacy_meta) = @@ -663,10 +659,9 @@ 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(ErrorKind::CorruptData)?, + ); } } _ => {} @@ -731,9 +726,9 @@ fn parse_redirect_target(redirect_path: &[u8], tombstone_id: &ObjectId) -> Resul Ok(tombstone_id.clone()) } else { let redirect_str = std::str::from_utf8(redirect_path) - .map_err(|_| Error::generic("invalid UTF-8 in redirect path"))?; + .map_err(|_| Error::new(ErrorKind::CorruptData, "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::CorruptData, "corrupt redirect path")) } } @@ -1009,7 +1004,7 @@ 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, "unexpected tombstone")), TieredGet::NotFound => Ok(None), } } @@ -1018,7 +1013,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, "unexpected tombstone")) + } TieredMetadata::NotFound => Ok(None), } } @@ -1087,7 +1084,10 @@ impl HighVolumeBackend for BigTableBackend { } } - Err(Error::generic("BigTable: race loop in put_non_tombstone")) + Err(Error::new( + ErrorKind::Internal, + "BigTable: race loop in put_non_tombstone", + )) } #[tracing::instrument(level = "debug", skip(self))] @@ -1198,7 +1198,8 @@ impl HighVolumeBackend for BigTableBackend { } } - Err(Error::generic( + Err(Error::new( + ErrorKind::Internal, "BigTable: race loop in delete_non_tombstone", )) } @@ -1269,13 +1270,15 @@ 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!( - "TTL duration overflow: {} plus {}s cannot be represented as SystemTime", - humantime::format_rfc3339_seconds(from), - ttl.as_secs() - ), - cause: None, + let deadline = from.checked_add(ttl).ok_or_else(|| { + Error::new( + ErrorKind::Internal, + format!( + "TTL duration overflow: {} plus {}s cannot be represented as SystemTime", + humantime::format_rfc3339_seconds(from), + ttl.as_secs() + ), + ) })?; system_time_to_micros(deadline) @@ -1288,19 +1291,10 @@ 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(ErrorKind::Internal)? .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(ErrorKind::Internal) } /// Converts a wall-clock time to Bigtable's microsecond timestamp, saturating at `i64::MAX` @@ -1351,10 +1345,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(e).context(ErrorKind::BackendFailure); } Err(e) => { retry_count += 1; @@ -1408,7 +1399,7 @@ fn apply_range(payload: Bytes, range: Option) -> Result<(Option assert_eq!(total, 22), + Err(error) if matches!(error.kind(), ErrorKind::RangeNotSatisfiable { total: 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 45b9b4a0..3f28c5f9 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::{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 + /// The default returns an [`ErrorKind::Unsupported`]. Backends that implement /// [`MultipartUploadBackend`] should override this to return `Ok(self)`. fn as_multipart_upload_backend(&self) -> Result<&dyn MultipartUploadBackend> { - Err(Error::NotImplemented) + Err(ErrorKind::Unsupported.into()) } } diff --git a/objectstore-service/src/backend/extensions.rs b/objectstore-service/src/backend/extensions.rs index 31503126..b4b9b237 100644 --- a/objectstore-service/src/backend/extensions.rs +++ b/objectstore-service/src/backend/extensions.rs @@ -6,11 +6,15 @@ //! structured error code and message from it (JSON for GCS JSON API, XML for GCS //! XML API and S3). -use reqwest::{Response, header}; +use std::borrow::Cow; +use std::error::Error as StdError; +use std::fmt; + +use reqwest::{Response, StatusCode, header}; use serde::Deserialize; use tracing::Instrument; -use crate::error::{BackendDetail, Error, Result}; +use crate::error::{Error, ErrorKind, Result}; use crate::stream; /// Extension trait that sends a request inside a tracing span. @@ -78,7 +82,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. +/// a backend-failure service error with the extracted error code and message. /// /// Implemented for both [`reqwest::Response`] and `Result` so it can be /// chained directly. @@ -91,7 +95,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 as a backend failure with 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 +128,10 @@ impl ResponseExt for Response { return Ok(self); }; self.drain_body().await; - return Err(Error::reqwest(context, e)); + return Err(e.into()); }; - Err(Error::BackendResponse { - context, - status, - detail, - }) + Err(BackendResponseError::new(context, status, detail).into()) } async fn drain_body(mut self) { @@ -144,8 +144,8 @@ impl ResponseExt for 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), + Some(ce) => ce.into(), + None => e.into(), }), } } @@ -185,3 +185,109 @@ async fn parse_xml_error(resp: Response) -> BackendDetail { BackendDetail::none() } } + +/// Structured error detail parsed from a backend HTTP error response. +/// +/// Formats conditionally: includes only the fields that are non-empty. +#[derive(Debug)] +struct BackendDetail { + /// Machine-readable error code (e.g., "InvalidArgument", "NoSuchKey"). + code: String, + /// Human-readable error message from the response body. + message: String, +} + +impl BackendDetail { + /// Creates a new [`BackendDetail`] with empty code and message. + fn none() -> Self { + Self { + code: String::new(), + message: String::new(), + } + } +} + +impl fmt::Display for BackendDetail { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match (self.code.is_empty(), self.message.is_empty()) { + (false, false) => write!(f, "{} (backend code {})", self.message, self.code), + (true, false) => f.write_str(&self.message), + (false, true) => write!(f, "backend code {}", self.code), + (true, true) => Ok(()), + } + } +} + +/// An HTTP error response received from a storage backend such as GCS or S3. +/// +/// Unlike [`reqwest::Error`], which covers transport-level failures, this type captures an +/// application-level error response where the backend returned a 4xx or 5xx status together with +/// a structured response body. It retains the request context, HTTP status, and parsed backend +/// error code and message. +#[derive(Debug)] +struct BackendResponseError { + context: Cow<'static, str>, + status: StatusCode, + detail: BackendDetail, +} + +impl BackendResponseError { + /// Creates a backend response error from its operation context, status, and detail. + pub fn new( + context: impl Into>, + status: StatusCode, + detail: BackendDetail, + ) -> Self { + Self { + context: context.into(), + status, + detail, + } + } +} + +impl fmt::Display for BackendResponseError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{} ({}). {}", self.context, self.status, self.detail) + } +} + +impl StdError for BackendResponseError {} + +impl From for Error { + fn from(source: BackendResponseError) -> Self { + Self::with_source(ErrorKind::BackendResponse(source.status), source) + } +} + +#[cfg(test)] +mod tests { + use std::error::Error as _; + + use reqwest::StatusCode; + + use super::{BackendDetail, BackendResponseError}; + use crate::error::{Error, ErrorKind}; + + #[test] + fn backend_response_preserves_status_and_structured_source() { + let error: Error = BackendResponseError::new( + "GCS: get object", + StatusCode::TOO_MANY_REQUESTS, + BackendDetail { + code: "rateLimitExceeded".to_owned(), + message: "too many requests".to_owned(), + }, + ) + .into(); + + assert_eq!( + error.kind(), + ErrorKind::BackendResponse(StatusCode::TOO_MANY_REQUESTS) + ); + assert_eq!( + error.source().unwrap().to_string(), + "GCS: get object (429 Too Many Requests). too many requests (backend code rateLimitExceeded)" + ); + } +} diff --git a/objectstore-service/src/backend/gcs.rs b/objectstore-service/src/backend/gcs.rs index 5204dca0..9241b5c4 100644 --- a/objectstore-service/src/backend/gcs.rs +++ b/objectstore-service/src/backend/gcs.rs @@ -2,12 +2,12 @@ use std::borrow::Cow; use std::collections::BTreeMap; +use std::error::Error as _; use std::future::Future; use std::sync::Arc; use std::time::SystemTime; use std::{fmt, io}; -use anyhow::Context; use futures_util::{StreamExt, TryStreamExt}; use gcp_auth::TokenProvider; use objectstore_types::headers; @@ -25,7 +25,7 @@ use crate::backend::common::{ use crate::change_stream::{ ChangeStream, ChangeStreamFactory, CostTrackerStreamConfig, flush_change_stream, }; -use crate::error::{Error, Result}; +use crate::error::{Error, ErrorKind, Result, ResultExt as _}; use crate::gcp_auth::PrefetchingTokenProvider; use crate::id::ObjectId; use crate::multipart::{ @@ -258,7 +258,8 @@ impl GcsObject { .metadata .remove(&GcsMetaKey::Expiration) .map(|s| s.parse()) - .transpose()? + .transpose() + .context(ErrorKind::CorruptData)? .unwrap_or_default(); let origin = self @@ -273,15 +274,16 @@ impl GcsObject { .transpose()?; let content_type = self.content_type; - let compression = self.content_encoding.map(|s| s.parse()).transpose()?; + let compression = self + .content_encoding + .map(|s| s.parse()) + .transpose() + .context(ErrorKind::CorruptData)?; let size = self .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(ErrorKind::CorruptData)?; let time_created = self.time_created; // At this point, all built-in metadata should have been removed from self.metadata. @@ -290,12 +292,10 @@ impl GcsObject { if let GcsMetaKey::Custom(custom_key) = key { custom.insert(custom_key, decode_gcs_meta_value(&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::CorruptData, + format!("GCS: unexpected built-in metadata key in object metadata: {key}"), + )); } } @@ -388,10 +388,7 @@ 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(ErrorKind::Internal)?, ); } @@ -401,10 +398,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(ErrorKind::Internal)?, ); } @@ -433,10 +427,7 @@ fn metadata_to_gcs_headers(metadata: &Metadata) -> Result { /// Decodes a stored GCS metadata value into its logical string. fn decode_gcs_meta_value(value: &str) -> Result { - headers::decode_header_str(value).map_err(|cause| Error::Generic { - context: "GCS: invalid percent-encoded UTF-8 in object metadata".to_owned(), - cause: Some(Box::new(cause)), - }) + headers::decode_header_str(value).context(ErrorKind::CorruptData) } /// Inserts a single `x-goog-meta-*` header, escaping the value for transport. @@ -456,10 +447,7 @@ 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)), - })?, + HeaderName::try_from(&header_name).context(ErrorKind::Internal)?, headers::encode_header_value(value), ); Ok(()) @@ -467,15 +455,19 @@ fn insert_gcs_meta_header( /// Returns `true` if the error is a transient reqwest failure worth retrying. 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, + if let Some(cause) = error + .source() + .and_then(|source| source.downcast_ref::()) + { + cause.is_timeout() + || cause.is_connect() + || cause.is_request() + || cause.status().is_some_and(status_is_retryable) + } else { + matches!( + error.kind(), + ErrorKind::BackendResponse(status) if status_is_retryable(status) + ) } } @@ -522,7 +514,9 @@ impl GcsBackend { Ok(Self { client: common::reqwest_client(), - endpoint: endpoint_str.parse().context("invalid GCS endpoint URL")?, + endpoint: endpoint_str + .parse() + .map_err(|error| anyhow::anyhow!("invalid GCS endpoint URL: {error}"))?, bucket, token_provider, change_stream, @@ -535,12 +529,14 @@ impl GcsBackend { let path = id.as_storage_path().to_string(); url.path_segments_mut() - .map_err(|()| Error::Generic { - context: format!( - "GCS: invalid endpoint URL, {} cannot be a base", - self.endpoint - ), - cause: None, + .map_err(|()| { + Error::new( + ErrorKind::Internal, + format!( + "GCS: invalid endpoint URL, {} cannot be a base", + self.endpoint + ), + ) })? .extend(&["storage", "v1", "b", &self.bucket, "o", &path]); @@ -552,12 +548,14 @@ impl GcsBackend { let mut url = self.endpoint.clone(); url.path_segments_mut() - .map_err(|()| Error::Generic { - context: format!( - "GCS: invalid endpoint URL, {} cannot be a base", - self.endpoint - ), - cause: None, + .map_err(|()| { + Error::new( + ErrorKind::Internal, + format!( + "GCS: invalid endpoint URL, {} cannot be a base", + self.endpoint + ), + ) })? .extend(&["upload", "storage", "v1", "b", &self.bucket, "o"]); @@ -577,12 +575,14 @@ 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!( - "GCS: invalid endpoint URL, {} cannot be a base", - self.endpoint - ), - cause: None, + let mut segments = url.path_segments_mut().map_err(|()| { + Error::new( + ErrorKind::Internal, + format!( + "GCS: invalid endpoint URL, {} cannot be a base", + self.endpoint + ), + ) })?; segments.push(&self.bucket); for part in id.as_storage_path().to_string().split('/') { @@ -641,8 +641,7 @@ impl GcsBackend { .request(Method::GET, object_url.clone()) .await? .send_traced() - .await - .map_err(|e| Error::reqwest("GCS: get metadata request", e))?; + .await?; if resp.status() == StatusCode::NOT_FOUND { resp.drain_body().await; @@ -653,8 +652,7 @@ impl GcsBackend { .check_error("GCS: get metadata status") .await? .json() - .await - .map_err(|e| Error::reqwest("GCS: get metadata parse", e))?; + .await?; Ok(Some(metadata)) }) @@ -740,10 +738,12 @@ impl GcsBackend { } // Bumping TTI is opportunistic. A concurrent metadata writer won the CAS race, // so leave its update intact and let a future read evaluate the TTI again. - Err(Error::BackendResponse { - status: StatusCode::PRECONDITION_FAILED, - .. - }) => Ok(false), + Err(error) + if error.kind() + == ErrorKind::BackendResponse(StatusCode::PRECONDITION_FAILED) => + { + Ok(false) + } Err(error) => Err(error), } }) @@ -782,10 +782,7 @@ 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(ErrorKind::Internal)?; let multipart = multipart::Form::new() .part( @@ -798,10 +795,7 @@ impl Backend for GcsBackend { "media", multipart::Part::stream(Body::wrap_stream(stream.boxed())) .mime_str(&metadata.content_type) - .map_err(|e| Error::Generic { - context: format!("invalid mime type: {}", metadata.content_type), - cause: Some(Box::new(e)), - })?, + .context(ErrorKind::InvalidMetadata)?, ); // GCS requires a multipart/related request. Its body looks identical to @@ -852,10 +846,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 = req.send_traced().await?; if resp.status() == StatusCode::RANGE_NOT_SATISFIABLE { let raw = resp @@ -864,10 +855,11 @@ 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!( - "GCS: 416 response with invalid Content-Range: {raw:?}" - )), + Some(total) => ErrorKind::RangeNotSatisfiable { total }.into(), + None => Error::new( + ErrorKind::BackendFailure, + format!("GCS: 416 response with invalid Content-Range: {raw:?}"), + ), }; resp.drain_body().await; return Err(err); @@ -884,9 +876,11 @@ 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::BackendFailure, + "GCS: 206 response missing valid Content-Range header", + ) })?, ) } else { @@ -919,8 +913,7 @@ impl Backend for GcsBackend { .request(Method::DELETE, object_url.clone()) .await? .send_traced() - .await - .map_err(|e| Error::reqwest("GCS: delete object", e))?; + .await?; // Do not error for objects that do not exist if resp.status() == StatusCode::NOT_FOUND { @@ -1071,10 +1064,10 @@ impl MultipartUploadBackend for GcsBackend { let mut headers = metadata_to_gcs_headers(metadata)?; headers.insert( header::CONTENT_TYPE, - metadata.content_type.parse().map_err(|e| Error::Generic { - context: "GCS: invalid content-type header value".into(), - cause: Some(Box::new(e)), - })?, + metadata + .content_type + .parse() + .context(ErrorKind::InvalidMetadata)?, ); headers.insert( header::CONTENT_LENGTH, @@ -1094,16 +1087,10 @@ impl MultipartUploadBackend for GcsBackend { .check_error("GCS: initiate multipart upload") .await?; - let body = resp - .bytes() - .await - .map_err(|e| Error::reqwest("GCS: read initiate multipart body", e))?; + let body = resp.bytes().await?; 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)), - })?; + quick_xml::de::from_reader(body.as_ref()).context(ErrorKind::CorruptData)?; xml.try_into() } @@ -1148,7 +1135,12 @@ 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::BackendFailure, + "GCS: upload part response missing ETag header", + ) + })?; resp.drain_body().await; @@ -1187,16 +1179,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?; 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)), - })?; + quick_xml::de::from_reader(body.as_ref()).context(ErrorKind::CorruptData)?; Ok(xml.into()) } @@ -1221,8 +1207,7 @@ impl MultipartUploadBackend for GcsBackend { .request(Method::DELETE, url) .await? .send_traced() - .await - .map_err(|e| Error::reqwest("GCS: abort multipart upload", e))?; + .await?; // XXX: real S3 would return 404 here if the upload has been recently completed and we // would have to handle it. It turns out GCS returns 204 instead, so we don't need to @@ -1251,10 +1236,7 @@ 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(ErrorKind::Internal)?; self.with_retry("complete_multipart", || { let url = url.clone(); @@ -1274,10 +1256,7 @@ impl MultipartUploadBackend for GcsBackend { // would have to handle it. It turns out GCS returns 200 instead, so we don't need to // handle that case. - let body = resp - .bytes() - .await - .map_err(|e| Error::reqwest("GCS: read complete multipart body", e))?; + let body = resp.bytes().await?; let error = quick_xml::de::from_reader::<_, XmlError>(body.as_ref()) .ok() @@ -1363,7 +1342,6 @@ mod tests { .await? .json::() .await - .map_err(|e| Error::reqwest("GCS: get metadata parse", e)) .map(|object| (object.generation, object.metageneration))?) } diff --git a/objectstore-service/src/backend/in_memory.rs b/objectstore-service/src/backend/in_memory.rs index 93e74cfc..2934d6be 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, "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(ErrorKind::RangeNotSatisfiable { 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(ErrorKind::RangeNotSatisfiable { 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::BackendFailure, "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::BackendFailure, "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::BackendFailure, "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..bbc672d5 100644 --- a/objectstore-service/src/backend/local_fs.rs +++ b/objectstore-service/src/backend/local_fs.rs @@ -1,6 +1,6 @@ //! Local filesystem backend for development and testing. -use std::io::ErrorKind; +use std::io; use std::path::PathBuf; use std::pin::pin; use std::time::SystemTime; @@ -15,7 +15,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 as _}; use crate::id::ObjectId; use crate::multipart::{ AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse, @@ -96,18 +96,15 @@ 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(ErrorKind::Internal)?; 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() == io::ErrorKind::NotFound => { objectstore_log::debug!("Object not found"); return Ok(None); } @@ -137,13 +134,15 @@ impl Backend for LocalFsBackend { 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, - })?; + serde_json::from_str(metadata_line.trim_end()).context(ErrorKind::CorruptData)?; 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::CorruptData, + "local-fs file corrupted: shorter than header", + ) + })?; metadata.size = Some(payload_size as usize); let (content_range, stream) = match range { @@ -151,7 +150,7 @@ impl Backend for LocalFsBackend { let content_range = byte_range .resolve(payload_size) - .ok_or(Error::RangeNotSatisfiable { + .ok_or(ErrorKind::RangeNotSatisfiable { total: payload_size, })?; let payload_start = metadata_line.len() as u64 + content_range.start; @@ -170,7 +169,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() == io::ErrorKind::NotFound { objectstore_log::debug!("Object not found"); } @@ -199,10 +198,7 @@ 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(ErrorKind::Internal)?; tokio::fs::write(meta_path, metadata_json).await?; Ok(upload_id) @@ -219,7 +215,10 @@ 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::BackendFailure, + "multipart upload not found", + )); } let etag = format!("\"etag-{part_number}-{content_length}\""); @@ -229,10 +228,7 @@ 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(ErrorKind::Internal)?; let part_path = dir.join(format!("{part_number}.part")); let file = OpenOptions::new() @@ -250,8 +246,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 +271,10 @@ 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::BackendFailure, + "multipart upload not found", + )); } let mut entries = tokio::fs::read_dir(&dir).await?; @@ -300,10 +299,7 @@ impl MultipartUploadBackend for LocalFsBackend { 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, - })?; + serde_json::from_str(header_line.trim_end()).context(ErrorKind::CorruptData)?; parts.push(Part { part_number: pn, @@ -353,17 +349,17 @@ 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::BackendFailure, + "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, - })?; + serde_json::from_slice(&meta_bytes).context(ErrorKind::CorruptData)?; // 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. @@ -383,10 +379,7 @@ impl MultipartUploadBackend for LocalFsBackend { 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, - })?; + serde_json::from_str(header_line.trim_end()).context(ErrorKind::CorruptData)?; let stored_etag = header["etag"].as_str().unwrap_or(""); if stored_etag != completed.etag { @@ -411,10 +404,7 @@ 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(ErrorKind::Internal)?; writer.write_all(metadata_json.as_bytes()).await?; writer.write_all(b"\n").await?; @@ -841,7 +831,7 @@ mod tests { .unwrap(); match backend.get_object(&id, Some(ByteRange::From(100))).await { - Err(Error::RangeNotSatisfiable { total: 5 }) => {} + Err(error) if matches!(error.kind(), ErrorKind::RangeNotSatisfiable { total: 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 f34beec0..b9f7486c 100644 --- a/objectstore-service/src/backend/s3_compatible.rs +++ b/objectstore-service/src/backend/s3_compatible.rs @@ -13,7 +13,7 @@ use super::extensions::{ResponseExt, SendTraced}; use crate::backend::common::{ self, Backend, DeleteResponse, GetResponse, MetadataResponse, PutResponse, }; -use crate::error::{Error, Result}; +use crate::error::{Error, ErrorKind, Result, ResultExt as _}; use crate::id::ObjectId; use crate::stream::ClientStream; @@ -152,9 +152,11 @@ where provider .get_token() .await - .map_err(|err| Error::Generic { - context: "S3: failed to get authentication token".to_owned(), - cause: Some(err.into()), + .map_err(|err| { + Error::new( + ErrorKind::BackendFailure, + format!("S3: failed to get authentication token: {err}"), + ) })? .as_str(), ); @@ -180,10 +182,7 @@ where let response = builder .send_traced() .await - .map_err(|cause| Error::Reqwest { - context: "S3: failed to send request".to_string(), - cause, - })?; + .context(ErrorKind::BackendFailure)?; if response.status() == StatusCode::NOT_FOUND { objectstore_log::debug!("Object not found"); @@ -198,10 +197,11 @@ 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!( - "S3: 416 response with invalid Content-Range: {raw:?}" - )), + Some(total) => ErrorKind::RangeNotSatisfiable { total }.into(), + None => Error::new( + ErrorKind::BackendFailure, + format!("S3: 416 response with invalid Content-Range: {raw:?}"), + ), }; response.drain_body().await; return Err(err); @@ -210,16 +210,19 @@ where let response = response.check_error("S3: failed to get object").await?; let headers = response.headers(); - let mut metadata = Metadata::from_headers(headers, GCS_CUSTOM_PREFIX)?; + let mut metadata = + Metadata::from_headers(headers, GCS_CUSTOM_PREFIX).context(ErrorKind::CorruptData)?; let content_range = if response.status() == StatusCode::PARTIAL_CONTENT { let range = headers .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::BackendFailure, + "S3: 206 response missing valid Content-Range header", + ) })?; metadata.size = Some(range.total as usize); Some(range) @@ -231,10 +234,7 @@ where .and_then(|value| value.to_str().ok()) .map(|value| value.parse::()) .transpose() - .map_err(|cause| Error::Generic { - context: "S3: failed to parse Content-Length from object response".to_string(), - cause: Some(Box::new(cause)), - })?; + .context(ErrorKind::CorruptData)?; if let Some(size) = size { metadata.size = Some(size); @@ -278,7 +278,10 @@ where format!("/{}/{}", self.bucket, id.as_storage_path()), ) .header("x-goog-metadata-directive", "REPLACE") - .headers(metadata_to_gcs_headers(metadata, GCS_CUSTOM_PREFIX)?) + .headers( + metadata_to_gcs_headers(metadata, GCS_CUSTOM_PREFIX) + .context(ErrorKind::InvalidMetadata)?, + ) .send_traced() .await .check_error("S3: update expiration time") @@ -328,7 +331,10 @@ impl Backend for S3CompatibleBackend { objectstore_log::debug!("Writing to s3_compatible backend"); self.request(Method::PUT, self.object_url(id)) .await? - .headers(metadata_to_gcs_headers(metadata, GCS_CUSTOM_PREFIX)?) + .headers( + metadata_to_gcs_headers(metadata, GCS_CUSTOM_PREFIX) + .context(ErrorKind::InvalidMetadata)?, + ) .body(Body::wrap_stream(stream)) .send_traced() .await @@ -369,10 +375,7 @@ impl Backend for S3CompatibleBackend { .await? .send_traced() .await - .map_err(|cause| Error::Reqwest { - context: "S3: failed to send delete request".to_string(), - cause, - })?; + .context(ErrorKind::BackendFailure)?; // 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..986325a7 100644 --- a/objectstore-service/src/backend/testing.rs +++ b/objectstore-service/src/backend/testing.rs @@ -24,7 +24,7 @@ //! _inner: &InMemoryBackend, //! _id: &ObjectId, //! ) -> Result { -//! Err(crate::error::Error::Io(std::io::Error::new( +//! Err(crate::error::Error::with_source(crate::error::ErrorKind::BackendFailure, std::io::Error::new( //! std::io::ErrorKind::ConnectionRefused, //! "simulated delete failure", //! ))) diff --git a/objectstore-service/src/backend/tiered.rs b/objectstore-service/src/backend/tiered.rs index 9887fe24..deceb207 100644 --- a/objectstore-service/src/backend/tiered.rs +++ b/objectstore-service/src/backend/tiered.rs @@ -115,7 +115,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 as _}; use crate::id::ObjectId; use crate::multipart::{ AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse, @@ -587,8 +587,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(ErrorKind::Internal)?; Ok(UploadId::new( base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(json), )?) @@ -601,8 +600,8 @@ 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::InvalidUploadId, e.to_string()))?; + serde_json::from_slice(&json).context(ErrorKind::CorruptData) } } @@ -821,7 +820,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( + return Err(Error::new( + ErrorKind::BackendFailure, "completed multipart object not found in long-term storage", )); } @@ -1151,10 +1151,13 @@ mod tests { _inner: &InMemoryBackend, _id: &ObjectId, ) -> Result { - Err(Error::Io(std::io::Error::new( - std::io::ErrorKind::ConnectionRefused, - "simulated long-term delete failure", - ))) + Err(Error::with_source( + ErrorKind::BackendFailure, + std::io::Error::new( + std::io::ErrorKind::ConnectionRefused, + "simulated long-term delete failure", + ), + )) } } @@ -1288,10 +1291,13 @@ 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( - std::io::ErrorKind::TimedOut, - "simulated compare_and_write failure", - ))) + Err(Error::with_source( + ErrorKind::BackendFailure, + std::io::Error::new( + std::io::ErrorKind::TimedOut, + "simulated compare_and_write failure", + ), + )) } } @@ -1939,10 +1945,13 @@ mod tests { .complete_multipart(id, upload_id, parts) .await .unwrap(); - Err(Error::Io(std::io::Error::new( - std::io::ErrorKind::TimedOut, - "simulated network error on complete_multipart", - ))) + Err(Error::with_source( + ErrorKind::BackendFailure, + std::io::Error::new( + std::io::ErrorKind::TimedOut, + "simulated network error on complete_multipart", + ), + )) } async fn get_metadata( @@ -1950,10 +1959,13 @@ mod tests { _inner: &InMemoryBackend, _id: &ObjectId, ) -> Result { - Err(Error::Io(std::io::Error::new( - std::io::ErrorKind::TimedOut, - "simulated network error on get_metadata", - ))) + Err(Error::with_source( + ErrorKind::BackendFailure, + std::io::Error::new( + std::io::ErrorKind::TimedOut, + "simulated network error on get_metadata", + ), + )) } } @@ -2058,10 +2070,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(Error::with_source( + ErrorKind::BackendFailure, + std::io::Error::new(std::io::ErrorKind::TimedOut, "simulated network error"), + )) } else { Ok(inner .complete_multipart(id, upload_id, parts) @@ -2194,10 +2206,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(Error::with_source( + ErrorKind::BackendFailure, + std::io::Error::new(std::io::ErrorKind::TimedOut, "simulated network error"), + )) } else { inner.get_metadata(id).await } diff --git a/objectstore-service/src/concurrency.rs b/objectstore-service/src/concurrency.rs index 7715568e..fe46cbc7 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::{AcquireError, Notify, OwnedSemaphorePermit, Semaphore}; -use crate::error::{Error, Result}; +use crate::error::{Error, ErrorKind, Panic, Result}; /// Interval for the periodic metrics emitter. const EMITTER_INTERVAL: Duration = Duration::from_secs(1); @@ -120,10 +120,10 @@ impl ConcurrencyLimiter { /// If a permit is free, returns immediately without touching the /// queue. Otherwise, acquires a queue ticket (bounded by the queue /// depth) and waits up to the configured timeout. Returns - /// [`Error::AtCapacity`] if the queue is full or on timeout. + /// [`ErrorKind::AtCapacity`] if the queue is full or on timeout. pub async fn acquire(&self) -> Result { if self.tasks_total == 0 { - return Err(Error::AtCapacity); + return Err(ErrorKind::AtCapacity.into()); } // Fast path: Instantly grab a free permit without parking. @@ -141,13 +141,13 @@ impl ConcurrencyLimiter { .queue .clone() .try_acquire_owned() - .map_err(|_| Error::AtCapacity)?; + .map_err(|_| 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(|_| ErrorKind::AtCapacity)? + .map_err(|_| ErrorKind::AtCapacity)?; Ok(ConcurrencyPermit { task_permit: Some(task_permit), @@ -158,13 +158,13 @@ impl ConcurrencyLimiter { /// Tries to acquire a single permit without waiting. /// - /// Returns [`Error::AtCapacity`] when no permits are available. + /// Returns [`ErrorKind::AtCapacity`] when no permits are available. pub fn try_acquire(&self) -> Result { let task_permit = self .tasks .clone() .try_acquire_owned() - .map_err(|_| Error::AtCapacity)?; + .map_err(|_| ErrorKind::AtCapacity)?; Ok(ConcurrencyPermit { task_permit: Some(task_permit), @@ -181,10 +181,10 @@ impl ConcurrencyLimiter { /// acquired under a single timeout deadline configured via /// [`with_timeout`](Self::with_timeout). /// - /// Returns [`Error::AtCapacity`] on timeout or when `max` is zero. + /// Returns [`ErrorKind::AtCapacity`] on timeout or when `max` is zero. pub async fn acquire_bulk(&self) -> Result { if self.tasks_total == 0 { - return Err(Error::AtCapacity); + return Err(ErrorKind::AtCapacity.into()); } let bulk_sem = self.bulk.clone(); @@ -198,8 +198,8 @@ impl ConcurrencyLimiter { let (task_permit, bulk_permit) = tokio::time::timeout(self.timeout, acquire) .await - .map_err(|_| Error::AtCapacity)? - .map_err(|_: AcquireError| Error::AtCapacity)?; + .map_err(|_| ErrorKind::AtCapacity)? + .map_err(|_: AcquireError| ErrorKind::AtCapacity)?; Ok(ConcurrencyPermit { task_permit: Some(task_permit), @@ -348,7 +348,7 @@ where let result = std::panic::AssertUnwindSafe(f) .catch_unwind() .await - .unwrap_or_else(|payload| Err(Error::panic(payload))); + .unwrap_or_else(|payload| Err(Panic::new(payload).into())); if let Err(ref e) = result { let error = e as &dyn std::error::Error; @@ -370,8 +370,9 @@ where ); rx.await.map_err(|_| { - objectstore_log::error!(!!&Error::Dropped, operation, "Task failed"); - Error::Dropped + let error = Error::new(ErrorKind::Internal, "task dropped"); + objectstore_log::error!(!!&error, operation, "Task failed"); + error })? } @@ -380,7 +381,6 @@ mod tests { use std::sync::atomic::{AtomicU32, Ordering}; use super::*; - use crate::error::Error; #[test] fn available_permits_tracks_held() { @@ -430,7 +430,7 @@ mod tests { let _permit = limiter.try_acquire().unwrap(); let result = limiter.try_acquire(); - assert!(matches!(result, Err(Error::AtCapacity))); + assert!(result.is_err_and(|error| error.kind() == ErrorKind::AtCapacity)); } #[test] @@ -505,7 +505,11 @@ mod tests { let p1 = limiter.try_acquire().unwrap(); let p2 = limiter.try_acquire().unwrap(); - assert!(matches!(limiter.try_acquire(), Err(Error::AtCapacity))); + assert!( + limiter + .try_acquire() + .is_err_and(|error| error.kind() == ErrorKind::AtCapacity) + ); drop(p1); assert!(limiter.try_acquire().is_ok()); @@ -520,7 +524,7 @@ mod tests { let start = tokio::time::Instant::now(); let result = limiter.acquire().await; - assert!(matches!(result, Err(Error::AtCapacity))); + assert!(result.is_err_and(|error| error.kind() == ErrorKind::AtCapacity)); assert_eq!(start.elapsed(), Duration::ZERO); drop(bulk_permits); } @@ -568,7 +572,7 @@ mod tests { tokio::time::sleep(Duration::from_secs(2)).await; let result = waiter.await.unwrap(); - assert!(matches!(result, Err(Error::AtCapacity))); + assert!(result.is_err_and(|error| error.kind() == ErrorKind::AtCapacity)); assert_eq!(limiter.queued_permits(), 0); } @@ -584,7 +588,7 @@ mod tests { assert_eq!(limiter.queued_permits(), 1); let result = limiter.acquire().await; - assert!(matches!(result, Err(Error::AtCapacity))); + assert!(result.is_err_and(|error| error.kind() == ErrorKind::AtCapacity)); } #[tokio::test(start_paused = true)] @@ -640,7 +644,7 @@ mod tests { let start = tokio::time::Instant::now(); let result = limiter.acquire().await; - assert!(matches!(result, Err(Error::AtCapacity))); + assert!(result.is_err_and(|error| error.kind() == ErrorKind::AtCapacity)); assert_eq!(start.elapsed(), Duration::ZERO); } @@ -768,7 +772,7 @@ mod tests { tokio::time::sleep(Duration::from_secs(2)).await; let result = waiter.await.unwrap(); - assert!(matches!(result, Err(Error::AtCapacity))); + assert!(result.is_err_and(|error| error.kind() == ErrorKind::AtCapacity)); assert_eq!(limiter.used_bulk_permits(), 0); } @@ -797,7 +801,7 @@ mod tests { let start = tokio::time::Instant::now(); let result = limiter.acquire_bulk().await; - assert!(matches!(result, Err(Error::AtCapacity))); + assert!(result.is_err_and(|error| error.kind() == ErrorKind::AtCapacity)); assert_eq!(start.elapsed(), Duration::ZERO); } @@ -815,7 +819,7 @@ mod tests { tokio::time::sleep(Duration::from_secs(2)).await; let result = waiter.await.unwrap(); - assert!(matches!(result, Err(Error::AtCapacity))); + assert!(result.is_err_and(|error| error.kind() == ErrorKind::AtCapacity)); drop(permit); } @@ -842,7 +846,7 @@ mod tests { // Third waiter exceeds queue depth — rejected instantly. let start = tokio::time::Instant::now(); let result = limiter.acquire().await; - assert!(matches!(result, Err(Error::AtCapacity))); + assert!(result.is_err_and(|error| error.kind() == ErrorKind::AtCapacity)); assert_eq!(start.elapsed(), Duration::ZERO); drop(bulk_permits); diff --git a/objectstore-service/src/error.rs b/objectstore-service/src/error.rs index 9643dd43..601b51f7 100644 --- a/objectstore-service/src/error.rs +++ b/objectstore-service/src/error.rs @@ -1,219 +1,288 @@ -//! Error types for service and backend operations. +//! Semantic errors for service and backend operations. //! -//! [`Error`] covers I/O, serialization, HTTP, metadata, authentication, -//! and backend-specific failures. [`Result`] is the corresponding alias. +//! [`Error`] deliberately exposes only a stable semantic [`ErrorKind`]. Its source chain and an +//! optional origin backtrace retain diagnostic detail without making backend implementation +//! details part of the service API. use std::any::Any; +use std::backtrace::Backtrace; +use std::borrow::Cow; +use std::error::Error as StdError; use std::fmt; use objectstore_log::Level; use reqwest::StatusCode; -use thiserror::Error as ThisError; -use crate::stream::ClientError; - -/// Structured error detail parsed from a backend HTTP error response. -/// -/// Formats conditionally: includes only the fields that are non-empty. +/// A panic captured from a service task. #[derive(Debug)] -pub struct BackendDetail { - /// Machine-readable error code (e.g., "InvalidArgument", "NoSuchKey"). - pub code: String, - /// Human-readable error message from the response body. - pub message: String, +pub struct Panic { + message: Cow<'static, str>, } -impl BackendDetail { - /// Creates a new [`BackendDetail`] with empty code and message. - pub fn none() -> Self { - Self { - code: String::new(), - message: String::new(), - } +impl Panic { + /// Extracts a message from a panic payload. + pub fn new(payload: Box) -> Self { + let message = if let Some(s) = payload.downcast_ref::<&str>() { + Cow::Borrowed(*s) + } else if let Some(s) = payload.downcast_ref::() { + Cow::Owned(s.clone()) + } else { + Cow::Borrowed("unknown panic") + }; + Self { message } } } -impl fmt::Display for BackendDetail { +impl fmt::Display for Panic { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - match (self.code.is_empty(), self.message.is_empty()) { - (false, false) => write!(f, "{} (backend code {})", self.message, self.code), - (true, false) => write!(f, "{}", self.message), - (false, true) => write!(f, "backend code {}", self.code), - (true, true) => Ok(()), - } + f.write_str(&self.message) } } -/// 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), - - /// 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, - }, +impl StdError for Panic {} - /// 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. - /// - /// This is an unexpected condition that can occur when the runtime drops the task for unknown - /// reasons. - #[error("task dropped")] - Dropped, - - /// 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, - - /// The requested byte range is not satisfiable for the object's size. - #[error("range not satisfiable (object size: {total} bytes)")] +/// The client-visible semantic classification of a service error. +#[non_exhaustive] +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ErrorKind { + /// Object metadata supplied by a client is invalid. + InvalidMetadata, + /// A multipart upload identifier is invalid. + InvalidUploadId, + /// A client-provided request stream failed. + ClientStream, + /// A requested byte range cannot be resolved against the object size. RangeNotSatisfiable { - /// Total size of the object in bytes. + /// Total object length in bytes. total: u64, }, - - /// The service has reached its concurrency limit and cannot accept more operations. - #[error("concurrency limit reached")] + /// The service cannot accept more work. AtCapacity, + /// The requested operation is unsupported. + Unsupported, + /// A storage backend operation failed. + BackendFailure, + /// A storage backend returned an HTTP error response. + BackendResponse(StatusCode), + /// A service task panicked. + Panic, + /// Persisted or remote data is corrupt. + CorruptData, + /// An unexpected internal service failure occurred. + Internal, +} - /// 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, +impl fmt::Display for ErrorKind { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::InvalidMetadata => f.write_str("invalid object metadata"), + Self::InvalidUploadId => f.write_str("invalid upload id"), + Self::ClientStream => f.write_str("invalid client stream"), + Self::RangeNotSatisfiable { total } => { + write!(f, "range not satisfiable (object size: {total} bytes)") + } + Self::AtCapacity => f.write_str("service at capacity"), + Self::Unsupported => f.write_str("unsupported operation"), + Self::BackendFailure => f.write_str("backend operation failed"), + Self::BackendResponse(status) => write!(f, "backend returned HTTP {status}"), + Self::CorruptData => f.write_str("corrupt stored data"), + Self::Panic => f.write_str("service task panicked"), + Self::Internal => f.write_str("internal service error"), + } + } +} - /// Invalid upload ID (e.g. path traversal attempt). - #[error(transparent)] - InvalidUploadId(#[from] objectstore_types::multipart::InvalidUploadId), +/// Opaque service error with a stable semantic kind. +pub struct Error { + kind: ErrorKind, + message: Option>, + source: Option>, + backtrace: Option, } impl Error { - /// Creates an [`Error::Panic`] 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() - } else if let Some(s) = payload.downcast_ref::() { - s.clone() - } else { - "unknown panic".to_owned() - }; - Self::Panic(msg) + /// Returns this error's semantic kind. + pub fn kind(&self) -> ErrorKind { + self.kind } - /// 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, - } + /// Returns the backtrace captured where this service error originated, if enabled. + pub fn backtrace(&self) -> Option<&Backtrace> { + self.backtrace.as_ref() } - /// 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, - } + /// Creates an error without an underlying source and with a specific message. + pub fn new(kind: ErrorKind, message: impl Into>) -> Self { + Self::build(kind, Some(message.into()), None) } - /// 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, + /// Creates an error with an underlying source. + pub fn with_source(kind: ErrorKind, source: E) -> Self + where + E: StdError + Send + Sync + 'static, + { + Self::build(kind, None, Some(Box::new(source))) + } + + fn build( + kind: ErrorKind, + message: Option>, + source: Option>, + ) -> Self { + Self { + kind, + message, + source, + backtrace: Some(Backtrace::force_capture()), } } /// 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, + match self.kind { + ErrorKind::InvalidMetadata + | ErrorKind::InvalidUploadId + | ErrorKind::ClientStream + | ErrorKind::RangeNotSatisfiable { .. } => Level::DEBUG, + ErrorKind::AtCapacity => Level::WARN, + ErrorKind::Unsupported + | ErrorKind::BackendFailure + | ErrorKind::BackendResponse(_) + | ErrorKind::CorruptData + | ErrorKind::Panic + | ErrorKind::Internal => Level::ERROR, + } + } +} + +impl fmt::Display for Error { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match &self.message { + Some(message) => f.write_str(message), + None => self.kind.fmt(f), } } } +impl fmt::Debug for Error { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("Error") + .field("kind", &self.kind) + .field("message", &self.message) + .field("source", &self.source) + .field("backtrace", &self.backtrace) + .finish() + } +} + +impl StdError for Error { + fn source(&self) -> Option<&(dyn StdError + 'static)> { + self.source.as_deref().map(|source| source as _) + } +} + +impl From for Error { + fn from(kind: ErrorKind) -> Self { + Self::build(kind, None, None) + } +} + +impl From for Error { + fn from(source: Panic) -> Self { + Self::with_source(ErrorKind::Panic, source) + } +} + +/// Adds a semantic kind when converting an external error into a service error. +pub trait ResultExt { + /// Converts an external error into a service error with `kind`. + fn context(self, kind: ErrorKind) -> Result; +} + +impl ResultExt for std::result::Result +where + E: StdError + Send + Sync + 'static, +{ + fn context(self, kind: ErrorKind) -> Result { + self.map_err(|source| Error::with_source(kind, source)) + } +} + +impl From for Error { + fn from(source: std::io::Error) -> Self { + Self::with_source(ErrorKind::BackendFailure, source) + } +} + +impl From for Error { + fn from(source: reqwest::Error) -> Self { + Self::with_source(ErrorKind::BackendFailure, source) + } +} + +impl From for Error { + fn from(source: gcp_auth::Error) -> Self { + Self::with_source(ErrorKind::BackendFailure, source) + } +} + +impl From for Error { + fn from(source: crate::stream::ClientError) -> Self { + Self::with_source(ErrorKind::ClientStream, source) + } +} + +impl From for Error { + fn from(source: objectstore_types::multipart::InvalidUploadId) -> Self { + Self::with_source(ErrorKind::InvalidUploadId, source) + } +} + /// Result type for service operations. pub type Result = std::result::Result; + +#[cfg(test)] +mod tests { + use std::error::Error as _; + use std::io; + + use super::{Error, ErrorKind, Panic}; + + #[test] + fn errors_always_capture_backtraces() { + let client: Error = ErrorKind::InvalidMetadata.into(); + let fault: Error = ErrorKind::BackendFailure.into(); + assert!(client.backtrace().is_some()); + assert!(fault.backtrace().is_some()); + } + + #[test] + fn opaque_error_preserves_source_and_origin_trace() { + let error = Error::with_source(ErrorKind::BackendFailure, io::Error::other("backend down")); + let standard_error: &dyn std::error::Error = &error; + + assert_eq!(error.kind(), ErrorKind::BackendFailure); + assert_eq!(standard_error.source().unwrap().to_string(), "backend down"); + assert!(error.backtrace().is_some()); + } + + #[test] + fn error_kind_default_message_includes_range_size() { + let error: Error = ErrorKind::RangeNotSatisfiable { total: 42 }.into(); + + assert_eq!( + error.to_string(), + "range not satisfiable (object size: 42 bytes)" + ); + } + + #[test] + fn panic_uses_the_payload_message() { + let panic = Panic::new(Box::new("task panicked")); + let error: Error = panic.into(); + + assert_eq!(error.kind(), ErrorKind::Panic); + assert_eq!(error.to_string(), "service task panicked"); + assert_eq!(error.source().unwrap().to_string(), "task panicked"); + } +} diff --git a/objectstore-service/src/service.rs b/objectstore-service/src/service.rs index b5b3e418..be0614ac 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::Result; +use crate::error::{ErrorKind, Result, ResultExt as _}; use crate::id::{ObjectContext, ObjectId}; use crate::multipart::{ AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse, @@ -205,7 +205,7 @@ impl StorageService { metadata: Metadata, stream: ClientStream, ) -> Result { - metadata.validate()?; + metadata.validate().context(ErrorKind::InvalidMetadata)?; let id = ObjectId::optional(context, key); let inner = Arc::clone(&self.inner); self.spawn("insert", async move { @@ -259,7 +259,7 @@ impl StorageService { id: ObjectId, metadata: Metadata, ) -> Result { - metadata.validate()?; + metadata.validate().context(ErrorKind::InvalidMetadata)?; self.inner.as_multipart_upload_backend()?; // Fail before clone/spawn if unsupported let inner = self.inner.clone(); self.spawn("initiate_multipart", async move { @@ -381,7 +381,6 @@ mod tests { use crate::backend::testing::{Hooks, TestBackend}; use crate::backend::tiered::TieredStorage; use crate::change_stream::ChangeStreamFactory; - use crate::error::Error; use crate::stream::{self, ClientStream}; fn make_context() -> ObjectContext { @@ -592,10 +591,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 { + let Err(error) = result else { panic!("expected Panic error"); }; - assert!(msg.contains("intentional panic in get_object"), "{msg}"); + assert_eq!(error.kind(), ErrorKind::Panic); + assert!( + error + .to_string() + .contains("intentional panic in get_object") + ); } /// In-memory backend with optional synchronization for `put_object`. @@ -734,7 +738,9 @@ mod tests { .await; assert!( - matches!(result, Err(Error::AtCapacity)), + result + .as_ref() + .is_err_and(|error| error.kind() == ErrorKind::AtCapacity), "expected AtCapacity, got {result:?}" ); @@ -788,12 +794,12 @@ 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(_)))); + assert!(result.is_err_and(|error| error.kind() == ErrorKind::Panic)); // Second operation should succeed in acquiring the permit (not AtCapacity). let result = service.get_object(id, None).await; assert!( - !matches!(result, Err(Error::AtCapacity)), + !result.is_err_and(|error| error.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..bc2a0301 100644 --- a/objectstore-service/src/stream.rs +++ b/objectstore-service/src/stream.rs @@ -66,8 +66,8 @@ 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 -/// than 500. +/// as [`ClientStream`](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. pub type ClientStream = BoxStream<'static, Result>; @@ -81,7 +81,8 @@ 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 [`ClientStream`](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 27eed920..5ce07088 100644 --- a/objectstore-service/src/streaming.rs +++ b/objectstore-service/src/streaming.rs @@ -13,14 +13,15 @@ //! regular requests. //! //! The regular acquire timeout applies: Operations that cannot acquire a permit within the -//! configured queue timeout fail with [`Error::AtCapacity`]. +//! configured queue timeout fail with [`AtCapacity`](crate::error::ErrorKind::AtCapacity). //! //! ## Concurrency Model //! //! [`StreamExecutor::execute`] uses `buffer_unordered` with the bulk budget as the concurrency //! bound. The input stream is pulled lazily and 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. +//! surfaces as a [`Panic`](crate::error::ErrorKind::Panic) for that item and does not affect +//! the others. use std::sync::Arc; @@ -229,8 +230,8 @@ impl StreamExecutor { /// that already hold a permit proceeds concurrently. /// /// Operations that cannot acquire a permit within the configured queue - /// timeout fail with [`Error::AtCapacity`]. Results are yielded in - /// completion order (not submission order). + /// timeout fail with [`AtCapacity`](crate::error::ErrorKind::AtCapacity). + /// Results are yielded in completion order (not submission order). pub fn execute( self, context: ObjectContext, @@ -342,7 +343,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}; @@ -586,7 +587,10 @@ mod tests { assert_eq!(outcomes.len(), 1); assert!( - matches!(&outcomes[0].1, Err(Error::AtCapacity)), + outcomes[0] + .1 + .as_ref() + .is_err_and(|error| error.kind() == ErrorKind::AtCapacity), "expected AtCapacity, got {:?}", outcomes[0].1, ); From 929da830d36b75564e7d4fe132b9ce46f6c1bf11 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Mon, 31 Aug 2026 16:06:28 +0200 Subject: [PATCH 2/4] test(service): Assert panic source details --- objectstore-service/src/service.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/objectstore-service/src/service.rs b/objectstore-service/src/service.rs index be0614ac..1cc093e6 100644 --- a/objectstore-service/src/service.rs +++ b/objectstore-service/src/service.rs @@ -595,10 +595,10 @@ mod tests { panic!("expected Panic error"); }; assert_eq!(error.kind(), ErrorKind::Panic); - assert!( - error - .to_string() - .contains("intentional panic in get_object") + assert_eq!(error.to_string(), "service task panicked"); + assert_eq!( + std::error::Error::source(&error).unwrap().to_string(), + "intentional panic in get_object" ); } From 81741fb112f24767a933bddc8bd441188803af30 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Tue, 1 Sep 2026 10:26:53 +0200 Subject: [PATCH 3/4] fix(service): Refine client error handling --- objectstore-service/src/backend/extensions.rs | 24 ++++++++++++++++++- objectstore-service/src/backend/tiered.rs | 13 +++++++++- 2 files changed, 35 insertions(+), 2 deletions(-) diff --git a/objectstore-service/src/backend/extensions.rs b/objectstore-service/src/backend/extensions.rs index b4b9b237..e058187a 100644 --- a/objectstore-service/src/backend/extensions.rs +++ b/objectstore-service/src/backend/extensions.rs @@ -205,6 +205,10 @@ impl BackendDetail { message: String::new(), } } + + fn is_empty(&self) -> bool { + self.code.is_empty() && self.message.is_empty() + } } impl fmt::Display for BackendDetail { @@ -248,7 +252,11 @@ impl BackendResponseError { impl fmt::Display for BackendResponseError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "{} ({}). {}", self.context, self.status, self.detail) + write!(f, "{} ({})", self.context, self.status)?; + if !self.detail.is_empty() { + write!(f, ". {}", self.detail)?; + } + Ok(()) } } @@ -290,4 +298,18 @@ mod tests { "GCS: get object (429 Too Many Requests). too many requests (backend code rateLimitExceeded)" ); } + + #[test] + fn backend_response_omits_separator_without_detail() { + let error = BackendResponseError::new( + "GCS: get object", + StatusCode::INTERNAL_SERVER_ERROR, + BackendDetail::none(), + ); + + assert_eq!( + error.to_string(), + "GCS: get object (500 Internal Server Error)" + ); + } } diff --git a/objectstore-service/src/backend/tiered.rs b/objectstore-service/src/backend/tiered.rs index deceb207..7a68f08d 100644 --- a/objectstore-service/src/backend/tiered.rs +++ b/objectstore-service/src/backend/tiered.rs @@ -601,7 +601,7 @@ impl TryFrom<&UploadId> for TieredUploadId { let json = base64::engine::general_purpose::URL_SAFE_NO_PAD .decode(value.as_bytes()) .map_err(|e| Error::new(ErrorKind::InvalidUploadId, e.to_string()))?; - serde_json::from_slice(&json).context(ErrorKind::CorruptData) + serde_json::from_slice(&json).context(ErrorKind::InvalidUploadId) } } @@ -1641,6 +1641,17 @@ mod tests { assert_eq!(decoded, id); } + #[test] + fn malformed_multipart_upload_ids_are_invalid_upload_ids() { + let invalid_base64 = UploadId::new("%%%".into()).unwrap(); + let malformed_json = UploadId::new("bm90IGpzb24".into()).unwrap(); + + for upload_id in [&invalid_base64, &malformed_json] { + let error = TieredUploadId::try_from(upload_id).unwrap_err(); + assert_eq!(error.kind(), ErrorKind::InvalidUploadId); + } + } + #[tokio::test] async fn multipart_single_part_roundtrip() { let (storage, hv, lt, _) = make_tiered_storage(); From a0d1ddc444659aeaf444e97e4f46579e6cfe9be4 Mon Sep 17 00:00:00 2001 From: Jan Michael Auer Date: Tue, 1 Sep 2026 18:09:30 +0200 Subject: [PATCH 4/4] ref(service): Add diagnostic error context --- .../examples/capture_service_error.rs | 88 ++++++ objectstore-server/src/endpoints/objects.rs | 8 +- objectstore-service/src/backend/bigtable.rs | 63 +++-- objectstore-service/src/backend/extensions.rs | 26 +- objectstore-service/src/backend/gcs.rs | 149 ++++++---- objectstore-service/src/backend/in_memory.rs | 28 +- objectstore-service/src/backend/local_fs.rs | 259 +++++++++++++----- .../src/backend/s3_compatible.rs | 52 ++-- objectstore-service/src/backend/tiered.rs | 9 +- objectstore-service/src/concurrency.rs | 2 +- objectstore-service/src/error.rs | 115 +++++--- objectstore-service/src/service.rs | 4 +- 12 files changed, 565 insertions(+), 238 deletions(-) create mode 100644 objectstore-server/examples/capture_service_error.rs diff --git a/objectstore-server/examples/capture_service_error.rs b/objectstore-server/examples/capture_service_error.rs new file mode 100644 index 00000000..9bd4e9b0 --- /dev/null +++ b/objectstore-server/examples/capture_service_error.rs @@ -0,0 +1,88 @@ +//! Sends a production-shaped service error to Sentry and waits for delivery. +//! +//! Configure this example with the same `OS__SENTRY__*` environment variables as the server. In +//! particular, `OS__SENTRY__DSN` must be set. An optional first argument selects the normal server +//! YAML configuration file. + +use std::error::Error as _; +use std::net::TcpListener; +use std::path::Path; +use std::time::Duration; + +use anyhow::ensure; +use objectstore_server::config::Config; +use objectstore_service::concurrency::spawn_metered; +use objectstore_service::error::{ErrorKind, ResultExt as _}; + +const FLUSH_TIMEOUT: Duration = Duration::from_secs(10); + +fn main() -> anyhow::Result<()> { + let config_path = std::env::args_os().nth(1); + let config = Config::load(config_path.as_deref().map(Path::new))?; + + rustls::crypto::ring::default_provider() + .install_default() + .map_err(|_| anyhow::anyhow!("failed to install rustls crypto provider"))?; + + // Keep the guard alive until after the explicit flush below. This is the same Sentry + // initialization used by the production server, including release, sampling, logs, tags, + // environment, and server name. + let _sentry_guard = objectstore_server::observability::init_sentry(&config) + .ok_or_else(|| anyhow::anyhow!("OS__SENTRY__DSN must be configured"))?; + + let runtime = tokio::runtime::Builder::new_multi_thread() + .thread_name("sentry-test-rt") + .enable_all() + .worker_threads(config.runtime.worker_threads) + .build()?; + let _runtime_guard = runtime.enter(); + + // This installs the same tracing-to-Sentry layer that reports service task failures in + // production. + objectstore_log::init(&config.logging); + + let endpoint = unused_local_endpoint()?; + let error = runtime.block_on(async move { + let result: objectstore_service::error::Result<()> = + spawn_metered("sentry_test_backend_request", (), async move { + reqwest::Client::builder() + .no_proxy() + .timeout(Duration::from_secs(2)) + .build() + .context(ErrorKind::BackendFailure, "building the Sentry test client")? + .get(endpoint) + .send() + .await + .context(ErrorKind::BackendFailure, "sending the Sentry test request")?; + Ok(()) + }) + .await; + + result.expect_err("request to a closed local port unexpectedly succeeded") + }); + + ensure!(error.kind() == ErrorKind::BackendFailure); + ensure!( + error.source().is_some(), + "service error lost its reqwest source" + ); + + let client = sentry::Hub::current() + .client() + .ok_or_else(|| anyhow::anyhow!("Sentry client was not initialized"))?; + ensure!( + client.flush(Some(FLUSH_TIMEOUT)), + "Sentry did not flush within {FLUSH_TIMEOUT:?}" + ); + + eprintln!("Sentry service-error event submitted and flushed"); + Ok(()) +} + +/// Reserves and releases a loopback port so the HTTP request produces a real connection error. +fn unused_local_endpoint() -> anyhow::Result { + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let address = listener.local_addr()?; + drop(listener); + Ok(format!("http://{address}/objectstore-sentry-test")) +} diff --git a/objectstore-server/src/endpoints/objects.rs b/objectstore-server/src/endpoints/objects.rs index 87dc36ac..ad12f9b0 100644 --- a/objectstore-server/src/endpoints/objects.rs +++ b/objectstore-server/src/endpoints/objects.rs @@ -95,7 +95,9 @@ async fn object_get( }; let stream = state.meter_stream(stream, &context); - let mut metadata_headers = metadata.to_headers("").context(ErrorKind::Internal)?; + let mut metadata_headers = metadata + .to_headers("") + .context(ErrorKind::Internal, "encoding object response metadata")?; let mut response = match content_range { Some(ref content_range) => { @@ -129,7 +131,9 @@ async fn object_head(service: AuthAwareService, Xt(id): Xt) -> ApiResu return Ok(StatusCode::NOT_FOUND.into_response()); }; - let mut headers = metadata.to_headers("").context(ErrorKind::Internal)?; + let mut headers = metadata + .to_headers("") + .context(ErrorKind::Internal, "encoding object response metadata")?; insert_content_length(&mut headers, &metadata); let mut response = (StatusCode::OK, headers).into_response(); diff --git a/objectstore-service/src/backend/bigtable.rs b/objectstore-service/src/backend/bigtable.rs index dbd44d8d..f57a5a78 100644 --- a/objectstore-service/src/backend/bigtable.rs +++ b/objectstore-service/src/backend/bigtable.rs @@ -486,7 +486,8 @@ fn object_mutations( // Record the payload size in the metadata before persisting it. metadata.size = Some(payload.len()); - let metadata_bytes = serde_json::to_vec(&metadata).context(ErrorKind::Internal)?; + let metadata_bytes = serde_json::to_vec(&metadata) + .context(ErrorKind::Internal, "encoding Bigtable object metadata")?; let mutations = [ // NB: We explicitly delete the row to clear metadata on overwrite. @@ -573,7 +574,8 @@ 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).context(ErrorKind::Internal)?, + value: serde_json::to_vec(&tombstone_meta) + .context(ErrorKind::Internal, "encoding Bigtable tombstone metadata")?, })), ]) } @@ -645,8 +647,10 @@ impl RowData { payload = cell.value; } COLUMN_TOMBSTONE_META => { - tombstone_meta_opt = - Some(serde_json::from_slice(&cell.value).context(ErrorKind::CorruptData)?); + tombstone_meta_opt = Some(serde_json::from_slice(&cell.value).context( + ErrorKind::CorruptData, + "decoding Bigtable tombstone metadata", + )?); } COLUMN_METADATA => { if let Ok(legacy_meta) = @@ -659,9 +663,10 @@ impl RowData { expiration_policy: legacy_meta.expiration_policy, }); } else { - metadata_opt = Some( - serde_json::from_slice(&cell.value).context(ErrorKind::CorruptData)?, - ); + metadata_opt = Some(serde_json::from_slice(&cell.value).context( + ErrorKind::CorruptData, + "decoding Bigtable object metadata", + )?); } } _ => {} @@ -726,9 +731,9 @@ fn parse_redirect_target(redirect_path: &[u8], tombstone_id: &ObjectId) -> Resul Ok(tombstone_id.clone()) } else { let redirect_str = std::str::from_utf8(redirect_path) - .map_err(|_| Error::new(ErrorKind::CorruptData, "invalid UTF-8 in redirect path"))?; + .context(ErrorKind::CorruptData, "decoding Bigtable redirect target")?; ObjectId::from_storage_path(redirect_str) - .ok_or_else(|| Error::new(ErrorKind::CorruptData, "corrupt redirect path")) + .ok_or_else(|| Error::new(ErrorKind::CorruptData, "parsing Bigtable redirect target")) } } @@ -1004,7 +1009,10 @@ impl Backend for BigTableBackend { TieredGet::Object(metadata, content_range, payload) => { Ok(Some((metadata, content_range, payload))) } - TieredGet::Tombstone(_) => Err(Error::new(ErrorKind::Internal, "unexpected tombstone")), + TieredGet::Tombstone(_) => Err(Error::new( + ErrorKind::Internal, + "unexpected Bigtable tombstone", + )), TieredGet::NotFound => Ok(None), } } @@ -1013,9 +1021,10 @@ 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::new(ErrorKind::Internal, "unexpected tombstone")) - } + TieredMetadata::Tombstone(_) => Err(Error::new( + ErrorKind::Internal, + "unexpected Bigtable tombstone", + )), TieredMetadata::NotFound => Ok(None), } } @@ -1086,7 +1095,7 @@ impl HighVolumeBackend for BigTableBackend { Err(Error::new( ErrorKind::Internal, - "BigTable: race loop in put_non_tombstone", + "Bigtable put race exhausted", )) } @@ -1200,7 +1209,7 @@ impl HighVolumeBackend for BigTableBackend { Err(Error::new( ErrorKind::Internal, - "BigTable: race loop in delete_non_tombstone", + "Bigtable delete race exhausted", )) } @@ -1270,16 +1279,9 @@ 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::new( - ErrorKind::Internal, - format!( - "TTL duration overflow: {} plus {}s cannot be represented as SystemTime", - humantime::format_rfc3339_seconds(from), - ttl.as_secs() - ), - ) - })?; + let deadline = from + .checked_add(ttl) + .ok_or_else(|| Error::new(ErrorKind::Internal, "calculating Bigtable expiration"))?; system_time_to_micros(deadline) } @@ -1291,10 +1293,12 @@ 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) - .context(ErrorKind::Internal)? + .context(ErrorKind::Internal, "converting Bigtable timestamp")? .as_millis(); - (millis * 1000).try_into().context(ErrorKind::Internal) + (millis * 1000) + .try_into() + .context(ErrorKind::Internal, "converting Bigtable timestamp") } /// Converts a wall-clock time to Bigtable's microsecond timestamp, saturating at `i64::MAX` @@ -1345,7 +1349,10 @@ 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(e).context(ErrorKind::BackendFailure); + return Err(e).context( + ErrorKind::BackendFailure, + format!("running Bigtable {context}"), + ); } Err(e) => { retry_count += 1; diff --git a/objectstore-service/src/backend/extensions.rs b/objectstore-service/src/backend/extensions.rs index e058187a..3f30476c 100644 --- a/objectstore-service/src/backend/extensions.rs +++ b/objectstore-service/src/backend/extensions.rs @@ -82,7 +82,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 -/// a backend-failure service error with the extracted error code and message. +/// a backend-response service error with the extracted error code and message. /// /// Implemented for both [`reqwest::Response`] and `Result` so it can be /// chained directly. @@ -128,7 +128,11 @@ impl ResponseExt for Response { return Ok(self); }; self.drain_body().await; - return Err(e.into()); + return Err(Error::with_context( + ErrorKind::BackendResponse(status), + context, + e, + )); }; Err(BackendResponseError::new(context, status, detail).into()) @@ -145,7 +149,7 @@ impl ResponseExt for Result { Ok(resp) => resp.check_error(context).await, Err(e) => Err(match stream::unpack_client_error(&e) { Some(ce) => ce.into(), - None => e.into(), + None => Error::with_context(ErrorKind::BackendFailure, context, e), }), } } @@ -264,7 +268,9 @@ impl StdError for BackendResponseError {} impl From for Error { fn from(source: BackendResponseError) -> Self { - Self::with_source(ErrorKind::BackendResponse(source.status), source) + let kind = ErrorKind::BackendResponse(source.status); + let context = source.context.clone(); + Self::with_context(kind, context, source) } } @@ -280,7 +286,7 @@ mod tests { #[test] fn backend_response_preserves_status_and_structured_source() { let error: Error = BackendResponseError::new( - "GCS: get object", + "getting a GCS object", StatusCode::TOO_MANY_REQUESTS, BackendDetail { code: "rateLimitExceeded".to_owned(), @@ -293,23 +299,27 @@ mod tests { error.kind(), ErrorKind::BackendResponse(StatusCode::TOO_MANY_REQUESTS) ); + assert_eq!( + error.to_string(), + "backend returned HTTP 429 Too Many Requests: getting a GCS object" + ); assert_eq!( error.source().unwrap().to_string(), - "GCS: get object (429 Too Many Requests). too many requests (backend code rateLimitExceeded)" + "getting a GCS object (429 Too Many Requests). too many requests (backend code rateLimitExceeded)" ); } #[test] fn backend_response_omits_separator_without_detail() { let error = BackendResponseError::new( - "GCS: get object", + "getting a GCS object", StatusCode::INTERNAL_SERVER_ERROR, BackendDetail::none(), ); assert_eq!( error.to_string(), - "GCS: get object (500 Internal Server Error)" + "getting a GCS object (500 Internal Server Error)" ); } } diff --git a/objectstore-service/src/backend/gcs.rs b/objectstore-service/src/backend/gcs.rs index 9241b5c4..625901c7 100644 --- a/objectstore-service/src/backend/gcs.rs +++ b/objectstore-service/src/backend/gcs.rs @@ -259,18 +259,18 @@ impl GcsObject { .remove(&GcsMetaKey::Expiration) .map(|s| s.parse()) .transpose() - .context(ErrorKind::CorruptData)? + .context(ErrorKind::CorruptData, "decoding GCS expiration policy")? .unwrap_or_default(); let origin = self .metadata .remove(&GcsMetaKey::Origin) - .map(|value| decode_gcs_meta_value(&value)) + .map(|value| decode_gcs_meta_value(&value, "decoding GCS origin metadata")) .transpose()?; let filename = self .metadata .remove(&GcsMetaKey::Filename) - .map(|value| decode_gcs_meta_value(&value)) + .map(|value| decode_gcs_meta_value(&value, "decoding GCS filename metadata")) .transpose()?; let content_type = self.content_type; @@ -278,23 +278,26 @@ impl GcsObject { .content_encoding .map(|s| s.parse()) .transpose() - .context(ErrorKind::CorruptData)?; + .context(ErrorKind::CorruptData, "decoding GCS compression")?; let size = self .size .map(|size| size.parse()) .transpose() - .context(ErrorKind::CorruptData)?; + .context(ErrorKind::CorruptData, "decoding GCS object size")?; let time_created = self.time_created; // At this point, all built-in metadata should have been removed from self.metadata. let mut custom = BTreeMap::new(); for (key, value) in self.metadata { if let GcsMetaKey::Custom(custom_key) = key { - custom.insert(custom_key, decode_gcs_meta_value(&value)?); + custom.insert( + custom_key, + decode_gcs_meta_value(&value, "decoding GCS custom metadata")?, + ); } else { return Err(Error::new( ErrorKind::CorruptData, - format!("GCS: unexpected built-in metadata key in object metadata: {key}"), + format!("unexpected GCS metadata key: {key}"), )); } } @@ -388,7 +391,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().context(ErrorKind::Internal)?, + formatted + .to_string() + .parse() + .context(ErrorKind::Internal, "encoding GCS custom-time header")?, ); } @@ -398,7 +404,7 @@ fn metadata_to_gcs_headers(metadata: &Metadata) -> Result { compression .to_string() .parse() - .context(ErrorKind::Internal)?, + .context(ErrorKind::Internal, "encoding GCS content-encoding header")?, ); } @@ -426,8 +432,8 @@ fn metadata_to_gcs_headers(metadata: &Metadata) -> Result { } /// Decodes a stored GCS metadata value into its logical string. -fn decode_gcs_meta_value(value: &str) -> Result { - headers::decode_header_str(value).context(ErrorKind::CorruptData) +fn decode_gcs_meta_value(value: &str, context: &'static str) -> Result { + headers::decode_header_str(value).context(ErrorKind::CorruptData, context) } /// Inserts a single `x-goog-meta-*` header, escaping the value for transport. @@ -447,7 +453,10 @@ fn insert_gcs_meta_header( ) -> Result<()> { let header_name = format!("x-goog-meta-{key}"); headers.insert( - HeaderName::try_from(&header_name).context(ErrorKind::Internal)?, + HeaderName::try_from(&header_name).context( + ErrorKind::Internal, + format!("encoding GCS metadata header {header_name}"), + )?, headers::encode_header_value(value), ); Ok(()) @@ -532,10 +541,7 @@ impl GcsBackend { .map_err(|()| { Error::new( ErrorKind::Internal, - format!( - "GCS: invalid endpoint URL, {} cannot be a base", - self.endpoint - ), + format!("building GCS object URL from {}", self.endpoint), ) })? .extend(&["storage", "v1", "b", &self.bucket, "o", &path]); @@ -551,10 +557,7 @@ impl GcsBackend { .map_err(|()| { Error::new( ErrorKind::Internal, - format!( - "GCS: invalid endpoint URL, {} cannot be a base", - self.endpoint - ), + format!("building GCS object URL from {}", self.endpoint), ) })? .extend(&["upload", "storage", "v1", "b", &self.bucket, "o"]); @@ -578,10 +581,7 @@ impl GcsBackend { let mut segments = url.path_segments_mut().map_err(|()| { Error::new( ErrorKind::Internal, - format!( - "GCS: invalid endpoint URL, {} cannot be a base", - self.endpoint - ), + format!("building GCS object URL from {}", self.endpoint), ) })?; segments.push(&self.bucket); @@ -596,7 +596,10 @@ impl GcsBackend { async fn request(&self, method: Method, url: impl IntoUrl) -> Result { let mut builder = self.client.request(method, url); if let Some(provider) = &self.token_provider { - let token = provider.token(TOKEN_SCOPES).await?; + let token = provider.token(TOKEN_SCOPES).await.context( + ErrorKind::BackendFailure, + "getting GCS authentication token", + )?; builder = builder.bearer_auth(token.as_str()); } Ok(builder) @@ -641,7 +644,8 @@ impl GcsBackend { .request(Method::GET, object_url.clone()) .await? .send_traced() - .await?; + .await + .context(ErrorKind::BackendFailure, "getting GCS object metadata")?; if resp.status() == StatusCode::NOT_FOUND { resp.drain_body().await; @@ -649,10 +653,14 @@ impl GcsBackend { } let metadata: GcsObject = resp - .check_error("GCS: get metadata status") + .check_error("getting GCS object metadata") .await? .json() - .await?; + .await + .context( + ErrorKind::CorruptData, + "decoding GCS object metadata response", + )?; Ok(Some(metadata)) }) @@ -729,7 +737,7 @@ impl GcsBackend { .json(&CustomTimeRequest { custom_time }) .send_traced() .await - .check_error("GCS: update custom time") + .check_error("updating GCS custom time") .await { Ok(response) => { @@ -782,7 +790,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).context(ErrorKind::Internal)?; + let metadata_json = serde_json::to_string(&gcs_metadata) + .context(ErrorKind::Internal, "encoding GCS upload metadata")?; let multipart = multipart::Form::new() .part( @@ -795,7 +804,7 @@ impl Backend for GcsBackend { "media", multipart::Part::stream(Body::wrap_stream(stream.boxed())) .mime_str(&metadata.content_type) - .context(ErrorKind::InvalidMetadata)?, + .context(ErrorKind::InvalidMetadata, "encoding GCS content type")?, ); // GCS requires a multipart/related request. Its body looks identical to @@ -810,7 +819,7 @@ impl Backend for GcsBackend { .header(header::CONTENT_TYPE, content_type) .send_traced() .await - .check_error("GCS: upload object") + .check_error("uploading a GCS object") .await?; let stored_size = read_stored_content_length(response).await; @@ -846,7 +855,10 @@ impl Backend for GcsBackend { if let Some(r) = range { req = req.header(header::RANGE, r.to_header_value()); } - let resp = req.send_traced().await?; + let resp = req + .send_traced() + .await + .context(ErrorKind::BackendFailure, "getting a GCS object payload")?; if resp.status() == StatusCode::RANGE_NOT_SATISFIABLE { let raw = resp @@ -856,16 +868,15 @@ impl Backend for GcsBackend { let total = raw.and_then(ContentRange::parse_unsatisfiable_total); let err = match total { Some(total) => ErrorKind::RangeNotSatisfiable { total }.into(), - None => Error::new( - ErrorKind::BackendFailure, - format!("GCS: 416 response with invalid Content-Range: {raw:?}"), - ), + None => { + Error::new(ErrorKind::BackendFailure, "invalid GCS 416 Content-Range") + } }; resp.drain_body().await; return Err(err); } - resp.check_error("GCS: get payload").await + resp.check_error("getting a GCS object payload").await }) .await?; @@ -877,10 +888,7 @@ impl Backend for GcsBackend { .and_then(|v| v.to_str().ok()) .and_then(|s| s.parse::().ok()) .ok_or_else(|| { - Error::new( - ErrorKind::BackendFailure, - "GCS: 206 response missing valid Content-Range header", - ) + Error::new(ErrorKind::BackendFailure, "missing GCS 206 Content-Range") })?, ) } else { @@ -913,7 +921,8 @@ impl Backend for GcsBackend { .request(Method::DELETE, object_url.clone()) .await? .send_traced() - .await?; + .await + .context(ErrorKind::BackendFailure, "deleting a GCS object")?; // Do not error for objects that do not exist if resp.status() == StatusCode::NOT_FOUND { @@ -921,7 +930,7 @@ impl Backend for GcsBackend { return Ok(false); } - resp.check_error("GCS: delete object") + resp.check_error("deleting a GCS object") .await? .drain_body() .await; @@ -1067,7 +1076,7 @@ impl MultipartUploadBackend for GcsBackend { metadata .content_type .parse() - .context(ErrorKind::InvalidMetadata)?, + .context(ErrorKind::InvalidMetadata, "encoding GCS content type")?, ); headers.insert( header::CONTENT_LENGTH, @@ -1084,13 +1093,19 @@ impl MultipartUploadBackend for GcsBackend { .headers(headers) .send_traced() .await - .check_error("GCS: initiate multipart upload") + .check_error("initiating a GCS multipart upload") .await?; - let body = resp.bytes().await?; + let body = resp.bytes().await.context( + ErrorKind::BackendFailure, + "reading GCS initiate-multipart response", + )?; let xml: XmlInitiateMultipartUploadResponse = - quick_xml::de::from_reader(body.as_ref()).context(ErrorKind::CorruptData)?; + quick_xml::de::from_reader(body.as_ref()).context( + ErrorKind::CorruptData, + "decoding GCS initiate-multipart response", + )?; xml.try_into() } @@ -1127,7 +1142,7 @@ impl MultipartUploadBackend for GcsBackend { let resp = builder .send_traced() .await - .check_error("GCS: upload part") + .check_error("uploading a GCS multipart part") .await?; let etag = resp @@ -1138,7 +1153,7 @@ impl MultipartUploadBackend for GcsBackend { .ok_or_else(|| { Error::new( ErrorKind::BackendFailure, - "GCS: upload part response missing ETag header", + "GCS upload-part response missing ETag", ) })?; @@ -1176,13 +1191,16 @@ impl MultipartUploadBackend for GcsBackend { .await? .send_traced() .await - .check_error("GCS: list parts") + .check_error("listing GCS multipart parts") .await?; - let body = resp.bytes().await?; + let body = resp + .bytes() + .await + .context(ErrorKind::BackendFailure, "reading GCS list-parts response")?; - let xml: XmlListPartsResponse = - quick_xml::de::from_reader(body.as_ref()).context(ErrorKind::CorruptData)?; + let xml: XmlListPartsResponse = quick_xml::de::from_reader(body.as_ref()) + .context(ErrorKind::CorruptData, "decoding GCS list-parts response")?; Ok(xml.into()) } @@ -1207,13 +1225,14 @@ impl MultipartUploadBackend for GcsBackend { .request(Method::DELETE, url) .await? .send_traced() - .await?; + .await + .context(ErrorKind::BackendFailure, "aborting a GCS multipart upload")?; // XXX: real S3 would return 404 here if the upload has been recently completed and we // would have to handle it. It turns out GCS returns 204 instead, so we don't need to // handle that case. - resp.check_error("GCS: abort multipart upload") + resp.check_error("aborting a GCS multipart upload") .await? .drain_body() .await; @@ -1236,7 +1255,10 @@ 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).context(ErrorKind::Internal)?; + let xml = quick_xml::se::to_string(&body).context( + ErrorKind::Internal, + "encoding GCS complete-multipart request", + )?; self.with_retry("complete_multipart", || { let url = url.clone(); @@ -1249,14 +1271,17 @@ impl MultipartUploadBackend for GcsBackend { .body(xml) .send_traced() .await - .check_error("GCS: complete multipart upload") + .check_error("completing a GCS multipart upload") .await?; // XXX: real S3 would return 404 here if the upload has been recently completed and we // would have to handle it. It turns out GCS returns 200 instead, so we don't need to // handle that case. - let body = resp.bytes().await?; + let body = resp.bytes().await.context( + ErrorKind::BackendFailure, + "reading GCS complete-multipart response", + )?; let error = quick_xml::de::from_reader::<_, XmlError>(body.as_ref()) .ok() @@ -1338,10 +1363,14 @@ mod tests { .await? .send_traced() .await - .check_error("GCS: get metadata request") + .check_error("getting GCS object metadata") .await? .json::() .await + .context( + ErrorKind::BackendFailure, + "decoding GCS object metadata response", + ) .map(|object| (object.generation, object.metageneration))?) } diff --git a/objectstore-service/src/backend/in_memory.rs b/objectstore-service/src/backend/in_memory.rs index 2934d6be..e3635886 100644 --- a/objectstore-service/src/backend/in_memory.rs +++ b/objectstore-service/src/backend/in_memory.rs @@ -128,9 +128,10 @@ 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::new(ErrorKind::Internal, "unexpected tombstone")) - } + Some(StoreEntry::Tombstone(_)) => Err(Error::new( + ErrorKind::Internal, + "unexpected in-memory tombstone", + )), Some(StoreEntry::Object(mut metadata, bytes)) => { let total = bytes.len() as u64; metadata.size = Some(bytes.len()); @@ -291,7 +292,12 @@ 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::new(ErrorKind::BackendFailure, "multipart upload not found"))?; + .ok_or_else(|| { + Error::new( + ErrorKind::BackendFailure, + "in-memory multipart upload not found", + ) + })?; upload.parts.insert( part_number, @@ -313,9 +319,12 @@ impl MultipartUploadBackend for InMemoryBackend { part_number_marker: Option, ) -> Result { let store = self.multipart_store.lock().unwrap(); - let upload = store - .get(&(id.clone(), upload_id.clone())) - .ok_or_else(|| Error::new(ErrorKind::BackendFailure, "multipart upload not found"))?; + let upload = store.get(&(id.clone(), upload_id.clone())).ok_or_else(|| { + Error::new( + ErrorKind::BackendFailure, + "in-memory multipart upload not found", + ) + })?; let iter = upload .parts @@ -379,7 +388,10 @@ impl MultipartUploadBackend for InMemoryBackend { let assembled = { let store = self.multipart_store.lock().unwrap(); let upload = store.get(&key).ok_or_else(|| { - Error::new(ErrorKind::BackendFailure, "multipart upload not found") + Error::new( + ErrorKind::BackendFailure, + "in-memory multipart upload not found", + ) })?; for completed in &parts { diff --git a/objectstore-service/src/backend/local_fs.rs b/objectstore-service/src/backend/local_fs.rs index bbc672d5..f24bcad2 100644 --- a/objectstore-service/src/backend/local_fs.rs +++ b/objectstore-service/src/backend/local_fs.rs @@ -85,31 +85,56 @@ impl Backend for LocalFsBackend { ) -> Result { let path = self.path.join(id.as_storage_path().to_string()); objectstore_log::debug!(path=%path.display(), "Writing to local_fs backend"); - tokio::fs::create_dir_all(path.parent().unwrap()).await?; + tokio::fs::create_dir_all(path.parent().unwrap()) + .await + .context( + ErrorKind::BackendFailure, + "creating local-fs object directory", + )?; let file = OpenOptions::new() .create(true) .write(true) .truncate(true) .open(path) - .await?; + .await + .context( + ErrorKind::BackendFailure, + "opening local-fs object for writing", + )?; let mut reader = pin!(StreamReader::new(stream)); let mut writer = BufWriter::new(file); - let metadata_json = serde_json::to_string(metadata).context(ErrorKind::Internal)?; - writer.write_all(metadata_json.as_bytes()).await?; - writer.write_all(b"\n").await?; + let metadata_json = serde_json::to_string(metadata) + .context(ErrorKind::Internal, "encoding local-fs object metadata")?; + writer.write_all(metadata_json.as_bytes()).await.context( + ErrorKind::BackendFailure, + "writing local-fs object metadata", + )?; + writer.write_all(b"\n").await.context( + ErrorKind::BackendFailure, + "writing local-fs object metadata", + )?; tokio::io::copy(&mut reader, &mut writer) .await .map_err(|e| match stream::unpack_client_error(&e) { Some(ce) => Error::from(ce), - None => Error::from(e), + None => Error::with_context( + ErrorKind::BackendFailure, + "writing local-fs object payload", + e, + ), })?; - writer.flush().await?; + writer + .flush() + .await + .context(ErrorKind::BackendFailure, "flushing local-fs object")?; let file = writer.into_inner(); - file.sync_data().await?; + file.sync_data() + .await + .context(ErrorKind::BackendFailure, "syncing local-fs object")?; drop(file); Ok(()) @@ -126,22 +151,30 @@ impl Backend for LocalFsBackend { objectstore_log::debug!("Object not found"); return Ok(None); } - err => err?, + err => err.context( + ErrorKind::BackendFailure, + "opening local-fs object for reading", + )?, }; let mut reader = BufReader::new(file); 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()).context(ErrorKind::CorruptData)?; + reader.read_line(&mut metadata_line).await.context( + ErrorKind::BackendFailure, + "reading local-fs object metadata", + )?; + let file_len = reader + .get_ref() + .metadata() + .await + .context(ErrorKind::BackendFailure, "reading local-fs object size")? + .len(); + let mut metadata: Metadata = serde_json::from_str(metadata_line.trim_end()) + .context(ErrorKind::CorruptData, "decoding local-fs object metadata")?; let payload_size = file_len .checked_sub(metadata_line.len() as u64) .ok_or_else(|| { - Error::new( - ErrorKind::CorruptData, - "local-fs file corrupted: shorter than header", - ) + Error::new(ErrorKind::CorruptData, "reading truncated local-fs object") })?; metadata.size = Some(payload_size as usize); @@ -154,7 +187,10 @@ impl Backend for LocalFsBackend { total: payload_size, })?; let payload_start = metadata_line.len() as u64 + content_range.start; - reader.seek(std::io::SeekFrom::Start(payload_start)).await?; + reader + .seek(std::io::SeekFrom::Start(payload_start)) + .await + .context(ErrorKind::BackendFailure, "seeking local-fs object payload")?; let limited = reader.take(content_range.len()); (Some(content_range), ReaderStream::new(limited).boxed()) } @@ -173,7 +209,8 @@ impl Backend for LocalFsBackend { { objectstore_log::debug!("Object not found"); } - Ok(result?) + result.context(ErrorKind::BackendFailure, "deleting local-fs object")?; + Ok(()) } } @@ -195,11 +232,18 @@ impl MultipartUploadBackend for LocalFsBackend { ) -> Result { let upload_id = UploadId::new(uuid::Uuid::now_v7().to_string())?; let dir = self.multipart_dir(id, &upload_id); - tokio::fs::create_dir_all(&dir).await?; + tokio::fs::create_dir_all(&dir).await.context( + ErrorKind::BackendFailure, + "creating local-fs multipart upload", + )?; let meta_path = dir.join("metadata.json"); - let metadata_json = serde_json::to_string(metadata).context(ErrorKind::Internal)?; - tokio::fs::write(meta_path, metadata_json).await?; + let metadata_json = serde_json::to_string(metadata) + .context(ErrorKind::Internal, "encoding local-fs multipart metadata")?; + tokio::fs::write(meta_path, metadata_json).await.context( + ErrorKind::BackendFailure, + "writing local-fs multipart metadata", + )?; Ok(upload_id) } @@ -214,10 +258,13 @@ impl MultipartUploadBackend for LocalFsBackend { body: ClientStream, ) -> Result { let dir = self.multipart_dir(id, upload_id); - if !tokio::fs::try_exists(&dir).await? { + if !tokio::fs::try_exists(&dir).await.context( + ErrorKind::BackendFailure, + "checking local-fs multipart upload", + )? { return Err(Error::new( ErrorKind::BackendFailure, - "multipart upload not found", + "local-fs multipart upload not found", )); } @@ -228,7 +275,8 @@ impl MultipartUploadBackend for LocalFsBackend { "uploaded_at": SystemTime::now(), "size": content_length, }); - let header_line = serde_json::to_string(&header).context(ErrorKind::Internal)?; + let header_line = serde_json::to_string(&header) + .context(ErrorKind::Internal, "encoding local-fs part header")?; let part_path = dir.join(format!("{part_number}.part")); let file = OpenOptions::new() @@ -236,27 +284,46 @@ impl MultipartUploadBackend for LocalFsBackend { .write(true) .truncate(true) .open(part_path) - .await?; + .await + .context( + ErrorKind::BackendFailure, + "opening local-fs multipart part for writing", + )?; let mut reader = pin!(StreamReader::new(body)); let mut writer = BufWriter::new(file); - writer.write_all(header_line.as_bytes()).await?; - writer.write_all(b"\n").await?; + writer + .write_all(header_line.as_bytes()) + .await + .context(ErrorKind::BackendFailure, "writing local-fs part header")?; + writer + .write_all(b"\n") + .await + .context(ErrorKind::BackendFailure, "writing local-fs part header")?; let _bytes_copied = tokio::io::copy(&mut reader, &mut writer) .await .map_err(|e| match stream::unpack_client_error(&e) { Some(ce) => Error::from(ce), - None => Error::from(e), + None => Error::with_context( + ErrorKind::BackendFailure, + "writing local-fs multipart part payload", + e, + ), })?; // TODO: validate bytes_copied against content_length and return a BadRequest-style // error. Needs a service-layer error variant that maps to HTTP 400 without abusing // ClientError (which is meant for stream errors). - writer.flush().await?; + writer.flush().await.context( + ErrorKind::BackendFailure, + "flushing local-fs multipart part", + )?; let file = writer.into_inner(); - file.sync_data().await?; + file.sync_data() + .await + .context(ErrorKind::BackendFailure, "syncing local-fs multipart part")?; drop(file); Ok(etag) @@ -270,17 +337,26 @@ impl MultipartUploadBackend for LocalFsBackend { part_number_marker: Option, ) -> Result { let dir = self.multipart_dir(id, upload_id); - if !tokio::fs::try_exists(&dir).await? { + if !tokio::fs::try_exists(&dir).await.context( + ErrorKind::BackendFailure, + "checking local-fs multipart upload", + )? { return Err(Error::new( ErrorKind::BackendFailure, - "multipart upload not found", + "local-fs multipart upload not found", )); } - let mut entries = tokio::fs::read_dir(&dir).await?; + let mut entries = tokio::fs::read_dir(&dir).await.context( + ErrorKind::BackendFailure, + "listing local-fs multipart parts", + )?; let mut parts = Vec::new(); - while let Some(entry) = entries.next_entry().await? { + while let Some(entry) = entries.next_entry().await.context( + ErrorKind::BackendFailure, + "listing local-fs multipart parts", + )? { let name = entry.file_name(); let name_str = name.to_string_lossy(); let Some(pn_str) = name_str.strip_suffix(".part") else { @@ -294,12 +370,17 @@ impl MultipartUploadBackend for LocalFsBackend { continue; } - let file = tokio::fs::File::open(entry.path()).await?; + let file = tokio::fs::File::open(entry.path()) + .await + .context(ErrorKind::BackendFailure, "opening local-fs multipart part")?; 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()).context(ErrorKind::CorruptData)?; + reader + .read_line(&mut header_line) + .await + .context(ErrorKind::BackendFailure, "reading local-fs part header")?; + let header: serde_json::Value = serde_json::from_str(header_line.trim_end()) + .context(ErrorKind::CorruptData, "decoding local-fs part header")?; parts.push(Part { part_number: pn, @@ -335,8 +416,14 @@ impl MultipartUploadBackend for LocalFsBackend { upload_id: &UploadId, ) -> Result { let dir = self.multipart_dir(id, upload_id); - if tokio::fs::try_exists(&dir).await? { - tokio::fs::remove_dir_all(dir).await?; + if tokio::fs::try_exists(&dir).await.context( + ErrorKind::BackendFailure, + "checking local-fs multipart upload", + )? { + tokio::fs::remove_dir_all(dir).await.context( + ErrorKind::BackendFailure, + "removing local-fs multipart upload", + )?; } Ok(()) } @@ -348,18 +435,26 @@ impl MultipartUploadBackend for LocalFsBackend { parts: Vec, ) -> Result { let dir = self.multipart_dir(id, upload_id); - if !tokio::fs::try_exists(&dir).await? { + if !tokio::fs::try_exists(&dir).await.context( + ErrorKind::BackendFailure, + "checking local-fs multipart upload", + )? { return Err(Error::new( ErrorKind::BackendFailure, - "multipart upload not found", + "local-fs 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).context(ErrorKind::CorruptData)?; + let meta_bytes = tokio::fs::read(&meta_path).await.context( + ErrorKind::BackendFailure, + "reading local-fs multipart metadata", + )?; + let metadata: Metadata = serde_json::from_slice(&meta_bytes).context( + ErrorKind::CorruptData, + "decoding local-fs 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. @@ -367,19 +462,27 @@ impl MultipartUploadBackend for LocalFsBackend { // Validate all parts (headers only) before writing anything for completed in &parts { let part_path = dir.join(format!("{}.part", completed.part_number)); - if !tokio::fs::try_exists(&part_path).await? { + if !tokio::fs::try_exists(&part_path).await.context( + ErrorKind::BackendFailure, + "checking local-fs multipart part", + )? { return Ok(Some(crate::multipart::CompleteMultipartError { code: "InvalidPart".into(), message: format!("part number {} was not uploaded", completed.part_number), })); } - let file = tokio::fs::File::open(&part_path).await?; + let file = tokio::fs::File::open(&part_path) + .await + .context(ErrorKind::BackendFailure, "opening local-fs multipart part")?; 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()).context(ErrorKind::CorruptData)?; + reader + .read_line(&mut header_line) + .await + .context(ErrorKind::BackendFailure, "reading local-fs part header")?; + let header: serde_json::Value = serde_json::from_str(header_line.trim_end()) + .context(ErrorKind::CorruptData, "decoding local-fs part header")?; let stored_etag = header["etag"].as_str().unwrap_or(""); if stored_etag != completed.etag { @@ -395,35 +498,67 @@ impl MultipartUploadBackend for LocalFsBackend { // Stream parts directly to the final object file let path = self.path.join(id.as_storage_path().to_string()); - tokio::fs::create_dir_all(path.parent().unwrap()).await?; + tokio::fs::create_dir_all(path.parent().unwrap()) + .await + .context( + ErrorKind::BackendFailure, + "creating local-fs object directory", + )?; let file = OpenOptions::new() .create(true) .write(true) .truncate(true) .open(path) - .await?; + .await + .context( + ErrorKind::BackendFailure, + "opening local-fs object for writing", + )?; let mut writer = BufWriter::new(file); - let metadata_json = serde_json::to_string(&metadata).context(ErrorKind::Internal)?; - writer.write_all(metadata_json.as_bytes()).await?; - writer.write_all(b"\n").await?; + let metadata_json = serde_json::to_string(&metadata) + .context(ErrorKind::Internal, "encoding local-fs object metadata")?; + writer.write_all(metadata_json.as_bytes()).await.context( + ErrorKind::BackendFailure, + "writing local-fs object metadata", + )?; + writer.write_all(b"\n").await.context( + ErrorKind::BackendFailure, + "writing local-fs object metadata", + )?; for completed in &parts { let part_path = dir.join(format!("{}.part", completed.part_number)); - let file = tokio::fs::File::open(&part_path).await?; + let file = tokio::fs::File::open(&part_path) + .await + .context(ErrorKind::BackendFailure, "opening local-fs multipart part")?; let mut reader = BufReader::new(file); let mut header_line = String::new(); - reader.read_line(&mut header_line).await?; - tokio::io::copy(&mut reader, &mut writer).await?; + reader + .read_line(&mut header_line) + .await + .context(ErrorKind::BackendFailure, "reading local-fs part header")?; + tokio::io::copy(&mut reader, &mut writer).await.context( + ErrorKind::BackendFailure, + "assembling local-fs object payload", + )?; } - writer.flush().await?; + writer + .flush() + .await + .context(ErrorKind::BackendFailure, "flushing local-fs object")?; let file = writer.into_inner(); - file.sync_data().await?; + file.sync_data() + .await + .context(ErrorKind::BackendFailure, "syncing local-fs object")?; drop(file); // Clean up multipart state - tokio::fs::remove_dir_all(dir).await?; + tokio::fs::remove_dir_all(dir).await.context( + ErrorKind::BackendFailure, + "removing local-fs multipart upload", + )?; Ok(None) } diff --git a/objectstore-service/src/backend/s3_compatible.rs b/objectstore-service/src/backend/s3_compatible.rs index b9f7486c..27676898 100644 --- a/objectstore-service/src/backend/s3_compatible.rs +++ b/objectstore-service/src/backend/s3_compatible.rs @@ -1,5 +1,7 @@ //! S3-compatible backend with generic protocol support. +use std::convert::Infallible; +use std::error::Error as StdError; use std::time::SystemTime; use std::{fmt, io}; @@ -72,8 +74,13 @@ pub trait Token: Send + Sync { /// Provides authentication tokens for S3-compatible requests. pub trait TokenProvider: Send + Sync + 'static { + /// Error returned when a token cannot be provided. + type Error: StdError + Send + Sync + 'static; + /// Returns a fresh token, fetching or refreshing it as needed. - fn get_token(&self) -> impl Future> + Send; + fn get_token( + &self, + ) -> impl Future> + Send; } /// Placeholder [`TokenProvider`] for unauthenticated backends. @@ -81,8 +88,10 @@ pub trait TokenProvider: Send + Sync + 'static { pub struct NoToken; impl TokenProvider for NoToken { + type Error = Infallible; + #[allow(refining_impl_trait)] - async fn get_token(&self) -> anyhow::Result { + async fn get_token(&self) -> std::result::Result { unimplemented!() } } @@ -152,12 +161,7 @@ where provider .get_token() .await - .map_err(|err| { - Error::new( - ErrorKind::BackendFailure, - format!("S3: failed to get authentication token: {err}"), - ) - })? + .context(ErrorKind::BackendFailure, "getting S3 authentication token")? .as_str(), ); } @@ -182,7 +186,7 @@ where let response = builder .send_traced() .await - .context(ErrorKind::BackendFailure)?; + .context(ErrorKind::BackendFailure, "sending an S3 object request")?; if response.status() == StatusCode::NOT_FOUND { objectstore_log::debug!("Object not found"); @@ -198,20 +202,17 @@ where let total = raw.and_then(ContentRange::parse_unsatisfiable_total); let err = match total { Some(total) => ErrorKind::RangeNotSatisfiable { total }.into(), - None => Error::new( - ErrorKind::BackendFailure, - format!("S3: 416 response with invalid Content-Range: {raw:?}"), - ), + None => Error::new(ErrorKind::BackendFailure, "invalid S3 416 Content-Range"), }; response.drain_body().await; return Err(err); } - let response = response.check_error("S3: failed to get object").await?; + let response = response.check_error("getting an S3 object").await?; let headers = response.headers(); - let mut metadata = - Metadata::from_headers(headers, GCS_CUSTOM_PREFIX).context(ErrorKind::CorruptData)?; + let mut metadata = Metadata::from_headers(headers, GCS_CUSTOM_PREFIX) + .context(ErrorKind::CorruptData, "decoding S3 object metadata")?; let content_range = if response.status() == StatusCode::PARTIAL_CONTENT { let range = headers @@ -219,10 +220,7 @@ where .and_then(|v| v.to_str().ok()) .and_then(|s| s.parse::().ok()) .ok_or_else(|| { - Error::new( - ErrorKind::BackendFailure, - "S3: 206 response missing valid Content-Range header", - ) + Error::new(ErrorKind::BackendFailure, "missing S3 206 Content-Range") })?; metadata.size = Some(range.total as usize); Some(range) @@ -234,7 +232,7 @@ where .and_then(|value| value.to_str().ok()) .map(|value| value.parse::()) .transpose() - .context(ErrorKind::CorruptData)?; + .context(ErrorKind::CorruptData, "decoding S3 Content-Length")?; if let Some(size) = size { metadata.size = Some(size); @@ -280,11 +278,11 @@ where .header("x-goog-metadata-directive", "REPLACE") .headers( metadata_to_gcs_headers(metadata, GCS_CUSTOM_PREFIX) - .context(ErrorKind::InvalidMetadata)?, + .context(ErrorKind::InvalidMetadata, "encoding S3 object metadata")?, ) .send_traced() .await - .check_error("S3: update expiration time") + .check_error("updating S3 expiration") .await? .drain_body() .await; @@ -333,12 +331,12 @@ impl Backend for S3CompatibleBackend { .await? .headers( metadata_to_gcs_headers(metadata, GCS_CUSTOM_PREFIX) - .context(ErrorKind::InvalidMetadata)?, + .context(ErrorKind::InvalidMetadata, "encoding S3 object metadata")?, ) .body(Body::wrap_stream(stream)) .send_traced() .await - .check_error("S3: failed to put object") + .check_error("uploading an S3 object") .await? .drain_body() .await; @@ -375,7 +373,7 @@ impl Backend for S3CompatibleBackend { .await? .send_traced() .await - .context(ErrorKind::BackendFailure)?; + .context(ErrorKind::BackendFailure, "sending an S3 delete request")?; // Do not error for objects that do not exist. if response.status() == StatusCode::NOT_FOUND { @@ -384,7 +382,7 @@ impl Backend for S3CompatibleBackend { } response - .check_error("S3: failed to delete object") + .check_error("deleting an S3 object") .await? .drain_body() .await; diff --git a/objectstore-service/src/backend/tiered.rs b/objectstore-service/src/backend/tiered.rs index 7a68f08d..824aa926 100644 --- a/objectstore-service/src/backend/tiered.rs +++ b/objectstore-service/src/backend/tiered.rs @@ -587,7 +587,8 @@ impl TryInto for TieredUploadId { type Error = Error; fn try_into(self) -> Result { - let json = serde_json::to_vec(&self).context(ErrorKind::Internal)?; + let json = + serde_json::to_vec(&self).context(ErrorKind::Internal, "encoding tiered upload ID")?; Ok(UploadId::new( base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(json), )?) @@ -600,8 +601,8 @@ 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::new(ErrorKind::InvalidUploadId, e.to_string()))?; - serde_json::from_slice(&json).context(ErrorKind::InvalidUploadId) + .kind(ErrorKind::InvalidUploadId)?; + serde_json::from_slice(&json).kind(ErrorKind::InvalidUploadId) } } @@ -822,7 +823,7 @@ impl MultipartUploadBackend for TieredStorage { ); return Err(Error::new( ErrorKind::BackendFailure, - "completed multipart object not found in long-term storage", + "tiered multipart object missing from long-term storage", )); } // Failed to `get_metadata`, cannot proceed. diff --git a/objectstore-service/src/concurrency.rs b/objectstore-service/src/concurrency.rs index fe46cbc7..96e5e4a7 100644 --- a/objectstore-service/src/concurrency.rs +++ b/objectstore-service/src/concurrency.rs @@ -370,7 +370,7 @@ where ); rx.await.map_err(|_| { - let error = Error::new(ErrorKind::Internal, "task dropped"); + let error = Error::new(ErrorKind::Internal, "service task dropped"); objectstore_log::error!(!!&error, operation, "Task failed"); error })? diff --git a/objectstore-service/src/error.rs b/objectstore-service/src/error.rs index 601b51f7..a4b15cf9 100644 --- a/objectstore-service/src/error.rs +++ b/objectstore-service/src/error.rs @@ -1,11 +1,10 @@ //! Semantic errors for service and backend operations. //! -//! [`Error`] deliberately exposes only a stable semantic [`ErrorKind`]. Its source chain and an -//! optional origin backtrace retain diagnostic detail without making backend implementation -//! details part of the service API. +//! [`Error`] deliberately exposes only a stable semantic [`ErrorKind`]. Human-readable context and +//! the source chain retain diagnostic detail without making backend implementation details part of +//! the service API. use std::any::Any; -use std::backtrace::Backtrace; use std::borrow::Cow; use std::error::Error as StdError; use std::fmt; @@ -93,11 +92,13 @@ impl fmt::Display for ErrorKind { } /// Opaque service error with a stable semantic kind. +/// +/// Its string representation is the kind followed by `: ` and human-readable context when context +/// is present. The underlying source is retained separately through [`StdError::source`]. pub struct Error { kind: ErrorKind, - message: Option>, + context: Option>, source: Option>, - backtrace: Option, } impl Error { @@ -106,14 +107,9 @@ impl Error { self.kind } - /// Returns the backtrace captured where this service error originated, if enabled. - pub fn backtrace(&self) -> Option<&Backtrace> { - self.backtrace.as_ref() - } - - /// Creates an error without an underlying source and with a specific message. - pub fn new(kind: ErrorKind, message: impl Into>) -> Self { - Self::build(kind, Some(message.into()), None) + /// Creates an error without an underlying source and with human-readable context. + pub fn new(kind: ErrorKind, context: impl Into>) -> Self { + Self::build(kind, Some(context.into()), None) } /// Creates an error with an underlying source. @@ -124,16 +120,26 @@ impl Error { Self::build(kind, None, Some(Box::new(source))) } + pub(crate) fn with_context( + kind: ErrorKind, + context: impl Into>, + source: E, + ) -> Self + where + E: StdError + Send + Sync + 'static, + { + Self::build(kind, Some(context.into()), Some(Box::new(source))) + } + fn build( kind: ErrorKind, - message: Option>, + context: Option>, source: Option>, ) -> Self { Self { kind, - message, + context, source, - backtrace: Some(Backtrace::force_capture()), } } @@ -157,10 +163,11 @@ impl Error { impl fmt::Display for Error { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - match &self.message { - Some(message) => f.write_str(message), - None => self.kind.fmt(f), + self.kind.fmt(f)?; + if let Some(context) = &self.context { + write!(f, ": {context}")?; } + Ok(()) } } @@ -168,9 +175,8 @@ impl fmt::Debug for Error { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("Error") .field("kind", &self.kind) - .field("message", &self.message) + .field("context", &self.context) .field("source", &self.source) - .field("backtrace", &self.backtrace) .finish() } } @@ -193,17 +199,49 @@ impl From for Error { } } -/// Adds a semantic kind when converting an external error into a service error. +/// Adds a semantic kind and optional context when converting an external error. pub trait ResultExt { - /// Converts an external error into a service error with `kind`. - fn context(self, kind: ErrorKind) -> Result; + /// Converts an external error into a service error with `kind` and human-readable context. + /// + /// The source error is retained, while the rendered service error contains the semantic kind + /// and context. + /// + /// ``` + /// use objectstore_service::error::{ErrorKind, ResultExt as _}; + /// + /// let result = std::fs::read("missing") + /// .context(ErrorKind::BackendFailure, "reading local object"); + /// let error = result.unwrap_err(); + /// assert_eq!( + /// error.to_string(), + /// "backend operation failed: reading local object" + /// ); + /// ``` + fn context(self, kind: ErrorKind, context: impl Into>) -> Result; + + /// Converts an external error into a service error with only `kind`. + /// + /// Use this when the source already identifies the failure or when the operation is expected to + /// be infallible. The source error is still retained. + /// + /// ``` + /// use objectstore_service::error::{ErrorKind, ResultExt as _}; + /// + /// let result = "invalid".parse::().kind(ErrorKind::InvalidMetadata); + /// assert_eq!(result.unwrap_err().to_string(), "invalid object metadata"); + /// ``` + fn kind(self, kind: ErrorKind) -> Result; } impl ResultExt for std::result::Result where E: StdError + Send + Sync + 'static, { - fn context(self, kind: ErrorKind) -> Result { + fn context(self, kind: ErrorKind, context: impl Into>) -> Result { + self.map_err(|source| Error::with_context(kind, context, source)) + } + + fn kind(self, kind: ErrorKind) -> Result { self.map_err(|source| Error::with_source(kind, source)) } } @@ -249,21 +287,26 @@ mod tests { use super::{Error, ErrorKind, Panic}; #[test] - fn errors_always_capture_backtraces() { - let client: Error = ErrorKind::InvalidMetadata.into(); - let fault: Error = ErrorKind::BackendFailure.into(); - assert!(client.backtrace().is_some()); - assert!(fault.backtrace().is_some()); - } - - #[test] - fn opaque_error_preserves_source_and_origin_trace() { + fn opaque_error_preserves_source() { let error = Error::with_source(ErrorKind::BackendFailure, io::Error::other("backend down")); let standard_error: &dyn std::error::Error = &error; assert_eq!(error.kind(), ErrorKind::BackendFailure); assert_eq!(standard_error.source().unwrap().to_string(), "backend down"); - assert!(error.backtrace().is_some()); + } + + #[test] + fn context_renders_after_kind() { + let error = Error::with_context( + ErrorKind::BackendFailure, + "reading local object", + io::Error::other("backend down"), + ); + + assert_eq!( + error.to_string(), + "backend operation failed: reading local object" + ); } #[test] diff --git a/objectstore-service/src/service.rs b/objectstore-service/src/service.rs index 1cc093e6..3df806bf 100644 --- a/objectstore-service/src/service.rs +++ b/objectstore-service/src/service.rs @@ -205,7 +205,7 @@ impl StorageService { metadata: Metadata, stream: ClientStream, ) -> Result { - metadata.validate().context(ErrorKind::InvalidMetadata)?; + metadata.validate().kind(ErrorKind::InvalidMetadata)?; let id = ObjectId::optional(context, key); let inner = Arc::clone(&self.inner); self.spawn("insert", async move { @@ -259,7 +259,7 @@ impl StorageService { id: ObjectId, metadata: Metadata, ) -> Result { - metadata.validate().context(ErrorKind::InvalidMetadata)?; + metadata.validate().kind(ErrorKind::InvalidMetadata)?; self.inner.as_multipart_upload_backend()?; // Fail before clone/spawn if unsupported let inner = self.inner.clone(); self.spawn("initiate_multipart", async move {