review

Author: Swatinem

objectstore-server/src/config.rs

@@ -31,8 +31,8 @@ pub struct Config {
     pub http_addr: SocketAddr,
 
     // storage config
-    pub small_storage: Storage,
-    pub large_storage: Storage,
+    pub high_volume_storage: Storage,
+    pub long_term_storage: Storage,
 
     // authentication config
     pub jwt_secret: String,
@@ -46,10 +46,10 @@ impl Default for Config {
         Self {
             http_addr: "0.0.0.0:8888".parse().unwrap(),
 
-            small_storage: Storage::FileSystem {
+            high_volume_storage: Storage::FileSystem {
                 path: PathBuf::from("data"),
             },
-            large_storage: Storage::FileSystem {
+            long_term_storage: Storage::FileSystem {
                 path: PathBuf::from("data"),
             },
 
@@ -96,13 +96,13 @@ mod tests {
     #[test]
     fn configurable_via_env() {
         figment::Jail::expect_with(|jail| {
-            jail.set_env("fss_large_storage__type", "s3compatible");
-            jail.set_env("fss_large_storage__endpoint", "http://localhost:8888");
-            jail.set_env("fss_large_storage__bucket", "whatever");
+            jail.set_env("fss_long_term_storage__type", "s3compatible");
+            jail.set_env("fss_long_term_storage__endpoint", "http://localhost:8888");
+            jail.set_env("fss_long_term_storage__bucket", "whatever");
 
             let config = Config::from_args(Args::default()).unwrap();
 
-            let Storage::S3Compatible { endpoint, bucket } = dbg!(config).large_storage else {
+            let Storage::S3Compatible { endpoint, bucket } = dbg!(config).long_term_storage else {
                 panic!("expected s3 storage");
             };
             assert_eq!(endpoint.as_deref(), Some("http://localhost:8888"));
@@ -118,7 +118,7 @@ mod tests {
         tempfile
             .write_all(
                 br#"
-            large_storage:
+            long_term_storage:
                 type: s3compatible
                 endpoint: http://localhost:8888
                 bucket: whatever
@@ -131,7 +131,7 @@ mod tests {
         };
         let config = Config::from_args(args).unwrap();
 
-        let Storage::S3Compatible { endpoint, bucket } = dbg!(config).large_storage else {
+        let Storage::S3Compatible { endpoint, bucket } = dbg!(config).long_term_storage else {
             panic!("expected s3 storage");
         };
         assert_eq!(endpoint.as_deref(), Some("http://localhost:8888"));

objectstore-server/src/state.rs

@@ -13,9 +13,9 @@ pub struct State {
 
 impl State {
     pub async fn new(config: Config) -> anyhow::Result<ServiceState> {
-        let small_storage = map_storage_config(&config.small_storage);
-        let large_storage = map_storage_config(&config.large_storage);
-        let service = StorageService::new(small_storage, large_storage).await?;
+        let high_volume = map_storage_config(&config.high_volume_storage);
+        let long_term = map_storage_config(&config.long_term_storage);
+        let service = StorageService::new(high_volume, long_term).await?;
 
         Ok(Arc::new(Self { config, service }))
     }

objectstore-service/src/lib.rs

@@ -20,17 +20,19 @@ use crate::backend::{BackendStream, BoxedBackend};
 pub use backend::BigTableConfig;
 pub use metadata::*;
 
-/// The threshold up until which we will go to the small backend.
-const SMALL_THRESHOLD: usize = 50 * 1024; // 50 KiB
+/// The threshold up until which we will go to the "high volume" backend.
+const BACKEND_SIZE_THRESHOLD: usize = 1024 * 1024; // 1 MiB
+const BACKEND_HIGH_VOLUME: u8 = 1;
+const BACKEND_LONG_TERM: u8 = 2;
 
 /// High-level asynchronous service for storing and retrieving objects.
 #[derive(Clone, Debug)]
 pub struct StorageService(Arc<StorageServiceInner>);
 
 #[derive(Debug)]
 struct StorageServiceInner {
-    small_backend: BoxedBackend,
-    large_backend: BoxedBackend,
+    high_volume_backend: BoxedBackend,
+    long_term_backend: BoxedBackend,
 }
 
 /// Configuration to initialize a [`StorageService`].
@@ -55,15 +57,15 @@ pub enum StorageConfig<'a> {
 impl StorageService {
     /// Creates a new `StorageService` with the specified configuration.
     pub async fn new(
-        small_config: StorageConfig<'_>,
-        large_config: StorageConfig<'_>,
+        high_volume_config: StorageConfig<'_>,
+        long_term_config: StorageConfig<'_>,
     ) -> anyhow::Result<Self> {
-        let small_backend = create_backend(small_config).await?;
-        let large_backend = create_backend(large_config).await?;
+        let high_volume_backend = create_backend(high_volume_config).await?;
+        let long_term_backend = create_backend(long_term_config).await?;
 
         let inner = StorageServiceInner {
-            small_backend,
-            large_backend,
+            high_volume_backend,
+            long_term_backend,
         };
         Ok(Self(Arc::new(inner)))
     }
@@ -77,12 +79,12 @@ impl StorageService {
         mut stream: BackendStream,
     ) -> anyhow::Result<ScopedKey> {
         let mut first_chunk = BytesMut::new();
-        let mut backend_id = 1; // 1 = small files backend
+        let mut backend_id = BACKEND_HIGH_VOLUME;
         while let Some(chunk) = stream.try_next().await? {
             first_chunk.extend_from_slice(&chunk);
 
-            if first_chunk.len() > SMALL_THRESHOLD {
-                backend_id = 2; // 2 = large files backend
+            if first_chunk.len() > BACKEND_SIZE_THRESHOLD {
+                backend_id = BACKEND_LONG_TERM;
                 break;
             }
         }
@@ -97,10 +99,22 @@ impl StorageService {
             key,
         };
 
-        self.0
-            .small_backend
-            .put_object(&key, metadata, stream)
-            .await?;
+        match backend_id {
+            BACKEND_HIGH_VOLUME => {
+                self.0
+                    .high_volume_backend
+                    .put_object(&key, metadata, stream)
+                    .await?
+            }
+            BACKEND_LONG_TERM => {
+                self.0
+                    .long_term_backend
+                    .put_object(&key, metadata, stream)
+                    .await?
+            }
+            _ => unreachable!(),
+        }
+
         Ok(key)
     }
 
@@ -110,17 +124,17 @@ impl StorageService {
         key: &ScopedKey,
     ) -> anyhow::Result<Option<(Metadata, BackendStream)>> {
         match key.key.backend {
-            1 => self.0.small_backend.get_object(key).await,
-            2 => self.0.large_backend.get_object(key).await,
+            BACKEND_HIGH_VOLUME => self.0.high_volume_backend.get_object(key).await,
+            BACKEND_LONG_TERM => self.0.long_term_backend.get_object(key).await,
             _ => anyhow::bail!("invalid backend"),
         }
     }
 
     /// Deletes an object stored at the given key, if it exists.
     pub async fn delete_object(&self, key: &ScopedKey) -> anyhow::Result<()> {
         match key.key.backend {
-            1 => self.0.small_backend.delete_object(key).await,
-            2 => self.0.large_backend.delete_object(key).await,
+            BACKEND_HIGH_VOLUME => self.0.high_volume_backend.delete_object(key).await,
+            BACKEND_LONG_TERM => self.0.long_term_backend.delete_object(key).await,
             _ => anyhow::bail!("invalid backend"),
         }
     }