From ba339b9c82538cc33f27261bbbe604ed65b411df Mon Sep 17 00:00:00 2001 From: Naomi Date: Sat, 24 Oct 2026 03:51:06 +0000 Subject: [PATCH 1/8] sql_store: S3 and filesystem object stores (slice 1b PR B) Adds the S3 and filesystem backends of SqlObjectStore. Objects are created at most once and never replaced. - S3: conditional create (If-None-Match: *) with the SHA-256 as user metadata, a startup probe that selects conditional or check-then-create mode, range reads and head. - Filesystem: temp file, fsync, hard link create under {datadir}/sql-store, with boot checks for the root. - Conformance tests shared by the in-memory, filesystem and S3 stores; S3 tests run against an in-process S3 API and, in the at-rest-harness job, against the harness MinIO. - The S3 block store's start-up size scan lists the tinycloud: key prefix. Details: TC-1491 --- .github/workflows/rust.yml | 5 + CHANGELOG.md | 2 + docs/sql-block-store-contract.md | 25 + tinycloud-node-server/src/storage/mod.rs | 1 + tinycloud-node-server/src/storage/s3.rs | 5 +- .../src/storage/sql_object.rs | 852 +++++++++++++ .../tests/sql_object_store.rs | 1095 +++++++++++++++++ 7 files changed, 1984 insertions(+), 1 deletion(-) create mode 100644 tinycloud-node-server/src/storage/sql_object.rs create mode 100644 tinycloud-node-server/tests/sql_object_store.rs diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 0d9aa875..4d6479c3 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -332,6 +332,11 @@ 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) + 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 fb211081..42e889dd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,8 @@ ### Changes - Added `sql_store` settings, validation, and manifest format and timestamp rules; `enabled = true` refuses to start (no uploader in this build). +- 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. + - Log sidecar: request events with status 400 and above (node and ingress) and the `db_failure_5xx` diagnostic event are capped per minute for each sidecar instance: 10 for each 5xx class (`db_failure_5xx`, `node_request_5xx`, `ingress_request_5xx`) and 3 for each 4xx class (`node_request_4xx`, `ingress_request_4xx`). When a cap holds events back, the sidecar emits one `errors_dropped` `node_warning` event per class per minute with exactly the fields `class` (one of those five names), `dropped` (integer, 1 or more) and `window_s` (60). The reserve for the capped classes is at most 585,403,200 bytes per 30 days per sidecar instance (the caps at worst-valid sizes, plus one `errors_dropped` event per class per minute), inside a 600,000,000-byte reserve. Every other event stays unsampled, including slow requests with a status below 400. Each cap is one Vector `lua` transform per class. The shipped fields are pseudonymous or fixed vocabulary; no SQL, ids, DIDs, paths or messages. The reviewed N0 compose template is `8df7e0f2…` (raw `192dfb7e…`); the lineage with the earlier values is in `docs/releasing.md`. The hosted proof's 500-request burst and its control sends use slow status-200 requests, which no cap covers. The volume policy and arithmetic are in `docs/runbooks/tc-959-n0-log-shipping.md`. Details: TC-1843. - SQL local durability and reservation before commit (`docs/sql-local-durability.md`). Postgres stays the master copy. File-backed SQL databases run `PRAGMA synchronous = FULL` (`[global.storage.sql] synchronous = "full"|"normal"`, env `TINYCLOUD_STORAGE__SQL__SYNCHRONOUS`; `normal` selects the earlier fsync behaviour), and the directories and renames that make a database file or a hydrated cache visible are fsynced. A write is checked before COMMIT and rolled back whole if it would pass the owner's remaining storage budget (`402`), and, when configured, the per-database cap (`max_database_bytes`, `507 sql_database_size_limit`) or the disk headroom (`min_free_bytes`, bounded by the WAL and checkpoint growth the transaction can cause and shared by all databases on the node, `507 sql_disk_headroom`). `max_database_bytes` and `min_free_bytes` default to `0` (off) and are set by config. A COMMIT that fails on disk answers `500 sql_commit_failed` and the node re-hydrates the database from Postgres. A write that fails because the disk is full answers `507 sql_disk_headroom`, whatever `min_free_bytes` is, not the owner's quota or the size limit. Every write request runs in one transaction (a single statement, a batch, a nested write (`WITH … INSERT`) and a request that carries a `PRAGMA` included), so such a request counts two more native SQL calls in request telemetry, and all of them get the same owner budget, size cap and disk gate. A request that wrote nothing is not gated and reserves nothing. The growth of the main file that the next checkpoint will cause stays held in the disk reservation from the commit until a checkpoint runs or the database's connection closes. No migration. Details: TC-1491. - `MemorySink` caps its size hint at 64 MiB and treats a failed reservation as no hint. Details: TC-1491. diff --git a/docs/sql-block-store-contract.md b/docs/sql-block-store-contract.md index fa8a1a8f..4cd26aab 100644 --- a/docs/sql-block-store-contract.md +++ b/docs/sql-block-store-contract.md @@ -698,3 +698,28 @@ 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, 400 or NotImplemented answer for the header, selects check-then-create mode. Any other failure is an error. The result is logged once at info level (target `tinycloud::sql_store`, the mode only). 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. + +### 12.2 Filesystem + +- The root is `{datadir}/sql-store`. Boot refuses a root that is not absolute, or 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, new directories and their parents are fsynced when created, and the temporary file is removed in every outcome. 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-node-server/src/storage/mod.rs b/tinycloud-node-server/src/storage/mod.rs index df299470..82848665 100644 --- a/tinycloud-node-server/src/storage/mod.rs +++ b/tinycloud-node-server/src/storage/mod.rs @@ -1,3 +1,4 @@ pub mod file_system; pub mod s3; pub mod size; +pub mod sql_object; diff --git a/tinycloud-node-server/src/storage/s3.rs b/tinycloud-node-server/src/storage/s3.rs index 33aaf0ff..6131e911 100644 --- a/tinycloud-node-server/src/storage/s3.rs +++ b/tinycloud-node-server/src/storage/s3.rs @@ -77,7 +77,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 @@ -104,6 +104,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 `tinycloud:`; only such keys count towards a space. + .prefix("tinycloud:") .into_paginator() .send(); // get the sum of all objects per space across all pages 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..f713f23c --- /dev/null +++ b/tinycloud-node-server/src/storage/sql_object.rs @@ -0,0 +1,852 @@ +//! 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, + fs::{File, OpenOptions}, + future::Future, + io::{ErrorKind, Read, Seek, SeekFrom, Write}, + path::{Component, Path, PathBuf}, + sync::{Arc, Mutex}, + 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 that serialise check-then-create inside this process. + gates: Mutex>>>, +} + +/// 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, + /// 409 ConditionalRequestConflict: another conditional write of the key + /// is in flight. + InFlight, + /// The provider does not implement the header (501, 400, NotImplemented). + Unsupported, + Failed(SqlStoreError), +} + +fn status(error: &SdkError) -> Option { + error.raw_response().map(|r| r.status().as_u16()) +} + +fn backend(op: &str, error: &SdkError) -> SqlStoreError { + // Status and error code only: the message and URL can carry the key. + let status = status(error).map_or_else(|| "none".to_owned(), |s| s.to_string()); + let code = ProvideErrorMetadata::code(error).unwrap_or("none"); + SqlStoreError::Backend(format!("s3 {op} failed (status {status}, code {code})")) +} + +fn classify_put(result: Result>) -> PutResult { + let error = match result { + Ok(_) => return PutResult::Created, + Err(error) => error, + }; + let code = ProvideErrorMetadata::code(&error).map(str::to_owned); + match (status(&error), code.as_deref()) { + (Some(412), _) | (_, Some("PreconditionFailed")) => PutResult::PreconditionFailed, + (Some(409), Some("ConditionalRequestConflict")) => PutResult::InFlight, + (Some(501 | 400), _) | (_, Some("NotImplemented")) => PutResult::Unsupported, + _ => 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. +/// The second answering 412 means the provider honours the condition; the +/// second succeeding, or either answering that the header is unsupported, +/// means it does not. A first 412 (the probe object exists from an earlier +/// start) already shows the condition is honoured. +async fn probe_with(mut put: F) -> Result +where + F: FnMut() -> Fut, + Fut: Future, +{ + for attempt in 0..2 { + match put().await { + PutResult::PreconditionFailed | PutResult::InFlight => { + return Ok(WriteMode::Conditional) + } + PutResult::Unsupported => return Ok(WriteMode::CheckThenWrite), + PutResult::Created if attempt == 1 => return Ok(WriteMode::CheckThenWrite), + 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) + } +} + +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("s3 returned an invalid content range".into()); + 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 a provider that answers 501, + /// 400 or NotImplemented for the header, selects the check-then-create + /// mode. Any other failure is an error, not a fallback. The object stays. + pub async fn probe(client: &Client, bucket: &str) -> Result { + let mode = + probe_with(|| put_object(client, bucket, PROBE_KEY, PROBE_BODY.to_vec(), true)).await?; + tracing::info!(target: LOG_TARGET, mode = mode.as_str(), "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 => { + return Err(SqlStoreError::Backend( + "s3 provider refused the conditional create".into(), + )) + } + 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 gate = self + .gates + .lock() + .unwrap() + .entry(key.as_str().to_owned()) + .or_default() + .clone(); + let outcome = async { + let _held = gate.lock().await; + 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())), + } + } + } + } + .await; + let mut gates = self.gates.lock().unwrap(); + // The map holds one reference and this call one; any other is a waiter. + if Arc::strong_count(&gate) == 2 { + gates.remove(key.as_str()); + } + outcome + } +} + +#[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); + } + 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(); + 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(range.resolve(total).map(|r| ObjectRange { + total_size: total, + start: r.start(), + data: data[r.start() as usize..=r.end() 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, +} + +/// Refuse a SQL store root the node must not use: one that is not absolute, +/// or that is equal to, inside, or a parent of the KV block-store root. Both +/// paths are compared as written (they must be absolute and free of `.` 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", + )); + } + for path in [root, blocks_root] { + if !path.is_absolute() { + return Err(SqlStoreError::Invalid( + "the SQL store root and the block store root must be absolute paths", + )); + } + if path + .components() + .any(|c| matches!(c, Component::CurDir | Component::ParentDir)) + { + return Err(SqlStoreError::Invalid( + "the SQL store root and the block store root must not contain . or .. 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`, fsyncing each new one +/// and its parent. +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 => {} + 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(resolved) = range.resolve(total) else { + return Ok(None); + }; + let len = usize::try_from(resolved.end() - resolved.start() + 1) + .map_err(|_| SqlStoreError::Invalid("range does not fit in memory"))?; + let mut data = vec![0u8; len]; + file.seek(SeekFrom::Start(resolved.start())) + .and_then(|_| file.read_exact(&mut data)) + .map_err(|e| io_error("read", &e))?; + Ok(Some(ObjectRange { + total_size: total, + start: resolved.start(), + 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)?; + 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_absolute_and_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", + "/", + "sql-store", + "./sql-store", + "/data/../data/sql-store", + ] { + assert!(check_fs_root(Path::new(bad), blocks).is_err(), "{bad}"); + } + // The block store root is checked the same way. + assert!(check_fs_root(Path::new("/data/sql-store"), Path::new("blocks")).is_err()); + assert!(check_fs_root( + Path::new("/data/sql-store"), + Path::new("/data/../data/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 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, 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)) + } + + #[test] + fn probe_selects_the_write_mode_from_the_second_answer() { + use PutResult::*; + let (mode, calls) = scripted(vec![Created, PreconditionFailed]); + assert_eq!((mode.unwrap(), calls), (WriteMode::Conditional, 2)); + // An object left by an earlier start answers the first request. + let (mode, calls) = scripted(vec![PreconditionFailed]); + assert_eq!((mode.unwrap(), calls), (WriteMode::Conditional, 1)); + let (mode, calls) = scripted(vec![Created, Created]); + assert_eq!((mode.unwrap(), calls), (WriteMode::CheckThenWrite, 2)); + let (mode, calls) = scripted(vec![Unsupported]); + assert_eq!((mode.unwrap(), calls), (WriteMode::CheckThenWrite, 1)); + let (mode, calls) = scripted(vec![Created, Unsupported]); + assert_eq!((mode.unwrap(), 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(_)))); + } +} 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..99bdf017 --- /dev/null +++ b/tinycloud-node-server/tests/sql_object_store.rs @@ -0,0 +1,1095 @@ +//! 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() +} + +/// 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 = 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 = body(2048); + 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()); + assert!(matches!( + FsSqlObjectStore::open("relative/sql-store".into(), &blocks), + Err(SqlStoreError::Invalid(_)) + )); +} + +// --------------------------------------------------------------------------- +// 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 rejects the header. + for (status, code) in [(501, "NotImplemented"), (400, "InvalidArgument")] { + let (_s3, mode) = probe_against(Conditional::Reject(status, code)).await?; + assert_eq!(mode?, WriteMode::CheckThenWrite, "{status}"); + } + + // Any other failure is an error, not a silent fallback. + for (status, code) in [(500, "InternalError"), (403, "AccessDenied")] { + let (_s3, mode) = probe_against(Conditional::Reject(status, code)).await?; + assert!(matches!(mode, Err(SqlStoreError::Backend(_))), "{status}"); + } + 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(()) +} + +#[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(()) +} From c097b8c1d0392543c5cde1da0764e30566def372 Mon Sep 17 00:00:00 2001 From: Naomi Date: Sat, 24 Oct 2026 04:01:45 +0000 Subject: [PATCH 2/8] sql_store: an empty ranged answer selects nothing Details: TC-1491 --- tinycloud-node-server/src/storage/sql_object.rs | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/tinycloud-node-server/src/storage/sql_object.rs b/tinycloud-node-server/src/storage/sql_object.rs index f713f23c..0bfaa5da 100644 --- a/tinycloud-node-server/src/storage/sql_object.rs +++ b/tinycloud-node-server/src/storage/sql_object.rs @@ -188,7 +188,7 @@ fn parse_sha256(value: &str) -> Result<[u8; 32], SqlStoreError> { /// `bytes first-last/total`. fn parse_content_range(value: &str) -> Result<(u64, u64, u64), SqlStoreError> { - let bad = || SqlStoreError::Backend("s3 returned an invalid content range".into()); + 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)?; @@ -378,6 +378,11 @@ impl SqlObjectStore for S3SqlObjectStore { .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)?; From 4f39bf1730dd5db347acdeeabf72b703b434f3cc Mon Sep 17 00:00:00 2001 From: Naomi Date: Sat, 24 Oct 2026 04:09:28 +0000 Subject: [PATCH 3/8] sql_store: name the failing stage in S3 backend errors Details: TC-1491 --- .../src/storage/sql_object.rs | 21 +++++++++++++++++-- 1 file changed, 19 insertions(+), 2 deletions(-) diff --git a/tinycloud-node-server/src/storage/sql_object.rs b/tinycloud-node-server/src/storage/sql_object.rs index 0bfaa5da..d64d3416 100644 --- a/tinycloud-node-server/src/storage/sql_object.rs +++ b/tinycloud-node-server/src/storage/sql_object.rs @@ -103,11 +103,28 @@ 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 { - // Status and error code only: the message and URL can carry the key. let status = status(error).map_or_else(|| "none".to_owned(), |s| s.to_string()); let code = ProvideErrorMetadata::code(error).unwrap_or("none"); - SqlStoreError::Backend(format!("s3 {op} failed (status {status}, code {code})")) + SqlStoreError::Backend(format!( + "s3 {op} failed ({}, status {status}, code {code})", + failure_kind(error) + )) } fn classify_put(result: Result>) -> PutResult { From 7245283118574b45c5bf73cd4b046bc30c477596 Mon Sep 17 00:00:00 2001 From: Naomi Date: Sat, 24 Oct 2026 04:14:52 +0000 Subject: [PATCH 4/8] sql_store: include the dispatch cause in S3 backend errors Details: TC-1491 --- tinycloud-node-server/src/storage/sql_object.rs | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/tinycloud-node-server/src/storage/sql_object.rs b/tinycloud-node-server/src/storage/sql_object.rs index d64d3416..4caabb62 100644 --- a/tinycloud-node-server/src/storage/sql_object.rs +++ b/tinycloud-node-server/src/storage/sql_object.rs @@ -121,8 +121,16 @@ fn failure_kind(error: &SdkError) -> &'static str { 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 = std::error::Error::source(failure); + while let Some(error) = source { + cause.push_str(&format!(": {error}")); + source = error.source(); + } + } SqlStoreError::Backend(format!( - "s3 {op} failed ({}, status {status}, code {code})", + "s3 {op} failed ({}, status {status}, code {code}){cause}", failure_kind(error) )) } From 267170f11a17853a7471634e587a140edde76843 Mon Sep 17 00:00:00 2001 From: Naomi Date: Sat, 24 Oct 2026 04:19:57 +0000 Subject: [PATCH 5/8] sql_store: compile fix for the dispatch cause Details: TC-1491 --- tinycloud-node-server/src/storage/sql_object.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tinycloud-node-server/src/storage/sql_object.rs b/tinycloud-node-server/src/storage/sql_object.rs index 4caabb62..85bd5102 100644 --- a/tinycloud-node-server/src/storage/sql_object.rs +++ b/tinycloud-node-server/src/storage/sql_object.rs @@ -123,7 +123,9 @@ fn backend(op: &str, error: &SdkError) -> SqlStoreEr let code = ProvideErrorMetadata::code(error).unwrap_or("none"); let mut cause = String::new(); if let SdkError::DispatchFailure(failure) = error { - let mut source = std::error::Error::source(failure); + 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(); From 9aaafea47548ae431388e5eb070f6dedbd3afabb Mon Sep 17 00:00:00 2001 From: Naomi Date: Sat, 24 Oct 2026 04:26:42 +0000 Subject: [PATCH 6/8] sql_store: serialise creators of one key in a process; test concurrent conditional PUTs on the provider Details: TC-1491 --- .../src/storage/sql_object.rs | 60 +++++++++---------- .../tests/sql_object_store.rs | 52 ++++++++++++++++ 2 files changed, 81 insertions(+), 31 deletions(-) diff --git a/tinycloud-node-server/src/storage/sql_object.rs b/tinycloud-node-server/src/storage/sql_object.rs index 85bd5102..9afdfead 100644 --- a/tinycloud-node-server/src/storage/sql_object.rs +++ b/tinycloud-node-server/src/storage/sql_object.rs @@ -82,7 +82,7 @@ pub struct S3SqlObjectStore { client: Client, bucket: String, mode: WriteMode, - /// Per-key gates that serialise check-then-create inside this process. + /// Per-key gates: creators of one key in this process go one at a time. gates: Mutex>>>, } @@ -319,34 +319,15 @@ impl S3SqlObjectStore { key: &ObjectKey, body: Vec, ) -> Result { - let gate = self - .gates - .lock() - .unwrap() - .entry(key.as_str().to_owned()) - .or_default() - .clone(); - let outcome = async { - let _held = gate.lock().await; - 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())), - } - } - } - } - .await; - let mut gates = self.gates.lock().unwrap(); - // The map holds one reference and this call one; any other is a waiter. - if Arc::strong_count(&gate) == 2 { - gates.remove(key.as_str()); + 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())), + }, } - outcome } } @@ -360,10 +341,27 @@ impl SqlObjectStore for S3SqlObjectStore { if &sha256(&body) != key.sha256() { return Err(SqlStoreError::BodyMismatch); } - match self.mode { - WriteMode::Conditional => self.create_conditionally(key, body).await, - WriteMode::CheckThenWrite => self.check_then_create(key, body).await, + // Creators of one key in this process go one at a time. + let gate = self + .gates + .lock() + .unwrap() + .entry(key.as_str().to_owned()) + .or_default() + .clone(); + let outcome = { + 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, + } + }; + let mut gates = self.gates.lock().unwrap(); + // The map holds one reference and this call one; any other is a waiter. + if Arc::strong_count(&gate) == 2 { + gates.remove(key.as_str()); } + outcome } async fn get_range( diff --git a/tinycloud-node-server/tests/sql_object_store.rs b/tinycloud-node-server/tests/sql_object_store.rs index 99bdf017..898506b1 100644 --- a/tinycloud-node-server/tests/sql_object_store.rs +++ b/tinycloud-node-server/tests/sql_object_store.rs @@ -1077,6 +1077,58 @@ async fn minio_check_then_write_conformance() -> Result<()> { 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<()> { From 8d6935857e3f913db2f62d7f94d07e27ab1ce1d0 Mon Sep 17 00:00:00 2001 From: Naomi Date: Sat, 24 Oct 2026 05:01:57 +0000 Subject: [PATCH 7/8] sql_store: select byte ranges with a local helper pinned to ByteRangeSpec Details: TC-1491 --- .../src/storage/sql_object.rs | 50 ++++++++++++++++--- 1 file changed, 43 insertions(+), 7 deletions(-) diff --git a/tinycloud-node-server/src/storage/sql_object.rs b/tinycloud-node-server/src/storage/sql_object.rs index 9afdfead..06618229 100644 --- a/tinycloud-node-server/src/storage/sql_object.rs +++ b/tinycloud-node-server/src/storage/sql_object.rs @@ -206,6 +206,20 @@ fn same_object(meta: &ObjectMeta, key: &ObjectKey, len: u64) -> Result 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() @@ -425,10 +439,10 @@ impl SqlObjectStore for S3SqlObjectStore { // A provider that ignored the header sent the whole object. None => { let total = data.len() as u64; - Ok(range.resolve(total).map(|r| ObjectRange { + Ok(select_range(range, total).map(|(first, last)| ObjectRange { total_size: total, - start: r.start(), - data: data[r.start() as usize..=r.end() as usize].to_vec(), + start: first, + data: data[first as usize..=last as usize].to_vec(), })) } } @@ -677,18 +691,18 @@ fn fs_get_range(path: &Path, range: ByteRangeSpec) -> Result )); } let total = metadata.len(); - let Some(resolved) = range.resolve(total) else { + let Some((first, last)) = select_range(range, total) else { return Ok(None); }; - let len = usize::try_from(resolved.end() - resolved.start() + 1) + 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(resolved.start())) + 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: resolved.start(), + start: first, data, })) } @@ -818,6 +832,28 @@ mod tests { } } + #[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!( From 635d73cf8c1b2cbaba666d404f8f35e50138d485 Mon Sep 17 00:00:00 2001 From: Naomi Date: Sat, 24 Oct 2026 05:55:27 +0000 Subject: [PATCH 8/8] sql_store: harden the object stores - The probe maps only 501 and NotImplemented to check-then-create; any other status is an error. The deciding answer's status and code are logged with the mode, and a rejected conditional create reports its status and code. - The per-key gate is released by a guard, so a cancelled create leaves no entry behind. - Relative filesystem roots are made absolute before the block store root check; the parent of an existing directory is fsynced on create. - The S3 scan prefix and the space id text share one constant, SPACE_ID_PREFIX. - Tests that planted different bytes derive them from the wanted bytes. - The new CI step has a 10 minute timeout. Contract section 12 updated. Details: TC-1491 --- .github/workflows/rust.yml | 1 + docs/sql-block-store-contract.md | 9 +- tinycloud-auth/src/resource.rs | 5 +- tinycloud-node-server/src/storage/s3.rs | 24 +- .../src/storage/sql_object.rs | 357 +++++++++++++----- .../tests/sql_object_store.rs | 73 +++- 6 files changed, 360 insertions(+), 109 deletions(-) diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 4d6479c3..c24ebbb7 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -333,6 +333,7 @@ jobs: 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) diff --git a/docs/sql-block-store-contract.md b/docs/sql-block-store-contract.md index 4cd26aab..c9104903 100644 --- a/docs/sql-block-store-contract.md +++ b/docs/sql-block-store-contract.md @@ -707,14 +707,15 @@ Slice 1b PR B adds the two backends of the object store trait in `tinycloud-node - 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, 400 or NotImplemented answer for the header, selects check-then-create mode. Any other failure is an error. The result is logged once at info level (target `tinycloud::sql_store`, the mode only). 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. +- 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`. Boot refuses a root that is not absolute, or that is equal to, inside, or a parent of the block store root. The filesystem store is not supported on Windows. +- 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, new directories and their parents are fsynced when created, and the temporary file is removed in every outcome. A leftover temporary file does not affect a later create. +- 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 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/s3.rs b/tinycloud-node-server/src/storage/s3.rs index 6131e911..3a414e6d 100644 --- a/tinycloud-node-server/src/storage/s3.rs +++ b/tinycloud-node-server/src/storage/s3.rs @@ -17,12 +17,24 @@ use rocket::{async_trait, http::hyper::Uri}; use serde::{Deserialize, Serialize}; use serde_with::{serde_as, DisplayFromStr}; use std::{collections::HashMap, io::Error as IoError, ops::AddAssign, pin::Pin, 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; use super::{file_system, size::SpaceSizes}; +/// Prefix of the boot scan's listing: every KV object key starts with it. +pub(crate) const KV_SCAN_PREFIX: &str = SPACE_ID_PREFIX; + +/// The key of a KV block: `{space}/{base64url(hash)}`. +pub(crate) fn kv_object_key(space: &SpaceId, id: &Hash) -> String { + format!( + "{}/{}", + space, + base64::encode_config(id.as_ref(), base64::URL_SAFE) + ) +} + /// Pinned so an SDK bump cannot change client defaults unnoticed. pub fn behavior_version() -> BehaviorVersion { BehaviorVersion::v2026_01_12() @@ -105,8 +117,8 @@ impl S3BlockStore { .list_objects_v2() .bucket(&config.bucket) // Every KV object key is `{space}/{hash}` and a space id string - // starts with `tinycloud:`; only such keys count towards a space. - .prefix("tinycloud:") + // 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 @@ -135,11 +147,7 @@ impl S3BlockStore { } fn key(&self, space: &SpaceId, id: &Hash) -> String { - format!( - "{}/{}", - space, - base64::encode_config(id.as_ref(), base64::URL_SAFE) - ) + kv_object_key(space, id) } async fn increment_size(&self, space: &SpaceId, size: u64) { diff --git a/tinycloud-node-server/src/storage/sql_object.rs b/tinycloud-node-server/src/storage/sql_object.rs index 06618229..1cf7262b 100644 --- a/tinycloud-node-server/src/storage/sql_object.rs +++ b/tinycloud-node-server/src/storage/sql_object.rs @@ -17,11 +17,12 @@ use std::{ collections::HashMap, + fmt, fs::{File, OpenOptions}, future::Future, io::{ErrorKind, Read, Seek, SeekFrom, Write}, path::{Component, Path, PathBuf}, - sync::{Arc, Mutex}, + sync::{Arc, Mutex, MutexGuard, PoisonError}, time::Duration, }; @@ -83,19 +84,79 @@ pub struct S3SqlObjectStore { bucket: String, mode: WriteMode, /// Per-key gates: creators of one key in this process go one at a time. - gates: Mutex>>>, + 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, + PreconditionFailed(Answer), /// 409 ConditionalRequestConflict: another conditional write of the key /// is in flight. - InFlight, - /// The provider does not implement the header (501, 400, NotImplemented). - Unsupported, + InFlight(Answer), + /// The provider does not implement the header (501, NotImplemented). + Unsupported(Answer), Failed(SqlStoreError), } @@ -142,11 +203,13 @@ fn classify_put(result: Result>) -> PutResult { Ok(_) => return PutResult::Created, Err(error) => error, }; - let code = ProvideErrorMetadata::code(&error).map(str::to_owned); - match (status(&error), code.as_deref()) { - (Some(412), _) | (_, Some("PreconditionFailed")) => PutResult::PreconditionFailed, - (Some(409), Some("ConditionalRequestConflict")) => PutResult::InFlight, - (Some(501 | 400), _) | (_, Some("NotImplemented")) => PutResult::Unsupported, + 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)), } } @@ -173,23 +236,28 @@ async fn put_object( classify_put(request.send().await) } -/// Decide the write mode from the answers to two conditional PUTs of one key. -/// The second answering 412 means the provider honours the condition; the -/// second succeeding, or either answering that the header is unsupported, -/// means it does not. A first 412 (the probe object exists from an earlier -/// start) already shows the condition is honoured. -async fn probe_with(mut put: F) -> Result +/// 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 | PutResult::InFlight => { - return Ok(WriteMode::Conditional) + 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::Unsupported => return Ok(WriteMode::CheckThenWrite), - PutResult::Created if attempt == 1 => return Ok(WriteMode::CheckThenWrite), PutResult::Created => {} PutResult::Failed(error) => return Err(error), } @@ -279,13 +347,20 @@ impl S3SqlObjectStore { /// 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 a provider that answers 501, - /// 400 or NotImplemented for the header, selects the check-then-create - /// mode. Any other failure is an error, not a fallback. The object stays. + /// 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 = + 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(), "sql object store write mode"); + tracing::info!( + target: LOG_TARGET, + mode = mode.as_str(), + answer = %answer, + "sql object store write mode" + ); Ok(mode) } @@ -311,14 +386,12 @@ impl S3SqlObjectStore { 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 => { + 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 => { - return Err(SqlStoreError::Backend( - "s3 provider refused the conditional create".into(), - )) + PutResult::Unsupported(answer) => { + return Err(SqlStoreError::Backend(format!("s3 put failed ({answer})"))) } PutResult::Failed(error) => return Err(error), } @@ -356,26 +429,12 @@ impl SqlObjectStore for S3SqlObjectStore { return Err(SqlStoreError::BodyMismatch); } // Creators of one key in this process go one at a time. - let gate = self - .gates - .lock() - .unwrap() - .entry(key.as_str().to_owned()) - .or_default() - .clone(); - let outcome = { - 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, - } - }; - let mut gates = self.gates.lock().unwrap(); - // The map holds one reference and this call one; any other is a waiter. - if Arc::strong_count(&gate) == 2 { - gates.remove(key.as_str()); + 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, } - outcome } async fn get_range( @@ -493,28 +552,29 @@ pub struct FsSqlObjectStore { root: PathBuf, } -/// Refuse a SQL store root the node must not use: one that is not absolute, -/// or that is equal to, inside, or a parent of the KV block-store root. Both -/// paths are compared as written (they must be absolute and free of `.` and -/// `..`); nothing is read from the filesystem. +/// `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", )); } - for path in [root, blocks_root] { - if !path.is_absolute() { + 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 be absolute paths", - )); - } - if path - .components() - .any(|c| matches!(c, Component::CurDir | Component::ParentDir)) - { - return Err(SqlStoreError::Invalid( - "the SQL store root and the block store root must not contain . or .. components", + "the SQL store root and the block store root must not contain .. components", )); } } @@ -523,12 +583,12 @@ pub fn check_fs_root(root: &Path, blocks_root: &Path) -> Result<(), SqlStoreErro "the SQL store root must differ from the block store root", )); } - if root.starts_with(blocks_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) { + if blocks_root.starts_with(&root) { return Err(SqlStoreError::Invalid( "the SQL store root must not contain the block store root", )); @@ -572,8 +632,10 @@ fn sync_dir(dir: &Path) -> Result<(), SqlStoreError> { .map_err(|e| io_error("directory sync", &e)) } -/// Create the directories from `root` down to `dir`, fsyncing each new one -/// and its parent. +/// 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) @@ -587,7 +649,7 @@ fn create_dirs(root: &Path, dir: &Path) -> Result<(), SqlStoreError> { sync_dir(¤t)?; sync_dir(&parent)?; } - Err(e) if e.kind() == ErrorKind::AlreadyExists => {} + Err(e) if e.kind() == ErrorKind::AlreadyExists => sync_dir(&parent)?, Err(e) => return Err(io_error("create directory", &e)), } } @@ -722,6 +784,7 @@ impl FsSqlObjectStore { /// 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() { @@ -778,7 +841,7 @@ mod tests { #[cfg(not(windows))] #[test] - fn fs_root_must_be_absolute_and_outside_the_block_store() { + fn fs_root_must_be_outside_the_block_store() { let blocks = Path::new("/data/blocks"); for ok in [ "/data/sql-store", @@ -794,14 +857,10 @@ mod tests { "/data/blocks/sql-store", "/data", "/", - "sql-store", - "./sql-store", "/data/../data/sql-store", ] { assert!(check_fs_root(Path::new(bad), blocks).is_err(), "{bad}"); } - // The block store root is checked the same way. - assert!(check_fs_root(Path::new("/data/sql-store"), Path::new("blocks")).is_err()); assert!(check_fs_root( Path::new("/data/sql-store"), Path::new("/data/../data/blocks") @@ -809,6 +868,37 @@ mod tests { .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"); @@ -874,7 +964,7 @@ mod tests { } } - fn scripted(answers: Vec) -> (Result, usize) { + 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(|| { @@ -889,20 +979,35 @@ mod tests { (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 (mode, calls) = scripted(vec![Created, PreconditionFailed]); - assert_eq!((mode.unwrap(), calls), (WriteMode::Conditional, 2)); + 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![PreconditionFailed]); - assert_eq!((mode.unwrap(), calls), (WriteMode::Conditional, 1)); + let (mode, calls) = scripted(vec![refused()]); + assert_eq!((mode.unwrap().0, calls), (WriteMode::Conditional, 1)); let (mode, calls) = scripted(vec![Created, Created]); - assert_eq!((mode.unwrap(), calls), (WriteMode::CheckThenWrite, 2)); - let (mode, calls) = scripted(vec![Unsupported]); - assert_eq!((mode.unwrap(), calls), (WriteMode::CheckThenWrite, 1)); - let (mode, calls) = scripted(vec![Created, Unsupported]); - assert_eq!((mode.unwrap(), calls), (WriteMode::CheckThenWrite, 2)); + 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] @@ -915,4 +1020,84 @@ mod tests { 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::{kv_object_key, 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 key = kv_object_key(&space, &hash(b"block")); + assert!(key.starts_with(KV_SCAN_PREFIX), "{key}"); + assert!(key.starts_with(&format!("{space}/"))); + // 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 index 898506b1..6cb40a8e 100644 --- a/tinycloud-node-server/tests/sql_object_store.rs +++ b/tinycloud-node-server/tests/sql_object_store.rs @@ -49,6 +49,13 @@ fn body(len: usize) -> Vec { (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) @@ -241,7 +248,11 @@ async fn conformance(store: Arc, planter: Option if let Some(planter) = planter { for (len, planted_len) in [(64usize, 64usize), (64, 7)] { let wanted = body(len); - let other = body(planted_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; @@ -375,7 +386,7 @@ async fn fs_leftover_temp_files_never_block_a_create() { PutOutcome::AlreadyPresent ); assert_eq!(names_in(parent), expected); - let planted = body(2048); + 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(); @@ -464,10 +475,16 @@ fn fs_open_refuses_a_root_that_overlaps_the_block_store() { )); } 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("relative/sql-store".into(), &blocks), + FsSqlObjectStore::open( + "tc1491-relative/blocks/sql-store".into(), + Path::new("./tc1491-relative/blocks") + ), Err(SqlStoreError::Invalid(_)) )); + assert!(!Path::new("tc1491-relative").exists()); } // --------------------------------------------------------------------------- @@ -995,16 +1012,52 @@ async fn mock_s3_probe_selects_the_mode_the_provider_supports() -> Result<()> { assert_eq!(mode?, WriteMode::CheckThenWrite); assert_eq!(s3.requests.lock().len(), 2); - // The provider rejects the header. - for (status, code) in [(501, "NotImplemented"), (400, "InvalidArgument")] { + // 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?; - assert_eq!(mode?, WriteMode::CheckThenWrite, "{status}"); + let Err(SqlStoreError::Backend(message)) = mode else { + panic!("{status} {code}: expected a backend error"); + }; + assert!(message.contains(&format!("status {status}")), "{message}"); } + Ok(()) +} - // Any other failure is an error, not a silent fallback. - for (status, code) in [(500, "InternalError"), (403, "AccessDenied")] { - let (_s3, mode) = probe_against(Conditional::Reject(status, code)).await?; - assert!(matches!(mode, Err(SqlStoreError::Backend(_))), "{status}"); +#[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(()) }