diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 0d9aa875..c24ebbb7 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -332,6 +332,12 @@ jobs: (set -a; . "$TC887_RUN_DIR/harness.env"; set +a cargo test --locked -p tinycloud-node --test s3_block_store -- --ignored) + - name: SQL object store (against the harness MinIO) + timeout-minutes: 10 + run: | + (set -a; . "$TC887_RUN_DIR/harness.env"; set +a + cargo test --locked -p tinycloud-node --test sql_object_store -- --ignored) + - name: Harness logs if: failure() run: | diff --git a/CHANGELOG.md b/CHANGELOG.md index 4bf054e7..3b7f7cb3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## [Unreleased] +- Added the SQL object store backends for S3 (conditional create with a startup probe and a check-then-create fallback) and the filesystem (atomic no-clobber create), with the contract rules in `docs/sql-block-store-contract.md` section 12. The S3 block store's start-up size scan lists the `tinycloud:` key prefix only. Details: TC-1491. + ## [1.24.0] - 2026-10-25 - KV puts persist content-addressed blobs before taking the per-space sequence lock, and committed `kv/sync` cursors stay ordered. A preparatory transaction authorizes the write and claims every distinct blob under a durable attempt, committing all claims or none; on PostgreSQL each claim statement also checks that the invocation's spaces are hosted. Uploads hold no pooled database connection, transaction or lock. Protected legacy meeting writes keep their pre-upload freeze barrier, which waits for commit or abort and fails after 30 seconds if stalled (TC-1260). diff --git a/docs/sql-block-store-contract.md b/docs/sql-block-store-contract.md index abb31bdc..5febc3a7 100644 --- a/docs/sql-block-store-contract.md +++ b/docs/sql-block-store-contract.md @@ -698,3 +698,29 @@ These keys sit under `[sql_store]`. They are checked at boot only when `enabled PR A tests: `object_format_must_match_its_kind`, `segment_from_must_follow_base_seq` and `object_state_flip_checks_the_format` (11.1); `created_at_is_stored_canonically`, `created_at_out_of_range_is_invalid_not_a_panic` and `revive_restarts_the_orphan_clock` (11.2); `sql_store_refuses_enable_mysql_and_non_canonical_spaces` (11.3); `sql_store_defaults_match_the_contract`, `sql_store_new_setting_defaults_and_pg_concurrency_floor`, `sql_store_new_settings_parse_and_are_bounded`, `sql_store_settings_are_checked_only_when_enabled` and `sql_store_new_settings_are_bounded_when_enabled` (11.4). The manifest tests run against SQLite and PostgreSQL. +## 12. Object store backends + +Slice 1b PR B adds the two backends of the object store trait in `tinycloud-node-server/src/storage/sql_object.rs`. Neither is started by the node yet. + +### 12.1 S3 + +- The store uses the block store's bucket and client settings. The object key is the key of section 3.1, unchanged. The object carries its lowercase hex SHA-256 as user metadata `sha256`. +- A body that does not hash to the key is refused with `BodyMismatch` before any request is sent. +- In conditional mode an object is created with one PUT carrying `If-None-Match: *`. A 412 answer is accepted only after a HEAD shows the stored object has the same length and the same `sha256` metadata (`AlreadyPresent`); any other stored object gives `Conflict`. A stored object without the metadata is a backend error, never a match. +- The startup probe writes a fixed small body to `sql/v1/_probe/v1` twice with `If-None-Match: *`. A 412 on the second request selects conditional mode. A second success, or a 501 or NotImplemented answer for the header, selects check-then-create mode. Any other failure, any other status included, is an error and the store does not start. The result is logged once at info level (target `tinycloud::sql_store`) with the mode and the status and code of the answer that decided it. The probe object is not deleted. +- In check-then-create mode the store sends a HEAD first. An absent object is created with an unconditional PUT; an existing one is compared as above. Creators of one key in one process run one at a time in both modes; creators in different processes can each report `Created` for the same key in check-then-create mode, and each stores the bytes the key names. +- In conditional mode a 409 `ConditionalRequestConflict` answer (another conditional write of the key is in flight) is retried up to 5 times; any other status is a backend error with its status and code. + +### 12.2 Filesystem + +- The root is `{datadir}/sql-store`. A relative root or block store root is made absolute against the current directory first (a lexical step that reads nothing from the filesystem; `..` components are refused). Boot then refuses a root that is equal to, inside, or a parent of the block store root. The filesystem store is not supported on Windows. +- The path of an object is the root joined with its key. Every key component must be an ordinary file name, and the resulting path must stay under the root. +- A create writes the body to a `.tmp-` file in the target directory, fsyncs it, and publishes it with a hard link, which fails instead of replacing an existing file. An existing file with the same length and SHA-256 gives `AlreadyPresent`, any other gives `Conflict`. The directory is fsynced and the temporary file is removed in every outcome. Each new directory and its parent are fsynced when created, and the parent of a directory that already exists is fsynced too. A leftover temporary file does not affect a later create. + +### 12.3 Ranges and head + +Both backends answer a range as the in-memory store does: the range is resolved against the object size and clamped to it, and a range that selects no byte (start at or beyond the end, an inverted range, a zero-length suffix, any range of an empty object) and a missing object both give `None`. `head` returns the length and SHA-256; the filesystem store hashes the file in bounded chunks. + +### 12.4 KV size scan + +The S3 block store's start-up size scan lists objects with the prefix `tinycloud:`. Space ids start with `tinycloud:`, and only keys whose part before the last `/` is a space id count towards a space, so the computed sizes are the same as for a scan of the whole bucket and `sql/v1/` objects are not listed. diff --git a/tinycloud-auth/src/resource.rs b/tinycloud-auth/src/resource.rs index f1262fec..d9befd34 100644 --- a/tinycloud-auth/src/resource.rs +++ b/tinycloud-auth/src/resource.rs @@ -281,9 +281,12 @@ pub enum ResourceCheckError { DoesNotExtendPath, } +/// The text every space id string starts with. +pub const SPACE_ID_PREFIX: &str = "tinycloud:"; + impl fmt::Display for SpaceId { fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { - write!(f, "tinycloud:{}:{}", self.suffix(), self.name) + write!(f, "{}{}:{}", SPACE_ID_PREFIX, self.suffix(), self.name) } } diff --git a/tinycloud-node-server/src/storage/mod.rs b/tinycloud-node-server/src/storage/mod.rs index 802dc7e4..439b0b2d 100644 --- a/tinycloud-node-server/src/storage/mod.rs +++ b/tinycloud-node-server/src/storage/mod.rs @@ -1,6 +1,7 @@ pub mod file_system; pub mod s3; pub mod size; +pub mod sql_object; use tinycloud_auth::resource::SpaceId; use tinycloud_core::hash::Hash; diff --git a/tinycloud-node-server/src/storage/s3.rs b/tinycloud-node-server/src/storage/s3.rs index 0fc0b57b..3d26f027 100644 --- a/tinycloud-node-server/src/storage/s3.rs +++ b/tinycloud-node-server/src/storage/s3.rs @@ -26,7 +26,7 @@ use std::{ sync::Arc, time::Instant, }; -use tinycloud_auth::resource::SpaceId; +use tinycloud_auth::resource::{SpaceId, SPACE_ID_PREFIX}; use tinycloud_core::{hash::Hash, storage::*}; use tokio_util::bytes::Bytes; @@ -37,6 +37,9 @@ use tokio::sync::Mutex; #[cfg(test)] use tokio::sync::Notify; +/// Prefix of the boot scan's listing: every KV object key starts with it. +pub(crate) const KV_SCAN_PREFIX: &str = SPACE_ID_PREFIX; + /// Pinned so an SDK bump cannot change client defaults unnoticed. pub fn behavior_version() -> BehaviorVersion { BehaviorVersion::v2026_01_12() @@ -115,7 +118,7 @@ impl StorageSetup for S3BlockStore { } } -async fn new_client(config: &S3BlockConfig) -> Client { +pub(crate) async fn new_client(config: &S3BlockConfig) -> Client { let general_config = aws_config().await; // Checksums only when an operation requires them: the SDK 1.x defaults // (CRC32 trailers, aws-chunked uploads, checksum-mode GETs) are rejected @@ -163,6 +166,9 @@ impl S3BlockStore { let mut pages = client .list_objects_v2() .bucket(&config.bucket) + // Every KV object key is `{space}/{hash}` and a space id string + // starts with `SPACE_ID_PREFIX`; only such keys count towards a space. + .prefix(KV_SCAN_PREFIX) .into_paginator() .send(); // get the sum of all objects per space across all pages, and the @@ -202,7 +208,7 @@ impl S3BlockStore { }) } - fn key_for(space: &SpaceId, id: &Hash, generation: i64) -> String { + pub(crate) fn key_for(space: &SpaceId, id: &Hash, generation: i64) -> String { let mut key = format!("{space}/"); // Hash digests fit in 64 bytes; encode on the stack so generated // reads allocate only their storage-key String. diff --git a/tinycloud-node-server/src/storage/sql_object.rs b/tinycloud-node-server/src/storage/sql_object.rs new file mode 100644 index 00000000..49ef78fd --- /dev/null +++ b/tinycloud-node-server/src/storage/sql_object.rs @@ -0,0 +1,1108 @@ +//! Object stores for the private SQL namespace `sql/v1/` (TC-1491 contract, +//! sections 3, 6 and 11.8). +//! +//! Both backends implement [`SqlObjectStore`] and share one rule: an object is +//! created at most once and never replaced. A key names its content (it ends +//! in the SHA-256 of the bytes), so a second creator of the same key either +//! stores identical bytes (`AlreadyPresent`) or finds different ones +//! (`Conflict`); nothing is ever overwritten. +//! +//! * [`S3SqlObjectStore`] writes into the block store's bucket, key as is, +//! with a conditional create (`If-None-Match: *`). [`S3SqlObjectStore::probe`] +//! checks at startup whether the provider honours the header and selects +//! the write mode; without it the store checks, then creates. +//! * [`FsSqlObjectStore`] writes below `{datadir}/sql-store`: the body goes to +//! a temporary file in the target directory, is fsynced, and is published +//! with a hard link, which fails instead of replacing an existing file. + +use std::{ + collections::HashMap, + fmt, + fs::{File, OpenOptions}, + future::Future, + io::{ErrorKind, Read, Seek, SeekFrom, Write}, + path::{Component, Path, PathBuf}, + sync::{Arc, Mutex, MutexGuard, PoisonError}, + time::Duration, +}; + +use async_trait::async_trait; +use aws_sdk_s3::{ + error::{ProvideErrorMetadata, SdkError}, + operation::{ + get_object::GetObjectError, head_object::HeadObjectError, put_object::PutObjectError, + }, + primitives::ByteStream, + Client, +}; +use sha2::{Digest, Sha256}; +use tinycloud_core::{ + sql::store::{ + key::KEY_ROOT, object::sha256, ObjectKey, ObjectMeta, ObjectRange, PutOutcome, + SqlObjectStore, SqlStoreError, + }, + storage::ByteRangeSpec, +}; + +use super::s3::{new_client, S3BlockConfig}; + +/// Tracing target of everything the SQL store logs. +const LOG_TARGET: &str = "tinycloud::sql_store"; + +/// How `put_if_absent` creates an object on S3. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum WriteMode { + /// One conditional PUT (`If-None-Match: *`); the provider refuses to + /// replace an existing object. + Conditional, + /// HEAD, then an unconditional PUT when the object is absent. + CheckThenWrite, +} + +impl WriteMode { + pub fn as_str(self) -> &'static str { + match self { + Self::Conditional => "conditional", + Self::CheckThenWrite => "check_then_write", + } + } +} + +// --------------------------------------------------------------------------- +// S3 +// --------------------------------------------------------------------------- + +/// Key of the object the startup probe writes. It is never deleted. +const PROBE_KEY: &str = "sql/v1/_probe/v1"; +const PROBE_BODY: &[u8] = b"tinycloud sql store probe v1"; +/// A conditional PUT that loses a race with an in-flight conditional PUT of +/// the same key may be answered 409; it is retried. +const CONFLICT_RETRIES: u32 = 5; + +pub struct S3SqlObjectStore { + client: Client, + bucket: String, + mode: WriteMode, + /// Per-key gates: creators of one key in this process go one at a time. + gates: Gates, +} + +type Gates = Mutex>>>; + +fn lock_gates(gates: &Gates) -> MutexGuard<'_, HashMap>>> { + gates.lock().unwrap_or_else(PoisonError::into_inner) +} + +/// A key's gate, removed from the map when the last holder or waiter lets go, +/// including when its future is dropped part-way. +struct GateGuard<'a> { + gates: &'a Gates, + key: &'a str, + gate: Arc>, +} + +impl<'a> GateGuard<'a> { + fn new(gates: &'a Gates, key: &'a str) -> Self { + let gate = lock_gates(gates).entry(key.to_owned()).or_default().clone(); + Self { gates, key, gate } + } + + async fn lock(&self) -> tokio::sync::MutexGuard<'_, ()> { + self.gate.lock().await + } +} + +impl Drop for GateGuard<'_> { + fn drop(&mut self) { + let mut gates = lock_gates(self.gates); + // The map holds one reference and this guard one; any other is a waiter. + if Arc::strong_count(&self.gate) == 2 { + gates.remove(self.key); + } + } +} + +/// The status and error code of a response. +struct Answer { + status: Option, + code: Option, +} + +impl Answer { + fn of(error: &SdkError) -> Self { + Self { + status: status(error), + code: ProvideErrorMetadata::code(error).map(str::to_owned), + } + } +} + +impl fmt::Display for Answer { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self.status { + Some(status) => write!(f, "status {status}")?, + None => write!(f, "status none")?, + } + write!(f, ", code {}", self.code.as_deref().unwrap_or("none")) + } +} + +/// What one PUT of the probe or of an object came back as. +enum PutResult { + Created, + /// 412: the object exists, the condition was honoured. + PreconditionFailed(Answer), + /// 409 ConditionalRequestConflict: another conditional write of the key + /// is in flight. + InFlight(Answer), + /// The provider does not implement the header (501, NotImplemented). + Unsupported(Answer), + Failed(SqlStoreError), +} + +fn status(error: &SdkError) -> Option { + error.raw_response().map(|r| r.status().as_u16()) +} + +/// Which stage failed, without the message or URL (they can carry the key). +fn failure_kind(error: &SdkError) -> &'static str { + match error { + SdkError::ConstructionFailure(_) => "construction", + SdkError::TimeoutError(_) => "timeout", + SdkError::DispatchFailure(f) if f.is_timeout() => "dispatch timeout", + SdkError::DispatchFailure(f) if f.is_io() => "dispatch io", + SdkError::DispatchFailure(f) if f.is_user() => "dispatch user", + SdkError::DispatchFailure(_) => "dispatch", + SdkError::ResponseError(_) => "response", + SdkError::ServiceError(_) => "service", + _ => "other", + } +} + +fn backend(op: &str, error: &SdkError) -> SqlStoreError { + let status = status(error).map_or_else(|| "none".to_owned(), |s| s.to_string()); + let code = ProvideErrorMetadata::code(error).unwrap_or("none"); + let mut cause = String::new(); + if let SdkError::DispatchFailure(failure) = error { + let mut source = failure + .as_connector_error() + .map(|e| e as &dyn std::error::Error); + while let Some(error) = source { + cause.push_str(&format!(": {error}")); + source = error.source(); + } + } + SqlStoreError::Backend(format!( + "s3 {op} failed ({}, status {status}, code {code}){cause}", + failure_kind(error) + )) +} + +fn classify_put(result: Result>) -> PutResult { + let error = match result { + Ok(_) => return PutResult::Created, + Err(error) => error, + }; + let answer = Answer::of(&error); + match (answer.status, answer.code.as_deref()) { + (Some(412), _) | (_, Some("PreconditionFailed")) => PutResult::PreconditionFailed(answer), + (Some(409), Some("ConditionalRequestConflict")) => PutResult::InFlight(answer), + // Only 501 and NotImplemented mean the header is not supported; any + // other status, 400 included, is a failure. + (Some(501), _) | (_, Some("NotImplemented")) => PutResult::Unsupported(answer), + _ => PutResult::Failed(backend("put", &error)), + } +} + +async fn put_object( + client: &Client, + bucket: &str, + key: &str, + body: Vec, + conditional: bool, +) -> PutResult { + let sha = hex::encode(sha256(&body)); + let request = client + .put_object() + .bucket(bucket) + .key(key) + .metadata("sha256", sha) + .body(ByteStream::from(body)); + let request = if conditional { + request.if_none_match("*") + } else { + request + }; + classify_put(request.send().await) +} + +/// Decide the write mode from the answers to two conditional PUTs of one key, +/// and name the answer that decided it. The second answering 412 means the +/// provider honours the condition; the second succeeding, or an answer of 501 +/// or NotImplemented, means it does not. A first 412 (the probe object exists +/// from an earlier start) already shows the condition is honoured. Any other +/// answer is an error. +async fn probe_with(mut put: F) -> Result<(WriteMode, String), SqlStoreError> +where + F: FnMut() -> Fut, + Fut: Future, +{ + for attempt in 0..2 { + match put().await { + PutResult::PreconditionFailed(answer) | PutResult::InFlight(answer) => { + return Ok((WriteMode::Conditional, answer.to_string())) + } + PutResult::Unsupported(answer) => { + return Ok((WriteMode::CheckThenWrite, answer.to_string())) + } + PutResult::Created if attempt == 1 => { + return Ok((WriteMode::CheckThenWrite, "created".to_owned())) + } + PutResult::Created => {} + PutResult::Failed(error) => return Err(error), + } + } + unreachable!("the second probe request always returns") +} + +/// An existing object is the one being created when length and SHA-256 agree. +fn same_object(meta: &ObjectMeta, key: &ObjectKey, len: u64) -> Result { + if meta.bytes == len && &meta.sha256 == key.sha256() { + Ok(PutOutcome::AlreadyPresent) + } else { + Err(SqlStoreError::Conflict) + } +} + +/// The first and last byte `range` selects in an object of `total` bytes, +/// clamped to the object; `None` when it selects no byte. +fn select_range(range: ByteRangeSpec, total: u64) -> Option<(u64, u64)> { + let last = total.checked_sub(1)?; + match range { + ByteRangeSpec::Inclusive { start, end } if start <= end && start < total => { + Some((start, end.min(last))) + } + ByteRangeSpec::From { start } if start < total => Some((start, last)), + ByteRangeSpec::Suffix { length } if length > 0 => Some((total - length.min(total), last)), + _ => None, + } +} + +fn parse_sha256(value: &str) -> Result<[u8; 32], SqlStoreError> { + hex::decode(value) + .ok() + .and_then(|bytes| <[u8; 32]>::try_from(bytes).ok()) + .ok_or_else(|| SqlStoreError::Backend("s3 object has an invalid sha256 metadata".into())) +} + +/// `bytes first-last/total`. +fn parse_content_range(value: &str) -> Result<(u64, u64, u64), SqlStoreError> { + let bad = || SqlStoreError::Backend(format!("s3 returned an invalid content range {value:?}")); + let rest = value.strip_prefix("bytes ").ok_or_else(bad)?; + let (bounds, total) = rest.split_once('/').ok_or_else(bad)?; + let (first, last) = bounds.split_once('-').ok_or_else(bad)?; + let first: u64 = first.parse().map_err(|_| bad())?; + let last: u64 = last.parse().map_err(|_| bad())?; + let total: u64 = total.parse().map_err(|_| bad())?; + if first > last || last >= total { + return Err(bad()); + } + Ok((first, last, total)) +} + +impl S3SqlObjectStore { + /// A store over the block store's bucket and client settings, with the + /// write mode chosen by [`Self::probe`]. + pub async fn open(config: &S3BlockConfig) -> Result { + let client = new_client(config).await; + let mode = Self::probe(&client, &config.bucket).await?; + Ok(Self::new(client, config.bucket.clone(), mode)) + } + + /// A store with the given write mode and no probe request. + pub async fn open_with_mode(config: &S3BlockConfig, mode: WriteMode) -> Self { + Self::new(new_client(config).await, config.bucket.clone(), mode) + } + + pub fn new(client: Client, bucket: String, mode: WriteMode) -> Self { + Self { + client, + bucket, + mode, + gates: Mutex::default(), + } + } + + pub fn mode(&self) -> WriteMode { + self.mode + } + + pub fn client(&self) -> &Client { + &self.client + } + + pub fn bucket(&self) -> &str { + &self.bucket + } + + /// Write a fixed small body to `sql/v1/_probe/v1` with `If-None-Match: *` + /// twice. A 412 on the second request means the provider honours + /// conditional writes; a second success, or an answer of 501 or + /// NotImplemented for the header, selects the check-then-create mode. + /// Any other failure, any other status included, is an error, not a + /// fallback. The object stays. The mode and the deciding answer's status + /// and code are logged once. + pub async fn probe(client: &Client, bucket: &str) -> Result { + let (mode, answer) = + probe_with(|| put_object(client, bucket, PROBE_KEY, PROBE_BODY.to_vec(), true)).await?; + tracing::info!( + target: LOG_TARGET, + mode = mode.as_str(), + answer = %answer, + "sql object store write mode" + ); + Ok(mode) + } + + async fn existing_matches( + &self, + key: &ObjectKey, + len: u64, + ) -> Result { + match self.head(key).await? { + Some(meta) => same_object(&meta, key, len), + None => Err(SqlStoreError::Backend( + "s3 object is missing after the create was refused".into(), + )), + } + } + + async fn create_conditionally( + &self, + key: &ObjectKey, + body: Vec, + ) -> Result { + let len = body.len() as u64; + for attempt in 0..=CONFLICT_RETRIES { + match put_object(&self.client, &self.bucket, key.as_str(), body.clone(), true).await { + PutResult::Created => return Ok(PutOutcome::Created), + PutResult::PreconditionFailed(_) => return self.existing_matches(key, len).await, + PutResult::InFlight(_) => { + tokio::time::sleep(Duration::from_millis(100 * (u64::from(attempt) + 1))).await + } + PutResult::Unsupported(answer) => { + return Err(SqlStoreError::Backend(format!("s3 put failed ({answer})"))) + } + PutResult::Failed(error) => return Err(error), + } + } + Err(SqlStoreError::Backend( + "s3 conditional create stayed in conflict".into(), + )) + } + + async fn check_then_create( + &self, + key: &ObjectKey, + body: Vec, + ) -> Result { + let len = body.len() as u64; + match self.head(key).await? { + Some(meta) => same_object(&meta, key, len), + None => match put_object(&self.client, &self.bucket, key.as_str(), body, false).await { + PutResult::Created => Ok(PutOutcome::Created), + PutResult::Failed(error) => Err(error), + _ => Err(SqlStoreError::Backend("s3 put was refused".into())), + }, + } + } +} + +#[async_trait] +impl SqlObjectStore for S3SqlObjectStore { + async fn put_if_absent( + &self, + key: &ObjectKey, + body: Vec, + ) -> Result { + if &sha256(&body) != key.sha256() { + return Err(SqlStoreError::BodyMismatch); + } + // Creators of one key in this process go one at a time. + let gate = GateGuard::new(&self.gates, key.as_str()); + let _held = gate.lock().await; + match self.mode { + WriteMode::Conditional => self.create_conditionally(key, body).await, + WriteMode::CheckThenWrite => self.check_then_create(key, body).await, + } + } + + async fn get_range( + &self, + key: &ObjectKey, + range: ByteRangeSpec, + ) -> Result, SqlStoreError> { + // Shapes that select nothing in any object need no request. + match range { + ByteRangeSpec::Inclusive { start, end } if start > end => return Ok(None), + ByteRangeSpec::Suffix { length: 0 } => return Ok(None), + _ => {} + } + let output = match self + .client + .get_object() + .bucket(&self.bucket) + .key(key.as_str()) + .range(range.to_http_value()) + .send() + .await + { + Ok(output) => output, + Err(error) => { + if matches!(error.as_service_error(), Some(GetObjectError::NoSuchKey(_))) + || status(&error) == Some(416) + || ProvideErrorMetadata::code(&error) == Some("InvalidRange") + { + return Ok(None); + } + return Err(backend("get", &error)); + } + }; + let content_range = output.content_range().map(str::to_owned); + let data = output + .body + .collect() + .await + .map_err(|_| SqlStoreError::Backend("s3 get body failed".into()))? + .into_bytes() + .to_vec(); + // A range that selects a byte always returns at least one: an empty + // answer means the object is empty, whatever header came with it. + if data.is_empty() { + return Ok(None); + } + match content_range { + Some(value) => { + let (first, last, total) = parse_content_range(&value)?; + if data.len() as u64 != last - first + 1 { + return Err(SqlStoreError::Backend( + "s3 get body does not match its range".into(), + )); + } + Ok(Some(ObjectRange { + total_size: total, + start: first, + data, + })) + } + // A provider that ignored the header sent the whole object. + None => { + let total = data.len() as u64; + Ok(select_range(range, total).map(|(first, last)| ObjectRange { + total_size: total, + start: first, + data: data[first as usize..=last as usize].to_vec(), + })) + } + } + } + + async fn head(&self, key: &ObjectKey) -> Result, SqlStoreError> { + let output = match self + .client + .head_object() + .bucket(&self.bucket) + .key(key.as_str()) + .send() + .await + { + Ok(output) => output, + Err(error) => { + if matches!(error.as_service_error(), Some(HeadObjectError::NotFound(_))) { + return Ok(None); + } + return Err(backend("head", &error)); + } + }; + let bytes = u64::try_from(output.content_length().unwrap_or_default()) + .map_err(|_| SqlStoreError::Backend("s3 object has a negative length".into()))?; + let sha = output + .metadata() + .and_then(|metadata| metadata.get("sha256")) + .ok_or_else(|| SqlStoreError::Backend("s3 object has no sha256 metadata".into()))?; + Ok(Some(ObjectMeta { + bytes, + sha256: parse_sha256(sha)?, + })) + } +} + +// --------------------------------------------------------------------------- +// Filesystem +// --------------------------------------------------------------------------- + +/// Chunk size for hashing an existing file. +const HASH_CHUNK: usize = 1 << 20; +/// Prefix of the temporary files a create writes next to its target. No key +/// component can start with it (a `.` in a key is always followed by two hex +/// digits). +const TEMP_PREFIX: &str = ".tmp-"; + +pub struct FsSqlObjectStore { + root: PathBuf, +} + +/// `path` made absolute against the current directory, lexically: `.` and +/// repeated separators are removed, nothing is read from the filesystem. +fn absolute_path(path: &Path) -> Result { + std::path::absolute(path) + .map_err(|_| SqlStoreError::Invalid("the SQL store root cannot be made absolute")) +} + +/// Refuse a SQL store root the node must not use: one that is equal to, +/// inside, or a parent of the KV block-store root. Both paths are made +/// absolute (relative to the current directory) and compared as written; they +/// must be free of `..` components, and nothing is read from the filesystem. +pub fn check_fs_root(root: &Path, blocks_root: &Path) -> Result<(), SqlStoreError> { + if cfg!(windows) { + return Err(SqlStoreError::Invalid( + "the filesystem SQL store is not supported on this platform", + )); + } + let root = absolute_path(root)?; + let blocks_root = absolute_path(blocks_root)?; + for path in [&root, &blocks_root] { + if path.components().any(|c| matches!(c, Component::ParentDir)) { + return Err(SqlStoreError::Invalid( + "the SQL store root and the block store root must not contain .. components", + )); + } + } + if root == blocks_root { + return Err(SqlStoreError::Invalid( + "the SQL store root must differ from the block store root", + )); + } + if root.starts_with(&blocks_root) { + return Err(SqlStoreError::Invalid( + "the SQL store root must not be inside the block store root", + )); + } + if blocks_root.starts_with(&root) { + return Err(SqlStoreError::Invalid( + "the SQL store root must not contain the block store root", + )); + } + Ok(()) +} + +fn io_error(op: &str, error: &std::io::Error) -> SqlStoreError { + // The kind only: a path would put keys in logs. + SqlStoreError::Backend(format!("filesystem {op} failed ({:?})", error.kind())) +} + +/// The file of `key` below `root`. Each component must be an ordinary file +/// name and the result must stay under `root`. +fn resolve_path(root: &Path, key: &str) -> Result { + let invalid = || SqlStoreError::Invalid("key is not a valid store path"); + if !key.starts_with(KEY_ROOT) { + return Err(invalid()); + } + let mut path = root.to_path_buf(); + for part in key.split('/') { + if part.is_empty() + || part == "." + || part == ".." + || part.starts_with(TEMP_PREFIX) + || part.contains(['\\', '\0']) + { + return Err(invalid()); + } + path.push(part); + } + if path == root || !path.starts_with(root) { + return Err(invalid()); + } + Ok(path) +} + +fn sync_dir(dir: &Path) -> Result<(), SqlStoreError> { + File::open(dir) + .and_then(|d| d.sync_all()) + .map_err(|e| io_error("directory sync", &e)) +} + +/// Create the directories from `root` down to `dir`. Each new directory and +/// its parent are fsynced; for a directory that already exists its parent is +/// fsynced too, so the whole chain is durable before the object is reported +/// created, whoever made the directories. +fn create_dirs(root: &Path, dir: &Path) -> Result<(), SqlStoreError> { + let relative = dir + .strip_prefix(root) + .map_err(|_| SqlStoreError::Invalid("key is not a valid store path"))?; + let mut current = root.to_path_buf(); + for part in relative.components() { + let parent = current.clone(); + current.push(part); + match std::fs::create_dir(¤t) { + Ok(()) => { + sync_dir(¤t)?; + sync_dir(&parent)?; + } + Err(e) if e.kind() == ErrorKind::AlreadyExists => sync_dir(&parent)?, + Err(e) => return Err(io_error("create directory", &e)), + } + } + Ok(()) +} + +/// Removes its file on drop, in every outcome of a create. +struct TempFile(PathBuf); + +impl Drop for TempFile { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.0); + } +} + +fn hash_file(file: &mut File) -> Result<[u8; 32], SqlStoreError> { + let mut hasher = Sha256::new(); + let mut chunk = vec![0u8; HASH_CHUNK]; + loop { + let n = file.read(&mut chunk).map_err(|e| io_error("read", &e))?; + if n == 0 { + return Ok(hasher.finalize().into()); + } + hasher.update(&chunk[..n]); + } +} + +fn fs_head(path: &Path) -> Result, SqlStoreError> { + let mut file = match File::open(path) { + Ok(file) => file, + Err(e) if e.kind() == ErrorKind::NotFound => return Ok(None), + Err(e) => return Err(io_error("open", &e)), + }; + let metadata = file.metadata().map_err(|e| io_error("stat", &e))?; + if !metadata.is_file() { + return Err(SqlStoreError::Backend( + "filesystem entry is not a file".into(), + )); + } + Ok(Some(ObjectMeta { + bytes: metadata.len(), + sha256: hash_file(&mut file)?, + })) +} + +fn fs_put( + path: &Path, + root: &Path, + sha: &[u8; 32], + body: &[u8], +) -> Result { + let dir = path + .parent() + .ok_or(SqlStoreError::Invalid("key is not a valid store path"))?; + create_dirs(root, dir)?; + let temp = TempFile(dir.join(format!( + "{TEMP_PREFIX}{}", + hex::encode(rand::random::<[u8; 16]>()) + ))); + { + let mut file = OpenOptions::new() + .write(true) + .create_new(true) + .open(&temp.0) + .map_err(|e| io_error("create temp file", &e))?; + file.write_all(body) + .and_then(|()| file.sync_all()) + .map_err(|e| io_error("write temp file", &e))?; + } + let outcome = match std::fs::hard_link(&temp.0, path) { + Ok(()) => Ok(PutOutcome::Created), + Err(e) if e.kind() == ErrorKind::AlreadyExists => match fs_head(path)? { + Some(meta) if meta.bytes == body.len() as u64 && &meta.sha256 == sha => { + Ok(PutOutcome::AlreadyPresent) + } + Some(_) => Err(SqlStoreError::Conflict), + None => Err(SqlStoreError::Backend( + "filesystem object is missing after the link was refused".into(), + )), + }, + Err(e) => Err(io_error("link", &e)), + }; + let synced = sync_dir(dir); + // The temporary file is removed when `temp` drops, after the sync. + match outcome { + Ok(outcome) => synced.map(|()| outcome), + Err(error) => Err(error), + } +} + +fn fs_get_range(path: &Path, range: ByteRangeSpec) -> Result, SqlStoreError> { + let mut file = match File::open(path) { + Ok(file) => file, + Err(e) if e.kind() == ErrorKind::NotFound => return Ok(None), + Err(e) => return Err(io_error("open", &e)), + }; + let metadata = file.metadata().map_err(|e| io_error("stat", &e))?; + if !metadata.is_file() { + return Err(SqlStoreError::Backend( + "filesystem entry is not a file".into(), + )); + } + let total = metadata.len(); + let Some((first, last)) = select_range(range, total) else { + return Ok(None); + }; + let len = usize::try_from(last - first + 1) + .map_err(|_| SqlStoreError::Invalid("range does not fit in memory"))?; + let mut data = vec![0u8; len]; + file.seek(SeekFrom::Start(first)) + .and_then(|_| file.read_exact(&mut data)) + .map_err(|e| io_error("read", &e))?; + Ok(Some(ObjectRange { + total_size: total, + start: first, + data, + })) +} + +async fn blocking(work: F) -> Result +where + T: Send + 'static, + F: FnOnce() -> Result + Send + 'static, +{ + tokio::task::spawn_blocking(work) + .await + .map_err(|_| SqlStoreError::Backend("filesystem task failed".into()))? +} + +impl FsSqlObjectStore { + /// Open the store at `root` (`{datadir}/sql-store`), creating it if + /// needed, after [`check_fs_root`] against the KV block-store root. + pub fn open(root: PathBuf, blocks_root: &Path) -> Result { + check_fs_root(&root, blocks_root)?; + let root = absolute_path(&root)?; + std::fs::create_dir_all(&root).map_err(|e| io_error("create root", &e))?; + sync_dir(&root)?; + if let Some(parent) = root.parent() { + sync_dir(parent)?; + } + Ok(Self { root }) + } + + pub fn root(&self) -> &Path { + &self.root + } + + /// The file an object is stored in. + pub fn path_for(&self, key: &ObjectKey) -> Result { + resolve_path(&self.root, key.as_str()) + } +} + +#[async_trait] +impl SqlObjectStore for FsSqlObjectStore { + async fn put_if_absent( + &self, + key: &ObjectKey, + body: Vec, + ) -> Result { + if &sha256(&body) != key.sha256() { + return Err(SqlStoreError::BodyMismatch); + } + let path = self.path_for(key)?; + let root = self.root.clone(); + let sha = *key.sha256(); + blocking(move || fs_put(&path, &root, &sha, &body)).await + } + + async fn get_range( + &self, + key: &ObjectKey, + range: ByteRangeSpec, + ) -> Result, SqlStoreError> { + let path = self.path_for(key)?; + blocking(move || fs_get_range(&path, range)).await + } + + async fn head(&self, key: &ObjectKey) -> Result, SqlStoreError> { + let path = self.path_for(key)?; + blocking(move || fs_head(&path)).await + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + + #[cfg(not(windows))] + #[test] + fn fs_root_must_be_outside_the_block_store() { + let blocks = Path::new("/data/blocks"); + for ok in [ + "/data/sql-store", + "/data/blocks2", + "/sql-store", + "/data/sql/store", + ] { + assert!(check_fs_root(Path::new(ok), blocks).is_ok(), "{ok}"); + } + for bad in [ + "/data/blocks", + "/data/blocks/", + "/data/blocks/sql-store", + "/data", + "/", + "/data/../data/sql-store", + ] { + assert!(check_fs_root(Path::new(bad), blocks).is_err(), "{bad}"); + } + assert!(check_fs_root( + Path::new("/data/sql-store"), + Path::new("/data/../data/blocks") + ) + .is_err()); + } + + #[cfg(not(windows))] + #[test] + fn relative_roots_are_made_absolute_before_the_check() { + // The default layout: `./data` for the data directory and the blocks. + let blocks = Path::new("./data/blocks"); + for ok in [ + "./data/sql-store", + "data/sql-store", + "sql-store", + "./data/blocks2", + ] { + assert!(check_fs_root(Path::new(ok), blocks).is_ok(), "{ok}"); + } + for bad in [ + "data/blocks", + "./data/blocks", + "./data/blocks/sql-store", + "data/blocks/sql-store", + "./data", + "data", + "", + ] { + assert!(check_fs_root(Path::new(bad), blocks).is_err(), "{bad:?}"); + } + // A relative root against an absolute block store root, and the reverse. + let cwd = std::env::current_dir().unwrap(); + assert!(check_fs_root(Path::new("sql-store"), &cwd.join("blocks")).is_ok()); + assert!(check_fs_root(Path::new("blocks/sql-store"), &cwd.join("blocks")).is_err()); + assert!(check_fs_root(&cwd.join("blocks/sql-store"), Path::new("blocks")).is_err()); + } + + #[test] + fn resolved_paths_stay_under_the_root() { + let root = Path::new("/var/tmp/sql-store"); + let ok = resolve_path(root, "sql/v1/a/b/c/snap/x").unwrap(); + assert_eq!(ok, root.join("sql/v1/a/b/c/snap/x")); + for bad in [ + "", + "sql/v1/", + "sql/v1/../../etc/passwd", + "sql/v1/a/./b", + "sql/v1//a", + "sql/v1/a/", + "/etc/passwd", + "../sql/v1/a", + "sql/v1/a\\b", + "sql/v1/a\0b", + "sql/v1/a/.tmp-0011", + "other/v1/a", + ] { + assert!(resolve_path(root, bad).is_err(), "{bad:?}"); + } + } + + #[test] + fn range_selection_matches_byte_range_spec() { + let edges = [0u64, 1, 2, 4, 5, 6, 9, 10, 11]; + let mut specs = Vec::new(); + for &a in &edges { + specs.push(ByteRangeSpec::From { start: a }); + specs.push(ByteRangeSpec::Suffix { length: a }); + for &b in &edges { + specs.push(ByteRangeSpec::Inclusive { start: a, end: b }); + } + } + for total in [0u64, 1, 2, 5, 10] { + for spec in &specs { + assert_eq!( + select_range(*spec, total), + spec.resolve(total).map(|r| (r.start(), r.end())), + "{spec:?} of {total}" + ); + } + } + } + + #[test] + fn content_ranges_are_parsed_strictly() { + assert_eq!( + parse_content_range("bytes 10-19/300").unwrap(), + (10, 19, 300) + ); + assert_eq!(parse_content_range("bytes 0-0/1").unwrap(), (0, 0, 1)); + for bad in [ + "", + "bytes */300", + "bytes 10-19/*", + "bytes 19-10/300", + "bytes 0-300/300", + "10-19/300", + "bytes 10/300", + ] { + assert!(parse_content_range(bad).is_err(), "{bad:?}"); + } + } + + fn scripted(answers: Vec) -> (Result<(WriteMode, String), SqlStoreError>, usize) { + let calls = AtomicUsize::new(0); + let answers = Mutex::new(answers.into_iter()); + let result = futures::executor::block_on(probe_with(|| { + calls.fetch_add(1, Ordering::SeqCst); + let next = answers + .lock() + .unwrap() + .next() + .expect("probe asked too often"); + async move { next } + })); + (result, calls.load(Ordering::SeqCst)) + } + + fn answer(status: u16, code: &str) -> Answer { + Answer { + status: Some(status), + code: Some(code.to_owned()), + } + } + + #[test] + fn probe_selects_the_write_mode_from_the_second_answer() { + use PutResult::*; + let refused = || PreconditionFailed(answer(412, "PreconditionFailed")); + let unsupported = || Unsupported(answer(501, "NotImplemented")); + let (mode, calls) = scripted(vec![Created, refused()]); + let (mode, decided_by) = mode.unwrap(); + assert_eq!((mode, calls), (WriteMode::Conditional, 2)); + assert_eq!(decided_by, "status 412, code PreconditionFailed"); + // An object left by an earlier start answers the first request. + let (mode, calls) = scripted(vec![refused()]); + assert_eq!((mode.unwrap().0, calls), (WriteMode::Conditional, 1)); + let (mode, calls) = scripted(vec![Created, Created]); + let (mode, decided_by) = mode.unwrap(); + assert_eq!((mode, calls), (WriteMode::CheckThenWrite, 2)); + assert_eq!(decided_by, "created"); + let (mode, calls) = scripted(vec![unsupported()]); + let (mode, decided_by) = mode.unwrap(); + assert_eq!((mode, calls), (WriteMode::CheckThenWrite, 1)); + assert_eq!(decided_by, "status 501, code NotImplemented"); + let (mode, calls) = scripted(vec![Created, unsupported()]); + assert_eq!((mode.unwrap().0, calls), (WriteMode::CheckThenWrite, 2)); + } + + #[test] + fn probe_failure_is_an_error_not_a_fallback() { + use PutResult::*; + let failure = || Failed(SqlStoreError::Backend("down".into())); + let (mode, calls) = scripted(vec![failure()]); + assert!(matches!(mode, Err(SqlStoreError::Backend(_)))); + assert_eq!(calls, 1); + let (mode, _) = scripted(vec![Created, failure()]); + assert!(matches!(mode, Err(SqlStoreError::Backend(_)))); + } + + #[test] + fn kv_keys_start_with_the_scan_prefix() { + use tinycloud_auth::resource::{SpaceId, SPACE_ID_PREFIX}; + use tinycloud_core::hash::hash; + + use super::super::s3::{S3BlockStore, KV_SCAN_PREFIX}; + + assert_eq!(KV_SCAN_PREFIX, SPACE_ID_PREFIX); + let space: SpaceId = + "tinycloud:pkh:eip155:1:0xf39Fd6e51aad88F6F4ce6aB8827279cffFb92266:default" + .parse() + .unwrap(); + assert!(space.to_string().starts_with(SPACE_ID_PREFIX)); + let digest = hash(b"block"); + let key = S3BlockStore::key_for(&space, &digest, 0); + assert!(key.starts_with(KV_SCAN_PREFIX), "{key}"); + assert!(key.starts_with(&format!("{space}/"))); + assert!(!key.contains('@'), "{key}"); + let next_key = S3BlockStore::key_for(&space, &digest, 7); + assert!(next_key.starts_with(KV_SCAN_PREFIX), "{next_key}"); + assert_eq!(next_key, format!("{key}@7")); + // SQL objects live outside the scanned prefix. + assert!(!KEY_ROOT.starts_with(KV_SCAN_PREFIX)); + } + + #[tokio::test] + async fn cancelled_creates_leave_no_gate_behind() { + // A server that accepts connections and never answers. + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let accepter = tokio::spawn(async move { + let mut held = Vec::new(); + while let Ok((stream, _)) = listener.accept().await { + held.push(stream); + } + }); + let client = Client::from_conf( + aws_sdk_s3::Config::builder() + .behavior_version(super::super::s3::behavior_version()) + .region(aws_sdk_s3::config::Region::new("us-east-1")) + .credentials_provider(aws_sdk_s3::config::Credentials::new( + "id", "secret", None, None, "test", + )) + .endpoint_url(format!("http://{addr}")) + .force_path_style(true) + .http_client(super::super::s3::http_client()) + .build(), + ); + let body = b"abc".to_vec(); + let key = ObjectKey::snapshot( + &tinycloud_core::sql::store::DbRef { + space: "tinycloud:s", + db_id: "d", + }, + &tinycloud_core::sql::store::Lineage::random(), + 1, + &sha256(&body), + ) + .unwrap(); + for mode in [WriteMode::Conditional, WriteMode::CheckThenWrite] { + let store = S3SqlObjectStore::new(client.clone(), "bucket".into(), mode); + tokio::time::timeout(Duration::from_secs(30), async { + // One creator holds the gate in its request, one waits for it; + // both are dropped by their timeouts. + let first = tokio::time::timeout( + Duration::from_millis(300), + store.put_if_absent(&key, body.clone()), + ); + let second = tokio::time::timeout( + Duration::from_millis(300), + store.put_if_absent(&key, body.clone()), + ); + let (first, second) = tokio::join!(first, second); + assert!( + first.is_err() && second.is_err(), + "the server never answers" + ); + }) + .await + .expect("the cancelled creates did not return"); + assert!(lock_gates(&store.gates).is_empty(), "{mode:?}"); + } + accepter.abort(); + } +} diff --git a/tinycloud-node-server/tests/sql_object_store.rs b/tinycloud-node-server/tests/sql_object_store.rs new file mode 100644 index 00000000..6cb40a8e --- /dev/null +++ b/tinycloud-node-server/tests/sql_object_store.rs @@ -0,0 +1,1200 @@ +//! TC-1491 slice 1b: the SQL object stores (`storage::sql_object`). +//! +//! One conformance set runs against the in-memory store, the filesystem store +//! and the S3 store. The S3 store runs against an in-process S3 API (no +//! network, every test of this file) and, in the `#[ignore]` tests, against +//! the at-rest harness MinIO with the same `TC887_S3_*` environment as +//! `tests/s3_block_store.rs`: +//! +//! ```sh +//! (set -a; . "$RUN_DIR/harness.env"; set +a; +//! cargo test -p tinycloud-node --test sql_object_store -- --ignored) +//! ``` +//! +//! Every call to the real S3 API is wrapped in a 30 s timeout so a regression +//! fails instead of hanging. + +use std::{collections::BTreeMap, future::Future, path::Path, sync::Arc, time::Duration}; + +use anyhow::{Context, Result}; +use async_trait::async_trait; +use axum::{ + body::{Body, Bytes}, + extract::State, + http::{HeaderMap, Method, StatusCode, Uri}, + response::Response, + Router, +}; +use parking_lot::Mutex; +use percent_encoding::percent_decode_str; +use tinycloud::storage::{ + s3::{S3BlockConfig, S3BlockStore}, + sql_object::{FsSqlObjectStore, S3SqlObjectStore, WriteMode}, +}; +use tinycloud_auth::resource::SpaceId; +use tinycloud_core::{ + sql::store::{ + object::sha256, DbRef, Lineage, MemorySqlObjectStore, ObjectKey, ObjectMeta, PutOutcome, + SqlObjectStore, SqlStoreError, + }, + storage::{ByteRangeSpec, StorageConfig, StoreSize}, +}; + +// --------------------------------------------------------------------------- +// Keys and bodies +// --------------------------------------------------------------------------- + +fn body(len: usize) -> Vec { + let salt = rand::random::() as usize; + (0..len).map(|i| ((i * 13 + salt) % 251) as u8).collect() +} + +/// Same length as `bytes`, different content. +fn differing(bytes: &[u8]) -> Vec { + let mut other = bytes.to_vec(); + other[0] ^= 1; + other +} + +/// A key of a fresh, random database and lineage: unique per call and per run. +fn key_for(bytes: &[u8]) -> ObjectKey { + key_in(&format!("tinycloud:sqltest:{}", tag()), "v2n", bytes) +} + +fn key_in(space: &str, db_id: &str, bytes: &[u8]) -> ObjectKey { + ObjectKey::snapshot( + &DbRef { space, db_id }, + &Lineage::random(), + 1, + &sha256(bytes), + ) + .unwrap() +} + +fn tag() -> String { + hex::encode(rand::random::<[u8; 6]>()) +} + +async fn whole(store: &S, key: &ObjectKey) -> Option> { + store + .get_range(key, ByteRangeSpec::From { start: 0 }) + .await + .unwrap() + .map(|r| r.data) +} + +async fn part( + store: &S, + key: &ObjectKey, + range: ByteRangeSpec, +) -> Option<(u64, u64, Vec)> { + store + .get_range(key, range) + .await + .unwrap() + .map(|r| (r.start, r.total_size, r.data)) +} + +// --------------------------------------------------------------------------- +// The conformance set +// --------------------------------------------------------------------------- + +/// Writes bytes under a key behind the store's back. +#[async_trait] +trait Planter: Send + Sync { + async fn plant(&self, key: &ObjectKey, bytes: Vec); +} + +async fn conformance(store: Arc, planter: Option<&dyn Planter>) { + let s = &*store; + + // Create, then the same bytes again: refused as present, bytes unchanged. + let b = body(300); + let k = key_for(&b); + assert_eq!(s.head(&k).await.unwrap(), None); + assert!(s + .get_range(&k, ByteRangeSpec::From { start: 0 }) + .await + .unwrap() + .is_none()); + assert_eq!( + s.put_if_absent(&k, b.clone()).await.unwrap(), + PutOutcome::Created + ); + assert_eq!( + s.put_if_absent(&k, b.clone()).await.unwrap(), + PutOutcome::AlreadyPresent + ); + assert_eq!(whole(s, &k).await.unwrap(), b); + assert_eq!( + s.head(&k).await.unwrap(), + Some(ObjectMeta { + bytes: 300, + sha256: sha256(&b) + }) + ); + + // A body that does not hash to the key is refused and stores nothing. + let mut wrong = b.clone(); + wrong[0] ^= 1; + assert!(matches!( + s.put_if_absent(&k, wrong.clone()).await, + Err(SqlStoreError::BodyMismatch) + )); + assert_eq!(whole(s, &k).await.unwrap(), b); + let k_unmade = key_for(&wrong); + assert!(matches!( + s.put_if_absent(&k_unmade, b.clone()).await, + Err(SqlStoreError::BodyMismatch) + )); + assert_eq!(s.head(&k_unmade).await.unwrap(), None); + + // Ranges, pinned to what the in-memory store answers. + use ByteRangeSpec::*; + let at = |start: usize, end: usize| (start as u64, 300, b[start..=end].to_vec()); + assert_eq!( + part(s, &k, Inclusive { start: 10, end: 19 }).await, + Some(at(10, 19)) + ); + assert_eq!( + part(s, &k, Inclusive { start: 0, end: 299 }).await, + Some(at(0, 299)) + ); + assert_eq!( + part( + s, + &k, + Inclusive { + start: 0, + end: 10_000 + } + ) + .await, + Some(at(0, 299)) + ); + assert_eq!( + part( + s, + &k, + Inclusive { + start: 299, + end: 299 + } + ) + .await, + Some(at(299, 299)) + ); + assert_eq!( + part( + s, + &k, + Inclusive { + start: 290, + end: 10_000 + } + ) + .await, + Some(at(290, 299)) + ); + assert_eq!(part(s, &k, From { start: 299 }).await, Some(at(299, 299))); + assert_eq!(part(s, &k, From { start: 0 }).await, Some(at(0, 299))); + assert_eq!(part(s, &k, Suffix { length: 1 }).await, Some(at(299, 299))); + assert_eq!(part(s, &k, Suffix { length: 5 }).await, Some(at(295, 299))); + assert_eq!(part(s, &k, Suffix { length: 300 }).await, Some(at(0, 299))); + assert_eq!( + part(s, &k, Suffix { length: 10_000 }).await, + Some(at(0, 299)) + ); + // Nothing is selected: start beyond the end, an inverted range, a + // zero-length suffix. + assert_eq!(part(s, &k, From { start: 300 }).await, None); + assert_eq!( + part( + s, + &k, + Inclusive { + start: 300, + end: 310 + } + ) + .await, + None + ); + assert_eq!(part(s, &k, Inclusive { start: 5, end: 4 }).await, None); + assert_eq!(part(s, &k, Suffix { length: 0 }).await, None); + + // The empty object: it exists, and no range of it selects anything. + let empty = key_for(b""); + assert_eq!( + s.put_if_absent(&empty, Vec::new()).await.unwrap(), + PutOutcome::Created + ); + assert_eq!( + s.put_if_absent(&empty, Vec::new()).await.unwrap(), + PutOutcome::AlreadyPresent + ); + assert_eq!( + s.head(&empty).await.unwrap(), + Some(ObjectMeta { + bytes: 0, + sha256: sha256(b"") + }) + ); + assert_eq!(part(s, &empty, From { start: 0 }).await, None); + assert_eq!(part(s, &empty, Suffix { length: 1 }).await, None); + assert_eq!(part(s, &empty, Inclusive { start: 0, end: 0 }).await, None); + + // Stored bytes that differ from the key's content are never replaced. + if let Some(planter) = planter { + for (len, planted_len) in [(64usize, 64usize), (64, 7)] { + let wanted = body(len); + let other = if planted_len == len { + differing(&wanted) + } else { + body(planted_len) + }; + assert_ne!(wanted, other); + let key = key_for(&wanted); + planter.plant(&key, other.clone()).await; + for _ in 0..2 { + assert!(matches!( + s.put_if_absent(&key, wanted.clone()).await, + Err(SqlStoreError::Conflict) + )); + assert_eq!(whole(s, &key).await.unwrap(), other); + } + } + } + + concurrent_creators(store, 40).await; +} + +/// 8 creators of one key, `rounds` times over fresh keys: exactly one wins. +async fn concurrent_creators(store: Arc, rounds: usize) { + for _ in 0..rounds { + let b = body(1024); + let k = key_for(&b); + let tasks: Vec<_> = (0..8) + .map(|_| { + let (store, k, b) = (store.clone(), k.clone(), b.clone()); + tokio::spawn(async move { store.put_if_absent(&k, b).await }) + }) + .collect(); + let mut created = 0; + let mut present = 0; + for task in tasks { + match task.await.unwrap().unwrap() { + PutOutcome::Created => created += 1, + PutOutcome::AlreadyPresent => present += 1, + } + } + assert_eq!((created, present), (1, 7)); + assert_eq!(whole(&*store, &k).await.unwrap(), b); + } +} + +// --------------------------------------------------------------------------- +// In-memory +// --------------------------------------------------------------------------- + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn memory_conformance() { + conformance(Arc::new(MemorySqlObjectStore::default()), None).await; +} + +// --------------------------------------------------------------------------- +// Filesystem +// --------------------------------------------------------------------------- + +/// `{tempdir}/sql-store` next to `{tempdir}/blocks`, on a real disk. +fn fs_dirs() -> (tempfile::TempDir, std::path::PathBuf, std::path::PathBuf) { + let dir = tempfile::Builder::new() + .prefix("sqlobj-") + .tempdir_in("/var/tmp") + .unwrap(); + let root = dir.path().join("sql-store"); + let blocks = dir.path().join("blocks"); + (dir, root, blocks) +} + +fn fs_store() -> (tempfile::TempDir, Arc) { + let (dir, root, blocks) = fs_dirs(); + let store = Arc::new(FsSqlObjectStore::open(root, &blocks).unwrap()); + (dir, store) +} + +struct FsPlanter(Arc); + +#[async_trait] +impl Planter for FsPlanter { + async fn plant(&self, key: &ObjectKey, bytes: Vec) { + let path = self.0.path_for(key).unwrap(); + std::fs::create_dir_all(path.parent().unwrap()).unwrap(); + std::fs::write(path, bytes).unwrap(); + } +} + +fn names_in(dir: &Path) -> Vec { + let mut names: Vec = std::fs::read_dir(dir) + .unwrap() + .map(|e| e.unwrap().file_name().into_string().unwrap()) + .collect(); + names.sort(); + names +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn fs_conformance() { + let (_dir, store) = fs_store(); + let planter = FsPlanter(store.clone()); + conformance(store, Some(&planter)).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn fs_leftover_temp_files_never_block_a_create() { + let (_dir, store) = fs_store(); + let b = body(2048); + let k = key_for(&b); + let path = store.path_for(&k).unwrap(); + let parent = path.parent().unwrap(); + // A crash between the temp write and the link leaves the full body (or a + // part of it) under a temp name. + std::fs::create_dir_all(parent).unwrap(); + std::fs::write(parent.join(".tmp-0123456789abcdef0123456789abcdef"), &b).unwrap(); + std::fs::write( + parent.join(".tmp-ffffffffffffffffffffffffffffffff"), + &b[..100], + ) + .unwrap(); + let leftovers = names_in(parent); + assert_eq!(leftovers.len(), 2); + + assert_eq!( + store.put_if_absent(&k, b.clone()).await.unwrap(), + PutOutcome::Created + ); + assert_eq!(std::fs::read(&path).unwrap(), b); + // The stray files are untouched; the create left none of its own. + let mut expected = leftovers.clone(); + expected.push(path.file_name().unwrap().to_str().unwrap().to_owned()); + expected.sort(); + assert_eq!(names_in(parent), expected); + + // Every outcome removes its temp file. + assert_eq!( + store.put_if_absent(&k, b.clone()).await.unwrap(), + PutOutcome::AlreadyPresent + ); + assert_eq!(names_in(parent), expected); + let planted = differing(&b); + let conflicted = key_for(&b); + let conflicted_path = store.path_for(&conflicted).unwrap(); + std::fs::create_dir_all(conflicted_path.parent().unwrap()).unwrap(); + std::fs::write(&conflicted_path, &planted).unwrap(); + assert!(matches!( + store.put_if_absent(&conflicted, b.clone()).await, + Err(SqlStoreError::Conflict) + )); + assert_eq!(names_in(conflicted_path.parent().unwrap()).len(), 1); + assert_eq!(std::fs::read(&conflicted_path).unwrap(), planted); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn fs_keys_never_resolve_outside_the_root() { + let (dir, store) = fs_store(); + for (space, db) in [ + ("tinycloud:../../escape", "v2n"), + ("tinycloud:x", ".."), + ("tinycloud:x", "a/b"), + ("/etc", "/passwd"), + ("tinycloud:x", "a\\b"), + ("..", "."), + ] { + let b = body(32); + let k = key_in(space, db, &b); + let path = store.path_for(&k).unwrap(); + assert!(path.starts_with(store.root()), "{path:?}"); + assert_eq!( + store.put_if_absent(&k, b.clone()).await.unwrap(), + PutOutcome::Created + ); + assert_eq!(std::fs::read(&path).unwrap(), b); + } + // Nothing was created beside the root. + assert_eq!(names_in(dir.path()), vec!["sql-store".to_owned()]); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn fs_large_objects_are_hashed_in_chunks_and_ranged() { + let (_dir, store) = fs_store(); + let b: Vec = (0..(3 << 20) + 17).map(|i| (i % 253) as u8).collect(); + let k = key_for(&b); + assert_eq!( + store.put_if_absent(&k, b.clone()).await.unwrap(), + PutOutcome::Created + ); + assert_eq!( + store.put_if_absent(&k, b.clone()).await.unwrap(), + PutOutcome::AlreadyPresent + ); + assert_eq!( + store.head(&k).await.unwrap(), + Some(ObjectMeta { + bytes: b.len() as u64, + sha256: sha256(&b) + }) + ); + let start = (1 << 20) - 5; + let got = part( + &*store, + &k, + ByteRangeSpec::Inclusive { + start: start as u64, + end: (start + 10) as u64, + }, + ) + .await + .unwrap(); + assert_eq!( + got, + (start as u64, b.len() as u64, b[start..=start + 10].to_vec()) + ); +} + +#[test] +fn fs_open_refuses_a_root_that_overlaps_the_block_store() { + let (dir, _root, blocks) = fs_dirs(); + for root in [ + blocks.clone(), + blocks.join("sql-store"), + dir.path().to_path_buf(), + ] { + assert!(matches!( + FsSqlObjectStore::open(root.clone(), &blocks), + Err(SqlStoreError::Invalid(_)) + )); + } + assert!(!blocks.join("sql-store").exists()); + // Relative paths are made absolute first: a relative root inside a + // relative block store root is refused, and nothing is created. + assert!(matches!( + FsSqlObjectStore::open( + "tc1491-relative/blocks/sql-store".into(), + Path::new("./tc1491-relative/blocks") + ), + Err(SqlStoreError::Invalid(_)) + )); + assert!(!Path::new("tc1491-relative").exists()); +} + +// --------------------------------------------------------------------------- +// S3: an in-process S3 API +// --------------------------------------------------------------------------- + +const MOCK_BUCKET: &str = "tc1491"; + +/// How the in-process API answers a PUT with `If-None-Match: *`. +#[derive(Clone, Copy)] +enum Conditional { + /// Refuses to replace an existing object (412), as S3 does. + Honour, + /// Replaces the object, as a provider ignoring the header does. + Ignore, + /// Answers with this status and error code. + Reject(u16, &'static str), +} + +/// key -> (body, x-amz-meta-sha256) +type Objects = BTreeMap, Option)>; + +#[derive(Clone)] +struct MockS3 { + objects: Arc>, + conditional: Arc>, + requests: Arc>>, +} + +fn reply(status: u16, headers: Vec<(&str, String)>, body: Vec) -> Response { + let mut builder = Response::builder().status(StatusCode::from_u16(status).unwrap()); + for (name, value) in headers { + builder = builder.header(name, value); + } + builder.body(Body::from(body)).unwrap() +} + +fn error_reply(status: u16, code: &str) -> Response { + reply( + status, + vec![("content-type", "application/xml".into())], + format!("{code}{code}").into_bytes(), + ) +} + +fn parse_range(value: &str) -> Option { + let (first, last) = value.strip_prefix("bytes=")?.split_once('-')?; + Some(match (first, last) { + ("", n) => ByteRangeSpec::Suffix { + length: n.parse().ok()?, + }, + (a, "") => ByteRangeSpec::From { + start: a.parse().ok()?, + }, + (a, b) => ByteRangeSpec::Inclusive { + start: a.parse().ok()?, + end: b.parse().ok()?, + }, + }) +} + +async fn mock_api( + State(s3): State, + method: Method, + uri: Uri, + headers: HeaderMap, + body: Bytes, +) -> Response { + let path = percent_decode_str(uri.path()) + .decode_utf8_lossy() + .into_owned(); + let rest = path.trim_start_matches('/'); + let (bucket, key) = rest.split_once('/').unwrap_or((rest, "")); + if bucket != MOCK_BUCKET { + return error_reply(404, "NoSuchBucket"); + } + let header = |name: &str| { + headers + .get(name) + .and_then(|v| v.to_str().ok()) + .map(str::to_owned) + }; + let mut objects = s3.objects.lock(); + match (&method, key.is_empty()) { + (&Method::GET, true) => { + let prefix = uri + .query() + .unwrap_or("") + .split('&') + .find_map(|pair| pair.strip_prefix("prefix=")) + .map(|p| percent_decode_str(p).decode_utf8_lossy().into_owned()) + .unwrap_or_default(); + s3.requests.lock().push(format!("LIST {prefix}")); + let contents: String = objects + .iter() + .filter(|(key, _)| key.starts_with(&prefix)) + .map(|(key, (data, _))| { + format!( + "{key}{}", + data.len() + ) + }) + .collect(); + reply( + 200, + vec![("content-type", "application/xml".into())], + format!( + "\ + {MOCK_BUCKET}false{contents}\ + " + ) + .into_bytes(), + ) + } + (&Method::PUT, false) => { + let conditional = header("if-none-match").as_deref() == Some("*"); + s3.requests + .lock() + .push(format!("PUT {key} conditional={conditional}")); + if conditional { + match *s3.conditional.lock() { + Conditional::Reject(status, code) => return error_reply(status, code), + Conditional::Honour if objects.contains_key(key) => { + return error_reply(412, "PreconditionFailed") + } + _ => {} + } + } + objects.insert(key.to_owned(), (body.to_vec(), header("x-amz-meta-sha256"))); + reply(200, vec![("etag", "\"mock\"".into())], Vec::new()) + } + (&Method::HEAD, false) => { + s3.requests.lock().push(format!("HEAD {key}")); + match objects.get(key) { + Some((data, sha)) => { + let mut headers = vec![("content-length", data.len().to_string())]; + if let Some(sha) = sha { + headers.push(("x-amz-meta-sha256", sha.clone())); + } + reply(200, headers, Vec::new()) + } + None => reply(404, Vec::new(), Vec::new()), + } + } + (&Method::GET, false) => { + s3.requests.lock().push(format!("GET {key}")); + let Some((data, _)) = objects.get(key) else { + return error_reply(404, "NoSuchKey"); + }; + let total = data.len() as u64; + let Some(spec) = header("range").as_deref().and_then(parse_range) else { + return reply( + 200, + vec![("content-length", total.to_string())], + data.clone(), + ); + }; + match spec.resolve(total) { + None => error_reply(416, "InvalidRange"), + Some(r) => { + let slice = data[r.start() as usize..=r.end() as usize].to_vec(); + reply( + 206, + vec![ + ("content-length", slice.len().to_string()), + ( + "content-range", + format!("bytes {}-{}/{total}", r.start(), r.end()), + ), + ], + slice, + ) + } + } + } + _ => error_reply(501, "NotImplemented"), + } +} + +async fn start_mock(conditional: Conditional) -> Result<(MockS3, S3BlockConfig)> { + // The S3 client reads credentials from the AWS_* environment. + std::env::set_var("AWS_ACCESS_KEY_ID", "tc1491"); + std::env::set_var("AWS_SECRET_ACCESS_KEY", "tc1491"); + std::env::set_var("AWS_REGION", "us-east-1"); + std::env::set_var("AWS_EC2_METADATA_DISABLED", "true"); + let s3 = MockS3 { + objects: Arc::default(), + conditional: Arc::new(Mutex::new(conditional)), + requests: Arc::default(), + }; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + let endpoint = format!("http://{}", listener.local_addr()?); + let app = Router::new().fallback(mock_api).with_state(s3.clone()); + tokio::spawn(async move { + let _ = axum::serve(listener, app).await; + }); + let config = serde_json::from_value(serde_json::json!({ + "bucket": MOCK_BUCKET, + "endpoint": endpoint, + }))?; + Ok((s3, config)) +} + +// --------------------------------------------------------------------------- +// S3: tests shared by the in-process API and the harness MinIO +// --------------------------------------------------------------------------- + +async fn limited(what: &str, work: impl Future) -> T { + tokio::time::timeout(Duration::from_secs(30), work) + .await + .unwrap_or_else(|_| panic!("{what} timed out after 30 s")) +} + +/// Every call bounded by 30 s. +struct Timed(S); + +#[async_trait] +impl SqlObjectStore for Timed { + async fn put_if_absent( + &self, + key: &ObjectKey, + body: Vec, + ) -> Result { + limited("put_if_absent", self.0.put_if_absent(key, body)).await + } + async fn get_range( + &self, + key: &ObjectKey, + range: ByteRangeSpec, + ) -> Result, SqlStoreError> { + limited("get_range", self.0.get_range(key, range)).await + } + async fn head(&self, key: &ObjectKey) -> Result, SqlStoreError> { + limited("head", self.0.head(key)).await + } +} + +/// Puts an object straight to S3, as an object written by something else. +struct S3Planter { + store: Arc, + with_metadata: bool, +} + +impl S3Planter { + async fn put_raw(&self, key: &ObjectKey, bytes: Vec) { + let request = self + .store + .client() + .put_object() + .bucket(self.store.bucket()) + .key(key.as_str()); + let request = if self.with_metadata { + request.metadata("sha256", hex::encode(sha256(&bytes))) + } else { + request + }; + limited( + "planting an object", + request + .body(aws_sdk_s3::primitives::ByteStream::from(bytes)) + .send(), + ) + .await + .unwrap(); + } +} + +#[async_trait] +impl Planter for S3Planter { + async fn plant(&self, key: &ObjectKey, bytes: Vec) { + self.put_raw(key, bytes).await + } +} + +async fn s3_conformance(config: &S3BlockConfig, mode: WriteMode) { + let store = Arc::new(S3SqlObjectStore::open_with_mode(config, mode).await); + let planter = S3Planter { + store: store.clone(), + with_metadata: true, + }; + struct Shared(Arc); + #[async_trait] + impl SqlObjectStore for Shared { + async fn put_if_absent( + &self, + key: &ObjectKey, + body: Vec, + ) -> Result { + self.0.put_if_absent(key, body).await + } + async fn get_range( + &self, + key: &ObjectKey, + range: ByteRangeSpec, + ) -> Result, SqlStoreError> { + self.0.get_range(key, range).await + } + async fn head(&self, key: &ObjectKey) -> Result, SqlStoreError> { + self.0.head(key).await + } + } + conformance(Arc::new(Timed(Shared(store))), Some(&planter)).await; +} + +/// An object without the `sha256` metadata is never taken for a match. +async fn s3_object_without_hash_metadata_is_an_error(config: &S3BlockConfig, mode: WriteMode) { + let store = Arc::new(Timed(S3SqlObjectStore::open_with_mode(config, mode).await)); + let raw = Arc::new(S3SqlObjectStore::open_with_mode(config, mode).await); + let planter = S3Planter { + store: raw, + with_metadata: false, + }; + let b = body(64); + let k = key_for(&b); + planter.plant(&k, b.clone()).await; + assert!(matches!( + store.head(&k).await, + Err(SqlStoreError::Backend(_)) + )); + assert!(matches!( + store.put_if_absent(&k, b.clone()).await, + Err(SqlStoreError::Backend(_)) + )); + assert_eq!(whole(&*store, &k).await.unwrap(), b); +} + +/// The KV boot scan lists `tinycloud:`-prefixed keys only: the sizes it +/// computes are the same with and without `sql/v1/` objects in the bucket. +async fn kv_scan_ignores_sql_objects(config: &S3BlockConfig) { + let store = Arc::new(S3SqlObjectStore::open_with_mode(config, WriteMode::Conditional).await); + let t = tag(); + let space = format!("tinycloud:key:scan:{t}"); + let sql_only = format!("tinycloud:key:sqlonly:{t}"); + let space_id: SpaceId = space.parse().unwrap(); + let sql_only_id: SpaceId = sql_only.parse().unwrap(); + let put_raw = |key: String, len: usize| { + let store = store.clone(); + async move { + limited( + "raw put", + store + .client() + .put_object() + .bucket(store.bucket()) + .key(key) + .body(aws_sdk_s3::primitives::ByteStream::from(vec![7u8; len])) + .send(), + ) + .await + .unwrap(); + } + }; + let open = || async { + limited("block store open", async { + StorageConfig::::open(config).await.unwrap() + }) + .await + }; + let sizes = |store: S3BlockStore| { + let (a, b) = (space_id.clone(), sql_only_id.clone()); + async move { + ( + store.total_size(&a).await.unwrap(), + store.total_size(&b).await.unwrap(), + ) + } + }; + + // KV-style objects: `{space}/{hash}`. + put_raw(format!("{space}/aaaa"), 5).await; + put_raw(format!("{space}/bbbb"), 7).await; + let before = sizes(open().await).await; + assert_eq!(before, (Some(12), None)); + + // SQL objects, through the store and as raw keys shaped like the + // namespace: none of them belongs to a space. + for len in [10, 4096, 1 << 20] { + let b = body(len); + let k = key_in(&space, "v2n", &b); + assert!(k.as_str().starts_with("sql/v1/")); + limited("sql put", store.put_if_absent(&k, b)) + .await + .unwrap(); + } + put_raw(format!("sql/v1/{space}/db/lineage/snap/x"), 1000).await; + put_raw(format!("sql/v1/{sql_only}/db/x"), 2000).await; + put_raw(format!("sql/v1/{sql_only}/x"), 3000).await; + let after = sizes(open().await).await; + assert_eq!(after, before); +} + +// --------------------------------------------------------------------------- +// S3: in-process API +// --------------------------------------------------------------------------- + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn mock_s3_conditional_conformance() -> Result<()> { + let (_s3, config) = start_mock(Conditional::Honour).await?; + s3_conformance(&config, WriteMode::Conditional).await; + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn mock_s3_check_then_write_conformance_with_a_provider_that_ignores_the_header() -> Result<()> +{ + let (_s3, config) = start_mock(Conditional::Ignore).await?; + s3_conformance(&config, WriteMode::CheckThenWrite).await; + Ok(()) +} + +#[tokio::test] +async fn mock_s3_object_without_hash_metadata_is_an_error() -> Result<()> { + for mode in [WriteMode::Conditional, WriteMode::CheckThenWrite] { + let (_s3, config) = start_mock(Conditional::Honour).await?; + s3_object_without_hash_metadata_is_an_error(&config, mode).await; + } + Ok(()) +} + +#[tokio::test] +async fn mock_s3_kv_scan_ignores_sql_objects() -> Result<()> { + let (s3, config) = start_mock(Conditional::Honour).await?; + kv_scan_ignores_sql_objects(&config).await; + // The scan asks for the KV key prefix only. + let lists: Vec = s3 + .requests + .lock() + .iter() + .filter(|r| r.starts_with("LIST")) + .cloned() + .collect(); + assert!( + !lists.is_empty() && lists.iter().all(|r| r == "LIST tinycloud:"), + "{lists:?}" + ); + Ok(()) +} + +#[tokio::test] +async fn mock_s3_body_mismatch_sends_no_request() -> Result<()> { + for mode in [WriteMode::Conditional, WriteMode::CheckThenWrite] { + let (s3, config) = start_mock(Conditional::Honour).await?; + let store = S3SqlObjectStore::open_with_mode(&config, mode).await; + let k = key_for(b"abc"); + assert!(matches!( + store.put_if_absent(&k, b"abd".to_vec()).await, + Err(SqlStoreError::BodyMismatch) + )); + assert!(s3.requests.lock().is_empty()); + } + Ok(()) +} + +#[tokio::test] +async fn mock_s3_conditional_create_is_one_request_and_a_412_is_checked() -> Result<()> { + let (s3, config) = start_mock(Conditional::Honour).await?; + let store = S3SqlObjectStore::open_with_mode(&config, WriteMode::Conditional).await; + let b = body(40); + let k = key_for(&b); + assert_eq!( + store.put_if_absent(&k, b.clone()).await?, + PutOutcome::Created + ); + assert_eq!(s3.requests.lock().len(), 1); + assert!(s3.requests.lock()[0].ends_with("conditional=true")); + s3.requests.lock().clear(); + assert_eq!( + store.put_if_absent(&k, b).await?, + PutOutcome::AlreadyPresent + ); + // The refused PUT, then the check of the stored object. + let log = s3.requests.lock().clone(); + assert_eq!(log.len(), 2, "{log:?}"); + assert!(log[0].starts_with("PUT ") && log[1].starts_with("HEAD ")); + Ok(()) +} + +#[tokio::test] +async fn mock_s3_check_then_write_never_sends_an_unconditional_put_for_an_existing_object( +) -> Result<()> { + let (s3, config) = start_mock(Conditional::Ignore).await?; + let store = S3SqlObjectStore::open_with_mode(&config, WriteMode::CheckThenWrite).await; + let b = body(40); + let k = key_for(&b); + assert_eq!( + store.put_if_absent(&k, b.clone()).await?, + PutOutcome::Created + ); + s3.requests.lock().clear(); + assert_eq!( + store.put_if_absent(&k, b).await?, + PutOutcome::AlreadyPresent + ); + let log = s3.requests.lock().clone(); + assert_eq!(log.len(), 1, "{log:?}"); + assert!(log[0].starts_with("HEAD ")); + Ok(()) +} + +async fn probe_against( + conditional: Conditional, +) -> Result<(MockS3, Result)> { + let (s3, config) = start_mock(conditional).await?; + let store = S3SqlObjectStore::open_with_mode(&config, WriteMode::Conditional).await; + let mode = S3SqlObjectStore::probe(store.client(), store.bucket()).await; + Ok((s3, mode)) +} + +#[tokio::test] +async fn mock_s3_probe_selects_the_mode_the_provider_supports() -> Result<()> { + // Honoured: the second request is refused. + let (s3, mode) = probe_against(Conditional::Honour).await?; + assert_eq!(mode?, WriteMode::Conditional); + assert_eq!( + *s3.requests.lock(), + vec![ + "PUT sql/v1/_probe/v1 conditional=true", + "PUT sql/v1/_probe/v1 conditional=true" + ] + ); + assert!(s3.objects.lock().contains_key("sql/v1/_probe/v1")); + + // Ignored: the second request succeeds. + let (s3, mode) = probe_against(Conditional::Ignore).await?; + assert_eq!(mode?, WriteMode::CheckThenWrite); + assert_eq!(s3.requests.lock().len(), 2); + + // The provider does not implement the header. + let (_s3, mode) = probe_against(Conditional::Reject(501, "NotImplemented")).await?; + assert_eq!(mode?, WriteMode::CheckThenWrite); + + // Any other failure, a 400 included, is an error, not a silent fallback. + for (status, code) in [ + (400, "InvalidArgument"), + (400, "RequestTimeout"), + (400, "ExpiredToken"), + (500, "InternalError"), + (403, "AccessDenied"), + ] { + let (_s3, mode) = probe_against(Conditional::Reject(status, code)).await?; + let Err(SqlStoreError::Backend(message)) = mode else { + panic!("{status} {code}: expected a backend error"); + }; + assert!(message.contains(&format!("status {status}")), "{message}"); + } + Ok(()) +} + +#[tokio::test] +async fn mock_s3_a_rejected_conditional_create_is_an_error_with_its_status_and_code() -> Result<()> +{ + for (status, code) in [ + (400, "InvalidArgument"), + (400, "RequestTimeout"), + (501, "NotImplemented"), + ] { + let (s3, config) = start_mock(Conditional::Reject(status, code)).await?; + let store = S3SqlObjectStore::open_with_mode(&config, WriteMode::Conditional).await; + let b = body(40); + let k = key_for(&b); + let Err(SqlStoreError::Backend(message)) = store.put_if_absent(&k, b).await else { + panic!("{status} {code}: expected a backend error"); + }; + assert!( + message.contains(&format!("status {status}")) && message.contains(code), + "{message}" + ); + // No unconditional PUT follows. + assert!(s3 + .requests + .lock() + .iter() + .all(|r| r.ends_with("conditional=true"))); + } + Ok(()) +} + +#[tokio::test] +async fn mock_s3_probe_object_from_an_earlier_start_is_not_overwritten() -> Result<()> { + let (s3, config) = start_mock(Conditional::Honour).await?; + let store = S3SqlObjectStore::open_with_mode(&config, WriteMode::Conditional).await; + for _ in 0..2 { + let mode = S3SqlObjectStore::probe(store.client(), store.bucket()).await?; + assert_eq!(mode, WriteMode::Conditional); + } + // First start: created, then refused. Second start: refused at once. + assert_eq!(s3.requests.lock().len(), 3); + assert_eq!(s3.objects.lock().len(), 1); + Ok(()) +} + +// --------------------------------------------------------------------------- +// S3: the at-rest harness MinIO +// --------------------------------------------------------------------------- + +fn env(name: &str) -> Result { + std::env::var(name).with_context(|| format!("{name} (load the harness env)")) +} + +fn minio_config() -> Result { + // The S3 client reads credentials from the AWS_* environment. + std::env::set_var("AWS_ACCESS_KEY_ID", env("TC887_S3_ACCESS_KEY_ID")?); + std::env::set_var("AWS_SECRET_ACCESS_KEY", env("TC887_S3_SECRET_ACCESS_KEY")?); + std::env::set_var("AWS_REGION", env("TC887_S3_REGION")?); + std::env::set_var("AWS_EC2_METADATA_DISABLED", "true"); + Ok(serde_json::from_value(serde_json::json!({ + "bucket": env("TC887_S3_BUCKET")?, + "endpoint": env("TC887_S3_ENDPOINT")?, + }))?) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[ignore = "needs the at-rest harness MinIO (TC887_S3_* env)"] +async fn minio_probe_selects_conditional_writes() -> Result<()> { + let config = minio_config()?; + let store = S3SqlObjectStore::open_with_mode(&config, WriteMode::Conditional).await; + for _ in 0..2 { + let mode = limited( + "probe", + S3SqlObjectStore::probe(store.client(), store.bucket()), + ) + .await?; + println!("sql object store probe selected {}", mode.as_str()); + assert_eq!(mode, WriteMode::Conditional); + } + // `open` runs the probe and keeps its answer. + let opened = limited("open", S3SqlObjectStore::open(&config)).await?; + assert_eq!(opened.mode(), WriteMode::Conditional); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[ignore = "needs the at-rest harness MinIO (TC887_S3_* env)"] +async fn minio_conditional_conformance() -> Result<()> { + s3_conformance(&minio_config()?, WriteMode::Conditional).await; + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[ignore = "needs the at-rest harness MinIO (TC887_S3_* env)"] +async fn minio_check_then_write_conformance() -> Result<()> { + s3_conformance(&minio_config()?, WriteMode::CheckThenWrite).await; + Ok(()) +} + +/// The provider itself: concurrent conditional PUTs of one key, with no +/// store in between. Exactly one is created; the others are refused with 412 +/// (or 409 while another write of the key is in flight), and none hangs. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[ignore = "needs the at-rest harness MinIO (TC887_S3_* env)"] +async fn minio_concurrent_conditional_puts_create_one_object() -> Result<()> { + let config = minio_config()?; + let store = Arc::new(S3SqlObjectStore::open_with_mode(&config, WriteMode::Conditional).await); + for round in 0..10 { + let b = body(1024); + let key = key_for(&b); + let tasks: Vec<_> = (0..8) + .map(|_| { + let (store, key, b) = (store.clone(), key.as_str().to_owned(), b.clone()); + tokio::spawn(async move { + let started = std::time::Instant::now(); + let result = limited( + "conditional put", + store + .client() + .put_object() + .bucket(store.bucket()) + .key(key) + .if_none_match("*") + .body(aws_sdk_s3::primitives::ByteStream::from(b)) + .send(), + ) + .await; + let outcome = match result { + Ok(_) => "created".to_owned(), + Err(e) => { + format!("status {:?}", e.raw_response().map(|r| r.status().as_u16())) + } + }; + (outcome, started.elapsed()) + }) + }) + .collect(); + let mut outcomes = Vec::new(); + for task in tasks { + outcomes.push(task.await?); + } + let created = outcomes.iter().filter(|(o, _)| o == "created").count(); + let refused = outcomes + .iter() + .filter(|(o, _)| o == "status Some(412)" || o == "status Some(409)") + .count(); + assert!(created == 1 && refused == 7, "round {round}: {outcomes:?}"); + } + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[ignore = "needs the at-rest harness MinIO (TC887_S3_* env)"] +async fn minio_object_without_hash_metadata_is_an_error() -> Result<()> { + let config = minio_config()?; + for mode in [WriteMode::Conditional, WriteMode::CheckThenWrite] { + s3_object_without_hash_metadata_is_an_error(&config, mode).await; + } + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[ignore = "needs the at-rest harness MinIO (TC887_S3_* env)"] +async fn minio_kv_scan_ignores_sql_objects() -> Result<()> { + kv_scan_ignores_sql_objects(&minio_config()?).await; + Ok(()) +}