khive-db 0.12.0

//! S3-compatible `BlobStore` backend (ADR-111 Amendment 2).
//!
//! S3 implementation of the `khive_storage::blob::BlobStore` trait, beside
//! `FsBlobStore` (`stores::blob`). Same content-addressed CAS
//! contract — `ContentRef`, dedup-on-identical-bytes, offline-maintenance-only
//! `delete`/`orphan_sweep` — over an S3-compatible object store via the
//! `object_store` crate's `aws` feature. No provider type crosses the
//! `khive-storage` trait boundary: callers still hold only `Arc<dyn
//! BlobStore>`.

use std::sync::Arc;
use std::time::Duration;

use async_trait::async_trait;
use bytes::Bytes;
use futures::stream::StreamExt;
use object_store::aws::AmazonS3Builder;
use object_store::path::Path as ObjectPath;
use object_store::{
    Error as ObjectStoreError, ObjectMeta, ObjectStore, ObjectStoreExt, PutMode, PutPayload,
};

use khive_storage::blob::{
    BlobOrphanSweepConfig, BlobOrphanSweepResult, BlobStore, ContentRef, MAX_BLOB_WHOLE_BYTES,
};
use khive_storage::error::StorageError;
use khive_storage::types::StorageResult;
use khive_storage::StorageCapability;

use crate::error::SqliteError;

/// Supported object size ceiling for v1: whole-buffer operations are accepted
/// only up to this size. `put` rejects a larger buffer;
/// `get_bounded_verified` enforces the requested limit while consuming the
/// provider stream. A streaming amendment is required for larger blobs.
pub const MAX_OBJECT_BYTES: u64 = MAX_BLOB_WHOLE_BYTES;

/// Default object-key prefix when a caller doesn't override it.
pub const DEFAULT_PREFIX: &str = "blobs";

/// Bounded outer-retry defaults for the idempotent content-addressed `put`
/// (ADR-111 Amendment 2). `object_store` classifies
/// `PutMode::Create` as non-idempotent and therefore never retries it
/// internally on a timeout -- a single slow request would otherwise surface
/// immediately as `StorageError::Driver` even though the object may have been
/// created. Content-addressed `put` IS safe to retry from the outside: the
/// same bytes always hash to the same key, and a retry that lands on an
/// `AlreadyExists` from the first attempt's own write is a dedup success, not
/// a conflict.
pub const DEFAULT_PUT_MAX_ATTEMPTS: u32 = 3;
/// Per-request deadline applied to every network call this store makes
/// (`head`, `get`, `put_opts`, `delete`). A call that does not resolve within
/// this window is treated as a timeout: `put`'s conditional create retries it
/// (if attempts remain) before mapping to `StorageError::Timeout`; every
/// other operation maps a timeout directly, since only the idempotent
/// content-addressed create is safe to retry from here.
pub const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
/// Base delay for jittered exponential backoff between `put` retry attempts.
pub const DEFAULT_PUT_RETRY_BASE_DELAY: Duration = Duration::from_millis(100);

/// Non-secret S3 backend configuration (ADR-111 Amendment 2). Credentials
/// are never part of this struct — `S3BlobStore::new` reads them from the
/// process environment only, never TOML.
#[derive(Clone, Debug)]
pub struct S3BlobStoreConfig {
    pub bucket: String,
    pub region: String,
    /// Compatibility knob for Cloudflare R2, MinIO, Tigris, and similar
    /// services. `None` means real AWS S3's normal regional endpoint.
    pub endpoint: Option<String>,
    pub prefix: String,
    /// Escape hatch for a trusted local test endpoint. `false` by default;
    /// an `http://` endpoint is rejected unless this is set.
    pub allow_http: bool,
}

impl S3BlobStoreConfig {
    pub fn new(bucket: impl Into<String>, region: impl Into<String>) -> Self {
        Self {
            bucket: bucket.into(),
            region: region.into(),
            endpoint: None,
            prefix: DEFAULT_PREFIX.to_string(),
            allow_http: false,
        }
    }

    pub fn with_endpoint(mut self, endpoint: impl Into<String>) -> Self {
        self.endpoint = Some(endpoint.into());
        self
    }

    pub fn with_prefix(mut self, prefix: impl Into<String>) -> Self {
        self.prefix = prefix.into();
        self
    }

    pub fn with_allow_http(mut self, allow_http: bool) -> Self {
        self.allow_http = allow_http;
        self
    }

    fn validate(&self) -> Result<(), SqliteError> {
        if self.bucket.trim().is_empty() {
            return Err(SqliteError::InvalidData(
                "S3 blob store: bucket must not be empty".to_string(),
            ));
        }
        if self.region.trim().is_empty() {
            return Err(SqliteError::InvalidData(
                "S3 blob store: region must not be empty".to_string(),
            ));
        }
        validate_prefix(&self.prefix)?;
        if let Some(endpoint) = &self.endpoint {
            if endpoint.starts_with("http://") && !self.allow_http {
                return Err(SqliteError::InvalidData(format!(
                    "S3 blob store: endpoint {endpoint:?} uses http:// but allow_http is false \
                     (set allow_http=true only for a trusted local test endpoint)"
                )));
            }
        }
        Ok(())
    }
}

/// A non-empty, canonical key prefix: no leading/trailing slash and no
/// empty, `.`, or `..` segment (ADR-111 Amendment 2).
fn validate_prefix(prefix: &str) -> Result<(), SqliteError> {
    if prefix.is_empty() {
        return Err(SqliteError::InvalidData(
            "S3 blob store: prefix must not be empty".to_string(),
        ));
    }
    if prefix.starts_with('/') || prefix.ends_with('/') {
        return Err(SqliteError::InvalidData(format!(
            "S3 blob store: prefix {prefix:?} must not have a leading or trailing slash"
        )));
    }
    for segment in prefix.split('/') {
        if segment.is_empty() || segment == "." || segment == ".." {
            return Err(SqliteError::InvalidData(format!(
                "S3 blob store: prefix {prefix:?} must not contain an empty, '.', or '..' segment"
            )));
        }
    }
    Ok(())
}

/// AWS credentials read from the process environment only (ADR-111
/// Amendment 2: never TOML, never printed, never in errors).
/// `AWS_ACCESS_KEY_ID`/`AWS_SECRET_ACCESS_KEY` are an all-or-nothing pair;
/// `AWS_SESSION_TOKEN` is optional. Startup errors name the missing
/// variable, never a value.
struct S3Credentials {
    access_key_id: String,
    secret_access_key: String,
    session_token: Option<String>,
}

/// Redacted on purpose: credentials must never appear in a panic message
/// (e.g. an `unwrap_err` on the wrong variant in a test) or a log line.
impl std::fmt::Debug for S3Credentials {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("S3Credentials")
            .field("access_key_id", &"<redacted>")
            .field("secret_access_key", &"<redacted>")
            .field(
                "session_token",
                &self.session_token.as_ref().map(|_| "<redacted>"),
            )
            .finish()
    }
}

impl S3Credentials {
    fn from_env() -> Result<Self, SqliteError> {
        let access_key_id = non_empty_env("AWS_ACCESS_KEY_ID");
        let secret_access_key = non_empty_env("AWS_SECRET_ACCESS_KEY");
        let (access_key_id, secret_access_key) = match (access_key_id, secret_access_key) {
            (Some(a), Some(s)) => (a, s),
            (Some(_), None) => {
                return Err(SqliteError::InvalidData(
                    "S3 blob store: AWS_ACCESS_KEY_ID is set but AWS_SECRET_ACCESS_KEY is \
                     missing -- both or neither must be set"
                        .to_string(),
                ));
            }
            (None, Some(_)) => {
                return Err(SqliteError::InvalidData(
                    "S3 blob store: AWS_SECRET_ACCESS_KEY is set but AWS_ACCESS_KEY_ID is \
                     missing -- both or neither must be set"
                        .to_string(),
                ));
            }
            (None, None) => {
                return Err(SqliteError::InvalidData(
                    "S3 blob store: AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY must both be \
                     set in the process environment"
                        .to_string(),
                ));
            }
        };
        Ok(Self {
            access_key_id,
            secret_access_key,
            session_token: non_empty_env("AWS_SESSION_TOKEN"),
        })
    }
}

fn non_empty_env(name: &str) -> Option<String> {
    std::env::var(name).ok().filter(|v| !v.is_empty())
}

fn map_object_store_err(e: ObjectStoreError, op: &'static str) -> StorageError {
    StorageError::driver(StorageCapability::Blob, op, e)
}

fn timeout_error(op: &'static str) -> StorageError {
    StorageError::Timeout {
        operation: op.into(),
    }
}

/// True for the `object_store::Error` shape produced once a request has
/// exhausted `object_store`'s own HTTP/transport retry budget (its
/// `client::retry` module -- private to that crate, so its `RetryError`
/// cannot be downcast to directly -- maps a timeout, connection failure, or
/// exhausted 5xx into this catch-all variant when no more specific HTTP
/// status applies). This is the only shape our own outer retry treats as
/// transient: `PermissionDenied`/`Unauthenticated` are credential failures
/// that a retry cannot fix (ADR-111 Amendment 2: never retried), and the
/// remaining variants (`InvalidPath`, `NotSupported`, `NotImplemented`,
/// `UnknownConfigurationKey`, `Precondition`, `NotModified`) are
/// request-shape errors, not transient ones.
fn is_outer_retryable(e: &ObjectStoreError) -> bool {
    matches!(e, ObjectStoreError::Generic { .. })
}

/// Jittered exponential backoff delay before retry attempt `attempt` (1-based:
/// the delay before the *second* attempt uses `attempt = 1`).
fn retry_backoff(base: Duration, attempt: u32) -> Duration {
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.subsec_nanos())
        .unwrap_or(0);
    // 1.0x-2.0x jitter multiplier, scaled by the attempt number for a simple
    // exponential-ish backoff without pulling in a `rand` runtime dependency.
    let jitter = 1.0 + (nanos % 1000) as f64 / 1000.0;
    base.mul_f64(jitter * attempt as f64)
}

/// An `S3BlobStore` backed by an S3-compatible object store (ADR-111
/// Amendment 2). Object keys follow the same two-level BLAKE3 shard shape as
/// `FsBlobStore`: `{prefix}/{hex[0..2]}/{hex[2..4]}/{hex}`.
///
/// # Bucket versioning is an operator obligation, not a preflight check
///
/// ADR-111 Amendment 2 requires the target bucket to be unversioned (a
/// versioned or versioning-suspended bucket lets `DELETE Object` leave a
/// delete marker while retaining prior bytes, breaking `delete`/
/// `orphan_sweep`'s physical-deletion contract). `object_store`'s
/// `ObjectStore` trait has no `GetBucketVersioning`-equivalent operation, and
/// adding one would mean a raw signed HTTP call or `aws-sdk-s3` outside the
/// dependency this amendment pins -- out of scope for v1. `S3BlobStore::new`
/// does **not** verify bucket versioning state; the operator provisioning
/// the bucket must ensure it is unversioned.
#[derive(Debug)]
pub struct S3BlobStore {
    client: Arc<dyn ObjectStore>,
    prefix: String,
    put_max_attempts: u32,
    request_timeout: Duration,
    put_retry_base_delay: Duration,
}

impl S3BlobStore {
    /// Build a store from non-secret config plus environment credentials.
    ///
    /// Addressing style (spec-gate rider, ADR-111 Amendment 2): AWS has
    /// deprecated path-style requests for new buckets, so when `endpoint` is
    /// omitted (real AWS S3), this uses virtual-hosted-style requests.  When
    /// an explicit `endpoint` is set (R2, MinIO, Tigris, or a local test
    /// double), it uses path-style, matching how those services are
    /// conventionally reached and how `object_store`'s own examples
    /// configure a custom endpoint.
    pub fn new(config: S3BlobStoreConfig) -> Result<Self, SqliteError> {
        config.validate()?;
        let credentials = S3Credentials::from_env()?;

        let mut builder = AmazonS3Builder::new()
            .with_bucket_name(&config.bucket)
            .with_region(&config.region)
            .with_access_key_id(&credentials.access_key_id)
            .with_secret_access_key(&credentials.secret_access_key)
            .with_allow_http(config.allow_http);
        if let Some(token) = &credentials.session_token {
            builder = builder.with_token(token);
        }
        builder = match &config.endpoint {
            Some(endpoint) => builder
                .with_endpoint(endpoint.clone())
                .with_virtual_hosted_style_request(false),
            None => builder.with_virtual_hosted_style_request(true),
        };
        // Conditional-create defaults to `S3ConditionalPut::ETagMatch`
        // (object_store's own default), which is what turns `PutMode::Create`
        // below into an `If-None-Match: *` request -- supported by AWS S3
        // and by R2/MinIO-class S3-compatible stores alike. Not overridden
        // here on purpose.

        let client = builder.build().map_err(|e| {
            SqliteError::InvalidData(format!("S3 blob store: failed to build client: {e}"))
        })?;

        Ok(Self {
            client: Arc::new(client),
            prefix: config.prefix,
            put_max_attempts: DEFAULT_PUT_MAX_ATTEMPTS,
            request_timeout: DEFAULT_REQUEST_TIMEOUT,
            put_retry_base_delay: DEFAULT_PUT_RETRY_BASE_DELAY,
        })
    }

    /// Test-only seam: build a store around an injected `ObjectStore` (a fake
    /// client) instead of a real `AmazonS3`, with retry/timeout parameters
    /// small enough for a unit test to exercise exhaustion paths quickly.
    /// `ObjectStore` (from `object_store`, not a khive-authored trait) is
    /// already the natural seam here -- `S3BlobStore` only ever holds an
    /// `Arc<dyn ObjectStore>`, never a concrete `AmazonS3`.
    #[cfg(test)]
    fn from_client_for_test(
        client: Arc<dyn ObjectStore>,
        prefix: impl Into<String>,
        put_max_attempts: u32,
        request_timeout: Duration,
        put_retry_base_delay: Duration,
    ) -> Result<Self, SqliteError> {
        let prefix = prefix.into();
        validate_prefix(&prefix)?;
        Ok(Self {
            client,
            prefix,
            put_max_attempts,
            request_timeout,
            put_retry_base_delay,
        })
    }

    fn shard_key(&self, content_ref: &ContentRef) -> ObjectPath {
        let hex = content_ref.as_str();
        ObjectPath::from(format!(
            "{}/{}/{}/{}",
            self.prefix,
            &hex[0..2],
            &hex[2..4],
            hex
        ))
    }

    /// Parse an object key back into a `ContentRef`, validating that it
    /// matches this store's exact `{prefix}/{h[0..2]}/{h[2..4]}/{h}` shard
    /// shape. Anything else under the prefix (a foreign key, a partial
    /// upload artifact from another tool, a key with extra path segments) is
    /// not recognized -- `orphan_sweep` must never delete it.
    fn parse_shard_key(&self, location: &ObjectPath) -> Option<ContentRef> {
        let full: &str = location.as_ref();
        let rest = full.strip_prefix(&self.prefix)?.strip_prefix('/')?;
        let mut parts = rest.split('/');
        let shard1 = parts.next()?;
        let shard2 = parts.next()?;
        let hex = parts.next()?;
        if parts.next().is_some() {
            return None;
        }
        let content_ref = ContentRef::from_hex(hex.to_string()).ok()?;
        let full_hex = content_ref.as_str();
        if &full_hex[0..2] != shard1 || &full_hex[2..4] != shard2 {
            return None;
        }
        Some(content_ref)
    }
}

#[async_trait]
impl BlobStore for S3BlobStore {
    async fn put(&self, bytes: Vec<u8>) -> StorageResult<ContentRef> {
        if bytes.len() as u64 > MAX_OBJECT_BYTES {
            return Err(StorageError::InvalidInput {
                capability: StorageCapability::Blob,
                operation: "put".into(),
                message: format!(
                    "payload of {} bytes exceeds the {MAX_OBJECT_BYTES}-byte v1 ceiling \
                     (ADR-111 Amendment 2)",
                    bytes.len()
                ),
            });
        }

        let digest = blake3::hash(&bytes);
        let content_ref = ContentRef::from_digest_bytes(digest.as_bytes());
        let key = self.shard_key(&content_ref);
        let payload_bytes = Bytes::from(bytes);

        // HEAD fast path: content-addressed dedup means an existing object
        // makes this put a no-op, same contract as FsBlobStore. A single
        // timed-out HEAD is not retried here -- the conditional create below
        // covers the same dedup outcome (`AlreadyExists`) if a HEAD timeout
        // masked an object that already exists.
        match tokio::time::timeout(self.request_timeout, self.client.head(&key)).await {
            Ok(Ok(_)) => return Ok(content_ref),
            Ok(Err(ObjectStoreError::NotFound { .. })) => {}
            Ok(Err(e)) => return Err(map_object_store_err(e, "put_head")),
            Err(_elapsed) => {}
        }

        // Bounded outer retry around the conditional create (ADR-111
        // Amendment 2). `object_store` marks `PutMode::Create`
        // non-idempotent and never retries it internally on a timeout, so a
        // single slow request would otherwise surface immediately as
        // `Driver` even though the service may have accepted the write. The
        // create IS safe to retry from here: it is content-addressed, so a
        // retry either lands its own create or observes `AlreadyExists` from
        // an attempt that actually succeeded -- both are dedup success.
        let mut attempt: u32 = 0;
        loop {
            attempt += 1;
            // `PutPayload` is consumed by `put_opts`; cloning `Bytes`
            // shares the same allocation across bounded retry attempts.
            let payload = PutPayload::from(payload_bytes.clone());
            let put_fut = self.client.put_opts(&key, payload, PutMode::Create.into());
            match tokio::time::timeout(self.request_timeout, put_fut).await {
                Ok(Ok(_)) => return Ok(content_ref),
                // A concurrent identical writer published this exact
                // content-addressed key (from another caller, or from a
                // prior attempt of this same retry loop that actually
                // succeeded despite a timed-out response). CAS semantics
                // make that a success, not a conflict.
                Ok(Err(ObjectStoreError::AlreadyExists { .. })) => return Ok(content_ref),
                // Exhausted the retry budget on a transient (non-timeout)
                // error: fall through to the catch-all arm below, which
                // preserves this final source under `Driver`.
                Ok(Err(e)) if is_outer_retryable(&e) && attempt < self.put_max_attempts => {
                    tokio::time::sleep(retry_backoff(self.put_retry_base_delay, attempt)).await;
                    continue;
                }
                Ok(Err(e)) => return Err(map_object_store_err(e, "put")),
                Err(_elapsed) if attempt < self.put_max_attempts => {
                    tokio::time::sleep(retry_backoff(self.put_retry_base_delay, attempt)).await;
                    continue;
                }
                // Exhausted the retry budget and the final attempt was itself
                // a timeout: the deadline, not a transient failure, is what
                // gave up -- classify as `Timeout`, not `Driver`.
                Err(_elapsed) => {
                    return Err(StorageError::Timeout {
                        operation: "put".into(),
                    });
                }
            }
        }
    }

    async fn get_bounded_verified(
        &self,
        content_ref: &ContentRef,
        max_bytes: u64,
    ) -> StorageResult<Vec<u8>> {
        // Reject an invalid portable-envelope request before constructing or
        // polling a provider future. This ordering is observable through the
        // backend-spy conformance test (ADR-160 D2).
        if max_bytes > MAX_BLOB_WHOLE_BYTES {
            return Err(StorageError::InvalidInput {
                capability: StorageCapability::Blob,
                operation: "get_bounded_verified".into(),
                message: format!(
                    "max_bytes {max_bytes} exceeds the {MAX_BLOB_WHOLE_BYTES}-byte portable whole-buffer envelope"
                ),
            });
        }

        let key = self.shard_key(content_ref);
        // One absolute provider deadline spans both GET setup and every body
        // poll; the body does not receive a fresh timeout window.
        let deadline = tokio::time::Instant::now() + self.request_timeout;
        let result = match tokio::time::timeout_at(deadline, self.client.get(&key)).await {
            Ok(Ok(result)) => result,
            Ok(Err(ObjectStoreError::NotFound { .. })) => {
                return Err(StorageError::NotFound {
                    capability: StorageCapability::Blob,
                    resource: "blob",
                    key: content_ref.to_string(),
                });
            }
            Ok(Err(other)) => {
                return Err(map_object_store_err(other, "get_bounded_verified"));
            }
            Err(_elapsed) => return Err(timeout_error("get_bounded_verified")),
        };

        let metadata_bytes = result.meta.size;
        if metadata_bytes > max_bytes {
            return Err(StorageError::BlobTooLarge {
                content_ref: content_ref.clone(),
                max_bytes,
                observed_at_least: metadata_bytes,
            });
        }

        let mut stream = result.into_stream();
        let consume = async {
            let mut bytes = Vec::with_capacity(metadata_bytes as usize);
            loop {
                match stream.next().await {
                    Some(Ok(chunk)) => {
                        let observed_at_least =
                            (bytes.len() as u64).saturating_add(chunk.len() as u64);
                        // Check before append: a false-small metadata value
                        // cannot make the local Vec grow beyond max_bytes.
                        if observed_at_least > max_bytes {
                            return Err(StorageError::BlobTooLarge {
                                content_ref: content_ref.clone(),
                                max_bytes,
                                observed_at_least,
                            });
                        }
                        bytes.extend_from_slice(&chunk);
                    }
                    Some(Err(e)) => {
                        return Err(map_object_store_err(e, "get_bounded_verified"));
                    }
                    None => break,
                }
            }

            let actual_bytes = bytes.len() as u64;
            if metadata_bytes != actual_bytes {
                return Err(StorageError::BlobSizeMismatch {
                    content_ref: content_ref.clone(),
                    metadata_bytes,
                    actual_bytes,
                });
            }

            let actual = ContentRef::from_digest_bytes(blake3::hash(&bytes).as_bytes());
            if actual != *content_ref {
                return Err(StorageError::BlobDigestMismatch {
                    expected: content_ref.clone(),
                    actual,
                });
            }
            Ok(bytes)
        };

        match tokio::time::timeout_at(deadline, consume).await {
            Ok(result) => result,
            Err(_elapsed) => Err(timeout_error("get_bounded_verified")),
        }
    }

    async fn exists(&self, content_ref: &ContentRef) -> StorageResult<bool> {
        let key = self.shard_key(content_ref);
        match tokio::time::timeout(self.request_timeout, self.client.head(&key)).await {
            Ok(Ok(_)) => Ok(true),
            Ok(Err(ObjectStoreError::NotFound { .. })) => Ok(false),
            Ok(Err(e)) => Err(map_object_store_err(e, "exists")),
            Err(_elapsed) => Err(timeout_error("exists")),
        }
    }

    async fn size(&self, content_ref: &ContentRef) -> StorageResult<Option<u64>> {
        let key = self.shard_key(content_ref);
        match tokio::time::timeout(self.request_timeout, self.client.head(&key)).await {
            Ok(Ok(meta)) => Ok(Some(meta.size)),
            Ok(Err(ObjectStoreError::NotFound { .. })) => Ok(None),
            Ok(Err(e)) => Err(map_object_store_err(e, "size")),
            Err(_elapsed) => Err(timeout_error("size")),
        }
    }

    async fn delete(&self, content_ref: &ContentRef) -> StorageResult<bool> {
        let key = self.shard_key(content_ref);
        // HEAD-then-DELETE (ADR-111 Amendment 2): S3 DELETE is idempotent
        // and does not report prior existence, so the trait's `true`/`false`
        // contract needs the preliminary HEAD. Offline-maintenance-only,
        // same as FsBlobStore -- see `BlobStore::delete`'s doc comment; this
        // backend adds no additional coordination against a racing entity
        // write.
        match tokio::time::timeout(self.request_timeout, self.client.head(&key)).await {
            Ok(Ok(_)) => {}
            Ok(Err(ObjectStoreError::NotFound { .. })) => return Ok(false),
            Ok(Err(e)) => return Err(map_object_store_err(e, "delete_head")),
            Err(_elapsed) => return Err(timeout_error("delete_head")),
        }
        match tokio::time::timeout(self.request_timeout, self.client.delete(&key)).await {
            Ok(Ok(_)) => Ok(true),
            Ok(Err(e)) => Err(map_object_store_err(e, "delete")),
            Err(_elapsed) => Err(timeout_error("delete")),
        }
    }

    // Offline-maintenance-only -- see `BlobStore::orphan_sweep`'s doc
    // comment for the concurrency hazard. `config.live_refs` is a snapshot;
    // this method performs no DB coordination, only a diff against whatever
    // set the caller handed it. Pagination is bounded (at most
    // `PAGE_SIZE` remote-listing entries held at once) and deletes run at
    // bounded concurrency; a list or delete failure aborts with an error
    // rather than reporting a partial scan as complete.
    //
    // Every network step -- each `stream.next()` poll and each concurrent
    // delete -- is wrapped in `self.request_timeout`, same as `put`/
    // `get_bounded_verified`/`exists`/`delete` above: a
    // deadline elapsing maps to `StorageError::Timeout`, and any other
    // error the provider returns keeps the `Driver` classification.
    async fn orphan_sweep(
        &self,
        config: &BlobOrphanSweepConfig,
    ) -> StorageResult<BlobOrphanSweepResult> {
        const PAGE_SIZE: usize = 1000;
        const DELETE_CONCURRENCY: usize = 32;

        let prefix_path = ObjectPath::from(self.prefix.clone());
        let mut stream = self.client.list(Some(&prefix_path));

        let mut scanned = 0u64;
        let mut deleted = 0u64;
        let mut would_delete = 0u64;

        loop {
            let mut page: Vec<ObjectMeta> = Vec::with_capacity(PAGE_SIZE);
            while page.len() < PAGE_SIZE {
                match tokio::time::timeout(self.request_timeout, stream.next()).await {
                    Ok(Some(Ok(meta))) => page.push(meta),
                    Ok(Some(Err(e))) => {
                        return Err(map_object_store_err(e, "orphan_sweep_list"));
                    }
                    Ok(None) => break,
                    Err(_elapsed) => return Err(timeout_error("orphan_sweep_list")),
                }
            }
            if page.is_empty() {
                break;
            }
            let page_len = page.len();

            let mut to_delete: Vec<ObjectPath> = Vec::new();
            for meta in page {
                scanned += 1;
                let Some(content_ref) = self.parse_shard_key(&meta.location) else {
                    // Foreign or malformed key under the prefix -- never
                    // recognized as live or orphaned, mirroring
                    // FsBlobStore's handling of stray `.tmp-*` files.
                    continue;
                };
                if config.live_refs.contains(&content_ref) {
                    continue;
                }
                would_delete += 1;
                if !config.dry_run {
                    to_delete.push(meta.location);
                }
            }

            if !to_delete.is_empty() {
                let request_timeout = self.request_timeout;
                let results: Vec<StorageResult<()>> =
                    futures::stream::iter(to_delete.into_iter().map(|path| {
                        let client = Arc::clone(&self.client);
                        async move {
                            match tokio::time::timeout(request_timeout, client.delete(&path)).await
                            {
                                Ok(Ok(())) => Ok(()),
                                Ok(Err(e)) => Err(map_object_store_err(e, "orphan_sweep_delete")),
                                Err(_elapsed) => Err(timeout_error("orphan_sweep_delete")),
                            }
                        }
                    }))
                    .buffer_unordered(DELETE_CONCURRENCY)
                    .collect()
                    .await;
                for r in results {
                    r?;
                    deleted += 1;
                }
            }

            if page_len < PAGE_SIZE {
                break;
            }
        }

        Ok(BlobOrphanSweepResult {
            scanned,
            deleted,
            would_delete,
            grace_period_skipped: 0,
        })
    }
}

#[cfg(test)]
#[path = "blob_s3_tests.rs"]
mod tests;