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.

22 changes: 13 additions & 9 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 @@ -90,15 +91,18 @@ impl ApiError {
ApiError::Auth(AuthError::NotPermitted) => StatusCode::FORBIDDEN,
ApiError::Auth(AuthError::InternalError(_)) => StatusCode::INTERNAL_SERVER_ERROR,

ApiError::Service(ServiceError::Client(_)) => StatusCode::BAD_REQUEST,
ApiError::Service(ServiceError::Metadata(_)) => StatusCode::BAD_REQUEST,
ApiError::Service(ServiceError::RangeNotSatisfiable { .. }) => {
StatusCode::RANGE_NOT_SATISFIABLE
}
ApiError::Service(ServiceError::InvalidUploadId(_)) => StatusCode::BAD_REQUEST,
ApiError::Service(ServiceError::AtCapacity) => StatusCode::TOO_MANY_REQUESTS,
ApiError::Service(ServiceError::NotImplemented) => StatusCode::NOT_IMPLEMENTED,
ApiError::Service(_) => StatusCode::INTERNAL_SERVER_ERROR,
ApiError::Service(error) => match error.kind {
ServiceErrorKind::ClientStream | ServiceErrorKind::InvalidInput => {
StatusCode::BAD_REQUEST
}
ServiceErrorKind::RangeNotSatisfiable => StatusCode::RANGE_NOT_SATISFIABLE,
ServiceErrorKind::BackendRateLimited => StatusCode::TOO_MANY_REQUESTS,
ServiceErrorKind::AtCapacity
| ServiceErrorKind::BackendTimeout
| ServiceErrorKind::BackendUnavailable => StatusCode::SERVICE_UNAVAILABLE,
ServiceErrorKind::NotImplemented => StatusCode::NOT_IMPLEMENTED,
ServiceErrorKind::Internal => StatusCode::INTERNAL_SERVER_ERROR,
},

ApiError::Internal(_) => StatusCode::INTERNAL_SERVER_ERROR,
}
Expand Down
10 changes: 8 additions & 2 deletions objectstore-server/src/endpoints/objects.rs
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ use axum::http::{HeaderMap, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::routing;
use axum::{Json, Router};
use objectstore_service::error::Error as ServiceError;
use objectstore_service::error::{Error as ServiceError, ErrorKind as ServiceErrorKind};
use objectstore_service::id::{ObjectContext, ObjectId};
use objectstore_types::metadata::Metadata;
use objectstore_types::range::ContentRange;
Expand Down Expand Up @@ -72,7 +72,13 @@ async fn object_get(
let (metadata, content_range, stream) = match result {
Ok(Some(result)) => result,
Ok(None) => return Ok(StatusCode::NOT_FOUND.into_response()),
Err(ApiError::Service(ServiceError::RangeNotSatisfiable { total })) => {
Err(ApiError::Service(e)) if e.kind == ServiceErrorKind::RangeNotSatisfiable => {
let Some(total) = e.range_total() else {
return Err(ApiError::Service(
e.kind(ServiceErrorKind::Internal)
.context("range error missing object size"),
));
};
let mut response = (
StatusCode::RANGE_NOT_SATISFIABLE,
[(
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 @@ -625,9 +625,10 @@ async fn test_bandwidth_scope_pct_limit() -> Result<()> {
}

#[tokio::test]
async fn test_batch_at_capacity_returns_429() -> Result<()> {
async fn test_batch_at_capacity_returns_503() -> Result<()> {
// With max_concurrency=0 the service has no permits available, so
// BatchExecutor::new() returns AtCapacity and the endpoint responds 429.
// BatchExecutor::new() returns AtCapacity and the endpoint responds 503:
// local load-shedding is surfaced as a temporary Service Unavailable.
let server = TestServer::with_config(Config {
service: Service {
max_concurrency: 0,
Expand Down Expand Up @@ -660,8 +661,8 @@ async fn test_batch_at_capacity_returns_429() -> Result<()> {

assert_eq!(
response.status(),
reqwest::StatusCode::TOO_MANY_REQUESTS,
"expected 429 when service has no available permits"
reqwest::StatusCode::SERVICE_UNAVAILABLE,
"expected 503 when service has no available permits"
);

Ok(())
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
38 changes: 33 additions & 5 deletions objectstore-service/docs/architecture.md
Original file line number Diff line number Diff line change
Expand Up @@ -180,11 +180,11 @@ prevent exhaustion of internal resources such as memory.

## Concurrency Limit

A concurrency limiter caps in-flight
backend operations. When all execution permits are held, new operations are
queued — adding latency instead of rejecting immediately. The queue itself is
bounded in both depth and time: operations that cannot be served within those
limits fail with [`Error::AtCapacity`](error::Error::AtCapacity).
A concurrency limiter caps in-flight backend operations. When all execution
permits are held, new operations are queued — adding latency instead of
rejecting immediately. The queue itself is bounded in both depth and time:
operations that cannot be served within those limits fail with an
[`ErrorKind::AtCapacity`](error::ErrorKind::AtCapacity) error.

The default execution limit is
[`DEFAULT_CONCURRENCY_LIMIT`](service::DEFAULT_CONCURRENCY_LIMIT). See
Expand All @@ -211,3 +211,31 @@ reservation, lazy pulling, memory bounds, and concurrency model.

More backpressure mechanisms (e.g. per-backend limits, adaptive throttling) may
be added here in the future.

# Error Model

Service and backend failures use a single [`Error`](error::Error) containing an
[`ErrorKind`](error::ErrorKind) classification and an `anyhow::Error`
diagnostic. The classification is deliberately independent of both the
concrete error type and HTTP semantics: it can be changed without altering the
diagnostic chain, while the server maps it onto a status code and
[`Error::level`](error::Error::level) maps it onto a log level.

Construction follows two paths:

- **Default (`?`)**: a foreign error converts through one of the `From` impls
into an [`ErrorKind::Internal`](error::ErrorKind::Internal) error while the
original error and its source chain become the diagnostic.
- **Override**: the [`ResultExt`](error::ResultExt) extension trait's `.kind(…)`
and `.context(…)` methods reclassify and annotate without a `map_err`.
Repeated `.context(…)` calls use anyhow semantics and retain every context
frame; `.kind(…)` changes only the independent classification.

`Error` delegates display, debug formatting, chain traversal, and downcasting
to anyhow. Backends classify transport and HTTP failures into the `Backend*`
kinds in [`check_error`](backend), so retryability is a simple check on
[`kind`](error::Error::kind). Unsatisfiable ranges are constructed with
[`Error::range_not_satisfiable`](error::Error::range_not_satisfiable) and expose
their required object size through
[`Error::range_total`](error::Error::range_total), allowing the server to emit a
valid `Content-Range` header without relying on an unchecked source convention.
86 changes: 41 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, 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::range_not_satisfiable(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,10 @@ 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.range_total();
assert_eq!(total, Some(22));
}
Ok(_) => panic!("expected RangeNotSatisfiable, got Ok"),
Err(e) => panic!("expected RangeNotSatisfiable, got {e:?}"),
}
Expand Down
8 changes: 4 additions & 4 deletions objectstore-service/src/backend/common.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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))
}
}

Expand Down
Loading
Loading