wip

Author: lcian

Cargo.lock

@@ -7,6 +7,7 @@ use axum::http::StatusCode;
 use axum::response::{IntoResponse, Response};
 use http::HeaderValue;
 use objectstore_service::error::Error as ServiceError;
+use objectstore_service::error::ErrorKind as ServiceErrorKind;
 use serde::{Deserialize, Serialize};
 use thiserror::Error;
 
@@ -92,18 +93,21 @@ impl ApiError {
                 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(_) => {
-                objectstore_log::error!(!!self, "error handling request");
-                StatusCode::INTERNAL_SERVER_ERROR
-            }
+            ApiError::Service(error) => match error.kind() {
+                ServiceErrorKind::ClientStream | ServiceErrorKind::InvalidInput => {
+                    StatusCode::BAD_REQUEST
+                }
+                ServiceErrorKind::RangeNotSatisfiable => StatusCode::RANGE_NOT_SATISFIABLE,
+                ServiceErrorKind::BackendRateLimited => StatusCode::TOO_MANY_REQUESTS,
+                ServiceErrorKind::AtCapacity
+                | ServiceErrorKind::BackendTimeout
+                | ServiceErrorKind::BackendUnavailable => StatusCode::SERVICE_UNAVAILABLE,
+                ServiceErrorKind::NotImplemented => StatusCode::NOT_IMPLEMENTED,
+                ServiceErrorKind::Internal => {
+                    objectstore_log::error!(!!self, "error handling request");
+                    StatusCode::INTERNAL_SERVER_ERROR
+                }
+            },
 
             ApiError::Internal(_) => {
                 objectstore_log::error!(!!self, "internal error");

objectstore-server/src/endpoints/common.rs

@@ -4,7 +4,9 @@ use axum::http::{HeaderMap, StatusCode};
 use axum::response::{IntoResponse, Response};
 use axum::routing;
 use axum::{Json, Router};
-use objectstore_service::error::Error as ServiceError;
+use objectstore_service::error::{
+    Error as ServiceError, ErrorKind as ServiceErrorKind, RangeNotSatisfiableError,
+};
 use objectstore_service::id::{ObjectContext, ObjectId};
 use objectstore_types::metadata::Metadata;
 use objectstore_types::range::ContentRange;
@@ -72,7 +74,11 @@ async fn object_get(
     let (metadata, content_range, stream) = match result {
         Ok(Some(result)) => result,
         Ok(None) => return Ok(StatusCode::NOT_FOUND.into_response()),
-        Err(ApiError::Service(ServiceError::RangeNotSatisfiable { total })) => {
+        Err(ApiError::Service(ref e)) if e.kind() == ServiceErrorKind::RangeNotSatisfiable => {
+            let total = e
+                .source()
+                .and_then(|s| s.downcast_ref::<RangeNotSatisfiableError>())
+                .map_or(0, |r| r.total);
             let mut response = (
                 StatusCode::RANGE_NOT_SATISFIABLE,
                 [(

objectstore-server/src/endpoints/objects.rs

@@ -534,9 +534,10 @@ async fn test_bandwidth_scope_pct_limit() -> Result<()> {
 }
 
 #[tokio::test]
-async fn test_batch_at_capacity_returns_429() -> Result<()> {
+async fn test_batch_at_capacity_returns_503() -> Result<()> {
     // With max_concurrency=0 the service has no permits available, so
-    // BatchExecutor::new() returns AtCapacity and the endpoint responds 429.
+    // BatchExecutor::new() returns AtCapacity and the endpoint responds 503:
+    // local load-shedding is surfaced as a temporary Service Unavailable.
     let server = TestServer::with_config(Config {
         service: Service { max_concurrency: 0 },
         auth: AuthZ {
@@ -566,8 +567,8 @@ async fn test_batch_at_capacity_returns_429() -> Result<()> {
 
     assert_eq!(
         response.status(),
-        reqwest::StatusCode::TOO_MANY_REQUESTS,
-        "expected 429 when service has no available permits"
+        reqwest::StatusCode::SERVICE_UNAVAILABLE,
+        "expected 503 when service has no available permits"
     );
 
     Ok(())

objectstore-server/tests/limits.rs

@@ -34,6 +34,7 @@ tokio = { workspace = true }
 tokio-util = { workspace = true, features = ["io", "rt"] }
 tonic = { workspace = true }
 tracing = { workspace = true }
+url = { workspace = true }
 uuid = { workspace = true, features = ["v7"] }
 
 [dev-dependencies]

objectstore-service/Cargo.toml

@@ -185,7 +185,7 @@ A semaphore caps the total number of in-flight backend operations across all
 callers. A permit is acquired before each operation is spawned and held until
 the task completes — including on panic — so the limit counts *running*
 operations, not queued ones. When no permits are available, the operation fails
-with [`Error::AtCapacity`](error::Error::AtCapacity).
+with an [`ErrorKind::AtCapacity`](error::ErrorKind::AtCapacity) error.
 
 The default limit is [`DEFAULT_CONCURRENCY_LIMIT`](service::DEFAULT_CONCURRENCY_LIMIT). Callers can override it via
 [`StorageService::with_concurrency_limit`].
@@ -211,3 +211,30 @@ reservation, lazy pulling, memory bounds, and concurrency model.
 
 More backpressure mechanisms (e.g. per-backend limits, adaptive throttling) may
 be added here in the future.
+
+# Error Model
+
+Service and backend failures use a single [`Error`](error::Error) type shaped
+like `anyhow::Error`: it carries an [`ErrorKind`](error::ErrorKind) (the
+*classification* of the failure), an optional boxed `source` error, and an
+optional `context` message. The [`ErrorKind`](error::ErrorKind) is deliberately
+independent of HTTP semantics — the server maps each kind onto a status code
+(see the server architecture docs), and [`Error::level`](error::Error::level)
+maps each kind onto a log level.
+
+Construction follows two paths:
+
+- **Default (`?`)**: a foreign error converts through one of the `From` impls
+  into an [`ErrorKind::Internal`](error::ErrorKind::Internal) error that keeps
+  the original as its `source`.
+- **Override**: the [`ResultExt`](error::ResultExt) extension trait's `.kind(…)`
+  and `.context(…)` methods reclassify and annotate without a `map_err`.
+  Chaining `.kind(k).context(c)` boxes the original error as `source` exactly
+  once — an already-converted `Error` passes through unchanged.
+
+Backends classify transport and HTTP failures into the `Backend*` kinds in
+[`check_error`](backend), so retryability is a simple check on
+[`kind`](error::Error::kind). The object's total size for a
+[`RangeNotSatisfiable`](error::ErrorKind::RangeNotSatisfiable) response is
+carried as a downcastable [`RangeNotSatisfiableError`](error::RangeNotSatisfiableError)
+`source` so the server can emit the `Content-Range` header.

objectstore-service/docs/architecture.md

@@ -47,7 +47,7 @@ use crate::backend::common::{
     Backend, DeleteResponse, GetResponse, HighVolumeBackend, MetadataResponse, PutResponse,
     TieredGet, TieredMetadata, TieredWrite, Tombstone,
 };
-use crate::error::{Error, Result};
+use crate::error::{Error, ErrorKind, RangeNotSatisfiableError, Result, ResultExt};
 use crate::gcp_auth::PrefetchingTokenProvider;
 use crate::id::ObjectId;
 use crate::stream::{ChunkedBytes, ClientStream};
@@ -466,8 +466,7 @@ fn object_mutations(mut metadata: Metadata, payload: Vec<u8>) -> Result<[v2::Mut
     // Record the payload size in the metadata before persisting it.
     metadata.size = Some(payload.len());
 
-    let metadata_bytes = serde_json::to_vec(&metadata)
-        .map_err(|cause| Error::serde("failed to serialize metadata", cause))?;
+    let metadata_bytes = serde_json::to_vec(&metadata).context("failed to serialize metadata")?;
 
     Ok([
         // NB: We explicitly delete the row to clear metadata on overwrite.
@@ -528,8 +527,7 @@ fn tombstone_mutations(tombstone: &Tombstone, now: SystemTime) -> Result<[v2::Mu
             family_name: family.to_owned(),
             column_qualifier: COLUMN_TOMBSTONE_META.to_owned(),
             timestamp_micros,
-            value: serde_json::to_vec(&tombstone_meta)
-                .map_err(|cause| Error::serde("failed to serialize tombstone", cause))?,
+            value: serde_json::to_vec(&tombstone_meta).context("failed to serialize tombstone")?,
         })),
     ])
 }
@@ -601,10 +599,10 @@ impl RowData {
                     payload = cell.value;
                 }
                 COLUMN_TOMBSTONE_META => {
-                    tombstone_meta_opt =
-                        Some(serde_json::from_slice(&cell.value).map_err(|cause| {
-                            Error::serde("failed to deserialize tombstone meta", cause)
-                        })?);
+                    tombstone_meta_opt = Some(
+                        serde_json::from_slice(&cell.value)
+                            .context("failed to deserialize tombstone meta")?,
+                    );
                 }
                 COLUMN_METADATA => {
                     if let Ok(legacy_meta) =
@@ -617,10 +615,10 @@ impl RowData {
                             expiration_policy: legacy_meta.expiration_policy,
                         });
                     } else {
-                        metadata_opt =
-                            Some(serde_json::from_slice(&cell.value).map_err(|cause| {
-                                Error::serde("failed to deserialize metadata", cause)
-                            })?);
+                        metadata_opt = Some(
+                            serde_json::from_slice(&cell.value)
+                                .context("failed to deserialize metadata")?,
+                        );
                     }
                 }
                 _ => {}
@@ -683,10 +681,11 @@ fn parse_redirect_target(redirect_path: &[u8], tombstone_id: &ObjectId) -> Resul
         objectstore_metrics::count!("bigtable.empty_redirect_read");
         Ok(tombstone_id.clone())
     } else {
-        let redirect_str = std::str::from_utf8(redirect_path)
-            .map_err(|_| Error::generic("invalid UTF-8 in redirect path"))?;
+        let redirect_str = std::str::from_utf8(redirect_path).map_err(|_| {
+            Error::new(ErrorKind::Internal).context("invalid UTF-8 in redirect path")
+        })?;
         ObjectId::from_storage_path(redirect_str)
-            .ok_or_else(|| Error::generic("corrupt redirect path"))
+            .ok_or_else(|| Error::new(ErrorKind::Internal).context("corrupt redirect path"))
     }
 }
 
@@ -930,7 +929,9 @@ impl Backend for BigTableBackend {
             TieredGet::Object(metadata, content_range, payload) => {
                 Ok(Some((metadata, content_range, payload)))
             }
-            TieredGet::Tombstone(_) => Err(Error::UnexpectedTombstone),
+            TieredGet::Tombstone(_) => {
+                Err(Error::new(ErrorKind::Internal).context("unexpected tombstone"))
+            }
             TieredGet::NotFound => Ok(None),
         }
     }
@@ -939,7 +940,9 @@ impl Backend for BigTableBackend {
     async fn get_metadata(&self, id: &ObjectId) -> Result<MetadataResponse> {
         match self.get_tiered_metadata(id).await? {
             TieredMetadata::Object(metadata) => Ok(Some(metadata)),
-            TieredMetadata::Tombstone(_) => Err(Error::UnexpectedTombstone),
+            TieredMetadata::Tombstone(_) => {
+                Err(Error::new(ErrorKind::Internal).context("unexpected tombstone"))
+            }
             TieredMetadata::NotFound => Ok(None),
         }
     }
@@ -1002,7 +1005,7 @@ impl HighVolumeBackend for BigTableBackend {
             }
         }
 
-        Err(Error::generic("BigTable: race loop in put_non_tombstone"))
+        Err(Error::new(ErrorKind::Internal).context("BigTable: race loop in put_non_tombstone"))
     }
 
     #[tracing::instrument(level = "debug", skip(self))]
@@ -1110,9 +1113,7 @@ impl HighVolumeBackend for BigTableBackend {
             }
         }
 
-        Err(Error::generic(
-            "BigTable: race loop in delete_non_tombstone",
-        ))
+        Err(Error::new(ErrorKind::Internal).context("BigTable: race loop in delete_non_tombstone"))
     }
 
     #[tracing::instrument(level = "debug", skip(self, write))]
@@ -1151,13 +1152,12 @@ impl HighVolumeBackend for BigTableBackend {
 /// required by BigTable, the resulting timestamp has millisecond precision, with the last digits at
 /// 0.
 fn ttl_to_micros(ttl: Duration, from: SystemTime) -> Result<i64> {
-    let deadline = from.checked_add(ttl).ok_or_else(|| Error::Generic {
-        context: format!(
+    let deadline = from.checked_add(ttl).ok_or_else(|| {
+        Error::new(ErrorKind::Internal).context(format!(
             "TTL duration overflow: {} plus {}s cannot be represented as SystemTime",
             humantime::format_rfc3339_seconds(from),
             ttl.as_secs()
-        ),
-        cause: None,
+        ))
     })?;
 
     system_time_to_micros(deadline)
@@ -1170,19 +1170,15 @@ fn ttl_to_micros(ttl: Duration, from: SystemTime) -> Result<i64> {
 fn system_time_to_micros(deadline: SystemTime) -> Result<i64> {
     let millis = deadline
         .duration_since(SystemTime::UNIX_EPOCH)
-        .map_err(|e| Error::Generic {
-            context: format!(
-                "unable to get duration since UNIX_EPOCH for SystemTime {}",
-                humantime::format_rfc3339_seconds(deadline)
-            ),
-            cause: Some(Box::new(e)),
-        })?
+        .context(format!(
+            "unable to get duration since UNIX_EPOCH for SystemTime {}",
+            humantime::format_rfc3339_seconds(deadline)
+        ))?
         .as_millis();
 
-    (millis * 1000).try_into().map_err(|e| Error::Generic {
-        context: format!("failed to convert {millis}ms to i64 microseconds"),
-        cause: Some(Box::new(e)),
-    })
+    (millis * 1000)
+        .try_into()
+        .context(format!("failed to convert {millis}ms to i64 microseconds"))
 }
 
 /// Converts a wall-clock time to Bigtable's microsecond timestamp, saturating at `i64::MAX`
@@ -1233,10 +1229,7 @@ where
             Ok(res) => return Ok(res),
             Err(e) if retry_count >= REQUEST_RETRY_COUNT || !is_retryable(&e) => {
                 objectstore_metrics::count!("bigtable.failures", action = context);
-                return Err(Error::Generic {
-                    context: format!("Bigtable: `{context}` failed"),
-                    cause: Some(Box::new(e)),
-                });
+                return Err(Error::from(e).context(format!("Bigtable: `{context}` failed")));
             }
             Err(e) => {
                 retry_count += 1;
@@ -1290,7 +1283,7 @@ fn apply_range(payload: Bytes, range: Option<ByteRange>) -> Result<(Option<Conte
     let total = payload.len() as u64;
     let content_range = byte_range
         .resolve(total)
-        .ok_or(Error::RangeNotSatisfiable { total })?;
+        .ok_or_else(|| Error::from(RangeNotSatisfiableError { total }))?;
 
     let sliced = payload.slice(content_range.start as usize..content_range.end as usize + 1);
     Ok((Some(content_range), sliced))
@@ -1865,11 +1858,11 @@ mod tests {
         // Legacy reads must error rather than leak tombstone data.
         assert!(matches!(
             backend.get_object(&hv_id, None).await,
-            Err(Error::UnexpectedTombstone)
+            Err(e) if e.kind() == ErrorKind::Internal
         ));
         assert!(matches!(
             backend.get_metadata(&hv_id).await,
-            Err(Error::UnexpectedTombstone)
+            Err(e) if e.kind() == ErrorKind::Internal
         ));
 
         // Idempotent retry: retry with the same target succeeds
@@ -2345,7 +2338,13 @@ mod tests {
         let id = put_range_test_object(&backend).await?;
 
         match backend.get_object(&id, Some(ByteRange::From(100))).await {
-            Err(Error::RangeNotSatisfiable { total }) => assert_eq!(total, 22),
+            Err(err) if err.kind() == ErrorKind::RangeNotSatisfiable => {
+                let total = err
+                    .source()
+                    .and_then(|s| s.downcast_ref::<crate::error::RangeNotSatisfiableError>())
+                    .map(|e| e.total);
+                assert_eq!(total, Some(22));
+            }
             Ok(_) => panic!("expected RangeNotSatisfiable, got Ok"),
             Err(e) => panic!("expected RangeNotSatisfiable, got {e:?}"),
         }

objectstore-service/src/backend/bigtable.rs

@@ -7,7 +7,7 @@ use objectstore_types::range::{ByteRange, ContentRange};
 
 use bytes::Bytes;
 
-use crate::error::{Error, Result};
+use crate::error::{Error, ErrorKind, Result};
 use crate::id::ObjectId;
 use crate::multipart::{
     AbortMultipartResponse, CompleteMultipartResponse, CompletedPart, InitiateMultipartResponse,
@@ -67,10 +67,10 @@ pub trait Backend: fmt::Debug + Send + Sync + 'static {
 
     /// Borrows this backend as a [`MultipartUploadBackend`] if supported.
     ///
-    /// The default returns [`Error::NotImplemented`]. Backends that implement
-    /// [`MultipartUploadBackend`] should override this to return `Ok(self)`.
+    /// The default returns an [`ErrorKind::NotImplemented`] error. Backends that
+    /// implement [`MultipartUploadBackend`] should override this to return `Ok(self)`.
     fn as_multipart_upload_backend(&self) -> Result<&dyn MultipartUploadBackend> {
-        Err(Error::NotImplemented)
+        Err(Error::new(ErrorKind::NotImplemented))
     }
 }