//! Native S3 client implementation of [`Storage`](icechunk_storage::Storage).
// Re-export AWS SDK types needed by consumers (e.g., icechunk's virtual_chunks)
pub use aws_sdk_s3;
use std::{
borrow::Cow, collections::HashMap, fmt, future::ready, ops::Range, pin::Pin,
sync::Arc, time::Duration,
};
use async_trait::async_trait;
use aws_config::{
BehaviorVersion, meta::region::RegionProviderChain, retry::RetryConfig,
timeout::TimeoutConfig,
};
use aws_credential_types::provider::error::CredentialsError;
use aws_sdk_s3::{
Client,
config::{
Builder, ConfigBag, IdentityCache, Intercept, ProvideCredentials, Region,
RuntimeComponents, StalledStreamProtectionConfig,
interceptors::{
BeforeDeserializationInterceptorContextMut,
BeforeTransmitInterceptorContextMut,
},
},
error::{BoxError, SdkError},
operation::{copy_object::CopyObjectError, put_object::PutObjectError},
primitives::ByteStream,
types::{CompletedMultipartUpload, CompletedPart, Delete, Object, ObjectIdentifier},
};
use aws_smithy_runtime::client::retries::classifiers::HttpStatusCodeClassifier;
use aws_smithy_types_convert::{
date_time::DateTimeExt as _, stream::PaginationStreamExt as _,
};
use bytes::Bytes;
use chrono::{DateTime, Utc};
use futures::{
Stream, StreamExt as _, TryStreamExt as _,
stream::{self, BoxStream, FuturesOrdered},
};
pub use icechunk_storage::s3_config::{
S3ChecksumAlgorithm, S3Credentials, S3CredentialsFetcher, S3Options,
S3StaticCredentials,
};
use icechunk_storage::{
DeleteObjectsResult, GetModifiedResult, ListInfo, RepositoryCreation, Settings,
Storage, StorageContext, StorageError, StorageErrorKind, StorageInfo, StorageResult,
VersionInfo, VersionedUpdateResult, obj_not_found_res, obj_store_error,
obj_store_error_res, other_error,
readback::{
ReadbackOutcome, WRITE_ID_METADATA_KEY, resolve_lost_response,
resolve_precondition, write_id_for,
},
sealed, split_in_multiple_equal_requests, strip_quotes, throttled_error,
};
use icechunk_types::ICResultExt as _;
use serde::{Deserialize, Serialize};
use tokio::sync::OnceCell;
use tokio_util::io::StreamReader;
use tracing::{error, instrument, trace, warn};
use typed_path::Utf8UnixPath;
use uuid::Uuid;
/// How object keys are laid out inside the bucket for a given repository.
///
/// Native-S3 repositories written before the fix for
/// <https://github.com/earth-mover/icechunk/issues/2239> stored every object
/// under a leading slash when the prefix was empty (`"/chunks/..."`, etc.).
/// We keep being able to read and write those repositories via [`KeyLayout::LegacyRoot`],
/// while all new repositories use the clean [`KeyLayout::Standard`] layout.
///
/// A repository has exactly one layout, decided once and never mixed.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum KeyLayout {
/// `prefix == ""` -> `"chunks/x"`; `prefix == "p"` -> `"p/chunks/x"`.
Standard,
/// Only used with an empty prefix: reproduces the historical `"/chunks/x"`.
LegacyRoot,
}
/// Well-known object names probed by [`S3Storage::probe_layout`] to auto-detect
/// the key layout of a pre-existing repository. These are fixed keys that exist
/// in every repository.
///
/// It's awful to have to have this tacit dependency here, this is icechunk-format stuff.
/// But... reality hits you hard. I don't want an explicit dependency between the crates
/// but there are some tests in `icechunk/src/refs.rs` that verify these strings are "right".
pub const DEFAULT_LAYOUT_ANCHORS: &[&str] = &["repo", "refs/branch.main/ref.json"];
/// Serializes the resolved [`KeyLayout`] as `Option<KeyLayout>`.
mod layout_cell_serde {
use super::{KeyLayout, OnceCell};
use serde::{Deserialize as _, Deserializer, Serialize as _, Serializer};
pub(super) fn serialize<S: Serializer>(
cell: &OnceCell<KeyLayout>,
serializer: S,
) -> Result<S::Ok, S::Error> {
cell.get().copied().serialize(serializer)
}
pub(super) fn deserialize<'de, D: Deserializer<'de>>(
deserializer: D,
) -> Result<OnceCell<KeyLayout>, D::Error> {
let resolved = Option::<KeyLayout>::deserialize(deserializer)?;
Ok(OnceCell::new_with(resolved))
}
}
#[derive(Debug, Serialize, Deserialize)]
pub struct S3Storage {
// config and credentials are stored so we are able to serialize and deserialize the struct
config: S3Options,
credentials: S3Credentials,
bucket: String,
prefix: String,
can_write: bool,
extra_read_headers: Vec<(String, String)>,
extra_write_headers: Vec<(String, String)>,
#[serde(default, with = "layout_cell_serde")]
key_layout: OnceCell<KeyLayout>,
/// Test/internal escape hatch permitting repository creation at an empty
/// prefix.
#[serde(skip)]
allow_empty_prefix_creation: bool,
#[serde(skip)]
/// We need to use `OnceCell` to allow async initialization, because serde
/// does not support async function calls from deserialization. This gives
/// us a way to lazily initialize the client.
client: OnceCell<Arc<Client>>,
}
impl fmt::Display for S3Storage {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
f,
"S3Storage(bucket={}, prefix={}, config={})",
self.bucket, self.prefix, self.config,
)
}
}
#[derive(Debug)]
struct ExtraHeadersInterceptor {
extra_read_headers: Vec<(String, String)>,
extra_write_headers: Vec<(String, String)>,
}
impl Intercept for ExtraHeadersInterceptor {
fn name(&self) -> &'static str {
"ExtraHeaders"
}
fn modify_before_retry_loop(
&self,
context: &mut BeforeTransmitInterceptorContextMut<'_>,
_runtime_components: &RuntimeComponents,
_cfg: &mut ConfigBag,
) -> Result<(), BoxError> {
let request = context.request_mut();
let headers = match request.method() {
"GET" | "HEAD" | "OPTIONS" | "TRACE" => &self.extra_read_headers,
_ => &self.extra_write_headers,
};
for (k, v) in headers.iter() {
request.headers_mut().try_insert(k.clone(), v.clone()).map_err(
|e| -> BoxError {
format!("invalid extra HTTP header {k:?}: {e}").into()
},
)?;
}
Ok(())
}
}
/// Appends the icechunk attribution fragment to the `user-agent` the SDK
/// built. It runs after the SDK's client-level `UserAgentInterceptor`,
/// so the fragment lands after the SDK tokens.
#[derive(Debug)]
struct AttributionInterceptor {
fragment: String,
}
impl Intercept for AttributionInterceptor {
fn name(&self) -> &'static str {
"IcechunkAttribution"
}
fn modify_before_signing(
&self,
context: &mut BeforeTransmitInterceptorContextMut<'_>,
_runtime_components: &RuntimeComponents,
_cfg: &mut ConfigBag,
) -> Result<(), BoxError> {
let headers = context.request_mut().headers_mut();
let ua = match headers.get("user-agent") {
Some(current) => format!("{current} {}", self.fragment),
None => self.fragment.clone(),
};
headers.try_insert("user-agent", ua)?;
Ok(())
}
}
/// Per-request config that appends `fragment` to the SDK's `user-agent`.
pub fn attribution_override_for(fragment: String) -> Builder {
Builder::new().interceptor(AttributionInterceptor { fragment })
}
fn attribution_override(ctx: &StorageContext<'_>) -> Builder {
attribution_override_for(ctx.attribution.user_agent_fragment())
}
/// Strips `x-amz-checksum-*` headers from HTTP 304 (Not Modified) responses.
///
/// R2 includes checksum headers (e.g. crc32, crc64nvme) on 304 responses, but
/// because 304 responses have no body the AWS SDK's checksum validation fails
/// with a mismatch, triggering transient-error retries with exponential backoff.
#[derive(Debug)]
struct StripChecksumOn304Interceptor;
impl Intercept for StripChecksumOn304Interceptor {
fn name(&self) -> &'static str {
"StripChecksumOn304"
}
fn modify_before_deserialization(
&self,
context: &mut BeforeDeserializationInterceptorContextMut<'_>,
_runtime_components: &RuntimeComponents,
_cfg: &mut ConfigBag,
) -> Result<(), BoxError> {
let response = context.response_mut();
if response.status().as_u16() == 304 {
let to_remove: Vec<_> = response
.headers()
.iter()
.filter(|(name, _)| name.starts_with("x-amz-checksum-"))
.map(|(name, _)| name.to_owned())
.collect();
for name in to_remove {
response.headers_mut().remove(&name);
}
}
Ok(())
}
}
/// Append the `/` that separates an endpoint path from the bucket name.
///
/// The SDK omits it, so `https://s3.hf.co/<ns>` addresses `/<ns><bucket>/<key>`.
fn endpoint_with_bucket_separator(endpoint_url: &str) -> Cow<'_, str> {
// The SDK appends the bucket name to any path past the root.
let has_path =
endpoint_url.split_once("://").is_some_and(|(_, rest)| rest.contains('/'));
if has_path && !endpoint_url.ends_with('/') {
Cow::Owned(format!("{endpoint_url}/"))
} else {
Cow::Borrowed(endpoint_url)
}
}
#[instrument(skip(credentials))]
pub async fn mk_client(
config: &S3Options,
credentials: S3Credentials,
extra_read_headers: Vec<(String, String)>,
extra_write_headers: Vec<(String, String)>,
settings: &Settings,
) -> Client {
let region = config
.region
.as_ref()
.map(|r| RegionProviderChain::first_try(Some(Region::new(r.clone()))))
.unwrap_or_else(RegionProviderChain::default_provider);
let endpoint = config.endpoint_url.as_deref().map(endpoint_with_bucket_separator);
let region = if endpoint.is_some() {
// GH793, the S3 SDK requires a region even though it may not make sense
// for S3-compatible object stores like Tigris or Ceph.
// So we set a fake region, using the `endpoint_url` as a sign that
// we are not talking to real S3
region.or_else(Region::new("region-was-not-set"))
} else {
region
};
let mut aws_config =
aws_config::defaults(BehaviorVersion::v2026_01_12()).region(region);
if let Some(endpoint) = endpoint {
aws_config = aws_config.endpoint_url(endpoint);
}
let stalled_stream = if config.network_stream_timeout_seconds == Some(0) {
StalledStreamProtectionConfig::disabled()
} else {
StalledStreamProtectionConfig::enabled()
.grace_period(Duration::from_secs(
config.network_stream_timeout_seconds.unwrap_or(10) as u64,
))
.build()
};
aws_config = aws_config.stalled_stream_protection(stalled_stream);
match credentials {
S3Credentials::FromEnv => {}
S3Credentials::Anonymous => aws_config = aws_config.no_credentials(),
S3Credentials::Static(credentials) => {
aws_config =
aws_config.credentials_provider(aws_credential_types::Credentials::new(
credentials.access_key_id,
credentials.secret_access_key,
credentials.session_token,
credentials.expires_after.map(|e| e.into()),
"user",
));
}
S3Credentials::Refreshable(fetcher) => {
aws_config =
aws_config.credentials_provider(ProvideRefreshableCredentials(fetcher));
}
}
let retry_config = RetryConfig::standard()
.with_max_attempts(settings.retries().max_tries().get() as u32)
.with_initial_backoff(Duration::from_millis(
settings.retries().initial_backoff_ms() as u64,
))
.with_max_backoff(Duration::from_millis(
settings.retries().max_backoff_ms() as u64
));
if let Some(timeouts) = settings.timeouts() {
let mut timeout_builder = TimeoutConfig::builder();
if let Some(ms) = timeouts.connect_timeout_ms {
timeout_builder =
timeout_builder.connect_timeout(Duration::from_millis(ms as u64));
}
if let Some(ms) = timeouts.read_timeout_ms {
timeout_builder =
timeout_builder.read_timeout(Duration::from_millis(ms as u64));
}
if let Some(ms) = timeouts.operation_timeout_ms {
timeout_builder =
timeout_builder.operation_timeout(Duration::from_millis(ms as u64));
}
if let Some(ms) = timeouts.operation_attempt_timeout_ms {
timeout_builder = timeout_builder
.operation_attempt_timeout(Duration::from_millis(ms as u64));
}
aws_config = aws_config.timeout_config(timeout_builder.build());
}
let mut s3_builder = Builder::from(&aws_config.load().await)
.force_path_style(config.force_path_style)
.retry_config(retry_config);
// credentials may take a while to refresh, defaults are too strict
let id_cache = IdentityCache::lazy()
.load_timeout(Duration::from_secs(120))
.buffer_time(Duration::from_secs(120))
.build();
s3_builder = s3_builder.identity_cache(id_cache);
// Add retry classifier for HTTP 408 (Request Timeout) and 429 (Too Many Requests).
// The default HttpStatusCodeClassifier only retries on 500, 502, 503, 504
// Note R2 sends 429 for "slowdown" while S3 sends 503.
// - R2: https://developers.cloudflare.com/r2/api/error-codes/
// - S3: https://docs.aws.amazon.com/AmazonS3/latest/API/ErrorResponses.html
// Tigris can occasionally respond with 499: "Client Closed Request"
static RETRY_CODES: &[u16] = &[408, 429, 499];
// This confusingly named `retry_classifier` method ends up calling
// `push_retry_classifier` after wrapping our custom classifier in `SharedRetryClassifier`.
// Ultimately, this is a push on to a `Vec<SharedRetryClassifier>`, and is thus additive
// to the existing default retry configuration.
// https://github.com/smithy-lang/smithy-rs/blob/cfcc39cf4b5bea665bba684b64bfca2b89e4bc73/rust-runtime/aws-smithy-runtime-api/src/client/runtime_components.rs#L755
// https://github.com/smithy-lang/smithy-rs/blob/cfcc39cf4b5bea665bba684b64bfca2b89e4bc73/rust-runtime/aws-smithy-runtime-api/src/client/runtime_components.rs#L370
s3_builder = s3_builder
.retry_classifier(HttpStatusCodeClassifier::new_from_codes(RETRY_CODES));
if !extra_read_headers.is_empty() || !extra_write_headers.is_empty() {
s3_builder = s3_builder.interceptor(ExtraHeadersInterceptor {
extra_read_headers,
extra_write_headers,
});
}
// R2 and Scaleway include x-amz-checksum-* headers on 304 Not Modified
// responses, but because 304 has no body the SDK's checksum validation
// fails and triggers expensive transient-error retries.
if config.endpoint_url.as_ref().is_some_and(|url| {
url.contains(".r2.cloudflarestorage.com") || url.contains(".scw.cloud")
}) {
s3_builder = s3_builder.interceptor(StripChecksumOn304Interceptor);
}
let config = s3_builder.build();
Client::from_conf(config)
}
fn to_sdk_checksum(a: S3ChecksumAlgorithm) -> aws_sdk_s3::types::ChecksumAlgorithm {
use aws_sdk_s3::types::ChecksumAlgorithm as Sdk;
match a {
S3ChecksumAlgorithm::Crc32 => Sdk::Crc32,
S3ChecksumAlgorithm::Crc32c => Sdk::Crc32C,
S3ChecksumAlgorithm::Crc64Nvme => Sdk::Crc64Nvme,
S3ChecksumAlgorithm::Sha1 => Sdk::Sha1,
S3ChecksumAlgorithm::Sha256 => Sdk::Sha256,
}
}
fn stream2stream(
s: ByteStream,
) -> Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>> {
let res = stream::try_unfold(s, move |mut stream| async move {
let next = stream.try_next().await?;
Ok(next.map(|bytes| (bytes, stream)))
});
Box::pin(res)
}
/// `Uuid::new_v4` in production; tests force a value via [`test_util`].
fn next_write_id() -> String {
#[cfg(feature = "test-util")]
if let Some(forced) = test_util::take_forced_write_id() {
return forced;
}
Uuid::new_v4().to_string()
}
/// Test-only write-id injection for the #2099 lost-response tests. Feature-gated
/// `pub` (not `#[cfg(test)]`) because the consumers are in the `icechunk` crate;
/// the forced id is thread-local and consumed by a single PUT.
///
/// This approach avoids exposing a new argument to pass an explicit write-id
/// where needed, and it should be an implementation detail anyway.
#[cfg(feature = "test-util")]
pub mod test_util {
use std::cell::RefCell;
thread_local! {
static FORCED_WRITE_ID: RefCell<Option<String>> = const { RefCell::new(None) };
}
/// Make the next conditional PUT on this thread stamp `id`.
pub fn force_next_write_id(id: impl Into<String>) {
FORCED_WRITE_ID.with(|c| *c.borrow_mut() = Some(id.into()));
}
pub(crate) fn take_forced_write_id() -> Option<String> {
FORCED_WRITE_ID.with(|c| c.borrow_mut().take())
}
}
/// Conditional header a PUT carries; shared by single-PUT and multipart.
enum Conditional<'a> {
IfNoneMatch,
IfMatch(&'a str),
}
fn conditional_for<'a>(
previous_version: Option<&'a VersionInfo>,
settings: &Settings,
) -> Option<Conditional<'a>> {
let pv = previous_version?;
match (
pv.etag(),
settings.unsafe_use_conditional_create(),
settings.unsafe_use_conditional_update(),
) {
(None, true, _) => Some(Conditional::IfNoneMatch),
(Some(etag), _, true) => Some(Conditional::IfMatch(etag)),
(_, _, _) => None,
}
}
/// Error codes meaning the conditional header tripped. `PreconditionFailed`
/// is minio, `ConditionalRequestConflict` is S3, `ConcurrentModification` is
/// Ceph Object Gateway.
fn is_precondition_code(code: &str) -> bool {
matches!(
code,
"PreconditionFailed" | "ConditionalRequestConflict" | "ConcurrentModification"
)
}
/// Identify a retryable error: any 5xx, HTTP 429, or an S3 code that means "try again"
fn is_transient_s3_error(status: Option<u16>, code: &str) -> bool {
matches!(status, Some(s) if s >= 500 || s == 429)
|| matches!(
code,
"InternalError"
| "ServiceUnavailable"
| "SlowDown"
| "RequestLimitExceeded"
| "Throttling"
| "ThrottlingException"
)
}
/// The code to report when a partial `DeleteObjects` response should be
/// retried whole: `Some` only if every per-key error is transient.
fn transient_per_key_code(errors: &[aws_sdk_s3::types::Error]) -> Option<&str> {
let first = errors.first()?.code()?;
errors
.iter()
.all(|e| e.code().is_some_and(|c| is_transient_s3_error(None, c)))
.then_some(first)
}
impl S3Storage {
/// Build an [`S3Storage`].
///
/// `extra_read_headers`/`extra_write_headers` are extra HTTP headers injected
/// by an SDK interceptor and split by HTTP method: `GET`/`HEAD`/`OPTIONS`/`TRACE`
/// carry the read headers, everything else the write headers.
///
/// `legacy_rooted_keys` declares the bucket's key layout:
/// - `None` — unknown: the layout is auto-detected by probing storage on first
/// use. The right choice when opening a repository whose layout you don't know.
/// - `Some(true)` — force the legacy leading-slash layout used before the fix
/// for <https://github.com/earth-mover/icechunk/issues/2239>. Only valid with
/// an empty prefix; errors otherwise.
/// - `Some(false)` — force the standard layout, skipping the probe.
#[expect(clippy::too_many_arguments)]
pub fn new(
config: S3Options,
bucket: String,
prefix: Option<String>,
credentials: S3Credentials,
can_write: bool,
extra_read_headers: Vec<(String, String)>,
extra_write_headers: Vec<(String, String)>,
legacy_rooted_keys: Option<bool>,
) -> Result<S3Storage, StorageError> {
let client = OnceCell::new();
let prefix = prefix.unwrap_or_default();
let prefix = prefix.strip_suffix("/").unwrap_or(prefix.as_str()).to_string();
// A known layout pre-seeds the cell so `probe_layout` is never reached;
// `None` leaves it empty to be detected lazily on first use.
let key_layout = match legacy_rooted_keys {
Some(true) => {
if !prefix.is_empty() {
return Err(other_error(
"legacy_rooted_keys is only valid with an empty prefix",
));
}
OnceCell::new_with(Some(KeyLayout::LegacyRoot))
}
Some(false) => OnceCell::new_with(Some(KeyLayout::Standard)),
None => OnceCell::new(),
};
Ok(S3Storage {
client,
config,
bucket,
prefix,
credentials,
can_write,
extra_read_headers,
extra_write_headers,
key_layout,
allow_empty_prefix_creation: false,
})
}
/// Test/internal escape hatch: permit creating a new repository at an empty
/// prefix (the bucket root), which [`Storage::can_create_repository`] would
/// otherwise refuse.
pub fn unsafe_allow_empty_prefix_creation(mut self) -> Self {
self.allow_empty_prefix_creation = true;
self
}
/// Get the client, initializing it if it hasn't been initialized yet. This is necessary because the
/// client is not serializeable and must be initialized after deserialization. Under normal construction
/// the original client is returned immediately.
#[instrument(skip_all)]
pub async fn get_client(&self, settings: &Settings) -> &Arc<Client> {
self.client
.get_or_init(|| async {
Arc::new(
mk_client(
&self.config,
self.credentials.clone(),
self.extra_read_headers.clone(),
self.extra_write_headers.clone(),
settings,
)
.await,
)
})
.await
}
/// Build the object key for a repository-relative path under the given layout.
fn key_for(&self, layout: KeyLayout, relpath: &str) -> String {
match layout {
// Only reachable with an empty prefix (enforced at construction and
// by `probe_layout`), so the prefix is intentionally ignored here.
KeyLayout::LegacyRoot => format!("/{relpath}"),
KeyLayout::Standard if self.prefix.is_empty() => relpath.to_string(),
KeyLayout::Standard => format!("{}/{}", self.prefix, relpath),
}
}
/// Key prefix to list. Strips one leading `/` from the relpath so it
/// doesn't double against the join (S3 matches `//` literally).
/// An interior `//` is left intact.
fn list_prefix(&self, layout: KeyLayout, relpath: &str) -> String {
self.key_for(layout, relpath.strip_prefix('/').unwrap_or(relpath))
}
/// Lists the keys that start with `key_prefix`. Ids are relative to `id_root`.
async fn list_keys(
&self,
ctx: &StorageContext<'_>,
key_prefix: String,
id_root: String,
) -> BoxStream<'static, StorageResult<ListInfo<String>>> {
let mut req = self
.get_client(ctx.settings)
.await
.list_objects_v2()
.bucket(self.bucket.clone())
.prefix(key_prefix);
if self.config.requester_pays {
req = req.request_payer(aws_sdk_s3::types::RequestPayer::Requester);
}
// FIXME: The SDK paginator drops per-request config, so listings carry no
// attribution fragment.
req.into_paginator()
.send()
.into_stream_03x()
.map_err(obj_store_error)
.try_filter_map(|page| {
let contents = page.contents.map(|cont| stream::iter(cont).map(Ok));
ready(Ok(contents))
})
.try_flatten()
.and_then(move |object| ready(object_to_list_info(id_root.as_str(), &object)))
.boxed()
}
/// Resolve this repository's [`KeyLayout`], probing storage at most once.
async fn layout(&self, ctx: &StorageContext<'_>) -> StorageResult<KeyLayout> {
self.key_layout.get_or_try_init(|| self.probe_layout(ctx)).await.copied()
}
/// Detect the key layout of the repository.
///
/// Only ever issues requests for empty-prefix repositories — a non-empty
/// prefix can never produce a leading slash, so the layout is unambiguously
/// [`KeyLayout::Standard`]. The probe HEADs a few fixed anchor files under
/// both layouts.
async fn probe_layout(&self, ctx: &StorageContext<'_>) -> StorageResult<KeyLayout> {
if !self.prefix.is_empty() {
// the legacy bug only triggered on empty prefixes
return Ok(KeyLayout::Standard);
}
// For each anchor, HEAD the clean and rooted keys and compare `ETag`s
// rather than mere existence. Some S3-compatible stores (e.g. MinIO) strip
// a leading slash, so HEAD("/repo") returns the *same* object as
// HEAD("repo"); that is key normalization, not a mixed repository, and must
// resolve to the clean layout.
let probes = DEFAULT_LAYOUT_ANCHORS.iter().map(|anchor| {
let clean_key = self.key_for(KeyLayout::Standard, anchor);
let rooted_key = self.key_for(KeyLayout::LegacyRoot, anchor);
async move {
let (clean, rooted) = futures::future::try_join(
self.head_etag(ctx, &clean_key),
self.head_etag(ctx, &rooted_key),
)
.await?;
Ok::<_, StorageError>(AnchorProbe { clean, rooted })
}
});
let anchors = futures::future::try_join_all(probes).await?;
layout_from_anchor_etags(&anchors, &self.bucket)
}
/// HEAD a single key, returning its `ETag` if the object exists, or `None` if
/// absent.
///
/// Treats both 404 and 400 as "absent" because some S3-compatible stores
/// (rustfs, some `MinIO` versions) reject leading-slash keys with 400 rather
/// than 404. The `ETag` lets [`Self::probe_layout`] tell a genuinely-rooted
/// object apart from a normalizing store that maps `"/x"` to the same object
/// as `"x"`.
async fn head_etag(
&self,
ctx: &StorageContext<'_>,
key: &str,
) -> StorageResult<Option<String>> {
let mut req = self
.get_client(ctx.settings)
.await
.head_object()
.bucket(self.bucket.clone())
.key(key);
if self.config.requester_pays {
req = req.request_payer(aws_sdk_s3::types::RequestPayer::Requester);
}
match req.customize().config_override(attribution_override(ctx)).send().await {
Ok(out) => Ok(Some(out.e_tag().unwrap_or_default().to_string())),
Err(sdk_err) => {
let absent = sdk_err.as_service_error().is_some_and(|e| e.is_not_found())
|| sdk_err
.raw_response()
.is_some_and(|r| matches!(r.status().as_u16(), 404 | 400));
if absent { Ok(None) } else { obj_store_error_res(sdk_err) }
}
}
}
async fn put_object_single<
I: IntoIterator<Item = (impl Into<String>, impl Into<String>)>,
>(
&self,
ctx: &StorageContext<'_>,
key: &str,
bytes: Bytes,
content_type: Option<impl Into<String>>,
metadata: I,
previous_version: Option<&VersionInfo>,
) -> StorageResult<VersionedUpdateResult> {
let mut req = self
.get_client(ctx.settings)
.await
.put_object()
.bucket(self.bucket.clone())
.key(key)
.body(bytes.into());
if let Some(algo) = self.config.checksum_algorithm {
req = req.checksum_algorithm(to_sdk_checksum(algo));
}
if ctx.settings.unsafe_use_metadata() {
if let Some(ct) = content_type {
req = req.content_type(ct);
};
for (k, v) in metadata {
req = req.metadata(k, v);
}
}
if let Some(klass) = ctx.settings.storage_class() {
let klass = klass.as_str().into();
req = req.storage_class(klass);
}
let conditional_applied = match conditional_for(previous_version, ctx.settings) {
Some(Conditional::IfNoneMatch) => {
req = req.if_none_match("*");
true
}
Some(Conditional::IfMatch(etag)) => {
req = req.if_match(strip_quotes(etag));
true
}
None => false,
};
let write_id = write_id_for(ctx.settings, conditional_applied, next_write_id);
if let Some(id) = write_id.as_deref() {
req = req.metadata(WRITE_ID_METADATA_KEY, id);
}
match req.customize().config_override(attribution_override(ctx)).send().await {
Ok(out) => {
let new_etag = out
.e_tag()
.ok_or(other_error("Object should have an etag".to_string()))?
.to_string();
let new_version = VersionInfo::from_etag_only(new_etag);
Ok(VersionedUpdateResult::Updated { new_version })
}
// minio returns this
Err(SdkError::ServiceError(err)) => {
let code = err.err().meta().code().unwrap_or_default();
if is_precondition_code(code) {
self.recover_precondition(ctx, key, write_id.as_deref()).await
} else {
obj_store_error_res(SdkError::<PutObjectError>::ServiceError(err))
}
}
// S3 API documents this
Err(SdkError::ResponseError(err)) => {
let status = err.raw().status().as_u16();
// see https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObject.html#API_PutObject_RequestSyntax
if status == 409 || status == 412 {
self.recover_precondition(ctx, key, write_id.as_deref()).await
} else {
obj_store_error_res(SdkError::<PutObjectError>::ResponseError(err))
}
}
Err(err) => obj_store_error_res(err),
}
}
async fn put_object_multipart<
I: IntoIterator<Item = (impl Into<String>, impl Into<String>)>,
>(
&self,
ctx: &StorageContext<'_>,
key: &str,
bytes: &Bytes,
content_type: Option<impl Into<String>>,
metadata: I,
previous_version: Option<&VersionInfo>,
) -> StorageResult<VersionedUpdateResult> {
let mut multi = self
.get_client(ctx.settings)
.await
.create_multipart_upload()
// We would like this, but it fails in MinIO
//.checksum_type(aws_sdk_s3::types::ChecksumType::FullObject)
//.checksum_algorithm(aws_sdk_s3::types::ChecksumAlgorithm::Crc64Nvme)
.bucket(self.bucket.clone())
.key(key);
// Write-id rides as metadata on `create_multipart_upload`, so decide
// here whether the later `complete` will carry a condition.
let will_apply_condition =
conditional_for(previous_version, ctx.settings).is_some();
let write_id = write_id_for(ctx.settings, will_apply_condition, next_write_id);
if ctx.settings.unsafe_use_metadata() {
if let Some(ct) = content_type {
multi = multi.content_type(ct);
};
for (k, v) in metadata {
multi = multi.metadata(k, v);
}
if let Some(id) = write_id.as_deref() {
multi = multi.metadata(WRITE_ID_METADATA_KEY, id);
}
}
if let Some(klass) = ctx.settings.storage_class() {
let klass = klass.as_str().into();
multi = multi.storage_class(klass);
}
let create_res = multi
.customize()
.config_override(attribution_override(ctx))
.send()
.await
.capture_box()?;
let upload_id = create_res.upload_id().ok_or(other_error(
"No upload_id in create multipart upload result".to_string(),
))?;
// We need to ensure all requests are the same size except for the last one, which can be
// smaller. This is a requirement for R2 compatibility
let parts = split_in_multiple_equal_requests(
&(0..bytes.len() as u64),
ctx.settings.concurrency().ideal_concurrent_request_size().get(),
ctx.settings.concurrency().max_concurrent_requests_for_object().get(),
)
.collect::<Vec<_>>();
let results = parts
.into_iter()
.enumerate()
.map(|(part_idx, range)| async move {
let body = bytes.slice(range.start as usize..range.end as usize).into();
let idx = part_idx as i32 + 1;
let mut req = self
.get_client(ctx.settings)
.await
.upload_part()
.upload_id(upload_id)
.bucket(self.bucket.clone())
.key(key)
.part_number(idx)
.body(body);
if let Some(algo) = self.config.checksum_algorithm {
req = req.checksum_algorithm(to_sdk_checksum(algo));
}
req.customize()
.config_override(attribution_override(ctx))
.send()
.await
.map(|res| (idx, res))
})
.collect::<FuturesOrdered<_>>();
let completed_parts = results
.map_ok(|(idx, res)| {
let etag = res.e_tag().unwrap_or("");
CompletedPart::builder()
.e_tag(strip_quotes(etag))
.part_number(idx)
.build()
})
.try_collect::<Vec<_>>()
.await
.capture_box()?;
let completed_parts =
CompletedMultipartUpload::builder().set_parts(Some(completed_parts)).build();
let mut req = self
.get_client(ctx.settings)
.await
.complete_multipart_upload()
.bucket(self.bucket.clone())
.key(key)
.upload_id(upload_id)
//.checksum_type(aws_sdk_s3::types::ChecksumType::FullObject)
.multipart_upload(completed_parts);
match conditional_for(previous_version, ctx.settings) {
Some(Conditional::IfNoneMatch) => req = req.if_none_match("*"),
Some(Conditional::IfMatch(etag)) => req = req.if_match(strip_quotes(etag)),
None => {}
}
match req.customize().config_override(attribution_override(ctx)).send().await {
Ok(out) => {
let new_etag = out
.e_tag()
.ok_or(other_error("Object should have an etag".to_string()))?
.to_string();
let new_version = VersionInfo::from_etag_only(new_etag);
Ok(VersionedUpdateResult::Updated { new_version })
}
Err(SdkError::ServiceError(err)) => {
let code = err.err().meta().code().unwrap_or_default();
// `NoSuchUpload` = SDK-retried complete after a lost response,
// or a failed upload; readback tells them apart.
if is_precondition_code(code) {
self.recover_precondition(ctx, key, write_id.as_deref()).await
} else if code == "NoSuchUpload" {
self.recover_lost_response(
ctx,
key,
write_id.as_deref(),
obj_store_error(SdkError::ServiceError(err)),
)
.await
} else {
obj_store_error_res(SdkError::ServiceError(err))
}
}
Err(SdkError::ResponseError(err)) => {
let status = err.raw().status().as_u16();
if status == 409 || status == 412 {
self.recover_precondition(ctx, key, write_id.as_deref()).await
} else if status == 404 {
// NoSuchUpload surfaced as a raw HTTP status.
self.recover_lost_response(
ctx,
key,
write_id.as_deref(),
obj_store_error(SdkError::<PutObjectError>::ResponseError(err)),
)
.await
} else {
obj_store_error_res(SdkError::<PutObjectError>::ResponseError(err))
}
}
Err(err) => obj_store_error_res(err),
}
}
/// Readback + resolve for a tripped conditional (precondition code or
/// 409/412 status).
async fn recover_precondition(
&self,
ctx: &StorageContext<'_>,
key: &str,
write_id: Option<&str>,
) -> StorageResult<VersionedUpdateResult> {
let outcome = self.read_back_after_conditional_failure(ctx, key, write_id).await;
resolve_precondition(outcome, key)
}
/// Readback + resolve for a failure only our own landed write can rescue.
async fn recover_lost_response(
&self,
ctx: &StorageContext<'_>,
key: &str,
write_id: Option<&str>,
original: StorageError,
) -> StorageResult<VersionedUpdateResult> {
let outcome = self.read_back_after_conditional_failure(ctx, key, write_id).await;
resolve_lost_response(outcome, key, original)
}
/// HEAD `key` and classify it against our write-id. `Err` only on a HEAD
/// failure.
async fn read_back_after_conditional_failure(
&self,
ctx: &StorageContext<'_>,
key: &str,
write_id: Option<&str>,
) -> StorageResult<ReadbackOutcome> {
let Some(write_id) = write_id else { return Ok(ReadbackOutcome::NotOurs) };
let mut head = self
.get_client(ctx.settings)
.await
.head_object()
.bucket(self.bucket.clone())
.key(key);
if self.config.requester_pays {
head = head.request_payer(aws_sdk_s3::types::RequestPayer::Requester);
}
let (stored_write_id, version) = match head
.customize()
.config_override(attribution_override(ctx))
.send()
.await
{
Ok(out) => (
out.metadata().and_then(|m| m.get(WRITE_ID_METADATA_KEY)).cloned(),
// S3 has no generation; etag is the only version identity.
out.e_tag()
.map(|e| VersionInfo::from_etag_only(e.to_string()))
.unwrap_or_else(VersionInfo::for_creation),
),
// Absent is conclusive (not a failed read-back): our write didn't land.
Err(sdk_err)
if sdk_err.as_service_error().is_some_and(|e| e.is_not_found())
|| sdk_err
.raw_response()
.is_some_and(|r| r.status().as_u16() == 404) =>
{
return Ok(ReadbackOutcome::NotOurs);
}
Err(sdk_err) => {
warn!(key, error = %sdk_err, "readback HEAD failed; propagating the HEAD error");
return obj_store_error_res(sdk_err);
}
};
Ok(ReadbackOutcome::classify(write_id, stored_write_id.as_deref(), &version))
}
}
pub fn range_to_header(range: &Range<u64>) -> String {
format!("bytes={}-{}", range.start, range.end - 1)
}
impl sealed::Sealed for S3Storage {}
#[async_trait]
#[typetag::serde]
impl Storage for S3Storage {
fn storage_info(&self) -> StorageInfo {
let mut fields = vec![("bucket", self.bucket.clone())];
if !self.prefix.is_empty() {
fields.push(("prefix", self.prefix.clone()));
}
fields.extend(self.config.info_fields());
StorageInfo { backend_type: "S3 (native)", fields }
}
async fn can_write(&self) -> StorageResult<bool> {
Ok(self.can_write)
}
async fn can_create_repository(&self) -> StorageResult<RepositoryCreation> {
if self.prefix.is_empty() && !self.allow_empty_prefix_creation {
Ok(RepositoryCreation::RefusedEmptyPrefix)
} else {
Ok(RepositoryCreation::Allowed)
}
}
async fn put_object(
&self,
ctx: &StorageContext<'_>,
path: &str,
bytes: Bytes,
content_type: Option<&str>,
metadata: Vec<(String, String)>,
previous_version: Option<&VersionInfo>,
) -> StorageResult<VersionedUpdateResult> {
let layout = self.layout(ctx).await?;
let path = self.key_for(layout, path);
if bytes.len() >= ctx.settings.minimum_size_for_multipart_upload() as usize {
self.put_object_multipart(
ctx,
path.as_str(),
&bytes,
content_type,
metadata,
previous_version,
)
.await
} else {
self.put_object_single(
ctx,
path.as_str(),
bytes,
content_type,
metadata,
previous_version,
)
.await
}
}
async fn copy_object(
&self,
ctx: &StorageContext<'_>,
from: &str,
to: &str,
content_type: Option<&str>,
version: &VersionInfo,
) -> StorageResult<VersionedUpdateResult> {
let layout = self.layout(ctx).await?;
let from = format!("{}/{}", self.bucket, self.key_for(layout, from));
let to = self.key_for(layout, to);
let mut req = self
.get_client(ctx.settings)
.await
.copy_object()
.bucket(self.bucket.clone())
.key(to)
.copy_source(from);
if ctx.settings.unsafe_use_conditional_update()
&& let Some(etag) = version.etag()
{
req = req.copy_source_if_match(strip_quotes(etag));
}
if let Some(klass) = ctx.settings.storage_class() {
let klass = klass.as_str().into();
req = req.storage_class(klass);
}
if let Some(ct) = content_type {
req = req.content_type(ct);
}
if self.config.requester_pays {
req = req.request_payer(aws_sdk_s3::types::RequestPayer::Requester);
}
match req.customize().config_override(attribution_override(ctx)).send().await {
Ok(_) => Ok(VersionedUpdateResult::Updated { new_version: version.clone() }),
Err(SdkError::ServiceError(err)) => {
let code = err.err().meta().code().unwrap_or_default();
if code == "PreconditionFailed"
|| code == "ConditionalRequestConflict"
// ConcurrentModification sent by Ceph Object Gateway
|| code == "ConcurrentModification"
{
Ok(VersionedUpdateResult::NotOnLatestVersion)
} else {
obj_store_error_res(SdkError::<CopyObjectError>::ServiceError(err))
}
}
// S3 API documents this
Err(SdkError::ResponseError(err)) => {
let status = err.raw().status().as_u16();
// see https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObject.html#API_PutObject_RequestSyntax
if status == 409 || status == 412 {
Ok(VersionedUpdateResult::NotOnLatestVersion)
} else {
obj_store_error_res(SdkError::<PutObjectError>::ResponseError(err))
}
}
Err(sdk_err) => match sdk_err.as_service_error() {
Some(_)
if sdk_err
.raw_response()
.is_some_and(|x| x.status().as_u16() == 404) =>
{
// needed for Cloudflare R2 public bucket URLs
// if object doesn't exist we get a 404 that isn't parsed by the AWS SDK
// into anything useful. So we need to parse the raw response, and match
// the status code.
obj_not_found_res()
}
_ => obj_store_error_res(sdk_err),
},
}
}
#[instrument(skip(self, ctx))]
async fn list_objects<'a>(
&'a self,
ctx: &StorageContext<'_>,
prefix: &str,
) -> StorageResult<BoxStream<'a, StorageResult<ListInfo<String>>>> {
let layout = self.layout(ctx).await?;
let prefix = self.list_prefix(layout, prefix);
Ok(self.list_keys(ctx, prefix.clone(), prefix).await)
}
#[instrument(skip(self, ctx, id_prefixes))]
async fn list_objects_with_id_prefixes<'a>(
&'a self,
ctx: &StorageContext<'_>,
prefix: &str,
id_prefixes: &[String],
) -> StorageResult<BoxStream<'a, StorageResult<ListInfo<String>>>> {
let layout = self.layout(ctx).await?;
let prefix = self.list_prefix(layout, prefix);
let mut listings = Vec::with_capacity(id_prefixes.len());
for id_prefix in id_prefixes {
let key_prefix = if prefix.is_empty() || prefix.ends_with('/') {
format!("{prefix}{id_prefix}")
} else {
format!("{prefix}/{id_prefix}")
};
listings.push(self.list_keys(ctx, key_prefix, prefix.clone()).await);
}
Ok(stream::select_all(listings).boxed())
}
fn lists_id_prefixes_natively(&self) -> bool {
true
}
#[instrument(skip(self, ctx, batch))]
async fn delete_batch(
&self,
ctx: &StorageContext<'_>,
prefix: &str,
batch: Vec<(String, u64)>,
) -> StorageResult<DeleteObjectsResult> {
fn join_prefix_id(prefix: &str, id: &str) -> String {
if prefix.is_empty() {
id.to_string()
} else {
format!("{}/{}", prefix.trim_end_matches('/'), id)
}
}
let layout = self.layout(ctx).await?;
let mut sizes = HashMap::new();
let mut ids = Vec::new();
for (id, size) in batch.into_iter() {
let key = self.key_for(layout, &join_prefix_id(prefix, &id));
if let Ok(ident) = ObjectIdentifier::builder().key(key.clone()).build() {
ids.push(ident);
sizes.insert(key, size);
}
}
let requested = ids.len();
let delete = Delete::builder()
.set_objects(Some(ids))
.build()
.map_err(|e| other_error(e.to_string()))?;
let mut req = self
.get_client(ctx.settings)
.await
.delete_objects()
.bucket(self.bucket.clone())
.delete(delete);
if let Some(algo) = self.config.checksum_algorithm {
req = req.checksum_algorithm(to_sdk_checksum(algo));
}
if self.config.requester_pays {
req = req.request_payer(aws_sdk_s3::types::RequestPayer::Requester);
}
// A throttle gets its own kind, so the caller can back off instead of
// counting it as a failure. The SDK has already retried it `max_tries`
// times with its own backoff by then.
let res = match req
.customize()
.config_override(attribution_override(ctx))
.send()
.await
{
Ok(res) => res,
Err(SdkError::ServiceError(err)) => {
use aws_sdk_s3::operation::RequestId as _;
let meta = err.err().meta();
let status = err.raw().status().as_u16();
let code = meta.code().unwrap_or("?");
let message = format!(
"DeleteObjects failed: status {status}, code {code}, message {}, request id {}",
meta.message().unwrap_or("?"),
meta.request_id().unwrap_or("?"),
);
return Err(if is_transient_s3_error(Some(status), code) {
throttled_error(code, message)
} else {
other_error(message)
});
}
Err(SdkError::ResponseError(err)) => {
return Err(other_error(format!(
"DeleteObjects failed: unparsable response, status {}",
err.raw().status().as_u16()
)));
}
Err(err) => return Err(obj_store_error(err)),
};
if let Some(errors) = res.errors.as_deref().filter(|e| !e.is_empty()) {
if let Some(code) = transient_per_key_code(errors) {
return Err(throttled_error(
code,
format!(
"DeleteObjects: {} of {requested} keys failed with {code}: {}",
errors.len(),
errors.first().and_then(|e| e.message()).unwrap_or("?"),
),
));
}
tracing::error!(
error = ?errors[0],
failed = errors.len(),
"Errors deleting objects",
);
}
let mut result = DeleteObjectsResult::default();
for deleted in res.deleted() {
if let Some(key) = deleted.key() {
let size = sizes.get(key).unwrap_or(&0);
result.deleted_bytes += *size;
result.deleted_objects += 1;
} else {
tracing::error!("Deleted object without key");
}
}
Ok(result)
}
#[instrument(skip(self, ctx))]
async fn get_object_last_modified(
&self,
ctx: &StorageContext<'_>,
path: &str,
) -> StorageResult<DateTime<Utc>> {
let layout = self.layout(ctx).await?;
let key = self.key_for(layout, path);
let mut req = self
.get_client(ctx.settings)
.await
.head_object()
.bucket(self.bucket.clone())
.key(key);
if self.config.requester_pays {
req = req.request_payer(aws_sdk_s3::types::RequestPayer::Requester);
}
let res = req
.customize()
.config_override(attribution_override(ctx))
.send()
.await
.capture_box()?;
let res = res
.last_modified
.ok_or(other_error("Object has no last_modified field".to_string()))?;
let res = res
.to_chrono_utc()
.map_err(|_| other_error("Invalid metadata timestamp".to_string()))?;
Ok(res)
}
#[instrument(skip(self, ctx))]
async fn get_object_conditional(
&self,
ctx: &StorageContext<'_>,
path: &str,
previous_version: Option<&VersionInfo>,
) -> StorageResult<GetModifiedResult> {
match self.get_object_range_conditional(ctx, path, None, previous_version).await {
Ok(Some((stream, new_version))) => {
let reader = StreamReader::new(stream.map_err(std::io::Error::other));
Ok(GetModifiedResult::Modified { data: Box::pin(reader), new_version })
}
Ok(None) => Ok(GetModifiedResult::OnLatestVersion),
Err(e) => Err(e),
}
}
async fn get_object_range(
&self,
ctx: &StorageContext<'_>,
path: &str,
range: Option<&Range<u64>>,
) -> StorageResult<(
Pin<Box<dyn Stream<Item = Result<Bytes, StorageError>> + Send>>,
VersionInfo,
)> {
self.get_object_range_conditional(ctx, path, range, None).await.map(|v| {
// If we got a result, then we can unwrap safely here:
// Errors would be in the other branch, and None is only expected
// if previous_version was passed in function call, but we set it to None
#[expect(clippy::expect_used)]
v.expect("Logic bug in get_object_range_conditional, should not get None")
})
}
}
impl S3Storage {
async fn get_object_range_conditional(
&self,
ctx: &StorageContext<'_>,
path: &str,
range: Option<&Range<u64>>,
previous_version: Option<&VersionInfo>,
) -> StorageResult<
Option<(
Pin<Box<dyn Stream<Item = Result<Bytes, StorageError>> + Send>>,
VersionInfo,
)>,
> {
let layout = self.layout(ctx).await?;
let client = self.get_client(ctx.settings).await;
let bucket = self.bucket.clone();
let key = self.key_for(layout, path);
let mut req = client.get_object().bucket(bucket).key(key);
if let Some(range) = range {
req = req.range(range_to_header(range));
}
if self.config.requester_pays {
req = req.request_payer(aws_sdk_s3::types::RequestPayer::Requester);
}
if let Some(previous_version) = previous_version.as_ref()
&& let Some(etag) = previous_version.etag()
{
req = req.if_none_match(strip_quotes(etag));
};
match req.customize().config_override(attribution_override(ctx)).send().await {
Ok(output) => match output.e_tag {
Some(etag) => {
let stream = stream2stream(output.body)
.map_err(|e| StorageError::capture(e.into()));
Ok(Some((Box::pin(stream), VersionInfo::from_etag_only(etag))))
}
None => Err(other_error("Object should have an etag".to_string())),
},
Err(sdk_err) => {
match sdk_err.as_service_error() {
Some(e) if e.is_no_such_key() => {
obj_not_found_res()
}
Some(_)
if sdk_err
.raw_response()
.is_some_and(|x| x.status().as_u16() == 404) =>
{
// needed for Cloudflare R2 public bucket URLs
// if object doesn't exist we get a 404 that isn't parsed by the AWS SDK
// into anything useful. So we need to parse the raw response, and match
// the status code.
obj_not_found_res()
}
Some(_)
// aws_sdk_s3 doesn't return an error when
// status 304 (Not Modified) happens, so
// check the http status code here and
// return None to make it easy to catch
// downstream
if sdk_err
.raw_response()
.is_some_and(|x| x.status().as_u16() == 304) =>
{
trace!("Received 304 (Not Modified). Treating requested object as up-to-date.");
Ok(None)
}
_ => obj_store_error_res(sdk_err),
}
}
}
}
}
#[derive(Debug)]
struct AnchorProbe<T> {
/// Modern layout using non-rooted keys
clean: T,
/// Legacy buggy layout using /chunks style keys on empty prefixes
rooted: T,
}
/// Resolve the layout from per-anchor HEAD results.
///
/// For each anchor, comparing `ETag`s (not mere existence) distinguishes a store
/// that normalizes `"/x"` to `"x"` (same object under both keys → clean layout)
/// from one that genuinely holds distinct objects at both keys (→ mixed).
fn layout_from_anchor_etags(
anchors: &[AnchorProbe<Option<String>>],
bucket: &str,
) -> StorageResult<KeyLayout> {
let seen = anchors.iter().fold(
AnchorProbe { clean: false, rooted: false },
|AnchorProbe { clean: clean_seen, rooted: rooted_seen },
AnchorProbe { clean, rooted }| {
let (clean_vote, rooted_vote) = match (clean, rooted) {
// Same physical object under both keys: the store normalized away
// the leading slash. Clean layout, not a mixed repository.
(Some(c), Some(r)) if c == r => (true, false),
// Genuinely distinct objects at both keys: ambiguous; flag as mixed.
(Some(_), Some(_)) => (true, true),
(Some(_), None) => (true, false),
(None, Some(_)) => (false, true),
(None, None) => (false, false),
};
AnchorProbe {
clean: clean_seen || clean_vote,
rooted: rooted_seen || rooted_vote,
}
},
);
decide_layout(&seen, bucket)
}
/// `bucket` is only used to build the mixed-layout error.
fn decide_layout(seen: &AnchorProbe<bool>, bucket: &str) -> StorageResult<KeyLayout> {
match seen {
// Both layouts present: ambiguous and unsafe
AnchorProbe { clean: true, rooted: true } => Err(other_error(format!(
"repository in bucket {bucket} has objects under both the standard and the \
legacy leading-slash key layouts; this is ambiguous and unsafe. \
See https://github.com/earth-mover/icechunk/issues/2239"
))),
AnchorProbe { clean: false, rooted: true } => Ok(KeyLayout::LegacyRoot),
// (true, false) => existing clean repo;
// (false, false) => empty/new repository. Both use the clean layout.
_ => Ok(KeyLayout::Standard),
}
}
fn object_to_list_info(prefix: &str, object: &Object) -> StorageResult<ListInfo<String>> {
let inner = || {
let key = object.key()?;
let last_modified = object.last_modified()?;
let created_at = last_modified.to_chrono_utc().ok()?;
let prefix = Utf8UnixPath::new(prefix);
let id = Utf8UnixPath::new(key).strip_prefix(prefix).ok()?.to_string();
let size_bytes = object.size.unwrap_or(0) as u64;
Some(ListInfo { id, created_at, size_bytes })
};
inner()
.ok_or_else(|| StorageError::capture(StorageErrorKind::BadPrefix(prefix.into())))
}
#[derive(Debug)]
struct ProvideRefreshableCredentials(Arc<dyn S3CredentialsFetcher>);
impl ProvideCredentials for ProvideRefreshableCredentials {
fn provide_credentials<'a>(
&'a self,
) -> aws_credential_types::provider::future::ProvideCredentials<'a>
where
Self: 'a,
{
aws_credential_types::provider::future::ProvideCredentials::new(self.provide())
}
}
impl ProvideRefreshableCredentials {
async fn provide(
&self,
) -> Result<aws_credential_types::Credentials, CredentialsError> {
let creds = self
.0
.get()
.await
.inspect_err(|err| error!(error = err, "Cannot load credentials"))
.map_err(CredentialsError::not_loaded)?;
let creds = aws_credential_types::Credentials::new(
creds.access_key_id,
creds.secret_access_key,
creds.session_token,
creds.expires_after.map(|e| e.into()),
"user",
);
Ok(creds)
}
}
// Factory functions
/// Build storage for an S3 (or S3-compatible, non-Tigris) bucket.
///
/// `extra_read_headers`/`extra_write_headers` are extra HTTP headers attached to
/// read/write requests respectively.
///
/// For `legacy_rooted_keys`, see [`S3Storage::new`]: `None` auto-detects the key
/// layout (the usual choice), `Some(true)` forces the legacy leading-slash layout,
/// and `Some(false)` forces the standard layout.
pub fn new_s3_storage(
config: S3Options,
bucket: String,
prefix: Option<String>,
credentials: Option<S3Credentials>,
extra_read_headers: Vec<(String, String)>,
extra_write_headers: Vec<(String, String)>,
legacy_rooted_keys: Option<bool>,
) -> StorageResult<Arc<dyn Storage + Send + Sync>> {
Ok(Arc::new(s3_storage(
config,
bucket,
prefix,
credentials,
extra_read_headers,
extra_write_headers,
legacy_rooted_keys,
)?))
}
/// Build storage for an S3 (or S3-compatible, non-Tigris) bucket.
///
/// `extra_read_headers`/`extra_write_headers` are extra HTTP headers attached to
/// read/write requests respectively.
///
/// For `legacy_rooted_keys`, see [`S3Storage::new`]: `None` auto-detects the key
/// layout (the usual choice), `Some(true)` forces the legacy leading-slash layout,
/// and `Some(false)` forces the standard layout.
pub fn s3_storage(
config: S3Options,
bucket: String,
prefix: Option<String>,
credentials: Option<S3Credentials>,
extra_read_headers: Vec<(String, String)>,
extra_write_headers: Vec<(String, String)>,
legacy_rooted_keys: Option<bool>,
) -> StorageResult<S3Storage> {
if let Some(endpoint) = &config.endpoint_url
&& (endpoint.contains("fly.storage.tigris.dev")
|| endpoint.contains("t3.storage.dev"))
{
return Err(other_error(
"Tigris Storage is not S3 compatible, use the Tigris specific constructor instead"
.to_string(),
));
}
S3Storage::new(
config,
bucket,
prefix,
credentials.unwrap_or(S3Credentials::FromEnv),
true,
extra_read_headers,
extra_write_headers,
legacy_rooted_keys,
)
}
/// Build storage for a Cloudflare R2 bucket.
///
/// `extra_read_headers`/`extra_write_headers` are extra HTTP headers attached to
/// read/write requests respectively.
///
/// For `legacy_rooted_keys`, see [`S3Storage::new`]: `None` auto-detects the key
/// layout (the usual choice), `Some(true)` forces the legacy leading-slash layout,
/// and `Some(false)` forces the standard layout.
#[expect(clippy::too_many_arguments)]
pub fn new_r2_storage(
config: S3Options,
bucket: Option<String>,
prefix: Option<String>,
account_id: Option<String>,
credentials: Option<S3Credentials>,
extra_read_headers: Vec<(String, String)>,
extra_write_headers: Vec<(String, String)>,
legacy_rooted_keys: Option<bool>,
) -> StorageResult<Arc<dyn Storage + Send + Sync>> {
Ok(Arc::new(r2_storage(
config,
bucket,
prefix,
account_id,
credentials,
extra_read_headers,
extra_write_headers,
legacy_rooted_keys,
)?))
}
/// Build storage for a Cloudflare R2 bucket.
///
/// `extra_read_headers`/`extra_write_headers` are extra HTTP headers attached to
/// read/write requests respectively.
///
/// For `legacy_rooted_keys`, see [`S3Storage::new`]: `None` auto-detects the key
/// layout (the usual choice), `Some(true)` forces the legacy leading-slash layout,
/// and `Some(false)` forces the standard layout.
#[expect(clippy::too_many_arguments)]
pub fn r2_storage(
config: S3Options,
bucket: Option<String>,
prefix: Option<String>,
account_id: Option<String>,
credentials: Option<S3Credentials>,
extra_read_headers: Vec<(String, String)>,
extra_write_headers: Vec<(String, String)>,
legacy_rooted_keys: Option<bool>,
) -> StorageResult<S3Storage> {
let (bucket, prefix) = match (bucket, prefix) {
(Some(bucket), Some(prefix)) => (bucket, Some(prefix)),
(None, Some(prefix)) => match prefix.split_once("/") {
Some((bucket, prefix)) => (bucket.to_string(), Some(prefix.to_string())),
None => (prefix, None),
},
(Some(bucket), None) => (bucket, None),
(None, None) => {
return Err(StorageErrorKind::R2ConfigurationError(
"Either bucket or prefix must be provided.".to_string(),
))
.capture();
}
};
if config.endpoint_url.is_none() && account_id.is_none() {
return Err(StorageErrorKind::R2ConfigurationError(
"Either endpoint_url or account_id must be provided.".to_string(),
))
.capture();
}
let mut config = config;
if config.region.is_none() {
config.region = Some("auto".to_string());
}
if config.endpoint_url.is_none() {
config.endpoint_url =
account_id.map(|x| format!("https://{x}.r2.cloudflarestorage.com"));
}
config.force_path_style = true;
S3Storage::new(
config,
bucket,
prefix,
credentials.unwrap_or(S3Credentials::FromEnv),
true,
extra_read_headers,
extra_write_headers,
legacy_rooted_keys,
)
}
/// The Hugging Face S3-compatible gateway.
const HF_GATEWAY_ENDPOINT: &str = "https://s3.hf.co";
/// Build storage for a Hugging Face Storage Bucket.
///
/// `namespace` owns the bucket: a Hugging Face user or organization. The gateway
/// scopes every operation to it through the endpoint path.
///
/// Hugging Face runs one gateway, so this overwrites any `endpoint_url` in
/// `config`.
///
/// `extra_read_headers`/`extra_write_headers` are extra HTTP headers attached to
/// read/write requests respectively.
///
/// For `legacy_rooted_keys`, see [`S3Storage::new`]: `None` auto-detects the key
/// layout (the usual choice), `Some(true)` forces the legacy leading-slash layout,
/// and `Some(false)` forces the standard layout.
#[expect(clippy::too_many_arguments)]
pub fn new_hf_storage(
config: S3Options,
bucket: String,
prefix: Option<String>,
namespace: &str,
credentials: Option<S3Credentials>,
extra_read_headers: Vec<(String, String)>,
extra_write_headers: Vec<(String, String)>,
legacy_rooted_keys: Option<bool>,
) -> StorageResult<Arc<dyn Storage + Send + Sync>> {
Ok(Arc::new(hf_storage(
config,
bucket,
prefix,
namespace,
credentials,
extra_read_headers,
extra_write_headers,
legacy_rooted_keys,
)?))
}
/// The concrete storage behind [`new_hf_storage`], so tests can read its config.
#[expect(clippy::too_many_arguments)]
fn hf_storage(
config: S3Options,
bucket: String,
prefix: Option<String>,
namespace: &str,
credentials: Option<S3Credentials>,
extra_read_headers: Vec<(String, String)>,
extra_write_headers: Vec<(String, String)>,
legacy_rooted_keys: Option<bool>,
) -> StorageResult<S3Storage> {
let mut config = config;
config.endpoint_url = Some(format!("{HF_GATEWAY_ENDPOINT}/{namespace}"));
if config.region.is_none() {
// the gateway serves one region and still requires the field
config.region = Some("us-east-1".to_string());
}
// the gateway does not serve virtual-hosted style URLs
config.force_path_style = true;
S3Storage::new(
config,
bucket,
prefix,
credentials.unwrap_or(S3Credentials::FromEnv),
true,
extra_read_headers,
extra_write_headers,
legacy_rooted_keys,
)
}
/// Build storage for a Tigris bucket.
///
/// `extra_read_headers`/`extra_write_headers` are extra HTTP headers attached to
/// read/write requests. The required `X-Tigris-*` consistency headers take
/// precedence on a name conflict.
///
/// For `legacy_rooted_keys`, see [`S3Storage::new`]: `None` auto-detects the key
/// layout (the usual choice), `Some(true)` forces the legacy leading-slash layout,
/// and `Some(false)` forces the standard layout.
#[expect(clippy::too_many_arguments)]
pub fn new_tigris_storage(
config: S3Options,
bucket: String,
prefix: Option<String>,
credentials: Option<S3Credentials>,
use_weak_consistency: bool,
extra_read_headers: Vec<(String, String)>,
extra_write_headers: Vec<(String, String)>,
legacy_rooted_keys: Option<bool>,
) -> StorageResult<Arc<dyn Storage + Send + Sync>> {
Ok(Arc::new(tigris_storage(
config,
bucket,
prefix,
credentials,
use_weak_consistency,
extra_read_headers,
extra_write_headers,
legacy_rooted_keys,
)?))
}
/// Merge user-supplied headers with headers Icechunk must set itself. On a
/// case-insensitive name conflict the required header wins (user entries with a
/// colliding name are dropped), since the required headers carry correctness
/// guarantees (e.g. Tigris consistency).
fn merge_required_headers(
user: Vec<(String, String)>,
required: Vec<(String, String)>,
) -> Vec<(String, String)> {
let mut merged: Vec<(String, String)> = user
.into_iter()
.filter(|(k, _)| !required.iter().any(|(rk, _)| rk.eq_ignore_ascii_case(k)))
.collect();
merged.extend(required);
merged
}
/// Build storage for a Tigris bucket.
///
/// `extra_read_headers`/`extra_write_headers` are extra HTTP headers attached to
/// read/write requests. The required `X-Tigris-*` consistency headers take
/// precedence on a name conflict.
///
/// For `legacy_rooted_keys`, see [`S3Storage::new`]: `None` auto-detects the key
/// layout (the usual choice), `Some(true)` forces the legacy leading-slash layout,
/// and `Some(false)` forces the standard layout.
#[expect(clippy::too_many_arguments)]
pub fn tigris_storage(
config: S3Options,
bucket: String,
prefix: Option<String>,
credentials: Option<S3Credentials>,
use_weak_consistency: bool,
extra_read_headers: Vec<(String, String)>,
extra_write_headers: Vec<(String, String)>,
legacy_rooted_keys: Option<bool>,
) -> StorageResult<S3Storage> {
let mut config = config;
if config.endpoint_url.is_none() {
config.endpoint_url = Some("https://t3.storage.dev".to_string());
}
let mut tigris_write_headers = Vec::with_capacity(2);
let mut tigris_read_headers = Vec::with_capacity(3);
if !use_weak_consistency {
// TODO: Tigris will need more than this to offer good eventually consistent behavior
// For example: we should use no-cache for branches and config file
if let Some(region) = config.region.as_ref() {
tigris_write_headers.push(("X-Tigris-Regions".to_string(), region.clone()));
tigris_write_headers
.push(("X-Tigris-Consistent".to_string(), "true".to_string()));
tigris_read_headers.push(("X-Tigris-Regions".to_string(), region.clone()));
tigris_read_headers
.push(("Cache-Control".to_string(), "no-cache".to_string()));
tigris_read_headers
.push(("X-Tigris-Consistent".to_string(), "true".to_string()));
} else {
return Err(other_error("Tigris storage requires a region to provide full consistency. Either set the region for the bucket or use the read-only, eventually consistent storage by passing `use_weak_consistency=True` (experts only)".to_string()));
}
}
S3Storage::new(
config,
bucket,
prefix,
credentials.unwrap_or(S3Credentials::FromEnv),
!use_weak_consistency, // notice eventually consistent storage can't do writes
merge_required_headers(extra_read_headers, tigris_read_headers),
merge_required_headers(extra_write_headers, tigris_write_headers),
legacy_rooted_keys,
)
}
#[cfg(test)]
mod tests {
use icechunk_macros::tokio_test;
use super::*;
// Readback classification is unit-tested in `icechunk_storage::readback`.
#[test]
fn transient_delete_errors_are_throttles() {
for (status, code) in [
(Some(500), "InternalError"),
(Some(503), "SlowDown"),
(Some(503), "ServiceUnavailable"),
(Some(429), "?"),
(Some(502), "?"),
(None, "InternalError"),
(None, "SlowDown"),
] {
assert!(is_transient_s3_error(status, code), "{status:?} {code}");
}
for (status, code) in [
(Some(403), "AccessDenied"),
(Some(404), "NoSuchKey"),
(Some(400), "MalformedXML"),
(None, "AccessDenied"),
(None, "?"),
] {
assert!(!is_transient_s3_error(status, code), "{status:?} {code}");
}
}
fn per_key_error(code: Option<&str>) -> aws_sdk_s3::types::Error {
let mut b = aws_sdk_s3::types::Error::builder().key("k").message("m");
if let Some(code) = code {
b = b.code(code);
}
b.build()
}
#[test]
fn partial_delete_is_retried_whole_only_when_every_key_error_is_transient() {
let errs =
[per_key_error(Some("InternalError")), per_key_error(Some("SlowDown"))];
assert_eq!(transient_per_key_code(&errs), Some("InternalError"));
let errs =
[per_key_error(Some("InternalError")), per_key_error(Some("AccessDenied"))];
assert_eq!(transient_per_key_code(&errs), None);
assert_eq!(transient_per_key_code(&[per_key_error(None)]), None);
assert_eq!(transient_per_key_code(&[]), None);
}
#[test]
fn precondition_codes() {
for code in
["PreconditionFailed", "ConditionalRequestConflict", "ConcurrentModification"]
{
assert!(is_precondition_code(code), "{code} must count as a precondition");
}
assert!(!is_precondition_code("NoSuchUpload"));
assert!(!is_precondition_code(""));
}
#[test]
fn endpoint_bucket_separator() {
// these endpoints need the separator the SDK omits
for (input, expected) in [
("https://s3.hf.co/my-namespace", "https://s3.hf.co/my-namespace/"),
("https://s3.hf.co/my-namespace/", "https://s3.hf.co/my-namespace/"),
("http://localhost:9000/a/b", "http://localhost:9000/a/b/"),
// no path, so the SDK builds the URL correctly
("https://fly.storage.tigris.dev", "https://fly.storage.tigris.dev"),
("http://localhost:9000", "http://localhost:9000"),
("http://localhost:9000/", "http://localhost:9000/"),
("localhost:9000", "localhost:9000"),
] {
assert_eq!(endpoint_with_bucket_separator(input), expected, "{input}");
}
}
/// GC fans its listing out over id prefixes only when the backend lists
/// them server-side; losing this override silently turns GC's listing
/// into a single stream.
#[test]
fn s3_lists_id_prefixes_natively() {
let storage = new_s3_storage(
S3Options::default(),
"my-bucket".to_string(),
Some("some/prefix".to_string()),
None,
Vec::new(),
Vec::new(),
None,
)
.unwrap();
assert!(storage.lists_id_prefixes_natively());
}
#[test]
fn hf_storage_sets_gateway_config() {
let storage = hf_storage(
S3Options::default(),
"my-bucket".to_string(),
Some("some/prefix".to_string()),
"my-namespace",
None,
Vec::new(),
Vec::new(),
None,
)
.unwrap();
assert_eq!(
storage.config.endpoint_url.as_deref(),
Some("https://s3.hf.co/my-namespace")
);
assert_eq!(storage.config.region.as_deref(), Some("us-east-1"));
assert!(storage.config.force_path_style);
assert_eq!(storage.bucket, "my-bucket");
}
#[tokio_test]
async fn test_serialize_s3_storage() {
let config = S3Options::default()
.with_region("us-west-2")
.with_endpoint_url("http://localhost:4200")
.with_allow_http(true);
let credentials = S3Credentials::Static(S3StaticCredentials {
access_key_id: "access_key_id".to_string(),
secret_access_key: "secret_access_key".to_string(),
session_token: Some("session_token".to_string()),
expires_after: None,
});
let storage = S3Storage::new(
config,
"bucket".to_string(),
Some("prefix".to_string()),
credentials,
true,
Vec::new(),
Vec::new(),
None,
)
.unwrap();
let serialized = serde_json::to_string(&storage).unwrap();
assert_eq!(
serialized,
r#"{"config":{"region":"us-west-2","endpoint_url":"http://localhost:4200","anonymous":false,"allow_http":true,"force_path_style":false,"network_stream_timeout_seconds":null,"requester_pays":false,"checksum_algorithm":null},"credentials":{"s3_credential_type":"static","access_key_id":"access_key_id","secret_access_key":"secret_access_key","session_token":"session_token","expires_after":null},"bucket":"bucket","prefix":"prefix","can_write":true,"extra_read_headers":[],"extra_write_headers":[],"key_layout":null}"#
);
let deserialized: S3Storage = serde_json::from_str(&serialized).unwrap();
assert_eq!(storage.config, deserialized.config);
}
/// Extra headers are serialized on the struct (so they survive the pickle /
/// distributed path), and round-trip intact.
#[tokio_test]
async fn test_serialize_s3_storage_with_headers() {
let read_headers = vec![("x-amz-meta-reader".to_string(), "r".to_string())];
let write_headers = vec![
("x-amz-acl".to_string(), "bucket-owner-full-control".to_string()),
("x-amz-meta-writer".to_string(), "w".to_string()),
];
let storage = S3Storage::new(
S3Options::default(),
"bucket".to_string(),
Some("prefix".to_string()),
S3Credentials::FromEnv,
true,
read_headers.clone(),
write_headers.clone(),
None,
)
.unwrap();
let serialized = serde_json::to_string(&storage).unwrap();
assert!(serialized.contains("x-amz-acl"), "got: {serialized}");
let deserialized: S3Storage = serde_json::from_str(&serialized).unwrap();
assert_eq!(deserialized.extra_read_headers, read_headers);
assert_eq!(deserialized.extra_write_headers, write_headers);
}
/// Tigris consistency headers win over user-supplied headers on a
/// case-insensitive name conflict; non-conflicting user headers survive.
#[test]
fn test_tigris_user_headers_merge() {
let storage = tigris_storage(
S3Options::default().with_region("iad"),
"bucket".to_string(),
Some("prefix".to_string()),
Some(S3Credentials::FromEnv),
false,
vec![("x-amz-meta-reader".to_string(), "r".to_string())],
vec![
("x-amz-acl".to_string(), "bucket-owner-full-control".to_string()),
// collides (case-insensitively) with the required Tigris header
("x-tigris-regions".to_string(), "hijacked".to_string()),
],
None,
)
.unwrap();
// user read header preserved alongside the injected Tigris read headers
assert!(
storage
.extra_read_headers
.contains(&("x-amz-meta-reader".to_string(), "r".to_string()))
);
assert!(
storage
.extra_read_headers
.iter()
.any(|(k, v)| k == "X-Tigris-Regions" && v == "iad")
);
// user x-amz-acl preserved; the colliding x-tigris-regions override dropped
assert!(storage.extra_write_headers.contains(&(
"x-amz-acl".to_string(),
"bucket-owner-full-control".to_string()
)));
assert!(!storage.extra_write_headers.iter().any(|(_, v)| v == "hijacked"));
assert!(
storage
.extra_write_headers
.iter()
.any(|(k, v)| k == "X-Tigris-Regions" && v == "iad")
);
}
fn storage_with_prefix(prefix: Option<&str>, legacy_rooted_keys: bool) -> S3Storage {
S3Storage::new(
S3Options::default(),
"bucket".to_string(),
prefix.map(str::to_string),
S3Credentials::FromEnv,
true,
Vec::new(),
Vec::new(),
// map the helper's bool: force legacy when set, else auto-detect
legacy_rooted_keys.then_some(true),
)
.unwrap()
}
/// empty prefix + Standard layout never produces a
/// leading slash, for every kind of object.
#[test]
fn test_key_for_standard_empty_prefix() {
let s = storage_with_prefix(Some(""), false);
for relpath in
["chunks/abc123", "repo", "refs/branch.main/ref.json", "config.yaml"]
{
let key = s.key_for(KeyLayout::Standard, relpath);
assert_eq!(key, relpath);
assert!(!key.starts_with('/'), "key {key:?} must not start with a slash");
}
}
#[test]
fn test_key_for_standard_nonempty_prefix() {
let s = storage_with_prefix(Some("foo"), false);
assert_eq!(s.key_for(KeyLayout::Standard, "chunks/abc123"), "foo/chunks/abc123");
// A trailing slash on the prefix is normalized away at construction.
let s = storage_with_prefix(Some("foo/"), false);
assert_eq!(s.key_for(KeyLayout::Standard, "chunks/abc123"), "foo/chunks/abc123");
}
/// `None` and `""` are the two ways #2239 was triggered; they must produce an
/// identical empty prefix (they converge at `unwrap_or_default` in `new`).
#[test]
fn test_none_and_empty_prefix_are_equivalent() {
let from_none = storage_with_prefix(None, false);
let from_empty = storage_with_prefix(Some(""), false);
for relpath in ["chunks/abc123", "repo"] {
assert_eq!(
from_none.key_for(KeyLayout::Standard, relpath),
from_empty.key_for(KeyLayout::Standard, relpath),
);
}
}
/// A leading slash on the relpath is collapsed at the join, but an interior
/// `//` is preserved
#[test]
fn test_list_prefix() {
// leading slash stripped for each layout
let s = storage_with_prefix(Some("foo"), false);
assert_eq!(s.list_prefix(KeyLayout::Standard, "/chunks"), "foo/chunks");
let s = storage_with_prefix(Some(""), false);
assert_eq!(s.list_prefix(KeyLayout::Standard, "/chunks"), "chunks");
let s = storage_with_prefix(Some(""), true);
assert_eq!(s.list_prefix(KeyLayout::LegacyRoot, "/chunks"), "/chunks");
// interior `//` is preserved, matching `key_for`
let s = storage_with_prefix(Some("a//b"), false);
assert_eq!(s.list_prefix(KeyLayout::Standard, "chunks/x"), "a//b/chunks/x");
assert_eq!(
s.list_prefix(KeyLayout::Standard, "chunks/x"),
s.key_for(KeyLayout::Standard, "chunks/x"),
);
}
/// `LegacyRoot` reproduces the exact pre-#2239 buggy layout, so we can keep
/// reading repositories written by old clients.
#[test]
fn test_key_for_legacy_root_reproduces_bug() {
let s = storage_with_prefix(Some(""), true);
assert_eq!(s.key_for(KeyLayout::LegacyRoot, "chunks/abc123"), "/chunks/abc123");
assert_eq!(s.key_for(KeyLayout::LegacyRoot, "repo"), "/repo");
}
#[test]
fn test_legacy_rooted_keys_rejected_with_nonempty_prefix() {
let err = S3Storage::new(
S3Options::default(),
"bucket".to_string(),
Some("foo".to_string()),
S3Credentials::FromEnv,
true,
Vec::new(),
Vec::new(),
Some(true),
)
.unwrap_err();
assert!(err.to_string().contains("empty prefix"), "got: {err}");
}
#[test]
fn test_decide_layout_four_way() {
let seen = |clean, rooted| AnchorProbe { clean, rooted };
// empty/new repo -> clean
assert_eq!(decide_layout(&seen(false, false), "b").unwrap(), KeyLayout::Standard);
// clean repo -> clean
assert_eq!(decide_layout(&seen(true, false), "b").unwrap(), KeyLayout::Standard);
// legacy rooted repo -> legacy
assert_eq!(
decide_layout(&seen(false, true), "b").unwrap(),
KeyLayout::LegacyRoot
);
// mixed -> hard error
let err = decide_layout(&seen(true, true), "mybucket").unwrap_err();
assert!(err.to_string().contains("both"), "got: {err}");
assert!(err.to_string().contains("mybucket"), "got: {err}");
}
#[test]
fn test_layout_from_anchor_etags() {
let et = |s: &str| Some(s.to_string());
let probe =
|clean: Option<String>, rooted: Option<String>| AnchorProbe { clean, rooted };
let layout = |anchors: &[AnchorProbe<Option<String>>]| {
layout_from_anchor_etags(anchors, "b")
};
// empty/new repo: nothing found -> clean
assert_eq!(layout(&[probe(None, None)]).unwrap(), KeyLayout::Standard);
// clean repo (clean key found, rooted key absent: AWS 404 or rustfs 400)
assert_eq!(layout(&[probe(et("E"), None)]).unwrap(), KeyLayout::Standard);
// genuine legacy rooted repo: only the rooted key exists
assert_eq!(layout(&[probe(None, et("E"))]).unwrap(), KeyLayout::LegacyRoot);
// normalizing store (MinIO): "/x" and "x" are the SAME object (equal ETag)
// -> clean, NOT a spurious mixed-layout error (regression for #2239 probe).
assert_eq!(layout(&[probe(et("E"), et("E"))]).unwrap(), KeyLayout::Standard);
// genuinely distinct objects at both keys -> mixed, hard error
assert!(layout(&[probe(et("CLEAN"), et("ROOTED"))]).is_err());
// multi-anchor: V2 repo on a normalizing store (repo found+normalized,
// refs anchor absent) -> clean
assert_eq!(
layout(&[probe(et("E"), et("E")), probe(None, None)]).unwrap(),
KeyLayout::Standard
);
}
/// Forcing the layout pre-seeds the cell and survives a serialize round-trip,
/// so a deserialized (forked/pickled) storage inherits it without probing.
#[test]
fn test_forced_layout_serializes_and_is_inherited() {
let s = storage_with_prefix(Some(""), true);
assert_eq!(s.key_layout.get().copied(), Some(KeyLayout::LegacyRoot));
let json = serde_json::to_string(&s).unwrap();
assert!(json.contains(r#""key_layout":"LegacyRoot""#), "got: {json}");
let restored: S3Storage = serde_json::from_str(&json).unwrap();
assert_eq!(restored.key_layout.get().copied(), Some(KeyLayout::LegacyRoot));
}
}