feat(aggregator): ipfs file uploader

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.

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"

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 }

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 }

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 }

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
                     );
                 }
             }

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
                     );
                 }
             }

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<CompressionAlgorithm>,
+    ) -> StdResult<ImmutablesLocation> {
+        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
                     );
                 }
             }

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<IpfsRpcServerConfig> {
+        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<IpfsRpcServerConfig>,
+
     /// 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> {
+        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<Self, Self::Err> {
+        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<IpfsRpcServerConfig> {
+        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 =

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<IpfsUploader> {
+        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<Vec<Arc<dyn AncillaryFileUploader>>> {
@@ -301,28 +317,36 @@ impl DependenciesBuilder {
     ) -> Result<Vec<Arc<dyn ImmutableFilesUploader>>> {
         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<Arc<dyn ImmutableFilesUploader>> =
+                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(),