diff --git a/Cargo.lock b/Cargo.lock index 0e14eec1597..38bfc448f0d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4351,7 +4351,7 @@ dependencies = [ [[package]] name = "mithril-aggregator" -version = "0.10.20" +version = "0.10.21" dependencies = [ "anyhow", "async-trait", @@ -4764,7 +4764,7 @@ dependencies = [ [[package]] name = "mithril-end-to-end" -version = "0.5.26" +version = "0.5.27" dependencies = [ "anyhow", "async-trait", diff --git a/docs/website/root/manual/develop/nodes/mithril-aggregator.md b/docs/website/root/manual/develop/nodes/mithril-aggregator.md index 2f736e9bb32..48998d4629b 100644 --- a/docs/website/root/manual/develop/nodes/mithril-aggregator.md +++ b/docs/website/root/manual/develop/nodes/mithril-aggregator.md @@ -588,6 +588,7 @@ Here is a list of the available parameters for the serve command: | `cardano_prover_max_hashes_allowed_by_request` | `--cardano-prover-max-hashes-allowed-by-request` | - | `CARDANO_PROVER_MAX_HASHES_ALLOWED_BY_REQUEST` | Maximum number of hashes allowed by request to the Cardano prover (applies to both transaction hashes and block hashes) | `100` | `100` | - | | `cardano_transactions_block_streamer_max_roll_forwards_per_poll` | `--cardano-transactions-block-streamer-max-roll-forwards-per-poll` | - | `CARDANO_TRANSACTIONS_BLOCK_STREAMER_MAX_ROLL_FORWARDS_PER_POLL` | Maximum number of roll forwards during a poll of the block streamer when importing transactions | `1000` | `1000` | - | | `preload_security_parameter` | - | - | `PRELOAD_SECURITY_PARAMETER` | Blocks offset, from the tip of the chain, to exclude during the cardano transactions preload
`[default: 2160]`. | `2160` | - | :heavy_check_mark: | +| `ipfs_rpc_server_config` | - | - | `IPFS_RPC_SERVER_CONFIG` | 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.
`ipns_key_name` (optional) if set, publishes the snapshot MFS folder to IPNS under this key name. The
key must already exist in the Kubo node keystore. | - | `{ "url": "http://localhost:5001/" }`or `{ "url": "http://localhost:5001/", "mfs_folder_name": "custom-folder", "ipns_key_name": "ipns-key" }` | - | | `enable_metrics_server` | `--enable-metrics-server` | - | `ENABLE_METRICS_SERVER` | Enable metrics HTTP server (Prometheus endpoint on /metrics) | `false` | - | - | | `metrics_server_ip` | `--metrics-server-ip` | - | `METRICS_SERVER_IP` | Metrics HTTP server IP | `0.0.0.0` | - | - | | `metrics_server_port` | `--metrics-server-port` | - | `METRICS_SERVER_PORT` | Metrics HTTP server listening port | `9090` | - | - | diff --git a/mithril-aggregator/Cargo.toml b/mithril-aggregator/Cargo.toml index 9e11fe2a65d..f9f2e74e73e 100644 --- a/mithril-aggregator/Cargo.toml +++ b/mithril-aggregator/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mithril-aggregator" -version = "0.10.20" +version = "0.10.21" description = "A Mithril Aggregator server" authors = { workspace = true } edition = { workspace = true } diff --git a/mithril-aggregator/src/configuration.rs b/mithril-aggregator/src/configuration.rs index 09fd56d5814..18d3d8088da 100644 --- a/mithril-aggregator/src/configuration.rs +++ b/mithril-aggregator/src/configuration.rs @@ -501,9 +501,11 @@ pub struct ServeCommandConfiguration { /// /// `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. + /// `ipns_key_name` (optional) if set, publishes the snapshot MFS folder to IPNS under this key name. The + /// key must already exist in the Kubo node keystore. #[example = "\ `{ \"url\": \"http://localhost:5001/\" }`\ - or `{ \"url\": \"http://localhost:5001/\", \"mfs_folder_name\": \"custom-folder\" }`\ + or `{ \"url\": \"http://localhost:5001/\", \"mfs_folder_name\": \"custom-folder\", \"ipns_key_name\": \"ipns-key\" }`\ "] #[serde( default, @@ -718,6 +720,12 @@ pub struct IpfsRpcServerConfig { /// Defaults to "mithril" if not specified. #[serde(default = "default_mfs_folder_name")] pub mfs_folder_name: String, + + /// Optional name of the IPNS key used to publish the snapshot MFS folder. + /// + /// The key must already exist in the IPFS node keystore. If it is missing or the publication + /// fails, a warning is logged and the snapshot upload still succeeds. + pub ipns_key_name: Option, } impl IpfsRpcServerConfig { @@ -1519,13 +1527,14 @@ mod test { #[test] fn deserializing_ipfs_rpc_server_parameters() { - let deserialized_without_mfs_dir: IpfsRpcServerConfig = + let deserialized_with_defaults: IpfsRpcServerConfig = serde_json::from_str(r#"{ "url": "http://localhost:5001/" }"#).unwrap(); assert_eq!( - deserialized_without_mfs_dir, + deserialized_with_defaults, IpfsRpcServerConfig { url: "http://localhost:5001/".to_string(), mfs_folder_name: default_mfs_folder_name(), + ipns_key_name: None, } ); @@ -1538,6 +1547,20 @@ mod test { IpfsRpcServerConfig { url: "http://localhost:5001/".to_string(), mfs_folder_name: "altered".to_string(), + ipns_key_name: None, + } + ); + + let deserialized_with_ipns_key_name: IpfsRpcServerConfig = serde_json::from_str( + r#"{ "url": "http://localhost:5001/", "ipns_key_name": "my-key" }"#, + ) + .unwrap(); + assert_eq!( + deserialized_with_ipns_key_name, + IpfsRpcServerConfig { + url: "http://localhost:5001/".to_string(), + mfs_folder_name: default_mfs_folder_name(), + ipns_key_name: Some("my-key".to_string()), } ); } diff --git a/mithril-aggregator/src/dependency_injection/builder/protocol/artifacts.rs b/mithril-aggregator/src/dependency_injection/builder/protocol/artifacts.rs index ccc24096cc7..2acd47dcaa7 100644 --- a/mithril-aggregator/src/dependency_injection/builder/protocol/artifacts.rs +++ b/mithril-aggregator/src/dependency_injection/builder/protocol/artifacts.rs @@ -282,6 +282,7 @@ impl DependenciesBuilder { Ok(IpfsUploader::new( Arc::new(rpc_api_client), IpfsMfsDirPath::from(ipfs_rpc_config.mfs_folder_name), + ipfs_rpc_config.ipns_key_name, FileUploadRetryPolicy::default(), &self.root_logger(), )) diff --git a/mithril-aggregator/src/file_uploaders/ipfs_uploader.rs b/mithril-aggregator/src/file_uploaders/ipfs_uploader.rs index 4407670bc3b..106a61b9455 100644 --- a/mithril-aggregator/src/file_uploaders/ipfs_uploader.rs +++ b/mithril-aggregator/src/file_uploaders/ipfs_uploader.rs @@ -1,9 +1,10 @@ use std::collections::HashSet; use std::path::{Path, PathBuf}; use std::sync::Arc; +use std::time::Duration; use anyhow::Context; -use slog::{Logger, trace}; +use slog::{Logger, debug, trace, warn}; use mithril_common::StdResult; use mithril_common::entities::FileUri; @@ -13,17 +14,24 @@ 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, + IpfsAddQuery, IpfsFilesLsQuery, IpfsFilesMkdirQuery, IpfsFilesStatQuery, IpfsKeyLsQuery, + IpfsNamePublishQuery, }; use crate::tools::kubo_rpc_client::{IpfsMfsDirPath, KuboRpcClient}; /// IPFS Content Identifier (CID) pub type IpfsCid = String; +/// Lifetime of the IPNS records published by the uploader in the DHT. +const IPNS_RECORD_LIFETIME: Duration = Duration::from_hours(5 * 24); +/// Cache TTL of the IPNS records published by the uploader. +const IPNS_RECORD_TTL: Duration = Duration::from_mins(30); + /// File uploader that stores files to IPFS pub struct IpfsUploader { rpc_client: Arc, ipfs_dir_path: IpfsMfsDirPath, + ipns_key_name: Option, retry_policy: FileUploadRetryPolicy, logger: Logger, } @@ -33,12 +41,14 @@ impl IpfsUploader { pub fn new( rpc_client: Arc, ipfs_dir_path: IpfsMfsDirPath, + ipns_key_name: Option, retry_policy: FileUploadRetryPolicy, logger: &Logger, ) -> Self { Self { rpc_client, ipfs_dir_path, + ipns_key_name, retry_policy, logger: logger.new_with_component_name::(), } @@ -59,6 +69,7 @@ impl IpfsUploader { .await .with_context(|| "listing files in IPFS MFS directory")?; + let mut uploaded_files = 0; for file_path in files { let filename = file_path @@ -82,9 +93,18 @@ impl IpfsUploader { format!(" Uploaded file path: {}", file_path.display()), ) .await?; + uploaded_files += 1; } - self.get_current_directory_cid().await + let dir_cid = self.get_current_directory_cid().await?; + debug!( + self.logger, "Batch upload to IPFS finished"; + "dir_cid" => %dir_cid, "uploaded_files" => uploaded_files, "already_present_files" => files.len() - uploaded_files + ); + + self.publish_directory_to_ipns(&dir_cid).await; + + Ok(dir_cid) } /// Get the current directory CID, reflecting the latest state of the directory @@ -146,6 +166,52 @@ impl IpfsUploader { Ok(FileUri(cid)) } + + /// Check that the given IPNS key is available in the IPFS node keystore, logging a warning + /// if it is not or if the check fails. + /// + /// Allows logging a cleaner message than just relying on the error returned by the IPFS node. + async fn check_ipns_key_availability(&self, ipns_key_name: &str) -> bool { + match self.rpc_client.list_ipns_key_names().await { + Ok(key_names) if key_names.iter().any(|name| name == ipns_key_name) => true, + Ok(_) => { + warn!( + self.logger, "IPNS key not found in the IPFS node, skipping publication of the directory to IPNS. The key must be imported in the IPFS node keystore"; + "key_name" => ipns_key_name + ); + false + } + Err(err) => { + warn!( + self.logger, "Failed to list the IPFS node keys, skipping publication of the directory to IPNS"; + "err" => ?err, "key_name" => ipns_key_name + ); + false + } + } + } + + async fn publish_directory_to_ipns(&self, dir_cid: &IpfsCid) { + if let Some(key) = &self.ipns_key_name { + if !self.check_ipns_key_availability(key).await { + return; + } + + trace!(self.logger, "Publishing dir to IPNS"; "dir_cid" => dir_cid, "key_name" => key); + + match self.rpc_client.publish_dir_to_ipns(dir_cid, key).await { + Ok(()) => { + debug!(self.logger, "Directory published to IPNS"; "dir_cid" => dir_cid, "key_name" => key); + } + Err(err) => { + warn!( + self.logger, "Failed to publish directory to an IPNS record"; + "err" => ?err, "dir_cid" => dir_cid, "key_name" => key + ); + } + } + } + } } #[async_trait::async_trait] @@ -181,6 +247,12 @@ pub trait IpfsBackendUploader: Sync + Send { mfs_dir_path: &IpfsMfsDirPath, file_path: &Path, ) -> StdResult>; + + /// Publish a directory to IPNS using a given key + async fn publish_dir_to_ipns(&self, dir_cid: &IpfsCid, key_name: &str) -> StdResult<()>; + + /// List the names of the keys available in the IPFS node keystore + async fn list_ipns_key_names(&self) -> StdResult>; } #[async_trait::async_trait] @@ -221,6 +293,21 @@ impl IpfsBackendUploader for KuboRpcClient { .await?; Ok(stat.map(|stat| stat.hash)) } + + async fn publish_dir_to_ipns(&self, dir_cid: &IpfsCid, key_name: &str) -> StdResult<()> { + let object_path = format!("/ipfs/{dir_cid}"); + self.send( + IpfsNamePublishQuery::publish_to_ipns(object_path, key_name) + .with_lifetime(IPNS_RECORD_LIFETIME) + .with_ttl(IPNS_RECORD_TTL), + ) + .await + } + + async fn list_ipns_key_names(&self) -> StdResult> { + let keys = self.send(IpfsKeyLsQuery::new()).await?; + Ok(keys.into_iter().map(|key| key.name).collect()) + } } #[cfg(test)] @@ -241,10 +328,21 @@ mod tests { Self::new( MockBuilder::configure(mock_config), mfs_dir.into(), + None, FileUploadRetryPolicy::never(), &TestLogger::stdout(), ) } + + fn with_ipns_key_name(mut self, key_name: &str) -> Self { + self.ipns_key_name = Some(key_name.to_string()); + self + } + + fn with_logger(mut self, logger: Logger) -> Self { + self.logger = logger; + self + } } #[tokio::test] @@ -335,6 +433,7 @@ mod tests { .with(eq(IpfsMfsDirPath::from(MFS_DIR))) .return_once(|_| Ok("directory-cid".to_string())) .once(); + mock.expect_publish_dir_to_ipns().never(); }); let directory_cid = uploader @@ -349,6 +448,47 @@ mod tests { assert_eq!("directory-cid", directory_cid); } + #[tokio::test] + async fn publish_directory_cid_to_ipns_using_key_name_if_configured() { + 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(HashSet::new())) + .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_get_dir_cid() + .with(eq(IpfsMfsDirPath::from(MFS_DIR))) + .return_once(|_| Ok("directory-cid".to_string())) + .once(); + mock.expect_list_ipns_key_names() + .return_once(|| Ok(vec!["self".to_string(), "test-key-name".to_string()])) + .once(); + mock.expect_publish_dir_to_ipns() + .with(eq("directory-cid".to_string()), eq("test-key-name")) + .return_once(|_, _| Ok(())) + .once(); + }) + .with_ipns_key_name("test-key-name"); + + let directory_cid = uploader + .batch_upload_to_dir(&[PathBuf::from("/local/new-file-1.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| { @@ -535,5 +675,109 @@ mod tests { .await .expect_err("directory CID retrieval failure should be returned"); } + + #[tokio::test] + async fn log_error_as_warning_and_continue_without_failing_if_publishing_to_ipns_fails() { + let (logger, log_inspector) = TestLogger::memory(); + let uploader = IpfsUploader::new_for_test(MFS_DIR, move |mock| { + mock.expect_create_dir().return_once(|_| Ok(())).once(); + mock.expect_list_directory_files() + .return_once(|_| Ok(HashSet::new())) + .once(); + mock.expect_upload_file().never(); + mock.expect_get_dir_cid() + .return_once(|_| Ok("directory-cid".to_string())) + .once(); + mock.expect_list_ipns_key_names() + .return_once(|| Ok(vec!["test-key-name".to_string()])) + .once(); + mock.expect_publish_dir_to_ipns() + .return_once(|_, _| Err(anyhow!("publish dir to ipns failure"))) + .once(); + }) + .with_ipns_key_name("test-key-name") + .with_logger(logger); + + let directory_cid = uploader.batch_upload_to_dir(&[]).await.unwrap(); + + assert_eq!("directory-cid", directory_cid); + assert!(log_inspector.contains_log("publish dir to ipns failure")); + } + + #[tokio::test] + async fn skip_publishing_to_ipns_without_failing_if_ipns_key_is_not_available() { + let (logger, log_inspector) = TestLogger::memory(); + let uploader = IpfsUploader::new_for_test(MFS_DIR, move |mock| { + mock.expect_create_dir().return_once(|_| Ok(())).once(); + mock.expect_list_directory_files() + .return_once(|_| Ok(HashSet::new())) + .once(); + mock.expect_upload_file().never(); + mock.expect_get_dir_cid() + .return_once(|_| Ok("directory-cid".to_string())) + .once(); + mock.expect_list_ipns_key_names() + .return_once(|| Ok(vec!["self".to_string()])) + .once(); + mock.expect_publish_dir_to_ipns().never(); + }) + .with_ipns_key_name("test-key-name") + .with_logger(logger); + + let directory_cid = uploader.batch_upload_to_dir(&[]).await.unwrap(); + + assert_eq!("directory-cid", directory_cid); + assert!(log_inspector.contains_log("IPNS key not found in the IPFS node")); + } + } + + mod check_ipns_key_availability { + use anyhow::anyhow; + + use super::*; + + #[tokio::test] + async fn key_is_available_if_listed_in_the_ipfs_node_keys() { + let (logger, log_inspector) = TestLogger::memory(); + let uploader = IpfsUploader::new_for_test("/test", |mock| { + mock.expect_list_ipns_key_names() + .return_once(|| Ok(vec!["self".to_string(), "test-key-name".to_string()])) + .once(); + }) + .with_logger(logger); + + assert!(uploader.check_ipns_key_availability("test-key-name").await); + assert!(!log_inspector.contains_log("test-key-name")); + } + + #[tokio::test] + async fn key_is_not_available_and_warn_if_not_listed_in_the_ipfs_node_keys() { + let (logger, log_inspector) = TestLogger::memory(); + let uploader = IpfsUploader::new_for_test("/test", |mock| { + mock.expect_list_ipns_key_names() + .return_once(|| Ok(vec!["self".to_string(), "another-key".to_string()])) + .once(); + }) + .with_logger(logger); + + assert!(!uploader.check_ipns_key_availability("test-key-name").await); + assert!(log_inspector.contains_log("IPNS key not found in the IPFS node")); + assert!(log_inspector.contains_log("test-key-name")); + } + + #[tokio::test] + async fn key_is_not_available_and_warn_if_listing_the_ipfs_node_keys_fails() { + let (logger, log_inspector) = TestLogger::memory(); + let uploader = IpfsUploader::new_for_test("/test", |mock| { + mock.expect_list_ipns_key_names() + .return_once(|| Err(anyhow!("list ipns keys failure"))) + .once(); + }) + .with_logger(logger); + + assert!(!uploader.check_ipns_key_availability("test-key-name").await); + assert!(log_inspector.contains_log("Failed to list the IPFS node keys")); + assert!(log_inspector.contains_log("list ipns keys failure")); + } } } diff --git a/mithril-aggregator/src/tools/kubo_rpc_client/api.rs b/mithril-aggregator/src/tools/kubo_rpc_client/api.rs index a2334af585c..25a4f87f023 100644 --- a/mithril-aggregator/src/tools/kubo_rpc_client/api.rs +++ b/mithril-aggregator/src/tools/kubo_rpc_client/api.rs @@ -9,6 +9,12 @@ use mithril_common::{StdError, StdResult}; use crate::tools::url_sanitizer::SanitizedUrlWithTrailingSlash; +/// Default timeout for the RPC request. +pub const DEFAULT_TIMEOUT_SECS: u64 = 10; + +/// Timeout for slow queries (large uploads, DHT publication). +pub const TIMEOUT_FOR_SLOW_QUERY_SECS: u64 = 3 * 60; + /// Trait for defining RPC queries to the Kubo IPFS node. #[async_trait::async_trait] pub trait KuboRpcQuery: Sync { @@ -30,9 +36,9 @@ pub trait KuboRpcQuery: Sync { Ok(request_builder) } - /// Timeout for the RPC request, defaults to 10 seconds. + /// Timeout for the RPC request, defaults to [DEFAULT_TIMEOUT_SECS] seconds. fn timeout(&self) -> Duration { - Duration::from_secs(10) + Duration::from_secs(DEFAULT_TIMEOUT_SECS) } /// Handles a successful RPC response and converts it to the expected response type. diff --git a/mithril-aggregator/src/tools/kubo_rpc_client/mod.rs b/mithril-aggregator/src/tools/kubo_rpc_client/mod.rs index ce7ff234def..4876be6cdaa 100644 --- a/mithril-aggregator/src/tools/kubo_rpc_client/mod.rs +++ b/mithril-aggregator/src/tools/kubo_rpc_client/mod.rs @@ -2,7 +2,7 @@ mod api; mod path; pub mod query; -pub use api::{KuboRpcClient, KuboRpcQuery}; +pub use api::{DEFAULT_TIMEOUT_SECS, KuboRpcClient, KuboRpcQuery, TIMEOUT_FOR_SLOW_QUERY_SECS}; pub use path::IpfsMfsDirPath; #[cfg(test)] 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 index 60ad7b78312..875e0b9b493 100644 --- a/mithril-aggregator/src/tools/kubo_rpc_client/query/ipfs_add.rs +++ b/mithril-aggregator/src/tools/kubo_rpc_client/query/ipfs_add.rs @@ -7,7 +7,9 @@ use serde::Deserialize; use mithril_common::StdResult; -use crate::tools::kubo_rpc_client::{IpfsMfsDirPath, KuboRpcQuery}; +use crate::tools::kubo_rpc_client::{ + DEFAULT_TIMEOUT_SECS, IpfsMfsDirPath, KuboRpcQuery, TIMEOUT_FOR_SLOW_QUERY_SECS, +}; /// Query to add a file to IPFS via the Kubo RPC API. /// @@ -30,9 +32,6 @@ pub struct IpfsAddResponse { } 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 { @@ -77,12 +76,12 @@ impl IpfsAddQuery { const BYTES_PER_MIB: u64 = 1024 * 1024; match file_size_in_bytes { - None => Duration::from_secs(Self::MAX_TIMEOUT_SECS), + None => Duration::from_secs(TIMEOUT_FOR_SLOW_QUERY_SECS), Some(size_in_bytes) => { let file_size_mib = size_in_bytes.div_ceil(BYTES_PER_MIB); - let timeout_secs = Self::BASE_TIMEOUT_SECS + let timeout_secs = DEFAULT_TIMEOUT_SECS .saturating_add(file_size_mib) - .min(Self::MAX_TIMEOUT_SECS); + .min(TIMEOUT_FOR_SLOW_QUERY_SECS); Duration::from_secs(timeout_secs) } @@ -276,7 +275,7 @@ mod tests { #[test] fn timeout_duration_adds_one_second_for_each_started_mib() { let cases = [ - (0, IpfsAddQuery::BASE_TIMEOUT_SECS), + (0, DEFAULT_TIMEOUT_SECS), (1, 11), (MIB, 11), (MIB + 1, 12), @@ -294,13 +293,13 @@ mod tests { } #[test] - fn timeout_duration_is_capped_at_three_minutes() { + fn timeout_duration_is_capped_at_slow_query_timeout() { 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), + (169 * MIB + 1, TIMEOUT_FOR_SLOW_QUERY_SECS), + (170 * MIB, TIMEOUT_FOR_SLOW_QUERY_SECS), + (171 * MIB, TIMEOUT_FOR_SLOW_QUERY_SECS), + (u64::MAX, TIMEOUT_FOR_SLOW_QUERY_SECS), ]; for (file_size, expected_timeout_secs) in cases { diff --git a/mithril-aggregator/src/tools/kubo_rpc_client/query/ipfs_key_ls.rs b/mithril-aggregator/src/tools/kubo_rpc_client/query/ipfs_key_ls.rs new file mode 100644 index 00000000000..d7474e211b3 --- /dev/null +++ b/mithril-aggregator/src/tools/kubo_rpc_client/query/ipfs_key_ls.rs @@ -0,0 +1,139 @@ +use anyhow::Context; +use reqwest::Response; +use serde::Deserialize; + +use mithril_common::StdResult; + +use crate::tools::kubo_rpc_client::KuboRpcQuery; + +/// Query to list the keys stored in the IPFS node keystore (used to sign IPNS records). +/// +/// see: https://docs.ipfs.tech/reference/kubo/rpc/#api-v0-key-ls +#[derive(Debug, Default)] +pub struct IpfsKeyLsQuery; + +#[derive(Debug, Clone, PartialEq, Eq, Deserialize)] +#[serde(rename_all = "PascalCase")] +struct IpfsKeyLsResponse { + keys: Option>, +} + +/// A key stored in the IPFS node keystore +#[derive(Debug, Clone, PartialEq, Eq, Deserialize)] +#[serde(rename_all = "PascalCase")] +pub struct IpfsKey { + /// Name of the key, local to the IPFS node + pub name: String, + /// Id of the key, usable to resolve the IPNS records signed with this key + pub id: String, +} + +impl IpfsKeyLsQuery { + /// Create a query that will list the keys of the IPFS node keystore. + pub fn new() -> Self { + Self + } +} + +#[async_trait::async_trait] +impl KuboRpcQuery for IpfsKeyLsQuery { + type Response = Vec; + + fn route(&self) -> String { + "api/v0/key/ls".to_string() + } + + async fn handle_success(&self, response: Response) -> StdResult { + let response: IpfsKeyLsResponse = response + .json() + .await + .with_context(|| "Failed to deserialize IPFS key ls response")?; + + Ok(response.keys.unwrap_or_default()) + } +} + +#[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_keys_list_if_request_succeeds() { + let (server, client) = setup_server_and_client(); + server.mock(|when, then| { + when.method(POST).path("/api/v0/key/ls"); + then.status(200).json_body(serde_json::json!({ + "Keys": [ + { "Name": "self", "Id": "k51qzi5uqu5djd98ozajs6h4ljp0p53lqf4fn4p3904t20yqjbc2nksxotpz66" }, + { "Name": "mithril-snapshot", "Id": "k51qzi5uqu5di91xjrh1a4hs0gd7l22tdoyh1odjjfu40vllpi1p9i2jert6sp" } + ] + })); + }); + + let response = client.send(IpfsKeyLsQuery::new()).await.unwrap(); + + assert_eq!( + vec![ + IpfsKey { + name: "self".to_string(), + id: "k51qzi5uqu5djd98ozajs6h4ljp0p53lqf4fn4p3904t20yqjbc2nksxotpz66" + .to_string(), + }, + IpfsKey { + name: "mithril-snapshot".to_string(), + id: "k51qzi5uqu5di91xjrh1a4hs0gd7l22tdoyh1odjjfu40vllpi1p9i2jert6sp" + .to_string(), + }, + ], + response + ); + } + + #[tokio::test] + async fn return_empty_list_when_keys_are_null() { + let (server, client) = setup_server_and_client(); + server.mock(|when, then| { + when.method(POST).path("/api/v0/key/ls"); + then.status(200).json_body(serde_json::json!({ "Keys": null })); + }); + + let response = client.send(IpfsKeyLsQuery::new()).await.unwrap(); + + assert_eq!(Vec::::new(), response); + } + + #[tokio::test] + async fn return_error_if_response_body_is_malformed() { + let (server, client) = setup_server_and_client(); + server.mock(|when, then| { + when.method(POST).path("/api/v0/key/ls"); + then.status(200).json_body(serde_json::json!({ "Keys": "invalid" })); + }); + + let err = client.send(IpfsKeyLsQuery::new()).await.unwrap_err(); + + assert!( + err.to_string().contains("Failed to deserialize IPFS key ls response"), + "unexpected error: {err:?}" + ); + } + + #[tokio::test] + async fn return_error_if_request_fails() { + let (server, client) = setup_server_and_client(); + server.mock(|when, then| { + when.method(POST).path("/api/v0/key/ls"); + then.status(500).json_body( + serde_json::json!({"Message":"keystore unavailable","Code":0,"Type":"error"}), + ); + }); + + let err = client.send(IpfsKeyLsQuery::new()).await.unwrap_err(); + + assert!(err.to_string().contains("keystore unavailable")); + } +} diff --git a/mithril-aggregator/src/tools/kubo_rpc_client/query/ipfs_name_publish.rs b/mithril-aggregator/src/tools/kubo_rpc_client/query/ipfs_name_publish.rs new file mode 100644 index 00000000000..c00ccb6a6d2 --- /dev/null +++ b/mithril-aggregator/src/tools/kubo_rpc_client/query/ipfs_name_publish.rs @@ -0,0 +1,166 @@ +use std::time::Duration; + +use reqwest::{RequestBuilder, Response}; + +use mithril_common::StdResult; + +use crate::tools::kubo_rpc_client::{KuboRpcQuery, TIMEOUT_FOR_SLOW_QUERY_SECS}; + +/// Query to publish an IPFS object to IPNS +/// +/// see: https://docs.ipfs.tech/reference/kubo/rpc/#api-v0-name-publish +#[derive(Debug)] +pub struct IpfsNamePublishQuery { + ipfs_object_path: String, + ipfs_key: String, + lifetime: Duration, + ttl: Duration, +} + +impl IpfsNamePublishQuery { + /// Default lifetime of the IPNS record in the DHT (using kubo default: 48 hours). + const DEFAULT_LIFETIME: Duration = Duration::from_hours(48); + /// Default cache TTL of the IPNS record (using kubo default: 5 minutes). + const DEFAULT_TTL: Duration = Duration::from_mins(5); + + /// Create a query that will publish the given IPFS object to IPNS. + pub fn publish_to_ipns, K: Into>( + ipfs_object_path: P, + ipfs_key: K, + ) -> Self { + Self { + ipfs_object_path: ipfs_object_path.into(), + ipfs_key: ipfs_key.into(), + lifetime: Self::DEFAULT_LIFETIME, + ttl: Self::DEFAULT_TTL, + } + } + + /// Set the lifetime of the IPNS record in the DHT. + /// + /// Sent to kubo with a seconds precision, sub-second parts are truncated. + pub fn with_lifetime(mut self, lifetime: Duration) -> Self { + self.lifetime = lifetime; + self + } + + /// Set the cache TTL of the IPNS record. + /// + /// Sent to kubo with a seconds precision, sub-second parts are truncated. + pub fn with_ttl(mut self, ttl: Duration) -> Self { + self.ttl = ttl; + self + } + + /// Format a duration to a format accepted by kubo (Go `time.ParseDuration`). + fn format_duration(duration: Duration) -> String { + format!("{}s", duration.as_secs()) + } +} + +#[async_trait::async_trait] +impl KuboRpcQuery for IpfsNamePublishQuery { + type Response = (); + + fn route(&self) -> String { + "api/v0/name/publish".to_string() + } + + async fn configure_request( + &self, + request_builder: RequestBuilder, + ) -> StdResult { + Ok(request_builder.query(&[ + ("arg", &self.ipfs_object_path), + ("key", &self.ipfs_key), + ("lifetime", &Self::format_duration(self.lifetime)), + ("ttl", &Self::format_duration(self.ttl)), + ])) + } + + fn timeout(&self) -> Duration { + Duration::from_secs(TIMEOUT_FOR_SLOW_QUERY_SECS) + } + + 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/name/publish") + .query_param( + "arg", + "/ipfs/QmYi7wrRFKVCcTB56A6Pep2j31Q5mHfmmu21RzHXu25RVR", + ) + .query_param("key", "my-key-name") + .query_param("lifetime", "172800s") + .query_param("ttl", "300s"); + then.status(200); + }); + + client + .send(IpfsNamePublishQuery::publish_to_ipns( + "/ipfs/QmYi7wrRFKVCcTB56A6Pep2j31Q5mHfmmu21RzHXu25RVR", + "my-key-name", + )) + .await + .unwrap(); + } + + #[tokio::test] + async fn send_configured_lifetime_and_ttl_in_seconds() { + let (server, client) = setup_server_and_client(); + server.mock(|when, then| { + when.method(POST) + .path("/api/v0/name/publish") + .query_param("lifetime", "120s") + .query_param("ttl", "1s"); + then.status(200); + }); + + client + .send( + IpfsNamePublishQuery::publish_to_ipns( + "/ipfs/QmYi7wrRFKVCcTB56A6Pep2j31Q5mHfmmu21RzHXu25RVR", + "my-key-name", + ) + .with_lifetime(Duration::from_mins(2)) + .with_ttl(Duration::from_millis(1500)), + ) + .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/name/publish"); + then.status(500).json_body( + serde_json::json!({"Message":"no key by the given name was found","Code":0,"Type":"error"}), + ); + }); + + let err = client + .send(IpfsNamePublishQuery::publish_to_ipns( + "/ipfs/QmYi7wrRFKVCcTB56A6Pep2j31Q5mHfmmu21RzHXu25RVR", + "my-key-name", + )) + .await + .unwrap_err(); + assert!(err.to_string().contains("no key by the given name was found")); + } +} diff --git a/mithril-aggregator/src/tools/kubo_rpc_client/query/mod.rs b/mithril-aggregator/src/tools/kubo_rpc_client/query/mod.rs index 42e3dfc97a8..915ddbf2a12 100644 --- a/mithril-aggregator/src/tools/kubo_rpc_client/query/mod.rs +++ b/mithril-aggregator/src/tools/kubo_rpc_client/query/mod.rs @@ -2,8 +2,12 @@ mod ipfs_add; mod ipfs_files_ls; mod ipfs_files_mkdir; mod ipfs_files_stat; +mod ipfs_key_ls; +mod ipfs_name_publish; pub use ipfs_add::*; pub use ipfs_files_ls::*; pub use ipfs_files_mkdir::*; pub use ipfs_files_stat::*; +pub use ipfs_key_ls::*; +pub use ipfs_name_publish::*; diff --git a/mithril-test-lab/ipfs-devnet/README.md b/mithril-test-lab/ipfs-devnet/README.md index 82ea4199259..88f26a0be5b 100644 --- a/mithril-test-lab/ipfs-devnet/README.md +++ b/mithril-test-lab/ipfs-devnet/README.md @@ -152,6 +152,10 @@ For node `N`, the default local ports are: | Gateway | `5100 + N` | `5101` | | Swarm | `5200 + N` | `5201` | +> [!IMPORTANT] +> `init` fails if any of these ports is already in use (e.g. by a swarm still running from this or another +> directory). Stop that swarm first, see [Troubleshooting](#some-kubo-nodes-are-still-running-after-stop). + #### Re-initialize an existing swarm If the target swarm directory already contains node configurations, use `--overwrite`: @@ -374,8 +378,11 @@ SWARM_DIR=./swarm ./ipfs-devnet.sh query -- --node 2 swarm peers Download the Kubo Web UI as a CAR archive and import it into the first swarm node, making it available inside the private IPFS network. +The CAR archive is downloaded from the [ipfs-webui GitHub release](https://github.com/ipfs/ipfs-webui/releases) whose +CID matches the Web UI expected by the running Kubo node. The command fails if no release matches this CID. + > [!IMPORTANT] -> The swarm must be running before using this command. The download may take some time. +> The swarm must be running before using this command, and `jq` must be installed. ```shell ./ipfs-devnet.sh load-webui [OPTIONS] diff --git a/mithril-test-lab/ipfs-devnet/VERSION b/mithril-test-lab/ipfs-devnet/VERSION index 22920c6163f..39b89960093 100644 --- a/mithril-test-lab/ipfs-devnet/VERSION +++ b/mithril-test-lab/ipfs-devnet/VERSION @@ -1,2 +1,2 @@ -0.1.2 +0.1.3 diff --git a/mithril-test-lab/ipfs-devnet/commands/ipfs-devnet-load-webui.sh b/mithril-test-lab/ipfs-devnet/commands/ipfs-devnet-load-webui.sh index 4b0c88c6ef2..b9b4d7839f0 100644 --- a/mithril-test-lab/ipfs-devnet/commands/ipfs-devnet-load-webui.sh +++ b/mithril-test-lab/ipfs-devnet/commands/ipfs-devnet-load-webui.sh @@ -10,10 +10,16 @@ readonly SCRIPT_DIRECTORY # shellcheck source=./lib/common.sh source "${SCRIPT_DIRECTORY}/lib/common.sh" +readonly GITHUB_API_RELEASES="https://api.github.com/repos/ipfs/ipfs-webui/releases" +readonly CONNECT_TIMEOUT_SECONDS=10 + display_help() { echo "Download and load the Kubo Web UI on the private IPFS network" echo - echo "IMPORTANT: The IPFS devnet must be running, and the download can be quite slow." + echo "The Web UI CAR archive is downloaded from the ipfs-webui GitHub release matching the Web UI CID" + echo "expected by the running Kubo node." + echo + echo "IMPORTANT: The IPFS devnet must be running." echo echo "Usage: $0 [OPTIONS]" echo @@ -34,13 +40,13 @@ find_web_ui_cid() { local location cid # leverage the 302 Found redirection which contains the CID in its location - # example: `Location: /ipfs/bafybeiciqeyipumpmhxzlxnbqdbbv6u5uij4hy4wax64dmj7kvrhusiq6y` + # example: `http://127.0.0.1:5001/ipfs/bafybeiciqeyipumpmhxzlxnbqdbbv6u5uij4hy4wax64dmj7kvrhusiq6y` location=$( - curl --fail --silent --show-error --head "http://127.0.0.1:${node_port}/webui/" | - awk 'BEGIN { IGNORECASE = 1 } /^Location:/ { sub(/\r$/, ""); print $2; exit }' + curl --fail --silent --show-error --head --output /dev/null --write-out '%{redirect_url}' \ + "http://127.0.0.1:${node_port}/webui/" ) || error_exit "Failed to query Kubo Web UI redirect." - cid=${location#/ipfs/} + cid=${location##*/ipfs/} if [[ -z "$cid" || "$cid" == "$location" ]]; then error_exit "Failed to extract Web UI CID from redirect location: '$location'" @@ -49,20 +55,53 @@ find_web_ui_cid() { printf '%s\n' "$cid" } -download_web_ui_car() { +fetch() { + curl --fail --silent --show-error --location --connect-timeout "$CONNECT_TIMEOUT_SECONDS" "$@" +} + +find_web_ui_car_url_from_github() { + local -r cid="$1" + local releases car_url + + releases=$(fetch -H "Accept: application/vnd.github+json" "${GITHUB_API_RELEASES}?per_page=100") || + error_exit "Failed to list ipfs-webui releases from '${GITHUB_API_RELEASES}'." + + # Each ipfs-webui release body states the CID of its build (e.g. "CID `bafybei...`") + car_url=$( + jq --raw-output --arg cid "$cid" ' + first( + .[] + | select((.body // "") | contains($cid)) + | .assets[] + | select(.name | endswith(".car")) + | .browser_download_url + ) // empty' <<< "$releases" + ) || error_exit "Failed to parse ipfs-webui releases from '${GITHUB_API_RELEASES}'." + + if [[ -z "$car_url" ]]; then + error_exit "No ipfs-webui GitHub release found for CID '${cid}' (searched ${GITHUB_API_RELEASES})." + fi + + printf '%s\n' "$car_url" +} + +download_web_ui_car_from_github() { local -r cid="$1" local -r download_dir="${2%/}" - local -r target_url="https://ipfs.io/ipfs/${cid}?format=car&dag-scope=all" local -r archive_path="${download_dir}/webui-${cid}.car" if [[ -f "$archive_path" ]]; then echo ">> Web UI CAR file already exists: ${archive_path}" >&2 else - echo ">> Downloading Kubo Web UI with CID ${cid} from 'ipfs.io'..." >&2 - curl --fail --silent --show-error --location -H "Accept: application/vnd.ipld.car" \ - --output "$archive_path" "$target_url" || - error_exit "Failed to download Kubo Web UI CAR file from '${target_url}'." + local target_url + target_url=$(find_web_ui_car_url_from_github "$cid") + + echo ">> Downloading Kubo Web UI with CID ${cid} from '${target_url}'..." >&2 + if ! fetch --output "$archive_path" "$target_url"; then + rm -f "$archive_path" + error_exit "Failed to download Kubo Web UI CAR file from '${target_url}'." + fi fi echo "$archive_path" @@ -72,8 +111,19 @@ load_web_ui_car() { local -r car_file_path="$1" local -r ipfs_bin_path="$2" local -r node_dir="${3%/}" - - IPFS_PATH="$node_dir" "$ipfs_bin_path" dag import "$car_file_path" + local -r expected_cid="$4" + local import_output + + import_output=$(IPFS_PATH="$node_dir" "$ipfs_bin_path" dag import --pin-roots --enc=json "$car_file_path") || + error_exit "Failed to import Web UI CAR file '${car_file_path}'." + echo "$import_output" + + # Example of expected output: `{"Root":{"Cid":{"/":"bafybeiciqeyipumpmhxzlxnbqdbbv6u5uij4hy4wax64dmj7kvrhusiq6y"},"PinErrorMsg":""}}` + if ! jq --null-input --exit-status --arg cid "$expected_cid" \ + '[inputs | .Root? | select(.Cid["/"] == $cid and .PinErrorMsg == "")] | any' <<< "$import_output" >/dev/null; then + rm -f "$car_file_path" + error_exit "Imported CAR file root does not match the expected Web UI CID '${expected_cid}', CAR file discarded." + fi } # --------------------------------------------------------------------------- @@ -107,7 +157,7 @@ if [[ "$#" -gt 0 ]]; then error_exit "Unexpected argument: $1" fi -check_requirements "curl" "awk" +check_requirements "curl" "jq" readonly DOWNLOAD_DIR=${DOWNLOAD_DIR:-"/tmp/kubo/"} readonly SWARM_DIR @@ -118,7 +168,7 @@ require_directory "$SWARM_DIR" "-s, --swarm-dir" # Use the first node of the network, it will be propagated from it to other nodes afterwards declare -r NODE_PORT=5001 -declare -r NODE_DIR="${SWARM_DIR}/kubo-node-1" +declare -r NODE_DIR="${SWARM_DIR%/}/kubo-node-1" declare -r IPFS_BIN="${SWARM_DIR%/}/bin/ipfs" require_directory "$NODE_DIR" "Kubo node directory" @@ -136,9 +186,9 @@ CID=$(find_web_ui_cid "$NODE_PORT") readonly CID echo ">> CID found: '$CID'" -CAR_FILE=$(download_web_ui_car "$CID" "$DOWNLOAD_DIR") +CAR_FILE=$(download_web_ui_car_from_github "$CID" "$DOWNLOAD_DIR") readonly CAR_FILE echo ">> Loading Web UI ..." -load_web_ui_car "$CAR_FILE" "$IPFS_BIN" "$NODE_DIR" +load_web_ui_car "$CAR_FILE" "$IPFS_BIN" "$NODE_DIR" "$CID" echo ">> Web UI loaded in the first swarm node ('$NODE_DIR')" diff --git a/mithril-test-lab/ipfs-devnet/commands/mkfiles/kubo-configure-swarm.sh b/mithril-test-lab/ipfs-devnet/commands/mkfiles/kubo-configure-swarm.sh index 060cfe9bd65..0b17b137998 100755 --- a/mithril-test-lab/ipfs-devnet/commands/mkfiles/kubo-configure-swarm.sh +++ b/mithril-test-lab/ipfs-devnet/commands/mkfiles/kubo-configure-swarm.sh @@ -10,6 +10,10 @@ readonly SCRIPT_DIRECTORY # shellcheck source=../lib/common.sh source "${SCRIPT_DIRECTORY}/../lib/common.sh" +readonly API_PORT_BASE=5000 +readonly GATEWAY_PORT_BASE=5100 +readonly SWARM_PORT_BASE=5200 + display_help() { echo "Configure a private swarm of Kubo IPFS node" echo @@ -30,6 +34,15 @@ generate_swarm_key() { od -An -N32 -tx1 /dev/urandom | tr -d ' \n' } +# Fail if something is already listening on the given local port, whatever process owns it +require_free_port() { + local -r port="$1" + + if (echo >"/dev/tcp/127.0.0.1/${port}") 2>/dev/null; then + error_exit "Port ${port} is already in use: another Kubo swarm may be running, please free the port then try again." + fi +} + # Init a node, removing existing content if overwrite is set, returning the node Peer Id init_node() { local -r node_id="$1" @@ -66,9 +79,9 @@ configure_node() { node_dir="$swarm_dir/kubo-node-$node_id" local api_port gateway_port swarm_port - api_port=$((5000 + node_id)) - gateway_port=$((5100 + node_id)) - swarm_port=$((5200 + node_id)) + api_port=$((API_PORT_BASE + node_id)) + gateway_port=$((GATEWAY_PORT_BASE + node_id)) + swarm_port=$((SWARM_PORT_BASE + node_id)) #---------- Write swarm key { @@ -117,7 +130,7 @@ configure_node_peers() { fi peer_id="${peer_ids[$((peer_node_id - 1))]}" - swarm_port=$((5200 + peer_node_id)) + swarm_port=$((SWARM_PORT_BASE + peer_node_id)) peers_json="${peers_json}${separator}{\"ID\":\"${peer_id}\",\"Addrs\":[\"/ip4/127.0.0.1/tcp/${swarm_port}\"]}" separator="," @@ -167,6 +180,13 @@ fi echo ">> Configuring ${NUMBER_OF_NODES} Kubo nodes in '${SWARM_DIR}'" +# Check ports before touching any node so a running swarm is never orphaned +for ((node_id = 1; node_id <= NUMBER_OF_NODES; node_id++)); do + require_free_port $((API_PORT_BASE + node_id)) + require_free_port $((GATEWAY_PORT_BASE + node_id)) + require_free_port $((SWARM_PORT_BASE + node_id)) +done + SWARM_KEY="$(generate_swarm_key)" readonly SWARM_KEY echo ">> Swarm key generated (not displayed)." diff --git a/mithril-test-lab/ipfs-devnet/commands/mkfiles/kubo-download.sh b/mithril-test-lab/ipfs-devnet/commands/mkfiles/kubo-download.sh index 1c016798edf..6f9ba3bb369 100755 --- a/mithril-test-lab/ipfs-devnet/commands/mkfiles/kubo-download.sh +++ b/mithril-test-lab/ipfs-devnet/commands/mkfiles/kubo-download.sh @@ -11,7 +11,6 @@ readonly SCRIPT_DIRECTORY source "${SCRIPT_DIRECTORY}/../lib/common.sh" readonly BIN_NAME="kubo" -readonly IPFS_DISTRIBUTIONS_CDN="https://dist.ipfs.tech" readonly GITHUB_RELEASES="https://github.com/ipfs/kubo/releases" readonly CONNECT_TIMEOUT_SECONDS=10 @@ -33,12 +32,6 @@ fetch() { curl --fail --silent --show-error --location --connect-timeout "$CONNECT_TIMEOUT_SECONDS" "$@" } -find_last_released_version_from_cdn() { - # The file have the following structure: one version per line, from earliest to latest, named "vXX.YY.ZZ[-rcN]" (e.g "v0.31.0-rc2" or "v0.33.2") - fetch "${IPFS_DISTRIBUTIONS_CDN}/${BIN_NAME}/versions" | - awk '/^v[0-9]+[.][0-9]+[.][0-9]+$/ { latest = $0 } END { if (latest != "") print latest; else exit 1 }' -} - find_last_released_version_from_github() { # The latest release url redirects to the url of its tag (e.g. "https://github.com/ipfs/kubo/releases/tag/v0.43.1") fetch --head --output /dev/null --write-out '%{url_effective}' "${GITHUB_RELEASES}/latest" | @@ -47,9 +40,8 @@ find_last_released_version_from_github() { find_last_released_version() { local latest_version - latest_version=$(find_last_released_version_from_cdn) || - latest_version=$(find_last_released_version_from_github) || - error_exit "Could not find a released version for '${BIN_NAME}' from '${IPFS_DISTRIBUTIONS_CDN}' or '${GITHUB_RELEASES}'." + latest_version=$(find_last_released_version_from_github) || + error_exit "Could not find a released version for '${BIN_NAME}' from '${GITHUB_RELEASES}'." echo "$latest_version" } @@ -124,14 +116,11 @@ download_bin_archive() { archive_name=$(format_archive_name "$version" "$os" "$arch") local -r archive_path="${download_dir}/${archive_name}" - # example url: https://dist.ipfs.tech/kubo/v0.42.0/kubo_v0.42.0_linux-arm64.tar.gz - local -r cdn_url="${IPFS_DISTRIBUTIONS_CDN}/${BIN_NAME}/${version}/${archive_name}" # example url: https://github.com/ipfs/kubo/releases/download/v0.42.0/kubo_v0.42.0_linux-arm64.tar.gz local -r github_url="${GITHUB_RELEASES}/download/${version}/${archive_name}" - download_bin_archive_from "$cdn_url" "$archive_path" || - download_bin_archive_from "$github_url" "$archive_path" || - error_exit "Failed to download '${BIN_NAME}' archive from '${cdn_url}' or '${github_url}'." + download_bin_archive_from "$github_url" "$archive_path" || + error_exit "Failed to download '${BIN_NAME}' archive from '${github_url}'." echo "$archive_path" } diff --git a/mithril-test-lab/ipfs-devnet/commands/mkfiles/mkscripts-lifecycle.sh b/mithril-test-lab/ipfs-devnet/commands/mkfiles/mkscripts-lifecycle.sh index 6e104c99658..97953f11205 100755 --- a/mithril-test-lab/ipfs-devnet/commands/mkfiles/mkscripts-lifecycle.sh +++ b/mithril-test-lab/ipfs-devnet/commands/mkfiles/mkscripts-lifecycle.sh @@ -57,6 +57,30 @@ error_exit() { exit 1 } +readonly READY_TIMEOUT_SECONDS=30 + +# A node is ready once its daemon has written the 'api' file and answers RPC calls. +# Without the 'api' file check, 'ipfs id' would run offline against the repository and succeed. +wait_for_node_ready() { + local -r node_path="$1" + local -r pid="$2" + local -r log_file="${node_path%/}/ipfs.log" + local i + + for ((i = 0; i < READY_TIMEOUT_SECONDS * 2; i++)); do + if [[ -f "${node_path%/}/api" ]] && IPFS_PATH="$node_path" "$IPFS_BIN" id >/dev/null 2>&1; then + echo ">> Kubo node ready: '$node_path'" + return 0 + fi + if ! kill -0 "$pid" 2>/dev/null; then + error_exit "Kubo node exited during startup: '$node_path' (see '$log_file')" + fi + sleep 0.5 + done + + error_exit "Kubo node not ready after ${READY_TIMEOUT_SECONDS}s: '$node_path' (see '$log_file')" +} + EOF printf 'readonly IPFS_BIN=%s\n' "$(shell_quote "$ipfs_bin")" @@ -114,6 +138,11 @@ for node_path in "${NODES_PATHS[@]}"; do printf '%s\n' "$pid" >"$pid_file" echo ">> Started Kubo node: '$node_path' pid=$pid" done + +# Wait in a separate loop so that all nodes boot in parallel +for node_path in "${NODES_PATHS[@]}"; do + wait_for_node_ready "$node_path" "$(cat "${node_path%/}/ipfs.pid")" +done EOF } > "$script_path" diff --git a/mithril-test-lab/mithril-end-to-end/Cargo.toml b/mithril-test-lab/mithril-end-to-end/Cargo.toml index 626f90918e9..e276366fb99 100644 --- a/mithril-test-lab/mithril-end-to-end/Cargo.toml +++ b/mithril-test-lab/mithril-end-to-end/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mithril-end-to-end" -version = "0.5.26" +version = "0.5.27" authors = { workspace = true } edition = { workspace = true } documentation = { workspace = true } diff --git a/mithril-test-lab/mithril-end-to-end/src/devnet/ipfs.rs b/mithril-test-lab/mithril-end-to-end/src/devnet/ipfs.rs index 70b47f33e53..c964e04fe9e 100644 --- a/mithril-test-lab/mithril-end-to-end/src/devnet/ipfs.rs +++ b/mithril-test-lab/mithril-end-to-end/src/devnet/ipfs.rs @@ -1,10 +1,10 @@ -use std::ffi::OsStr; -use std::path::{Path, PathBuf}; -use std::process::Stdio; +use std::ops::{Deref, DerefMut}; +use std::path::{Display, Path, PathBuf}; +use std::process::{ExitStatus, Stdio}; use anyhow::{Context, anyhow}; use reqwest::Url; -use slog_scope::info; +use slog_scope::{info, warn}; use tokio::process::Command; use mithril_common::StdResult; @@ -23,18 +23,19 @@ pub enum IpfsDevnetMode { Detached, } -#[derive(Debug, Clone, Default)] +#[derive(Debug, Clone)] pub struct IpfsDevnet { - devnet_script_path: PathBuf, - swarm_dir: PathBuf, topology: Vec, mode: IpfsDevnetMode, + command_builder: IpfsCommandBuilder, } -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone)] pub struct KuboNode { pub rpc_url: Url, pub working_dir: PathBuf, + swarm_index: u8, + command_builder: IpfsCommandBuilder, } #[derive(Debug, Clone)] @@ -46,23 +47,131 @@ pub struct IpfsDevnetBootstrapArgs { pub mode: IpfsDevnetMode, } -impl IpfsDevnet { - fn build_command>(&self, sub_command: C) -> StdResult { +#[derive(Debug, Clone, Eq, PartialEq)] +pub struct IpnsKey { + pub id: String, + pub name: String, +} + +#[derive(Debug, Clone)] +struct IpfsCommandBuilder { + devnet_script_path: PathBuf, + swarm_dir: PathBuf, +} + +impl IpfsCommandBuilder { + fn new(devnet_script_path: &Path, swarm_dir: &Path) -> Self { + Self { + devnet_script_path: devnet_script_path.to_owned(), + swarm_dir: swarm_dir.to_owned(), + } + } + + fn display_script_path(&self) -> Display<'_> { + self.devnet_script_path.display() + } + + fn build(&self, sub_command: &str, additional_args: &[&str]) -> IpfsCommand { + let mut context = format!("{IPFS_DEVNET_SCRIPT_NAME} {sub_command}"); + + if !additional_args.is_empty() { + context.push(' '); + context.push_str(&additional_args.join(" ")); + } + let mut command = Command::new(self.devnet_script_path.clone()); command .arg(sub_command) + .args(additional_args) .env("SWARM_DIR", self.swarm_dir.as_os_str()) .stdout(Stdio::piped()) .stderr(Stdio::piped()) .kill_on_drop(true); - Ok(command) + + IpfsCommand { command, context } + } + + fn build_query_command(&self, node_swarm_index: u8, query_args: &[&str]) -> IpfsCommand { + let node_swarm_index_str = node_swarm_index.to_string(); + // First `--` forwards arguments to the generated `$SWARM_DIR/query.sh` command + // Second `--` ends `query.sh` options, ensuring that all remaining arguments are forwarded to the executed `ipfs` command. + let additional_args = + [vec!["--", "-n", &node_swarm_index_str, "--"], query_args.to_vec()].concat(); + + self.build("query", &additional_args) + } +} + +struct IpfsCommand { + command: Command, + context: String, +} + +impl IpfsCommand { + /// Execute the IPFS command and return the exit status. + /// + /// Stdout and stderr are forwarded to the slog scope logger. + async fn execute(mut self) -> StdResult { + let exit_status = self + .command + .spawn() + .with_context(|| format!("Failed to spawn '{}'", self.context))? + .wait_forwarding_output_to_slog_scope(&self.context) + .await + .with_context(|| format!("Execution of '{}' failed", self.context))?; + + Ok(exit_status) + } + + /// Execute the IPFS command and return the output. + /// + /// If the command fails, an error is returned. + async fn execute_with_output(mut self) -> StdResult { + let output = self + .command + .output() + .await + .with_context(|| format!("Failed to spawn '{}'", self.context))?; + + if output.status.success() { + Ok(output) + } else { + Err(anyhow!( + "Execution of '{}' failed with status {}: {}", + self.context, + output.status, + String::from_utf8_lossy(&output.stderr) + )) + } + } +} + +impl Deref for IpfsCommand { + type Target = Command; + + fn deref(&self) -> &Self::Target { + &self.command + } +} + +impl DerefMut for IpfsCommand { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.command } +} - fn build_topology(number_of_nodes: u8, swarm_dir: &Path) -> Vec { +impl IpfsDevnet { + fn build_topology( + number_of_nodes: u8, + swarm_dir: &Path, + command_builder: &IpfsCommandBuilder, + ) -> Vec { (1..=number_of_nodes) .map(|n| KuboNode { rpc_url: Url::parse(&format!("http://127.0.0.1:{}", 5000_u16 + n as u16)).unwrap(), working_dir: swarm_dir.join(format!("kubo-node-{n}")), + swarm_index: n, + command_builder: command_builder.clone(), }) .collect() } @@ -102,14 +211,15 @@ impl IpfsDevnet { } } + let command_builder = IpfsCommandBuilder::new(&devnet_script_path, &swarm_dir); Ok(IpfsDevnet { - devnet_script_path, - swarm_dir: bootstrap_args.swarm_target_dir.to_owned(), topology: Self::build_topology( bootstrap_args.number_of_nodes, &bootstrap_args.swarm_target_dir, + &command_builder, ), mode: bootstrap_args.mode, + command_builder, }) } @@ -120,43 +230,31 @@ impl IpfsDevnet { IPFS_DEVNET_SCRIPT_NAME, &bootstrap_args.devnet_scripts_dir, )?; + let command_builder = + IpfsCommandBuilder::new(&devnet_script_path, &bootstrap_args.swarm_target_dir); - let mut bootstrap_command = Command::new(&devnet_script_path); - bootstrap_command - .arg("init") - .arg("--overwrite") - .arg("--number") - .arg(bootstrap_args.number_of_nodes.to_string()) - .arg("--swarm-dir") - .arg(bootstrap_args.swarm_target_dir.as_os_str()); + let mut bootstrap_command = command_builder.build( + "init", + &["--overwrite", "--number", &bootstrap_args.number_of_nodes.to_string()], + ); if let Some(kubo_version) = &bootstrap_args.kubo_version { bootstrap_command.env("KUBO_VERSION", kubo_version.to_string()); } - bootstrap_command - .current_dir(&bootstrap_args.devnet_scripts_dir) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .kill_on_drop(true); + bootstrap_command.current_dir(&bootstrap_args.devnet_scripts_dir); - info!("Bootstrapping the IPFS Devnet"; "script" => &devnet_script_path.display(), "cmd" => "init"); + info!("Bootstrapping the IPFS Devnet"; "script" => command_builder.display_script_path(), "cmd" => "init"); - let exit_status = bootstrap_command - .spawn() - .with_context(|| format!("{IPFS_DEVNET_SCRIPT_NAME} failed to start"))? - .wait_forwarding_output_to_slog_scope(IPFS_DEVNET_SCRIPT_NAME) - .await - .with_context(|| format!("{IPFS_DEVNET_SCRIPT_NAME} failed to run"))?; - match exit_status.code() { + match bootstrap_command.execute().await?.code() { Some(0) => Ok(IpfsDevnet { - devnet_script_path, - swarm_dir: bootstrap_args.swarm_target_dir.to_owned(), topology: Self::build_topology( bootstrap_args.number_of_nodes, &bootstrap_args.swarm_target_dir, + &command_builder, ), mode: bootstrap_args.mode, + command_builder, }), Some(code) => Err(anyhow!(RetryableDevnetError(format!( "IPFS Bootstrap devnet exited with status code: {code}" @@ -165,10 +263,6 @@ impl IpfsDevnet { } } - pub fn swarm_dir(&self) -> PathBuf { - self.swarm_dir.clone() - } - pub fn topology(&self) -> &[KuboNode] { &self.topology } @@ -176,17 +270,11 @@ impl IpfsDevnet { pub async fn start(&self) -> StdResult<()> { // Note: running the start command on an already running devnet does nothing, so running it // against a "Detached" devnet poses no risk (if stopped, it will start, if running, it will do nothing). - let mut run_command = self.build_command("start")?; + let run_command = self.command_builder.build("start", &[]); - info!("Starting the IPFS devnet"; "script" => &self.devnet_script_path.display(), "cmd" => "start"); + info!("Starting the IPFS devnet"; "script" => &self.command_builder.display_script_path(), "cmd" => "start"); - let status = run_command - .spawn() - .with_context(|| "Failed to start the IPFS devnet")? - .wait_forwarding_output_to_slog_scope(&format!("{IPFS_DEVNET_SCRIPT_NAME} start")) - .await - .with_context(|| "Error while starting the IPFS devnet")?; - match status.code() { + match run_command.execute().await?.code() { Some(0) => Ok(()), Some(code) => Err(anyhow!(RetryableDevnetError(format!( "Run IPFS devnet exited with status code: {code}" @@ -201,20 +289,113 @@ impl IpfsDevnet { return Ok(()); } - let mut run_command = self.build_command("stop")?; + let run_command = self.command_builder.build("stop", &[]); - info!("Stopping the IPFS devnet"; "script" => &self.devnet_script_path.display(), "cmd" => "stop"); + info!("Stopping the IPFS devnet"; "script" => &self.command_builder.display_script_path(), "cmd" => "stop"); - let exit_status = run_command - .spawn() - .with_context(|| "Failed to stop the IPFS devnet")? - .wait_forwarding_output_to_slog_scope(&format!("{IPFS_DEVNET_SCRIPT_NAME} stop")) - .await - .with_context(|| "Error while stopping the IPFS devnet")?; - match exit_status.code() { + match run_command.execute().await?.code() { Some(0) => Ok(()), Some(code) => Err(anyhow!("Stop IPFS devnet exited with status code: {code}")), None => Err(anyhow!("Stop IPFS devnet terminated by signal")), } } } + +impl KuboNode { + /// List IPNS keys of the Kubo node keystore + pub async fn list_ipns_keys(&self) -> StdResult> { + // Output lines example: + // k51qzi5uqu5dhmp0n2e4olicqyqure0kk8qq87y8s2o8qyyeah2hugrmo9boq7 self + // k51qzi5uqu5dldmb73ienkcee5fm46ldactwcuu0wzqizcwgas9en94spxr16j mithril-snapshot + let output = self + .command_builder + .build_query_command(self.swarm_index, &["key", "list", "-l"]) + .execute_with_output() + .await?; + + Ok(String::from_utf8_lossy(&output.stdout) + .lines() + .filter_map(|line| line.trim().split_once(char::is_whitespace)) + .map(|(id, name)| IpnsKey { + id: id.to_string(), + name: name.to_string(), + }) + .collect()) + } + + /// Get the id of the IPNS key with the given name + /// + /// Note: The key name is local to the IPFS node and can't be used to resolve IPNS names, only the key id can. + pub async fn get_ipns_key_id(&self, key_name: &str) -> StdResult> { + let keys = self.list_ipns_keys().await?; + Ok(keys.into_iter().find(|key| key.name == key_name).map(|key| key.id)) + } + + pub async fn create_ipns_keypair(&self, key_name: &str) -> StdResult<()> { + if self.get_ipns_key_id(key_name).await?.is_some() { + warn!("Create IPNS keypair - key with name '{key_name}' already exists, reusing it"); + return Ok(()); + } + + let query_command = self.command_builder.build_query_command( + self.swarm_index, + &["key", "gen", "--type=ed25519", key_name], + ); + + info!("Creating IPNS keypair"; "script" => &self.command_builder.display_script_path(), "cmd" => "query -- key gen"); + + match query_command.execute().await?.code() { + Some(0) => Ok(()), + Some(code) => Err(anyhow!(RetryableDevnetError(format!( + "Create IPNS keypair exited with status code: {code}" + )))), + None => Err(anyhow!("Create IPNS keypair terminated by signal")), + } + } + + /// Resolve an **IPNS** name to an IPFS path. + pub async fn ipns_name_resolve(&self, ipns_name: &str) -> StdResult { + // Example output: + // /ipfs/QmfYpbWwaAmrTfapNZHYDkutDxckvqA5hzjrbfYDwymaKw + let output = self + .command_builder + .build_query_command( + self.swarm_index, + &["name", "resolve", "--nocache", ipns_name], + ) + .execute_with_output() + .await?; + + Ok(String::from_utf8_lossy(&output.stdout).trim().to_string()) + } + + /// Resolve an **IPFS** path (useful to retrieve the CID of a file if you only know the path and its folder CID). + pub async fn ipfs_resolve(&self, ipfs_path: &str) -> StdResult { + // Example output: + // /ipfs/bafkreibeffv3u3yqulwvgxqjcfsigocas35t5rjykzk5nmyev2x2fzwe3y + let output = self + .command_builder + .build_query_command(self.swarm_index, &["resolve", ipfs_path]) + .execute_with_output() + .await?; + + Ok(String::from_utf8_lossy(&output.stdout).trim().to_string()) + } + + /// Resolve the given IPNS name and check that the resolved directory contains the given file, + /// returning the resolved directory path. + pub async fn resolve_file_in_ipns_dir( + &self, + ipns_name: &str, + ipfs_file_name: &str, + ) -> StdResult { + // Note: we could use `ipfs resolve` directly with the ipns key id, but this would make the + // query prone to cache issues since this request does not allow ignoring the cache (contrary + // to `ipfs name resolve` which has a `--nocache` option). + let ipfs_dir_path = self.ipns_name_resolve(ipns_name).await?; + self.ipfs_resolve(&format!("{ipfs_dir_path}/{ipfs_file_name}")) + .await?; + + Ok(ipfs_dir_path) + } +} diff --git a/mithril-test-lab/mithril-end-to-end/src/devnet/mod.rs b/mithril-test-lab/mithril-end-to-end/src/devnet/mod.rs index 9a34dd90621..1a01bbbf268 100644 --- a/mithril-test-lab/mithril-end-to-end/src/devnet/mod.rs +++ b/mithril-test-lab/mithril-end-to-end/src/devnet/mod.rs @@ -2,7 +2,7 @@ mod cardano; mod ipfs; pub use cardano::{Devnet, DevnetBootstrapArgs, DevnetTopology, FullNode, PoolNode}; -pub use ipfs::{IpfsDevnet, IpfsDevnetBootstrapArgs, IpfsDevnetMode, KuboNode}; +pub use ipfs::{IpfsDevnet, IpfsDevnetBootstrapArgs, IpfsDevnetMode, IpnsKey, KuboNode}; #[derive(thiserror::Error, Debug, PartialEq, Eq)] #[error("Retryable devnet error: `{0}`")] diff --git a/mithril-test-lab/mithril-end-to-end/src/mithril/aggregator.rs b/mithril-test-lab/mithril-end-to-end/src/mithril/aggregator.rs index b6367959e70..e723fcf348c 100644 --- a/mithril-test-lab/mithril-end-to-end/src/mithril/aggregator.rs +++ b/mithril-test-lab/mithril-end-to-end/src/mithril/aggregator.rs @@ -19,7 +19,7 @@ use crate::utils::{EnvVars, MithrilCommand, NodeVersion}; use crate::{ ANCILLARY_MANIFEST_SECRET_KEY, AggregateSignatureType, DEVNET_DMQ_MAGIC_ID, DEVNET_MAGIC_ID, DmqNodeFlavor, ERA_MARKERS_SECRET_KEY, ERA_MARKERS_VERIFICATION_KEY, FullNode, GenesisKeys, - KuboNode, PROTOCOL_CONFIGURATION_MARKERS_SECRET_KEY, + IPNS_KEY_NAME_FOR_SNAPSHOT, KuboNode, PROTOCOL_CONFIGURATION_MARKERS_SECRET_KEY, PROTOCOL_CONFIGURATION_MARKERS_VERIFICATION_KEY, RetryableDevnetError, }; @@ -64,6 +64,7 @@ pub struct Aggregator { process: RwLock>, chain_observer: Arc, full_node: FullNode, + ipfs_kubo_node: Option, aggregate_signature_type: AggregateSignatureType, startup_protocol_parameters: ProtocolParameters, } @@ -223,8 +224,8 @@ impl Aggregator { env.insert( "IPFS_RPC_SERVER_CONFIG", format!( - r#"{{"url": "{}", "mfs_folder_name": "mithril-snapshots"}}"#, - node.rpc_url + r#"{{"url": "{}", "mfs_folder_name": "mithril-snapshots", "ipns_key_name": "{}"}}"#, + node.rpc_url, IPNS_KEY_NAME_FOR_SNAPSHOT ), ); } @@ -258,6 +259,7 @@ impl Aggregator { process: RwLock::new(None), chain_observer, full_node: aggregator_config.full_node.clone(), + ipfs_kubo_node: aggregator_config.ipfs_kubo_node.cloned(), aggregate_signature_type: aggregator_config.aggregate_signature_type, startup_protocol_parameters: aggregator_config.startup_protocol_parameters.clone(), }) @@ -324,6 +326,7 @@ impl Aggregator { process: RwLock::new(None), chain_observer: other.chain_observer.clone(), full_node: other.full_node.clone(), + ipfs_kubo_node: other.ipfs_kubo_node.clone(), aggregate_signature_type: other.aggregate_signature_type, startup_protocol_parameters: other.startup_protocol_parameters.clone(), } @@ -365,6 +368,10 @@ impl Aggregator { &self.full_node } + pub fn ipfs_kubo_node(&self) -> Option<&KuboNode> { + self.ipfs_kubo_node.as_ref() + } + /// Get the version of the mithril-aggregator binary. pub fn version(&self) -> &NodeVersion { &self.version diff --git a/mithril-test-lab/mithril-end-to-end/src/mithril/infrastructure.rs b/mithril-test-lab/mithril-end-to-end/src/mithril/infrastructure.rs index 231c88d63b7..27e0626a20e 100644 --- a/mithril-test-lab/mithril-end-to-end/src/mithril/infrastructure.rs +++ b/mithril-test-lab/mithril-end-to-end/src/mithril/infrastructure.rs @@ -1,3 +1,4 @@ +use anyhow::Context; use serde::{Deserialize, Serialize}; use slog_scope::info; use std::collections::BTreeSet; @@ -18,8 +19,8 @@ use crate::mithril::relay_signer::RelaySignerConfiguration; use crate::toolkit::ScenarioToolkit; use crate::{ AggregateSignatureType, Aggregator, AggregatorConfig, Client, DEVNET_MAGIC_ID, Devnet, - DmqNodeFlavor, FullNode, GenesisKeys, IpfsDevnet, KuboNode, PoolNode, RelayAggregator, - RelayPassive, RelaySigner, Signer, + DmqNodeFlavor, FullNode, GenesisKeys, IPNS_KEY_NAME_FOR_SNAPSHOT, IpfsDevnet, KuboNode, + PoolNode, RelayAggregator, RelayPassive, RelaySigner, Signer, }; use super::signer::SignerConfig; @@ -376,6 +377,17 @@ impl MithrilInfrastructure { certificate_chain_aggregator_endpoint: Option, ) -> StdResult { let aggregator_name = Aggregator::name_suffix(index); + let is_leader = index == 0; + + if is_leader && let Some(kubo_node) = ipfs_kubo_node { + kubo_node + .create_ipns_keypair(IPNS_KEY_NAME_FOR_SNAPSHOT) + .await + .with_context(|| { + format!("Failed to create IPNS keypair for aggregator {aggregator_name}") + })?; + } + let aggregator_artifacts_dir = config .artifacts_dir .join(format!("mithril-aggregator-{aggregator_name}")); diff --git a/mithril-test-lab/mithril-end-to-end/src/mithril/mod.rs b/mithril-test-lab/mithril-end-to-end/src/mithril/mod.rs index bae555991a1..d61b03527b5 100644 --- a/mithril-test-lab/mithril-end-to-end/src/mithril/mod.rs +++ b/mithril-test-lab/mithril-end-to-end/src/mithril/mod.rs @@ -31,3 +31,5 @@ pub const ANCILLARY_MANIFEST_VERIFICATION_KEY: &str = "5b33322c3235332c3138362c3 pub const ANCILLARY_MANIFEST_SECRET_KEY: &str = "5b3131382c3138342c3232342c3137332c3136302c3234312c36312c3134342c36342c39332c3130362c3232392c38332c3133342c3138392c34302c3138392c3231302c32352c3138342c3136302c3134312c3233372c32362c3136382c35342c3233392c3230342c3133392c3131392c31332c3139395d"; pub const PROTOCOL_CONFIGURATION_MARKERS_VERIFICATION_KEY: &str = "5b33322c3235332c3138362c3230312c3137372c31312c3131372c3133352c3138372c3136372c3138312c3138382c32322c35392c3230362c3130352c3233312c3135302c3231352c33302c37382c3231322c37362c31362c3235322c3138302c37322c3133342c3133372c3234372c3136312c36385d"; pub const PROTOCOL_CONFIGURATION_MARKERS_SECRET_KEY: &str = "5b3131382c3138342c3232342c3137332c3136302c3234312c36312c3134342c36342c39332c3130362c3232392c38332c3133342c3138392c34302c3138392c3231302c32352c3138342c3136302c3134312c3233372c32362c3136382c35342c3233392c3230342c3133392c3131392c31332c3139395d"; + +pub const IPNS_KEY_NAME_FOR_SNAPSHOT: &str = "mithril-snapshot"; diff --git a/mithril-test-lab/mithril-end-to-end/src/scenario/full.rs b/mithril-test-lab/mithril-end-to-end/src/scenario/full.rs index 52f6a7d7a4f..2b7bdc2fefe 100644 --- a/mithril-test-lab/mithril-end-to-end/src/scenario/full.rs +++ b/mithril-test-lab/mithril-end-to-end/src/scenario/full.rs @@ -386,7 +386,7 @@ impl FullScenario { .is_certified_and_verified( aggregator, &mut client, - infrastructure.client_ipfs_node().map(|n| n.rpc_url.clone()), + infrastructure.client_ipfs_node(), expected_epoch_min, infrastructure.signers().len(), ) diff --git a/mithril-test-lab/mithril-end-to-end/src/scenario/minimal.rs b/mithril-test-lab/mithril-end-to-end/src/scenario/minimal.rs index 2757cb3e66b..2fb09681b0a 100644 --- a/mithril-test-lab/mithril-end-to-end/src/scenario/minimal.rs +++ b/mithril-test-lab/mithril-end-to-end/src/scenario/minimal.rs @@ -185,7 +185,7 @@ impl MinimalScenario { .is_certified_and_verified( aggregator, &mut client, - infrastructure.client_ipfs_node().map(|n| n.rpc_url.clone()), + infrastructure.client_ipfs_node(), expected_epoch_min, infrastructure.signers().len(), ) diff --git a/mithril-test-lab/mithril-end-to-end/src/toolkit/check/cardano_database.rs b/mithril-test-lab/mithril-end-to-end/src/toolkit/check/cardano_database.rs index eac8340f64f..ee7a02640b3 100644 --- a/mithril-test-lab/mithril-end-to-end/src/toolkit/check/cardano_database.rs +++ b/mithril-test-lab/mithril-end-to-end/src/toolkit/check/cardano_database.rs @@ -4,12 +4,14 @@ use slog_scope::{info, warn}; use mithril_common::{ StdResult, - entities::{Epoch, EpochSpecifier}, + entities::{Epoch, EpochSpecifier, ImmutableFileNumber}, messages::{CardanoDatabaseDigestListMessage, CardanoDatabaseSnapshotListItemMessage}, }; use crate::{ - Aggregator, CardanoDbV2Command, Client, ClientCommand, poll_until, + Aggregator, CardanoDbV2Command, Client, ClientCommand, KuboNode, + mithril::IPNS_KEY_NAME_FOR_SNAPSHOT, + poll_until, toolkit::{CheckCertificateToolkit, ScenarioToolkitContext}, utils::AttemptResult, }; @@ -30,7 +32,7 @@ impl CheckCardanoDatabaseToolkit { &self, aggregator: &Aggregator, client: &mut Client, - ipfs_rpc_url: Option, + client_ipfs_node: Option<&KuboNode>, expected_epoch_min: Epoch, total_signers_expected: usize, ) -> StdResult<()> { @@ -45,7 +47,30 @@ impl CheckCardanoDatabaseToolkit { ) .await?; self.node_producing_cardano_database_digests_map(aggregator).await?; - self.verify_with_client(client, ipfs_rpc_url, &artifact.hash).await?; + self.verify_with_client( + client, + client_ipfs_node.map(|n| n.rpc_url.clone()), + &artifact.hash, + ) + .await?; + + if aggregator.is_leader() + && let Some(client_kubo_node) = client_ipfs_node + { + let aggregator_kubo_node = aggregator.ipfs_kubo_node().with_context(|| { + format!( + "IPFS devnet is enabled but {} has no Kubo node", + aggregator.name() + ) + })?; + self.check_immutable_published_to_ipns( + aggregator, + aggregator_kubo_node, + client_kubo_node, + artifact.beacon.immutable_file_number, + ) + .await?; + } Ok(()) } @@ -169,4 +194,71 @@ impl CheckCardanoDatabaseToolkit { Ok(()) } + + pub async fn check_immutable_published_to_ipns( + &self, + aggregator: &Aggregator, + aggregator_kubo_node: &KuboNode, + client_kubo_node: &KuboNode, + immutable_file_number: ImmutableFileNumber, + ) -> StdResult<()> { + info!( + "Checking IPNS publication of immutable file #{immutable_file_number}"; + "aggregator" => &aggregator.name(), + ); + + let ipns_key_id = aggregator_kubo_node + .get_ipns_key_id(IPNS_KEY_NAME_FOR_SNAPSHOT) + .await? + .with_context(|| { + format!( + "Failed to get IPNS key ID for key '{IPNS_KEY_NAME_FOR_SNAPSHOT}' and {}", + aggregator.name() + ) + })?; + info!("IPNS key ID for key '{IPNS_KEY_NAME_FOR_SNAPSHOT}': '{ipns_key_id}'"; "aggregator" => &aggregator.name()); + + let ipns_name = format!("/ipns/{ipns_key_id}"); + let ipfs_file_name_for_immutable = format!("{immutable_file_number:0>5}.tar.zst"); + let mut last_error = None; + + match poll_until!( + self.context.existing_artifact_fetch_timeout(), + self.context.poll_backoff(), + { + match client_kubo_node.resolve_file_in_ipns_dir( + &ipns_name, + &ipfs_file_name_for_immutable, + ) + .await + { + Ok(ipfs_dir_path) => Ok(Some(ipfs_dir_path)), + Err(error) => { + warn!("IPNS publication not yet available, retrying"; "error" => ?error, "aggregator" => &aggregator.name()); + last_error = Some(error); + Ok(None) + } + } + } + ) { + AttemptResult::Ok(ipfs_dir_path) => { + info!( + "Success, '{ipns_name}' resolves to '{ipfs_dir_path}' that contains immutable {immutable_file_number:0>5}"; + "aggregator" => &aggregator.name() + ); + Ok(()) + } + AttemptResult::Err(error) => Err(error), + AttemptResult::Timeout(..) => { + let error = anyhow!( + "Timeout exhausted while checking that '{ipns_name}' contains immutable {immutable_file_number:0>5}" + ); + Err(match last_error { + Some(last_error) => last_error.context(error), + None => error, + }) + } + } + .with_context(|| format!("Checking IPNS publication of aggregator `{}`", aggregator.name())) + } }