objectstore_service/backend/

bigtable.rs

//! BigTable backend for high-volume, low-latency storage of small objects.
//!
//! # Row Format
//!
//! Object row keys use the object's storage path. An object row contains either an **object** or a
//! **tombstone** — never both. The two layouts are mutually exclusive and distinguished by
//! column presence:
//!
//! | Column | Family    | Content                     | Present when       |
//! |--------|-----------|-----------------------------|--------------------|
//! | `p`    | `fg`/`fm` | Compressed payload bytes    | Object row only    |
//! | `m`    | `fg`/`fm` | [`Metadata`] JSON           | Object row only    |
//! | `r`    | `fg`/`fm` | Redirect path to LT storage | Tombstone row only |
//!
//! The `r` column signals a tombstone row: its **value** is the long-term `ObjectId`
//! serialized via `as_storage_path()`. Callers can resolve the LT object directly from the
//! `r` value without reconstructing it from the row key. Its column family and timestamp
//! carry the tombstone's concrete expiration deadline.
//!
//! `p`/`m` and `r` are mutually exclusive. Every write begins with a `DeleteFromRow`
//! mutation that clears all columns before writing the new cells, so mixed rows cannot exist.
//!
//! Upload markers use a separate `uploads` namespace and contain only a one-byte
//! `u` cell in `fg`, timestamped with the upload deadline.
//!
//! ## Legacy Tombstone Format
//!
//! Tombstones written before the `r` column layout used the object-row format with an
//! empty `p` column and `"is_redirect_tombstone": true` in the `m` JSON. Both formats are
//! supported for reading. A `bigtable.legacy_tombstone_read` metric is emitted on each legacy
//! read. Legacy tombstones expire naturally by TTL/GC; a successful conditional
//! expiry extension upgrades them to the `r` format. Tombstone metadata in the historical `t`
//! column is ignored; the corresponding `r` cell contains all information needed by readers.

use std::fmt;
use std::future::Future;
use std::sync::Arc;
use std::time::Duration;

use bigtable_rs::bigtable::{BigTableConnection, Error as BigTableError, RowCell};
use bigtable_rs::google::bigtable::v2::{self, mutation};
use bytes::Bytes;
use futures_util::TryStreamExt;
use objectstore_types::metadata::Metadata;
use objectstore_types::range::{ByteRange, ContentRange};
use objectstore_types::time::Timestamp;
use serde::{Deserialize, Serialize};
use tonic::Code;
use tracing::Instrument;

use crate::backend::common::{
    self, Backend, DeleteResponse, ExpiryUpdate, GetResponse, HighVolumeBackend, MetadataResponse,
    PutResponse, SetExpiryResponse, TieredGet, TieredMetadata, TieredUpdate, TieredWrite,
    Tombstone,
};
use crate::change_stream::{
    ChangeStream, ChangeStreamFactory, CostTrackerStreamConfig, flush_change_stream,
};
use crate::error::{Error, ErrorKind, Result, ResultExt as _};
use crate::gcp_auth::PrefetchingTokenProvider;
use crate::id::ObjectId;
use crate::stream::{ChunkedBytes, ClientStream};

/// Configuration for [`BigTableBackend`].
///
/// Stores objects in [Google Cloud Bigtable], a NoSQL wide-column database optimized for
/// high-throughput, low-latency workloads with small objects. Authentication uses Application
/// Default Credentials (ADC).
///
/// **Note**: The table must be pre-created with the following column families:
/// - `fg`: timestamp-based garbage collection (`maxage=1s`)
/// - `fm`: manual garbage collection (`no GC policy`)
///
/// [Google Cloud Bigtable]: https://cloud.google.com/bigtable
///
/// # Example
///
/// ```yaml
/// storage:
///   type: bigtable
///   project_id: my-project
///   instance_name: objectstore
///   table_name: objectstore
/// ```
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct BigTableConfig {
    /// Optional custom Bigtable endpoint.
    ///
    /// Useful for testing with emulators. If `None`, uses the default Bigtable endpoint.
    ///
    /// # Default
    ///
    /// `None` (uses default Bigtable endpoint)
    ///
    /// # Environment Variables
    ///
    /// - `OS__STORAGE__TYPE=bigtable`
    /// - `OS__STORAGE__ENDPOINT=localhost:8086` (optional)
    pub endpoint: Option<String>,

    /// GCP project ID.
    ///
    /// The Google project ID (not project number) containing the Bigtable instance.
    ///
    /// # Environment Variables
    ///
    /// - `OS__STORAGE__PROJECT_ID=my-project`
    pub project_id: String,

    /// Bigtable instance name.
    ///
    /// # Environment Variables
    ///
    /// - `OS__STORAGE__INSTANCE_NAME=my-instance`
    pub instance_name: String,

    /// Bigtable table name.
    ///
    /// The table must exist before starting the server.
    ///
    /// # Environment Variables
    ///
    /// - `OS__STORAGE__TABLE_NAME=objectstore`
    pub table_name: String,

    /// Optional number of connections to maintain to Bigtable.
    ///
    /// # Default
    ///
    /// `None` (defaults to 1)
    ///
    /// # Environment Variables
    ///
    /// - `OS__STORAGE__CONNECTIONS=16` (optional)
    pub connections: Option<usize>,

    /// Timeout for an individual Bigtable RPC attempt.
    ///
    /// # Default
    ///
    /// `2s`
    ///
    /// # Environment Variables
    ///
    /// - `OS__STORAGE__RPC_TIMEOUT=2s`
    /// - `OS__STORAGE__HIGH_VOLUME__RPC_TIMEOUT=2s` (tiered storage)
    #[serde(default = "default_rpc_timeout", with = "humantime_serde")]
    pub rpc_timeout: Duration,

    /// Reports what this backend stores, for per-usecase cost attribution.
    ///
    /// # Default
    ///
    /// `None`, which disables reporting for this backend.
    ///
    /// # Environment Variables
    ///
    /// - `OS__STORAGE__COGS__SHARED_RESOURCE_ID=bigtable_objectstore`
    /// - `OS__STORAGE__COGS__SAMPLE_RATE=1.0` (optional)
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub cogs: Option<CostTrackerStreamConfig>,
}

fn default_rpc_timeout() -> Duration {
    Duration::from_secs(2)
}

/// Maximum age for connections (GRPC channels) to Bigtable, after which they will be swapped with
/// new ones in the background.
/// This is intended to avoid latency spikes that could occur every hour or so, when the server
/// closes long standing connections ([source](https://web.archive.org/web/20260211140930/https://docs.cloud.google.com/bigtable/docs/performance#cold-starts:~:text=return%20an%20error.-,Cold%20start,-at%20client%20initialization)).
/// `tonic` already handles reconnections transparently, but lazily, meaning that the first requests
/// that attempt to use a certain channel after the server has closed it will pay the cost of the
/// reconnection, resulting in increased latency for those requests.
const MAX_CHANNEL_AGE: Option<Duration> = Some(Duration::from_mins(50));
/// Permission scopes required for accessing the BigTable data API.
const TOKEN_SCOPES: &[&str] = &["https://www.googleapis.com/auth/bigtable.data"];

/// How often to retry failed requests.
const REQUEST_RETRY_COUNT: usize = 2;
/// How many times to retry a CAS mutation before giving up and returning an error.
const CAS_RETRY_COUNT: usize = 3;

/// Column that stores the raw payload (compressed).
const COLUMN_PAYLOAD: &[u8] = b"p";
/// Column that stores metadata in JSON.
const COLUMN_METADATA: &[u8] = b"m";
/// Column that stores the redirect path for tombstone rows.
const COLUMN_REDIRECT: &[u8] = b"r";
/// Regex to match all non-payload columns (`m`, `r`) for metadata-only reads.
const FILTER_META: &[u8] = b"^[mr]$";

/// Column which marks the presence of an ongoing resumable upload.
const COLUMN_UPLOAD: &[u8] = b"u";

/// Column family that uses timestamp-based garbage collection.
///
/// We require a GC rule on this family to automatically delete rows.
/// See: <https://cloud.google.com/bigtable/docs/gc-cell-level>
const FAMILY_GC: &str = "fg";
/// Column family that uses manual garbage collection.
const FAMILY_MANUAL: &str = "fm";

/// BigTable storage backend for high-volume, low-latency object storage.
pub struct BigTableBackend {
    bigtable: BigTableConnection,

    instance_path: String,
    table_path: String,
    table_name: String,

    change_stream: Arc<dyn ChangeStream>,
}

impl fmt::Debug for BigTableBackend {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("BigTableBackend")
            .field("instance_path", &self.instance_path)
            .field("table_path", &self.table_path)
            .field("table_name", &self.table_name)
            .finish_non_exhaustive()
    }
}

/// Creates a row filter that matches a single column by exact qualifier.
fn column_filter(column: &[u8]) -> v2::RowFilter {
    v2::RowFilter {
        filter: Some(v2::row_filter::Filter::ColumnQualifierRegexFilter(
            [b"^", column, b"$"].concat(),
        )),
    }
}

/// Creates a row filter matching the legacy tombstone format: `m` column JSON starts with
/// `{"is_redirect_tombstone":true`.
///
/// After legacy tombstones expire naturally this filter becomes dead code in both callers.
fn legacy_tombstone_filter() -> v2::RowFilter {
    v2::RowFilter {
        filter: Some(v2::row_filter::Filter::Chain(v2::row_filter::Chain {
            filters: vec![
                column_filter(COLUMN_METADATA),
                v2::RowFilter {
                    filter: Some(v2::row_filter::Filter::ValueRegexFilter(
                        b"^\\{\"is_redirect_tombstone\":true[,}].*".to_vec(),
                    )),
                },
            ],
        })),
    }
}

/// Wraps `inner` so that it only matches live (non-expired) cells.
///
/// Uses the rounded access time directly. Legacy fractional cell timestamps can be excluded
/// up to one second before their rounded metadata deadline; new writes are second-aligned.
fn live_row_filter(inner: v2::RowFilter, now: Timestamp) -> v2::RowFilter {
    v2::RowFilter {
        filter: Some(v2::row_filter::Filter::Interleave(
            v2::row_filter::Interleave {
                filters: vec![
                    // Manual family: never expires.
                    v2::RowFilter {
                        filter: Some(v2::row_filter::Filter::Chain(v2::row_filter::Chain {
                            filters: vec![
                                v2::RowFilter {
                                    filter: Some(v2::row_filter::Filter::FamilyNameRegexFilter(
                                        format!("^{FAMILY_MANUAL}$"),
                                    )),
                                },
                                inner.clone(),
                            ],
                        })),
                    },
                    // GC family: only match non-expired cells.
                    v2::RowFilter {
                        filter: Some(v2::row_filter::Filter::Chain(v2::row_filter::Chain {
                            filters: vec![
                                v2::RowFilter {
                                    filter: Some(v2::row_filter::Filter::FamilyNameRegexFilter(
                                        format!("^{FAMILY_GC}$"),
                                    )),
                                },
                                v2::RowFilter {
                                    filter: Some(v2::row_filter::Filter::TimestampRangeFilter(
                                        v2::TimestampRange {
                                            start_timestamp_micros: now.as_micros() as i64,
                                            end_timestamp_micros: 0,
                                        },
                                    )),
                                },
                                inner,
                            ],
                        })),
                    },
                ],
            },
        )),
    }
}

/// Builds a raw row filter that matches any live tombstone row, new- or legacy-format.
///
/// New format: presence of the `r` column.
/// Legacy format: `is_redirect_tombstone: true` in the `m` column JSON.
///
/// After legacy tombstones expire naturally this simplifies to just
/// `column_filter(COLUMN_REDIRECT)`.
fn tombstone_filter(access_time: Timestamp) -> v2::RowFilter {
    let filter = v2::RowFilter {
        filter: Some(v2::row_filter::Filter::Interleave(
            v2::row_filter::Interleave {
                filters: vec![column_filter(COLUMN_REDIRECT), legacy_tombstone_filter()],
            },
        )),
    };
    live_row_filter(filter, access_time)
}

/// Returns a [`MutatePredicate`] that matches any live tombstone row.
///
/// Mutations will not run on live tombstones (`predicate_matched == false`). They _will_
/// run on expired tombstones as well as non-tombstones. Used by
/// [`BigTableBackend::put_non_tombstone`] and [`BigTableBackend::compare_and_write`] as
/// the `CheckAndMutateRow` predicate.
///
/// This predicate cannot distinguish an empty row from a row holding a regular object; a caller
/// that needs to know whether its mutation hit anything wants [`non_tombstone_predicate`].
fn tombstone_predicate(access_time: Timestamp) -> MutatePredicate {
    MutatePredicate::Exclude(tombstone_filter(access_time))
}

/// Returns a [`MutatePredicate`] that is the logical negation of [`tombstone_predicate`];
/// it matches everything _except_ live tombstones.
///
/// Mutations run only when the predicate matches (`predicate_matched == true`). They will
/// run on expired tombstones as well as non-tombstones. The match result doubles as a
/// "was a row removed?" signal. Used by [`BigTableBackend::delete_non_tombstone`] as the
/// `CheckAndMutateRow` predicate.
///
/// Built as a `Condition` filter:
/// - Predicate: [`tombstone_filter`] -> is a live tombstone present?
/// - True branch: `BlockAllFilter` -> match nothing; live tombstones should be preserved
/// - False branch: `PassAllFilter` -> match everything: expired tombstones and non-tombstones
fn non_tombstone_predicate(access_time: Timestamp) -> MutatePredicate {
    MutatePredicate::Include(v2::RowFilter {
        filter: Some(v2::row_filter::Filter::Condition(Box::new(
            v2::row_filter::Condition {
                predicate_filter: Some(Box::new(tombstone_filter(access_time))),
                true_filter: Some(Box::new(v2::RowFilter {
                    filter: Some(v2::row_filter::Filter::BlockAllFilter(true)),
                })),
                false_filter: Some(Box::new(v2::RowFilter {
                    filter: Some(v2::row_filter::Filter::PassAllFilter(true)),
                })),
            },
        ))),
    })
}

/// Builds an anchored regex pattern (`^…$`) that matches `value` literally.
///
/// Uses [`regex::escape`] so that metacharacters in storage paths (`.`, `/`, etc.)
/// are treated as literal bytes.
fn exact_value_regex(value: &str) -> Vec<u8> {
    format!("^{}$", regex::escape(value)).into_bytes()
}

/// Matches tombstones whose redirect resolves to `target`.
///
/// ## Predicate Matches
///
/// Must be used with `true_mutations` and `predicate_matched == true`.
///
/// ## Details
///
/// Always includes an exact match on the `r` (redirect) column:
/// - Chain: `r` column present AND value == `target` storage path
///
/// When `target == own_id` (the caller expects a legacy identity redirect), the
/// exact match is wrapped in an Interleave with two additional fallbacks:
/// - Chain: `r` column present AND value == `b""` (empty-sentinel written before the redirect
///   column stored the path)
/// - Chain: `m` column present AND value matches `{"is_redirect_tombstone":true...}` regex
///   (legacy metadata format predating the dedicated `r` column)
fn redirect_target_filter(
    target: &ObjectId,
    own_id: &ObjectId,
    access_time: Timestamp,
) -> v2::RowFilter {
    let target_path = exact_value_regex(&target.as_storage_path().to_string());

    let exact_match = v2::RowFilter {
        filter: Some(v2::row_filter::Filter::Chain(v2::row_filter::Chain {
            filters: vec![
                column_filter(COLUMN_REDIRECT),
                v2::RowFilter {
                    filter: Some(v2::row_filter::Filter::ValueRegexFilter(target_path)),
                },
            ],
        })),
    };

    if target != own_id {
        return live_row_filter(exact_match, access_time);
    }

    let empty_redirect_match = v2::RowFilter {
        filter: Some(v2::row_filter::Filter::Chain(v2::row_filter::Chain {
            filters: vec![
                column_filter(COLUMN_REDIRECT),
                v2::RowFilter {
                    filter: Some(v2::row_filter::Filter::ValueRegexFilter(b"^$".to_vec())),
                },
            ],
        })),
    };

    // Also match legacy tombstones that resolve to the HV id:
    // - empty `r` value (written before the redirect column stored the path)
    // - legacy `m` column format (`is_redirect_tombstone: true`)
    let filter = v2::RowFilter {
        filter: Some(v2::row_filter::Filter::Interleave(
            v2::row_filter::Interleave {
                filters: vec![exact_match, empty_redirect_match, legacy_tombstone_filter()],
            },
        )),
    };
    live_row_filter(filter, access_time)
}

/// Returns a [`MutatePredicate`] that matches tombstones whose redirect resolves to either `old` or `new`.
///
/// Mutations run only when the predicate matches (`predicate_matched == true`):
/// equivalent to `t == old || t == new`. Built as an Interleave of two
/// [`redirect_target_filter`] calls — yields cells iff at least one branch matches.
/// An absent row or non-tombstone row yields 0 cells, so `predicate_matched = false` (conflict).
fn update_predicate(
    old: &ObjectId,
    new: &ObjectId,
    own_id: &ObjectId,
    access_time: Timestamp,
) -> MutatePredicate {
    MutatePredicate::Include(v2::RowFilter {
        filter: Some(v2::row_filter::Filter::Interleave(
            v2::row_filter::Interleave {
                filters: vec![
                    redirect_target_filter(old, own_id, access_time),
                    redirect_target_filter(new, own_id, access_time),
                ],
            },
        )),
    })
}

/// Returns a [`MutatePredicate`] that matches rows where no conflicting tombstone exists.
///
/// Mutations run only when the row is conflict-free (`predicate_matched == false`):
/// no tombstone is present, or the tombstone's redirect already points to `target`.
///
/// Built as an inverted `Condition` filter:
/// - Predicate: [`redirect_target_filter`]`(target)` — tombstone already points to `target`?
/// - True branch: `BlockAllFilter` → 0 cells (already at target, safe state).
/// - False branch: [`tombstone_filter`] → 0 cells when no tombstone exists.
///
/// Both safe states yield 0 cells, so `predicate_matched = false` in both cases.
fn optional_target_predicate(
    target: &ObjectId,
    own_id: &ObjectId,
    access_time: Timestamp,
) -> MutatePredicate {
    MutatePredicate::Exclude(v2::RowFilter {
        filter: Some(v2::row_filter::Filter::Condition(Box::new(
            v2::row_filter::Condition {
                predicate_filter: Some(Box::new(redirect_target_filter(
                    target,
                    own_id,
                    access_time,
                ))),
                true_filter: Some(Box::new(v2::RowFilter {
                    filter: Some(v2::row_filter::Filter::BlockAllFilter(true)),
                })),
                false_filter: Some(Box::new(tombstone_filter(access_time))),
            },
        ))),
    })
}

fn exact_expiry_filter(start: i64) -> Result<v2::RowFilter> {
    let end = start.checked_add(1).ok_or_else(|| {
        Error::new(
            ErrorKind::Internal,
            "building Bigtable expiration predicate",
        )
    })?;
    Ok(v2::RowFilter {
        filter: Some(v2::row_filter::Filter::Chain(v2::row_filter::Chain {
            filters: vec![
                v2::RowFilter {
                    filter: Some(v2::row_filter::Filter::FamilyNameRegexFilter(format!(
                        "^{FAMILY_GC}$"
                    ))),
                },
                v2::RowFilter {
                    filter: Some(v2::row_filter::Filter::TimestampRangeFilter(
                        v2::TimestampRange {
                            start_timestamp_micros: start,
                            end_timestamp_micros: end,
                        },
                    )),
                },
            ],
        })),
    })
}

/// Matches an inline row whose metadata cell has the observed expiry timestamp.
fn inline_expiry_predicate(
    observed_expiry: i64,
    access_time: Timestamp,
) -> Result<MutatePredicate> {
    let inline_at_expiry = v2::RowFilter {
        filter: Some(v2::row_filter::Filter::Chain(v2::row_filter::Chain {
            filters: vec![
                column_filter(COLUMN_METADATA),
                exact_expiry_filter(observed_expiry)?,
            ],
        })),
    };

    Ok(MutatePredicate::Include(v2::RowFilter {
        filter: Some(v2::row_filter::Filter::Condition(Box::new(
            v2::row_filter::Condition {
                predicate_filter: Some(Box::new(tombstone_filter(access_time))),
                true_filter: Some(Box::new(v2::RowFilter {
                    filter: Some(v2::row_filter::Filter::BlockAllFilter(true)),
                })),
                false_filter: Some(Box::new(inline_at_expiry)),
            },
        ))),
    }))
}

fn redirect_expiry_predicate(
    target: &ObjectId,
    own_id: &ObjectId,
    observed_expiry: i64,
    access_time: Timestamp,
) -> Result<MutatePredicate> {
    Ok(MutatePredicate::Include(v2::RowFilter {
        filter: Some(v2::row_filter::Filter::Chain(v2::row_filter::Chain {
            filters: vec![
                redirect_target_filter(target, own_id, access_time),
                exact_expiry_filter(observed_expiry)?,
            ],
        })),
    }))
}

/// The condition under which a [`BigTableBackend::check_and_mutate`] write proceeds.
///
/// Each variant pairs a row filter with the state that makes the write safe:
/// `Include` writes when the row matches; `Exclude` writes when it does not.
#[derive(Clone, Debug)]
enum MutatePredicate {
    /// Write proceeds when the filter matches the row.
    ///
    /// Mutations run in `true_mutations`; succeeds when `predicate_matched == true`.
    Include(v2::RowFilter),
    /// Write proceeds when the filter does not match the row.
    ///
    /// Mutations run in `false_mutations`; succeeds when `predicate_matched == false`.
    Exclude(v2::RowFilter),
}

/// Creates a row filter that reads all non-payload columns (`m`, `r`).
///
/// Used by metadata-only reads to avoid fetching the (potentially large) payload column
/// while still being able to detect both new- and legacy-format tombstones.
fn metadata_filter() -> v2::RowFilter {
    v2::RowFilter {
        filter: Some(v2::row_filter::Filter::ColumnQualifierRegexFilter(
            FILTER_META.to_owned(),
        )),
    }
}

fn mutation(mutation: mutation::Mutation) -> v2::Mutation {
    v2::Mutation {
        mutation: Some(mutation),
    }
}

/// Creates a `DeleteFromRow` mutation wrapped in the outer [`v2::Mutation`] envelope.
fn delete_row_mutation() -> v2::Mutation {
    mutation(mutation::Mutation::DeleteFromRow(
        mutation::DeleteFromRow {},
    ))
}

/// Builds the three mutations that write an object row: clear existing data,
/// then set the payload and metadata cells. Returns them with the resulting row size.
///
/// Used by both [`BigTableBackend::put_row`] (unconditional write) and
/// [`BigTableBackend::put_non_tombstone`] (conditional write).
fn object_mutations(
    path: &[u8],
    mut metadata: Metadata,
    payload: Vec<u8>,
) -> Result<([v2::Mutation; 3], u64)> {
    let (family, timestamp_micros) = match metadata.time_expires {
        None => (FAMILY_MANUAL, -1),
        Some(deadline) => (FAMILY_GC, deadline.as_micros() as i64),
    };

    // Record the payload size in the metadata before persisting it.
    metadata.size = Some(payload.len());

    let metadata_bytes = serde_json::to_vec(&metadata)
        .context(ErrorKind::Internal, "encoding Bigtable object metadata")?;

    let mutations = [
        // NB: We explicitly delete the row to clear metadata on overwrite.
        delete_row_mutation(),
        mutation(mutation::Mutation::SetCell(mutation::SetCell {
            family_name: family.to_owned(),
            column_qualifier: COLUMN_PAYLOAD.to_owned(),
            timestamp_micros,
            value: payload,
        })),
        mutation(mutation::Mutation::SetCell(mutation::SetCell {
            family_name: family.to_owned(),
            column_qualifier: COLUMN_METADATA.to_owned(),
            timestamp_micros,
            value: metadata_bytes,
        })),
    ];

    let size = row_size(path, &mutations);
    Ok((mutations, size))
}

/// Approximates the bytes a row occupies, as its key plus every cell value written.
///
/// This function does not distinguish between object rows and tombstone rows. It does not
/// include Bigtable's own overhead.
fn row_size(path: &[u8], mutations: &[v2::Mutation]) -> u64 {
    let cells: usize = mutations
        .iter()
        .filter_map(|m| match &m.mutation {
            Some(mutation::Mutation::SetCell(cell)) => Some(cell.value.len()),
            _ => None,
        })
        .sum();

    (path.len() + cells) as u64
}

/// Builds the two mutations that write a tombstone row: clear existing data,
/// then set the redirect cell.
///
/// Used by both unconditional tombstone writes and the conditional expiry-extension paths.
fn tombstone_mutations(tombstone: &Tombstone) -> [v2::Mutation; 2] {
    let (family, timestamp_micros) = match tombstone.time_expires {
        None => (FAMILY_MANUAL, -1),
        Some(deadline) => (FAMILY_GC, deadline.as_micros() as i64),
    };

    [
        delete_row_mutation(),
        mutation(mutation::Mutation::SetCell(mutation::SetCell {
            family_name: family.to_owned(),
            column_qualifier: COLUMN_REDIRECT.to_owned(),
            timestamp_micros,
            value: tombstone.target.as_storage_path().to_string().into_bytes(),
        })),
    ]
}

/// Subset of [`Metadata`] that indicates a row is a tombstone instead of a real object.
///
/// Used to construct [`RowData`].
#[derive(Debug, Deserialize)]
struct LegacyTombstoneMeta {
    /// Internal redirect tombstone marker.
    ///
    /// When `true`, this object is a legacy tombstone. This implies:
    ///  - the payload is empty
    ///  - metadata is not meaningful
    ///  - the `r` column is not present
    #[serde(default)]
    is_redirect_tombstone: bool,
}

/// Parsed data from a BigTable row's cells.
enum RowData {
    /// A regular object row with payload and metadata.
    Object {
        metadata: Metadata,
        payload: Vec<u8>,
        /// Original GC timestamp for exact CAS matching, including legacy fractional seconds.
        expiry_micros: i64,
    },
    /// A tombstone row indicating the real payload lives on the long-term backend.
    Tombstone {
        target: Vec<u8>,
        time_expires: Option<Timestamp>,
        /// Original GC timestamp for exact CAS matching, including legacy fractional seconds.
        expiry_micros: i64,
    },
}

impl RowData {
    /// Parses a set of row cells into a [`RowData`].
    ///
    /// New-format tombstones are identified by the presence of the `r` column.
    /// Legacy tombstones (written before the column migration) are identified by
    /// `is_redirect_tombstone: true` in the `m` column JSON; a
    /// `bigtable.legacy_tombstone_read` metric is emitted on each such read.
    fn from_cells(cells: Vec<RowCell>) -> Result<Self> {
        let mut metadata_opt: Option<Metadata> = None;
        let mut redirect_detected = false;
        let mut redirect_target = Vec::new();
        let mut expire_at = None;
        let mut expiry_micros = 0;
        let mut payload = Vec::new();

        for cell in cells {
            // NB: All cells are written with the same timestamp; last write is safe.

            // Only derive expiration from GC-family cells — manual-family cells
            // use server-assigned timestamps that don't represent expiration.
            if cell.family_name == FAMILY_GC {
                expiry_micros = cell.timestamp_micros;
                expire_at = Some(Timestamp::from_unix_micros(expiry_micros).context(
                    ErrorKind::CorruptData,
                    "decoding Bigtable expiration timestamp",
                )?);
            }

            match cell.qualifier.as_slice() {
                COLUMN_REDIRECT => {
                    redirect_detected = true;
                    redirect_target = cell.value;
                }
                COLUMN_PAYLOAD => {
                    payload = cell.value;
                }
                COLUMN_METADATA => {
                    if let Ok(legacy_meta) =
                        serde_json::from_slice::<LegacyTombstoneMeta>(&cell.value)
                        && legacy_meta.is_redirect_tombstone
                    {
                        redirect_detected = true;
                        objectstore_metrics::count!("bigtable.legacy_tombstone_read");
                    } else {
                        metadata_opt = Some(serde_json::from_slice(&cell.value).context(
                            ErrorKind::CorruptData,
                            "decoding Bigtable object metadata",
                        )?);
                    }
                }
                _ => {}
            }
        }

        Ok(if redirect_detected {
            RowData::Tombstone {
                target: redirect_target,
                time_expires: expire_at,
                expiry_micros,
            }
        } else {
            // Metadata may have been skipped by a payload-only internal read.
            let mut metadata = metadata_opt.unwrap_or_default();
            metadata.time_expires = expire_at;
            RowData::Object {
                metadata,
                payload,
                expiry_micros,
            }
        })
    }

    /// Returns the resolved expiration timestamp for this row, regardless of variant.
    fn time_expires(&self) -> Option<Timestamp> {
        match self {
            RowData::Object { metadata, .. } => metadata.time_expires,
            RowData::Tombstone { time_expires, .. } => *time_expires,
        }
    }

    /// Returns `true` if this row is expired as of the given `time`.
    ///
    /// Only applies to rows with an expiration deadline.
    fn expires_before(&self, time: Timestamp) -> bool {
        self.time_expires().is_some_and(|ts| ts < time)
    }
}

/// Parses the raw `r` column bytes into a redirect target [`ObjectId`].
///
/// For tombstones with an empty `r` value, falls back to the ID of the tombstone
/// itself and emits a `bigtable.empty_redirect_read` metric so deployments can
/// track when it is safe to remove the legacy empty-value code path.
fn parse_redirect_target(redirect_path: &[u8], tombstone_id: &ObjectId) -> Result<ObjectId> {
    if redirect_path.is_empty() {
        objectstore_metrics::count!("bigtable.empty_redirect_read");
        Ok(tombstone_id.clone())
    } else {
        let redirect_str = std::str::from_utf8(redirect_path)
            .context(ErrorKind::CorruptData, "decoding Bigtable redirect target")?;
        ObjectId::from_storage_path(redirect_str)
            .ok_or_else(|| Error::new(ErrorKind::CorruptData, "parsing Bigtable redirect target"))
    }
}

impl BigTableBackend {
    /// Creates a new [`BigTableBackend`] from the given `config`.
    ///
    /// Pass an `endpoint` in the config to connect to a local emulator; omit it to use real GCP
    /// credentials. `connections` controls the gRPC connection pool size (defaults to 1).
    /// A `PingAndWarm` request is sent through the pool every 10 seconds in an effort
    /// to keep the connections active.
    pub async fn new(
        config: BigTableConfig,
        streams: &ChangeStreamFactory,
    ) -> anyhow::Result<Self> {
        let BigTableConfig {
            endpoint,
            project_id,
            instance_name,
            table_name,
            connections,
            rpc_timeout,
            cogs,
        } = config;
        let change_stream = streams.build(cogs.as_ref());

        let bigtable = if let Some(ref endpoint) = endpoint {
            BigTableConnection::new_with_emulator(
                endpoint,
                &project_id,
                &instance_name,
                false, // is_read_only
                connections.unwrap_or(1),
                Some(rpc_timeout),
            )?
        } else {
            let token_provider = PrefetchingTokenProvider::gcp_auth(TOKEN_SCOPES).await?;
            BigTableConnection::new_with_managed_transport(
                &project_id,
                &instance_name,
                false, // is_read_only
                Some(rpc_timeout),
                Arc::new(token_provider),
                connections.unwrap_or(1),
                true, // prime_channels
                None, // app_profile_id
                MAX_CHANNEL_AGE,
                Some(Duration::from_secs(10)), // periodic PingAndWarm
            )
            .await?
        };

        let client = bigtable.client();

        Ok(Self {
            bigtable,
            instance_path: format!("projects/{project_id}/instances/{instance_name}"),
            table_path: client.get_full_table_name(&table_name),
            table_name,
            change_stream,
        })
    }

    /// Reads a single row by key, returning parsed row data.
    ///
    /// Returns `None` if the row is absent or has expired.
    #[tracing::instrument(level = "debug", fields(action), skip_all)]
    async fn read_row(
        &self,
        path: &[u8],
        action: &'static str,
        access_time: Timestamp,
        filter: Option<v2::RowFilter>,
    ) -> Result<Option<RowData>> {
        let request = v2::ReadRowsRequest {
            table_name: self.table_path.clone(),
            rows: Some(v2::RowSet {
                row_keys: vec![path.to_owned()],
                row_ranges: vec![],
            }),
            filter,
            rows_limit: 1,
            ..Default::default()
        };

        let response = retry(action, || async {
            self.bigtable.client().read_rows(request.clone()).await
        })
        .await?;
        debug_assert!(response.len() <= 1, "Expected at most one row");

        let Some((_, cells)) = response.into_iter().next() else {
            objectstore_log::debug!("Object not found");
            return Ok(None);
        };

        let row = RowData::from_cells(cells)?;
        Ok(if row.expires_before(access_time) {
            None
        } else {
            Some(row)
        })
    }

    #[tracing::instrument(level = "debug", fields(action), skip_all)]
    async fn mutate(
        &self,
        path: Vec<u8>,
        mutations: impl Into<Vec<v2::Mutation>>,
        action: &'static str,
    ) -> Result<v2::MutateRowResponse> {
        let request = v2::MutateRowRequest {
            table_name: self.table_path.clone(),
            row_key: path,
            mutations: mutations.into(),
            ..Default::default()
        };

        let response = retry(action, || async {
            self.bigtable.client().mutate_row(request.clone()).await
        })
        .await?;

        Ok(response.into_inner())
    }

    /// Writes an object row, returning the size of the row it wrote.
    async fn put_row(
        &self,
        path: Vec<u8>,
        metadata: Metadata,
        payload: Vec<u8>,
        action: &'static str,
    ) -> Result<(v2::MutateRowResponse, u64)> {
        let (mutations, size) = object_mutations(&path, metadata, payload)?;
        let response = self.mutate(path, mutations, action).await?;
        Ok((response, size))
    }

    /// Executes a `CheckAndMutateRow` request.
    #[tracing::instrument(level = "debug", fields(action = context), skip_all)]
    async fn check_and_mutate(
        &self,
        row_key: Vec<u8>,
        predicate: MutatePredicate,
        mutations: impl Into<Vec<v2::Mutation>>,
        context: &'static str,
    ) -> Result<bool> {
        let (filter, true_mutations, false_mutations, success_on_match) = match predicate {
            MutatePredicate::Include(f) => (f, mutations.into(), vec![], true),
            MutatePredicate::Exclude(f) => (f, vec![], mutations.into(), false),
        };

        let request = v2::CheckAndMutateRowRequest {
            table_name: self.table_path.clone(),
            row_key,
            predicate_filter: Some(filter),
            true_mutations,
            false_mutations,
            ..Default::default()
        };

        let future = retry(context, || async {
            self.bigtable
                .client()
                .check_and_mutate_row(request.clone())
                .await
        });

        Ok(future.await?.predicate_matched == success_on_match)
    }
}

#[async_trait::async_trait]
impl Backend for BigTableBackend {
    fn name(&self) -> &'static str {
        "bigtable"
    }

    #[tracing::instrument(level = "debug", fields(?id), skip_all)]
    async fn put_object(
        &self,
        id: &ObjectId,
        metadata: &Metadata,
        mut stream: ClientStream,
        _access_time: Timestamp,
    ) -> Result<PutResponse> {
        objectstore_log::debug!("Writing to Bigtable backend");
        let path = id.as_storage_path().to_string().into_bytes();

        let mut payload = ChunkedBytes::new(0);
        while let Some(chunk) = stream.try_next().await? {
            payload.push(chunk);
        }

        let (_, size) = self
            .put_row(path, metadata.clone(), payload.into_bytes().into(), "put")
            .await?;
        self.change_stream.write(id, size, metadata.time_expires);

        Ok(())
    }

    #[tracing::instrument(level = "debug", skip(self))]
    async fn get_object(
        &self,
        id: &ObjectId,
        access_time: Timestamp,
        range: Option<ByteRange>,
    ) -> Result<GetResponse> {
        match self.get_tiered_object(id, access_time, range).await? {
            TieredGet::Object(metadata, content_range, payload) => {
                Ok(Some((metadata, content_range, payload)))
            }
            TieredGet::Tombstone(_) => Err(ErrorKind::UnexpectedTombstone.into()),
            TieredGet::NotFound => Ok(None),
        }
    }

    #[tracing::instrument(level = "debug", skip(self))]
    async fn get_metadata(
        &self,
        id: &ObjectId,
        access_time: Timestamp,
    ) -> Result<MetadataResponse> {
        match self.get_tiered_metadata(id, access_time).await? {
            TieredMetadata::Object(metadata) => Ok(Some(metadata)),
            TieredMetadata::Tombstone(_) => Err(ErrorKind::UnexpectedTombstone.into()),
            TieredMetadata::NotFound => Ok(None),
        }
    }

    async fn set_expiry(
        &self,
        id: &ObjectId,
        target: ExpiryUpdate,
        access_time: Timestamp,
    ) -> Result<SetExpiryResponse> {
        self.compare_and_update(id, None, TieredUpdate::SetExpiry(target), access_time)
            .await
    }

    #[tracing::instrument(level = "debug", skip(self))]
    async fn delete_object(
        &self,
        id: &ObjectId,
        _access_time: Timestamp,
    ) -> Result<DeleteResponse> {
        objectstore_log::debug!("Deleting from Bigtable backend");

        let path = id.as_storage_path().to_string().into_bytes();
        self.mutate(path, [delete_row_mutation()], "delete").await?;
        self.change_stream.delete(id);

        Ok(())
    }

    async fn join(&self) {
        flush_change_stream(&self.change_stream).await;
    }
}

#[async_trait::async_trait]
impl HighVolumeBackend for BigTableBackend {
    async fn create_upload_marker(
        &self,
        revision: &ObjectId,
        time_expires: Timestamp,
    ) -> Result<()> {
        let mutations = vec![mutation(mutation::Mutation::SetCell(mutation::SetCell {
            family_name: FAMILY_GC.to_owned(),
            column_qualifier: COLUMN_UPLOAD.to_vec(),
            timestamp_micros: time_expires.as_micros() as i64,
            value: vec![1],
        }))];
        self.mutate(
            revision.as_upload_path().to_string().into_bytes(),
            mutations,
            "create_upload_marker",
        )
        .await?;
        Ok(())
    }

    async fn has_upload_marker(&self, revision: &ObjectId, access_time: Timestamp) -> Result<bool> {
        let request = v2::ReadRowsRequest {
            table_name: self.table_path.clone(),
            rows: Some(v2::RowSet {
                row_keys: vec![revision.as_upload_path().to_string().into_bytes()],
                row_ranges: vec![],
            }),
            filter: Some(live_row_filter(column_filter(COLUMN_UPLOAD), access_time)),
            rows_limit: 1,
            ..Default::default()
        };
        let rows = retry("has_upload_marker", || async {
            self.bigtable.client().read_rows(request.clone()).await
        })
        .await?;
        Ok(!rows.is_empty())
    }

    async fn delete_upload_marker(
        &self,
        revision: &ObjectId,
        access_time: Timestamp,
    ) -> Result<bool> {
        self.check_and_mutate(
            revision.as_upload_path().to_string().into_bytes(),
            MutatePredicate::Include(live_row_filter(column_filter(COLUMN_UPLOAD), access_time)),
            vec![delete_row_mutation()],
            "delete_upload_marker",
        )
        .await
    }

    #[tracing::instrument(level = "debug", fields(?id), skip_all)]
    async fn put_non_tombstone(
        &self,
        id: &ObjectId,
        metadata: &Metadata,
        payload: Bytes,
        access_time: Timestamp,
    ) -> Result<Option<Tombstone>> {
        objectstore_log::debug!("Conditional put to Bigtable backend");

        let path = id.as_storage_path().to_string().into_bytes();
        let (mutations, size) = object_mutations(&path, metadata.clone(), payload.to_vec())?;

        for _ in 0..CAS_RETRY_COUNT {
            let write_succeeded = self
                .check_and_mutate(
                    path.clone(),
                    tombstone_predicate(access_time),
                    mutations.clone(),
                    "put_non_tombstone",
                )
                .await?;

            if write_succeeded {
                self.change_stream.write(id, size, metadata.time_expires);
                return Ok(None);
            }

            // A tombstone was present: read its data for the caller.
            let row = self
                .read_row(
                    &path,
                    "put_non_tombstone",
                    access_time,
                    Some(metadata_filter()),
                )
                .await?;

            match row {
                Some(RowData::Tombstone {
                    target,
                    time_expires,
                    ..
                }) => {
                    return Ok(Some(Tombstone {
                        target: parse_redirect_target(&target, id)?,
                        time_expires,
                    }));
                }
                // Race: Tombstone was replaced by an object, retry to overwrite
                Some(RowData::Object { .. }) => continue,
                // Race: Tombstone was deleted, retry to write.
                None => continue,
            }
        }

        Err(Error::new(
            ErrorKind::Internal,
            "Bigtable put race exhausted",
        ))
    }

    #[tracing::instrument(level = "debug", skip(self))]
    async fn get_tiered_object(
        &self,
        id: &ObjectId,
        access_time: Timestamp,
        range: Option<ByteRange>,
    ) -> Result<TieredGet> {
        objectstore_log::debug!("Reading from Bigtable backend");
        let path = id.as_storage_path().to_string().into_bytes();

        let Some(row) = self
            .read_row(&path, "get_tiered_object", access_time, None)
            .await?
        else {
            return Ok(TieredGet::NotFound);
        };

        Ok(match row {
            RowData::Tombstone {
                target,
                time_expires,
                ..
            } => TieredGet::Tombstone(Tombstone {
                target: parse_redirect_target(&target, id)?,
                time_expires,
            }),
            RowData::Object {
                metadata, payload, ..
            } => {
                let mut metadata = metadata;
                let payload = Bytes::from(payload);
                if metadata.size.is_none() {
                    // If object size wasn't written into the metadata, re-compute it now
                    metadata.size = Some(payload.len());
                }

                let (content_range, payload) = apply_range(payload, range)?;
                TieredGet::Object(metadata, content_range, crate::stream::single(payload))
            }
        })
    }

    #[tracing::instrument(level = "debug", skip(self))]
    async fn get_tiered_metadata(
        &self,
        id: &ObjectId,
        access_time: Timestamp,
    ) -> Result<TieredMetadata> {
        objectstore_log::debug!("Reading metadata from Bigtable backend");
        let path = id.as_storage_path().to_string().into_bytes();

        // Read metadata and tombstone columns — skip the (potentially large) payload.
        // NB: `metadata.size` will only be populated if the size was added to the metadata before
        // writing to Bigtable.
        let row_opt = self
            .read_row(
                &path,
                "get_tiered_metadata",
                access_time,
                Some(metadata_filter()),
            )
            .await?;
        let Some(row) = row_opt else {
            return Ok(TieredMetadata::NotFound);
        };

        Ok(match row {
            RowData::Tombstone {
                target,
                time_expires,
                ..
            } => TieredMetadata::Tombstone(Tombstone {
                target: parse_redirect_target(&target, id)?,
                time_expires,
            }),
            RowData::Object { metadata, .. } => TieredMetadata::Object(metadata),
        })
    }

    #[tracing::instrument(level = "debug", skip(self))]
    async fn compare_and_update(
        &self,
        id: &ObjectId,
        current: Option<&ObjectId>,
        update: TieredUpdate,
        access_time: Timestamp,
    ) -> Result<SetExpiryResponse> {
        let TieredUpdate::SetExpiry(expiry_target) = update;
        let path = id.as_storage_path().to_string().into_bytes();

        // Inline extension needs metadata and payload from the same read so a
        // successful conditional rewrite can preserve the payload verbatim.
        let Some(row) = self
            .read_row(&path, "set_expiry", access_time, None)
            .await?
        else {
            return Ok(SetExpiryResponse::NotFound);
        };

        let (expire_at, predicate, mutations): (_, _, Vec<_>) = match row {
            RowData::Object {
                metadata,
                payload,
                expiry_micros,
            } => {
                if current.is_some() {
                    return Ok(SetExpiryResponse::Rejected); // wrong row kind
                }
                let Some(old_expiry) = metadata.time_expires else {
                    return Ok(SetExpiryResponse::Rejected);
                };

                if old_expiry < access_time {
                    return Ok(SetExpiryResponse::NotFound); // already expired
                }
                let Some(expire_at) = expiry_target.resolve(metadata.time_created, access_time)?
                else {
                    return Ok(SetExpiryResponse::Rejected);
                };
                if old_expiry >= expire_at {
                    return Ok(SetExpiryResponse::Satisfied(expire_at)); // already satisfied
                }

                // Observing a live cell here is not atomic with wall-clock
                // expiry or Bigtable GC. The conditional write may still lose
                // to either and then returns false.
                let predicate = inline_expiry_predicate(expiry_micros, access_time)?;
                let mut metadata = metadata;
                metadata.expiration_policy = common::extended_expiration_policy(
                    metadata.expiration_policy,
                    metadata.time_created,
                    old_expiry,
                    expire_at,
                )?;
                metadata.time_expires = Some(expire_at);
                let (mutations, _) = object_mutations(&path, metadata, payload)?;
                (expire_at, predicate, mutations.into())
            }
            RowData::Tombstone {
                target,
                time_expires,
                expiry_micros,
            } => {
                let Some(expected) = current else {
                    return Ok(SetExpiryResponse::Rejected); // wrong row kind
                };
                let Some(old_expiry) = time_expires else {
                    return Ok(SetExpiryResponse::Rejected);
                };

                let redirect_target = parse_redirect_target(&target, id)?;
                if old_expiry < access_time {
                    return Ok(SetExpiryResponse::NotFound);
                }
                if redirect_target != *expected {
                    return Ok(SetExpiryResponse::Rejected); // wrong target
                }
                let Some(expire_at) = expiry_target.resolve(None, access_time)? else {
                    return Ok(SetExpiryResponse::Rejected);
                };
                if old_expiry >= expire_at {
                    return Ok(SetExpiryResponse::Satisfied(expire_at)); // already satisfied
                }

                let predicate =
                    redirect_expiry_predicate(expected, id, expiry_micros, access_time)?;
                let tombstone = Tombstone {
                    target: redirect_target,
                    time_expires: Some(expire_at),
                };
                (expire_at, predicate, tombstone_mutations(&tombstone).into())
            }
        };

        let applied = self
            .check_and_mutate(path, predicate, mutations, "set_expiry")
            .await?;

        if applied {
            self.change_stream.update(id, Some(expire_at));
        }

        Ok(if applied {
            SetExpiryResponse::Satisfied(expire_at)
        } else {
            SetExpiryResponse::Rejected
        })
    }

    #[tracing::instrument(level = "debug", skip(self))]
    async fn delete_non_tombstone(
        &self,
        id: &ObjectId,
        access_time: Timestamp,
    ) -> Result<Option<Tombstone>> {
        objectstore_log::debug!("Conditional delete from Bigtable backend");

        let path = id.as_storage_path().to_string().into_bytes();

        for _ in 0..CAS_RETRY_COUNT {
            let deleted = self
                .check_and_mutate(
                    path.clone(),
                    non_tombstone_predicate(access_time),
                    [delete_row_mutation()],
                    "delete_non_tombstone",
                )
                .await?;

            if deleted {
                self.change_stream.delete(id);
                return Ok(None);
            }

            // Nothing was deleted: either a tombstone is in the way, or the row is absent.
            // Read the row to find out which, and to hand the tombstone to the caller.
            let row = self
                .read_row(
                    &path,
                    "delete_non_tombstone",
                    access_time,
                    Some(metadata_filter()),
                )
                .await?;

            match row {
                Some(RowData::Tombstone {
                    target,
                    time_expires,
                    ..
                }) => {
                    return Ok(Some(Tombstone {
                        target: parse_redirect_target(&target, id)?,
                        time_expires,
                    }));
                }
                // Race: An object appeared since the predicate ran, delete the new object now.
                Some(RowData::Object { .. }) => continue,
                // The row is absent or expired, nothing left to do.
                None => return Ok(None),
            }
        }

        Err(Error::new(
            ErrorKind::Internal,
            "Bigtable delete race exhausted",
        ))
    }

    #[tracing::instrument(level = "debug", skip(self, write))]
    async fn compare_and_write(
        &self,
        id: &ObjectId,
        current: Option<&ObjectId>,
        write: TieredWrite,
        access_time: Timestamp,
    ) -> Result<bool> {
        objectstore_log::debug!("CAS put to Bigtable backend");

        let path = id.as_storage_path().to_string().into_bytes();
        let predicate = match (current, write.target()) {
            (Some(old), Some(new)) => update_predicate(old, new, id, access_time),
            (Some(target), None) => optional_target_predicate(target, id, access_time),
            (None, Some(target)) => optional_target_predicate(target, id, access_time),
            (None, None) => tombstone_predicate(access_time),
        };

        // Get the correct set of mutations to apply as well as the new expiration date.
        // If we're deleting something, `expires_at` is `None`. If we're writing something
        // without an expiration date, `expires_at` is `Some(None)`.
        let (mutations, expires_at): (Vec<v2::Mutation>, Option<Option<Timestamp>>) = match write {
            TieredWrite::Tombstone(tombstone) => (
                tombstone_mutations(&tombstone).into(),
                Some(tombstone.time_expires),
            ),
            TieredWrite::Object(m, p) => {
                let expires_at = m.time_expires;
                let (mutations, _) = object_mutations(&path, m, p.to_vec())?;
                (mutations.into(), Some(expires_at))
            }
            TieredWrite::Delete => (vec![delete_row_mutation()], None),
        };

        let written = self
            .check_and_mutate(
                path.clone(),
                predicate,
                mutations.clone(),
                "compare_and_write",
            )
            .await?;

        match (written, expires_at) {
            // Don't record anything if the write didn't succeed
            (false, _) => {}
            // We wrote something (the inner `expires_at` is `None` for manual GC)
            (true, Some(expires_at)) => {
                self.change_stream
                    .write(id, row_size(&path, &mutations), expires_at)
            }
            // We deleted something
            (true, None) => self.change_stream.delete(id),
        }

        Ok(written)
    }
}

/// Retries a BigTable RPC on transient errors.
async fn retry<T, F>(context: &'static str, f: impl Fn() -> F) -> Result<T>
where
    F: Future<Output = Result<T, BigTableError>> + Send,
{
    let mut retry_count = 0usize;

    loop {
        let attempt_span = tracing::debug_span!(
            "bigtable.request",
            action = context,
            grpc.status = tracing::field::Empty,
        );
        let attempt = async {
            let result = f().await;
            let span = tracing::Span::current();
            match &result {
                Ok(_) => span.record("grpc.status", "ok"),
                Err(BigTableError::RpcError(status)) => {
                    span.record("grpc.status", tracing::field::debug(status.code()))
                }
                // Non-RPC error; the error event carries the details.
                Err(_) => &span,
            };
            result
        };

        match attempt.instrument(attempt_span).await {
            Ok(res) => return Ok(res),
            Err(e) if retry_count >= REQUEST_RETRY_COUNT || !is_retryable(&e) => {
                objectstore_metrics::count!("bigtable.failures", action = context);
                return Err(e).context(
                    ErrorKind::BackendFailure,
                    format!("running Bigtable {context}"),
                );
            }
            Err(e) => {
                retry_count += 1;
                objectstore_metrics::count!("bigtable.retries", action = context);
                objectstore_log::warn!(!!&e, retry_count, context, "Retrying request");
            }
        }
    }
}

fn is_retryable(error: &BigTableError) -> bool {
    match error {
        // Transient errors on auth token refresh
        BigTableError::GCPAuthError(_) => true,
        // Transient GRPC network failures
        BigTableError::TransportError(_) => true,
        // These could also indicate transient network failures
        BigTableError::IoError(_) => true,
        BigTableError::TimeoutError(_) => true,

        // See https://docs.cloud.google.com/bigtable/docs/status-codes
        BigTableError::RpcError(status) => match status.code() {
            // Generic retriable status
            Code::Unavailable => true,
            // Timeouts
            Code::Cancelled => true,
            Code::DeadlineExceeded => true,
            // Token might have refreshed too late
            Code::Unauthenticated => true,
            // Unspecified, attempt to retry anyways
            Code::Aborted => true,
            Code::Internal => true,
            Code::FailedPrecondition => true,
            Code::Unknown => true,
            _ => false,
        },
        _ => false,
    }
}

/// Resolves an optional byte range against a payload buffer, returning the
/// applicable content range and the (potentially narrowed) payload.
///
/// When `range` is `None`, returns the full payload unchanged. Uses
/// `Bytes::slice` to avoid copying data.
fn apply_range(payload: Bytes, range: Option<ByteRange>) -> Result<(Option<ContentRange>, Bytes)> {
    let Some(byte_range) = range else {
        return Ok((None, payload));
    };

    let total = payload.len() as u64;
    let content_range = byte_range
        .resolve(total)
        .ok_or(ErrorKind::RangeNotSatisfiable { total })?;

    let sliced = payload.slice(content_range.start as usize..content_range.end as usize + 1);
    Ok((Some(content_range), sliced))
}

#[cfg(test)]
mod tests {
    use std::collections::BTreeMap;

    use anyhow::Result;
    #[cfg(feature = "storage-cogs")]
    use objectstore_inventory_tracker::{OpType, test_utils::DummyProducer};
    use objectstore_types::metadata::ExpirationPolicy;
    use objectstore_types::scope::{Scope, Scopes};

    use super::*;
    use crate::backend::common::ExpiryTarget;
    use crate::id::ObjectContext;
    use crate::stream;

    // NB: Most of these tests require a BigTable emulator running. This is done
    // automatically in CI.
    //
    // Refer to the readme for how to set up the emulator.

    fn test_config() -> BigTableConfig {
        BigTableConfig {
            endpoint: Some("localhost:8086".into()),
            project_id: "testing".into(),
            instance_name: "objectstore".into(),
            table_name: "objectstore".into(),
            connections: None,
            rpc_timeout: default_rpc_timeout(),
            cogs: None,
        }
    }

    async fn create_test_backend() -> Result<BigTableBackend> {
        BigTableBackend::new(test_config(), &ChangeStreamFactory::default()).await
    }

    #[cfg(feature = "storage-cogs")]
    async fn create_test_backend_with_change_stream() -> Result<(BigTableBackend, DummyProducer)> {
        let (streams, producer) = crate::change_stream::dummy_factory();
        let config = BigTableConfig {
            cogs: Some(CostTrackerStreamConfig {
                shared_resource_id: "bigtable_objectstore".into(),
                sample_rate: 1.0,
            }),
            ..test_config()
        };

        Ok((BigTableBackend::new(config, &streams).await?, producer))
    }

    fn make_id() -> ObjectId {
        ObjectId::random(ObjectContext {
            usecase: "testing".into(),
            scopes: Scopes::from_iter([Scope::create("testing", "value").unwrap()]),
        })
    }

    async fn create_object(
        backend: &BigTableBackend,
        id: &ObjectId,
        metadata: &Metadata,
        payload: &[u8],
        now: Timestamp,
    ) -> Result<()> {
        let path = id.as_storage_path().to_string().into_bytes();
        // Resolve `time_expires` from `now` (as `from_insert_headers` does) unless the test set
        // it explicitly, so `object_mutations` has an expiration to persist.
        let mut metadata = metadata.clone();
        if metadata.time_expires.is_none() {
            metadata.time_expires = metadata.expiration_policy.expires_in().map(|ttl| now + ttl);
        }
        let (mutations, _) = object_mutations(&path, metadata, payload.to_vec())?;
        backend.mutate(path, mutations, "test-setup").await?;
        Ok(())
    }

    async fn create_tombstone(
        backend: &BigTableBackend,
        id: &ObjectId,
        tombstone: &Tombstone,
    ) -> Result<()> {
        let path = id.as_storage_path().to_string().into_bytes();
        let mutations = tombstone_mutations(tombstone);
        backend.mutate(path, mutations, "test-setup").await?;
        Ok(())
    }

    /// Writes a legacy-format tombstone row directly into Bigtable.
    async fn write_legacy_tombstone(
        backend: &BigTableBackend,
        id: &ObjectId,
        expiration_policy: ExpirationPolicy,
        time_expires: Option<Timestamp>,
    ) -> Result<()> {
        let meta = if expiration_policy.is_manual() {
            r#"{"is_redirect_tombstone":true}"#.to_owned()
        } else {
            let policy_json = serde_json::to_string(&expiration_policy).unwrap();
            format!(r#"{{"is_redirect_tombstone":true,"expiration_policy":{policy_json}}}"#)
        };

        let (family, timestamp_micros) = if expiration_policy.is_manual() {
            (FAMILY_MANUAL, -1)
        } else {
            let t =
                time_expires.unwrap_or(Timestamp::now() + expiration_policy.expires_in().unwrap());
            (FAMILY_GC, t.as_micros() as i64)
        };

        let path = id.as_storage_path().to_string().into_bytes();
        let mutations = [mutation(mutation::Mutation::SetCell(mutation::SetCell {
            family_name: family.to_owned(),
            column_qualifier: COLUMN_METADATA.to_owned(),
            timestamp_micros,
            value: meta.into_bytes(),
        }))];

        backend.mutate(path, mutations, "test-setup").await?;

        Ok(())
    }

    /// Writes a historical `r`/`t` tombstone row with an empty `r` value directly.
    async fn write_empty_redirect_tombstone(
        backend: &BigTableBackend,
        id: &ObjectId,
    ) -> Result<()> {
        let path = id.as_storage_path().to_string().into_bytes();
        let mutations = [
            mutation(mutation::Mutation::SetCell(mutation::SetCell {
                family_name: FAMILY_MANUAL.to_owned(),
                column_qualifier: COLUMN_REDIRECT.to_owned(),
                timestamp_micros: -1,
                value: b"".to_vec(), // empty — legacy format
            })),
            mutation(mutation::Mutation::SetCell(mutation::SetCell {
                family_name: FAMILY_MANUAL.to_owned(),
                column_qualifier: b"t".to_vec(),
                timestamp_micros: -1,
                value: b"{}".to_vec(),
            })),
        ];

        backend.mutate(path, mutations, "test-setup").await?;

        Ok(())
    }

    // --- Section 1: Object Operations ---

    /// Verifies the full roundtrip: put → get_object (payload + metadata) → get_metadata (metadata).
    #[tokio::test]
    async fn test_roundtrip() -> Result<()> {
        let backend = create_test_backend().await?;

        let id = make_id();
        let metadata = Metadata {
            content_type: "text/plain".into(),
            time_created: Some(Timestamp::now()),
            custom: BTreeMap::from_iter([("hello".into(), "world".into())]),
            ..Default::default()
        };

        backend
            .put_object(
                &id,
                &metadata,
                stream::single("hello, world"),
                Timestamp::now(),
            )
            .await?;

        let (obj_meta, _, stream) = backend
            .get_object(&id, Timestamp::now(), None)
            .await?
            .unwrap();
        let payload = stream::read_to_vec(stream).await?;
        assert_eq!(payload, b"hello, world");
        assert_eq!(obj_meta.content_type, metadata.content_type);
        assert_eq!(obj_meta.custom, metadata.custom);

        let head_meta = backend.get_metadata(&id, Timestamp::now()).await?.unwrap();
        assert_eq!(head_meta.content_type, metadata.content_type);
        assert_eq!(head_meta.custom, metadata.custom);

        Ok(())
    }

    /// Verifies that a server-resolved `time_expires` is persisted verbatim, not recomputed.
    #[tokio::test]
    async fn test_time_expires_roundtrip() -> Result<()> {
        let backend = create_test_backend().await?;

        let id = make_id();
        let ttl = Duration::from_hours(2 * 24);
        let expires = Timestamp::now() + ttl;
        let metadata = Metadata {
            expiration_policy: ExpirationPolicy::TimeToLive(ttl),
            time_expires: Some(expires),
            ..Default::default()
        };
        create_object(&backend, &id, &metadata, b"data", Timestamp::now()).await?;

        let meta = backend.get_metadata(&id, Timestamp::now()).await?.unwrap();
        assert_eq!(meta.time_expires, Some(expires));

        Ok(())
    }

    /// Verifies that absent rows return None or succeed silently for all read/delete operations.
    #[tokio::test]
    async fn test_nonexistent() -> Result<()> {
        let backend = create_test_backend().await?;

        let id = make_id();
        assert!(
            backend
                .get_object(&id, Timestamp::now(), None)
                .await?
                .is_none()
        );
        assert!(backend.get_metadata(&id, Timestamp::now()).await?.is_none());
        backend.delete_object(&id, Timestamp::now()).await?;

        Ok(())
    }

    #[tokio::test]
    async fn test_overwrite() -> Result<()> {
        let backend = create_test_backend().await?;

        let id = make_id();
        let first_metadata = Metadata {
            custom: BTreeMap::from_iter([("invalid".into(), "invalid".into())]),
            ..Default::default()
        };
        create_object(&backend, &id, &first_metadata, b"hello", Timestamp::now()).await?;

        let second_metadata = Metadata {
            custom: BTreeMap::from_iter([("hello".into(), "world".into())]),
            ..Default::default()
        };
        backend
            .put_object(
                &id,
                &second_metadata,
                stream::single("world"),
                Timestamp::now(),
            )
            .await?;

        let (meta, _, stream) = backend
            .get_object(&id, Timestamp::now(), None)
            .await?
            .unwrap();
        let payload = stream::read_to_vec(stream).await?;
        assert_eq!(payload, b"world");
        assert_eq!(meta.custom, second_metadata.custom);

        Ok(())
    }

    #[tokio::test]
    async fn test_read_after_delete() -> Result<()> {
        let backend = create_test_backend().await?;

        let id = make_id();
        let metadata = Metadata::default();
        create_object(&backend, &id, &metadata, b"hello", Timestamp::now()).await?;
        backend.delete_object(&id, Timestamp::now()).await?;

        assert!(
            backend
                .get_object(&id, Timestamp::now(), None)
                .await?
                .is_none()
        );

        Ok(())
    }

    /// Backend reads are side-effect-free; explicit extension preserves payload.
    #[tokio::test]
    async fn test_set_expiry() -> Result<()> {
        for is_ttl in [false, true] {
            let backend = create_test_backend().await?;
            let tti = Duration::from_hours(2 * 24);
            let mut metadata = Metadata {
                expiration_policy: if is_ttl {
                    ExpirationPolicy::TimeToLive(tti)
                } else {
                    ExpirationPolicy::TimeToIdle(tti)
                },
                ..Default::default()
            };

            // Backdate `now` so the written expiry (past_now + tti) is stale but not expired.
            let past_now = Timestamp::now() - tti + Duration::from_mins(1);

            let id = make_id();
            metadata.time_created = Some(past_now);
            metadata.time_expires = Some(past_now + tti);
            let path = id.as_storage_path().to_string().into_bytes();
            let (mutations, _) = object_mutations(&path, metadata, b"hello, world".to_vec())?;
            // Simulate a legacy fractional deadline. Renewal must match the raw GC timestamp.
            let mutations = mutations.map(|mut mutation| {
                if let Some(mutation::Mutation::SetCell(cell)) = &mut mutation.mutation {
                    cell.timestamp_micros -= 500_000;
                }
                mutation
            });
            backend.mutate(path, mutations, "test-setup").await?;

            let (observed, _, _) = backend
                .get_object(&id, Timestamp::now(), None)
                .await?
                .unwrap();
            let observed_expiry = observed.time_expires.unwrap();
            assert_eq!(
                backend
                    .get_metadata(&id, Timestamp::now())
                    .await?
                    .unwrap()
                    .time_expires,
                Some(observed_expiry),
                "backend reads must not renew TTI"
            );

            let requested = past_now + tti + tti;
            for target in [
                ExpiryTarget::At(requested),
                ExpiryTarget::FromCreation(tti + tti),
            ] {
                assert_eq!(
                    backend
                        .set_expiry(&id, target.into(), Timestamp::now())
                        .await?,
                    SetExpiryResponse::Satisfied(requested)
                );
            }
            assert_eq!(
                backend
                    .get_metadata(&id, Timestamp::now())
                    .await?
                    .unwrap()
                    .time_expires,
                Some(requested)
            );
            let (updated, _, stream) = backend
                .get_object(&id, Timestamp::now(), None)
                .await?
                .unwrap();
            assert_eq!(
                updated.expiration_policy,
                if is_ttl {
                    ExpirationPolicy::TimeToLive(tti + tti)
                } else {
                    ExpirationPolicy::TimeToIdle(tti)
                }
            );
            let payload = stream::read_to_vec(stream).await?;
            assert_eq!(payload, b"hello, world");
        }
        Ok(())
    }

    #[tokio::test]
    async fn test_expiry_outcomes() -> Result<()> {
        let backend = create_test_backend().await?;
        let access_time = Timestamp::now();
        let deadline = access_time + Duration::from_hours(1);
        for (expiry, expected) in [
            (None, SetExpiryResponse::Rejected),
            (
                Some(access_time - Duration::from_secs(1)),
                SetExpiryResponse::NotFound,
            ),
            (Some(deadline), SetExpiryResponse::Rejected),
        ] {
            let id = make_id();
            let metadata = Metadata {
                time_expires: expiry,
                ..Default::default()
            };
            create_object(&backend, &id, &metadata, b"payload", access_time).await?;
            assert_eq!(
                backend
                    .set_expiry(
                        &id,
                        ExpiryTarget::FromCreation(Duration::from_hours(2)).into(),
                        access_time
                    )
                    .await?,
                expected
            );
        }
        Ok(())
    }

    #[tokio::test]
    async fn test_expiry_conflict() -> Result<()> {
        let backend = create_test_backend().await?;
        let missing = make_id();
        assert_eq!(
            backend
                .set_expiry(
                    &missing,
                    ExpiryTarget::At(Timestamp::now() + Duration::from_hours(2)).into(),
                    Timestamp::now()
                )
                .await?,
            SetExpiryResponse::NotFound
        );

        let id = make_id();
        let observed_expiry = Timestamp::now() + Duration::from_hours(1);
        let original = Metadata {
            expiration_policy: ExpirationPolicy::TimeToLive(Duration::from_hours(1)),
            time_expires: Some(observed_expiry),
            ..Default::default()
        };
        create_object(&backend, &id, &original, b"original", Timestamp::now()).await?;

        let path = id.as_storage_path().to_string().into_bytes();
        let mut extended = original.clone();
        extended.time_expires = Some(observed_expiry + Duration::from_hours(1));
        let (extension, _) = object_mutations(&path, extended, b"original".to_vec())?;
        let predicate =
            inline_expiry_predicate(observed_expiry.as_micros() as i64, Timestamp::now())?;

        let mut replacement = original.clone();
        replacement.time_expires = Some(observed_expiry + Duration::from_mins(1));
        create_object(
            &backend,
            &id,
            &replacement,
            b"replacement",
            Timestamp::now(),
        )
        .await?;
        assert!(
            !backend
                .check_and_mutate(path, predicate, extension, "test-expiry-conflict")
                .await?
        );
        let (_, _, payload) = backend
            .get_object(&id, Timestamp::now(), None)
            .await?
            .unwrap();
        assert_eq!(stream::read_to_vec(payload).await?, b"replacement");
        Ok(())
    }

    #[tokio::test]
    async fn test_redirect_expiry() -> Result<()> {
        let backend = create_test_backend().await?;
        let id = make_id();
        let target = ObjectId::random(id.context().clone());
        let wrong_target = ObjectId::random(id.context().clone());
        let old_expiry = Timestamp::now() + Duration::from_hours(1);
        let path = id.as_storage_path().to_string().into_bytes();
        let mutations = tombstone_mutations(&Tombstone {
            target: target.clone(),
            time_expires: Some(old_expiry),
        })
        .map(|mut mutation| {
            if let Some(mutation::Mutation::SetCell(cell)) = &mut mutation.mutation {
                cell.timestamp_micros -= 500_000;
            }
            mutation
        });
        backend.mutate(path, mutations, "test-setup").await?;

        let later = old_expiry + Duration::from_hours(2);
        assert_eq!(
            backend
                .compare_and_update(
                    &id,
                    Some(&wrong_target),
                    TieredUpdate::SetExpiry(ExpiryTarget::At(later).into()),
                    Timestamp::now(),
                )
                .await?,
            SetExpiryResponse::Rejected
        );
        assert_eq!(
            backend
                .compare_and_update(
                    &id,
                    Some(&target),
                    TieredUpdate::SetExpiry(ExpiryTarget::At(later).into()),
                    Timestamp::now()
                )
                .await?,
            SetExpiryResponse::Satisfied(later)
        );
        let requested = old_expiry + Duration::from_mins(30);
        assert_eq!(
            backend
                .compare_and_update(
                    &id,
                    Some(&target),
                    TieredUpdate::SetExpiry(ExpiryTarget::At(requested).into()),
                    Timestamp::now(),
                )
                .await?,
            SetExpiryResponse::Satisfied(requested)
        );
        let TieredMetadata::Tombstone(tombstone) =
            backend.get_tiered_metadata(&id, Timestamp::now()).await?
        else {
            panic!("expected tombstone");
        };
        assert_eq!(tombstone.time_expires, Some(later));
        Ok(())
    }

    // --- Section 2: Expiration ---

    #[tokio::test]
    async fn test_ttl_immediate() -> Result<()> {
        // NB: We create a TTL that immediately expires in this test. This might be optimized away
        // in a future implementation, so we will have to update this test accordingly.

        let backend = create_test_backend().await?;

        let id = make_id();
        let metadata = Metadata {
            expiration_policy: ExpirationPolicy::TimeToLive(Duration::from_secs(0)),
            time_expires: Some(Timestamp::now() - Duration::from_secs(1)),
            ..Default::default()
        };
        create_object(&backend, &id, &metadata, b"hello, world", Timestamp::now()).await?;

        assert!(
            backend
                .get_object(&id, Timestamp::now(), None)
                .await?
                .is_none()
        );

        Ok(())
    }

    #[tokio::test]
    async fn test_tti_immediate() -> Result<()> {
        // NB: We create a TTI that immediately expires in this test. This might be optimized away
        // in a future implementation, so we will have to update this test accordingly.

        let backend = create_test_backend().await?;

        let id = make_id();
        let metadata = Metadata {
            expiration_policy: ExpirationPolicy::TimeToIdle(Duration::from_secs(0)),
            time_expires: Some(Timestamp::now() - Duration::from_secs(1)),
            ..Default::default()
        };
        create_object(&backend, &id, &metadata, b"hello, world", Timestamp::now()).await?;

        assert!(
            backend
                .get_object(&id, Timestamp::now(), None)
                .await?
                .is_none()
        );

        Ok(())
    }

    // --- Section 3: Tiered Operations ---

    /// Covers all three row states for `get_tiered_object` and `get_tiered_metadata`.
    ///
    /// - **empty**: both return NotFound.
    /// - **object**: put_object, both return the Object variant with correct payload/metadata.
    /// - **tombstone**: CAS-write with a distinct `lt_id`, both return the Tombstone variant
    ///   with `target == lt_id`.
    #[tokio::test]
    async fn test_tiered_get() -> Result<()> {
        let backend = create_test_backend().await?;

        // empty
        let id = make_id();
        assert!(matches!(
            backend
                .get_tiered_object(&id, Timestamp::now(), None)
                .await?,
            TieredGet::NotFound
        ));
        assert!(matches!(
            backend.get_tiered_metadata(&id, Timestamp::now()).await?,
            TieredMetadata::NotFound
        ));

        // object
        let id = make_id();
        let put_meta = Metadata {
            content_type: "text/plain".into(),
            custom: BTreeMap::from_iter([("k".into(), "v".into())]),
            ..Default::default()
        };
        create_object(&backend, &id, &put_meta, b"payload", Timestamp::now()).await?;

        let TieredGet::Object(obj_meta, _, obj_stream) = backend
            .get_tiered_object(&id, Timestamp::now(), None)
            .await?
        else {
            panic!("expected TieredGet::Object");
        };
        let obj_payload = stream::read_to_vec(obj_stream).await?;
        assert_eq!(obj_payload, b"payload");
        assert_eq!(obj_meta.content_type, put_meta.content_type);
        assert_eq!(obj_meta.custom, put_meta.custom);

        let TieredMetadata::Object(head_meta) =
            backend.get_tiered_metadata(&id, Timestamp::now()).await?
        else {
            panic!("expected TieredMetadata::Object");
        };
        assert_eq!(head_meta.content_type, put_meta.content_type);
        assert_eq!(head_meta.custom, put_meta.custom);

        // tombstone
        let hv_id = make_id();
        let lt_id = ObjectId::random(hv_id.context().clone());
        let tombstone = Tombstone {
            target: lt_id.clone(),
            time_expires: None,
        };
        create_tombstone(&backend, &hv_id, &tombstone).await?;

        match backend
            .get_tiered_object(&hv_id, Timestamp::now(), None)
            .await?
        {
            TieredGet::Tombstone(get_t) => assert_eq!(get_t.target, lt_id),
            other => panic!("expected TieredGet::Tombstone, got {other:?}"),
        }
        match backend
            .get_tiered_metadata(&hv_id, Timestamp::now())
            .await?
        {
            TieredMetadata::Tombstone(meta_t) => assert_eq!(meta_t.target, lt_id,),
            other => panic!("expected TieredMetadata::Tombstone, got {other:?}"),
        }

        Ok(())
    }

    /// Covers all three row states for `put_non_tombstone`.
    ///
    /// - **empty**: returns None, object is readable.
    /// - **object**: overwrites with new payload, returns None.
    /// - **tombstone**: returns Some(Tombstone) with the correct target; tombstone still intact.
    #[tokio::test]
    async fn test_put_non_tombstone() -> Result<()> {
        let backend = create_test_backend().await?;

        // empty: put_non_tombstone on absent row succeeds and makes object readable.
        let id = make_id();
        let metadata = Metadata::default();
        let result = backend
            .put_non_tombstone(
                &id,
                &metadata,
                Bytes::from_static(b"first"),
                Timestamp::now(),
            )
            .await?;
        assert_eq!(result, None, "expected None on empty row");
        let (_, _, stream) = backend
            .get_object(&id, Timestamp::now(), None)
            .await?
            .unwrap();
        assert_eq!(&stream::read_to_vec(stream).await?, b"first");

        // object: put_non_tombstone on existing object replaces payload, returns None.
        let id = make_id();
        create_object(&backend, &id, &metadata, b"old", Timestamp::now()).await?;
        let result = backend
            .put_non_tombstone(&id, &metadata, Bytes::from_static(b"new"), Timestamp::now())
            .await?;
        assert_eq!(result, None, "expected None when overwriting object");
        let (_, _, stream) = backend
            .get_object(&id, Timestamp::now(), None)
            .await?
            .unwrap();
        assert_eq!(&stream::read_to_vec(stream).await?, b"new");

        // tombstone: put_non_tombstone returns Some(Tombstone) and leaves tombstone intact.
        let hv_id = make_id();
        let lt_id = ObjectId::random(hv_id.context().clone());
        let tombstone = Tombstone {
            target: lt_id.clone(),
            time_expires: None,
        };
        create_tombstone(&backend, &hv_id, &tombstone).await?;
        let result = backend
            .put_non_tombstone(&hv_id, &metadata, Bytes::new(), Timestamp::now())
            .await?;
        let returned = result.expect("expected Some(Tombstone) when row is a tombstone");
        assert_eq!(returned.target, lt_id);
        assert!(
            matches!(
                backend
                    .get_tiered_metadata(&hv_id, Timestamp::now())
                    .await?,
                TieredMetadata::Tombstone(_)
            ),
            "tombstone must still exist after put_non_tombstone"
        );

        Ok(())
    }

    /// Covers all three row states for `delete_non_tombstone`.
    ///
    /// - **empty**: returns None.
    /// - **object**: returns None, row gone.
    /// - **tombstone**: returns Some(Tombstone) with correct target; tombstone still intact.
    ///
    /// Verifies that the `r` column is correctly detected by both the `ReadRows` column
    /// filter and the `CheckAndMutate` `non_tombstone_predicate`.
    #[tokio::test]
    async fn test_delete_non_tombstone() -> Result<()> {
        let backend = create_test_backend().await?;

        // empty
        let id = make_id();
        assert_eq!(
            backend.delete_non_tombstone(&id, Timestamp::now()).await?,
            None
        );

        // object
        let id = make_id();
        let metadata = Metadata::default();
        create_object(&backend, &id, &metadata, b"hello, world", Timestamp::now()).await?;
        assert_eq!(
            backend.delete_non_tombstone(&id, Timestamp::now()).await?,
            None
        );
        assert!(
            backend
                .get_object(&id, Timestamp::now(), None)
                .await?
                .is_none()
        );

        // tombstone
        let id = make_id();
        let tombstone = Tombstone {
            target: id.clone(),
            time_expires: None,
        };
        create_tombstone(&backend, &id, &tombstone).await?;
        let tombstone = backend
            .delete_non_tombstone(&id, Timestamp::now())
            .await?
            .expect("expected Some(tombstone)");
        assert_eq!(tombstone.target, id, "tombstone target must be returned");
        assert!(
            matches!(
                backend.get_tiered_metadata(&id, Timestamp::now()).await?,
                TieredMetadata::Tombstone(_)
            ),
            "tombstone must still exist after delete_non_tombstone"
        );

        Ok(())
    }

    // --- Section 4: Compare-and-Write ---

    /// Creating a tombstone on an empty row succeeds; a retry of the same CAS also succeeds.
    ///
    /// After creation, both tiered and legacy APIs reflect the tombstone.
    #[tokio::test]
    async fn test_cas_create_tombstone() -> Result<()> {
        let backend = create_test_backend().await?;

        let hv_id = make_id();
        let lt_id = ObjectId::random(hv_id.context().clone());
        let time_expires = Some(Timestamp::now() + Duration::from_hours(1));
        let tombstone = Tombstone {
            target: lt_id.clone(),
            time_expires,
        };

        // First create succeeds.
        let committed = backend
            .compare_and_write(
                &hv_id,
                None,
                TieredWrite::Tombstone(tombstone.clone()),
                Timestamp::now(),
            )
            .await?;
        assert!(committed, "expected CAS success on empty row");

        // Tiered reads must see the tombstone with the correct target and deadline.
        let TieredMetadata::Tombstone(t) = backend
            .get_tiered_metadata(&hv_id, Timestamp::now())
            .await?
        else {
            panic!("expected TieredMetadata::Tombstone");
        };
        assert_eq!(t.target, lt_id, "target must round-trip via r column");
        assert_eq!(t.time_expires, time_expires);
        match backend
            .get_tiered_object(&hv_id, Timestamp::now(), None)
            .await?
        {
            TieredGet::Tombstone(t) => assert_eq!(t.target, lt_id, "round-trip via r column"),
            other => panic!("expected TieredGet::Tombstone, got {other:?}"),
        }

        // Legacy reads must error rather than leak tombstone data.
        assert!(
            backend
                .get_object(&hv_id, Timestamp::now(), None)
                .await
                .is_err_and(|error| error.kind() == ErrorKind::UnexpectedTombstone)
        );
        assert!(
            backend
                .get_metadata(&hv_id, Timestamp::now())
                .await
                .is_err_and(|error| error.kind() == ErrorKind::UnexpectedTombstone)
        );

        // Idempotent retry: retry with the same target succeeds
        let second = backend
            .compare_and_write(
                &hv_id,
                None,
                TieredWrite::Tombstone(tombstone),
                Timestamp::now(),
            )
            .await?;
        assert!(second, "idempotent retry");

        Ok(())
    }

    /// Swapping a tombstone target: wrong expected → false, correct expected → true.
    #[tokio::test]
    async fn test_cas_swap_tombstone() -> Result<()> {
        let backend = create_test_backend().await?;

        let hv_id = make_id();
        let old_lt_id = ObjectId::random(hv_id.context().clone());
        let wrong_lt_id = ObjectId::random(hv_id.context().clone());
        let new_lt_id = ObjectId::random(hv_id.context().clone());

        let tombstone = Tombstone {
            target: old_lt_id.clone(),
            time_expires: None,
        };
        create_tombstone(&backend, &hv_id, &tombstone).await?;

        // Wrong target: CAS fails, tombstone unchanged.
        let write = TieredWrite::Tombstone(Tombstone {
            target: new_lt_id.clone(),
            time_expires: None,
        });
        let swapped = backend
            .compare_and_write(&hv_id, Some(&wrong_lt_id), write.clone(), Timestamp::now())
            .await?;
        assert!(!swapped, "expected CAS failure due to wrong target");
        match backend
            .get_tiered_metadata(&hv_id, Timestamp::now())
            .await?
        {
            TieredMetadata::Tombstone(t) => assert_eq!(t.target, old_lt_id),
            other => panic!("expected tombstone, got {other:?}"),
        }

        // Correct target: CAS succeeds, target updated.
        let swapped = backend
            .compare_and_write(&hv_id, Some(&old_lt_id), write.clone(), Timestamp::now())
            .await?;
        assert!(swapped, "expected CAS success with correct target");
        match backend
            .get_tiered_metadata(&hv_id, Timestamp::now())
            .await?
        {
            TieredMetadata::Tombstone(t) => assert_eq!(t.target, new_lt_id),
            other => panic!("expected tombstone, got {other:?}"),
        }

        // Idempotent retry: same A→B swap returns true.
        let retry = backend
            .compare_and_write(&hv_id, Some(&old_lt_id), write, Timestamp::now())
            .await?;
        assert!(retry, "idempotent retry");

        Ok(())
    }

    /// Swapping a tombstone for inline object data: wrong expected → false, correct → true.
    #[tokio::test]
    async fn test_cas_swap_inline() -> Result<()> {
        let backend = create_test_backend().await?;

        let id = make_id();
        let lt_id = ObjectId::random(id.context().clone());
        let wrong_id = ObjectId::random(id.context().clone());

        let tombstone = Tombstone {
            target: lt_id.clone(),
            time_expires: None,
        };
        create_tombstone(&backend, &id, &tombstone).await?;

        // Wrong target: CAS fails, tombstone intact.
        let write = TieredWrite::Object(Metadata::default(), Bytes::new());
        let swapped = backend
            .compare_and_write(&id, Some(&wrong_id), write, Timestamp::now())
            .await?;
        assert!(!swapped, "expected CAS failure with wrong target");
        assert!(matches!(
            backend.get_tiered_metadata(&id, Timestamp::now()).await?,
            TieredMetadata::Tombstone(_)
        ));

        // Correct target: CAS succeeds, row becomes an inline object.
        let payload = Bytes::from_static(b"hello inline");
        let write = TieredWrite::Object(Metadata::default(), payload.clone());
        let swapped = backend
            .compare_and_write(&id, Some(&lt_id), write.clone(), Timestamp::now())
            .await?;
        assert!(swapped, "expected CAS success with correct target");
        let TieredGet::Object(_, _, stream) = backend
            .get_tiered_object(&id, Timestamp::now(), None)
            .await?
        else {
            panic!("expected inline object after swap");
        };
        assert_eq!(&stream::read_to_vec(stream).await?, payload.as_ref());

        // Idempotent retry: row is already inline (no tombstone), same CAS returns true.
        let retry = backend
            .compare_and_write(&id, Some(&lt_id), write, Timestamp::now())
            .await?;
        assert!(retry, "idempotent retry");

        Ok(())
    }

    /// CAS-write an object onto an empty row (expected=None, write=Object) succeeds.
    #[tokio::test]
    async fn test_cas_create_object_on_empty_row() -> Result<()> {
        let backend = create_test_backend().await?;

        let id = make_id();
        let payload = Bytes::from_static(b"cas object");
        let write = TieredWrite::Object(Metadata::default(), payload.clone());
        let committed = backend
            .compare_and_write(&id, None, write, Timestamp::now())
            .await?;
        assert!(committed, "expected CAS success on empty row");

        let TieredGet::Object(_, _, stream) = backend
            .get_tiered_object(&id, Timestamp::now(), None)
            .await?
        else {
            panic!("expected Object after CAS-create");
        };
        assert_eq!(&stream::read_to_vec(stream).await?, payload.as_ref());

        Ok(())
    }

    /// CAS-delete: wrong expected → false; correct expected → true, row gone.
    #[tokio::test]
    async fn test_cas_delete() -> Result<()> {
        let backend = create_test_backend().await?;

        let id = make_id();
        let lt_id = ObjectId::random(id.context().clone());
        let wrong_id = ObjectId::random(id.context().clone());

        let tombstone = Tombstone {
            target: lt_id.clone(),
            time_expires: None,
        };
        create_tombstone(&backend, &id, &tombstone).await?;

        // Wrong target: fails, row preserved.
        let deleted = backend
            .compare_and_write(&id, Some(&wrong_id), TieredWrite::Delete, Timestamp::now())
            .await?;
        assert!(!deleted, "expected CAS failure with wrong target");
        assert!(matches!(
            backend.get_tiered_metadata(&id, Timestamp::now()).await?,
            TieredMetadata::Tombstone(_)
        ));

        // Correct target: succeeds, row gone.
        let deleted = backend
            .compare_and_write(&id, Some(&lt_id), TieredWrite::Delete, Timestamp::now())
            .await?;
        assert!(deleted, "expected CAS delete success");
        assert!(matches!(
            backend.get_tiered_metadata(&id, Timestamp::now()).await?,
            TieredMetadata::NotFound
        ));

        // Idempotent retry: row is already absent (no tombstone), same delete returns true.
        let retry = backend
            .compare_and_write(&id, Some(&lt_id), TieredWrite::Delete, Timestamp::now())
            .await?;
        assert!(retry, "idempotent retry");

        // Inline object replaced tombstone: Safe to delete since it is an idempotent operation.
        let id2 = make_id();
        let fake_lt_id = ObjectId::random(id2.context().clone());
        let metadata = Metadata::default();
        create_object(&backend, &id2, &metadata, b"data", Timestamp::now()).await?;
        let deleted = backend
            .compare_and_write(
                &id2,
                Some(&fake_lt_id),
                TieredWrite::Delete,
                Timestamp::now(),
            )
            .await?;
        assert!(deleted, "expected idempotent deletion");

        Ok(())
    }

    // --- Section 5: Legacy Tombstone Compatibility ---

    /// Legacy Manual and TTL tombstones are correctly read via the tiered APIs.
    ///
    /// Uses `Manual` expiration so `timestamp_micros = -1` (server-assigned ≈ write time)
    /// does not trigger immediate expiry.
    #[tokio::test]
    async fn test_legacy_tombstone_reads() -> Result<()> {
        let backend = create_test_backend().await?;

        // Manual policy: get_tiered_metadata returns a non-expiring tombstone.
        let id = make_id();
        write_legacy_tombstone(&backend, &id, ExpirationPolicy::Manual, None).await?;

        let TieredMetadata::Tombstone(t) =
            backend.get_tiered_metadata(&id, Timestamp::now()).await?
        else {
            panic!("expected tombstone");
        };
        assert_eq!(t.time_expires, None);
        assert!(matches!(
            backend
                .get_tiered_object(&id, Timestamp::now(), None)
                .await?,
            TieredGet::Tombstone(_)
        ));

        // TTL policy: get_tiered_metadata reconstructs the concrete deadline.
        //
        // A future cell timestamp (now + TTL) is required so `expires_before` does not
        // immediately filter the row.
        let id = make_id();
        let ttl = Duration::from_hours(2 * 24);
        write_legacy_tombstone(&backend, &id, ExpirationPolicy::TimeToLive(ttl), None).await?;

        let TieredMetadata::Tombstone(t) =
            backend.get_tiered_metadata(&id, Timestamp::now()).await?
        else {
            panic!("expected TieredMetadata::Tombstone");
        };
        assert!(t.time_expires.is_some());

        Ok(())
    }

    /// A conditional extension upgrades a legacy TTI tombstone to `r`.
    #[tokio::test]
    async fn test_legacy_tombstone_tti_upgrade() -> Result<()> {
        let backend = create_test_backend().await?;
        let id = make_id();
        let path = id.as_storage_path().to_string().into_bytes();

        let tti = Duration::from_hours(2 * 24);

        // Place time_expires near expiry but still in the future.
        let old_deadline = Timestamp::now() + Duration::from_mins(1);
        write_legacy_tombstone(
            &backend,
            &id,
            ExpirationPolicy::TimeToIdle(tti),
            Some(old_deadline),
        )
        .await?;

        // A read observes the legacy row but leaves it unchanged.
        let TieredMetadata::Tombstone(_) =
            backend.get_tiered_metadata(&id, Timestamp::now()).await?
        else {
            panic!("expected tombstone");
        };
        assert_eq!(
            backend
                .read_row(&path, "test-verify", Timestamp::now(), None)
                .await?
                .and_then(|row| row.time_expires()),
            Some(old_deadline)
        );

        let requested = Timestamp::now() + tti;
        assert_eq!(
            backend
                .compare_and_update(
                    &id,
                    Some(&id),
                    TieredUpdate::SetExpiry(ExpiryTarget::At(requested).into()),
                    Timestamp::now()
                )
                .await?,
            SetExpiryResponse::Satisfied(requested)
        );

        // After extension, the row uses the requested timestamp.
        let new_deadline = match backend
            .read_row(&path, "test-verify", Timestamp::now(), None)
            .await?
        {
            Some(RowData::Tombstone { time_expires, .. }) => time_expires.unwrap(),
            _ => panic!("expected tombstone row after extension"),
        };

        assert!(
            new_deadline > old_deadline,
            "explicit extension should extend tombstone expiry: {old_deadline:?} -> {new_deadline:?}"
        );

        Ok(())
    }

    /// Legacy tombstones are handled correctly by all conditional write operations.
    ///
    /// Covers: `put_non_tombstone`, `delete_non_tombstone`, CAS-delete for both the
    /// legacy-metadata format and the empty-redirect format.
    #[tokio::test]
    async fn test_legacy_tombstone_conditional_ops() -> Result<()> {
        let backend = create_test_backend().await?;

        // put_non_tombstone returns Some(target == id) for a legacy tombstone.
        let id = make_id();
        write_legacy_tombstone(&backend, &id, ExpirationPolicy::Manual, None).await?;
        let t_opt = backend
            .put_non_tombstone(&id, &Metadata::default(), Bytes::new(), Timestamp::now())
            .await?;
        assert_eq!(t_opt.map(|t| t.target).as_ref(), Some(&id));

        // delete_non_tombstone returns Some(target == id) for a legacy tombstone.
        let id = make_id();
        write_legacy_tombstone(&backend, &id, ExpirationPolicy::Manual, None).await?;
        let t_opt = backend.delete_non_tombstone(&id, Timestamp::now()).await?;
        assert_eq!(t_opt.map(|t| t.target).as_ref(), Some(&id));

        // CAS-delete succeeds on a legacy-metadata tombstone (target resolves to hv_id).
        let id = make_id();
        write_legacy_tombstone(&backend, &id, ExpirationPolicy::Manual, None).await?;
        let deleted = backend
            .compare_and_write(&id, Some(&id), TieredWrite::Delete, Timestamp::now())
            .await?;
        assert!(
            deleted,
            "CAS-delete must succeed on legacy-metadata tombstone"
        );
        assert!(matches!(
            backend.get_tiered_metadata(&id, Timestamp::now()).await?,
            TieredMetadata::NotFound
        ));

        // CAS-delete succeeds on an empty-redirect tombstone (target resolves to hv_id).
        let id = make_id();
        write_empty_redirect_tombstone(&backend, &id).await?;
        let deleted = backend
            .compare_and_write(&id, Some(&id), TieredWrite::Delete, Timestamp::now())
            .await?;
        assert!(
            deleted,
            "CAS-delete must succeed on empty-redirect tombstone"
        );
        assert!(matches!(
            backend.get_tiered_metadata(&id, Timestamp::now()).await?,
            TieredMetadata::NotFound
        ));

        Ok(())
    }

    /// An empty `r` value falls back to the HV id when resolving the tombstone target.
    #[tokio::test]
    async fn test_empty_redirect_falls_back_to_hv_id() -> Result<()> {
        let backend = create_test_backend().await?;
        let id = make_id();

        write_empty_redirect_tombstone(&backend, &id).await?;
        match backend.get_tiered_metadata(&id, Timestamp::now()).await? {
            TieredMetadata::Tombstone(t) => assert_eq!(t.target, id, "must fall back to hv_id"),
            other => panic!("expected tombstone, got {other:?}"),
        }

        Ok(())
    }

    // --- Section 6: Expired Tombstone Handling ---

    /// CAS with `current=None` must succeed when the row holds an expired
    /// tombstone. The physical row still exists but is logically gone.
    #[tokio::test]
    async fn test_cas_create_tombstone_over_expired() -> Result<()> {
        let backend = create_test_backend().await?;

        let id = make_id();
        let old_lt_id = ObjectId::random(id.context().clone());
        let old_tombstone = Tombstone {
            target: old_lt_id,
            time_expires: Some(Timestamp::now() - Duration::from_secs(1)),
        };
        create_tombstone(&backend, &id, &old_tombstone).await?;

        let new_lt_id = ObjectId::random(id.context().clone());
        let new_tombstone = Tombstone {
            target: new_lt_id.clone(),
            time_expires: Some(Timestamp::now() + Duration::from_hours(1)),
        };
        let committed = backend
            .compare_and_write(
                &id,
                None,
                TieredWrite::Tombstone(new_tombstone),
                Timestamp::now(),
            )
            .await?;
        assert!(
            committed,
            "CAS with current=None must succeed over an expired tombstone"
        );

        let TieredMetadata::Tombstone(t) =
            backend.get_tiered_metadata(&id, Timestamp::now()).await?
        else {
            panic!("expected new tombstone to be readable");
        };
        assert_eq!(t.target, new_lt_id);

        Ok(())
    }

    /// `put_non_tombstone` must succeed when the row holds only an expired
    /// tombstone — the expired row is logically absent.
    #[tokio::test]
    async fn test_put_non_tombstone_over_expired() -> Result<()> {
        let backend = create_test_backend().await?;

        let id = make_id();
        let lt_id = ObjectId::random(id.context().clone());
        let tombstone = Tombstone {
            target: lt_id,
            time_expires: Some(Timestamp::now() - Duration::from_secs(1)),
        };
        create_tombstone(&backend, &id, &tombstone).await?;

        let result = backend
            .put_non_tombstone(
                &id,
                &Metadata::default(),
                Bytes::from_static(b"data"),
                Timestamp::now(),
            )
            .await?;
        assert_eq!(
            result, None,
            "put_non_tombstone must succeed (return None) over an expired tombstone"
        );

        let (_, _, stream) = backend
            .get_object(&id, Timestamp::now(), None)
            .await?
            .unwrap();
        assert_eq!(&stream::read_to_vec(stream).await?, b"data");

        Ok(())
    }

    // --- Range Request Tests ---

    async fn put_range_test_object(backend: &BigTableBackend) -> Result<ObjectId> {
        let id = make_id();
        let metadata = Metadata {
            content_type: "text/plain".into(),
            ..Default::default()
        };
        let payload = b"Hello, range requests!";
        backend
            .put_object(
                &id,
                &metadata,
                stream::single(payload.as_slice()),
                Timestamp::now(),
            )
            .await?;
        Ok(id)
    }

    #[tokio::test]
    async fn get_object_range_bounded() -> Result<()> {
        let backend = create_test_backend().await?;
        let id = put_range_test_object(&backend).await?;

        let (_, content_range, stream) = backend
            .get_object(&id, Timestamp::now(), Some(ByteRange::Bounded(7, 11)))
            .await?
            .unwrap();
        let data = stream::read_to_vec(stream).await?;
        assert_eq!(&data, b"range");

        let content_range = content_range.unwrap();
        assert_eq!(content_range.start, 7);
        assert_eq!(content_range.end, 11);
        assert_eq!(content_range.total, 22);

        Ok(())
    }

    #[tokio::test]
    async fn get_object_range_from() -> Result<()> {
        let backend = create_test_backend().await?;
        let id = put_range_test_object(&backend).await?;

        let (_, content_range, stream) = backend
            .get_object(&id, Timestamp::now(), Some(ByteRange::From(7)))
            .await?
            .unwrap();
        let data = stream::read_to_vec(stream).await?;
        assert_eq!(&data, b"range requests!");

        let content_range = content_range.unwrap();
        assert_eq!(content_range.start, 7);
        assert_eq!(content_range.end, 21);
        assert_eq!(content_range.total, 22);

        Ok(())
    }

    #[tokio::test]
    async fn get_object_range_last() -> Result<()> {
        let backend = create_test_backend().await?;
        let id = put_range_test_object(&backend).await?;

        let (_, content_range, stream) = backend
            .get_object(&id, Timestamp::now(), Some(ByteRange::Last(9)))
            .await?
            .unwrap();
        let data = stream::read_to_vec(stream).await?;
        assert_eq!(&data, b"requests!");

        let content_range = content_range.unwrap();
        assert_eq!(content_range.start, 13);
        assert_eq!(content_range.end, 21);
        assert_eq!(content_range.total, 22);

        Ok(())
    }

    #[tokio::test]
    async fn get_object_range_unsatisfiable() -> Result<()> {
        let backend = create_test_backend().await?;
        let id = put_range_test_object(&backend).await?;

        match backend
            .get_object(&id, Timestamp::now(), Some(ByteRange::From(100)))
            .await
        {
            Err(error) if matches!(error.kind(), ErrorKind::RangeNotSatisfiable { total: 22 }) => {}
            Ok(_) => panic!("expected RangeNotSatisfiable, got Ok"),
            Err(e) => panic!("expected RangeNotSatisfiable, got {e:?}"),
        }

        Ok(())
    }

    #[tokio::test]
    async fn get_object_no_range_returns_full_payload() -> Result<()> {
        let backend = create_test_backend().await?;
        let id = put_range_test_object(&backend).await?;

        let (_, content_range, stream) = backend
            .get_object(&id, Timestamp::now(), None)
            .await?
            .unwrap();
        let data = stream::read_to_vec(stream).await?;
        assert_eq!(&data, b"Hello, range requests!");
        assert!(content_range.is_none());

        Ok(())
    }

    #[test]
    fn row_size_counts_the_key_and_every_cell() {
        let path = b"attachments/org.1/objects/abc";
        let (mutations, size) =
            object_mutations(path, Metadata::default(), b"0123456789".to_vec()).unwrap();

        // The key, the 10-byte payload, and the serialized metadata. `object_mutations`
        // stamps the size into the metadata before serializing it, so the expected length
        // has to account for that too.
        let stamped = Metadata {
            size: Some(10),
            ..Default::default()
        };
        let metadata_len = serde_json::to_vec(&stamped).unwrap().len();
        let expected = (path.len() + 10 + metadata_len) as u64;

        assert_eq!(row_size(path, &mutations), expected);
        assert_eq!(size, expected, "the size handed back matches the mutations");
    }

    #[test]
    fn row_size_is_nonzero_for_tombstones() {
        let path = b"attachments/org.1/objects/abc";
        let time_expires = Timestamp::now() + Duration::from_secs(60);
        let tombstone = Tombstone {
            target: ObjectId::from_storage_path("attachments/org.1/objects/abc/0199").unwrap(),
            time_expires: Some(time_expires),
        };
        let mutations = tombstone_mutations(&tombstone);

        assert_eq!(mutations.len(), 2);
        let set_cell = mutations[1].mutation.as_ref().unwrap();
        let mutation::Mutation::SetCell(set_cell) = set_cell else {
            panic!("expected redirect SetCell mutation");
        };
        assert_eq!(set_cell.family_name, FAMILY_GC);
        assert_eq!(set_cell.column_qualifier, COLUMN_REDIRECT);
        assert_eq!(set_cell.timestamp_micros, time_expires.as_micros() as i64);
        assert!(row_size(path, &mutations) > path.len() as u64);
    }

    #[cfg(feature = "storage-cogs")]
    #[tokio::test]
    async fn change_stream_reports_writes_and_deletes() -> Result<()> {
        let (backend, producer) = create_test_backend_with_change_stream().await?;
        let id = make_id();
        let metadata = Metadata {
            expiration_policy: ExpirationPolicy::TimeToLive(Duration::from_secs(3600)),
            time_expires: Some(Timestamp::now() + Duration::from_secs(3600)),
            ..Default::default()
        };

        backend
            .put_object(
                &id,
                &metadata,
                stream::single::<crate::stream::ClientError>(b"hello".to_vec()),
                Timestamp::now(),
            )
            .await?;
        backend.delete_object(&id, Timestamp::now()).await?;

        let records = producer.records();
        assert_eq!(records.len(), 2);

        assert_eq!(records[0].op_type, OpType::Write);
        assert_eq!(records[0].app_feature, "testing");
        assert_eq!(records[0].shared_resource_id, "bigtable_objectstore");
        // Key plus payload plus metadata, so strictly more than the payload alone.
        assert!(records[0].size.unwrap() > b"hello".len() as u64);
        assert!(records[0].expiration_time.is_some());

        assert_eq!(records[1].op_type, OpType::Delete);
        assert_eq!(records[1].size, None);
        assert_eq!(records[1].record_id, records[0].record_id);

        Ok(())
    }

    #[cfg(feature = "storage-cogs")]
    #[tokio::test]
    async fn delete_non_tombstone_reclaims_expired_rows() -> Result<()> {
        let (backend, producer) = create_test_backend_with_change_stream().await?;

        // An expired tombstone is past its lifetime: reclaimed, not handed to the caller.
        // This test serves as documentation of that potentially surprising behavior.
        let id = make_id();
        let tombstone = Tombstone {
            target: ObjectId::random(id.context().clone()),
            time_expires: Some(Timestamp::now() - Duration::from_secs(1)),
        };
        create_tombstone(&backend, &id, &tombstone).await?;
        assert_eq!(
            backend.delete_non_tombstone(&id, Timestamp::now()).await?,
            None,
            "an expired tombstone must not be returned to the caller"
        );
        let records = producer.records();
        assert_eq!(records.len(), 1, "the expired tombstone must be reclaimed");
        assert_eq!(records[0].op_type, OpType::Delete);

        // The same holds for an object row (expired or otherwise).
        let id = make_id();
        let metadata = Metadata {
            expiration_policy: ExpirationPolicy::TimeToLive(Duration::from_secs(0)),
            time_expires: Some(Timestamp::now() - Duration::from_secs(1)),
            ..Default::default()
        };
        create_object(&backend, &id, &metadata, b"gone", Timestamp::now()).await?;
        producer.clear();
        assert_eq!(
            backend.delete_non_tombstone(&id, Timestamp::now()).await?,
            None
        );
        let records = producer.records();
        assert_eq!(records.len(), 1, "the object row must be reclaimed");
        assert_eq!(records[0].op_type, OpType::Delete);

        Ok(())
    }

    #[cfg(feature = "storage-cogs")]
    #[tokio::test]
    async fn change_stream_reports_tombstone_rows() -> Result<()> {
        let (backend, producer) = create_test_backend_with_change_stream().await?;
        let id = make_id();
        let target = new_test_revision(&id);

        let tombstone = Tombstone {
            target: target.clone(),
            time_expires: Some(Timestamp::now() + Duration::from_secs(3600)),
        };
        let written = backend
            .compare_and_write(
                &id,
                None,
                TieredWrite::Tombstone(tombstone),
                Timestamp::now(),
            )
            .await?;
        assert!(written);

        let records = producer.records();
        assert_eq!(records.len(), 1);
        assert_eq!(records[0].op_type, OpType::Write);
        assert!(
            records[0].size.unwrap() > 0,
            "tombstone rows occupy storage and must not report zero"
        );
        assert!(records[0].expiration_time.is_some());

        Ok(())
    }

    #[cfg(feature = "storage-cogs")]
    #[tokio::test]
    async fn change_stream_reports_expiry_extension_as_an_update() -> Result<()> {
        let (backend, producer) = create_test_backend_with_change_stream().await?;
        let id = make_id();
        let metadata = Metadata {
            expiration_policy: ExpirationPolicy::TimeToIdle(Duration::from_secs(3600)),
            time_expires: Some(Timestamp::now() + Duration::from_secs(1)),
            ..Default::default()
        };

        backend
            .put_object(
                &id,
                &metadata,
                stream::single::<crate::stream::ClientError>(b"hello".to_vec()),
                Timestamp::now(),
            )
            .await?;
        producer.clear();

        backend
            .set_expiry(
                &id,
                ExpiryTarget::At(Timestamp::now() + Duration::from_secs(3600)).into(),
                Timestamp::now(),
            )
            .await?;

        let records = producer.records();
        assert_eq!(records.len(), 1, "expected exactly one extension report");
        assert_eq!(records[0].op_type, OpType::Update);
        assert_eq!(
            records[0].size, None,
            "an extension does not change the size"
        );
        assert!(records[0].expiration_time.is_some());

        Ok(())
    }

    #[cfg(feature = "storage-cogs")]
    fn new_test_revision(id: &ObjectId) -> ObjectId {
        ObjectId {
            context: id.context.clone(),
            key: format!("{}/{}", id.key, uuid::Uuid::now_v7()),
        }
    }
}