Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

28 changes: 16 additions & 12 deletions objectstore-server/src/endpoints/common.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -94,18 +95,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::TaskPanic | ServiceErrorKind::Internal => {
objectstore_log::error!(!!self, "error handling request");
StatusCode::INTERNAL_SERVER_ERROR
}
},

ApiError::Internal(_) => {
objectstore_log::error!(!!self, "internal error");
Expand Down
5 changes: 3 additions & 2 deletions objectstore-server/src/endpoints/multipart.rs
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ use bytes::Bytes;
use futures::StreamExt;
use http::HeaderValue;
use http::header;
use objectstore_service::error::Error as ServiceError;
use objectstore_service::error::{ErrorKind as ServiceErrorKind, ResultExt};
use objectstore_service::id::{ObjectContext, ObjectId};
use objectstore_service::multipart::{CompletedPart, PartNumber, UploadId};
use objectstore_types::metadata::Metadata;
Expand Down Expand Up @@ -96,7 +96,8 @@ async fn initiate_inner(
headers: HeaderMap,
) -> ApiResult<Response> {
// 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, "").kind(ServiceErrorKind::InvalidInput)?;

state
.config
Expand Down
16 changes: 12 additions & 4 deletions objectstore-server/src/endpoints/objects.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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, ResultExt,
};
use objectstore_service::id::{ObjectContext, ObjectId};
use objectstore_types::metadata::Metadata;
use objectstore_types::range::ContentRange;
Expand Down Expand Up @@ -43,7 +45,8 @@ async fn objects_post(
headers: HeaderMap,
MeteredBody(body): MeteredBody,
) -> ApiResult<Response> {
let metadata = Metadata::from_insert_headers(&headers, "").map_err(ServiceError::from)?;
let metadata =
Metadata::from_insert_headers(&headers, "").kind(ServiceErrorKind::InvalidInput)?;

state
.config
Expand Down Expand Up @@ -72,7 +75,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,
[(
Expand Down Expand Up @@ -175,7 +182,8 @@ async fn object_put(
headers: HeaderMap,
MeteredBody(body): MeteredBody,
) -> ApiResult<Response> {
let metadata = Metadata::from_insert_headers(&headers, "").map_err(ServiceError::from)?;
let metadata =
Metadata::from_insert_headers(&headers, "").kind(ServiceErrorKind::InvalidInput)?;

let ObjectId { context, key } = id;

Expand Down
9 changes: 5 additions & 4 deletions objectstore-server/tests/limits.rs
Original file line number Diff line number Diff line change
Expand Up @@ -532,9 +532,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 {
Expand Down Expand Up @@ -564,8 +565,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(())
Expand Down
1 change: 1 addition & 0 deletions objectstore-service/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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]
Expand Down
22 changes: 21 additions & 1 deletion objectstore-service/docs/architecture.md
Original file line number Diff line number Diff line change
Expand Up @@ -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`].
Expand All @@ -211,3 +211,23 @@ 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 (usually) 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.
89 changes: 44 additions & 45 deletions objectstore-service/src/backend/bigtable.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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};
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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")?,
})),
])
}
Expand Down Expand Up @@ -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) =
Expand All @@ -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")?,
);
}
}
_ => {}
Expand Down Expand Up @@ -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"))
}
}

Expand Down Expand Up @@ -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),
}
}
Expand All @@ -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),
}
}
Expand Down Expand Up @@ -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))]
Expand Down Expand Up @@ -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))]
Expand Down Expand Up @@ -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)
Expand All @@ -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`
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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))
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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:?}"),
}
Expand Down
Loading
Loading