Share the remote storage as a crate

libs/remote_storage/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "remote_storage"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+anyhow = { version = "1.0", features = ["backtrace"] }
+tokio = { version = "1.17", features = ["sync", "macros", "fs", "io-util"] }
+tokio-util = { version = "0.7", features = ["io"] }
+tracing = "0.1.27"
+rusoto_core = "0.48"
+rusoto_s3 = "0.48"
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1"
+async-trait = "0.1"
+
+workspace_hack = { version = "0.1", path = "../../workspace_hack" }
+
+[dev-dependencies]
+tempfile = "3.2"

libs/remote_storage/src/lib.rs

@@ -0,0 +1,232 @@
+//! A set of generic storage abstractions for the page server to use when backing up and restoring its state from the external storage.
+//! No other modules from this tree are supposed to be used directly by the external code.
+//!
+//! [`RemoteStorage`] trait a CRUD-like generic abstraction to use for adapting external storages with a few implementations:
+//!   * [`local_fs`] allows to use local file system as an external storage
+//!   * [`s3_bucket`] uses AWS S3 bucket as an external storage
+//!
+mod local_fs;
+mod s3_bucket;
+
+use std::{
+    borrow::Cow,
+    collections::HashMap,
+    ffi::OsStr,
+    num::{NonZeroU32, NonZeroUsize},
+    path::{Path, PathBuf},
+};
+
+use anyhow::Context;
+use tokio::io;
+use tracing::info;
+
+pub use self::{
+    local_fs::LocalFs,
+    s3_bucket::{S3Bucket, S3ObjectKey},
+};
+
+/// How many different timelines can be processed simultaneously when synchronizing layers with the remote storage.
+/// During regular work, pageserver produces one layer file per timeline checkpoint, with bursts of concurrency
+/// during start (where local and remote timelines are compared and initial sync tasks are scheduled) and timeline attach.
+/// Both cases may trigger timeline download, that might download a lot of layers. This concurrency is limited by the clients internally, if needed.
+pub const DEFAULT_REMOTE_STORAGE_MAX_CONCURRENT_SYNCS: usize = 50;
+pub const DEFAULT_REMOTE_STORAGE_MAX_SYNC_ERRORS: u32 = 10;
+/// Currently, sync happens with AWS S3, that has two limits on requests per second:
+/// ~200 RPS for IAM services
+/// https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/UsingWithRDS.IAMDBAuth.html
+/// ~3500 PUT/COPY/POST/DELETE or 5500 GET/HEAD S3 requests
+/// https://aws.amazon.com/premiumsupport/knowledge-center/s3-request-limit-avoid-throttling/
+pub const DEFAULT_REMOTE_STORAGE_S3_CONCURRENCY_LIMIT: usize = 100;
+
+/// Storage (potentially remote) API to manage its state.
+/// This storage tries to be unaware of any layered repository context,
+/// providing basic CRUD operations for storage files.
+#[async_trait::async_trait]
+pub trait RemoteStorage: Send + Sync {
+    /// A way to uniquely reference a file in the remote storage.
+    type RemoteObjectId;
+
+    /// Attempts to derive the storage path out of the local path, if the latter is correct.
+    fn remote_object_id(&self, local_path: &Path) -> anyhow::Result<Self::RemoteObjectId>;
+
+    /// Gets the download path of the given storage file.
+    fn local_path(&self, remote_object_id: &Self::RemoteObjectId) -> anyhow::Result<PathBuf>;
+
+    /// Lists all items the storage has right now.
+    async fn list(&self) -> anyhow::Result<Vec<Self::RemoteObjectId>>;
+
+    /// Streams the local file contents into remote into the remote storage entry.
+    async fn upload(
+        &self,
+        from: impl io::AsyncRead + Unpin + Send + Sync + 'static,
+        // S3 PUT request requires the content length to be specified,
+        // otherwise it starts to fail with the concurrent connection count increasing.
+        from_size_bytes: usize,
+        to: &Self::RemoteObjectId,
+        metadata: Option<StorageMetadata>,
+    ) -> anyhow::Result<()>;
+
+    /// Streams the remote storage entry contents into the buffered writer given, returns the filled writer.
+    /// Returns the metadata, if any was stored with the file previously.
+    async fn download(
+        &self,
+        from: &Self::RemoteObjectId,
+        to: &mut (impl io::AsyncWrite + Unpin + Send + Sync),
+    ) -> anyhow::Result<Option<StorageMetadata>>;
+
+    /// Streams a given byte range of the remote storage entry contents into the buffered writer given, returns the filled writer.
+    /// Returns the metadata, if any was stored with the file previously.
+    async fn download_byte_range(
+        &self,
+        from: &Self::RemoteObjectId,
+        start_inclusive: u64,
+        end_exclusive: Option<u64>,
+        to: &mut (impl io::AsyncWrite + Unpin + Send + Sync),
+    ) -> anyhow::Result<Option<StorageMetadata>>;
+
+    async fn delete(&self, path: &Self::RemoteObjectId) -> anyhow::Result<()>;
+}
+
+/// TODO kb
+pub enum GenericRemoteStorage {
+    Local(LocalFs),
+    S3(S3Bucket),
+}
+
+impl GenericRemoteStorage {
+    pub fn new(
+        working_directory: PathBuf,
+        storage_config: &RemoteStorageConfig,
+    ) -> anyhow::Result<Self> {
+        match &storage_config.storage {
+            RemoteStorageKind::LocalFs(root) => {
+                info!("Using fs root '{}' as a remote storage", root.display());
+                LocalFs::new(root.clone(), working_directory).map(GenericRemoteStorage::Local)
+            }
+            RemoteStorageKind::AwsS3(s3_config) => {
+                info!("Using s3 bucket '{}' in region '{}' as a remote storage, prefix in bucket: '{:?}', bucket endpoint: '{:?}'",
+                    s3_config.bucket_name, s3_config.bucket_region, s3_config.prefix_in_bucket, s3_config.endpoint);
+                S3Bucket::new(s3_config, working_directory).map(GenericRemoteStorage::S3)
+            }
+        }
+    }
+}
+
+/// Extra set of key-value pairs that contain arbitrary metadata about the storage entry.
+/// Immutable, cannot be changed once the file is created.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct StorageMetadata(HashMap<String, String>);
+
+fn strip_path_prefix<'a>(prefix: &'a Path, path: &'a Path) -> anyhow::Result<&'a Path> {
+    if prefix == path {
+        anyhow::bail!(
+            "Prefix and the path are equal, cannot strip: '{}'",
+            prefix.display()
+        )
+    } else {
+        path.strip_prefix(prefix).with_context(|| {
+            format!(
+                "Path '{}' is not prefixed with '{}'",
+                path.display(),
+                prefix.display(),
+            )
+        })
+    }
+}
+
+/// External backup storage configuration, enough for creating a client for that storage.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RemoteStorageConfig {
+    /// Max allowed number of concurrent sync operations between the API user and the remote storage.
+    pub max_concurrent_syncs: NonZeroUsize,
+    /// Max allowed errors before the sync task is considered failed and evicted.
+    pub max_sync_errors: NonZeroU32,
+    /// The storage connection configuration.
+    pub storage: RemoteStorageKind,
+}
+
+/// A kind of a remote storage to connect to, with its connection configuration.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum RemoteStorageKind {
+    /// Storage based on local file system.
+    /// Specify a root folder to place all stored files into.
+    LocalFs(PathBuf),
+    /// AWS S3 based storage, storing all files in the S3 bucket
+    /// specified by the config
+    AwsS3(S3Config),
+}
+
+/// AWS S3 bucket coordinates and access credentials to manage the bucket contents (read and write).
+#[derive(Clone, PartialEq, Eq)]
+pub struct S3Config {
+    /// Name of the bucket to connect to.
+    pub bucket_name: String,
+    /// The region where the bucket is located at.
+    pub bucket_region: String,
+    /// A "subfolder" in the bucket, to use the same bucket separately by multiple remote storage users at once.
+    pub prefix_in_bucket: Option<String>,
+    /// A base URL to send S3 requests to.
+    /// By default, the endpoint is derived from a region name, assuming it's
+    /// an AWS S3 region name, erroring on wrong region name.
+    /// Endpoint provides a way to support other S3 flavors and their regions.
+    ///
+    /// Example: `http://127.0.0.1:5000`
+    pub endpoint: Option<String>,
+    /// AWS S3 has various limits on its API calls, we need not to exceed those.
+    /// See [`DEFAULT_REMOTE_STORAGE_S3_CONCURRENCY_LIMIT`] for more details.
+    pub concurrency_limit: NonZeroUsize,
+}
+
+impl std::fmt::Debug for S3Config {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("S3Config")
+            .field("bucket_name", &self.bucket_name)
+            .field("bucket_region", &self.bucket_region)
+            .field("prefix_in_bucket", &self.prefix_in_bucket)
+            .field("concurrency_limit", &self.concurrency_limit)
+            .finish()
+    }
+}
+
+pub fn path_with_suffix_extension(original_path: impl AsRef<Path>, suffix: &str) -> PathBuf {
+    let new_extension = match original_path
+        .as_ref()
+        .extension()
+        .map(OsStr::to_string_lossy)
+    {
+        Some(extension) => Cow::Owned(format!("{extension}.{suffix}")),
+        None => Cow::Borrowed(suffix),
+    };
+    original_path
+        .as_ref()
+        .with_extension(new_extension.as_ref())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_path_with_suffix_extension() {
+        let p = PathBuf::from("/foo/bar");
+        assert_eq!(
+            &path_with_suffix_extension(&p, "temp").to_string_lossy(),
+            "/foo/bar.temp"
+        );
+        let p = PathBuf::from("/foo/bar");
+        assert_eq!(
+            &path_with_suffix_extension(&p, "temp.temp").to_string_lossy(),
+            "/foo/bar.temp.temp"
+        );
+        let p = PathBuf::from("/foo/bar.baz");
+        assert_eq!(
+            &path_with_suffix_extension(&p, "temp.temp").to_string_lossy(),
+            "/foo/bar.baz.temp.temp"
+        );
+        let p = PathBuf::from("/foo/bar.baz");
+        assert_eq!(
+            &path_with_suffix_extension(&p, ".temp").to_string_lossy(),
+            "/foo/bar.baz..temp"
+        );
+    }
+}

libs/remote_storage/src/local_fs.rs

@@ -0,0 +1,854 @@
+//! Local filesystem acting as a remote storage.
+//! Multiple API users can use the same "storage" of this kind by using different storage roots.
+//!
+//! This storage used in tests, but can also be used in cases when a certain persistent
+//! volume is mounted to the local FS.
+
+use std::{
+    future::Future,
+    path::{Path, PathBuf},
+    pin::Pin,
+};
+
+use anyhow::{bail, ensure, Context};
+use tokio::{
+    fs,
+    io::{self, AsyncReadExt, AsyncSeekExt, AsyncWriteExt},
+};
+use tracing::*;
+
+use crate::path_with_suffix_extension;
+
+use super::{strip_path_prefix, RemoteStorage, StorageMetadata};
+
+pub struct LocalFs {
+    working_directory: PathBuf,
+    storage_root: PathBuf,
+}
+
+impl LocalFs {
+    /// Attempts to create local FS storage, along with its root directory.
+    pub fn new(root: PathBuf, working_directory: PathBuf) -> anyhow::Result<Self> {
+        if !root.exists() {
+            std::fs::create_dir_all(&root).with_context(|| {
+                format!(
+                    "Failed to create all directories in the given root path '{}'",
+                    root.display(),
+                )
+            })?;
+        }
+        Ok(Self {
+            working_directory,
+            storage_root: root,
+        })
+    }
+
+    fn resolve_in_storage(&self, path: &Path) -> anyhow::Result<PathBuf> {
+        if path.is_relative() {
+            Ok(self.storage_root.join(path))
+        } else if path.starts_with(&self.storage_root) {
+            Ok(path.to_path_buf())
+        } else {
+            bail!(
+                "Path '{}' does not belong to the current storage",
+                path.display()
+            )
+        }
+    }
+
+    async fn read_storage_metadata(
+        &self,
+        file_path: &Path,
+    ) -> anyhow::Result<Option<StorageMetadata>> {
+        let metadata_path = storage_metadata_path(file_path);
+        if metadata_path.exists() && metadata_path.is_file() {
+            let metadata_string = fs::read_to_string(&metadata_path).await.with_context(|| {
+                format!(
+                    "Failed to read metadata from the local storage at '{}'",
+                    metadata_path.display()
+                )
+            })?;
+
+            serde_json::from_str(&metadata_string)
+                .with_context(|| {
+                    format!(
+                        "Failed to deserialize metadata from the local storage at '{}'",
+                        metadata_path.display()
+                    )
+                })
+                .map(|metadata| Some(StorageMetadata(metadata)))
+        } else {
+            Ok(None)
+        }
+    }
+}
+
+#[async_trait::async_trait]
+impl RemoteStorage for LocalFs {
+    type RemoteObjectId = PathBuf;
+
+    fn remote_object_id(&self, local_path: &Path) -> anyhow::Result<Self::RemoteObjectId> {
+        Ok(self.storage_root.join(
+            strip_path_prefix(&self.working_directory, local_path)
+                .context("local path does not belong to this storage")?,
+        ))
+    }
+
+    fn local_path(&self, storage_path: &Self::RemoteObjectId) -> anyhow::Result<PathBuf> {
+        let relative_path = strip_path_prefix(&self.storage_root, storage_path)
+            .context("local path does not belong to this storage")?;
+        Ok(self.working_directory.join(relative_path))
+    }
+
+    async fn list(&self) -> anyhow::Result<Vec<Self::RemoteObjectId>> {
+        get_all_files(&self.storage_root).await
+    }
+
+    async fn upload(
+        &self,
+        from: impl io::AsyncRead + Unpin + Send + Sync + 'static,
+        from_size_bytes: usize,
+        to: &Self::RemoteObjectId,
+        metadata: Option<StorageMetadata>,
+    ) -> anyhow::Result<()> {
+        let target_file_path = self.resolve_in_storage(to)?;
+        create_target_directory(&target_file_path).await?;
+        // We need this dance with sort of durable rename (without fsyncs)
+        // to prevent partial uploads. This was really hit when pageserver shutdown
+        // cancelled the upload and partial file was left on the fs
+        let temp_file_path = path_with_suffix_extension(&target_file_path, "temp");
+        let mut destination = io::BufWriter::new(
+            fs::OpenOptions::new()
+                .write(true)
+                .create(true)
+                .open(&temp_file_path)
+                .await
+                .with_context(|| {
+                    format!(
+                        "Failed to open target fs destination at '{}'",
+                        target_file_path.display()
+                    )
+                })?,
+        );
+
+        let from_size_bytes = from_size_bytes as u64;
+        // Require to read 1 byte more than the expected to check later, that the stream and its size match.
+        let mut buffer_to_read = from.take(from_size_bytes + 1);
+
+        let bytes_read = io::copy(&mut buffer_to_read, &mut destination)
+            .await
+            .with_context(|| {
+                format!(
+                    "Failed to upload file (write temp) to the local storage at '{}'",
+                    temp_file_path.display()
+                )
+            })?;
+
+        ensure!(
+            bytes_read == from_size_bytes,
+            "Provided stream has actual size {} fthat is smaller than the given stream size {}",
+            bytes_read,
+            from_size_bytes
+        );
+
+        ensure!(
+            buffer_to_read.read(&mut [0]).await? == 0,
+            "Provided stream has bigger size than the given stream size {}",
+            from_size_bytes
+        );
+
+        destination.flush().await.with_context(|| {
+            format!(
+                "Failed to upload (flush temp) file to the local storage at '{}'",
+                temp_file_path.display()
+            )
+        })?;
+
+        fs::rename(temp_file_path, &target_file_path)
+            .await
+            .with_context(|| {
+                format!(
+                    "Failed to upload (rename) file to the local storage at '{}'",
+                    target_file_path.display()
+                )
+            })?;
+
+        if let Some(storage_metadata) = metadata {
+            let storage_metadata_path = storage_metadata_path(&target_file_path);
+            fs::write(
+                &storage_metadata_path,
+                serde_json::to_string(&storage_metadata.0)
+                    .context("Failed to serialize storage metadata as json")?,
+            )
+            .await
+            .with_context(|| {
+                format!(
+                    "Failed to write metadata to the local storage at '{}'",
+                    storage_metadata_path.display()
+                )
+            })?;
+        }
+
+        Ok(())
+    }
+
+    async fn download(
+        &self,
+        from: &Self::RemoteObjectId,
+        to: &mut (impl io::AsyncWrite + Unpin + Send + Sync),
+    ) -> anyhow::Result<Option<StorageMetadata>> {
+        let file_path = self.resolve_in_storage(from)?;
+
+        if file_path.exists() && file_path.is_file() {
+            let mut source = io::BufReader::new(
+                fs::OpenOptions::new()
+                    .read(true)
+                    .open(&file_path)
+                    .await
+                    .with_context(|| {
+                        format!(
+                            "Failed to open source file '{}' to use in the download",
+                            file_path.display()
+                        )
+                    })?,
+            );
+            io::copy(&mut source, to).await.with_context(|| {
+                format!(
+                    "Failed to download file '{}' from the local storage",
+                    file_path.display()
+                )
+            })?;
+            source.flush().await?;
+
+            self.read_storage_metadata(&file_path).await
+        } else {
+            bail!(
+                "File '{}' either does not exist or is not a file",
+                file_path.display()
+            )
+        }
+    }
+
+    async fn download_byte_range(
+        &self,
+        from: &Self::RemoteObjectId,
+        start_inclusive: u64,
+        end_exclusive: Option<u64>,
+        to: &mut (impl io::AsyncWrite + Unpin + Send + Sync),
+    ) -> anyhow::Result<Option<StorageMetadata>> {
+        if let Some(end_exclusive) = end_exclusive {
+            ensure!(
+                end_exclusive > start_inclusive,
+                "Invalid range, start ({}) is bigger then end ({:?})",
+                start_inclusive,
+                end_exclusive
+            );
+            if start_inclusive == end_exclusive.saturating_sub(1) {
+                return Ok(None);
+            }
+        }
+        let file_path = self.resolve_in_storage(from)?;
+
+        if file_path.exists() && file_path.is_file() {
+            let mut source = io::BufReader::new(
+                fs::OpenOptions::new()
+                    .read(true)
+                    .open(&file_path)
+                    .await
+                    .with_context(|| {
+                        format!(
+                            "Failed to open source file '{}' to use in the download",
+                            file_path.display()
+                        )
+                    })?,
+            );
+            source
+                .seek(io::SeekFrom::Start(start_inclusive))
+                .await
+                .context("Failed to seek to the range start in a local storage file")?;
+            match end_exclusive {
+                Some(end_exclusive) => {
+                    io::copy(&mut source.take(end_exclusive - start_inclusive), to).await
+                }
+                None => io::copy(&mut source, to).await,
+            }
+            .with_context(|| {
+                format!(
+                    "Failed to download file '{}' range from the local storage",
+                    file_path.display()
+                )
+            })?;
+
+            self.read_storage_metadata(&file_path).await
+        } else {
+            bail!(
+                "File '{}' either does not exist or is not a file",
+                file_path.display()
+            )
+        }
+    }
+
+    async fn delete(&self, path: &Self::RemoteObjectId) -> anyhow::Result<()> {
+        let file_path = self.resolve_in_storage(path)?;
+        if file_path.exists() && file_path.is_file() {
+            Ok(fs::remove_file(file_path).await?)
+        } else {
+            bail!(
+                "File '{}' either does not exist or is not a file",
+                file_path.display()
+            )
+        }
+    }
+}
+
+fn storage_metadata_path(original_path: &Path) -> PathBuf {
+    path_with_suffix_extension(original_path, "metadata")
+}
+
+fn get_all_files<'a, P>(
+    directory_path: P,
+) -> Pin<Box<dyn Future<Output = anyhow::Result<Vec<PathBuf>>> + Send + Sync + 'a>>
+where
+    P: AsRef<Path> + Send + Sync + 'a,
+{
+    Box::pin(async move {
+        let directory_path = directory_path.as_ref();
+        if directory_path.exists() {
+            if directory_path.is_dir() {
+                let mut paths = Vec::new();
+                let mut dir_contents = fs::read_dir(directory_path).await?;
+                while let Some(dir_entry) = dir_contents.next_entry().await? {
+                    let file_type = dir_entry.file_type().await?;
+                    let entry_path = dir_entry.path();
+                    if file_type.is_symlink() {
+                        debug!("{:?} us a symlink, skipping", entry_path)
+                    } else if file_type.is_dir() {
+                        paths.extend(get_all_files(entry_path).await?.into_iter())
+                    } else {
+                        paths.push(dir_entry.path());
+                    }
+                }
+                Ok(paths)
+            } else {
+                bail!("Path '{}' is not a directory", directory_path.display())
+            }
+        } else {
+            Ok(Vec::new())
+        }
+    })
+}
+
+async fn create_target_directory(target_file_path: &Path) -> anyhow::Result<()> {
+    let target_dir = match target_file_path.parent() {
+        Some(parent_dir) => parent_dir,
+        None => bail!(
+            "File path '{}' has no parent directory",
+            target_file_path.display()
+        ),
+    };
+    if !target_dir.exists() {
+        fs::create_dir_all(target_dir).await?;
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod pure_tests {
+    use tempfile::tempdir;
+
+    use super::*;
+
+    #[test]
+    fn storage_path_positive() -> anyhow::Result<()> {
+        let workdir = tempdir()?.path().to_owned();
+
+        let storage_root = PathBuf::from("somewhere").join("else");
+        let storage = LocalFs {
+            working_directory: workdir.clone(),
+            storage_root: storage_root.clone(),
+        };
+
+        let local_path = workdir
+            .join("timelines")
+            .join("some_timeline")
+            .join("file_name");
+        let expected_path = storage_root.join(local_path.strip_prefix(&workdir)?);
+
+        assert_eq!(
+            expected_path,
+            storage.remote_object_id(&local_path).expect("Matching path should map to storage path normally"),
+            "File paths from workdir should be stored in local fs storage with the same path they have relative to the workdir"
+        );
+
+        Ok(())
+    }
+
+    #[test]
+    fn storage_path_negatives() -> anyhow::Result<()> {
+        #[track_caller]
+        fn storage_path_error(storage: &LocalFs, mismatching_path: &Path) -> String {
+            match storage.remote_object_id(mismatching_path) {
+                Ok(wrong_path) => panic!(
+                    "Expected path '{}' to error, but got storage path: {:?}",
+                    mismatching_path.display(),
+                    wrong_path,
+                ),
+                Err(e) => format!("{:?}", e),
+            }
+        }
+
+        let workdir = tempdir()?.path().to_owned();
+        let storage_root = PathBuf::from("somewhere").join("else");
+        let storage = LocalFs {
+            working_directory: workdir.clone(),
+            storage_root,
+        };
+
+        let error_string = storage_path_error(&storage, &workdir);
+        assert!(error_string.contains("does not belong to this storage"));
+        assert!(error_string.contains(workdir.to_str().unwrap()));
+
+        let mismatching_path_str = "/something/else";
+        let error_message = storage_path_error(&storage, Path::new(mismatching_path_str));
+        assert!(
+            error_message.contains(mismatching_path_str),
+            "Error should mention wrong path"
+        );
+        assert!(
+            error_message.contains(workdir.to_str().unwrap()),
+            "Error should mention server workdir"
+        );
+        assert!(error_message.contains("does not belong to this storage"));
+
+        Ok(())
+    }
+
+    #[test]
+    fn local_path_positive() -> anyhow::Result<()> {
+        let workdir = tempdir()?.path().to_owned();
+        let storage_root = PathBuf::from("somewhere").join("else");
+        let storage = LocalFs {
+            working_directory: workdir.clone(),
+            storage_root: storage_root.clone(),
+        };
+
+        let name = "not a metadata";
+        let local_path = workdir.join("timelines").join("some_timeline").join(name);
+        assert_eq!(
+            local_path,
+            storage
+                .local_path(&storage_root.join(local_path.strip_prefix(&workdir)?))
+                .expect("For a valid input, valid local path should be parsed"),
+            "Should be able to parse metadata out of the correctly named remote delta file"
+        );
+
+        let local_metadata_path = workdir
+            .join("timelines")
+            .join("some_timeline")
+            .join("metadata");
+        let remote_metadata_path = storage.remote_object_id(&local_metadata_path)?;
+        assert_eq!(
+            local_metadata_path,
+            storage
+                .local_path(&remote_metadata_path)
+                .expect("For a valid input, valid local path should be parsed"),
+            "Should be able to parse metadata out of the correctly named remote metadata file"
+        );
+
+        Ok(())
+    }
+
+    #[test]
+    fn local_path_negatives() -> anyhow::Result<()> {
+        #[track_caller]
+        #[allow(clippy::ptr_arg)] // have to use &PathBuf due to `storage.local_path` parameter requirements
+        fn local_path_error(storage: &LocalFs, storage_path: &PathBuf) -> String {
+            match storage.local_path(storage_path) {
+                Ok(wrong_path) => panic!(
+                    "Expected local path input {:?} to cause an error, but got file path: {:?}",
+                    storage_path, wrong_path,
+                ),
+                Err(e) => format!("{:?}", e),
+            }
+        }
+
+        let storage_root = PathBuf::from("somewhere").join("else");
+        let storage = LocalFs {
+            working_directory: tempdir()?.path().to_owned(),
+            storage_root,
+        };
+
+        let totally_wrong_path = "wrong_wrong_wrong";
+        let error_message = local_path_error(&storage, &PathBuf::from(totally_wrong_path));
+        assert!(error_message.contains(totally_wrong_path));
+
+        Ok(())
+    }
+
+    #[test]
+    fn download_destination_matches_original_path() -> anyhow::Result<()> {
+        let workdir = tempdir()?.path().to_owned();
+        let original_path = workdir
+            .join("timelines")
+            .join("some_timeline")
+            .join("some name");
+
+        let storage_root = PathBuf::from("somewhere").join("else");
+        let dummy_storage = LocalFs {
+            working_directory: workdir,
+            storage_root,
+        };
+
+        let storage_path = dummy_storage.remote_object_id(&original_path)?;
+        let download_destination = dummy_storage.local_path(&storage_path)?;
+
+        assert_eq!(
+            original_path, download_destination,
+            "'original path -> storage path -> matching fs path' transformation should produce the same path as the input one for the correct path"
+        );
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod fs_tests {
+    use super::*;
+
+    use std::{collections::HashMap, io::Write};
+    use tempfile::tempdir;
+
+    #[tokio::test]
+    async fn upload_file() -> anyhow::Result<()> {
+        let workdir = tempdir()?.path().to_owned();
+        let storage = create_storage()?;
+
+        let (file, size) = create_file_for_upload(
+            &storage.working_directory.join("whatever"),
+            "whatever_contents",
+        )
+        .await?;
+        let target_path = PathBuf::from("/").join("somewhere").join("else");
+        match storage.upload(file, size, &target_path, None).await {
+            Ok(()) => panic!("Should not allow storing files with wrong target path"),
+            Err(e) => {
+                let message = format!("{:?}", e);
+                assert!(message.contains(&target_path.display().to_string()));
+                assert!(message.contains("does not belong to the current storage"));
+            }
+        }
+        assert!(storage.list().await?.is_empty());
+
+        let target_path_1 = upload_dummy_file(&workdir, &storage, "upload_1", None).await?;
+        assert_eq!(
+            storage.list().await?,
+            vec![target_path_1.clone()],
+            "Should list a single file after first upload"
+        );
+
+        let target_path_2 = upload_dummy_file(&workdir, &storage, "upload_2", None).await?;
+        assert_eq!(
+            list_files_sorted(&storage).await?,
+            vec![target_path_1.clone(), target_path_2.clone()],
+            "Should list a two different files after second upload"
+        );
+
+        Ok(())
+    }
+
+    fn create_storage() -> anyhow::Result<LocalFs> {
+        LocalFs::new(tempdir()?.path().to_owned(), tempdir()?.path().to_owned())
+    }
+
+    #[tokio::test]
+    async fn download_file() -> anyhow::Result<()> {
+        let workdir = tempdir()?.path().to_owned();
+
+        let storage = create_storage()?;
+        let upload_name = "upload_1";
+        let upload_target = upload_dummy_file(&workdir, &storage, upload_name, None).await?;
+
+        let mut content_bytes = io::BufWriter::new(std::io::Cursor::new(Vec::new()));
+        let metadata = storage.download(&upload_target, &mut content_bytes).await?;
+        assert!(
+            metadata.is_none(),
+            "No metadata should be returned for no metadata upload"
+        );
+
+        content_bytes.flush().await?;
+        let contents = String::from_utf8(content_bytes.into_inner().into_inner())?;
+        assert_eq!(
+            dummy_contents(upload_name),
+            contents,
+            "We should upload and download the same contents"
+        );
+
+        let non_existing_path = PathBuf::from("somewhere").join("else");
+        match storage.download(&non_existing_path, &mut io::sink()).await {
+            Ok(_) => panic!("Should not allow downloading non-existing storage files"),
+            Err(e) => {
+                let error_string = e.to_string();
+                assert!(error_string.contains("does not exist"));
+                assert!(error_string.contains(&non_existing_path.display().to_string()));
+            }
+        }
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn download_file_range_positive() -> anyhow::Result<()> {
+        let workdir = tempdir()?.path().to_owned();
+
+        let storage = create_storage()?;
+        let upload_name = "upload_1";
+        let upload_target = upload_dummy_file(&workdir, &storage, upload_name, None).await?;
+
+        let mut full_range_bytes = io::BufWriter::new(std::io::Cursor::new(Vec::new()));
+        let metadata = storage
+            .download_byte_range(&upload_target, 0, None, &mut full_range_bytes)
+            .await?;
+        assert!(
+            metadata.is_none(),
+            "No metadata should be returned for no metadata upload"
+        );
+        full_range_bytes.flush().await?;
+        assert_eq!(
+            dummy_contents(upload_name),
+            String::from_utf8(full_range_bytes.into_inner().into_inner())?,
+            "Download full range should return the whole upload"
+        );
+
+        let mut zero_range_bytes = io::BufWriter::new(std::io::Cursor::new(Vec::new()));
+        let same_byte = 1_000_000_000;
+        let metadata = storage
+            .download_byte_range(
+                &upload_target,
+                same_byte,
+                Some(same_byte + 1), // exclusive end
+                &mut zero_range_bytes,
+            )
+            .await?;
+        assert!(
+            metadata.is_none(),
+            "No metadata should be returned for no metadata upload"
+        );
+        zero_range_bytes.flush().await?;
+        assert!(
+            zero_range_bytes.into_inner().into_inner().is_empty(),
+            "Zero byte range should not download any part of the file"
+        );
+
+        let uploaded_bytes = dummy_contents(upload_name).into_bytes();
+        let (first_part_local, second_part_local) = uploaded_bytes.split_at(3);
+
+        let mut first_part_remote = io::BufWriter::new(std::io::Cursor::new(Vec::new()));
+        let metadata = storage
+            .download_byte_range(
+                &upload_target,
+                0,
+                Some(first_part_local.len() as u64),
+                &mut first_part_remote,
+            )
+            .await?;
+        assert!(
+            metadata.is_none(),
+            "No metadata should be returned for no metadata upload"
+        );
+
+        first_part_remote.flush().await?;
+        let first_part_remote = first_part_remote.into_inner().into_inner();
+        assert_eq!(
+            first_part_local,
+            first_part_remote.as_slice(),
+            "First part bytes should be returned when requested"
+        );
+
+        let mut second_part_remote = io::BufWriter::new(std::io::Cursor::new(Vec::new()));
+        let metadata = storage
+            .download_byte_range(
+                &upload_target,
+                first_part_local.len() as u64,
+                Some((first_part_local.len() + second_part_local.len()) as u64),
+                &mut second_part_remote,
+            )
+            .await?;
+        assert!(
+            metadata.is_none(),
+            "No metadata should be returned for no metadata upload"
+        );
+
+        second_part_remote.flush().await?;
+        let second_part_remote = second_part_remote.into_inner().into_inner();
+        assert_eq!(
+            second_part_local,
+            second_part_remote.as_slice(),
+            "Second part bytes should be returned when requested"
+        );
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn download_file_range_negative() -> anyhow::Result<()> {
+        let workdir = tempdir()?.path().to_owned();
+
+        let storage = create_storage()?;
+        let upload_name = "upload_1";
+        let upload_target = upload_dummy_file(&workdir, &storage, upload_name, None).await?;
+
+        let start = 10000;
+        let end = 234;
+        assert!(start > end, "Should test an incorrect range");
+        match storage
+            .download_byte_range(&upload_target, start, Some(end), &mut io::sink())
+            .await
+        {
+            Ok(_) => panic!("Should not allow downloading wrong ranges"),
+            Err(e) => {
+                let error_string = e.to_string();
+                assert!(error_string.contains("Invalid range"));
+                assert!(error_string.contains(&start.to_string()));
+                assert!(error_string.contains(&end.to_string()));
+            }
+        }
+
+        let non_existing_path = PathBuf::from("somewhere").join("else");
+        match storage
+            .download_byte_range(&non_existing_path, 1, Some(3), &mut io::sink())
+            .await
+        {
+            Ok(_) => panic!("Should not allow downloading non-existing storage file ranges"),
+            Err(e) => {
+                let error_string = e.to_string();
+                assert!(error_string.contains("does not exist"));
+                assert!(error_string.contains(&non_existing_path.display().to_string()));
+            }
+        }
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn delete_file() -> anyhow::Result<()> {
+        let workdir = tempdir()?.path().to_owned();
+
+        let storage = create_storage()?;
+        let upload_name = "upload_1";
+        let upload_target = upload_dummy_file(&workdir, &storage, upload_name, None).await?;
+
+        storage.delete(&upload_target).await?;
+        assert!(storage.list().await?.is_empty());
+
+        match storage.delete(&upload_target).await {
+            Ok(()) => panic!("Should not allow deleting non-existing storage files"),
+            Err(e) => {
+                let error_string = e.to_string();
+                assert!(error_string.contains("does not exist"));
+                assert!(error_string.contains(&upload_target.display().to_string()));
+            }
+        }
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn file_with_metadata() -> anyhow::Result<()> {
+        let workdir = tempdir()?.path().to_owned();
+
+        let storage = create_storage()?;
+        let upload_name = "upload_1";
+        let metadata = StorageMetadata(HashMap::from([
+            ("one".to_string(), "1".to_string()),
+            ("two".to_string(), "2".to_string()),
+        ]));
+        let upload_target =
+            upload_dummy_file(&workdir, &storage, upload_name, Some(metadata.clone())).await?;
+
+        let mut content_bytes = io::BufWriter::new(std::io::Cursor::new(Vec::new()));
+        let full_download_metadata = storage.download(&upload_target, &mut content_bytes).await?;
+
+        content_bytes.flush().await?;
+        let contents = String::from_utf8(content_bytes.into_inner().into_inner())?;
+        assert_eq!(
+            dummy_contents(upload_name),
+            contents,
+            "We should upload and download the same contents"
+        );
+
+        assert_eq!(
+            full_download_metadata.as_ref(),
+            Some(&metadata),
+            "We should get the same metadata back for full download"
+        );
+
+        let uploaded_bytes = dummy_contents(upload_name).into_bytes();
+        let (first_part_local, _) = uploaded_bytes.split_at(3);
+
+        let mut first_part_remote = io::BufWriter::new(std::io::Cursor::new(Vec::new()));
+        let partial_download_metadata = storage
+            .download_byte_range(
+                &upload_target,
+                0,
+                Some(first_part_local.len() as u64),
+                &mut first_part_remote,
+            )
+            .await?;
+        first_part_remote.flush().await?;
+        let first_part_remote = first_part_remote.into_inner().into_inner();
+        assert_eq!(
+            first_part_local,
+            first_part_remote.as_slice(),
+            "First part bytes should be returned when requested"
+        );
+
+        assert_eq!(
+            partial_download_metadata.as_ref(),
+            Some(&metadata),
+            "We should get the same metadata back for partial download"
+        );
+
+        Ok(())
+    }
+
+    async fn upload_dummy_file(
+        workdir: &Path,
+        storage: &LocalFs,
+        name: &str,
+        metadata: Option<StorageMetadata>,
+    ) -> anyhow::Result<PathBuf> {
+        let timeline_path = workdir.join("timelines").join("some_timeline");
+        let relative_timeline_path = timeline_path.strip_prefix(&workdir)?;
+        let storage_path = storage.storage_root.join(relative_timeline_path).join(name);
+
+        let from_path = storage.working_directory.join(name);
+        let (file, size) = create_file_for_upload(&from_path, &dummy_contents(name)).await?;
+        storage.upload(file, size, &storage_path, metadata).await?;
+        Ok(storage_path)
+    }
+
+    async fn create_file_for_upload(
+        path: &Path,
+        contents: &str,
+    ) -> anyhow::Result<(io::BufReader<fs::File>, usize)> {
+        std::fs::create_dir_all(path.parent().unwrap())?;
+        let mut file_for_writing = std::fs::OpenOptions::new()
+            .write(true)
+            .create_new(true)
+            .open(path)?;
+        write!(file_for_writing, "{}", contents)?;
+        drop(file_for_writing);
+        let file_size = path.metadata()?.len() as usize;
+        Ok((
+            io::BufReader::new(fs::OpenOptions::new().read(true).open(&path).await?),
+            file_size,
+        ))
+    }
+
+    fn dummy_contents(name: &str) -> String {
+        format!("contents for {}", name)
+    }
+
+    async fn list_files_sorted(storage: &LocalFs) -> anyhow::Result<Vec<PathBuf>> {
+        let mut files = storage.list().await?;
+        files.sort();
+        Ok(files)
+    }
+}

libs/remote_storage/src/s3_bucket.rs

@@ -0,0 +1,465 @@
+//! AWS S3 storage wrapper around `rusoto` library.
+//!
+//! Respects `prefix_in_bucket` property from [`S3Config`],
+//! allowing multiple api users to independently work with the same S3 bucket, if
+//! their bucket prefixes are both specified and different.
+
+use std::path::{Path, PathBuf};
+
+use anyhow::Context;
+use rusoto_core::{
+    credential::{InstanceMetadataProvider, StaticProvider},
+    HttpClient, Region,
+};
+use rusoto_s3::{
+    DeleteObjectRequest, GetObjectRequest, ListObjectsV2Request, PutObjectRequest, S3Client,
+    StreamingBody, S3,
+};
+use tokio::{io, sync::Semaphore};
+use tokio_util::io::ReaderStream;
+use tracing::debug;
+
+use crate::{strip_path_prefix, RemoteStorage, S3Config};
+
+use super::StorageMetadata;
+
+const S3_PREFIX_SEPARATOR: char = '/';
+
+#[derive(Debug, Eq, PartialEq, PartialOrd, Ord, Hash)]
+pub struct S3ObjectKey(String);
+
+impl S3ObjectKey {
+    fn key(&self) -> &str {
+        &self.0
+    }
+
+    fn download_destination(&self, workdir: &Path, prefix_to_strip: Option<&str>) -> PathBuf {
+        let path_without_prefix = match prefix_to_strip {
+            Some(prefix) => self.0.strip_prefix(prefix).unwrap_or_else(|| {
+                panic!(
+                    "Could not strip prefix '{}' from S3 object key '{}'",
+                    prefix, self.0
+                )
+            }),
+            None => &self.0,
+        };
+
+        workdir.join(
+            path_without_prefix
+                .split(S3_PREFIX_SEPARATOR)
+                .collect::<PathBuf>(),
+        )
+    }
+}
+
+/// AWS S3 storage.
+pub struct S3Bucket {
+    workdir: PathBuf,
+    client: S3Client,
+    bucket_name: String,
+    prefix_in_bucket: Option<String>,
+    // Every request to S3 can be throttled or cancelled, if a certain number of requests per second is exceeded.
+    // Same goes to IAM, which is queried before every S3 request, if enabled. IAM has even lower RPS threshold.
+    // The helps to ensure we don't exceed the thresholds.
+    concurrency_limiter: Semaphore,
+}
+
+impl S3Bucket {
+    /// Creates the S3 storage, errors if incorrect AWS S3 configuration provided.
+    pub fn new(aws_config: &S3Config, workdir: PathBuf) -> anyhow::Result<Self> {
+        debug!(
+            "Creating s3 remote storage for S3 bucket {}",
+            aws_config.bucket_name
+        );
+        let region = match aws_config.endpoint.clone() {
+            Some(custom_endpoint) => Region::Custom {
+                name: aws_config.bucket_region.clone(),
+                endpoint: custom_endpoint,
+            },
+            None => aws_config
+                .bucket_region
+                .parse::<Region>()
+                .context("Failed to parse the s3 region from config")?,
+        };
+        let request_dispatcher = HttpClient::new().context("Failed to create S3 http client")?;
+
+        let access_key_id = std::env::var("AWS_ACCESS_KEY_ID").ok();
+        let secret_access_key = std::env::var("AWS_SECRET_ACCESS_KEY").ok();
+
+        let client = if access_key_id.is_none() && secret_access_key.is_none() {
+            debug!("Using IAM-based AWS access");
+            S3Client::new_with(request_dispatcher, InstanceMetadataProvider::new(), region)
+        } else {
+            debug!("Using credentials-based AWS access");
+            S3Client::new_with(
+                request_dispatcher,
+                StaticProvider::new_minimal(
+                    access_key_id.unwrap_or_default(),
+                    secret_access_key.unwrap_or_default(),
+                ),
+                region,
+            )
+        };
+
+        let prefix_in_bucket = aws_config.prefix_in_bucket.as_deref().map(|prefix| {
+            let mut prefix = prefix;
+            while prefix.starts_with(S3_PREFIX_SEPARATOR) {
+                prefix = &prefix[1..]
+            }
+
+            let mut prefix = prefix.to_string();
+            while prefix.ends_with(S3_PREFIX_SEPARATOR) {
+                prefix.pop();
+            }
+            prefix
+        });
+
+        Ok(Self {
+            client,
+            workdir,
+            bucket_name: aws_config.bucket_name.clone(),
+            prefix_in_bucket,
+            concurrency_limiter: Semaphore::new(aws_config.concurrency_limit.get()),
+        })
+    }
+}
+
+#[async_trait::async_trait]
+impl RemoteStorage for S3Bucket {
+    type RemoteObjectId = S3ObjectKey;
+
+    fn remote_object_id(&self, local_path: &Path) -> anyhow::Result<Self::RemoteObjectId> {
+        let relative_path = strip_path_prefix(&self.workdir, local_path)?;
+        let mut key = self.prefix_in_bucket.clone().unwrap_or_default();
+        for segment in relative_path {
+            key.push(S3_PREFIX_SEPARATOR);
+            key.push_str(&segment.to_string_lossy());
+        }
+        Ok(S3ObjectKey(key))
+    }
+
+    fn local_path(&self, storage_path: &Self::RemoteObjectId) -> anyhow::Result<PathBuf> {
+        Ok(storage_path.download_destination(&self.workdir, self.prefix_in_bucket.as_deref()))
+    }
+
+    async fn list(&self) -> anyhow::Result<Vec<Self::RemoteObjectId>> {
+        let mut document_keys = Vec::new();
+
+        let mut continuation_token = None;
+        loop {
+            let _guard = self
+                .concurrency_limiter
+                .acquire()
+                .await
+                .context("Concurrency limiter semaphore got closed during S3 list")?;
+            let fetch_response = self
+                .client
+                .list_objects_v2(ListObjectsV2Request {
+                    bucket: self.bucket_name.clone(),
+                    prefix: self.prefix_in_bucket.clone(),
+                    continuation_token,
+                    ..ListObjectsV2Request::default()
+                })
+                .await?;
+            document_keys.extend(
+                fetch_response
+                    .contents
+                    .unwrap_or_default()
+                    .into_iter()
+                    .filter_map(|o| Some(S3ObjectKey(o.key?))),
+            );
+
+            match fetch_response.continuation_token {
+                Some(new_token) => continuation_token = Some(new_token),
+                None => break,
+            }
+        }
+
+        Ok(document_keys)
+    }
+
+    async fn upload(
+        &self,
+        from: impl io::AsyncRead + Unpin + Send + Sync + 'static,
+        from_size_bytes: usize,
+        to: &Self::RemoteObjectId,
+        metadata: Option<StorageMetadata>,
+    ) -> anyhow::Result<()> {
+        let _guard = self
+            .concurrency_limiter
+            .acquire()
+            .await
+            .context("Concurrency limiter semaphore got closed during S3 upload")?;
+        self.client
+            .put_object(PutObjectRequest {
+                body: Some(StreamingBody::new_with_size(
+                    ReaderStream::new(from),
+                    from_size_bytes,
+                )),
+                bucket: self.bucket_name.clone(),
+                key: to.key().to_owned(),
+                metadata: metadata.map(|m| m.0),
+                ..PutObjectRequest::default()
+            })
+            .await?;
+        Ok(())
+    }
+
+    async fn download(
+        &self,
+        from: &Self::RemoteObjectId,
+        to: &mut (impl io::AsyncWrite + Unpin + Send + Sync),
+    ) -> anyhow::Result<Option<StorageMetadata>> {
+        let _guard = self
+            .concurrency_limiter
+            .acquire()
+            .await
+            .context("Concurrency limiter semaphore got closed during S3 download")?;
+        let object_output = self
+            .client
+            .get_object(GetObjectRequest {
+                bucket: self.bucket_name.clone(),
+                key: from.key().to_owned(),
+                ..GetObjectRequest::default()
+            })
+            .await?;
+
+        if let Some(body) = object_output.body {
+            let mut from = io::BufReader::new(body.into_async_read());
+            io::copy(&mut from, to).await?;
+        }
+
+        Ok(object_output.metadata.map(StorageMetadata))
+    }
+
+    async fn download_byte_range(
+        &self,
+        from: &Self::RemoteObjectId,
+        start_inclusive: u64,
+        end_exclusive: Option<u64>,
+        to: &mut (impl io::AsyncWrite + Unpin + Send + Sync),
+    ) -> anyhow::Result<Option<StorageMetadata>> {
+        // S3 accepts ranges as https://www.w3.org/Protocols/rfc2616/rfc2616-sec14.html#sec14.35
+        // and needs both ends to be exclusive
+        let end_inclusive = end_exclusive.map(|end| end.saturating_sub(1));
+        let range = Some(match end_inclusive {
+            Some(end_inclusive) => format!("bytes={}-{}", start_inclusive, end_inclusive),
+            None => format!("bytes={}-", start_inclusive),
+        });
+        let _guard = self
+            .concurrency_limiter
+            .acquire()
+            .await
+            .context("Concurrency limiter semaphore got closed during S3 range download")?;
+        let object_output = self
+            .client
+            .get_object(GetObjectRequest {
+                bucket: self.bucket_name.clone(),
+                key: from.key().to_owned(),
+                range,
+                ..GetObjectRequest::default()
+            })
+            .await?;
+
+        if let Some(body) = object_output.body {
+            let mut from = io::BufReader::new(body.into_async_read());
+            io::copy(&mut from, to).await?;
+        }
+
+        Ok(object_output.metadata.map(StorageMetadata))
+    }
+
+    async fn delete(&self, path: &Self::RemoteObjectId) -> anyhow::Result<()> {
+        let _guard = self
+            .concurrency_limiter
+            .acquire()
+            .await
+            .context("Concurrency limiter semaphore got closed during S3 delete")?;
+        self.client
+            .delete_object(DeleteObjectRequest {
+                bucket: self.bucket_name.clone(),
+                key: path.key().to_owned(),
+                ..DeleteObjectRequest::default()
+            })
+            .await?;
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use tempfile::tempdir;
+
+    use super::*;
+
+    #[test]
+    fn download_destination() -> anyhow::Result<()> {
+        let workdir = tempdir()?.path().to_owned();
+        let local_path = workdir.join("one").join("two").join("test_name");
+        let relative_path = local_path.strip_prefix(&workdir)?;
+
+        let key = S3ObjectKey(format!(
+            "{}{}",
+            S3_PREFIX_SEPARATOR,
+            relative_path
+                .iter()
+                .map(|segment| segment.to_str().unwrap())
+                .collect::<Vec<_>>()
+                .join(&S3_PREFIX_SEPARATOR.to_string()),
+        ));
+
+        assert_eq!(
+            local_path,
+            key.download_destination(&workdir, None),
+            "Download destination should consist of s3 path joined with the workdir prefix"
+        );
+
+        Ok(())
+    }
+
+    #[test]
+    fn storage_path_positive() -> anyhow::Result<()> {
+        let workdir = tempdir()?.path().to_owned();
+
+        let segment_1 = "matching";
+        let segment_2 = "file";
+        let local_path = &workdir.join(segment_1).join(segment_2);
+
+        let storage = dummy_storage(workdir);
+
+        let expected_key = S3ObjectKey(format!(
+            "{}{S3_PREFIX_SEPARATOR}{segment_1}{S3_PREFIX_SEPARATOR}{segment_2}",
+            storage.prefix_in_bucket.as_deref().unwrap_or_default(),
+        ));
+
+        let actual_key = storage
+            .remote_object_id(local_path)
+            .expect("Matching path should map to S3 path normally");
+        assert_eq!(
+            expected_key,
+            actual_key,
+            "S3 key from the matching path should contain all segments after the workspace prefix, separated with S3 separator"
+        );
+
+        Ok(())
+    }
+
+    #[test]
+    fn storage_path_negatives() -> anyhow::Result<()> {
+        #[track_caller]
+        fn storage_path_error(storage: &S3Bucket, mismatching_path: &Path) -> String {
+            match storage.remote_object_id(mismatching_path) {
+                Ok(wrong_key) => panic!(
+                    "Expected path '{}' to error, but got S3 key: {:?}",
+                    mismatching_path.display(),
+                    wrong_key,
+                ),
+                Err(e) => e.to_string(),
+            }
+        }
+
+        let workdir = tempdir()?.path().to_owned();
+        let storage = dummy_storage(workdir.clone());
+
+        let error_message = storage_path_error(&storage, &workdir);
+        assert!(
+            error_message.contains("Prefix and the path are equal"),
+            "Message '{}' does not contain the required string",
+            error_message
+        );
+
+        let mismatching_path = PathBuf::from("somewhere").join("else");
+        let error_message = storage_path_error(&storage, &mismatching_path);
+        assert!(
+            error_message.contains(mismatching_path.to_str().unwrap()),
+            "Error should mention wrong path"
+        );
+        assert!(
+            error_message.contains(workdir.to_str().unwrap()),
+            "Error should mention server workdir"
+        );
+        assert!(
+            error_message.contains("is not prefixed with"),
+            "Message '{}' does not contain a required string",
+            error_message
+        );
+
+        Ok(())
+    }
+
+    #[test]
+    fn local_path_positive() -> anyhow::Result<()> {
+        let workdir = tempdir()?.path().to_owned();
+        let storage = dummy_storage(workdir.clone());
+        let timeline_dir = workdir.join("timelines").join("test_timeline");
+        let relative_timeline_path = timeline_dir.strip_prefix(&workdir)?;
+
+        let s3_key = create_s3_key(
+            &relative_timeline_path.join("not a metadata"),
+            storage.prefix_in_bucket.as_deref(),
+        );
+        assert_eq!(
+            s3_key.download_destination(&workdir, storage.prefix_in_bucket.as_deref()),
+            storage
+                .local_path(&s3_key)
+                .expect("For a valid input, valid S3 info should be parsed"),
+            "Should be able to parse metadata out of the correctly named remote delta file"
+        );
+
+        let s3_key = create_s3_key(
+            &relative_timeline_path.join("metadata"),
+            storage.prefix_in_bucket.as_deref(),
+        );
+        assert_eq!(
+            s3_key.download_destination(&workdir, storage.prefix_in_bucket.as_deref()),
+            storage
+                .local_path(&s3_key)
+                .expect("For a valid input, valid S3 info should be parsed"),
+            "Should be able to parse metadata out of the correctly named remote metadata file"
+        );
+
+        Ok(())
+    }
+
+    #[test]
+    fn download_destination_matches_original_path() -> anyhow::Result<()> {
+        let workdir = tempdir()?.path().to_owned();
+        let original_path = workdir
+            .join("timelines")
+            .join("some_timeline")
+            .join("some name");
+
+        let dummy_storage = dummy_storage(workdir);
+
+        let key = dummy_storage.remote_object_id(&original_path)?;
+        let download_destination = dummy_storage.local_path(&key)?;
+
+        assert_eq!(
+            original_path, download_destination,
+            "'original path -> storage key -> matching fs path' transformation should produce the same path as the input one for the correct path"
+        );
+
+        Ok(())
+    }
+
+    fn dummy_storage(workdir: PathBuf) -> S3Bucket {
+        S3Bucket {
+            workdir,
+            client: S3Client::new("us-east-1".parse().unwrap()),
+            bucket_name: "dummy-bucket".to_string(),
+            prefix_in_bucket: Some("dummy_prefix/".to_string()),
+            concurrency_limiter: Semaphore::new(1),
+        }
+    }
+
+    fn create_s3_key(relative_file_path: &Path, prefix: Option<&str>) -> S3ObjectKey {
+        S3ObjectKey(relative_file_path.iter().fold(
+            prefix.unwrap_or_default().to_string(),
+            |mut path_string, segment| {
+                path_string.push(S3_PREFIX_SEPARATOR);
+                path_string.push_str(segment.to_str().unwrap());
+                path_string
+            },
+        ))
+    }
+}