diff --git a/CHANGELOG.md b/CHANGELOG.md index b63124459c6..ed91ddac050 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,9 +13,9 @@ As a minor extension, we have adopted a slightly different versioning convention - Support for bytes encoding of the SNARK aggregate signatures in the certificates. - Reduced the encoded size of the SNARK proofs by serializing their bytes as CBOR byte strings. - Support for reading on-chain protocol configurations for both Aggregator and Signer. - -- **UNSTABLE** Reworked the Mithril aggregator's file archiver to produce byte-stable archives across systems. - - Existing archives must be regenerated by the Mithril aggregator to ensure byte stability. + - Reworked the Mithril aggregator's file archiver to produce byte-stable archives across systems. + - Existing archives must be regenerated by the Mithril aggregator to ensure byte stability. + - Support IPFS for uploads of immutable files in the Mithril aggregator. - **REMOVED** support for `Gzip` compression/decompression in the Mithril aggregator and client: - The aggregator no longer produces or supports `Gzip` compression for snapshot-related artifacts: immutable files and ancillaries. diff --git a/Cargo.lock b/Cargo.lock index bad4d87307a..e6026e12a93 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4214,7 +4214,7 @@ dependencies = [ [[package]] name = "mithril-aggregator" -version = "0.10.0" +version = "0.10.1" dependencies = [ "anyhow", "async-trait", @@ -4418,7 +4418,7 @@ dependencies = [ [[package]] name = "mithril-client" -version = "0.14.19" +version = "0.14.20" dependencies = [ "anyhow", "async-trait", @@ -4456,7 +4456,7 @@ dependencies = [ [[package]] name = "mithril-client-cli" -version = "0.13.21" +version = "0.13.22" dependencies = [ "anyhow", "async-trait", @@ -4513,7 +4513,7 @@ dependencies = [ [[package]] name = "mithril-common" -version = "0.7.18" +version = "0.7.19" dependencies = [ "anyhow", "async-trait", diff --git a/internal/cardano-node/mithril-cardano-node-internal-database/Cargo.toml b/internal/cardano-node/mithril-cardano-node-internal-database/Cargo.toml index b44b872a882..9a4ef2de1bf 100644 --- a/internal/cardano-node/mithril-cardano-node-internal-database/Cargo.toml +++ b/internal/cardano-node/mithril-cardano-node-internal-database/Cargo.toml @@ -15,7 +15,7 @@ anyhow = { workspace = true } async-trait = { workspace = true } digest = { workspace = true } hex = { workspace = true } -mithril-common = { path = "../../../mithril-common", version = "0.7.18" } +mithril-common = { path = "../../../mithril-common", version = "0.7.19" } serde = { workspace = true } serde_json = { workspace = true } sha2 = "0.10.9" diff --git a/internal/mithril-aggregator-client/Cargo.toml b/internal/mithril-aggregator-client/Cargo.toml index 048776d8eac..dcf5b8884ad 100644 --- a/internal/mithril-aggregator-client/Cargo.toml +++ b/internal/mithril-aggregator-client/Cargo.toml @@ -13,7 +13,7 @@ include = ["**/*.rs", "Cargo.toml", "README.md"] [dependencies] anyhow = { workspace = true } async-trait = { workspace = true } -mithril-common = { path = "../../mithril-common", version = "0.7.18" } +mithril-common = { path = "../../mithril-common", version = "0.7.19" } reqwest = { workspace = true } semver = { workspace = true } serde = { workspace = true } diff --git a/internal/mithril-aggregator-discovery/Cargo.toml b/internal/mithril-aggregator-discovery/Cargo.toml index 43f6eff84c6..1e06d6de833 100644 --- a/internal/mithril-aggregator-discovery/Cargo.toml +++ b/internal/mithril-aggregator-discovery/Cargo.toml @@ -14,7 +14,7 @@ include = ["**/*.rs", "Cargo.toml", "README.md", ".gitignore"] anyhow = { workspace = true } async-trait = { workspace = true } mithril-aggregator-client = { path = "../mithril-aggregator-client", version = "0.2.4" } -mithril-common = { path = "../../mithril-common", version = "0.7.18" } +mithril-common = { path = "../../mithril-common", version = "0.7.19" } rand = { version = "0.10.2" } reqwest = { workspace = true } serde = { workspace = true } diff --git a/mithril-aggregator/Cargo.toml b/mithril-aggregator/Cargo.toml index 06cbc7d8abe..81739dd6e2b 100644 --- a/mithril-aggregator/Cargo.toml +++ b/mithril-aggregator/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mithril-aggregator" -version = "0.10.0" +version = "0.10.1" description = "A Mithril Aggregator server" authors = { workspace = true } edition = { workspace = true } diff --git a/mithril-aggregator/src/artifact_builder/cardano_database_artifacts/ancillary.rs b/mithril-aggregator/src/artifact_builder/cardano_database_artifacts/ancillary.rs index 4c07c17ef97..eb38e91afdc 100644 --- a/mithril-aggregator/src/artifact_builder/cardano_database_artifacts/ancillary.rs +++ b/mithril-aggregator/src/artifact_builder/cardano_database_artifacts/ancillary.rs @@ -177,9 +177,8 @@ impl AncillaryArtifactBuilder { } Err(e) => { error!( - self.logger, - "Failed to upload ancillary archive"; - "error" => e.to_string() + self.logger, "Failed to upload ancillary archive"; + "error" => ?e ); } } diff --git a/mithril-aggregator/src/artifact_builder/cardano_database_artifacts/digest.rs b/mithril-aggregator/src/artifact_builder/cardano_database_artifacts/digest.rs index 885d84f7b3b..b8380b484a0 100644 --- a/mithril-aggregator/src/artifact_builder/cardano_database_artifacts/digest.rs +++ b/mithril-aggregator/src/artifact_builder/cardano_database_artifacts/digest.rs @@ -251,9 +251,8 @@ impl DigestArtifactBuilder { } Err(e) => { error!( - self.logger, - "Failed to upload digest file"; - "error" => e.to_string() + self.logger, "Failed to upload digest file"; + "error" => ?e ); } } diff --git a/mithril-aggregator/src/artifact_builder/cardano_database_artifacts/immutable.rs b/mithril-aggregator/src/artifact_builder/cardano_database_artifacts/immutable.rs index 70203f73e74..c894dc89ce3 100644 --- a/mithril-aggregator/src/artifact_builder/cardano_database_artifacts/immutable.rs +++ b/mithril-aggregator/src/artifact_builder/cardano_database_artifacts/immutable.rs @@ -7,13 +7,15 @@ use slog::{Logger, error}; use mithril_common::{ StdResult, - entities::{CompressionAlgorithm, ImmutableFileNumber, ImmutablesLocation, MultiFilesUri}, + entities::{ + CompressionAlgorithm, ImmutableFileNumber, ImmutablesLocation, MultiFilesUri, TemplateUri, + }, logging::LoggerExtensions, }; use crate::{ DumbUploader, FileUploader, - file_uploaders::{CloudUploader, LocalUploader}, + file_uploaders::{CloudUploader, IpfsUploader, LocalUploader}, services::Snapshotter, }; @@ -99,6 +101,24 @@ impl ImmutableFilesUploader for LocalUploader { } } +#[async_trait] +impl ImmutableFilesUploader for IpfsUploader { + async fn batch_upload( + &self, + filepaths: &[PathBuf], + compression_algorithm: Option, + ) -> StdResult { + let directory_cid = self.batch_upload_to_dir(filepaths).await?; + + Ok(ImmutablesLocation::Ipfs { + uri: MultiFilesUri::Template(TemplateUri(format!( + "{directory_cid}/{{immutable_file_number}}.tar.zst" + ))), + compression_algorithm, + }) + } +} + #[async_trait] impl ImmutableFilesUploader for CloudUploader { async fn batch_upload( @@ -241,9 +261,8 @@ impl ImmutableArtifactBuilder { } Err(e) => { error!( - self.logger, - "Failed to upload immutable archive"; - "error" => e.to_string() + self.logger, "Failed to upload immutable archive"; + "error" => ?e ); } } diff --git a/mithril-aggregator/src/configuration.rs b/mithril-aggregator/src/configuration.rs index 4c9d7af2df4..7a9c683ed3b 100644 --- a/mithril-aggregator/src/configuration.rs +++ b/mithril-aggregator/src/configuration.rs @@ -130,6 +130,11 @@ pub trait ConfigurationSource { panic!("snapshot_use_cdn_domain is not implemented."); } + /// Configuration parameters to connect to a Kubo IPFS RPC API, setting this will enable IPFS upload for immutable snapshots + fn ipfs_rpc_server_config(&self) -> Option { + panic!("ipfs_rpc_server_config is not implemented."); + } + /// Server listening IP fn server_ip(&self) -> String { panic!("server_ip is not implemented."); @@ -534,6 +539,20 @@ pub struct ServeCommandConfiguration { /// Use CDN domain to construct snapshot urls if snapshot_uploader_type is Gcp pub snapshot_use_cdn_domain: bool, + /// URL of a Kubo IPFS RPC API, setting this will enable IPFS upload for immutable snapshots + /// + /// `mfs_folder_name` (optional) allows overriding the name of the folder in the IPFS MFS (Mutable File System) + /// where the snapshots will be stored. Defaults to "mithril" if not specified. + #[example = "\ + `{ \"url\": \"http://localhost:5001/\" }`\ + or `{ \"url\": \"http://localhost:5001/\", \"mfs_folder_name\": \"custom-folder\" }`\ + "] + #[serde( + default, + deserialize_with = "serde_deserialization::string_or_struct_optional" + )] + pub ipfs_rpc_server_config: Option, + /// Server listening IP pub server_ip: String, @@ -716,6 +735,44 @@ pub enum SnapshotUploaderType { Local, } +/// Configuration for connecting to a Kubo IPFS RPC server. +/// +/// This struct holds the connection details and settings for uploading +/// immutable snapshots to IPFS via the Kubo RPC API. +#[derive(Debug, Clone, Deserialize, PartialEq, Eq)] +pub struct IpfsRpcServerConfig { + /// URL of the Kubo IPFS RPC API endpoint. + /// + /// Example: `http://localhost:5001/` + pub url: String, + + /// Name of the folder in the MFS (Mutable File System) where snapshots will be stored. + /// + /// Defaults to "mithril" if not specified. + #[serde(default = "default_mfs_folder_name")] + pub mfs_folder_name: String, +} + +impl IpfsRpcServerConfig { + /// Parsed URL of a Kubo IPFS RPC API + pub fn sanitized_url(&self) -> StdResult { + SanitizedUrlWithTrailingSlash::parse(&self.url) + .with_context(|| "Invalid IPFS RPC server URL") + } +} + +fn default_mfs_folder_name() -> String { + "mithril".to_string() +} + +impl FromStr for IpfsRpcServerConfig { + type Err = serde_json::Error; + + fn from_str(s: &str) -> Result { + serde_json::from_str(s) + } +} + /// Configuration to connect to the Blockfrost API. /// /// Currently only used to fetch the ticker and name for registered pools. @@ -793,6 +850,7 @@ impl ServeCommandConfiguration { snapshot_uploader_type: SnapshotUploaderType::Local, snapshot_bucket_name: None, snapshot_use_cdn_domain: false, + ipfs_rpc_server_config: None, server_ip: "0.0.0.0".to_string(), server_port: 8000, public_server_url: None, @@ -911,6 +969,10 @@ impl ConfigurationSource for ServeCommandConfiguration { self.snapshot_use_cdn_domain } + fn ipfs_rpc_server_config(&self) -> Option { + self.ipfs_rpc_server_config.clone() + } + fn server_ip(&self) -> String { self.server_ip.clone() } @@ -1519,6 +1581,31 @@ mod test { assert!(!config.is_follower_aggregator()); } + #[test] + fn deserializing_ipfs_rpc_server_parameters() { + let deserialized_without_mfs_dir: IpfsRpcServerConfig = + serde_json::from_str(r#"{ "url": "http://localhost:5001/" }"#).unwrap(); + assert_eq!( + deserialized_without_mfs_dir, + IpfsRpcServerConfig { + url: "http://localhost:5001/".to_string(), + mfs_folder_name: default_mfs_folder_name(), + } + ); + + let deserialized_with_mfs_dir: IpfsRpcServerConfig = serde_json::from_str( + r#"{ "url": "http://localhost:5001/", "mfs_folder_name": "altered" }"#, + ) + .unwrap(); + assert_eq!( + deserialized_with_mfs_dir, + IpfsRpcServerConfig { + url: "http://localhost:5001/".to_string(), + mfs_folder_name: "altered".to_string(), + } + ); + } + #[test] fn deserializing_blockfrost_parameters() { let deserialized_without_base_url: BlockfrostParameters = diff --git a/mithril-aggregator/src/dependency_injection/builder/protocol/artifacts.rs b/mithril-aggregator/src/dependency_injection/builder/protocol/artifacts.rs index c579f2d2320..2b1c3768489 100644 --- a/mithril-aggregator/src/dependency_injection/builder/protocol/artifacts.rs +++ b/mithril-aggregator/src/dependency_injection/builder/protocol/artifacts.rs @@ -13,11 +13,12 @@ use crate::artifact_builder::{ DigestSnapshotter, ImmutableArtifactBuilder, ImmutableFilesUploader, MithrilStakeDistributionArtifactBuilder, }; -use crate::configuration::AncillaryFilesSignerConfig; +use crate::configuration::{AncillaryFilesSignerConfig, IpfsRpcServerConfig}; use crate::dependency_injection::builder::SNAPSHOT_ARTIFACTS_DIR; use crate::dependency_injection::{DependenciesBuilder, DependenciesBuilderError, Result}; use crate::file_uploaders::{ - CloudRemotePath, CloudUploader, FileUploadRetryPolicy, GCloudBackendUploader, LocalUploader, + CloudRemotePath, CloudUploader, FileUploadRetryPolicy, GCloudBackendUploader, IpfsUploader, + LocalUploader, }; use crate::get_dependency; use crate::http_server::{CARDANO_DATABASE_DOWNLOAD_PATH, SNAPSHOT_DOWNLOAD_PATH}; @@ -29,6 +30,7 @@ use crate::services::{ SignedEntityServiceArtifactsDependencies, Snapshotter, }; use crate::tools::DEFAULT_GCP_CREDENTIALS_JSON_ENV_VAR; +use crate::tools::kubo_rpc_client::{IpfsMfsDirPath, KuboRpcClient}; use crate::{DumbUploader, ExecutionEnvironment, FileUploader, SnapshotUploaderType}; impl DependenciesBuilder { @@ -253,6 +255,20 @@ impl DependenciesBuilder { )) } + async fn build_ipfs_uploader( + &self, + ipfs_rpc_config: IpfsRpcServerConfig, + ) -> Result { + let rpc_api_client = + KuboRpcClient::new(ipfs_rpc_config.sanitized_url()?, self.root_logger())?; + Ok(IpfsUploader::new( + Arc::new(rpc_api_client), + IpfsMfsDirPath::from(ipfs_rpc_config.mfs_folder_name), + FileUploadRetryPolicy::default(), + &self.root_logger(), + )) + } + async fn build_cardano_database_ancillary_uploaders( &self, ) -> Result>> { @@ -301,28 +317,36 @@ impl DependenciesBuilder { ) -> Result>> { let logger = self.root_logger(); if self.configuration.environment() == ExecutionEnvironment::Production { - match self.configuration.snapshot_uploader_type() { - SnapshotUploaderType::Gcp => { - let allow_overwrite = false; - let remote_folder_path = - CloudRemotePath::new("cardano-database").join("immutable"); - - Ok(vec![Arc::new( - self.build_gcp_uploader(remote_folder_path, allow_overwrite).await?, - )]) - } - SnapshotUploaderType::Local => { - let server_url_prefix = self.configuration.get_server_url()?; - let immutable_url_prefix = server_url_prefix - .sanitize_join(&format!("{CARDANO_DATABASE_DOWNLOAD_PATH}/immutable/"))?; - - Ok(vec![Arc::new(LocalUploader::new_without_copy( - immutable_url_prefix, - FileUploadRetryPolicy::default(), - logger, - ))]) - } + let mut uploaders: Vec> = + match self.configuration.snapshot_uploader_type() { + SnapshotUploaderType::Gcp => { + let allow_overwrite = false; + let remote_folder_path = + CloudRemotePath::new("cardano-database").join("immutable"); + + vec![Arc::new( + self.build_gcp_uploader(remote_folder_path, allow_overwrite).await?, + )] + } + SnapshotUploaderType::Local => { + let server_url_prefix = self.configuration.get_server_url()?; + let immutable_url_prefix = server_url_prefix.sanitize_join(&format!( + "{CARDANO_DATABASE_DOWNLOAD_PATH}/immutable/" + ))?; + + vec![Arc::new(LocalUploader::new_without_copy( + immutable_url_prefix, + FileUploadRetryPolicy::default(), + logger, + ))] + } + }; + + if let Some(ipfs_rpc_config) = self.configuration.ipfs_rpc_server_config() { + uploaders.push(Arc::new(self.build_ipfs_uploader(ipfs_rpc_config).await?)); } + + Ok(uploaders) } else { Ok(vec![Arc::new(DumbUploader::new( FileUploadRetryPolicy::never(), diff --git a/mithril-aggregator/src/file_uploaders/interface.rs b/mithril-aggregator/src/file_uploaders/interface.rs index 5aa68a4f516..416ad9c82d9 100644 --- a/mithril-aggregator/src/file_uploaders/interface.rs +++ b/mithril-aggregator/src/file_uploaders/interface.rs @@ -47,20 +47,35 @@ pub trait FileUploader: Sync + Send { /// Upload a file with retries according to the retry policy. async fn upload(&self, filepath: &Path) -> StdResult { let retry_policy = self.retry_policy(); + retry( + async || self.upload_without_retry(filepath).await, + retry_policy, + format!(" Uploaded file path: {}", filepath.display()), + ) + .await + } +} - let mut nb_attempts = 0; - loop { - nb_attempts += 1; - match self.upload_without_retry(filepath).await { - Ok(result) => return Ok(result), - Err(e) if nb_attempts >= retry_policy.attempts => { - return Err(e.context(format!( - "Upload failed after {nb_attempts} attempts. Uploaded file path: {}", - filepath.display() - ))); - } - _ => tokio::time::sleep(retry_policy.delay_between_attempts).await, +pub(super) async fn retry( + f: F, + retry_policy: FileUploadRetryPolicy, + additional_err_context: String, +) -> StdResult +where + F: Fn() -> Fut, + Fut: Future>, +{ + let mut nb_attempts = 0; + loop { + nb_attempts += 1; + match f().await { + Ok(result) => return Ok(result), + Err(e) if nb_attempts >= retry_policy.attempts => { + return Err(e.context(format!( + "Upload failed after {nb_attempts} attempts.{additional_err_context}" + ))); } + _ => tokio::time::sleep(retry_policy.delay_between_attempts).await, } } } diff --git a/mithril-aggregator/src/file_uploaders/ipfs_uploader.rs b/mithril-aggregator/src/file_uploaders/ipfs_uploader.rs new file mode 100644 index 00000000000..ffdf51434b8 --- /dev/null +++ b/mithril-aggregator/src/file_uploaders/ipfs_uploader.rs @@ -0,0 +1,521 @@ +use std::collections::HashMap; +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use anyhow::Context; +use slog::{Logger, trace}; + +use mithril_common::StdResult; +use mithril_common::entities::FileUri; +use mithril_common::logging::LoggerExtensions; + +use crate::FileUploader; +use crate::file_uploaders::FileUploadRetryPolicy; +use crate::file_uploaders::interface::retry; +use crate::tools::kubo_rpc_client::query::{ + IpfsAddQuery, IpfsFilesLsQuery, IpfsFilesMkdirQuery, IpfsFilesStatQuery, +}; +use crate::tools::kubo_rpc_client::{IpfsMfsDirPath, KuboRpcClient}; + +/// IPFS Content Identifier (CID) +pub type IpfsCid = String; + +/// File uploader that stores files to IPFS +pub struct IpfsUploader { + rpc_client: Arc, + ipfs_dir_path: IpfsMfsDirPath, + retry_policy: FileUploadRetryPolicy, + logger: Logger, +} + +impl IpfsUploader { + /// Create a new IPFS uploader + pub fn new( + rpc_client: Arc, + ipfs_dir_path: IpfsMfsDirPath, + retry_policy: FileUploadRetryPolicy, + logger: &Logger, + ) -> Self { + Self { + rpc_client, + ipfs_dir_path, + retry_policy, + logger: logger.new_with_component_name::(), + } + } + + /// Upload a batch of files at once to the target IPFS MFS directory, returning the updated CID + /// of the directory + /// + /// Compared to uploading the files one by one: + /// - it checks if the target directory exists in IPFS only once + /// - it batches the existence checks of the files by listing them using one `files ls` query + pub async fn batch_upload_to_dir(&self, files: &[PathBuf]) -> StdResult { + self.ensure_directory_exists().await?; + + let existing_entries = self + .rpc_client + .list_directory_files(&self.ipfs_dir_path) + .await + .with_context(|| "listing files in IPFS MFS directory")?; + + for file_path in files { + let filename = + file_path + .file_name() + .and_then(|name| name.to_str()) + .with_context(|| { + format!( + "File path '{}' has no valid UTF-8 filename", + file_path.display() + ) + })?; + + if existing_entries.contains_key(filename) { + continue; + } + + // We are bypassing retry-capable [FileUploader::upload] to avoid already batched checks, so we need to retry manually + retry( + || async { self.upload_missing_file_once(file_path).await }, + self.retry_policy(), + format!(" Uploaded file path: {}", file_path.display()), + ) + .await?; + } + + self.get_current_directory_cid().await + } + + /// Get the current directory CID, reflecting the latest state of the directory + pub async fn get_current_directory_cid(&self) -> StdResult { + self.rpc_client.get_dir_cid(&self.ipfs_dir_path).await + } + + async fn ensure_directory_exists(&self) -> StdResult<()> { + self.rpc_client + .create_dir(&self.ipfs_dir_path) + .await + .with_context(|| { + format!( + "Failed to create directory '{}' in IPFS", + self.ipfs_dir_path + ) + }) + } + + async fn upload_file_once(&self, filepath: &Path) -> StdResult { + self.ensure_directory_exists().await?; + + if let Some(cid) = self + .rpc_client + .file_exists(&self.ipfs_dir_path, filepath) + .await + .with_context(|| { + format!( + "Failed to check if file '{}' exists in IPFS", + filepath.display() + ) + })? + { + trace!(self.logger, "File already exists in IPFS"; "cid" => %cid); + return Ok(FileUri(cid)); + } + + self.upload_missing_file_once(filepath).await + } + + // Note: this method assumes that the target directory exists and the file does not exist in IPFS + async fn upload_missing_file_once(&self, filepath: &Path) -> StdResult { + trace!(self.logger, "Uploading file to IPFS"; "file_path" => %filepath.display()); + + let cid = self + .rpc_client + .upload_file(filepath, &self.ipfs_dir_path) + .await + .with_context(|| format!("Failed to upload file '{}' to IPFS", filepath.display()))?; + trace!( + self.logger, "File upload to IPFS finished"; + "file_path" => %filepath.display(), "cid" => %cid + ); + + Ok(FileUri(cid)) + } +} + +#[async_trait::async_trait] +impl FileUploader for IpfsUploader { + async fn upload_without_retry(&self, filepath: &Path) -> StdResult { + self.upload_file_once(filepath).await + } + + fn retry_policy(&self) -> FileUploadRetryPolicy { + self.retry_policy.clone() + } +} + +/// Backend trait for IPFS operations +#[cfg_attr(test, mockall::automock)] +#[async_trait::async_trait] +pub trait IpfsBackendUploader: Sync + Send { + /// Create a directory in IPFS + async fn create_dir(&self, dir_path: &IpfsMfsDirPath) -> StdResult<()>; + + /// List all paths in an MFS directory + async fn list_directory_files( + &self, + dir_path: &IpfsMfsDirPath, + ) -> StdResult>; + + /// Get the CID of a directory + async fn get_dir_cid(&self, dir_path: &IpfsMfsDirPath) -> StdResult; + + /// Upload a file to IPFS and return its CID + async fn upload_file(&self, file_path: &Path, mfs_path: &IpfsMfsDirPath) -> StdResult; + + /// Check if a file exists in a given MFS directory and return its CID if it does + async fn file_exists( + &self, + mfs_dir_path: &IpfsMfsDirPath, + file_path: &Path, + ) -> StdResult>; +} + +#[async_trait::async_trait] +impl IpfsBackendUploader for KuboRpcClient { + async fn create_dir(&self, dir_path: &IpfsMfsDirPath) -> StdResult<()> { + self.send(IpfsFilesMkdirQuery::create_mfs_directory(dir_path)).await + } + + async fn list_directory_files( + &self, + dir_path: &IpfsMfsDirPath, + ) -> StdResult> { + let response = self.send(IpfsFilesLsQuery::new(dir_path)).await?; + Ok(response.unwrap_or_default()) + } + + async fn get_dir_cid(&self, dir_path: &IpfsMfsDirPath) -> StdResult { + let stat = self + .send(IpfsFilesStatQuery::new(dir_path.as_ref())) + .await? + .with_context(|| format!("Directory {dir_path} does not exist in IPFS node",))?; + Ok(stat.hash) + } + + async fn upload_file(&self, file_path: &Path, mfs_path: &IpfsMfsDirPath) -> StdResult { + let res = self + .send(IpfsAddQuery::new_with_mfs_reference(file_path, mfs_path)) + .await?; + Ok(res.hash) + } + + async fn file_exists( + &self, + mfs_dir_path: &IpfsMfsDirPath, + file_path: &Path, + ) -> StdResult> { + let stat = self + .send(IpfsFilesStatQuery::new( + mfs_dir_path.join_file_name_from(file_path)?, + )) + .await?; + Ok(stat.map(|stat| stat.hash)) + } +} + +#[cfg(test)] +mod tests { + use mockall::predicate::eq; + + use mithril_common::test::mock_extensions::MockBuilder; + + use crate::test::TestLogger; + + use super::*; + + impl IpfsUploader { + fn new_for_test>( + mfs_dir: P, + mock_config: impl FnOnce(&mut MockIpfsBackendUploader), + ) -> Self { + Self::new( + MockBuilder::configure(mock_config), + mfs_dir.into(), + FileUploadRetryPolicy::never(), + &TestLogger::stdout(), + ) + } + } + + #[tokio::test] + async fn existing_file_is_not_uploaded_and_its_cid_is_returned() { + let uploader = IpfsUploader::new_for_test(IpfsMfsDirPath::from("/test/dir"), |mock| { + mock.expect_create_dir().returning(|_| Ok(())); + mock.expect_file_exists() + .with( + eq(IpfsMfsDirPath::from("/test/dir")), + eq(Path::new("/a/dummy-file.txt")), + ) + .returning(|_, _| Ok(Some("existing".to_string()))); + mock.expect_upload_file().never(); + }); + + let result = uploader + .upload_without_retry(Path::new("/a/dummy-file.txt")) + .await + .unwrap(); + + assert_eq!(FileUri("existing".to_string()), result); + } + + #[tokio::test] + async fn non_existing_file_is_uploaded_and_its_cid_is_returned() { + let uploader = IpfsUploader::new_for_test(IpfsMfsDirPath::from("/test/dir"), |mock| { + mock.expect_create_dir() + .with(eq(IpfsMfsDirPath::from("/test/dir"))) + .returning(|_| Ok(())); + mock.expect_file_exists() + .with( + eq(IpfsMfsDirPath::from("/test/dir")), + eq(Path::new("/a/dummy-file.txt")), + ) + .returning(|_, _| Ok(None)); + mock.expect_upload_file() + .with( + eq(Path::new("/a/dummy-file.txt")), + eq(IpfsMfsDirPath::from("/test/dir")), + ) + .returning(|_, _| Ok("test_cid".to_string())); + }); + + uploader + .upload_without_retry(Path::new("/a/dummy-file.txt")) + .await + .unwrap(); + } + + mod batch_upload { + use std::time::Duration; + + use anyhow::anyhow; + + use super::*; + + const MFS_DIR: &str = "/test/dir"; + + #[tokio::test] + async fn uploads_only_missing_files_and_returns_directory_cid() { + let existing_files = HashMap::from([( + "already-uploaded.txt".to_string(), + "existing-cid".to_string(), + )]); + + let uploader = IpfsUploader::new_for_test(MFS_DIR, move |mock| { + mock.expect_create_dir() + .with(eq(IpfsMfsDirPath::from(MFS_DIR))) + .return_once(|_| Ok(())) + .once(); + mock.expect_list_directory_files() + .with(eq(IpfsMfsDirPath::from(MFS_DIR))) + .return_once(move |_| Ok(existing_files)) + .once(); + mock.expect_file_exists().never(); + mock.expect_upload_file() + .with( + eq(PathBuf::from("/local/new-file-1.txt")), + eq(IpfsMfsDirPath::from(MFS_DIR)), + ) + .return_once(|_, _| Ok("file-1-cid".to_string())) + .once(); + mock.expect_upload_file() + .with( + eq(PathBuf::from("/other/new-file-2.txt")), + eq(IpfsMfsDirPath::from(MFS_DIR)), + ) + .return_once(|_, _| Ok("file-2-cid".to_string())) + .once(); + mock.expect_get_dir_cid() + .with(eq(IpfsMfsDirPath::from(MFS_DIR))) + .return_once(|_| Ok("directory-cid".to_string())) + .once(); + }); + + let directory_cid = uploader + .batch_upload_to_dir(&[ + PathBuf::from("/local/already-uploaded.txt"), + PathBuf::from("/local/new-file-1.txt"), + PathBuf::from("/other/new-file-2.txt"), + ]) + .await + .unwrap(); + + assert_eq!("directory-cid", directory_cid); + } + + #[tokio::test] + async fn support_retry() { + let mut uploader = IpfsUploader::new_for_test(MFS_DIR, move |mock| { + mock.expect_create_dir() + .with(eq(IpfsMfsDirPath::from(MFS_DIR))) + .return_once(|_| Ok(())) + .once(); + mock.expect_list_directory_files() + .with(eq(IpfsMfsDirPath::from(MFS_DIR))) + .return_once(move |_| Ok(HashMap::new())) + .once(); + mock.expect_upload_file() + .with( + eq(PathBuf::from("/local/new-file.txt")), + eq(IpfsMfsDirPath::from(MFS_DIR)), + ) + .return_once(|_, _| Err(anyhow!("first upload failed"))) + .once(); + mock.expect_upload_file() + .with( + eq(PathBuf::from("/local/new-file.txt")), + eq(IpfsMfsDirPath::from(MFS_DIR)), + ) + .return_once(|_, _| Ok("file-cid".to_string())) + .once(); + mock.expect_get_dir_cid() + .with(eq(IpfsMfsDirPath::from(MFS_DIR))) + .return_once(|_| Ok("directory-cid".to_string())) + .once(); + }); + uploader.retry_policy = FileUploadRetryPolicy { + attempts: 2, + delay_between_attempts: Duration::from_millis(5), + }; + + let directory_cid = uploader + .batch_upload_to_dir(&[PathBuf::from("/local/new-file.txt")]) + .await + .unwrap(); + + assert_eq!("directory-cid", directory_cid); + } + + #[tokio::test] + async fn empty_batch_returns_current_directory_cid_without_uploading_files() { + let uploader = IpfsUploader::new_for_test(MFS_DIR, |mock| { + mock.expect_create_dir() + .with(eq(IpfsMfsDirPath::from(MFS_DIR))) + .return_once(|_| Ok(())) + .once(); + mock.expect_list_directory_files() + .with(eq(IpfsMfsDirPath::from(MFS_DIR))) + .return_once(|_| Ok(HashMap::new())) + .once(); + mock.expect_file_exists().never(); + mock.expect_upload_file().never(); + mock.expect_get_dir_cid() + .with(eq(IpfsMfsDirPath::from(MFS_DIR))) + .return_once(|_| Ok("directory-cid".to_string())) + .once(); + }); + + let directory_cid = uploader.batch_upload_to_dir(&[]).await.unwrap(); + + assert_eq!("directory-cid", directory_cid); + } + + #[tokio::test] + async fn returns_error_when_a_file_path_has_no_filename() { + let uploader = IpfsUploader::new_for_test(MFS_DIR, |mock| { + mock.expect_create_dir().return_once(|_| Ok(())).once(); + mock.expect_list_directory_files() + .return_once(|_| Ok(HashMap::new())) + .once(); + mock.expect_file_exists().never(); + mock.expect_upload_file().never(); + mock.expect_get_dir_cid().never(); + }); + + uploader + .batch_upload_to_dir(&[PathBuf::new()]) + .await + .expect_err("a path without a filename should fail"); + } + + #[tokio::test] + async fn returns_error_when_directory_creation_fails() { + let uploader = IpfsUploader::new_for_test(MFS_DIR, |mock| { + mock.expect_create_dir() + .with(eq(IpfsMfsDirPath::from(MFS_DIR))) + .return_once(|_| Err(anyhow!("create directory failure"))) + .once(); + mock.expect_list_directory_files().never(); + mock.expect_upload_file().never(); + mock.expect_get_dir_cid().never(); + }); + + uploader + .batch_upload_to_dir(&[PathBuf::from("new-file.txt")]) + .await + .expect_err("directory creation failure should be returned"); + } + + #[tokio::test] + async fn returns_error_when_listing_directory_files_fails() { + let uploader = IpfsUploader::new_for_test(MFS_DIR, |mock| { + mock.expect_create_dir().return_once(|_| Ok(())).once(); + mock.expect_list_directory_files() + .with(eq(IpfsMfsDirPath::from(MFS_DIR))) + .return_once(|_| Err(anyhow!("list directory failure"))) + .once(); + mock.expect_upload_file().never(); + mock.expect_get_dir_cid().never(); + }); + + uploader + .batch_upload_to_dir(&[PathBuf::from("new-file.txt")]) + .await + .expect_err("directory listing failure should be returned"); + } + + #[tokio::test] + async fn returns_error_when_uploading_a_missing_file_fails() { + let uploader = IpfsUploader::new_for_test(MFS_DIR, |mock| { + mock.expect_create_dir().return_once(|_| Ok(())).once(); + mock.expect_list_directory_files() + .return_once(|_| Ok(HashMap::new())) + .once(); + mock.expect_file_exists().never(); + mock.expect_upload_file() + .with( + eq(PathBuf::from("/local/new-file.txt")), + eq(IpfsMfsDirPath::from(MFS_DIR)), + ) + .return_once(|_, _| Err(anyhow!("upload failure"))) + .once(); + mock.expect_get_dir_cid().never(); + }); + + uploader + .batch_upload_to_dir(&[PathBuf::from("/local/new-file.txt")]) + .await + .expect_err("file upload failure should be returned"); + } + + #[tokio::test] + async fn returns_error_when_retrieving_directory_cid_fails() { + let uploader = IpfsUploader::new_for_test(MFS_DIR, |mock| { + mock.expect_create_dir().return_once(|_| Ok(())).once(); + mock.expect_list_directory_files() + .return_once(|_| Ok(HashMap::new())) + .once(); + mock.expect_upload_file().never(); + mock.expect_get_dir_cid() + .with(eq(IpfsMfsDirPath::from(MFS_DIR))) + .return_once(|_| Err(anyhow!("get directory CID failure"))) + .once(); + }); + + uploader + .batch_upload_to_dir(&[]) + .await + .expect_err("directory CID retrieval failure should be returned"); + } + } +} diff --git a/mithril-aggregator/src/file_uploaders/mod.rs b/mithril-aggregator/src/file_uploaders/mod.rs index 5851aa953cb..702a128bed4 100644 --- a/mithril-aggregator/src/file_uploaders/mod.rs +++ b/mithril-aggregator/src/file_uploaders/mod.rs @@ -1,9 +1,11 @@ mod cloud_uploader; mod dumb_uploader; mod interface; +mod ipfs_uploader; mod local_uploader; pub use cloud_uploader::{CloudRemotePath, CloudUploader, GCloudBackendUploader}; pub use dumb_uploader::*; pub use interface::{FileUploadRetryPolicy, FileUploader}; +pub use ipfs_uploader::IpfsUploader; pub use local_uploader::LocalUploader; diff --git a/mithril-aggregator/src/tools/kubo_rpc_client/api.rs b/mithril-aggregator/src/tools/kubo_rpc_client/api.rs new file mode 100644 index 00000000000..13607d81d58 --- /dev/null +++ b/mithril-aggregator/src/tools/kubo_rpc_client/api.rs @@ -0,0 +1,353 @@ +use std::time::Duration; + +use anyhow::Context; +use reqwest::{RequestBuilder, Response, StatusCode, Url}; +use slog::{Logger, trace}; + +use mithril_common::logging::LoggerExtensions; +use mithril_common::{StdError, StdResult}; + +use crate::tools::url_sanitizer::SanitizedUrlWithTrailingSlash; + +/// Trait for defining RPC queries to the Kubo IPFS node. +#[async_trait::async_trait] +pub trait KuboRpcQuery: Sync { + /// The type of the successful response from this query. + type Response; + + /// Returns the API route path for this query. + fn route(&self) -> String; + + /// Configures the RPC request before sending. + /// + /// Override this method to add query parameters, headers, or body to the request. + /// + /// Note: Configuration made by the [KuboRpcClient] won't be overridden since they are applied after this method. + async fn configure_request( + &self, + request_builder: RequestBuilder, + ) -> StdResult { + Ok(request_builder) + } + + /// Timeout for the RPC request. + fn timeout(&self) -> Duration { + Duration::from_secs(1) + } + + /// Handles a successful RPC response and converts it to the expected response type. + async fn handle_success(&self, response: Response) -> StdResult; + + /// Handles an error RPC response + /// + /// Default to returning an error with the response text, but can be overridden to handle errors + /// differently. + async fn handle_error(&self, response: Response) -> StdResult { + let status_code = response.status(); + let response_text = response.text().await.unwrap_or_else(|e| e.to_string()); + + Err(format_response_error(status_code, &response_text)) + } +} + +/// HTTP client for sending RPC requests to a Kubo IPFS node. +/// +/// This requester handles the HTTP communication layer, including request construction, +/// timeout management, and error handling. +pub struct KuboRpcClient { + rpc_base_url: Url, + client: reqwest::Client, + logger: Logger, +} + +impl KuboRpcClient { + /// Creates a new Kubo RPC client. + pub fn new(rpc_base_url: SanitizedUrlWithTrailingSlash, logger: Logger) -> StdResult { + Ok(Self { + rpc_base_url: rpc_base_url.into(), + client: reqwest::Client::builder() + .build() + .with_context(|| "RPC client creation failed")?, + logger: logger.new_with_component_name::(), + }) + } + + /// Sends an RPC query to the Kubo node and returns the parsed response. + /// + /// Returns an error if the request fails, times out, or the response indicates failure. + pub async fn send(&self, query: Q) -> StdResult { + let route = query.route(); + let endpoint = join_endpoint(&self.rpc_base_url, &route)?; + trace!(self.logger, "Kubo RPC POST"; "endpoint" => %endpoint); + + let request_builder = query + .configure_request(self.client.post(endpoint)) + .await + .with_context(|| { + format!("Failed to configure request for Kubo RPC endpoint: '{route}'") + })? + .timeout(query.timeout()); + + let response = request_builder + .send() + .await + .with_context(|| format!("Failed to send request to Kubo RPC endpoint: '{route}'"))?; + + if response.status().is_success() { + query.handle_success(response).await + } else { + query.handle_error(response).await + } + } +} + +fn join_endpoint(base_url: &Url, endpoint: &str) -> StdResult { + let normalized_endpoint = if let Some(stripped_endpoint) = endpoint.strip_prefix("/") { + stripped_endpoint + } else { + endpoint + }; + + base_url + .join(normalized_endpoint) + .with_context(|| format!("Could not join `{base_url}` to URL `{endpoint}`")) +} + +pub(super) fn format_response_error(status: StatusCode, response_text: &str) -> StdError { + anyhow::anyhow!("Request to Kubo RPC failed: {status}: '{response_text}'") +} + +pub(super) async fn handle_file_not_exist_error( + query_name: &str, + response: Response, +) -> StdResult> { + let status = response.status(); + let body = response + .text() + .await + .with_context(|| format!("Failed to read IPFS {query_name} error response"))?; + + if body.contains("file does not exist") { + Ok(None) + } else { + Err(format_response_error(status, &body)) + } +} + +#[cfg(test)] +mod tests { + use httpmock::Method::POST; + + use crate::tools::kubo_rpc_client::test_tools::setup_server_and_client; + + use super::*; + + struct QueryWithoutResponse; + + #[async_trait::async_trait] + impl KuboRpcQuery for QueryWithoutResponse { + type Response = (); + + fn route(&self) -> String { + "foo".to_string() + } + + async fn handle_success(&self, _response: Response) -> StdResult { + Ok(()) + } + } + + struct QueryWithResponseAndParam { + param: String, + } + + #[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, serde::Deserialize)] + struct QueryResponse { + foo: String, + bar: u32, + } + + #[async_trait::async_trait] + impl KuboRpcQuery for QueryWithResponseAndParam { + type Response = QueryResponse; + + fn route(&self) -> String { + "/foo".to_string() + } + + async fn configure_request( + &self, + request_builder: RequestBuilder, + ) -> StdResult { + Ok(request_builder.query(&[("param", &self.param)])) + } + + async fn handle_success(&self, response: Response) -> StdResult { + let json = response.json().await?; + Ok(json) + } + } + + #[test] + fn join_an_endpoint_with_a_leading_slash_should_keep_existing_components() { + let base_url = Url::parse("http://localhost:5001/api/v0/").unwrap(); + + assert_eq!( + "http://localhost:5001/api/v0/foo", + join_endpoint(&base_url, "/foo").unwrap().as_str() + ); + } + + #[tokio::test] + async fn minimal_request_with_only_route() { + let (server, client) = setup_server_and_client(); + server.mock(|when, then| { + when.method(POST).path("/foo"); + then.status(200); + }); + + client.send(QueryWithoutResponse).await.unwrap(); + } + + #[tokio::test] + async fn minimal_request_with_route_and_param() { + let expected_response = QueryResponse { + foo: "pika".to_string(), + bar: 123, + }; + let (server, client) = setup_server_and_client(); + server.mock(|when, then| { + when.method(POST).path("/foo").query_param("param", "bar"); + then.status(200).json_body_obj(&expected_response); + }); + + let response = client + .send(QueryWithResponseAndParam { + param: "bar".to_string(), + }) + .await + .unwrap(); + + assert_eq!(expected_response, response); + } + + #[tokio::test] + async fn query_times_out_when_response_exceeds_configured_timeout() { + struct TimeoutQuery; + + #[async_trait::async_trait] + impl KuboRpcQuery for TimeoutQuery { + type Response = (); + + fn route(&self) -> String { + "will_timeout".to_string() + } + + fn timeout(&self) -> Duration { + Duration::from_millis(10) + } + + async fn handle_success(&self, _response: Response) -> StdResult { + Ok(()) + } + } + + let (server, client) = setup_server_and_client(); + let _server_mock = server.mock(|when, then| { + when.any_request(); + then.delay(Duration::from_millis(100)); + }); + + let error = client.send(TimeoutQuery).await.unwrap_err(); + + assert!( + format!("{error:?}").contains("operation timed out"), + "Expected error message to contain 'operation timed out'\ngot '{error:?}'", + ) + } + + mod errors { + use super::*; + + macro_rules! assert_error_text_contains { + ($error: expr, $expect_contains: expr) => { + let error = &$error; + assert!( + error.contains($expect_contains), + "Expected error message to contain '{}'\ngot '{error:?}'", + $expect_contains, + ); + }; + } + + #[tokio::test] + async fn handle_json_errors() { + let json = serde_json::json!({"title": "error", "message":"an error"}); + let (server, client) = setup_server_and_client(); + server.mock(|when, then| { + when.any_request(); + then.status(400).json_body(json.clone()); + }); + + let err = client.send(QueryWithoutResponse).await.unwrap_err(); + assert_error_text_contains!(err.to_string(), &json.to_string()); + } + + #[tokio::test] + async fn handle_malformed_json() { + let malformed_json = r###"{"title": "error" "message":"an error"}"###; + let (server, client) = setup_server_and_client(); + server.mock(|when, then| { + when.any_request(); + then.status(400) + .body(malformed_json) + .header("Content-Type", "application/json"); + }); + + let err = client.send(QueryWithoutResponse).await.unwrap_err(); + assert_error_text_contains!(err.to_string(), &malformed_json.to_string()); + } + + #[tokio::test] + async fn handle_text_error() { + let (server, client) = setup_server_and_client(); + server.mock(|when, then| { + when.any_request(); + then.status(400).body("an error message"); + }); + + let err = client.send(QueryWithoutResponse).await.unwrap_err(); + assert_error_text_contains!(err.to_string(), ": 'an error message'"); + } + + #[tokio::test] + async fn handle_4xx_error() { + let (server, client) = setup_server_and_client(); + server.mock(|when, then| { + when.any_request(); + then.status(400).body("an error message"); + }); + + let err = client.send(QueryWithoutResponse).await.unwrap_err(); + assert_error_text_contains!( + err.to_string(), + "Request to Kubo RPC failed: 400 Bad Request:" + ); + } + + #[tokio::test] + async fn handle_5xx_error() { + let (server, client) = setup_server_and_client(); + server.mock(|when, then| { + when.any_request(); + then.status(500).body("an error message"); + }); + + let err = client.send(QueryWithoutResponse).await.unwrap_err(); + assert_error_text_contains!( + err.to_string(), + "Request to Kubo RPC failed: 500 Internal Server Error:" + ); + } + } +} diff --git a/mithril-aggregator/src/tools/kubo_rpc_client/mod.rs b/mithril-aggregator/src/tools/kubo_rpc_client/mod.rs new file mode 100644 index 00000000000..ce7ff234def --- /dev/null +++ b/mithril-aggregator/src/tools/kubo_rpc_client/mod.rs @@ -0,0 +1,27 @@ +mod api; +mod path; +pub mod query; + +pub use api::{KuboRpcClient, KuboRpcQuery}; +pub use path::IpfsMfsDirPath; + +#[cfg(test)] +mod test_tools { + use httpmock::MockServer; + + use crate::test::TestLogger; + use crate::tools::url_sanitizer::SanitizedUrlWithTrailingSlash; + + use super::KuboRpcClient; + + pub(super) fn setup_server_and_client() -> (MockServer, KuboRpcClient) { + let server = MockServer::start(); + let client = KuboRpcClient::new( + SanitizedUrlWithTrailingSlash::parse(&server.base_url()).unwrap(), + TestLogger::stdout(), + ) + .unwrap(); + + (server, client) + } +} diff --git a/mithril-aggregator/src/tools/kubo_rpc_client/path.rs b/mithril-aggregator/src/tools/kubo_rpc_client/path.rs new file mode 100644 index 00000000000..639ca6ad166 --- /dev/null +++ b/mithril-aggregator/src/tools/kubo_rpc_client/path.rs @@ -0,0 +1,154 @@ +use std::fmt::Display; +use std::path::Path; + +use anyhow::Context; + +use mithril_common::StdResult; + +/// A path to a Mutable File System directory in IPFS. +/// +/// It enforces that the path is absolute and has a trailing slash. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)] +#[serde(transparent)] +pub struct IpfsMfsDirPath(String); + +impl IpfsMfsDirPath { + /// Return a path for a file in this directory whose name is taken from `file_path`. + /// + /// Only the final component of `file_path` is used. Returns an error if the path + /// has no file name. + pub fn join_file_name_from>(&self, file_path: P) -> StdResult { + let filename = file_path + .as_ref() + .file_name() + .with_context(|| { + format!( + "Failed to get filename from path: {}", + file_path.as_ref().display() + ) + })? + .to_string_lossy(); + Ok(format!("{}{filename}", self.0)) + } +} + +impl Display for IpfsMfsDirPath { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.0) + } +} + +impl<'de> serde::Deserialize<'de> for IpfsMfsDirPath { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + let path = String::deserialize(deserializer)?; + Ok(Self::from(path)) + } +} + +impl AsRef for IpfsMfsDirPath { + fn as_ref(&self) -> &str { + &self.0 + } +} + +impl From for IpfsMfsDirPath { + /// Converts a [String] into an `IpfsMfsDirPath`. + /// + /// It ensures the following: + /// - The resulting path always starts with a '/' if it does not already. + /// - The resulting path always ends with a '/' if it does not already. + fn from(path: String) -> Self { + let mut path = path; + if !path.starts_with('/') { + path.insert(0, '/'); + } + + if !path.ends_with('/') { + path.push('/'); + } + + Self(path) + } +} + +impl From<&str> for IpfsMfsDirPath { + /// Converts a [str] into an `IpfsMfsDirPath`. + /// + /// It ensures the following: + /// - The resulting path always starts with a '/' if it does not already. + /// - The resulting path always ends with a '/' if it does not already. + fn from(value: &str) -> Self { + value.to_string().into() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn enforces_absolute_and_trailing_slash() { + assert_eq!("/", IpfsMfsDirPath::from("").as_ref()); + assert_eq!("/", IpfsMfsDirPath::from("/").as_ref()); + assert_eq!("/dir/", IpfsMfsDirPath::from("/dir").as_ref()); + assert_eq!("/dir/", IpfsMfsDirPath::from("/dir/").as_ref()); + assert_eq!("/dir/subdir/", IpfsMfsDirPath::from("/dir/subdir").as_ref()); + assert_eq!( + "/dir/subdir/", + IpfsMfsDirPath::from("/dir/subdir/").as_ref() + ); + } + + #[test] + fn deserializing_enforces_absolute_and_trailing_slash() { + assert_eq!( + IpfsMfsDirPath::from("/"), + serde_json::from_str(r#""""#).unwrap() + ); + assert_eq!( + IpfsMfsDirPath::from("/"), + serde_json::from_str(r#""/""#).unwrap(), + ); + assert_eq!( + IpfsMfsDirPath::from("/dir/"), + serde_json::from_str(r#""/dir""#).unwrap(), + ); + assert_eq!( + IpfsMfsDirPath::from("/dir/"), + serde_json::from_str(r#""/dir/""#).unwrap(), + ); + assert_eq!( + IpfsMfsDirPath::from("/dir with spaces/"), + serde_json::from_str(r#""/dir with spaces/""#).unwrap() + ); + assert_eq!( + IpfsMfsDirPath::from("/dir/subdir/"), + serde_json::from_str(r#""/dir/subdir""#).unwrap() + ); + assert_eq!( + IpfsMfsDirPath::from("/dir/subdir/"), + serde_json::from_str(r#""/dir/subdir/""#).unwrap() + ); + } + + #[test] + fn join_file_name_from_builds_mfs_path_from_directory_and_file_name() { + let path = IpfsMfsDirPath::from("/test/dir") + .join_file_name_from("/local/archive/dummy-file.txt") + .unwrap(); + + assert_eq!("/test/dir/dummy-file.txt", path); + } + + #[test] + fn join_file_name_from_returns_error_when_path_has_no_file_name() { + let error = IpfsMfsDirPath::from("/test/dir") + .join_file_name_from(Path::new("")) + .unwrap_err(); + + assert!(error.to_string().contains("Failed to get filename from path")); + } +} diff --git a/mithril-aggregator/src/tools/kubo_rpc_client/query/ipfs_add.rs b/mithril-aggregator/src/tools/kubo_rpc_client/query/ipfs_add.rs new file mode 100644 index 00000000000..60ad7b78312 --- /dev/null +++ b/mithril-aggregator/src/tools/kubo_rpc_client/query/ipfs_add.rs @@ -0,0 +1,325 @@ +use std::path::{Path, PathBuf}; +use std::time::Duration; + +use anyhow::Context; +use reqwest::{RequestBuilder, Response}; +use serde::Deserialize; + +use mithril_common::StdResult; + +use crate::tools::kubo_rpc_client::{IpfsMfsDirPath, KuboRpcQuery}; + +/// Query to add a file to IPFS via the Kubo RPC API. +/// +/// see: https://docs.ipfs.tech/reference/kubo/rpc/#api-v0-add +#[derive(Debug)] +pub struct IpfsAddQuery { + file_path: PathBuf, + to_files: Option, + enable_no_copy: bool, +} + +/// Response from the IPFS add operation. +#[derive(Debug, Clone, PartialEq, Eq, Deserialize)] +#[serde(rename_all = "PascalCase")] +pub struct IpfsAddResponse { + /// Name of the added file + pub name: String, + /// Hash of the added file (CID) + pub hash: String, +} + +impl IpfsAddQuery { + const BASE_TIMEOUT_SECS: u64 = 10; + const MAX_TIMEOUT_SECS: u64 = 3 * 60; + + /// Create a query that will add the given file to IPFS. + #[cfg(test)] + pub fn new>(file_path: P) -> Self { + Self { + file_path: file_path.as_ref().to_path_buf(), + to_files: None, + enable_no_copy: false, + } + } + + /// Create a query that will add the given file to IPFS and reference it in the MFS. + pub fn new_with_mfs_reference>( + file_path: P1, + mfs_path: &IpfsMfsDirPath, + ) -> Self { + Self { + file_path: file_path.as_ref().to_path_buf(), + to_files: Some(mfs_path.clone()), + enable_no_copy: false, + } + } + + /// Tells IPFS to not copy the file to its internal storage, but instead to reference it directly. + /// + /// **Test only** until we properly implement this feature (we must symlink the file into the + /// IPFS root directory and set the `abspath` multipart header to the symlinked path, else the + /// request will be rejected). + #[cfg(test)] + pub fn no_copy(mut self) -> Self { + self.enable_no_copy = true; + self + } + + /// Calculates the request timeout from the file size. + /// + /// The timeout starts at 10 seconds and increases by one second for every started MiB, up to + /// a maximum of three minutes. + /// If the file size cannot be determined, the maximum timeout is used. + /// + /// Note: expected file size in production for immutables archive is 1-100 MiB (mostly 2-20 Mib). + fn timeout_duration(file_size_in_bytes: Option) -> Duration { + const BYTES_PER_MIB: u64 = 1024 * 1024; + + match file_size_in_bytes { + None => Duration::from_secs(Self::MAX_TIMEOUT_SECS), + Some(size_in_bytes) => { + let file_size_mib = size_in_bytes.div_ceil(BYTES_PER_MIB); + let timeout_secs = Self::BASE_TIMEOUT_SECS + .saturating_add(file_size_mib) + .min(Self::MAX_TIMEOUT_SECS); + + Duration::from_secs(timeout_secs) + } + } + } +} + +#[async_trait::async_trait] +impl KuboRpcQuery for IpfsAddQuery { + type Response = IpfsAddResponse; + + fn route(&self) -> String { + "api/v0/add".to_string() + } + + async fn configure_request( + &self, + mut request_builder: RequestBuilder, + ) -> StdResult { + let mut part = reqwest::multipart::Part::file(&self.file_path).await?; + + if let Some(mfs_path) = &self.to_files { + request_builder = request_builder.query(&[("to-files", mfs_path)]); + } + + if self.enable_no_copy { + request_builder = request_builder.query(&[("nocopy", true)]); + + // Add the "abspath" header to the file part + // It sets the absolute path of the file being added to IPFS and MUST point to a path + // inside the IPFS root directory. + // + // It may be different from the file path provided, e.g.: file_path could point to the + // real location of the file to be added while `abspath` points to the symlinked path of + // the same file but inside the IPFS root directory. + let mut part_headers = reqwest::header::HeaderMap::new(); + let abspath = std::path::absolute(&self.file_path) + .with_context(|| "Failed to get absolute path of file")?; + part_headers.insert( + "abspath", + abspath + .to_str() + .with_context(|| "Failed to convert absolute file path to string")? + .try_into()?, + ); + part = part.headers(part_headers); + } + + let request_builder = request_builder + .multipart(reqwest::multipart::Form::new().part("file", part)) + .query(&[("pin", true)]) + // Parameters below ensure deterministic CID. + .query(&[ + ("inline", false), + ("preserve-mode", false), + ("preserve-mtime", false), + ("raw-leaves", true), + ("trickle", false), + ("wrap-with-directory", false), + ]) + .query(&[("cid-version", 1)]) + .query(&[("hash", "sha2-256"), ("chunker", "size-262144")]); + + Ok(request_builder) + } + + fn timeout(&self) -> Duration { + let file_size_in_bytes = self.file_path.metadata().map(|m| m.len()).ok(); + Self::timeout_duration(file_size_in_bytes) + } + + async fn handle_success(&self, response: Response) -> StdResult { + response + .json() + .await + .with_context(|| "Failed to deserialize IPFS add response") + } +} + +#[cfg(test)] +mod tests { + use httpmock::Method::POST; + + use mithril_common::temp_dir_create; + + use crate::tools::kubo_rpc_client::test_tools::setup_server_and_client; + + use super::*; + + #[tokio::test] + async fn return_add_data_if_request_succeeds() { + let test_dir = temp_dir_create!(); + let file = test_dir.join("test.txt"); + std::fs::File::create(&file).unwrap(); + + let (server, client) = setup_server_and_client(); + server.mock(|when, then| { + when.method(POST).path("/api/v0/add").query_param("pin", "true"); + then.status(200).json_body(serde_json::json!({"Name":"test.txt","Hash":"QmYi7wrRFKVCcTB56A6Pep2j31Q5mHfmmu21RzHXu25RVR","Size":"23"})); + }); + + let response = client.send(IpfsAddQuery::new(file)).await.unwrap(); + assert_eq!( + IpfsAddResponse { + name: "test.txt".to_string(), + hash: "QmYi7wrRFKVCcTB56A6Pep2j31Q5mHfmmu21RzHXu25RVR".to_string(), + }, + response + ); + } + + #[tokio::test] + async fn return_error_if_request_fails() { + let test_dir = temp_dir_create!(); + let file = test_dir.join("test.txt"); + std::fs::File::create(&file).unwrap(); + + let (server, client) = setup_server_and_client(); + server.mock(|when, then| { + when.method(POST).path("/api/v0/add"); + then.status(500).json_body( + serde_json::json!({"Message":"paths must start with a leading slash","Code":0,"Type":"error"}), + ); + }); + + let err = client.send(IpfsAddQuery::new(file)).await.unwrap_err(); + assert!( + err.to_string().contains("paths must start with a leading slash"), + "unexpected error: {err}" + ); + } + + #[tokio::test] + async fn make_produced_cid_more_deterministic_by_enforcing_parameters_that_may_affect_cid() { + let test_dir = temp_dir_create!(); + let file = test_dir.join("test.txt"); + std::fs::File::create(&file).unwrap(); + + let (server, client) = setup_server_and_client(); + server.mock(|when, then| { + when.method(POST).path("/api/v0/add") + .query_param("cid-version", "1") + .query_param("hash", "sha2-256") + .query_param("raw-leaves", "true") + .query_param("chunker", "size-262144") + .query_param("inline", "false") + .query_param("trickle", "false") + .query_param("preserve-mode", "false") + .query_param("preserve-mtime", "false") + .query_param("wrap-with-directory", "false"); + then.status(200).json_body(serde_json::json!({"Name":"test.txt","Hash":"QmYi7wrRFKVCcTB56A6Pep2j31Q5mHfmmu21RzHXu25RVR","Size":"23"})); + }); + + client.send(IpfsAddQuery::new(file)).await.unwrap(); + } + + #[tokio::test] + async fn no_copy_adds_query_parameter_and_abspath_multipart_header() { + let test_dir = temp_dir_create!(); + let file = test_dir.join("test.txt"); + std::fs::File::create(&file).unwrap(); + + let abs_filepath = file.canonicalize().unwrap().to_string_lossy().to_string(); + + let (server, client) = setup_server_and_client(); + server.mock(|when, then| { + when.method(POST) + .path("/api/v0/add") + .query_param("nocopy", "true") + .body_includes(format!("abspath: {abs_filepath}")); + then.status(200) + .json_body(serde_json::json!({"Name":"test.txt","Hash":"whatever"})); + }); + + client.send(IpfsAddQuery::new(file).no_copy()).await.unwrap(); + } + + mod timeout { + use super::*; + + const MIB: u64 = 1024 * 1024; + + #[test] + fn timeout_duration_returns_maximum_when_file_size_is_unknown() { + assert_eq!( + Duration::from_secs(180), + IpfsAddQuery::timeout_duration(None) + ); + } + + #[test] + fn timeout_duration_adds_one_second_for_each_started_mib() { + let cases = [ + (0, IpfsAddQuery::BASE_TIMEOUT_SECS), + (1, 11), + (MIB, 11), + (MIB + 1, 12), + (50 * MIB, 60), + (100 * MIB, 110), + ]; + + for (file_size, expected_timeout_secs) in cases { + assert_eq!( + Duration::from_secs(expected_timeout_secs), + IpfsAddQuery::timeout_duration(Some(file_size)), + "unexpected timeout for file size {file_size}" + ); + } + } + + #[test] + fn timeout_duration_is_capped_at_three_minutes() { + let cases = [ + (169 * MIB, 179), + (169 * MIB + 1, IpfsAddQuery::MAX_TIMEOUT_SECS), + (170 * MIB, IpfsAddQuery::MAX_TIMEOUT_SECS), + (171 * MIB, IpfsAddQuery::MAX_TIMEOUT_SECS), + (u64::MAX, IpfsAddQuery::MAX_TIMEOUT_SECS), + ]; + + for (file_size, expected_timeout_secs) in cases { + assert_eq!( + Duration::from_secs(expected_timeout_secs), + IpfsAddQuery::timeout_duration(Some(file_size)), + "unexpected timeout for file size {file_size}" + ); + } + } + + #[test] + fn timeout_is_based_on_the_uploaded_file_size() { + let test_dir = temp_dir_create!(); + let file_path = test_dir.join("43-mib-file.bin"); + std::fs::File::create(&file_path).unwrap().set_len(43 * MIB).unwrap(); + + let query = IpfsAddQuery::new(file_path); + assert_eq!(Duration::from_secs(53), query.timeout()); + } + } +} diff --git a/mithril-aggregator/src/tools/kubo_rpc_client/query/ipfs_files_ls.rs b/mithril-aggregator/src/tools/kubo_rpc_client/query/ipfs_files_ls.rs new file mode 100644 index 00000000000..151c2ad1cea --- /dev/null +++ b/mithril-aggregator/src/tools/kubo_rpc_client/query/ipfs_files_ls.rs @@ -0,0 +1,203 @@ +use anyhow::Context; +use reqwest::{RequestBuilder, Response}; +use serde::Deserialize; +use std::collections::HashMap; +use std::time::Duration; + +use mithril_common::StdResult; + +use crate::tools::kubo_rpc_client::api::handle_file_not_exist_error; +use crate::tools::kubo_rpc_client::{IpfsMfsDirPath, KuboRpcQuery}; + +/// Query to list directories in an MFS (Mutable File System) in IPFS via the Kubo RPC API. +/// +/// Returns a map of file names to their hashes / CIDs. +/// +/// see: https://docs.ipfs.tech/reference/kubo/rpc/#api-v0-files-ls +#[derive(Debug)] +pub struct IpfsFilesLsQuery { + dir: IpfsMfsDirPath, +} + +#[derive(Debug, Clone, PartialEq, Eq, Deserialize)] +#[serde(rename_all = "PascalCase")] +struct IpfsLsResponse { + entries: Option>, +} + +#[derive(Debug, Clone, PartialEq, Eq, Deserialize)] +#[serde(rename_all = "PascalCase")] +struct IpfsLsResponseItem { + /// Name of the added file + name: String, + /// Hash of the added file (CID) + hash: String, +} + +impl IpfsFilesLsQuery { + /// Create a query that will get the list a directory in the MFS. + pub fn new(mfs_dir: &IpfsMfsDirPath) -> Self { + Self { + dir: mfs_dir.clone(), + } + } +} + +#[async_trait::async_trait] +impl KuboRpcQuery for IpfsFilesLsQuery { + type Response = Option>; + + fn route(&self) -> String { + "api/v0/files/ls".to_string() + } + + async fn configure_request( + &self, + request_builder: RequestBuilder, + ) -> StdResult { + Ok(request_builder + .query(&[("arg", &self.dir)]) + // enable long listing (else hashes are empty) and disable sorting (handled rust-side) + .query(&[("long", "true"), ("U", "true")])) + } + + fn timeout(&self) -> Duration { + Duration::from_secs(10) + } + + async fn handle_success(&self, response: Response) -> StdResult { + let response: IpfsLsResponse = response + .json() + .await + .with_context(|| "Failed to deserialize IPFS ls response")?; + + match response.entries { + Some(entries) => Ok(Some( + entries.into_iter().map(|item| (item.name, item.hash)).collect(), + )), + // If entries are null, the directory exists but is empty + None => Ok(Some(HashMap::::new())), + } + } + + async fn handle_error(&self, response: Response) -> StdResult { + handle_file_not_exist_error("files ls", response).await + } +} + +#[cfg(test)] +mod tests { + use httpmock::Method::POST; + + use crate::tools::kubo_rpc_client::test_tools::setup_server_and_client; + + use super::*; + + #[tokio::test] + async fn return_empty_list_when_directory_is_empty() { + let (server, client) = setup_server_and_client(); + server.mock(|when, then| { + when.method(POST) + .path("/api/v0/files/ls") + .query_param("arg", "/test/") + .query_param("long", "true") + .query_param("U", "true"); + // Kubo returns null if the directory exists but is empty + then.status(200).json_body(serde_json::json!({ "Entries": null })); + }); + + let response = client + .send(IpfsFilesLsQuery::new(&IpfsMfsDirPath::from("/test"))) + .await + .unwrap(); + assert_eq!(Some(HashMap::::new()), response); + } + + #[tokio::test] + async fn return_items_list_if_request_succeeds() { + let (server, client) = setup_server_and_client(); + server.mock(|when, then| { + when.method(POST) + .path("/api/v0/files/ls") + .query_param("arg", "/test/") + .query_param("long", "true") + .query_param("U", "true"); + then.status(200).json_body(serde_json::json!({ + "Entries":[ + {"Name":"00000.tar.zst","Type":0,"Size":28486,"Hash":"QmePDH8sb7dux6VEvACJYS3m76D4Cc8eyfhejs7wcDFwWi"}, + {"Name":"00001.tar.zst","Type":0,"Size":28557,"Hash":"QmXtTUpZervXkza1KmmmnfkxqJmLxZAsRXUfVKQHPBBkFA"}, + {"Name":"00002.tar.zst","Type":0,"Size":29518,"Hash":"QmYd4yX3ms9jeLcDd9r3DZaMYuKb7TNX5dL1VD1wyNgKac"}, + {"Name":"00003.tar.zst","Type":0,"Size":28951,"Hash":"Qmbh4AHrNT8GMrJYLAyku88zhoZAyJcsRRyXNFbDJsbCp9"}, + {"Name":"sub-dir","Type":1,"Size":0,"Hash":"QmX5UvqhAYnEqAGx41SCovCg4x6NTF5XEVBMLarqk8J4x7"} + ] + })); + }); + + let response = client + .send(IpfsFilesLsQuery::new(&IpfsMfsDirPath::from("/test"))) + .await + .unwrap(); + + assert_eq!( + Some(HashMap::::from([ + ( + "00000.tar.zst".to_string(), + "QmePDH8sb7dux6VEvACJYS3m76D4Cc8eyfhejs7wcDFwWi".to_string(), + ), + ( + "00001.tar.zst".to_string(), + "QmXtTUpZervXkza1KmmmnfkxqJmLxZAsRXUfVKQHPBBkFA".to_string(), + ), + ( + "00002.tar.zst".to_string(), + "QmYd4yX3ms9jeLcDd9r3DZaMYuKb7TNX5dL1VD1wyNgKac".to_string(), + ), + ( + "00003.tar.zst".to_string(), + "Qmbh4AHrNT8GMrJYLAyku88zhoZAyJcsRRyXNFbDJsbCp9".to_string(), + ), + ( + "sub-dir".to_string(), + "QmX5UvqhAYnEqAGx41SCovCg4x6NTF5XEVBMLarqk8J4x7".to_string(), + ), + ])), + response + ); + } + + #[tokio::test] + async fn return_none_if_request_fails_with_not_exist_message() { + let (server, client) = setup_server_and_client(); + server.mock(|when, then| { + when.method(POST) + .path("/api/v0/files/ls") + .query_param("arg", "/test/"); + then.status(500).json_body( + serde_json::json!({"Message":"file does not exist","Code":0,"Type":"error"}), + ); + }); + + let response = client + .send(IpfsFilesLsQuery::new(&IpfsMfsDirPath::from("/test"))) + .await + .unwrap(); + assert_eq!(None, response); + } + + #[tokio::test] + async fn return_error_if_request_fails_with_other_message() { + let (server, client) = setup_server_and_client(); + server.mock(|when, then| { + when.method(POST).path("/api/v0/files/ls").query_param("arg", "/test/"); + then.status(500).json_body( + serde_json::json!({"Message":"paths must start with a leading slash","Code":0,"Type":"error"}), + ); + }); + + let err = client + .send(IpfsFilesLsQuery::new(&IpfsMfsDirPath::from("/test"))) + .await + .unwrap_err(); + assert!(err.to_string().contains("paths must start with a leading slash")); + } +} diff --git a/mithril-aggregator/src/tools/kubo_rpc_client/query/ipfs_files_mkdir.rs b/mithril-aggregator/src/tools/kubo_rpc_client/query/ipfs_files_mkdir.rs new file mode 100644 index 00000000000..e745bfea9a0 --- /dev/null +++ b/mithril-aggregator/src/tools/kubo_rpc_client/query/ipfs_files_mkdir.rs @@ -0,0 +1,95 @@ +use reqwest::{RequestBuilder, Response}; + +use mithril_common::StdResult; + +use crate::tools::kubo_rpc_client::{IpfsMfsDirPath, KuboRpcQuery}; + +/// Query to make an MFS (Mutable File System) directory in IPFS via the Kubo RPC API. +/// +/// Note: `parents` flag is set to true by default; this has two effects: +/// - parent directories will be created if they do not exist. +/// - the command will succeed even if the directory already exists. +/// +/// see: https://docs.ipfs.tech/reference/kubo/rpc/#api-v0-files-mkdir +#[derive(Debug)] +pub struct IpfsFilesMkdirQuery { + ipfs_absolute_path: IpfsMfsDirPath, +} + +impl IpfsFilesMkdirQuery { + /// Create a query that will create the given IPFS absolute path as an MFS directory. + pub fn create_mfs_directory(ipfs_absolute_path: &IpfsMfsDirPath) -> Self { + Self { + ipfs_absolute_path: ipfs_absolute_path.clone(), + } + } +} + +#[async_trait::async_trait] +impl KuboRpcQuery for IpfsFilesMkdirQuery { + type Response = (); + + fn route(&self) -> String { + "api/v0/files/mkdir".to_string() + } + + async fn configure_request( + &self, + request_builder: RequestBuilder, + ) -> StdResult { + Ok(request_builder + .query(&[("arg", &self.ipfs_absolute_path)]) + .query(&[("parents", true)])) + } + + async fn handle_success(&self, _response: Response) -> StdResult { + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use httpmock::Method::POST; + + use crate::tools::kubo_rpc_client::test_tools::setup_server_and_client; + + use super::*; + + #[tokio::test] + async fn succeeds_when_server_returns_200() { + let (server, client) = setup_server_and_client(); + server.mock(|when, then| { + when.method(POST) + .path("/api/v0/files/mkdir") + .query_param("arg", "/test/") + .query_param("parents", "true"); + then.status(200); + }); + + client + .send(IpfsFilesMkdirQuery::create_mfs_directory( + &IpfsMfsDirPath::from("/test"), + )) + .await + .unwrap(); + } + + #[tokio::test] + async fn return_error_if_request_fails_with_other_message() { + let (server, client) = setup_server_and_client(); + server.mock(|when, then| { + when.method(POST).path("/api/v0/files/mkdir"); + then.status(500).json_body( + serde_json::json!({"Message":"paths must start with a leading slash","Code":0,"Type":"error"}), + ); + }); + + let err = client + .send(IpfsFilesMkdirQuery::create_mfs_directory( + &IpfsMfsDirPath::from("/test"), + )) + .await + .unwrap_err(); + assert!(err.to_string().contains("paths must start with a leading slash")); + } +} diff --git a/mithril-aggregator/src/tools/kubo_rpc_client/query/ipfs_files_stat.rs b/mithril-aggregator/src/tools/kubo_rpc_client/query/ipfs_files_stat.rs new file mode 100644 index 00000000000..6c3ebd73e64 --- /dev/null +++ b/mithril-aggregator/src/tools/kubo_rpc_client/query/ipfs_files_stat.rs @@ -0,0 +1,136 @@ +use anyhow::Context; +use reqwest::{RequestBuilder, Response}; +use serde::Deserialize; + +use mithril_common::StdResult; + +use crate::tools::kubo_rpc_client::KuboRpcQuery; +use crate::tools::kubo_rpc_client::api::handle_file_not_exist_error; + +/// Query to display file status in an MFS (Mutable File System) in IPFS via the Kubo RPC API. +/// +/// see: https://docs.ipfs.tech/reference/kubo/rpc/#api-v0-files-stat +#[derive(Debug)] +pub struct IpfsFilesStatQuery { + path_in_ipfs: String, +} + +/// Response from the IPFS files stat operation. +#[derive(Debug, Clone, PartialEq, Eq, Deserialize)] +#[serde(rename_all = "PascalCase")] +pub struct IpfsStatResponse { + /// Hash (CID) of the file + pub hash: String, + /// Size of the file in bytes + pub size: u64, + /// Cumulative size including blocks + pub cumulative_size: u64, + /// Type of the IPFS object + pub r#type: MfsStatType, +} + +/// Type of MFS (Mutable File System) entry in IPFS. +#[derive(Debug, Clone, PartialEq, Eq, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum MfsStatType { + /// A file entry + File, + /// A directory entry + Directory, +} + +impl IpfsFilesStatQuery { + /// Create a query that will get the status of a file in the MFS. + pub fn new>(path_in_ipfs: P) -> Self { + Self { + path_in_ipfs: path_in_ipfs.as_ref().to_string(), + } + } +} + +#[async_trait::async_trait] +impl KuboRpcQuery for IpfsFilesStatQuery { + type Response = Option; + + fn route(&self) -> String { + "api/v0/files/stat".to_string() + } + + async fn configure_request( + &self, + request_builder: RequestBuilder, + ) -> StdResult { + Ok(request_builder.query(&[("arg", &self.path_in_ipfs)])) + } + + async fn handle_success(&self, response: Response) -> StdResult { + response + .json() + .await + .map(Some) + .with_context(|| "Failed to deserialize IPFS stat response") + } + + async fn handle_error(&self, response: Response) -> StdResult { + handle_file_not_exist_error("files stat", response).await + } +} + +#[cfg(test)] +mod tests { + use httpmock::Method::POST; + + use crate::tools::kubo_rpc_client::test_tools::setup_server_and_client; + + use super::*; + + #[tokio::test] + async fn return_stat_data_if_request_succeeds() { + let (server, client) = setup_server_and_client(); + server.mock(|when, then| { + when.method(POST).path("/api/v0/files/stat").query_param("arg", "/test"); + then.status(200).json_body(serde_json::json!({"Hash": "QmHash", "Size": 1, "CumulativeSize": 2, "Type": "file"})); + }); + + let response = client.send(IpfsFilesStatQuery::new("/test")).await.unwrap(); + assert_eq!( + Some(IpfsStatResponse { + hash: "QmHash".to_string(), + size: 1, + cumulative_size: 2, + r#type: MfsStatType::File + }), + response + ); + } + + #[tokio::test] + async fn return_none_if_request_fails_with_not_exist_message() { + let (server, client) = setup_server_and_client(); + server.mock(|when, then| { + when.method(POST) + .path("/api/v0/files/stat") + .query_param("arg", "/test"); + then.status(500).json_body( + serde_json::json!({"Message":"file does not exist","Code":0,"Type":"error"}), + ); + }); + + let response = client.send(IpfsFilesStatQuery::new("/test")).await.unwrap(); + assert_eq!(None, response); + } + + #[tokio::test] + async fn return_error_if_request_fails_with_other_message() { + let (server, client) = setup_server_and_client(); + server.mock(|when, then| { + when.method(POST).path("/api/v0/files/stat").query_param("arg", "/test"); + then.status(500).json_body( + serde_json::json!({"Message":"paths must start with a leading slash","Code":0,"Type":"error"}), + ); + }); + + let err = client.send(IpfsFilesStatQuery::new("/test")).await.unwrap_err(); + assert!(err.to_string().contains("paths must start with a leading slash")); + } +} diff --git a/mithril-aggregator/src/tools/kubo_rpc_client/query/mod.rs b/mithril-aggregator/src/tools/kubo_rpc_client/query/mod.rs new file mode 100644 index 00000000000..42e3dfc97a8 --- /dev/null +++ b/mithril-aggregator/src/tools/kubo_rpc_client/query/mod.rs @@ -0,0 +1,9 @@ +mod ipfs_add; +mod ipfs_files_ls; +mod ipfs_files_mkdir; +mod ipfs_files_stat; + +pub use ipfs_add::*; +pub use ipfs_files_ls::*; +pub use ipfs_files_mkdir::*; +pub use ipfs_files_stat::*; diff --git a/mithril-aggregator/src/tools/mod.rs b/mithril-aggregator/src/tools/mod.rs index 93b3777bdde..6387f9b2c70 100644 --- a/mithril-aggregator/src/tools/mod.rs +++ b/mithril-aggregator/src/tools/mod.rs @@ -1,6 +1,7 @@ mod certificates_hash_migrator; mod era; mod genesis; +pub mod kubo_rpc_client; mod protocol_configuration; pub mod signer_importer; mod single_signature_authenticator; diff --git a/mithril-client-cli/Cargo.toml b/mithril-client-cli/Cargo.toml index 0f511c8704e..96799e17658 100644 --- a/mithril-client-cli/Cargo.toml +++ b/mithril-client-cli/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mithril-client-cli" -version = "0.13.21" +version = "0.13.22" description = "A Mithril Client" authors = { workspace = true } edition = { workspace = true } diff --git a/mithril-client-cli/src/commands/cardano_db/show.rs b/mithril-client-cli/src/commands/cardano_db/show.rs index 8d0391c0d23..c6d7e12b18f 100644 --- a/mithril-client-cli/src/commands/cardano_db/show.rs +++ b/mithril-client-cli/src/commands/cardano_db/show.rs @@ -141,6 +141,14 @@ fn immutables_location_iter( template_uri.0 )), }, + ImmutablesLocation::Ipfs { + uri, + compression_algorithm: _, + } => match uri { + MultiFilesUri::Template(template_uri) => { + Some(format!("IPFS, template_uri: \"{}\"", template_uri.0)) + } + }, ImmutablesLocation::Unknown => None, }) } diff --git a/mithril-client/Cargo.toml b/mithril-client/Cargo.toml index e326fe8ba20..12e98a052fa 100644 --- a/mithril-client/Cargo.toml +++ b/mithril-client/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mithril-client" -version = "0.14.19" +version = "0.14.20" description = "Mithril client library" authors = { workspace = true } edition = { workspace = true } @@ -63,7 +63,7 @@ chrono = { workspace = true } flume = { version = "0.12.0", optional = true } futures = "0.3.32" mithril-aggregator-client = { path = "../internal/mithril-aggregator-client", version = "0.2.4" } -mithril-common = { path = "../mithril-common", version = "0.7.18", default-features = false } +mithril-common = { path = "../mithril-common", version = "0.7.19", default-features = false } reqwest = { workspace = true, default-features = false, features = ["charset", "http2", "stream", "system-proxy"] } serde = { workspace = true } serde_json = { workspace = true } diff --git a/mithril-client/src/cardano_database_client/download_unpack/internal_downloader.rs b/mithril-client/src/cardano_database_client/download_unpack/internal_downloader.rs index 0d8aea5f218..497555a7b7e 100644 --- a/mithril-client/src/cardano_database_client/download_unpack/internal_downloader.rs +++ b/mithril-client/src/cardano_database_client/download_unpack/internal_downloader.rs @@ -173,7 +173,8 @@ impl InternalArtifactDownloader { } } // Note: unknown locations should have been filtered out by `sanitized_locations` - ImmutablesLocation::Unknown => unreachable!(), + // IPFS locations are excluded as we do not have a downloader for them yet + ImmutablesLocation::Ipfs { .. } | ImmutablesLocation::Unknown => unreachable!(), }; locations_to_try.push(location_to_try); diff --git a/mithril-common/Cargo.toml b/mithril-common/Cargo.toml index aa42ed99055..ef0bb7b95c6 100644 --- a/mithril-common/Cargo.toml +++ b/mithril-common/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mithril-common" -version = "0.7.18" +version = "0.7.19" description = "Common types, interfaces, and utilities for Mithril nodes." authors = { workspace = true } edition = { workspace = true } diff --git a/mithril-common/src/entities/cardano_database.rs b/mithril-common/src/entities/cardano_database.rs index 0a2ba932747..df42c36b490 100644 --- a/mithril-common/src/entities/cardano_database.rs +++ b/mithril-common/src/entities/cardano_database.rs @@ -124,6 +124,15 @@ pub enum ImmutablesLocation { #[serde(skip_serializing_if = "Option::is_none")] compression_algorithm: Option, }, + /// IPFS location. + Ipfs { + /// URI of the IPFS location. + uri: MultiFilesUri, + + /// Compression algorithm of the Cardano database artifacts. + #[serde(skip_serializing_if = "Option::is_none")] + compression_algorithm: Option, + }, /// Catchall for unknown location variants. #[serde(other)] Unknown, @@ -322,7 +331,8 @@ mod tests { } #[test] - fn should_not_display_compression_algorithm_in_json_immutable_location_when_none() { + fn should_not_display_compression_algorithm_in_json_immutable_cloud_storage_location_when_none() + { let json = serde_json::json!(ImmutablesLocation::CloudStorage { uri: MultiFilesUri::Template(TemplateUri("https://example.com".to_string())), compression_algorithm: None, @@ -332,4 +342,16 @@ mod tests { r#"{"type":"cloud_storage","uri":{"Template":"https://example.com"}}"# ); } + + #[test] + fn should_not_display_compression_algorithm_in_json_immutable_ipfs_location_when_none() { + let json = serde_json::json!(ImmutablesLocation::Ipfs { + uri: MultiFilesUri::Template(TemplateUri("https://example.com".to_string())), + compression_algorithm: None, + }); + assert_eq!( + json.to_string(), + r#"{"type":"ipfs","uri":{"Template":"https://example.com"}}"# + ); + } } diff --git a/mithril-common/src/messages/cardano_database.rs b/mithril-common/src/messages/cardano_database.rs index 1874f23254f..44bba566f0a 100644 --- a/mithril-common/src/messages/cardano_database.rs +++ b/mithril-common/src/messages/cardano_database.rs @@ -52,7 +52,13 @@ impl ImmutablesMessagePart { let sanitized_locations: Vec<_> = self .locations .iter() - .filter(|l| !matches!(l, ImmutablesLocation::Unknown)) + .filter(|l| { + !matches!( + l, + // Temporarily exclude IPFS locations until the download is implemented. + ImmutablesLocation::Unknown | ImmutablesLocation::Ipfs { .. } + ) + }) .cloned() .collect(); @@ -197,6 +203,12 @@ mod tests { "uri": { "Template": "https://host-2/immutables-{immutable_file_number}" } + }, + { + "type": "ipfs", + "uri": { + "Template": "QmSiDJUe8MsPRwbAJuT5PtRmxXLJVsUK1j2ZSN3SKkAN6Z/{immutable_file_number}" + } } ] }, @@ -251,6 +263,12 @@ mod tests { )), compression_algorithm: None, }, + ImmutablesLocation::Ipfs { + uri: MultiFilesUri::Template(TemplateUri( + "QmSiDJUe8MsPRwbAJuT5PtRmxXLJVsUK1j2ZSN3SKkAN6Z/{immutable_file_number}".to_string(), + )), + compression_algorithm: None, + }, ], }, ancillary: AncillaryMessagePart { @@ -391,6 +409,22 @@ mod tests { .sanitized_locations() .expect_err("Should fail since all locations are unknown."); } + + // Temporarily exclude IPFS locations until the download is implemented. + #[test] + fn fails_if_all_locations_are_ipfs() { + ImmutablesMessagePart { + locations: vec![ImmutablesLocation::Ipfs { + uri: MultiFilesUri::Template(TemplateUri( + "Whatever_CID/{immutable_file_number}.tar.zst".to_string(), + )), + compression_algorithm: None, + }], + average_size_uncompressed: 512, + } + .sanitized_locations() + .expect_err("Should fail since all locations are IPFS."); + } } mod sanitize_ancillary_locations {