diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 7d67f8f4..33de19f6 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -355,6 +355,11 @@ jobs: (set -a; . "$TC887_RUN_DIR/harness.env"; set +a cargo test --locked -p tinycloud-node --test s3_block_store -- --ignored) + - name: S3 block store with encryption at rest (against the harness MinIO) + run: | + (set -a; . "$TC887_RUN_DIR/harness.env"; set +a + cargo test --locked -p tinycloud-node --test at_rest_s3_store -- --ignored) + - name: Harness logs if: failure() run: | diff --git a/CHANGELOG.md b/CHANGELOG.md index 50a6bbc5..b2e33bb8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,7 +2,7 @@ ## [Unreleased] -- Encryption at rest for KV blocks, FileSystem store (TC-887 S3, off by default). With `at_rest.mode = "encrypt"`/`"strict"` and keys set up, new KV blocks are sealed (TCE1) under anonymous names and served back as plaintext; legacy blocks stay readable in `encrypt`, are refused in `strict`, and a node with locked keys answers 503 for sealed data. **Behaviour change on nodes with at-rest keys only:** a writing invocation is now authorized (and its spaces checked as hosted) before any key work, so a put to a space the node does not host, or without authority there, answers **401 instead of 404**; nodes without keys are unchanged. See `docs/at-rest/INTERFACES.md` §7. +- Encryption at rest for KV blocks, FileSystem and S3 stores (TC-887 S3, off by default). With `at_rest.mode = "encrypt"`/`"strict"` and keys set up, new KV blocks are sealed (TCE1) under anonymous names and served back as plaintext; S3 range reads fetch the header and then only the covering segments, pinned to one object version. Legacy blocks stay readable in `encrypt`, are refused in `strict`, and a node with locked keys answers 503 for sealed data. **Behaviour change on nodes with at-rest keys only:** a writing invocation is now authorized (and its spaces checked as hosted) before any key work, so a put to a space the node does not host, or without authority there, answers **401 instead of 404**; nodes without keys are unchanged. See `docs/at-rest/INTERFACES.md` §7. - New migration `m20261017_000000_epoch_order_space_index` adds `idx_epoch_order_space_parent_child ON epoch_order (space, parent, child)` for the head-epoch anti-join that every write runs under its space lock (TC-1274). On a fresh database the build is instant. On PostgreSQL, if a valid index of that name and shape already exists, the migration issues no DDL and takes no lock on `epoch_order`. If the index of that name is INVALID or has a different shape, the boot fails, and nothing is applied, with an error naming the fix. On an existing database **without** the index, the migration builds it inside the boot's migration transaction. That holds a SHARE lock that blocks writes for the whole build, and the lock wait is bounded by the 30 s `lock_timeout` (TC-1176). **Runbook (PostgreSQL, before upgrading a node with history):** in autocommit, run `CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_epoch_order_space_parent_child ON epoch_order (space, parent, child);`. Then confirm `SELECT indisvalid FROM pg_index WHERE indexrelid = 'idx_epoch_order_space_parent_child'::regclass;` returns `t`. If it returns `f`, run `DROP INDEX CONCURRENTLY idx_epoch_order_space_parent_child;` and build it again. `down` keeps the index. MySQL is covered only by the generated statement, because the full MySQL migration chain does not run (TC-1314). - Request events include validated/echoed UUID correlation, typed replacement flags, node-keyed target account/space pseudonyms and per-request database acquisition/execution timings and attempt counts. Driver/actor instrumentation includes failures and request-awaited audit batches; forwarding requires explicit first-party origin classification. A shared core `Pseudonymizer` preserves existing account/space hashes, adds the service/space-scoped database hash primitive for F6, and is derived once at startup for injection into consumers. See `docs/request-telemetry.md` for the wire contract, key custody, restart joinability and coverage/exclusions (TC-959 N2). - CACAO/SIWE delegation ingestion normalizes ECDSA recovery IDs 0/1 to 27/28 before hashing and persisting the canonical DAG-CBOR bytes (TC-1071). Equivalent low-s signatures now share one delegation CID, including the CID used for revocation lookup. High-s signatures are rejected during decoding, matching the existing EIP-191 verifier and EIP-2 low-s convention; UCAN/JWS bytes remain unchanged. Existing stored delegation IDs are not rewritten. diff --git a/docs/at-rest-escrow.md b/docs/at-rest-escrow.md index a083ef3c..71da932a 100644 --- a/docs/at-rest-escrow.md +++ b/docs/at-rest-escrow.md @@ -146,7 +146,7 @@ The same settings as compose env: `TINYCLOUD_AT_REST__MODE`, `TINYCLOUD_AT_REST_ - it keeps a copy of the sealed key, the bundle, the ceremony record and each space's wrapped data key in the block store (`_tc/at-rest/` in S3, `{datadir}/at-rest/_tc/at-rest/` on disk). Data-key objects are named by a keyed hash, not the space; - in this mode a copy that can't be written is logged and the boot goes on (nothing depends on it yet). With `encrypt` or `strict`, every boot-time copy must be written: a block store that refuses the write stops the boot; - nothing is encrypted. -- **`encrypt` and `strict` need a block store with a seal path.** The FileSystem store seals KV blocks (TC-887 S3); with S3 block storage, `encrypt` and `strict` are refused at boot in this build, so the node won't claim to encrypt while it writes plaintext. With a seal path, `encrypt` or `strict` refuses to start unless the recipients equal the bundle's, `verified_fp` equals its EFP, and the stored ceremony record verifies under the key inside the TEE. A Static-root node with no recipients may use `escrow_none = true` instead, which acknowledges that no escrow exists. +- **`encrypt` and `strict` need a block store with a seal path.** Both the FileSystem and the S3 store seal KV blocks (TC-887 S3), so both declare one; without it `encrypt` and `strict` are refused at boot, so the node never claims to encrypt while it writes plaintext. With a seal path, `encrypt` or `strict` refuses to start unless the recipients equal the bundle's, `verified_fp` equals its EFP, and the stored ceremony record verifies under the key inside the TEE. A Static-root node with no recipients may use `escrow_none = true` instead, which acknowledges that no escrow exists. - **In a TEE, the escrow recipients and `verified_fp` must be literal values in the measured compose.** With any at-rest config, the node reads its `app-compose.json` from the guest agent (`/Info`, `tcb_info.app_compose`), checks it against `compose_hash`, and refuses to boot unless `TINYCLOUD_AT_REST__ESCROW__RECIPIENTS` and `TINYCLOUD_AT_REST__ESCROW__VERIFIED_FP` are set there as literals (no `${…}`, no pass-through) equal to the effective config. Values from the encrypted environment, a file or anywhere else are refused. If the app-compose can't be read or parsed, the node refuses to boot. The `/Info` shape must be confirmed on a live CVM before the escrow ceremony. - **The escrow bundle is made or changed only at boot, never by a request.** Every endpoint below only reads it. - **An ARK never gets a second bundle for other holders.** The node refuses to boot (`escrow_holders_changed`) if the configured recipients differ from any existing bundle for its ARK: the active bundle, any stored escrow row, a ceremony record in the block store whose MAC verifies under the ARK, or a bundle object in the block store naming the ARK. This holds after a restore too, when the database lost its escrow rows. The block-store bundle object is unauthenticated, so it can only refuse, never be used. Changing a holder after the ceremony needs an ARK rotation (TC-1345). diff --git a/docs/at-rest/INTERFACES.md b/docs/at-rest/INTERFACES.md index d7a4e9ba..0abe38d6 100644 --- a/docs/at-rest/INTERFACES.md +++ b/docs/at-rest/INTERFACES.md @@ -299,7 +299,9 @@ redacted. - **Accepted residual (header trust):** `Content-Length`/`Content-Range`/416 totals come from the unverified header. A forged header can only produce a wrong total or a body that aborts with no unauthenticated bytes (test `storage::sealed::tests::forged_header_length_aborts_without_extra_bytes`), - never wrong bytes. + never wrong bytes. On S3 a range read's segment GET is pinned to the header GET's ETag + (`If-Match`): if the object is replaced or removed in between, the read starts over once, then + answers 503. It never falls back to the legacy name once a sealed header was read. - **Quota:** plaintext bytes. Sealed objects are scanned by `space_tag` and joined to a space when the object-name key is unlocked (lazily, per lookup), so the first quota check after an unlock counts legacy + sealed. While locked, a space reports legacy bytes only (writes are 503 then). diff --git a/tinycloud-node-server/src/storage/mod.rs b/tinycloud-node-server/src/storage/mod.rs index 853e4d62..ccad5473 100644 --- a/tinycloud-node-server/src/storage/mod.rs +++ b/tinycloud-node-server/src/storage/mod.rs @@ -9,18 +9,14 @@ use std::sync::Arc; use crate::{at_rest::AtRestRuntime, BlockStores}; use tinycloud_core::storage::either::Either; -/// Links the opened block store to the node's at-rest state (TC-887). -/// -/// The FileSystem store seals KV blocks when the mode and keys allow, so it -/// declares a seal path. The S3 store does not seal yet and declares none, -/// so `at_rest::setup` refuses a non-`off` mode with S3 storage rather than -/// write plaintext. +/// Links the opened block store to the node's at-rest state (TC-887). Both +/// stores seal KV blocks when the mode and keys allow, so storage declares a +/// seal path. pub fn attach_at_rest(stores: &mut BlockStores, runtime: Arc) { + runtime.declare_seal_path(); + let handle = sealed::AtRestHandle::new(runtime); match stores { - Either::B(fs) => { - runtime.declare_seal_path(); - fs.attach_at_rest(sealed::AtRestHandle::new(runtime)); - } - Either::A(_) => {} + Either::A(s3) => s3.attach_at_rest(handle), + Either::B(fs) => fs.attach_at_rest(handle), } } diff --git a/tinycloud-node-server/src/storage/s3.rs b/tinycloud-node-server/src/storage/s3.rs index 5935a343..a131868d 100644 --- a/tinycloud-node-server/src/storage/s3.rs +++ b/tinycloud-node-server/src/storage/s3.rs @@ -8,18 +8,30 @@ use aws_smithy_http::{byte_stream::Error as ByteStreamError, endpoint::Endpoint} use aws_types::sdk_config::SdkConfig; use futures::{ future::Either as AsyncEither, + io::AsyncReadExt, stream::{IntoAsyncRead, MapErr, TryStreamExt}, }; 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, time::Instant}; +use std::{ + collections::HashMap, future::Future, io::Error as IoError, ops::AddAssign, time::Instant, +}; use tinycloud_auth::resource::SpaceId; -use tinycloud_core::{at_rest::plaintext_len, hash::Hash, storage::*}; +use tinycloud_core::{ + at_rest::{plaintext_len, DataKeyProvider, HEADER_LEN}, + hash::Hash, + storage::*, +}; use super::{ file_system, - sealed::{self, scanned_object, ScannedObject}, + sealed::{ + self, contains_name, open_sealed, plan_range, read_plan, scanned_object, seal_file, + seal_vec, space_tag_of, write_target, AtRestHandle, AtRestStoreError, AtRestView, + ContainsName, LoggedDecrypt, ReadPlan, ScannedObject, SealFileError, SealedName, + WriteTarget, + }, size::SpaceSizes, }; @@ -32,6 +44,41 @@ pub struct S3BlockStore { pub client: Client, pub bucket: String, sizes: SpaceSizes, + at_rest: Option, +} + +/// An S3 object body as an `AsyncRead`. +type Body = IntoAsyncRead IoError>>; + +/// Plaintext from a legacy object, or authenticated plaintext from a sealed one. +/// The decrypting side is boxed: it carries the AES key schedule and segment +/// buffers, and `Content` values move by value through deep request paths. +pub type S3Readable = AsyncEither>>; + +fn body(stream: ByteStream) -> Body { + stream + .map_err(convert as fn(ByteStreamError) -> IoError) + .into_async_read() +} + +/// A staged block on its way into the store. +enum Staged { + File(file_system::TempFileStage), + Bytes(Vec), +} + +/// The outcome of a ranged GET. +enum RangeGet { + Missing, + InvalidRange, + /// `If-Match` failed: the object was replaced since the ETag was read. + Changed, + Got { + range: ResolvedByteRange, + total: u64, + etag: Option, + body: Body, + }, } #[serde_as] @@ -120,6 +167,7 @@ impl S3BlockStore { client, bucket: config.bucket.clone(), sizes, + at_rest: None, }) } @@ -131,11 +179,383 @@ impl S3BlockStore { ) } - async fn increment_size(&self, space: &SpaceId, size: u64) { - self.sizes.increment_size(space, None, size).await; + /// Links the store to the node's at-rest state (TC-887). Without it the + /// store behaves as at-rest mode `off`. + pub fn attach_at_rest(&mut self, handle: AtRestHandle) { + self.at_rest = Some(handle); } - async fn decrement_size(&self, space: &SpaceId, size: u64) { - self.sizes.decrement_size(space, None, size).await; + + fn view(&self) -> AtRestView { + AtRestView::of(self.at_rest.as_ref()) + } + + async fn head_exists(&self, key: &str) -> Result { + Ok(self.object_size(key).await?.is_some()) + } + + /// The whole object: (stored length, body), or `None` if absent. + async fn get_full(&self, key: &str) -> Result, S3StoreError> { + match self + .client + .get_object() + .bucket(&self.bucket) + .key(key) + .send() + .await + { + Ok(o) => Ok(Some((o.content_length().try_into()?, body(o.body)))), + Err(SdkError::ServiceError { + err: + GetObjectError { + kind: GetObjectErrorKind::NoSuchKey(_), + .. + }, + .. + }) => Ok(None), + Err(e) => Err(S3Error::from(e).into()), + } + } + + async fn get_range( + &self, + key: &str, + range: ByteRangeSpec, + if_match: Option<&str>, + ) -> Result { + match self + .client + .get_object() + .bucket(&self.bucket) + .key(key) + .range(range.to_http_value()) + .set_if_match(if_match.map(str::to_owned)) + .send() + .await + { + Ok(output) => { + let content_range = output.content_range().ok_or_else(|| { + S3StoreError::InvalidContentRange("missing header".to_string()) + })?; + let (range, total) = parse_content_range(content_range)?; + Ok(RangeGet::Got { + range, + total, + etag: output.e_tag().map(str::to_owned), + body: body(output.body), + }) + } + Err(SdkError::ServiceError { + err: + GetObjectError { + kind: GetObjectErrorKind::NoSuchKey(_), + .. + }, + .. + }) => Ok(RangeGet::Missing), + Err(SdkError::ServiceError { err, .. }) if err.code() == Some("InvalidRange") => { + Ok(RangeGet::InvalidRange) + } + Err(SdkError::ServiceError { err, .. }) if err.code() == Some("PreconditionFailed") => { + Ok(RangeGet::Changed) + } + Err(error) => Err(S3Error::from(error).into()), + } + } + + async fn put(&self, key: &str, body: ByteStream) -> Result<(), S3StoreError> { + self.client + .put_object() + .bucket(&self.bucket) + .key(key) + .body(body) + .send() + .await + .map_err(S3Error::from)?; + Ok(()) + } + + async fn persist_staged( + &self, + space: &SpaceId, + hash: &Hash, + staged: Staged, + ) -> Result<(), S3StoreError> { + match write_target(&self.view(), space, hash).await? { + WriteTarget::Legacy => { + let key = self.key(space, hash); + if self.head_exists(&key).await? { + return Ok(()); + } + let size = match staged { + Staged::File(t_file) => { + let size = t_file.size().await?; + let (_file, path) = t_file.into_inner(); + self.put(&key, ByteStream::from_path(&path).await?).await?; + size + } + Staged::Bytes(b) => { + let size = b.len() as u64; + self.put(&key, ByteStream::from(b)).await?; + size + } + }; + self.sizes.increment_size(space, None, size).await; + } + WriteTarget::Sealed { name, key } => { + // Only the sealed name dedups: a legacy copy never suppresses + // the sealed write. + let object = name.key(); + if self.head_exists(&object).await? { + return Ok(()); + } + let size = match staged { + Staged::File(t_file) => { + let size = t_file.size().await?; + let (_file, path) = t_file.into_inner(); + let sealed = seal_file(&key, space, hash, &path, size).await?; + self.put(&object, ByteStream::from_path(sealed.path()).await?) + .await?; + size + } + Staged::Bytes(b) => { + let size = b.len() as u64; + let ct = seal_vec(key, space, hash, b).await?; + self.put(&object, ByteStream::from(ct)).await?; + size + } + }; + self.sizes + .increment_size(space, Some(&name.space_tag), size) + .await; + } + } + Ok(()) + } + + async fn read_legacy( + &self, + space: &SpaceId, + id: &Hash, + ) -> Result>, S3StoreError> { + Ok(self + .get_full(&self.key(space, id)) + .await? + .map(|(len, b)| Content::new(len, AsyncEither::Left(b)))) + } + + async fn read_sealed( + &self, + name: &SealedName, + provider: &dyn DataKeyProvider, + space: &SpaceId, + id: &Hash, + ) -> Result>, S3StoreError> { + let Some((len, b)) = self.get_full(&name.key()).await? else { + return Ok(None); + }; + let (size, reader) = open_sealed(provider, space, id, b, len) + .await + .inspect_err(|e| e.log_at(name))? + .into_inner(); + Ok(Some(Content::new( + size, + AsyncEither::Right(Box::new(LoggedDecrypt::new(reader, name))), + ))) + } + + async fn read_block( + &self, + space: &SpaceId, + id: &Hash, + ) -> Result>, S3StoreError> { + match read_plan(&self.view(), space, id)? { + ReadPlan::Legacy => self.read_legacy(space, id).await, + ReadPlan::LegacyWhileLocked => match self.read_legacy(space, id).await? { + Some(c) => Ok(Some(c)), + None => Err(AtRestStoreError::Unavailable("at-rest keys are locked").into()), + }, + ReadPlan::SealedThenLegacy { name, provider } => { + match self.read_sealed(&name, &*provider, space, id).await? { + Some(c) => Ok(Some(c)), + None => self.read_legacy(space, id).await, + } + } + ReadPlan::Sealed { name, provider } => { + match self.read_sealed(&name, &*provider, space, id).await? { + Some(c) => Ok(Some(c)), + None if self.head_exists(&self.key(space, id)).await? => { + Err(AtRestStoreError::LegacyRefused.into()) + } + None => Ok(None), + } + } + } + } + + async fn range_legacy( + &self, + space: &SpaceId, + id: &Hash, + spec: ByteRangeSpec, + ) -> Result>, S3StoreError> { + let key = self.key(space, id); + match self.get_range(&key, spec, None).await? { + RangeGet::Got { + range, total, body, .. + } => Ok(Some(RangeRead::Content { + total_size: total, + range, + content: Content::new(range.len(), AsyncEither::Left(body)), + })), + RangeGet::Missing | RangeGet::Changed => Ok(None), + RangeGet::InvalidRange => { + let head = self + .client + .head_object() + .bucket(&self.bucket) + .key(&key) + .send() + .await; + match head { + Ok(output) => Ok(Some(RangeRead::Unsatisfiable { + total_size: output.content_length().try_into()?, + })), + Err(SdkError::ServiceError { + err: + HeadObjectError { + kind: HeadObjectErrorKind::NotFound(_), + .. + }, + .. + }) => Ok(None), + Err(error) => Err(S3Error::from(error).into()), + } + } + } + } + + /// A range of the block's sealed object, or `None` if it is absent: one + /// GET for the header, one (pinned to the first one's ETag) for the + /// segments. Totals and the 416 decision use the plaintext length. If the + /// object is replaced or removed between the two, the read starts over + /// once, then fails (503); once a header was read it never falls back to + /// the legacy name. + async fn range_sealed( + &self, + name: &SealedName, + provider: &dyn DataKeyProvider, + space: &SpaceId, + id: &Hash, + spec: ByteRangeSpec, + ) -> Result>, S3StoreError> { + let object = name.key(); + let logged = |e: AtRestStoreError| { + e.log_at(name); + e + }; + let header_span = ByteRangeSpec::Inclusive { + start: 0, + end: HEADER_LEN as u64 - 1, + }; + let mut seen_header = false; + for _ in 0..2 { + let (total, etag, mut head_body) = + match self.get_range(&object, header_span, None).await? { + RangeGet::Missing | RangeGet::Changed if seen_header => break, + RangeGet::Missing | RangeGet::Changed => return Ok(None), + RangeGet::InvalidRange => { + return Err(logged(AtRestStoreError::Integrity( + "object shorter than its header", + )) + .into()) + } + RangeGet::Got { + total, etag, body, .. + } => (total, etag, body), + }; + seen_header = true; + let mut header = [0u8; HEADER_LEN]; + head_body.read_exact(&mut header).await.map_err(|_| { + logged(AtRestStoreError::Integrity( + "object shorter than its header", + )) + })?; + let plan = plan_range(provider, space, &header, total, spec) + .await + .map_err(logged)?; + let Some(r) = plan.range else { + return Ok(Some(RangeRead::Unsatisfiable { + total_size: plan.plaintext_len(), + })); + }; + let span = ByteRangeSpec::Inclusive { + start: r.ciphertext_start(), + end: r.ciphertext_end(), + }; + let body = match self.get_range(&object, span, etag.as_deref()).await? { + RangeGet::Got { range, body, .. } + if range.start() == r.ciphertext_start() + && range.end() == r.ciphertext_end() => + { + body + } + RangeGet::Got { .. } => { + return Err(logged(AtRestStoreError::Integrity( + "ranged GET returned another span", + )) + .into()) + } + // Replaced, removed or shrunk since the header was read. + RangeGet::Changed | RangeGet::Missing | RangeGet::InvalidRange => continue, + }; + let reader = plan.reader(space, id, r, body).map_err(logged)?; + let range = r.plaintext_range(); + return Ok(Some(RangeRead::Content { + total_size: plan.plaintext_len(), + range, + content: Content::new( + range.len(), + AsyncEither::Right(Box::new(LoggedDecrypt::new(reader, name))), + ), + })); + } + Err(AtRestStoreError::Unavailable("the object changed during the read").into()) + } + + async fn range_block( + &self, + space: &SpaceId, + id: &Hash, + spec: ByteRangeSpec, + ) -> Result>, S3StoreError> { + match read_plan(&self.view(), space, id)? { + ReadPlan::Legacy => self.range_legacy(space, id, spec).await, + ReadPlan::LegacyWhileLocked => match self.range_legacy(space, id, spec).await? { + Some(r) => Ok(Some(r)), + None => Err(AtRestStoreError::Unavailable("at-rest keys are locked").into()), + }, + ReadPlan::SealedThenLegacy { name, provider } => { + match self + .range_sealed(&name, &*provider, space, id, spec) + .await? + { + Some(r) => Ok(Some(r)), + None => self.range_legacy(space, id, spec).await, + } + } + ReadPlan::Sealed { name, provider } => { + match self + .range_sealed(&name, &*provider, space, id, spec) + .await? + { + Some(r) => Ok(Some(r)), + None if self.head_exists(&self.key(space, id)).await? => { + Err(AtRestStoreError::LegacyRefused.into()) + } + None => Ok(None), + } + } + } } /// The stored size of an object, or `None` if absent. HeadObject, not @@ -195,31 +615,32 @@ pub enum S3StoreError { AtRest(#[from] sealed::AtRestStoreError), } +impl From for S3StoreError { + fn from(e: SealFileError) -> Self { + match e { + SealFileError::AtRest(e) => Self::AtRest(e), + SealFileError::Io(e) => Self::Io(e), + } + } +} + #[async_trait] impl ImmutableReadStore for S3BlockStore { type Error = S3StoreError; - type Readable = IntoAsyncRead IoError>>; + type Readable = S3Readable; async fn contains(&self, space: &SpaceId, id: &Hash) -> Result { - match self - .client - .head_object() - .bucket(&self.bucket) - .key(self.key(space, id)) - .send() - .await - { - Ok(_) => Ok(true), - Err(SdkError::ServiceError { - err: - HeadObjectError { - kind: HeadObjectErrorKind::NotFound(_), - .. - }, - .. - }) => Ok(false), - Err(e) => Err(S3Error::from(e).into()), + // The name a write would use (INTERFACES §7). + match contains_name(&self.view(), space, id)? { + ContainsName::Legacy => self.head_exists(&self.key(space, id)).await, + ContainsName::Sealed(name) => self.head_exists(&name.key()).await, } } + fn prepares(&self) -> bool { + super::sealed::prepares(&self.view()) + } + async fn prepare_read(&self, space: &SpaceId) -> Result<(), Self::Error> { + Ok(super::sealed::prepare_read(&self.view(), space).await?) + } async fn read( &self, @@ -227,30 +648,9 @@ impl ImmutableReadStore for S3BlockStore { id: &Hash, ) -> Result>, Self::Error> { let start = Instant::now(); - let res = self - .client - .get_object() - .bucket(&self.bucket) - .key(self.key(space, id)) - .send() - .await; - let result = match res { - Ok(o) => Ok(Some(Content::new( - o.content_length().try_into()?, - o.body - .map_err(convert as fn(ByteStreamError) -> IoError) - .into_async_read(), - ))), - Err(SdkError::ServiceError { - err: - GetObjectError { - kind: GetObjectErrorKind::NoSuchKey(_), - .. - }, - .. - }) => Ok(None), - Err(e) => Err(S3Error::from(e).into()), - }; + // Boxed: the sealed read path's future is large, and the trait method + // future is built on the caller's stack before async_trait boxes it. + let result = Box::pin(self.read_block(space, id)).await; crate::prometheus::observe_stage( crate::prometheus::InvocationStage::BlockRead, crate::prometheus::StageOutcome::from(result.is_ok()), @@ -265,67 +665,7 @@ impl ImmutableReadStore for S3BlockStore { id: &Hash, range: ByteRangeSpec, ) -> Result>, Self::Error> { - let key = self.key(space, id); - let output = self - .client - .get_object() - .bucket(&self.bucket) - .key(key) - .range(range.to_http_value()) - .send() - .await; - match output { - Ok(output) => { - let content_range = output.content_range().ok_or_else(|| { - S3StoreError::InvalidContentRange("missing header".to_string()) - })?; - let (range, total_size) = parse_content_range(content_range)?; - let content = Content::new( - range.len(), - output - .body - .map_err(convert as fn(ByteStreamError) -> IoError) - .into_async_read(), - ); - Ok(Some(RangeRead::Content { - total_size, - range, - content, - })) - } - Err(SdkError::ServiceError { - err: - GetObjectError { - kind: GetObjectErrorKind::NoSuchKey(_), - .. - }, - .. - }) => return Ok(None), - Err(SdkError::ServiceError { err, .. }) if err.code() == Some("InvalidRange") => { - let head = self - .client - .head_object() - .bucket(&self.bucket) - .key(self.key(space, id)) - .send() - .await; - match head { - Ok(output) => Ok(Some(RangeRead::Unsatisfiable { - total_size: output.content_length().try_into()?, - })), - Err(SdkError::ServiceError { - err: - HeadObjectError { - kind: HeadObjectErrorKind::NotFound(_), - .. - }, - .. - }) => Ok(None), - Err(error) => Err(S3Error::from(error).into()), - } - } - Err(error) => Err(S3Error::from(error).into()), - } + Box::pin(self.range_block(space, id, range)).await } } @@ -355,79 +695,60 @@ fn parse_content_range(value: &str) -> Result<(ResolvedByteRange, u64), S3StoreE Ok((range, total_size)) } +/// Times a block write, as before. +async fn observed_write(write: F) -> Result +where + F: Future>, +{ + let start = Instant::now(); + let result = write.await; + crate::prometheus::observe_stage( + crate::prometheus::InvocationStage::BlockWrite, + crate::prometheus::StageOutcome::from(result.is_ok()), + start.elapsed(), + ); + result +} + #[async_trait] impl ImmutableWriteStore for S3BlockStore { type Error = S3StoreError; + async fn prepare_write(&self, space: &SpaceId) -> Result<(), Self::Error> { + Ok(sealed::prepare_write(&self.view(), space).await?) + } async fn persist( &self, space: &SpaceId, staged: HashBuffer<::Writable>, ) -> Result { - let start = Instant::now(); - let result = async move { + observed_write(async move { let (mut h, f) = staged.into_inner(); let hash = h.finalize(); - - if !self.contains(space, &hash).await? { - let size = f.len() as u64; - self.client - .put_object() - .bucket(&self.bucket) - .key(self.key(space, &hash)) - .body(ByteStream::from(f)) - .send() - .await - .map_err(S3Error::from)?; - self.increment_size(space, size).await; - } + Box::pin(self.persist_staged(space, &hash, Staged::Bytes(f))).await?; Ok(hash) - } - .await; - crate::prometheus::observe_stage( - crate::prometheus::InvocationStage::BlockWrite, - crate::prometheus::StageOutcome::from(result.is_ok()), - start.elapsed(), - ); - result + }) + .await } } #[async_trait] impl ImmutableWriteStore for S3BlockStore { type Error = S3StoreError; + async fn prepare_write(&self, space: &SpaceId) -> Result<(), Self::Error> { + Ok(sealed::prepare_write(&self.view(), space).await?) + } async fn persist( &self, space: &SpaceId, staged: HashBuffer<::Writable>, ) -> Result { - let start = Instant::now(); - let result = async move { + observed_write(async move { let (mut h, f) = staged.into_inner(); let hash = h.finalize(); - - if !self.contains(space, &hash).await? { - let size = f.size().await?; - let (_file, path) = f.into_inner(); - - self.client - .put_object() - .bucket(&self.bucket) - .key(self.key(space, &hash)) - .body(ByteStream::from_path(&path).await?) - .send() - .await - .map_err(S3Error::from)?; - self.increment_size(space, size).await; - } + Box::pin(self.persist_staged(space, &hash, Staged::File(f))).await?; Ok(hash) - } - .await; - crate::prometheus::observe_stage( - crate::prometheus::InvocationStage::BlockWrite, - crate::prometheus::StageOutcome::from(result.is_ok()), - start.elapsed(), - ); - result + }) + .await } } @@ -436,54 +757,25 @@ impl ImmutableWriteStore Result<(), Self::Error> { + Ok(sealed::prepare_write(&self.view(), space).await?) + } async fn persist( &self, space: &SpaceId, staged: HashBuffer< as ImmutableStaging>::Writable>, ) -> Result { - let start = Instant::now(); - let result = async move { + observed_write(async move { let (mut h, f) = staged.into_inner(); let hash = h.finalize(); - - if !self.contains(space, &hash).await? { - match f { - AsyncEither::Left(t_file) => { - let size = t_file.size().await?; - let (_file, path) = t_file.into_inner(); - self.client - .put_object() - .bucket(&self.bucket) - .key(self.key(space, &hash)) - .body(ByteStream::from_path(&path).await?) - .send() - .await - .map_err(S3Error::from)?; - self.increment_size(space, size).await; - } - AsyncEither::Right(b) => { - let size = b.len() as u64; - self.client - .put_object() - .bucket(&self.bucket) - .key(self.key(space, &hash)) - .body(ByteStream::from(b)) - .send() - .await - .map_err(S3Error::from)?; - self.increment_size(space, size).await; - } - } + let staged = match f { + AsyncEither::Left(t_file) => Staged::File(t_file), + AsyncEither::Right(b) => Staged::Bytes(b), }; + Box::pin(self.persist_staged(space, &hash, staged)).await?; Ok(hash) - } - .await; - crate::prometheus::observe_stage( - crate::prometheus::InvocationStage::BlockWrite, - crate::prometheus::StageOutcome::from(result.is_ok()), - start.elapsed(), - ); - result + }) + .await } } @@ -491,12 +783,27 @@ impl ImmutableWriteStore Result, Self::Error> { + // A sealed copy first (if keys can name it), then the legacy object. + if let Some(nk) = self.view().object_name_key() { + let name = SealedName::new(&nk, space, id); + if let Some(size) = self.object_size(&name.key()).await? { + self.delete(&name.key()).await?; + self.sizes + .decrement_size( + space, + Some(&name.space_tag), + plaintext_len(size).unwrap_or(0), + ) + .await; + return Ok(Some(())); + } + } let key = self.key(space, id); let Some(size) = self.object_size(&key).await? else { return Ok(None); }; self.delete(&key).await?; - self.decrement_size(space, size).await; + self.sizes.decrement_size(space, None, size).await; Ok(Some(())) } } @@ -505,7 +812,11 @@ impl ImmutableDeleteStore for S3BlockStore { impl StoreSize for S3BlockStore { type Error = S3StoreError; async fn total_size(&self, space: &SpaceId) -> Result, Self::Error> { - Ok(self.sizes.get_size(space, None).await) + let view = self.view(); + Ok(self + .sizes + .get_size(space, space_tag_of(&view, space).as_deref()) + .await) } } diff --git a/tinycloud-node-server/src/storage/size.rs b/tinycloud-node-server/src/storage/size.rs index 740bb0cc..81f85ace 100644 --- a/tinycloud-node-server/src/storage/size.rs +++ b/tinycloud-node-server/src/storage/size.rs @@ -123,5 +123,10 @@ mod tests { sizes.decrement_size(&c, Some("tagC"), 40).await; assert_eq!(sizes.get_size(&c, Some("tagC")).await, Some(40)); assert_eq!(sizes.get_size(&c, None).await, Some(40)); + // A second removal of the same bytes (double remove) floors at zero + // under both entries; it never wraps. + sizes.decrement_size(&c, Some("tagC"), 40).await; + sizes.decrement_size(&c, None, 40).await; + assert_eq!(sizes.get_size(&c, Some("tagC")).await, Some(0)); } } diff --git a/tinycloud-node-server/tests/at_rest_s3_store.rs b/tinycloud-node-server/tests/at_rest_s3_store.rs new file mode 100644 index 00000000..0577d5dd --- /dev/null +++ b/tinycloud-node-server/tests/at_rest_s3_store.rs @@ -0,0 +1,468 @@ +//! TC-887 S3c: the S3 block store with encryption at rest, against a real +//! S3 API (the native MinIO of `scripts/at-rest-harness`). Ignored by +//! default; run with the harness environment loaded: +//! +//! ```sh +//! (set -a; . "$RUN_DIR/harness.env"; set +a; +//! cargo test -p tinycloud-node --test at_rest_s3_store -- --ignored) +//! ``` + +use std::sync::Arc; + +use anyhow::{ensure, Context, Result}; +use futures::io::{AsyncReadExt, AsyncWriteExt}; +use parking_lot::RwLock; +use tinycloud::{ + config::AtRestMode, + storage::{ + file_system::TempFileSystemStage, + s3::{S3BlockConfig, S3BlockStore, S3StoreError}, + sealed::{AtRestHandle, AtRestSource, AtRestStoreError, AtRestView}, + }, +}; +use tinycloud_auth::resource::SpaceId; +use tinycloud_core::{ + at_rest::{ciphertext_len, DataKeyProvider, FixedKeyProvider, SEGMENT_LEN}, + hash::Hash, + storage::{ + memory::MemoryStaging, ByteRangeSpec, HashBuffer, ImmutableDeleteStore, ImmutableReadStore, + ImmutableStaging, ImmutableWriteStore, RangeRead, StorageConfig, StorageSetup, StoreSize, + }, +}; + +const SEG: usize = SEGMENT_LEN; + +struct Switch(RwLock); + +impl AtRestSource for Switch { + fn view(&self) -> AtRestView { + self.0.read().clone() + } +} + +fn view(mode: AtRestMode, encrypting: bool, keys: Option) -> AtRestView { + AtRestView { + mode, + encrypting, + provider: keys.map(|k| Arc::new(k) as Arc), + } +} + +fn keys() -> FixedKeyProvider { + FixedKeyProvider::new([31; 32]) +} + +fn env(name: &str) -> Result { + std::env::var(name).with_context(|| format!("{name} (load the harness env)")) +} + +async fn open_store(v: AtRestView) -> Result<(S3BlockStore, Arc)> { + // 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"); + let config: S3BlockConfig = serde_json::from_value(serde_json::json!({ + "bucket": env("TC887_S3_BUCKET")?, + "endpoint": env("TC887_S3_ENDPOINT")?, + }))?; + let mut store = config.open().await?; + let switch = Arc::new(Switch(RwLock::new(v))); + store.attach_at_rest(AtRestHandle::new(switch.clone())); + Ok((store, switch)) +} + +async fn store(v: AtRestView) -> Result<(S3BlockStore, Arc, SpaceId)> { + let (store, switch) = open_store(v).await?; + let space: SpaceId = format!( + "tinycloud:key:s3store:{}", + hex::encode(rand::random::<[u8; 6]>()) + ) + .parse()?; + store.create(&space).await?; + Ok((store, switch, space)) +} + +fn value(n: usize) -> Vec { + (0..n).map(|i| (i * 13 % 251) as u8).collect() +} + +async fn put(store: &S3BlockStore, space: &SpaceId, data: &[u8]) -> Result { + ImmutableWriteStore::::prepare_write(store, space).await?; + let mut staged = HashBuffer::new(Vec::new()); + staged.write_all(data).await.unwrap(); + ImmutableWriteStore::::persist(store, space, staged).await +} + +async fn get(store: &S3BlockStore, space: &SpaceId, h: &Hash) -> Result> { + let content = store.read(space, h).await?.context("block missing")?; + let len = content.len(); + let mut out = Vec::new(); + Box::pin(content).read_to_end(&mut out).await?; + ensure!( + out.len() as u64 == len, + "Content::len is not the plaintext length" + ); + Ok(out) +} + +async fn raw(store: &S3BlockStore, key: &str) -> Result> { + Ok(store + .client + .get_object() + .bucket(&store.bucket) + .key(key) + .send() + .await? + .body + .collect() + .await? + .into_bytes() + .to_vec()) +} + +#[tokio::test] +#[ignore = "needs the at-rest harness MinIO (TC887_S3_* env)"] +async fn sealed_blocks_round_trip_and_range_over_s3() -> Result<()> { + let (store, _, space) = store(view(AtRestMode::Encrypt, true, Some(keys()))).await?; + let nk = keys().object_name_key()?; + let mut total = 0u64; + for n in [0, 17, SEG, SEG + 1, 3 * SEG + 9] { + let data = value(n); + let h = put(&store, &space, &data).await?; + total += n as u64; + // Stored under the sealed name only, as TCE1. + let sealed = tinycloud_core::at_rest::sealed_object_name(&nk, &space.to_string(), &h); + let bytes = raw(&store, &sealed).await?; + ensure!( + bytes.len() as u64 == ciphertext_len(n as u64)?, + "{n}: stored size" + ); + ensure!(bytes.starts_with(b"TCE1"), "{n}: not TCE1"); + ensure!(get(&store, &space, &h).await? == data, "{n}: read back"); + if n > 0 { + let edges = [0, 1, SEG.saturating_sub(1), SEG, SEG + 1, n - 1]; + for &a in edges.iter().filter(|&&a| a < n) { + for &b in edges.iter().filter(|&&b| b >= a && b < n) { + let spec = ByteRangeSpec::Inclusive { + start: a as u64, + end: b as u64, + }; + match store.read_range(&space, &h, spec).await?.context("range")? { + RangeRead::Content { + total_size, + range, + content, + } => { + ensure!(total_size == n as u64, "{n}: range total"); + ensure!((range.start(), range.end()) == (a as u64, b as u64)); + let mut out = Vec::new(); + Box::pin(content).read_to_end(&mut out).await?; + ensure!(out == data[a..=b], "{n}: range {a}..={b}"); + } + RangeRead::Unsatisfiable { .. } => anyhow::bail!("{n}: {a}..={b} 416"), + } + } + } + } + match store + .read_range(&space, &h, ByteRangeSpec::From { start: n as u64 }) + .await? + .context("range")? + { + RangeRead::Unsatisfiable { total_size } => { + ensure!( + total_size == n as u64, + "{n}: 416 total is the plaintext length" + ) + } + RangeRead::Content { .. } => anyhow::bail!("{n}: range past the end was served"), + } + } + ensure!( + store.total_size(&space).await? == Some(total), + "plaintext quota" + ); + // A fresh store over the same bucket counts the sealed objects' plaintext + // bytes at boot, by space tag, with no legacy objects for the space. + let (reopened, _) = open_store(view(AtRestMode::Encrypt, true, Some(keys()))).await?; + ensure!( + reopened.total_size(&space).await? == Some(total), + "boot-scan plaintext quota" + ); + Ok(()) +} + +#[tokio::test] +#[ignore = "needs the at-rest harness MinIO (TC887_S3_* env)"] +async fn legacy_locked_strict_and_tamper_over_s3() -> Result<()> { + let (store, switch, space) = store(AtRestView::off()).await?; + let legacy = value(5000); + let hl = put(&store, &space, &legacy).await?; + + // encrypt: legacy still readable; a new block is sealed. + *switch.0.write() = view(AtRestMode::Encrypt, true, Some(keys())); + ensure!( + get(&store, &space, &hl).await? == legacy, + "legacy in encrypt" + ); + let sealed = value(SEG + 7); + let hs = put(&store, &space, &sealed).await?; + + // Locked: writes and sealed reads are 503; legacy is served. + *switch.0.write() = view(AtRestMode::Encrypt, false, Some(keys().locked())); + let unavailable = |e: &S3StoreError| matches!(e, S3StoreError::AtRest(e) if e.status() == rocket::http::Status::ServiceUnavailable); + ensure!( + put(&store, &space, &value(9)) + .await + .as_ref() + .is_err_and(unavailable), + "locked write" + ); + ensure!( + store + .read(&space, &hs) + .await + .as_ref() + .is_err_and(unavailable), + "locked sealed read" + ); + ensure!( + get(&store, &space, &hl).await? == legacy, + "legacy while locked" + ); + + // strict: legacy refused, sealed served. + *switch.0.write() = view(AtRestMode::Strict, true, Some(keys())); + ensure!( + matches!( + store.read(&space, &hl).await, + Err(S3StoreError::AtRest(AtRestStoreError::LegacyRefused)) + ), + "strict refuses legacy" + ); + ensure!( + get(&store, &space, &hs).await? == sealed, + "strict reads sealed" + ); + + // A tampered sealed object fails closed. + let nk = keys().object_name_key()?; + let key = tinycloud_core::at_rest::sealed_object_name(&nk, &space.to_string(), &hs); + let mut bytes = raw(&store, &key).await?; + bytes[64 + 3] ^= 1; + store + .client + .put_object() + .bucket(&store.bucket) + .key(&key) + .body(bytes.into()) + .send() + .await?; + let content = store.read(&space, &hs).await?.context("block")?; + let mut out = Vec::new(); + let err = Box::pin(content).read_to_end(&mut out).await.unwrap_err(); + ensure!( + err.kind() == std::io::ErrorKind::InvalidData && out.is_empty(), + "tamper" + ); + + // Removing the sealed block drops it. + ensure!(store.remove(&space, &hs).await? == Some(()), "remove"); + ensure!(store.read(&space, &hs).await?.is_none(), "removed"); + Ok(()) +} + +async fn put_file(store: &S3BlockStore, space: &SpaceId, data: &[u8]) -> Result { + ImmutableWriteStore::::prepare_write(store, space).await?; + let mut staged = TempFileSystemStage.stage(space).await?; + staged.write_all(data).await?; + staged.flush().await?; + Ok(ImmutableWriteStore::::persist(store, space, staged).await?) +} + +async fn range( + store: &S3BlockStore, + space: &SpaceId, + h: &Hash, + spec: ByteRangeSpec, +) -> Result<(u64, Vec)> { + match store.read_range(space, h, spec).await?.context("range")? { + RangeRead::Content { + total_size, + content, + .. + } => { + let mut out = Vec::new(); + Box::pin(content).read_to_end(&mut out).await?; + Ok((total_size, out)) + } + RangeRead::Unsatisfiable { .. } => anyhow::bail!("{spec:?} unsatisfiable"), + } +} + +fn status(e: &S3StoreError) -> Option { + match e { + S3StoreError::AtRest(e) => Some(e.status()), + _ => None, + } +} + +/// Temp-file staging seals on S3 too; ranges across the tail and From / +/// Suffix ranges at segment edges match the plaintext. +#[tokio::test] +#[ignore = "needs the at-rest harness MinIO (TC887_S3_* env)"] +async fn temp_file_staging_seals_and_ranges_over_s3() -> Result<()> { + let (store, _, space) = store(view(AtRestMode::Encrypt, true, Some(keys()))).await?; + for n in [3 * SEG, 2 * SEG + 1, 2 * SEG - 1] { + let data = value(n); + let h = put_file(&store, &space, &data).await?; + let nk = keys().object_name_key()?; + let sealed = tinycloud_core::at_rest::sealed_object_name(&nk, &space.to_string(), &h); + ensure!( + raw(&store, &sealed).await?.starts_with(b"TCE1"), + "{n}: not sealed" + ); + ensure!(get(&store, &space, &h).await? == data, "{n}: read back"); + for at in [0, 1, SEG - 1, SEG, SEG + 1, 2 * SEG - 1, 2 * SEG, n - 1] { + if at >= n { + continue; + } + for spec in [ + ByteRangeSpec::From { start: at as u64 }, + ByteRangeSpec::Suffix { + length: (n - at) as u64, + }, + ] { + let (total, out) = range(&store, &space, &h, spec).await?; + ensure!(total == n as u64 && out == data[at..], "{n}: {spec:?}"); + } + } + } + Ok(()) +} + +/// Back to `off` with keys: sealed blocks still read; `off` with locked keys +/// answers 503 for a sealed-only block. `strict` with locked keys answers 503 +/// for legacy and sealed blocks alike. +#[tokio::test] +#[ignore = "needs the at-rest harness MinIO (TC887_S3_* env)"] +async fn off_after_encrypt_and_strict_while_locked_over_s3() -> Result<()> { + let (store, switch, space) = store(AtRestView::off()).await?; + let hl = put(&store, &space, &value(70)).await?; + *switch.0.write() = view(AtRestMode::Encrypt, true, Some(keys())); + let sealed = value(SEG + 2); + let hs = put(&store, &space, &sealed).await?; + + *switch.0.write() = view(AtRestMode::Off, false, Some(keys())); + ensure!( + get(&store, &space, &hs).await? == sealed, + "off reads sealed" + ); + *switch.0.write() = view(AtRestMode::Off, false, Some(keys().locked())); + let e = store.read(&space, &hs).await.map(|_| ()).unwrap_err(); + ensure!( + status(&e) == Some(rocket::http::Status::ServiceUnavailable), + "off + locked: {e}" + ); + + *switch.0.write() = view(AtRestMode::Strict, false, Some(keys().locked())); + for h in [hl, hs] { + let e = store.read(&space, &h).await.map(|_| ()).unwrap_err(); + ensure!( + status(&e) == Some(rocket::http::Status::ServiceUnavailable), + "strict + locked: {e}" + ); + } + Ok(()) +} + +/// `contains` checks the name a write would use (INTERFACES §7). +#[tokio::test] +#[ignore = "needs the at-rest harness MinIO (TC887_S3_* env)"] +async fn contains_checks_the_name_a_write_would_use_over_s3() -> Result<()> { + let (store, switch, space) = store(AtRestView::off()).await?; + let data = value(90); + let h = put(&store, &space, &data).await?; + ensure!(store.contains(&space, &h).await?, "legacy in off"); + *switch.0.write() = view(AtRestMode::Encrypt, true, Some(keys())); + ensure!(!store.contains(&space, &h).await?, "legacy only in encrypt"); + put(&store, &space, &data).await?; + ensure!(store.contains(&space, &h).await?, "sealed in encrypt"); + Ok(()) +} + +/// A sealed object replaced (same block, another salt) between a range +/// read's header GET and its segment GET: the segment GET is pinned to the +/// header's ETag, so the read starts over or answers 503, and never returns +/// wrong bytes or a false integrity 500. +#[tokio::test] +#[ignore = "needs the at-rest harness MinIO (TC887_S3_* env)"] +async fn a_replaced_object_never_yields_a_false_integrity_error() -> Result<()> { + let (store, _, space) = store(view(AtRestMode::Encrypt, true, Some(keys()))).await?; + let data = value(3 * SEG + 5); + let h = put(&store, &space, &data).await?; + let nk = keys().object_name_key()?; + let key = tinycloud_core::at_rest::sealed_object_name(&nk, &space.to_string(), &h); + let space_str = space.to_string(); + let identity = tinycloud_core::at_rest::BlobIdentity::KvBlock { + space: &space_str, + content_hash: &h, + }; + let dek = keys().current_key(&space_str).await?; + let versions = [ + raw(&store, &key).await?, + tinycloud_core::at_rest::seal(&dek, &identity, data.clone())?, + ]; + ensure!(versions[0] != versions[1], "two encryptions differ by salt"); + + let writer = { + let (client, bucket, key) = (store.client.clone(), store.bucket.clone(), key.clone()); + let versions = versions.clone(); + tokio::spawn(async move { + for i in 0..200usize { + client + .put_object() + .bucket(&bucket) + .key(&key) + .body(versions[i % 2].clone().into()) + .send() + .await + .expect("put"); + } + }) + }; + let (mut served, mut retried_out) = (0, 0); + while !writer.is_finished() { + let spec = ByteRangeSpec::Inclusive { + start: (SEG - 3) as u64, + end: (2 * SEG + 3) as u64, + }; + match store.read_range(&space, &h, spec).await { + Ok(Some(RangeRead::Content { content, .. })) => { + // The segment GET is pinned to one version: its body is that + // version's bytes even if the object is replaced meanwhile, + // so a body never aborts and never yields wrong bytes. + let mut out = Vec::new(); + Box::pin(content) + .read_to_end(&mut out) + .await + .context("a served range aborted mid-stream")?; + ensure!(out == data[SEG - 3..=2 * SEG + 3], "wrong bytes served"); + served += 1; + } + Ok(other) => anyhow::bail!("unexpected {:?}", other.is_some()), + Err(e) => { + ensure!( + status(&e) == Some(rocket::http::Status::ServiceUnavailable), + "a replaced object must not be an integrity error: {e}" + ); + retried_out += 1; + } + } + } + writer.await?; + println!("served {served}, gave up after a retry {retried_out}"); + ensure!(served > 0, "no range read completed"); + Ok(()) +}