gonzalo_store_s3/

lib.rs

//! S3-compatible object-store substrate. One JSON object per record at
//! key `namespace/collection/id.json`.

use async_trait::async_trait;
use aws_sdk_s3::Client;
use aws_sdk_s3::error::ProvideErrorMetadata;
use aws_sdk_s3::operation::head_object::HeadObjectError;
use aws_sdk_s3::primitives::DateTime;
use std::collections::{BTreeMap, BTreeSet};
use std::sync::{Arc, RwLock};
use std::time::SystemTime;

use gonzalo_core::{
    BlobEntry, BlobStore, ContentHash, CoreError, DEFAULT_ANCESTOR_CAP, DeletePlan, DeleteResult,
    Identity, KeyPrefix, PurgePlan, PutPlan, PutResult, Record, RecordKey, Result, Revision,
    decode_segment, now_ms, object_key, plan_delete, plan_purge, plan_put, plan_put_raw,
    validate_ancestor_cap,
};

/// Key prefix under which content-addressed blobs live (`blobs/<hash>`), kept
/// separate from record objects (`namespace/collection/id.json`).
const BLOB_PREFIX: &str = "blobs/";

/// What one `ListObjectsV2` traversal saw: the record keys under the prefix,
/// and which of them carry a tombstone marker (ADR 0025).
#[derive(Default)]
struct Listing {
    records: Vec<RecordKey>,
    /// Marked keys, each with the ETag its marker carried in this listing.
    /// Removing a marker is conditional on that ETag, so a marker rewritten
    /// since is left alone (see `delete_marker_if_match`).
    markers: BTreeMap<RecordKey, String>,
}

pub struct S3Store {
    client: Client,
    bucket: String,
    /// Maximum `Record::ancestors` length kept on every committed write
    /// (spec §3.9). Defaults to [`DEFAULT_ANCESTOR_CAP`].
    cap: usize,
    /// Encoded `(namespace, collection)` pairs known to carry the marked flag
    /// (ADR 0025), shared across [`handle`](S3Store::handle) clones.
    ///
    /// Only ever added to: the flag is set once and never cleared, so a cached
    /// `true` cannot go stale. An *unmarked* collection is deliberately not
    /// cached — it re-checks, which costs one `HeadObject` on a path that is
    /// already paying a read per key.
    marked: Arc<RwLock<BTreeSet<(String, String)>>>,
}

impl S3Store {
    /// Build a store from an explicit client and bucket. Use
    /// [`S3Store::connect`] for the common env/endpoint path.
    pub fn new(client: Client, bucket: impl Into<String>) -> Self {
        Self {
            client,
            bucket: bucket.into(),
            cap: DEFAULT_ANCESTOR_CAP,
            marked: Arc::new(RwLock::new(BTreeSet::new())),
        }
    }

    /// Override the ancestor cap (spec §3.9). A cap of `0` is rejected.
    pub fn with_ancestor_cap(mut self, cap: usize) -> Result<Self> {
        self.cap = validate_ancestor_cap(cap)?;
        Ok(self)
    }

    /// Connect using the ambient AWS config (env, profile, IRSA, etc.). If
    /// `endpoint` is `Some`, target an S3-compatible server (MinIO/Garage, etc.)
    /// with path-style addressing; if `region` is `Some`, override the ambient
    /// region (else the AWS env/profile region applies).
    pub async fn connect(
        bucket: impl Into<String>,
        endpoint: Option<String>,
        region: Option<String>,
    ) -> Self {
        let base = aws_config::load_from_env().await;
        let mut builder = aws_sdk_s3::config::Builder::from(&base);
        if let Some(ep) = endpoint {
            builder = builder.endpoint_url(ep).force_path_style(true);
        }
        if let Some(r) = region {
            builder = builder.region(aws_sdk_s3::config::Region::new(r));
        }
        let client = Client::from_conf(builder.build());
        Self::new(client, bucket)
    }

    /// An owned handle to the same bucket, for a spawned task. The client is a
    /// cheap clone sharing one connection pool.
    fn handle(&self) -> Self {
        Self {
            client: self.client.clone(),
            bucket: self.bucket.clone(),
            cap: self.cap,
            // The Arc, not the set: a handle must see what its parent learned.
            marked: Arc::clone(&self.marked),
        }
    }

    async fn read(&self, key: &RecordKey) -> Result<Option<Record>> {
        Ok(self.read_with_etag(key).await?.map(|(rec, _)| rec))
    }

    /// Like [`read`](Self::read) but also returns the object's S3 ETag, which
    /// [`write_planned`](Self::write_planned) feeds back as the write
    /// precondition to make every compare-and-swap atomic (closing the
    /// read-then-write TOCTOU). Returns tombstones: this is a raw read.
    async fn read_with_etag(&self, key: &RecordKey) -> Result<Option<(Record, String)>> {
        let obj = object_key(key);
        match self
            .client
            .get_object()
            .bucket(&self.bucket)
            .key(&obj)
            .send()
            .await
        {
            Ok(resp) => {
                // A missing ETag used to default to `""`, which later went out
                // as `If-Match: ""`. A server that omits the ETag and reads an
                // empty `If-Match` as "no condition" would turn every
                // compare-and-swap into a blind overwrite or delete, so refuse
                // instead of writing unconditionally (gonzalo#286).
                let etag = resp
                    .e_tag()
                    .ok_or_else(|| {
                        CoreError::Backend(format!(
                            "s3: GetObject for {obj} returned no ETag; conditional writes are unsafe against this backend"
                        ))
                    })?
                    .to_string();
                let data = resp
                    .body
                    .collect()
                    .await
                    .map_err(|e| CoreError::Backend(e.to_string()))?
                    .into_bytes();
                let record =
                    serde_json::from_slice(&data).map_err(|e| CoreError::Serde(e.to_string()))?;
                Ok(Some((record, etag)))
            }
            Err(e) => {
                let svc = e.into_service_error();
                if svc.is_no_such_key() {
                    Ok(None)
                } else {
                    Err(CoreError::Backend(svc.to_string()))
                }
            }
        }
    }

    /// Every record key under `prefix`, tombstones included (the raw listing).
    async fn list_keys(&self, prefix: &KeyPrefix) -> Result<Vec<RecordKey>> {
        Ok(self.list_objects(prefix).await?.records)
    }

    /// One `ListObjectsV2` traversal, sorting what it sees into record keys and
    /// the keys that carry a tombstone marker (ADR 0025). Markers are siblings
    /// of their records, so the same prefix covers both and no second traversal
    /// is needed. Paginates off the continuation token (see
    /// [`next_continuation`]).
    async fn list_objects(&self, prefix: &KeyPrefix) -> Result<Listing> {
        let mut s3_prefix = String::new();
        if let Some(ns) = &prefix.namespace {
            s3_prefix.push_str(&gonzalo_core::segment(ns));
            s3_prefix.push('/');
            if let Some(col) = &prefix.collection {
                s3_prefix.push_str(&gonzalo_core::segment(col));
                s3_prefix.push('/');
            }
        }
        let mut out = Listing::default();
        let mut continuation: Option<String> = None;
        loop {
            let mut req = self.client.list_objects_v2().bucket(&self.bucket);
            if !s3_prefix.is_empty() {
                req = req.prefix(&s3_prefix);
            }
            if let Some(token) = &continuation {
                req = req.continuation_token(token);
            }
            let resp = req
                .send()
                .await
                .map_err(|e| CoreError::Backend(e.into_service_error().to_string()))?;
            for obj in resp.contents() {
                let Some(k) = obj.key() else { continue };
                if let Some(key) = parse_object_key(k) {
                    if prefix.matches(&key) {
                        out.records.push(key);
                    }
                } else if let Some(key) = parse_marker_key(k)
                    && prefix.matches(&key)
                    && let Some(etag) = obj.e_tag()
                {
                    out.markers.insert(key, etag.to_string());
                }
            }
            match next_continuation(resp.is_truncated(), resp.next_continuation_token()) {
                Some(token) => continuation = Some(token),
                None => break,
            }
        }
        Ok(out)
    }

    /// Serialize `record` and `PutObject` it at its key under `pre`. A writer
    /// that changed the object after our read makes this a 412, reported as
    /// [`WriteOutcome::LostRace`] instead of clobbering its write.
    async fn put_record_if(&self, record: &Record, pre: Precondition) -> Result<WriteOutcome> {
        let bytes =
            serde_json::to_vec_pretty(record).map_err(|e| CoreError::Serde(e.to_string()))?;
        let mut req = self
            .client
            .put_object()
            .bucket(&self.bucket)
            .key(object_key(&record.key))
            .body(bytes.into());
        req = match pre {
            Precondition::IfAbsent => req.if_none_match("*"),
            Precondition::IfMatch(tag) => req.if_match(tag),
        };
        match req.send().await {
            Ok(_) => Ok(WriteOutcome::Applied),
            Err(e) => {
                let svc = e.into_service_error();
                match race_kind(svc.code()) {
                    Some(kind) => Ok(WriteOutcome::LostRace(kind)),
                    None => Err(CoreError::Backend(svc.to_string())),
                }
            }
        }
    }

    /// `DeleteObject` at `key` only if it still carries `etag` (`If-Match`).
    async fn delete_record_if_match(&self, key: &RecordKey, etag: String) -> Result<WriteOutcome> {
        match self
            .client
            .delete_object()
            .bucket(&self.bucket)
            .key(object_key(key))
            .if_match(etag)
            .send()
            .await
        {
            Ok(_) => Ok(WriteOutcome::Applied),
            Err(e) => {
                let svc = e.into_service_error();
                match race_kind(svc.code()) {
                    Some(kind) => Ok(WriteOutcome::LostRace(kind)),
                    None => Err(CoreError::Backend(svc.to_string())),
                }
            }
        }
    }

    /// Whether `list` may treat an unmarked key in this collection as live —
    /// true once the collection carries the marked flag (ADR 0025).
    ///
    /// A bucket written before markers existed has tombstones with no marker,
    /// and trusting their absence there would resurrect deleted records. So
    /// trust is earned per collection, by a pass that read every key and
    /// backfilled what was missing.
    pub async fn collection_marked(&self, namespace: &str, collection: &str) -> Result<bool> {
        let pair = (
            gonzalo_core::segment(namespace),
            gonzalo_core::segment(collection),
        );
        if self.marked.read().unwrap().contains(&pair) {
            return Ok(true);
        }
        let found = match self
            .client
            .head_object()
            .bucket(&self.bucket)
            .key(marked_flag_key(namespace, collection))
            .send()
            .await
        {
            Ok(_) => true,
            Err(e) => {
                let svc = e.into_service_error();
                if svc.is_not_found() {
                    false
                } else {
                    return Err(CoreError::Backend(svc.to_string()));
                }
            }
        };
        if found {
            self.marked.write().unwrap().insert(pair);
        }
        Ok(found)
    }

    /// Record that every tombstone in this collection carries a marker.
    ///
    /// Only for a caller that just read every key in the collection and wrote
    /// the markers that were missing — the flag is what later listings trust
    /// instead of reading.
    pub async fn mark_collection(&self, namespace: &str, collection: &str) -> Result<()> {
        self.client
            .put_object()
            .bucket(&self.bucket)
            .key(marked_flag_key(namespace, collection))
            .body(b"1".to_vec().into())
            .send()
            .await
            .map_err(|e| CoreError::Backend(e.into_service_error().to_string()))?;
        self.marked.write().unwrap().insert((
            gonzalo_core::segment(namespace),
            gonzalo_core::segment(collection),
        ));
        Ok(())
    }

    /// Write the tombstone marker for `key` (ADR 0025).
    /// Unconditional: the marker carries no version, so writing one that
    /// already exists is a no-op rather than a race to lose.
    async fn put_marker(&self, key: &RecordKey, revision: &Revision) -> Result<()> {
        // The body is the tombstone's revision, which makes each marker
        // distinguishable from the next one written for the same key: a
        // tombstone's counter always exceeds the record it replaced, so no two
        // markers for a key ever share an ETag. `delete_marker_if_match` relies
        // on that to tell "the marker I resolved" from "a marker written since".
        let body = serde_json::to_vec(revision).map_err(|e| CoreError::Serde(e.to_string()))?;
        self.client
            .put_object()
            .bucket(&self.bucket)
            .key(marker_key(key))
            .body(body.into())
            .send()
            .await
            .map(|_| ())
            .map_err(|e| CoreError::Backend(e.into_service_error().to_string()))
    }

    /// Remove the marker for `key` only if it still carries `etag`.
    ///
    /// The conditional is what makes removal safe to do concurrently with a
    /// delete. A listing resolves a marker to "the record is live" or "the
    /// record is gone", then removes it — but between those two steps another
    /// writer may have deleted the record, writing *its* marker first and the
    /// tombstone after. An unconditional removal landing in that window would
    /// strand a tombstone with no marker, which a flagged collection reads as
    /// live. The new marker carries a different revision, so its ETag differs
    /// and this removal turns into a no-op (412) instead.
    async fn delete_marker_if_match(&self, key: &RecordKey, etag: &str) -> Result<()> {
        match self
            .client
            .delete_object()
            .bucket(&self.bucket)
            .key(marker_key(key))
            .if_match(etag)
            .send()
            .await
        {
            Ok(_) => Ok(()),
            Err(e) => {
                let svc = e.into_service_error();
                // A marker rewritten since the listing is not ours to remove,
                // and one already gone needs no removing.
                if race_kind(svc.code()).is_some() || svc.code() == Some("NoSuchKey") {
                    Ok(())
                } else {
                    Err(CoreError::Backend(svc.to_string()))
                }
            }
        }
    }

    /// Whether `key` currently carries a tombstone marker.
    ///
    /// For inspecting the layout — tests and operational debugging. `list`
    /// reads markers out of the listing it already made instead, which is the
    /// entire point of putting them in the key space (ADR 0025).
    pub async fn marker_exists(&self, key: &RecordKey) -> Result<bool> {
        match self
            .client
            .head_object()
            .bucket(&self.bucket)
            .key(marker_key(key))
            .send()
            .await
        {
            Ok(_) => Ok(true),
            Err(e) => {
                let svc = e.into_service_error();
                if svc.is_not_found() {
                    Ok(false)
                } else {
                    Err(CoreError::Backend(svc.to_string()))
                }
            }
        }
    }

    /// Write a marker for `key` without writing a tombstone — the state a
    /// crash between the two leaves behind. Test seam for the stale-marker
    /// cases of ADR 0025; production code always pairs the two writes.
    #[doc(hidden)]
    pub async fn write_marker_for_test(&self, key: &RecordKey) -> Result<()> {
        self.put_marker(key, &Revision::initial(b"stale marker"))
            .await
    }

    /// Remove a marker while leaving its tombstone — the pre-ADR-0025 layout.
    /// Test seam for the upgrade case; nothing in production strips a marker
    /// from a record that is still deleted.
    #[doc(hidden)]
    pub async fn remove_marker_for_test(&self, key: &RecordKey) -> Result<()> {
        self.client
            .delete_object()
            .bucket(&self.bucket)
            .key(marker_key(key))
            .send()
            .await
            .map(|_| ())
            .map_err(|e| CoreError::Backend(e.into_service_error().to_string()))
    }

    /// Read `keys` in bounded-concurrency batches and report which are live.
    /// A key absent from the map was not read.
    ///
    /// The reads run concurrently because one at a time multiplied every key's
    /// latency by a round trip, which is what made a large collection slow
    /// rather than merely expensive (gonzalo#286).
    async fn read_liveness(&self, keys: &[RecordKey]) -> Result<BTreeMap<RecordKey, Liveness>> {
        let mut live = BTreeMap::new();
        for batch in keys.chunks(LIST_READ_CONCURRENCY) {
            let mut reads = tokio::task::JoinSet::new();
            for key in batch {
                // A handle per task: the client is a cheap clone (it shares one
                // connection pool), and a task needs to own what it reads.
                let store = self.handle();
                let key = key.clone();
                reads.spawn(async move {
                    let liveness = listed_as_live(store.read(&key).await)?;
                    Ok::<_, CoreError>((key, liveness))
                });
            }
            while let Some(joined) = reads.join_next().await {
                let (key, liveness) = joined.map_err(|e| CoreError::Backend(e.to_string()))??;
                live.insert(key, liveness);
            }
        }
        Ok(live)
    }

    /// The compare-and-swap loop behind every mutating method. Each attempt
    /// reads the object and its ETag, asks `plan` what to do given that exact
    /// record, and carries it out with a conditional write gated on the same
    /// ETag. A lost race (412, 409, or the object vanishing to a concurrent
    /// purge) means a concurrent writer changed the object after our read, so
    /// the next attempt re-reads and re-plans. The planner therefore always
    /// decides against the true current record, as it does under the fs and
    /// git stores' locks. Gives up after [`MAX_WRITE_ATTEMPTS`] attempts.
    async fn write_planned<T, P>(&self, key: &RecordKey, plan: P) -> Result<T>
    where
        T: Send,
        P: Fn(Option<&Record>) -> Result<Planned<T>> + Sync,
    {
        let plan = &plan;
        // What the previous attempt sent, with the answer it would have
        // returned. A conditional PutObject can apply on the server and still
        // look like a failure — the response times out, the SDK retries, and
        // the server answers 412 because the object now carries the new ETag.
        // Without this, the re-read finds our own write and the planner calls
        // it someone else's, returning a Conflict for a write that committed.
        // `delete_as` and `purge` already converge here; this makes writes
        // agree with them and with `MemStore` (gonzalo#286).
        let pending: std::sync::Mutex<Option<(Record, T)>> = std::sync::Mutex::new(None);
        let pending = &pending;
        retry_on_lost_race(key, move || async move {
            let current = self.read_with_etag(key).await?;

            // Take the previous attempt's record before any await, so the lock
            // is never held across one.
            let previous = pending.lock().unwrap().take();
            if let Some((sent, answer)) = previous
                && own_write_landed(&sent, current.as_ref().map(|(rec, _)| rec))
            {
                return Ok(Step::Done(answer));
            }

            let etag = current.as_ref().map(|(_, tag)| tag.as_str());
            let (outcome, answer) = match plan(current.as_ref().map(|(rec, _)| rec))? {
                Planned::Finish(answer) => return Ok(Step::Done(answer)),
                Planned::Put(record, answer) => {
                    // The marker goes first (ADR 0025). A crash after this
                    // leaves a stale marker, which costs one wasted read; a
                    // crash after the tombstone would leave one *unmarked*,
                    // which is the single state a flagged collection cannot
                    // survive — `list` would show the deleted record as live.
                    if record.is_tombstone() {
                        self.put_marker(key, &record.revision).await?;
                    }
                    let outcome = self.put_record_if(&record, precondition(etag)).await?;
                    if let WriteOutcome::LostRace(_) = outcome {
                        // Remember it, so the next attempt can recognise its own
                        // write if this "failure" was really an ambiguous commit.
                        *pending.lock().unwrap() = Some((record, answer));
                        return Ok(match outcome {
                            WriteOutcome::LostRace(kind) => Step::Retry(kind),
                            WriteOutcome::Applied => unreachable!("checked above"),
                        });
                    }
                    // A recreation leaves the old marker behind on purpose. It
                    // is now stale — the key is live — which costs the next
                    // listing one read and nothing else, and that listing
                    // removes it under the ETag it saw. Removing it here
                    // instead would race a concurrent delete: the other writer
                    // puts its marker before its tombstone, and a removal
                    // landing between the two would strand an unmarked
                    // tombstone. Writers only ever add markers.
                    (outcome, answer)
                }
                Planned::Remove(answer) => {
                    // Planners only remove a record they were given, so an ETag
                    // was read. The empty fallback can never match; it would
                    // just 412 and re-plan.
                    let tag = etag.unwrap_or_default().to_string();
                    // Purge leaves the marker too, for the same reason, and the
                    // next listing sweeps it as an orphan.
                    (self.delete_record_if_match(key, tag).await?, answer)
                }
            };
            Ok(match outcome {
                WriteOutcome::Applied => Step::Done(answer),
                WriteOutcome::LostRace(kind) => Step::Retry(kind),
            })
        })
        .await
    }
}

/// The S3 precondition that enforces OCC atomically at write time, chosen from
/// the caller's `expected` revision and the ETag read for the object.
#[derive(Debug, PartialEq, Eq)]
enum Precondition {
    /// Create only if the object is still absent (`If-None-Match: *`).
    IfAbsent,
    /// Replace only if the object still carries this ETag (`If-Match: <etag>`).
    IfMatch(String),
}

/// Map the ETag read for the object to the write precondition. If an object
/// was read (a live record *or a tombstone*), the write must replace exactly
/// that version (`If-Match`). If none was read, it must still be absent
/// (`If-None-Match: *`). This depends on what's stored, not on the caller's
/// `expected`: recreation is `put(_, None)` over an existing tombstone object,
/// so keying off `expected` would pick `IfAbsent` and fail with 412 every time.
fn precondition(etag: Option<&str>) -> Precondition {
    match etag {
        Some(tag) => Precondition::IfMatch(tag.to_string()),
        None => Precondition::IfAbsent,
    }
}

/// Whether an S3 error code denotes a failed write precondition (HTTP 412) —
/// i.e. a concurrent writer won the race, which OCC surfaces as a `Conflict`.
fn is_precondition_failed(code: Option<&str>) -> bool {
    matches!(code, Some("PreconditionFailed"))
}

/// Why a conditional write did not apply. The kinds differ in what the caller
/// should do next (gonzalo#286).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum RaceKind {
    /// `412`: the object changed after our read, or `NoSuchKey`: it vanished to
    /// a concurrent purge. Another writer already finished, so re-read and
    /// re-plan straight away.
    Settled,
    /// `409 ConditionalRequestConflict`: AWS returns this while a conflicting
    /// conditional write is still **in flight**. Retrying immediately can burn
    /// every attempt inside one in-flight window, so back off first.
    InFlight,
}

/// Classify a lost race by the S3 error code, or `None` if it isn't one.
fn race_kind(code: Option<&str>) -> Option<RaceKind> {
    match code {
        Some("ConditionalRequestConflict") => Some(RaceKind::InFlight),
        Some("PreconditionFailed" | "NoSuchKey") => Some(RaceKind::Settled),
        _ => None,
    }
}

/// Consumer view of a raw read: a tombstone reads as absent (spec §3.2).
fn visible(record: Option<Record>) -> Option<Record> {
    record.filter(|r| !r.is_tombstone())
}

/// What a read said about a key the listing turned up.
#[derive(Clone, Debug, PartialEq, Eq)]
enum Liveness {
    /// Listed.
    Live,
    /// Hidden. Carries the tombstone's revision, which is what its marker
    /// names (ADR 0025), so a pass that finds one unmarked can write it.
    Tombstoned(Revision),
    /// Hidden: purged between the listing and the read.
    Absent,
}

/// Whether consumer `list` includes a key, given the raw read of its object.
/// Tombstones and keys purged since the listing (NotFound) are excluded. An
/// object that fails to decode stays listed, so `get` on that key surfaces the
/// error. Any other read error (network, 5xx, permission) fails the whole
/// `list`: on s3 it may be transient, and listing a key that may be a
/// tombstone would be wrong. fs and git differ: they keep every unreadable
/// entry listed.
fn listed_as_live(read: Result<Option<Record>>) -> Result<Liveness> {
    match read {
        Ok(Some(record)) if record.is_tombstone() => Ok(Liveness::Tombstoned(record.revision)),
        Ok(Some(_)) => Ok(Liveness::Live),
        Ok(None) => Ok(Liveness::Absent),
        Err(CoreError::Serde(_)) => Ok(Liveness::Live),
        Err(e) => Err(e),
    }
}

/// One planner decision, translated into the S3 action that carries it out
/// and the answer to return once that action lands.
// `Record` is the natural payload for a write step; boxing it would ripple
// through every `*_step` signature for no benefit at this call volume.
#[allow(clippy::large_enum_variant)]
#[derive(Debug, PartialEq, Eq)]
enum Planned<T> {
    /// No write needed; return this answer.
    Finish(T),
    /// `PutObject` this record under [`precondition`], then return the answer.
    Put(Record, T),
    /// `DeleteObject` with `If-Match` on the read ETag, then return the answer.
    Remove(T),
}

/// Translate a [`PutPlan`] (from `plan_put` or `plan_put_raw`).
fn put_step(key: &RecordKey, plan: PutPlan) -> Result<Planned<PutResult>> {
    match plan {
        PutPlan::Write(record) => {
            let committed = PutResult::Committed(record.revision.clone());
            Ok(Planned::Put(record, committed))
        }
        PutPlan::Conflict(conflict) => Ok(Planned::Finish(PutResult::Conflict(conflict))),
        PutPlan::NotFound => Err(CoreError::NotFound(key.clone())),
        // Only consumer `plan_put` produces this: for a `RecordKind::Tombstone`
        // record (deletes go through `delete_as`, replication through `put_raw`)
        // and for a create over a manifest tombstone.
        PutPlan::Rejected(reason) => Err(CoreError::Invalid(reason.to_string())),
    }
}

/// Translate a [`DeletePlan`]: a tombstone is written with `PutObject`.
fn delete_step(plan: DeletePlan) -> Planned<DeleteResult> {
    match plan {
        DeletePlan::Write(tombstone) => Planned::Put(tombstone, DeleteResult::Deleted),
        DeletePlan::Noop => Planned::Finish(DeleteResult::Deleted),
        DeletePlan::Conflict(conflict) => Planned::Finish(DeleteResult::Conflict(conflict)),
    }
}

/// Translate a [`PurgePlan`]: the only plan that physically removes an object.
fn purge_step(plan: PurgePlan) -> Planned<DeleteResult> {
    match plan {
        PurgePlan::Remove => Planned::Remove(DeleteResult::Deleted),
        PurgePlan::Noop => Planned::Finish(DeleteResult::Deleted),
        PurgePlan::Conflict(conflict) => Planned::Finish(DeleteResult::Conflict(conflict)),
    }
}

/// Most read → plan → conditional-write attempts one call makes before giving
/// up on a key that other writers keep changing underneath it.
const MAX_WRITE_ATTEMPTS: usize = 8;

/// How many of consumer `list`'s per-key reads run at once. Hiding tombstones
/// costs a `GetObject` per key (spec §8.4); doing them one at a time multiplied
/// every key by a round trip. Bounded so a large collection cannot open an
/// unbounded number of connections (gonzalo#286).
const LIST_READ_CONCURRENCY: usize = 16;

/// Result of one attempt: finished with an answer, or lost the race (412) and
/// must re-read and re-plan.
#[derive(Debug, PartialEq, Eq)]
enum Step<T> {
    Done(T),
    Retry(RaceKind),
}

/// Run `attempt` until it finishes, at most [`MAX_WRITE_ATTEMPTS`] times. An
/// error from an attempt ends the loop immediately. Each retry re-reads and
/// re-plans, so the planner always decides against the true current record,
/// just as it does under the fs and git stores' locks.
async fn retry_on_lost_race<T, F, Fut>(key: &RecordKey, mut attempt: F) -> Result<T>
where
    F: FnMut() -> Fut,
    Fut: std::future::Future<Output = Result<Step<T>>>,
{
    for round in 0..MAX_WRITE_ATTEMPTS {
        match attempt().await? {
            Step::Done(value) => return Ok(value),
            // A settled race (412, or the object vanished to a purge) means the
            // other writer has finished: re-read and re-plan immediately.
            Step::Retry(RaceKind::Settled) => {}
            // A 409 means a conflicting conditional write is still in flight.
            // Retrying straight away can burn every attempt inside that one
            // window, so wait a little, with jitter so racing writers don't
            // line up again (gonzalo#286).
            Step::Retry(RaceKind::InFlight) => {
                tokio::time::sleep(inflight_backoff(round)).await;
            }
        }
    }
    Err(CoreError::Backend(format!(
        "s3: conditional write for {key} lost {MAX_WRITE_ATTEMPTS} consecutive races"
    )))
}

/// Whether the record a previous attempt sent is what the store now holds, so
/// that attempt's write applied after all (gonzalo#286).
///
/// A conditional `PutObject` can apply on the server and still look like a
/// failure: the response times out, the SDK retries, and the server answers
/// `412` because the object already carries the new ETag. Re-planning against
/// that state reports a `Conflict` for the caller's own committed write.
/// Comparing whole records is exact — two different writes never produce the
/// same record, because the revision hashes the body and a rewrite of the same
/// body at the same revision is the same record.
fn own_write_landed(sent: &Record, current: Option<&Record>) -> bool {
    current.is_some_and(|rec| rec == sent)
}

/// How long to wait before retrying after a `409 ConditionalRequestConflict`:
/// a doubling delay from ~4 ms, capped, with jitter so two writers that
/// collided do not wake together and collide again (gonzalo#286).
///
/// The jitter is derived from the clock rather than a random-number generator,
/// which would be a dependency for a few bits of entropy that nothing depends
/// on for correctness: a bad draw only costs another attempt.
fn inflight_backoff(round: usize) -> std::time::Duration {
    const BASE_MS: u64 = 4;
    const CAP_MS: u64 = 250;
    let step = BASE_MS.saturating_mul(1 << round.min(6)).min(CAP_MS);
    let jitter = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| u64::from(d.subsec_nanos()) % step.max(1))
        .unwrap_or(0);
    std::time::Duration::from_millis(step + jitter)
}

/// Result of one conditional S3 write: it applied, or its precondition failed
/// (412) because a concurrent writer changed the object after our read.
#[derive(Debug, PartialEq, Eq)]
enum WriteOutcome {
    Applied,
    /// The write did not apply because another writer got there first; the
    /// kind says whether that writer has finished (gonzalo#286).
    LostRace(RaceKind),
}

/// Decide the continuation token for the next `list_objects_v2` page, driving
/// pagination off token *presence* rather than the `is_truncated` flag. A
/// well-behaved backend only returns a token when there is more to fetch, but a
/// misbehaving one can report `is_truncated = true` yet omit the token; keying
/// off the flag would then re-request page 1 forever. So: if a token is present
/// we continue with it, otherwise we terminate — regardless of `is_truncated`.
/// This guarantees the pagination loop always makes progress or stops.
fn next_continuation(_is_truncated: Option<bool>, token: Option<&str>) -> Option<String> {
    token.map(str::to_string)
}

#[async_trait]
impl gonzalo_core::Store for S3Store {
    // ---- consumer surface (tombstones hidden) ----

    async fn get(&self, key: &RecordKey) -> Result<Option<Record>> {
        Ok(visible(self.read(key).await?))
    }

    async fn list(&self, prefix: &KeyPrefix) -> Result<Vec<RecordKey>> {
        // A tombstone lives at the deleted record's own object key, so the
        // listing cannot tell it from a live record by the record object alone.
        // A marker beside it can (ADR 0025): one traversal classifies every
        // key, and only marked keys are read — one read per *tombstone*, which
        // `collect` bounds, instead of one per *record*, which nothing did.
        let listing = self.list_objects(prefix).await?;

        // Trust is earned per collection, because that is the granularity a
        // pass can prove, so decide per collection. A pass spanning several
        // enumerated each of them fully and may flag each of them.
        let mut by_collection: BTreeMap<(String, String), Vec<RecordKey>> = BTreeMap::new();
        for key in listing.records {
            by_collection
                .entry((key.namespace.clone(), key.collection.clone()))
                .or_default()
                .push(key);
        }
        // A collection can hold markers and no records at all — every record
        // purged, one marker left behind by a purge that died between its two
        // deletes. Grouping on records alone would never visit it, so the
        // orphan would outlive every listing.
        for marker in listing.markers.keys() {
            by_collection
                .entry((marker.namespace.clone(), marker.collection.clone()))
                .or_default();
        }

        let mut out = Vec::new();
        for ((namespace, collection), keys) in by_collection {
            let marked = self.collection_marked(&namespace, &collection).await?;
            // Fast path: read only what a marker points at. Slow path: read
            // every key, exactly as before markers existed, and repair what the
            // old layout left behind.
            let to_read: Vec<RecordKey> = if marked {
                keys.iter()
                    .filter(|k| listing.markers.contains_key(k))
                    .cloned()
                    .collect()
            } else {
                keys.clone()
            };
            let live = self.read_liveness(&to_read).await?;

            for key in &keys {
                match live.get(key) {
                    // Not read: unmarked in a flagged collection, so live.
                    None => out.push(key.clone()),
                    Some(Liveness::Live) => {
                        if let Some(etag) = listing.markers.get(key) {
                            // Live but marked — a marker left by a recreation,
                            // or by a crash before the tombstone landed. Remove
                            // it under the ETag this listing saw, so a delete
                            // that has written a *new* marker since keeps it.
                            self.delete_marker_if_match(key, etag).await?;
                        }
                        out.push(key.clone());
                    }
                    Some(Liveness::Tombstoned(revision)) => {
                        if !listing.markers.contains_key(key) {
                            // A tombstone the old layout left unmarked. The
                            // pass that just proved it is the cheapest place to
                            // repair it.
                            self.put_marker(key, revision).await?;
                        }
                    }
                    Some(Liveness::Absent) => {
                        // Purged between the listing and the read, so its
                        // marker — if any — is now an orphan.
                        if let Some(etag) = listing.markers.get(key) {
                            self.delete_marker_if_match(key, etag).await?;
                        }
                    }
                }
            }

            // A marker whose record was not in the listing at all is an orphan
            // left by a purge, which removes the record and never the marker.
            let listed: BTreeSet<&RecordKey> = keys.iter().collect();
            for (orphan, etag) in listing
                .markers
                .iter()
                .filter(|(m, _)| m.namespace == namespace && m.collection == collection)
                .filter(|(m, _)| !listed.contains(m))
            {
                self.delete_marker_if_match(orphan, etag).await?;
            }

            if !marked {
                self.mark_collection(&namespace, &collection).await?;
            }
        }
        // One deterministic order for the whole listing, so what a caller sees
        // never depends on which read returned first.
        out.sort();
        Ok(out)
    }

    async fn put(&self, record: Record, expected: Option<Revision>) -> Result<PutResult> {
        // Consumer write: a tombstone counts as absent, so `expected = None`
        // over one is a recreation that re-stamps the revision (`plan_put`).
        let key = record.key.clone();
        self.write_planned(&key, |current| {
            put_step(
                &key,
                // The clock is read per attempt: a retry after a lost race is
                // a later write, and its stamps should say so (gonzalo#293).
                plan_put(
                    current,
                    record.clone(),
                    expected.clone(),
                    now_ms(),
                    self.cap,
                ),
            )
        })
        .await
    }

    // `delete` is the trait's provided method: `delete_as(key, expected, None)`.
    async fn delete_as(
        &self,
        key: &RecordKey,
        expected: Option<Revision>,
        author: Option<Identity>,
    ) -> Result<DeleteResult> {
        // Deletion writes a tombstone (spec §3.1). `now_ms()` is taken per
        // attempt, so a retried delete stamps when it actually landed.
        self.write_planned(key, |current| {
            Ok(delete_step(plan_delete(
                current,
                expected.clone(),
                now_ms(),
                self.cap,
                author.as_ref(),
            )))
        })
        .await
    }

    // ---- replication surface (tombstones visible) ----

    async fn get_raw(&self, key: &RecordKey) -> Result<Option<Record>> {
        self.read(key).await
    }

    async fn list_raw(&self, prefix: &KeyPrefix) -> Result<Vec<RecordKey>> {
        self.list_keys(prefix).await
    }

    async fn put_raw(&self, record: Record, expected: Option<Revision>) -> Result<PutResult> {
        // Replication write: stores the caller's revision verbatim, never
        // re-stamps, and a tombstone is a real current record it must name in
        // `expected` (`plan_put_raw`).
        let key = record.key.clone();
        self.write_planned(&key, |current| {
            put_step(
                &key,
                plan_put_raw(
                    current,
                    record.clone(),
                    expected.clone(),
                    now_ms(),
                    self.cap,
                ),
            )
        })
        .await
    }

    async fn purge(&self, key: &RecordKey, expected: Revision) -> Result<DeleteResult> {
        // The only physical removal: a conditional DeleteObject, and only when
        // the stored revision is still `expected`, so a recreation that lands
        // mid-purge survives.
        self.write_planned(key, |current| {
            Ok(purge_step(plan_purge(current, &expected)))
        })
        .await
    }
}

#[async_trait]
impl BlobStore for S3Store {
    async fn put_blob(&self, content: &[u8]) -> Result<ContentHash> {
        let hash = ContentHash::of(content);
        let key = format!("{BLOB_PREFIX}{}", hash.0);
        // Content-addressed + write-if-absent: an existing blob at this key is
        // byte-identical, so `If-None-Match: *` turns a re-upload into a no-op
        // (a 412 just means it's already stored). Idempotent and bandwidth-cheap.
        match self
            .client
            .put_object()
            .bucket(&self.bucket)
            .key(&key)
            .if_none_match("*")
            .body(content.to_vec().into())
            .send()
            .await
        {
            Ok(_) => Ok(hash),
            Err(e) => {
                let svc = e.into_service_error();
                if is_precondition_failed(svc.code()) {
                    Ok(hash) // already present — no-op
                } else {
                    Err(CoreError::Backend(svc.to_string()))
                }
            }
        }
    }

    async fn get_blob(&self, hash: &ContentHash) -> Result<Option<Vec<u8>>> {
        let key = format!("{BLOB_PREFIX}{}", hash.0);
        match self
            .client
            .get_object()
            .bucket(&self.bucket)
            .key(&key)
            .send()
            .await
        {
            Ok(resp) => {
                let data = resp
                    .body
                    .collect()
                    .await
                    .map_err(|e| CoreError::Backend(e.to_string()))?
                    .into_bytes();
                Ok(Some(data.to_vec()))
            }
            Err(e) => {
                let svc = e.into_service_error();
                if svc.is_no_such_key() {
                    Ok(None)
                } else {
                    Err(CoreError::Backend(svc.to_string()))
                }
            }
        }
    }

    async fn list_blobs(&self) -> Result<Vec<BlobEntry>> {
        let mut out = Vec::new();
        let mut continuation: Option<String> = None;
        loop {
            let mut req = self
                .client
                .list_objects_v2()
                .bucket(&self.bucket)
                .prefix(BLOB_PREFIX);
            if let Some(token) = &continuation {
                req = req.continuation_token(token);
            }
            let resp = req
                .send()
                .await
                .map_err(|e| CoreError::Backend(e.into_service_error().to_string()))?;
            for obj in resp.contents() {
                if let Some(k) = obj.key()
                    && let Some(hash) = blob_hash_from_key(k)
                {
                    out.push(blob_entry_from_listing(k, hash, obj.last_modified())?);
                }
            }
            match next_continuation(resp.is_truncated(), resp.next_continuation_token()) {
                Some(token) => continuation = Some(token),
                None => break,
            }
        }
        Ok(out)
    }

    async fn delete_blob(&self, hash: &ContentHash) -> Result<()> {
        let key = format!("{BLOB_PREFIX}{}", hash.0);
        self.client
            .delete_object()
            .bucket(&self.bucket)
            .key(&key)
            .send()
            .await
            .map_err(|e| CoreError::Backend(e.into_service_error().to_string()))?;
        Ok(()) // S3 delete of an absent key succeeds — idempotent
    }

    async fn has_blob(&self, hash: &ContentHash) -> Result<bool> {
        let key = format!("{BLOB_PREFIX}{}", hash.0);
        match self
            .client
            .head_object()
            .bucket(&self.bucket)
            .key(&key)
            .send()
            .await
        {
            Ok(_) => Ok(true),
            Err(e) => {
                // Read the HTTP status before `into_service_error` drops the raw
                // response; it is the most diagnosable part of a 403 (#330).
                let status = e.raw_response().map(|r| r.status().as_u16());
                let svc = e.into_service_error();
                if head_error_is_absent(&svc) {
                    Ok(false)
                } else {
                    Err(CoreError::Backend(format!(
                        "head blob {}: {}",
                        hash.0,
                        describe_head_error(&svc, status)
                    )))
                }
            }
        }
    }
}

/// Whether a `HeadObject` error code means "no such blob" as opposed to a fault
/// worth surfacing. S3 answers a missing key with 404 `NotFound`; anything else
/// is a real error and must not be reported as absence.
fn head_means_absent(code: &str) -> bool {
    matches!(code, "NotFound" | "NoSuchKey")
}

/// Whether a `HeadObject` service error means "no such blob".
///
/// Split out of [`S3Store::has_blob`] so the *combination* of the two checks is
/// testable, not just the string half (gonzalo#330). The earlier unit test only
/// covered [`head_means_absent`], so an edit that mapped `AccessDenied` or a 500
/// to absence would have passed the suite untouched.
///
/// Getting this wrong is asymmetric. Reporting a present blob as absent only
/// makes a writer re-upload it, never delete it. Reporting a *fault* as absence
/// is worse: it hides an outage behind a plausible answer, and a GC run that
/// believes a blob is gone will not re-pin it.
fn head_error_is_absent(e: &HeadObjectError) -> bool {
    e.is_not_found() || head_means_absent(e.code().unwrap_or_default())
}

/// A `HeadObject` failure, described for whoever reads the log.
///
/// `HeadObjectError`'s own `Display` renders the `Unhandled` variant as
/// "unhandled error", which tells an operator nothing about a 403 (gonzalo#330).
/// Lead with the service's error code and the HTTP status, so the common
/// misconfigurations — wrong credentials, wrong bucket policy — are readable
/// from the message alone.
fn describe_head_error(e: &HeadObjectError, status: Option<u16>) -> String {
    let code = e.code().unwrap_or("unknown code");
    let mut out = match status {
        Some(status) => format!("{code} (HTTP {status})"),
        None => code.to_string(),
    };
    match e.message() {
        Some(message) => {
            out.push_str(": ");
            out.push_str(message);
        }
        // No service message: fall back to the SDK's own rendering rather than
        // stopping at a bare code.
        None => out.push_str(&format!(": {e}")),
    }
    out
}

/// Map one listed blob object to a [`BlobEntry`].
///
/// Split out of `list_blobs` so its two failure branches are reachable from a
/// unit test (gonzalo#330): the crate has no fake S3 client and its integration
/// tests need a live endpoint, so otherwise only CI's HA soak exercises this code
/// at all. Both branches fail the whole listing loudly, which is right — a
/// listing we cannot date is not one GC should sweep from. Only the
/// missing-timestamp branch is testable; see the note on the conversion below.
fn blob_entry_from_listing(
    key: &str,
    hash: ContentHash,
    last_modified: Option<&DateTime>,
) -> Result<BlobEntry> {
    // ListObjectsV2 always carries LastModified; a response without one is a
    // malformed listing, not a blob we can reason about the age of.
    let dt = last_modified
        .ok_or_else(|| CoreError::Backend(format!("blob {key} listed without a LastModified")))?;
    // Unreachable on a 64-bit target, and deliberately kept anyway.
    // `SystemTime::try_from(DateTime)` fails only if the offset from the epoch
    // overflows `SystemTime`, and on a 64-bit `timespec` platform both extremes
    // of the `i64` seconds a `DateTime` can hold convert cleanly -- measured on
    // the Linux builder: `i64::MAX` yields an entry dated `i64::MAX` ms and
    // `i64::MIN` one dated `-i64::MAX` ms, neither an error. So no unit test
    // covers this arm; a 32-bit target is where it earns its keep (gonzalo#330).
    let modified = SystemTime::try_from(*dt).map_err(|e| {
        CoreError::Backend(format!(
            "blob {key} has an unrepresentable LastModified: {e}"
        ))
    })?;
    Ok(BlobEntry::from_system_time(hash, modified))
}

/// Parse a blob object key `blobs/<hash>` back into a [`ContentHash`]. Returns
/// `None` for anything that isn't exactly one segment under `blobs/` — so a
/// record object that happens to live in a `blobs` namespace
/// (`blobs/<col>/<id>.json`, which still has a `/`) is never mistaken for a blob.
fn blob_hash_from_key(key: &str) -> Option<ContentHash> {
    let rest = key.strip_prefix(BLOB_PREFIX)?;
    if rest.is_empty() || rest.contains('/') || rest.contains('.') {
        return None;
    }
    Some(ContentHash(rest.to_string()))
}

/// Parse `namespace/collection/id.json` back into a `RecordKey`, decoding each
/// component (the exact inverse of `object_key`). Returns `None` for objects
/// that don't match the expected three-part `.json` shape. Since every literal
/// `/` in a component is escaped, splitting on `/` always yields exactly the
/// three separators' worth of parts.
fn parse_object_key(s: &str) -> Option<RecordKey> {
    let rest = s.strip_suffix(".json")?;
    let parts: Vec<&str> = rest.split('/').collect();
    if parts.len() == 3 {
        Some(RecordKey::new(
            decode_segment(parts[0]),
            decode_segment(parts[1]),
            decode_segment(parts[2]),
        ))
    } else {
        None
    }
}

/// Suffix appended to a record's object key to form its tombstone marker
/// (ADR 0025). `segment()` escapes `.`, so no encoded id can end in this and
/// the two key spaces are disjoint by construction.
const MARKER_SUFFIX: &str = ".tombstone";

/// The marker object key for `key`: `namespace/collection/id.json.tombstone`.
///
/// An object here means "the record at this key may be a tombstone —
/// read it to find out". Its *absence* is only meaningful in a collection
/// carrying [`marked_flag_key`], since a bucket written before ADR 0025 has
/// tombstones with no marker at all.
fn marker_key(key: &RecordKey) -> String {
    format!("{}{MARKER_SUFFIX}", object_key(key))
}

/// The record a marker object belongs to, or `None` if `s` is not a marker.
fn parse_marker_key(s: &str) -> Option<RecordKey> {
    parse_object_key(s.strip_suffix(MARKER_SUFFIX)?)
}

/// The per-collection flag object key. Its presence says every tombstone in
/// this collection carries a marker, so `list` may treat an unmarked key as
/// live. The name holds no `.` and does not end in `.json`, so
/// [`parse_object_key`] can never produce it from a caller's record key.
fn marked_flag_key(namespace: &str, collection: &str) -> String {
    format!(
        "{}/{}/_tombstone_markers",
        gonzalo_core::segment(namespace),
        gonzalo_core::segment(collection)
    )
}

#[cfg(test)]
mod tests {
    use super::*;

    // Review Focus 5. A 403, a 500 or throttling must surface as Err. Reporting
    // `false` would make a writer re-upload on every transient fault and hide a
    // broken store behind what looks like a missing blob.
    #[test]
    fn a_head_error_other_than_not_found_is_not_absence() {
        assert!(!head_means_absent("AccessDenied"));
        assert!(!head_means_absent("InternalError"));
        assert!(head_means_absent("NotFound"));
    }

    /// A `HeadObjectError` carrying `code`, as the SDK builds an unmodelled
    /// service error. `generic` is the sanctioned constructor; matching the
    /// `Unhandled` variant directly is deprecated.
    fn head_error(code: &str) -> HeadObjectError {
        HeadObjectError::generic(
            aws_sdk_s3::error::ErrorMetadata::builder()
                .code(code)
                .message("from the service")
                .build(),
        )
    }

    // gonzalo#330: the test above only covers the string half of `has_blob`'s
    // decision. This one covers the predicate `has_blob` actually calls, so an
    // edit that mapped a fault to `Ok(false)` fails here.
    #[test]
    fn only_a_not_found_head_error_is_absence() {
        // The modelled 404, which `is_not_found()` answers.
        assert!(head_error_is_absent(&HeadObjectError::NotFound(
            aws_sdk_s3::types::error::NotFound::builder().build()
        )));
        // An unmodelled error whose code still means absence, which only the
        // `head_means_absent` arm catches.
        assert!(head_error_is_absent(&head_error("NoSuchKey")));

        // Faults. None of these may be reported as absence.
        for code in ["AccessDenied", "InternalError", "SlowDown", ""] {
            assert!(
                !head_error_is_absent(&head_error(code)),
                "code {code:?} must not read as absence"
            );
        }
    }

    // gonzalo#330: `HeadObjectError`'s own Display renders an unmodelled error
    // as "unhandled error", which does not tell an operator a 403 happened.
    #[test]
    fn a_head_fault_is_described_with_its_code_and_status() {
        let msg = describe_head_error(&head_error("AccessDenied"), Some(403));
        assert!(msg.contains("AccessDenied"), "{msg}");
        assert!(msg.contains("403"), "{msg}");
        assert!(msg.contains("from the service"), "{msg}");

        // No status available (a response error, not a service error).
        let msg = describe_head_error(&head_error("InternalError"), None);
        assert!(msg.contains("InternalError"), "{msg}");
        assert!(!msg.contains("HTTP"), "no status should be invented: {msg}");
    }

    // gonzalo#330: `list_blobs`'s two failure branches. Neither was reachable in
    // a unit test before, and only CI's HA soak touches `list_blobs` at all.
    #[test]
    fn a_listed_blob_without_a_last_modified_fails_the_listing() {
        let err = blob_entry_from_listing("blobs/abc", ContentHash("abc".into()), None)
            .expect_err("a listing we cannot date must not be silently dropped");
        let msg = err.to_string();
        assert!(msg.contains("blobs/abc"), "{msg}");
        assert!(msg.contains("without a LastModified"), "{msg}");
    }

    #[test]
    fn a_well_formed_listed_blob_maps_to_an_entry() {
        let dt = DateTime::from_secs(1_700_000_000);
        let entry = blob_entry_from_listing("blobs/ok", ContentHash("ok".into()), Some(&dt))
            .expect("a dated blob maps cleanly");
        assert_eq!(entry.hash, ContentHash("ok".into()));
    }

    #[test]
    fn marker_key_is_the_record_key_plus_a_suffix() {
        let k = RecordKey::new("ns", "col", "id");
        assert_eq!(marker_key(&k), "ns/col/id.json.tombstone");
        assert_eq!(parse_marker_key(&marker_key(&k)), Some(k));
    }

    #[test]
    fn a_marker_key_is_not_a_record_key() {
        // `segment()` escapes `.`, so no id can produce a key ending in
        // `.json.tombstone` — the two key spaces are disjoint by construction.
        let k = RecordKey::new("ns", "col", "id");
        assert_eq!(parse_object_key(&marker_key(&k)), None);
        for id in ["id.json.tombstone", "id.json", "a.b", "..", "50%"] {
            let key = RecordKey::new("ns", "col", id);
            assert_ne!(object_key(&key), marker_key(&k));
            assert_eq!(parse_marker_key(&object_key(&key)), None);
        }
    }

    #[test]
    fn marked_flag_key_is_not_a_record_or_marker_key() {
        let flag = marked_flag_key("ns", "col");
        assert_eq!(flag, "ns/col/_tombstone_markers");
        assert_eq!(parse_object_key(&flag), None);
        assert_eq!(parse_marker_key(&flag), None);
    }

    #[test]
    fn marked_flag_key_escapes_its_components() {
        // Same encoding as record keys, so a namespace containing `/` cannot
        // reach another collection's flag.
        assert_eq!(
            marked_flag_key("a/b", "c.d"),
            "a%2Fb/c%2Ed/_tombstone_markers"
        );
    }

    #[test]
    fn parse_roundtrips_object_key() {
        let k = RecordKey::new("ns", "col", "id");
        assert_eq!(parse_object_key(&object_key(&k)), Some(k));
    }

    #[test]
    fn parse_roundtrips_special_char_keys() {
        // Keys with `.`, `/`, spaces, and `%` must survive the object-key
        // round-trip and stay distinct (no collision onto one object).
        for k in [
            RecordKey::new("a/b", "c.d", "e/f"),
            RecordKey::new("ns", "col", "v1.0"),
            RecordKey::new("ns", "col", "v1_0"),
            RecordKey::new("50% off", "café", "🚀"),
        ] {
            assert_eq!(parse_object_key(&object_key(&k)), Some(k));
        }
        assert_ne!(
            object_key(&RecordKey::new("ns", "col", "v1.0")),
            object_key(&RecordKey::new("ns", "col", "v1_0")),
        );
    }

    #[test]
    fn parse_rejects_non_json_or_wrong_depth() {
        assert_eq!(parse_object_key("a/b/c.txt"), None);
        assert_eq!(parse_object_key("a/b.json"), None);
        assert_eq!(parse_object_key("a/b/c/d.json"), None);
    }

    use gonzalo_core::store::Conflict;
    use gonzalo_core::{Body, Identity, Meta, RecordKind, plan_put, tombstone_of};
    use std::cell::Cell;
    use std::collections::BTreeMap;

    fn live(key: &RecordKey, payload: &[u8]) -> Record {
        Record {
            key: key.clone(),
            kind: RecordKind::Topic,
            revision: Revision::initial(payload),
            parent: None,
            body: Body::Inline(payload.to_vec()),
            meta: Meta {
                author: Identity::new("tester"),
                origin_system: "test".into(),
                created: 0,
                updated: 0,
                labels: BTreeMap::new(),
            },
            links: Vec::new(),
            ancestors: Vec::new(),
            deleted_at: None,
            deleted_blob: None,
            deleted_kind: None,
        }
    }

    fn tomb(key: &RecordKey, payload: &[u8]) -> Record {
        tombstone_of(&live(key, payload), 1_000, DEFAULT_ANCESTOR_CAP, None)
    }

    // ---- hardening before other S3 backends are qualified (#286) ----

    #[test]
    fn an_ambiguous_commit_is_recognised_as_our_own_write() {
        // The write applied, the response timed out, the retry saw 412. The
        // re-read returns exactly what we sent, so the attempt succeeded.
        let k = RecordKey::new("ns", "col", "ambiguous");
        let sent = live(&k, b"v1");
        assert!(own_write_landed(&sent, Some(&sent)));

        // Someone else's write, or nothing at all, is not ours.
        assert!(!own_write_landed(&sent, Some(&live(&k, b"someone-else"))));
        assert!(!own_write_landed(&sent, Some(&tomb(&k, b"v1"))));
        assert!(!own_write_landed(&sent, None));
    }

    #[test]
    fn inflight_backoff_grows_and_stays_bounded() {
        // Each round waits at least as long as the one before, and never more
        // than twice the cap, so eight attempts cannot stall a caller.
        let mut previous = std::time::Duration::ZERO;
        for round in 0..MAX_WRITE_ATTEMPTS {
            let wait = inflight_backoff(round);
            assert!(wait >= std::time::Duration::from_millis(4), "round {round}");
            assert!(
                wait < std::time::Duration::from_millis(500),
                "round {round}"
            );
            if round > 0 {
                assert!(
                    wait >= previous / 2,
                    "round {round} fell far below the previous wait"
                );
            }
            previous = wait;
        }
    }

    fn conflict(key: &RecordKey) -> Box<Conflict> {
        Box::new(Conflict {
            key: key.clone(),
            expected: Some(Revision::initial(b"stale")),
            current: live(key, b"current"),
        })
    }

    /// An `S3Store` that never touches the network: building a client does no
    /// I/O, so the builder can be unit-tested without an endpoint.
    fn offline_store() -> S3Store {
        let conf = aws_sdk_s3::Config::builder()
            .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest())
            .region(aws_sdk_s3::config::Region::new("us-east-1"))
            .build();
        S3Store::new(Client::from_conf(conf), "offline")
    }

    // ---- preconditions ----

    #[test]
    fn create_uses_if_absent() {
        // No object was read → create-only.
        assert_eq!(precondition(None), Precondition::IfAbsent);
    }

    #[test]
    fn update_uses_if_match_on_the_read_etag() {
        assert_eq!(
            precondition(Some("\"abc123\"")),
            Precondition::IfMatch("\"abc123\"".to_string())
        );
    }

    #[test]
    fn recreation_over_tombstone_uses_if_match() {
        // `put(record, None)` over a tombstone is a recreation the planner
        // writes. The tombstone object exists, so the write must be `If-Match`
        // on its ETag. `If-None-Match: *` would 412 on every attempt.
        let k = RecordKey::new("ns", "col", "recreate");
        let plan = plan_put(
            Some(&tomb(&k, b"old")),
            live(&k, b"new"),
            None,
            now_ms(),
            DEFAULT_ANCESTOR_CAP,
        );
        assert!(matches!(plan, PutPlan::Write(_)));
        assert_eq!(
            precondition(Some("\"tomb-etag\"")),
            Precondition::IfMatch("\"tomb-etag\"".to_string())
        );
    }

    // ---- consumer read filtering ----

    #[test]
    fn visible_hides_tombstones_and_keeps_live() {
        let k = RecordKey::new("ns", "col", "vis");
        assert_eq!(visible(None), None);
        assert_eq!(visible(Some(tomb(&k, b"x"))), None);
        let rec = live(&k, b"x");
        assert_eq!(visible(Some(rec.clone())), Some(rec));
    }

    #[test]
    fn list_filter_excludes_tombstones_and_vanished_keys() {
        let k = RecordKey::new("ns", "col", "listed");
        assert_eq!(
            listed_as_live(Ok(Some(live(&k, b"x")))).unwrap(),
            Liveness::Live
        );
        // A tombstone reports the revision its marker names, so a listing that
        // finds one unmarked can repair it without a second read.
        let t = tomb(&k, b"x");
        assert_eq!(
            listed_as_live(Ok(Some(t.clone()))).unwrap(),
            Liveness::Tombstoned(t.revision)
        );
        // Purged between the listing and the read.
        assert_eq!(listed_as_live(Ok(None)).unwrap(), Liveness::Absent);
    }

    #[test]
    fn list_filter_keeps_undecodable_objects_and_propagates_backend_errors() {
        // An object that fails to decode stays listed; `get` surfaces the error.
        // Any other read error fails the list (s3 rule; fs and git keep such
        // entries listed).
        assert_eq!(
            listed_as_live(Err(CoreError::Serde("bad json".into()))).unwrap(),
            Liveness::Live
        );
        assert!(matches!(
            listed_as_live(Err(CoreError::Backend("503".into()))),
            Err(CoreError::Backend(_))
        ));
    }

    // ---- plan → S3 action ----

    #[test]
    fn put_step_writes_and_commits_the_planned_revision() {
        let k = RecordKey::new("ns", "col", "put-write");
        let rec = live(&k, b"v");
        match put_step(&k, PutPlan::Write(rec.clone())) {
            Ok(Planned::Put(stored, answer)) => {
                assert_eq!(answer, PutResult::Committed(rec.revision.clone()));
                assert_eq!(stored, rec);
            }
            other => panic!("expected Put, got {other:?}"),
        }
    }

    #[test]
    fn put_step_finishes_on_conflict() {
        let k = RecordKey::new("ns", "col", "put-conflict");
        match put_step(&k, PutPlan::Conflict(conflict(&k))) {
            Ok(Planned::Finish(PutResult::Conflict(c))) => assert_eq!(c, conflict(&k)),
            other => panic!("expected Finish(Conflict), got {other:?}"),
        }
    }

    #[test]
    fn put_step_maps_not_found_to_error() {
        let k = RecordKey::new("ns", "col", "put-missing");
        match put_step(&k, PutPlan::NotFound) {
            Err(CoreError::NotFound(got)) => assert_eq!(got, k),
            other => panic!("expected NotFound, got {other:?}"),
        }
    }

    #[test]
    fn put_step_maps_rejected_to_an_invalid_call() {
        // `Invalid`, not `Backend`: the call can never succeed, and a daemon in
        // front of this store answers 400 rather than 500 (#299).
        let k = RecordKey::new("ns", "col", "put-rejected");
        let reason = gonzalo_core::CONSUMER_TOMBSTONE_REJECTED;
        match put_step(&k, PutPlan::Rejected(reason)) {
            Err(CoreError::Invalid(msg)) => assert_eq!(msg, reason),
            other => panic!("expected Invalid, got {other:?}"),
        }
    }

    #[test]
    fn delete_step_maps_each_plan() {
        let k = RecordKey::new("ns", "col", "del");
        let t = tomb(&k, b"x");
        assert_eq!(
            delete_step(DeletePlan::Write(t.clone())),
            Planned::Put(t, DeleteResult::Deleted)
        );
        assert_eq!(
            delete_step(DeletePlan::Noop),
            Planned::Finish(DeleteResult::Deleted)
        );
        assert_eq!(
            delete_step(DeletePlan::Conflict(conflict(&k))),
            Planned::Finish(DeleteResult::Conflict(conflict(&k)))
        );
    }

    #[test]
    fn purge_step_maps_each_plan() {
        let k = RecordKey::new("ns", "col", "purge");
        assert_eq!(
            purge_step(PurgePlan::Remove),
            Planned::Remove(DeleteResult::Deleted)
        );
        assert_eq!(
            purge_step(PurgePlan::Noop),
            Planned::Finish(DeleteResult::Deleted)
        );
        assert_eq!(
            purge_step(PurgePlan::Conflict(conflict(&k))),
            Planned::Finish(DeleteResult::Conflict(conflict(&k)))
        );
    }

    // ---- lost-race classification ----

    #[test]
    fn lost_race_covers_412_409_and_vanished_objects() {
        // 409 is the one that must back off; the others are settled races
        // the loop retries immediately (#286).
        assert_eq!(
            race_kind(Some("ConditionalRequestConflict")),
            Some(RaceKind::InFlight)
        );
        assert_eq!(
            race_kind(Some("PreconditionFailed")),
            Some(RaceKind::Settled)
        );
        assert_eq!(race_kind(Some("NoSuchKey")), Some(RaceKind::Settled));
        assert_eq!(race_kind(Some("AccessDenied")), None);
        assert_eq!(race_kind(None), None);
    }

    // ---- lost-race retry loop ----

    #[tokio::test]
    async fn retry_gives_up_after_max_attempts() {
        let k = RecordKey::new("ns", "col", "hot");
        let calls = Cell::new(0usize);
        let calls_ref = &calls;
        let out: Result<()> = retry_on_lost_race(&k, move || async move {
            calls_ref.set(calls_ref.get() + 1);
            Ok(Step::Retry(RaceKind::Settled))
        })
        .await;
        assert_eq!(calls.get(), MAX_WRITE_ATTEMPTS);
        assert_eq!(MAX_WRITE_ATTEMPTS, 8);
        match out {
            Err(CoreError::Backend(msg)) => assert_eq!(
                msg,
                format!("s3: conditional write for {k} lost 8 consecutive races")
            ),
            other => panic!("expected Backend error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn retry_returns_as_soon_as_an_attempt_lands() {
        // Losing every attempt but the last still succeeds, with no extra call.
        let k = RecordKey::new("ns", "col", "warm");
        let calls = Cell::new(0usize);
        let calls_ref = &calls;
        let out = retry_on_lost_race(&k, move || async move {
            calls_ref.set(calls_ref.get() + 1);
            if calls_ref.get() < MAX_WRITE_ATTEMPTS {
                Ok(Step::Retry(RaceKind::Settled))
            } else {
                Ok(Step::Done("landed"))
            }
        })
        .await;
        assert_eq!(out.unwrap(), "landed");
        assert_eq!(calls.get(), MAX_WRITE_ATTEMPTS);
    }

    #[tokio::test]
    async fn retry_propagates_errors_without_retrying() {
        let k = RecordKey::new("ns", "col", "broken");
        let calls = Cell::new(0usize);
        let calls_ref = &calls;
        let out: Result<()> = retry_on_lost_race(&k, move || async move {
            calls_ref.set(calls_ref.get() + 1);
            Err(CoreError::Backend("access denied".into()))
        })
        .await;
        assert!(matches!(out, Err(CoreError::Backend(m)) if m == "access denied"));
        assert_eq!(calls.get(), 1);
    }

    // ---- ancestor cap ----

    #[tokio::test]
    async fn new_store_uses_default_ancestor_cap() {
        assert_eq!(offline_store().cap, DEFAULT_ANCESTOR_CAP);
    }

    #[tokio::test]
    async fn with_ancestor_cap_sets_cap() {
        match offline_store().with_ancestor_cap(3) {
            Ok(store) => assert_eq!(store.cap, 3),
            Err(e) => panic!("cap 3 must be accepted: {e}"),
        }
    }

    #[tokio::test]
    async fn with_ancestor_cap_rejects_zero() {
        assert!(matches!(
            offline_store().with_ancestor_cap(0),
            Err(CoreError::Backend(_))
        ));
    }

    #[test]
    fn next_continuation_terminates_when_token_absent() {
        // The pagination bug: a backend reports more pages but omits the token.
        // Keying off `is_truncated` would loop forever; we must terminate.
        assert_eq!(next_continuation(Some(true), None), None);
        // No token, not truncated → also terminate (the normal last page).
        assert_eq!(next_continuation(Some(false), None), None);
        assert_eq!(next_continuation(None, None), None);
    }

    #[test]
    fn next_continuation_advances_when_token_present() {
        // A token means fetch the next page, regardless of the flag's value.
        assert_eq!(
            next_continuation(Some(true), Some("t1")),
            Some("t1".to_string())
        );
        assert_eq!(
            next_continuation(Some(false), Some("t2")),
            Some("t2".to_string())
        );
        assert_eq!(next_continuation(None, Some("t3")), Some("t3".to_string()));
    }

    #[test]
    fn precondition_failed_is_classified_by_code() {
        assert!(is_precondition_failed(Some("PreconditionFailed")));
        assert!(!is_precondition_failed(Some("AccessDenied")));
        assert!(!is_precondition_failed(None));
    }

    #[test]
    fn blob_key_roundtrips_and_rejects_records() {
        let h = ContentHash::of(b"slice bytes");
        let key = format!("{BLOB_PREFIX}{}", h.0);
        assert_eq!(blob_hash_from_key(&key), Some(h));
        // A record object under a `blobs` namespace has a nested path + `.json`
        // and must never be read back as a blob hash.
        assert_eq!(blob_hash_from_key("blobs/col/id.json"), None);
        assert_eq!(blob_hash_from_key("ns/col/id.json"), None);
        assert_eq!(blob_hash_from_key("blobs/"), None);
    }
}