feat(TC-887): sealed KV block helpers and plaintext sizes by space tag

tinycloud-node-server/Cargo.toml

@@ -117,7 +117,8 @@ path = "../tinycloud-at-rest-escrow/"
 [dev-dependencies.tinycloud-core]
 path = "../tinycloud-core/"
 default-features = false
-features = ["sqlite", "postgres", "mysql", "tokio", "verification-count-test-hook"]
+# TC-887: `test-util` provides at_rest::FixedKeyProvider for sealed-store tests.
+features = ["sqlite", "postgres", "mysql", "tokio", "verification-count-test-hook", "test-util"]
 
 [dev-dependencies]
 # TC-887: escrow holders decrypting their shares in tests/at_rest_ceremony.rs.

tinycloud-node-server/src/routes/mod.rs

@@ -137,6 +137,36 @@ fn database_failure(
     }
 }
 
+/// TC-887: the status for an invocation whose block-store step failed for
+/// at-rest reasons (503 keys unavailable, 500 integrity/unsupported), if so.
+fn at_rest_invoke_status(
+    e: &TxStoreError<BlockStores, BlockStage, tinycloud_core::keys::StaticSecret>,
+) -> Option<Status> {
+    match e {
+        TxStoreError::StoreRead(err) | TxStoreError::StoreWrite(err) => {
+            crate::storage::sealed::block_store_at_rest_error(err).map(|e| e.status())
+        }
+        _ => None,
+    }
+}
+
+/// A KV read failure: 503/500 for at-rest causes (TC-887), 500 otherwise.
+pub(crate) fn kv_read_failure(
+    e: tinycloud_core::storage::either::EitherError<
+        DbErr,
+        <BlockStores as ImmutableReadStore>::Error,
+    >,
+) -> (Status, String) {
+    use tinycloud_core::storage::either::EitherError;
+    let status = match &e {
+        EitherError::B(err) => crate::storage::sealed::block_store_at_rest_error(err)
+            .map(|e| e.status())
+            .unwrap_or(Status::InternalServerError),
+        EitherError::A(_) => Status::InternalServerError,
+    };
+    (status, e.to_string())
+}
+
 /// `/invoke` failure: plain text for everything except storage rejections,
 /// which carry the structured TC-619 JSON body.
 #[derive(Debug)]
@@ -426,7 +456,7 @@ pub async fn signed_kv_get(
             if kv_result.is_ok() { "ok" } else { "error" },
             kv_start.elapsed(),
         );
-        return match kv_result.map_err(|e| (Status::InternalServerError, e.to_string()))? {
+        return match kv_result.map_err(kv_read_failure)? {
             Some((metadata, hash, content)) => {
                 validate_signed_kv_hash_binding(&ticket, &hash)?;
                 Ok(SignedKvReadResponse::from_range(metadata, hash, content))
@@ -441,7 +471,7 @@ pub async fn signed_kv_get(
         if kv_result.is_ok() { "ok" } else { "error" },
         kv_start.elapsed(),
     );
-    match kv_result.map_err(|e| (Status::InternalServerError, e.to_string()))? {
+    match kv_result.map_err(kv_read_failure)? {
         Some((md, hash, content)) => {
             validate_signed_kv_hash_binding(&ticket, &hash)?;
             Ok(SignedKvReadResponse::Full {
@@ -2229,6 +2259,9 @@ async fn invoke_impl(
                         database_failure(error, write_may_have_applied, e.to_string()).into()
                     );
                 }
+                if let Some(status) = at_rest_invoke_status(&e) {
+                    return Err((status, e.to_string()).into());
+                }
                 Err((
                     match &e {
                         TxStoreError::Tx(TxError::SpaceNotFound) => Status::NotFound,

tinycloud-node-server/src/routes/public.rs

@@ -238,7 +238,7 @@ pub async fn public_kv_get(
     let result = tinycloud
         .public_kv_get(&space_id, &key)
         .await
-        .map_err(|e| (Status::InternalServerError, e.to_string()))?;
+        .map_err(super::kv_read_failure)?;
 
     match result {
         Some((md, hash, content)) => {
@@ -286,7 +286,7 @@ pub async fn public_kv_head(
     let result = tinycloud
         .public_kv_get(&space_id, &key)
         .await
-        .map_err(|e| (Status::InternalServerError, e.to_string()))?;
+        .map_err(super::kv_read_failure)?;
 
     match result {
         Some((md, hash, _content)) => {

tinycloud-node-server/src/storage/file_system.rs

@@ -16,7 +16,7 @@ use std::{
 };
 use tempfile::{NamedTempFile, PathPersistError};
 use tinycloud_auth::{resource::SpaceId, ssi::dids::DIDBuf};
-use tinycloud_core::{hash::Hash, storage::*};
+use tinycloud_core::{at_rest::plaintext_len, hash::Hash, storage::*};
 use tokio::{
     fs::{create_dir_all, metadata, remove_file, File},
     io::{AsyncReadExt, AsyncSeekExt},
@@ -34,7 +34,10 @@ pub struct FileSystemStore {
 impl FileSystemStore {
     async fn new(path: PathBuf) -> Result<Self, IoError> {
         // get the size of the directory
-        let sizes = store_sizes(&path).await?.into();
+        let sizes = SpaceSizes::from_scan(
+            store_sizes(&path).await?,
+            sealed_sizes(&path.join(SEALED_DIR)).await?,
+        );
         Ok(Self { path, sizes })
     }
 
@@ -46,10 +49,10 @@ impl FileSystemStore {
     }
 
     async fn increment_size(&self, space: &SpaceId, size: u64) {
-        self.sizes.increment_size(space, size).await;
+        self.sizes.increment_size(space, None, size).await;
     }
     async fn decrement_size(&self, space: &SpaceId, size: u64) {
-        self.sizes.decrement_size(space, size).await;
+        self.sizes.decrement_size(space, None, size).await;
     }
 }
 
@@ -107,6 +110,8 @@ pub enum FileSystemStoreError {
     Io(#[from] IoError),
     #[error(transparent)]
     Persist(#[from] PathPersistError),
+    #[error(transparent)]
+    AtRest(#[from] super::sealed::AtRestStoreError),
 }
 
 #[async_trait]
@@ -168,7 +173,7 @@ impl ImmutableReadStore for FileSystemStore {
 impl StoreSize for FileSystemStore {
     type Error = FileSystemStoreError;
     async fn total_size(&self, space: &SpaceId) -> Result<Option<u64>, Self::Error> {
-        Ok(self.sizes.get_size(space).await)
+        Ok(self.sizes.get_size(space, None).await)
     }
 }
 
@@ -177,11 +182,15 @@ async fn store_sizes<P: AsRef<Path>>(path: &P) -> Result<HashMap<SpaceId, u64>,
     ReadDirStream::new(tokio::fs::read_dir(path).await?)
         // for every entry in the store dir
         .try_fold(HashMap::new(), |mut acc, entry| async move {
-            // if its a directory and the suffix is a valid string
+            // if its a directory and the suffix is a valid string (the sealed
+            // `v1` directory is scanned by `sealed_sizes`)
             if let (true, Ok(ref suffix)) = (
                 entry.metadata().await?.is_dir(),
                 entry.file_name().into_string(),
             ) {
+                if suffix == SEALED_DIR {
+                    return Ok(acc);
+                }
                 let mut ds = ReadDirStream::new(tokio::fs::read_dir(entry.path()).await?);
                 let did: DIDBuf = ["did:", suffix.as_str()].concat().parse().map_err(|_| {
                     IoError::new(ErrorKind::InvalidData, format!("Invalid DID: {suffix}"))
@@ -207,6 +216,42 @@ async fn store_sizes<P: AsRef<Path>>(path: &P) -> Result<HashMap<SpaceId, u64>,
         .await
 }
 
+/// The sealed-object directory under the store root: `v1/{space_tag}/{object_tag}`.
+const SEALED_DIR: &str = "v1";
+
+/// Plaintext bytes per space tag under `v1/` (TC-887). Missing directory: none.
+async fn sealed_sizes(path: &Path) -> Result<HashMap<String, u64>, IoError> {
+    let mut sizes = HashMap::new();
+    let mut invalid = super::sealed::InvalidLengthLog::default();
+    let mut tags = match tokio::fs::read_dir(path).await {
+        Ok(dir) => ReadDirStream::new(dir),
+        Err(e) if e.kind() == ErrorKind::NotFound => return Ok(sizes),
+        Err(e) => return Err(e),
+    };
+    while let Some(tag) = tags.try_next().await? {
+        let (true, Ok(name)) = (
+            tag.metadata().await?.is_dir(),
+            tag.file_name().into_string(),
+        ) else {
+            continue;
+        };
+        let mut total = 0u64;
+        let mut objects = ReadDirStream::new(tokio::fs::read_dir(tag.path()).await?);
+        while let Some(object) = objects.try_next().await? {
+            let meta = object.metadata().await?;
+            if meta.is_file() {
+                total += plaintext_len(meta.len()).unwrap_or_else(|| {
+                    invalid.record(meta.len());
+                    0
+                });
+            }
+        }
+        sizes.insert(name, total);
+    }
+    invalid.finish();
+    Ok(sizes)
+}
+
 async fn space_size<P: AsRef<Path>>(path: &P) -> Result<u64, IoError> {
     // get the sum size of all files in this directory (do not recurse into subdirectories)
     ReadDirStream::new(tokio::fs::read_dir(path).await?)
@@ -492,4 +537,34 @@ mod test {
             None
         );
     }
+
+    /// A store holding sealed objects under `v1/` still opens, and counts
+    /// their plaintext bytes per space tag, not under a space.
+    #[tokio::test]
+    async fn boot_scan_counts_sealed_objects_by_tag() {
+        use tinycloud_core::at_rest::ciphertext_len;
+        let dir = tempfile::tempdir().unwrap();
+        let space_id: SpaceId = "tinycloud:key:test:default".parse().unwrap();
+        let legacy = dir
+            .path()
+            .join(space_id.suffix())
+            .join(space_id.name().as_str());
+        std::fs::create_dir_all(&legacy).unwrap();
+        std::fs::write(legacy.join("obj"), [1u8; 11]).unwrap();
+        let tag = dir.path().join(SEALED_DIR).join("tagA");
+        std::fs::create_dir_all(&tag).unwrap();
+        for (name, plain) in [("o1", 100u64), ("o2", 70_000)] {
+            let len = ciphertext_len(plain).unwrap() as usize;
+            std::fs::write(tag.join(name), vec![0u8; len]).unwrap();
+        }
+        // Not a valid TCE1 length: counted as 0.
+        std::fs::write(tag.join("bad"), [0u8; 70]).unwrap();
+
+        let store = FileSystemConfig::new(dir.path()).open().await.unwrap();
+        assert_eq!(store.total_size(&space_id).await.unwrap(), Some(11));
+        assert_eq!(
+            store.sizes.get_size(&space_id, Some("tagA")).await,
+            Some(11 + 100 + 70_000)
+        );
+    }
 }

tinycloud-node-server/src/storage/mod.rs

@@ -1,4 +1,5 @@
 pub mod at_rest_mirror;
 pub mod file_system;
 pub mod s3;
+pub mod sealed;
 pub mod size;

tinycloud-node-server/src/storage/s3.rs

@@ -15,9 +15,13 @@ use serde::{Deserialize, Serialize};
 use serde_with::{serde_as, DisplayFromStr};
 use std::{collections::HashMap, io::Error as IoError, ops::AddAssign, time::Instant};
 use tinycloud_auth::resource::SpaceId;
-use tinycloud_core::{hash::Hash, storage::*};
+use tinycloud_core::{at_rest::plaintext_len, hash::Hash, storage::*};
 
-use super::{file_system, size::SpaceSizes};
+use super::{
+    file_system,
+    sealed::{self, scanned_object, ScannedObject},
+    size::SpaceSizes,
+};
 
 async fn aws_config() -> SdkConfig {
     aws_config::from_env().load().await
@@ -70,31 +74,48 @@ pub(crate) async fn new_client(config: &S3BlockConfig) -> Client {
 impl S3BlockStore {
     async fn new_(config: &S3BlockConfig) -> Result<Self, S3Error> {
         let client = new_client(config).await;
-        let sizes = client
+        let (spaces, sealed, invalid) = client
             .list_objects_v2()
             .bucket(&config.bucket)
             .into_paginator()
             .send()
-            // get the sum of all objects in each page
-            .try_fold(HashMap::new(), |mut acc, page| async move {
-                // get the sum of all objects per space in this particular page
-                for (space, obj_size) in page.contents.into_iter().flatten().filter_map(|content| {
-                    content.key().and_then(|key| {
-                        let (o, _) = key.rsplit_once('/')?;
-                        let space: SpaceId = o.parse().ok()?;
-                        if content.size() > 0 {
-                            Some((space, content.size() as u64))
-                        } else {
-                            None
+            // sum stored plaintext bytes per space (legacy names) and per
+            // space tag (sealed names) in each page
+            .try_fold(
+                (
+                    HashMap::new(),
+                    HashMap::new(),
+                    sealed::InvalidLengthLog::default(),
+                ),
+                |(mut spaces, mut sealed, mut invalid), page| async move {
+                    for content in page.contents.into_iter().flatten() {
+                        let (Some(key), size) = (content.key(), content.size().max(0) as u64)
+                        else {
+                            continue;
+                        };
+                        match scanned_object(key) {
+                            ScannedObject::Legacy(space) if size > 0 => {
+                                spaces.entry(space).or_insert(0).add_assign(size);
+                            }
+                            ScannedObject::Sealed(tag) => {
+                                let plaintext = plaintext_len(size).unwrap_or_else(|| {
+                                    invalid.record(size);
+                                    0
+                                });
+                                sealed
+                                    .entry(tag.to_owned())
+                                    .or_insert(0)
+                                    .add_assign(plaintext);
+                            }
+                            _ => {}
                         }
-                    })
-                }) {
-                    acc.entry(space).or_insert(0).add_assign(obj_size);
-                }
-                Ok(acc)
-            })
-            .await?
-            .into();
+                    }
+                    Ok((spaces, sealed, invalid))
+                },
+            )
+            .await?;
+        invalid.finish();
+        let sizes = SpaceSizes::from_scan(spaces, sealed);
         Ok(S3BlockStore {
             client,
             bucket: config.bucket.clone(),
@@ -111,10 +132,10 @@ impl S3BlockStore {
     }
 
     async fn increment_size(&self, space: &SpaceId, size: u64) {
-        self.sizes.increment_size(space, size).await;
+        self.sizes.increment_size(space, None, size).await;
     }
     async fn decrement_size(&self, space: &SpaceId, size: u64) {
-        self.sizes.decrement_size(space, size).await;
+        self.sizes.decrement_size(space, None, size).await;
     }
 
     /// The stored size of an object, or `None` if absent. HeadObject, not
@@ -170,6 +191,8 @@ pub enum S3StoreError {
     Length(#[from] std::num::TryFromIntError),
     #[error("invalid S3 Content-Range response: {0}")]
     InvalidContentRange(String),
+    #[error(transparent)]
+    AtRest(#[from] sealed::AtRestStoreError),
 }
 
 #[async_trait]
@@ -482,7 +505,7 @@ impl ImmutableDeleteStore for S3BlockStore {
 impl StoreSize for S3BlockStore {
     type Error = S3StoreError;
     async fn total_size(&self, space: &SpaceId) -> Result<Option<u64>, Self::Error> {
-        Ok(self.sizes.get_size(space).await)
+        Ok(self.sizes.get_size(space, None).await)
     }
 }