From 945664dee487a8e30b1308722c9d532294d22b82 Mon Sep 17 00:00:00 2001 From: DJO <790521+Alenar@users.noreply.github.com> Date: Wed, 9 Sep 2026 18:11:54 +0200 Subject: [PATCH 1/4] refactor(ipfs): standardize IPFS URI handling and simplify IPFS downloader Using `ipfs://` url scheme is more standard and allows usage as is, without prefixing, with both `files/stat` and `cat` kubo RPC endpoints. --- .../cardano_database_artifacts/immutable.rs | 34 +++++++++++++- .../src/file_uploaders/ipfs_uploader.rs | 2 +- mithril-client/src/file_downloader/ipfs.rs | 44 ++++++++----------- 3 files changed, 52 insertions(+), 28 deletions(-) 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 c894dc89ce3..8e66539eaac 100644 --- a/mithril-aggregator/src/artifact_builder/cardano_database_artifacts/immutable.rs +++ b/mithril-aggregator/src/artifact_builder/cardano_database_artifacts/immutable.rs @@ -112,7 +112,7 @@ impl ImmutableFilesUploader for IpfsUploader { Ok(ImmutablesLocation::Ipfs { uri: MultiFilesUri::Template(TemplateUri(format!( - "{directory_cid}/{{immutable_file_number}}.tar.zst" + "ipfs://{directory_cid}/{{immutable_file_number}}.tar.zst" ))), compression_algorithm, }) @@ -299,6 +299,7 @@ mod tests { use mithril_cardano_node_internal_database::test::DummyCardanoDbBuilder; use mithril_common::{ entities::TemplateUri, + temp_dir_create, test::{TempDir, assert_equivalent, equivalent_to}, }; use mithril_file_archiver::FileArchiver; @@ -887,6 +888,8 @@ mod tests { } mod batch_upload { + use std::collections::HashMap; + use mithril_common::test::TempDir; use crate::file_uploaders::FileUploadRetryPolicy; @@ -971,6 +974,35 @@ mod tests { .await .expect_err("Should return an error when not template found"); } + + #[tokio::test] + async fn ipfs_batch_upload_yield_ipfs_urls() { + let test_dir = temp_dir_create!(); + let uploader = IpfsUploader::new_for_test("dir", move |mock| { + mock.expect_create_dir().returning(|_| Ok(())); + mock.expect_list_directory_files() + .returning(move |_| Ok(HashMap::new())); + mock.expect_file_exists().never(); + mock.expect_upload_file().returning(|_, _| Ok("file-cid".to_string())); + mock.expect_get_dir_cid() + .returning(|_| Ok("directory-cid".to_string())); + }); + + let archive_1 = create_fake_archive(&test_dir, "00001.tar.zst"); + + let location = IpfsUploader::batch_upload(&uploader, &[archive_1], None) + .await + .unwrap(); + assert_eq!( + ImmutablesLocation::Ipfs { + uri: MultiFilesUri::Template(TemplateUri( + "ipfs://directory-cid/{immutable_file_number}.tar.zst".to_string() + )), + compression_algorithm: None + }, + location + ); + } } mod immutable_file_number_extractor { diff --git a/mithril-aggregator/src/file_uploaders/ipfs_uploader.rs b/mithril-aggregator/src/file_uploaders/ipfs_uploader.rs index ffdf51434b8..3fe6c2061f2 100644 --- a/mithril-aggregator/src/file_uploaders/ipfs_uploader.rs +++ b/mithril-aggregator/src/file_uploaders/ipfs_uploader.rs @@ -235,7 +235,7 @@ mod tests { use super::*; impl IpfsUploader { - fn new_for_test>( + pub(crate) fn new_for_test>( mfs_dir: P, mock_config: impl FnOnce(&mut MockIpfsBackendUploader), ) -> Self { diff --git a/mithril-client/src/file_downloader/ipfs.rs b/mithril-client/src/file_downloader/ipfs.rs index ac308f335e3..093ef9138a7 100644 --- a/mithril-client/src/file_downloader/ipfs.rs +++ b/mithril-client/src/file_downloader/ipfs.rs @@ -76,8 +76,8 @@ impl IpfsFileDownloader { .with_context(|| format!("Could not build Kubo RPC endpoint for route '{route}'")) } - /// POST to a Kubo RPC `route` with the given IPFS path (`/`) as its - /// `arg` query parameter. + /// POST to a Kubo RPC `route` with the given IPFS path (`ipfs:///`) as + /// its `arg` query parameter. async fn post( &self, route: &str, @@ -135,7 +135,7 @@ impl IpfsFileDownloader { let response = self .post( "api/v0/files/stat", - &with_ipfs_namespace_prefix(ipfs_path), + ipfs_path, Some(self.existence_check_timeout), ) .await?; @@ -163,11 +163,6 @@ impl IpfsFileDownloader { } } -// files/stat does not work against directory CID without the `/ipfs/` prefix -fn with_ipfs_namespace_prefix(cid: &str) -> String { - format!("/ipfs/{cid}") -} - #[async_trait] impl FileDownloader for IpfsFileDownloader { async fn download_unpack( @@ -178,9 +173,7 @@ impl FileDownloader for IpfsFileDownloader { compression_algorithm: Option, download_event_type: DownloadEvent, ) -> StdResult<()> { - // `location` is a path in the form `/` (confirmed against a live - // Kubo node), which is exactly the `arg` shape `cat` expect and only need to be prefixed - // with `/ipfs/` for `files/stat`. + // `location` is expected a path in the form `ipfs:///`. let ipfs_path = location.as_str(); let downloaded = self.open_stream(ipfs_path).await?; @@ -225,12 +218,12 @@ mod tests { let target_dir = temp_dir_create!(); let content = "Hello, world!"; let size = content.len() as u64; - let ipfs_path = "QmDummyDirCid/00006.tar.zst"; + let ipfs_path = "ipfs://QmDummyDirCid/00006.tar.zst"; let server = MockServer::start(); server.mock(|when, then| { when.method(POST) .path("/api/v0/files/stat") - .query_param("arg", with_ipfs_namespace_prefix(ipfs_path)); + .query_param("arg", ipfs_path); then.status(200) .json_body(serde_json::json!({"Key": "bafkreidummy", "Size": size})); }); @@ -277,10 +270,10 @@ mod tests { #[tokio::test] async fn missing_block_reported_through_500_body_is_turned_into_a_not_found_error() { let target_dir = temp_dir_create!(); - let ipfs_path = "QmRoiDvkuGRg4tjWabNp4Y5jxbUS8FFNFn9pDopqfbtfW2/00007.tar.zst"; + let ipfs_path = "ipfs://QmRoiDvkuGRg4tjWabNp4Y5jxbUS8FFNFn9pDopqfbtfW2/00007.tar.zst"; let server = MockServer::start(); server.mock(|when, then| { - when.method(POST).path("/api/v0/files/stat").query_param("arg", with_ipfs_namespace_prefix(ipfs_path)); + when.method(POST).path("/api/v0/files/stat").query_param("arg", ipfs_path); then.status(500).json_body(serde_json::json!({ "Message": "no link named \"00007.tar.zst\" under QmRoiDvkuGRg4tjWabNp4Y5jxbUS8FFNFn9pDopqfbtfW2", "Code": 0, @@ -303,10 +296,9 @@ mod tests { .unwrap_err(); assert!( - error.to_string().contains(&format!( - "Location='{}' not found", - with_ipfs_namespace_prefix(ipfs_path) - )), + error + .to_string() + .contains(&format!("Location='{ipfs_path}' not found")), "unexpected error: {error:?}" ); } @@ -314,12 +306,12 @@ mod tests { #[tokio::test] async fn files_stat_timeout_raise_unreachable_file_downloader_error() { let target_dir = temp_dir_create!(); - let ipfs_path = "QmRoiDvkuGRg4tjWabNp4Y5jxbUS8FFNFn9pDopqfbtfW2/00007.tar.zst"; + let ipfs_path = "ipfs://QmRoiDvkuGRg4tjWabNp4Y5jxbUS8FFNFn9pDopqfbtfW2/00007.tar.zst"; let server = MockServer::start(); server.mock(|when, then| { when.method(POST) .path("/api/v0/files/stat") - .query_param("arg", with_ipfs_namespace_prefix(ipfs_path)); + .query_param("arg", ipfs_path); then.delay(Duration::from_millis(100)); }); let ipfs_file_downloader = downloader(&server, FeedbackSender::new(&[])) @@ -341,7 +333,7 @@ mod tests { assert_eq!( Some(&FileDownloaderUnreachable { source: "IPFS", - uri: with_ipfs_namespace_prefix(ipfs_path) + uri: ipfs_path.to_string(), }), error.downcast_ref::() ); @@ -350,13 +342,13 @@ mod tests { #[tokio::test] async fn cat_does_not_apply_timeout() { let target_dir = temp_dir_create!(); - let ipfs_path = "QmRoiDvkuGRg4tjWabNp4Y5jxbUS8FFNFn9pDopqfbtfW2/00007.tar.zst"; + let ipfs_path = "ipfs://QmRoiDvkuGRg4tjWabNp4Y5jxbUS8FFNFn9pDopqfbtfW2/00007.tar.zst"; let server = MockServer::start(); server.mock(|when, then| { when.method(POST) .path("/api/v0/files/stat") - .query_param("arg", with_ipfs_namespace_prefix(ipfs_path)); + .query_param("arg", ipfs_path); then.status(200) .json_body(serde_json::json!({"Key": "bafkreidummy", "Size": 1})); }); @@ -384,12 +376,12 @@ mod tests { #[tokio::test] async fn context_deadline_exceeded_is_not_mistaken_for_a_missing_file() { let target_dir = temp_dir_create!(); - let ipfs_path = "QmRoiDvkuGRg4tjWabNp4Y5jxbUS8FFNFn9pDopqfbtfW2/fake-cardano-cli.sh"; + let ipfs_path = "ipfs://QmRoiDvkuGRg4tjWabNp4Y5jxbUS8FFNFn9pDopqfbtfW2/fake-cardano-cli.sh"; let server = MockServer::start(); server.mock(|when, then| { when.method(POST) .path("/api/v0/files/stat") - .query_param("arg", with_ipfs_namespace_prefix(ipfs_path)); + .query_param("arg", ipfs_path); then.status(500).json_body(serde_json::json!({ "Message": "context deadline exceeded", "Code": 0, From 97a1c42efed2eedcb5aa20033771ca2837fff272 Mon Sep 17 00:00:00 2001 From: DJO <790521+Alenar@users.noreply.github.com> Date: Wed, 9 Sep 2026 18:13:08 +0200 Subject: [PATCH 2/4] fix(ipfs-devnet): peer configuration was not using the updated port making the nodes unable to connect to each other. --- .../ipfs-devnet/commands/mkfiles/kubo-configure-swarm.sh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 561dfccea43..060cfe9bd65 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 @@ -117,7 +117,7 @@ configure_node_peers() { fi peer_id="${peer_ids[$((peer_node_id - 1))]}" - swarm_port=$((4000 + peer_node_id)) + swarm_port=$((5200 + peer_node_id)) peers_json="${peers_json}${separator}{\"ID\":\"${peer_id}\",\"Addrs\":[\"/ip4/127.0.0.1/tcp/${swarm_port}\"]}" separator="," From ba02bde54b816f8fcc0372ba8d14a7da72c4172f Mon Sep 17 00:00:00 2001 From: DJO <790521+Alenar@users.noreply.github.com> Date: Wed, 9 Sep 2026 18:26:36 +0200 Subject: [PATCH 3/4] fix(aggregator): avoid double logs of single signatures after validation error the two validators (`/register_signatures` http route handler and `SignatureProcessor`) already include them in a "full_payload" log trace. So we do not need to repeat it, especially since they quite huge, this polute the logs. --- mithril-aggregator/src/multi_signer.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/mithril-aggregator/src/multi_signer.rs b/mithril-aggregator/src/multi_signer.rs index b5e9fa38ab9..e890106512b 100644 --- a/mithril-aggregator/src/multi_signer.rs +++ b/mithril-aggregator/src/multi_signer.rs @@ -78,7 +78,7 @@ impl MultiSignerImpl { protocol_multi_signer .verify_single_signature(&message, single_signature) .with_context(|| { - format!("Multi Signer can not verify single signature for message '{message:?}' and single signature {single_signature:#?}") + format!("Multi Signer can not verify single signature for message '{message:?}' from party_id '{}'", single_signature.party_id) }) } } From f0c164f61da0f106bc77eebfc69b29f305d99c8f Mon Sep 17 00:00:00 2001 From: DJO <790521+Alenar@users.noreply.github.com> Date: Wed, 9 Sep 2026 18:33:22 +0200 Subject: [PATCH 4/4] fix(aggregator): add missing compute cache pool at startup for prover service Else if an aggregator restarts, it won't be able to compute proofs for "blocks and transaction" artifacts until a new artifact is produced (which can take several minutes). --- .../builder/protocol/artifacts.rs | 34 ++++++++++++++----- 1 file changed, 26 insertions(+), 8 deletions(-) diff --git a/mithril-aggregator/src/dependency_injection/builder/protocol/artifacts.rs b/mithril-aggregator/src/dependency_injection/builder/protocol/artifacts.rs index 2b1c3768489..ccc24096cc7 100644 --- a/mithril-aggregator/src/dependency_injection/builder/protocol/artifacts.rs +++ b/mithril-aggregator/src/dependency_injection/builder/protocol/artifacts.rs @@ -3,6 +3,7 @@ use semver::Version; use std::path::PathBuf; use std::sync::Arc; +use mithril_common::StdResult; use mithril_common::crypto_helper::ManifestSigner; use mithril_file_archiver::FileArchiver; @@ -78,16 +79,33 @@ impl DependenciesBuilder { logger, )); - // Compute the cache pool for prover service + // Compute the cache pool for both new and legacy prover services // This is done here to avoid circular dependencies between the prover service and the signed entity service // TODO: Make this part of a warmup phase of the aggregator? - if let Some(signed_entity) = - signed_entity_service.get_last_cardano_transaction_snapshot().await? - { - legacy_prover_service - .compute_cache(signed_entity.artifact.block_number) - .await?; - } + let warm_up_legacy_cache = async { + if let Some(signed_entity) = + signed_entity_service.get_last_cardano_transaction_snapshot().await? + { + legacy_prover_service + .compute_cache(signed_entity.artifact.block_number) + .await?; + } + StdResult::Ok(()) + }; + + let warm_up_cache = async { + if let Some(signed_entity) = signed_entity_service + .get_last_cardano_blocks_transactions_snapshot() + .await? + { + prover_service + .compute_cache(signed_entity.artifact.block_number_signed) + .await?; + } + StdResult::Ok(()) + }; + + tokio::try_join!(warm_up_legacy_cache, warm_up_cache)?; Ok(signed_entity_service) }