feat: add the MultidappClaimer as an experimental feature for sunodo

CHANGELOG.md

@@ -8,6 +8,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Added support for the experimental multidapp claimer.
+- Added the `DAPP_CONTRACT_ADDRESS` environment variable to the `authority-claimer`. If let unset, the service instantiates the MultidappClaimer, that reads dapp addresses from Redis.
+
 ### Changed
 
 - Disabled the `authority-claimer` when `CARTESI_EXPERIMENTAL_SUNODO_VALIDATOR_ENABLED` is set to `true`.

internal/node/services.go

@@ -117,6 +117,8 @@ func newAuthorityClaimer(c config.NodeConfig, workDir string) services.CommandSe
 	s.Env = append(s.Env, fmt.Sprintf("HISTORY_ADDRESS=%v", c.ContractsHistoryAddress))
 	s.Env = append(s.Env, fmt.Sprintf("AUTHORITY_ADDRESS=%v", c.ContractsAuthorityAddress))
 	s.Env = append(s.Env, fmt.Sprintf("INPUT_BOX_ADDRESS=%v", c.ContractsInputBoxAddress))
+	s.Env = append(s.Env, fmt.Sprintf("DAPP_CONTRACT_ADDRESS=%v",
+		c.ContractsApplicationAddress))
 	s.Env = append(s.Env, fmt.Sprintf("GENESIS_BLOCK=%v",
 		c.ContractsInputBoxDeploymentBlockNumber))
 	s.Env = append(s.Env, fmt.Sprintf("AUTHORITY_CLAIMER_HTTP_SERVER_PORT=%v",

offchain/advance-runner/src/broker.rs

@@ -55,7 +55,7 @@ impl BrokerFacade {
         let client = Broker::new(config).await.context(BrokerInternalSnafu)?;
         let inputs_stream = RollupsInputsStream::new(&dapp_metadata);
         let outputs_stream = RollupsOutputsStream::new(&dapp_metadata);
-        let claims_stream = RollupsClaimsStream::new(dapp_metadata.chain_id);
+        let claims_stream = RollupsClaimsStream::new(&dapp_metadata);
         Ok(Self {
             client,
             inputs_stream,

offchain/authority-claimer/README.md

@@ -2,3 +2,27 @@
 
 This service submits rollups claims consumed from the broker to the blockchain using the [tx-manager crate](https://github.com/cartesi/tx-manager).
 It runs at the end of every epoch, when new claims are inserted on the broker.
+
+### Multi-dapp Mode
+
+(This is an **experimental** feature! Don't try it unless you know exactly what you are doing!)
+
+The `authority-claimer` can be configured to run in "multidapp mode".
+To do so, the `DAPP_CONTRACT_ADDRESS` environment variable must be left unset.
+This will force the claimer to instantiate a `MultidappBrokerListener` instead of a `DefaultBrokerListener`.
+
+In multidapp mode, the claimer reads claims from the broker for multiple applications.
+All dapps must share the same History contract and the same chain ID.
+
+Instead of using evironment variables,
+    the claimer will get the list of application addresses from Redis,
+    through the `experimental-dapp-addresses-config` key.
+You must set this key with a string of comma separated (`", "`)
+    hex encoded addresses (without `"0x"`)
+    **before** starting the `authority-claimer`.
+
+You may rewrite the list of addresses at any time,
+The claimer will adjust accordingly.
+The `authority-claimer` stops with an error if the list is empty.
+
+Example key value: `"0202020202020202020202020202020202020202, 0505050505050505050505050505050505050505"`.

offchain/authority-claimer/src/claimer.rs

@@ -4,7 +4,9 @@
 use async_trait::async_trait;
 use snafu::ResultExt;
 use std::fmt::Debug;
-use tracing::{info, trace};
+use tracing::{debug, info};
+
+use rollups_events::Address;
 
 use crate::{
     checker::DuplicateChecker, listener::BrokerListener,
@@ -31,6 +33,9 @@ pub enum ClaimerError<
     D: DuplicateChecker,
     T: TransactionSender,
 > {
+    #[snafu(display("invalid app address {:?}", app_address))]
+    InvalidAppAddress { app_address: Address },
+
     #[snafu(display("broker listener error"))]
     BrokerListenerError { source: B::Error },
 
@@ -84,22 +89,22 @@ where
     type Error = ClaimerError<B, D, T>;
 
     async fn start(mut self) -> Result<(), Self::Error> {
-        trace!("Starting the authority claimer loop");
+        debug!("Starting the authority claimer loop");
         loop {
             let rollups_claim = self
                 .broker_listener
                 .listen()
                 .await
                 .context(BrokerListenerSnafu)?;
-            trace!("Got a claim from the broker: {:?}", rollups_claim);
+            debug!("Got a claim from the broker: {:?}", rollups_claim);
 
             let is_duplicated_rollups_claim = self
                 .duplicate_checker
                 .is_duplicated_rollups_claim(&rollups_claim)
                 .await
                 .context(DuplicatedClaimSnafu)?;
             if is_duplicated_rollups_claim {
-                trace!("It was a duplicated claim");
+                info!("It was a duplicated claim");
                 continue;
             }
 

offchain/authority-claimer/src/config/cli.rs

@@ -6,9 +6,10 @@ use eth_tx_manager::{
     config::{TxEnvCLIConfig as TxManagerCLIConfig, TxManagerConfig},
     Priority,
 };
+use ethers::utils::hex;
 use log::{LogConfig, LogEnvCliConfig};
 use redacted::Redacted;
-use rollups_events::{BrokerCLIConfig, BrokerConfig};
+use rollups_events::{Address, BrokerCLIConfig, BrokerConfig};
 use rusoto_core::Region;
 use snafu::ResultExt;
 use std::{fs, str::FromStr};
@@ -47,6 +48,10 @@ pub(crate) struct AuthorityClaimerCLI {
     #[command(flatten)]
     pub contracts_config: ContractsCLIConfig,
 
+    /// Address of rollups dapp
+    #[arg(long, env)]
+    pub dapp_contract_address: Option<String>,
+
     /// Genesis block for reading blockchain events
     #[arg(long, env, default_value_t = 1)]
     pub genesis_block: u64,
@@ -72,13 +77,25 @@ impl TryFrom<AuthorityClaimerCLI> for AuthorityClaimerConfig {
             ContractsConfig::try_from(cli_config.contracts_config)
                 .context(ContractsSnafu)?;
 
+        let dapp_contract_address: Option<Address> = cli_config
+            .dapp_contract_address
+            .map(|raw_dapp_contract_address| {
+                let address: [u8; 20] =
+                    hex::decode(&raw_dapp_contract_address[2..])
+                        .expect("Dapp json parse error")
+                        .try_into()
+                        .expect("Dapp address with wrong size");
+                address.into()
+            });
+
         Ok(AuthorityClaimerConfig {
             tx_manager_config,
             tx_signing_config,
             tx_manager_priority: Priority::Normal,
             broker_config,
             log_config,
             contracts_config,
+            dapp_address: dapp_contract_address,
             genesis_block: cli_config.genesis_block,
         })
     }

offchain/authority-claimer/src/config/mod.rs

@@ -13,7 +13,7 @@ use eth_tx_manager::{config::TxManagerConfig, Priority};
 use http_server::HttpServerConfig;
 use log::LogConfig;
 use redacted::Redacted;
-use rollups_events::BrokerConfig;
+use rollups_events::{Address, BrokerConfig};
 use rusoto_core::Region;
 
 #[derive(Debug, Clone)]
@@ -30,6 +30,7 @@ pub struct AuthorityClaimerConfig {
     pub broker_config: BrokerConfig,
     pub log_config: LogConfig,
     pub contracts_config: ContractsConfig,
+    pub dapp_address: Option<Address>,
     pub genesis_block: u64,
 }
 

offchain/authority-claimer/src/lib.rs

@@ -10,8 +10,9 @@ pub mod sender;
 pub mod signer;
 
 use config::Config;
+use listener::{BrokerListener, MultidappBrokerListener};
 use snafu::Error;
-use tracing::trace;
+use tracing::{info, trace};
 
 use crate::{
     checker::DefaultDuplicateChecker,
@@ -22,21 +23,39 @@ use crate::{
 };
 
 pub async fn run(config: Config) -> Result<(), Box<dyn Error>> {
-    // Creating the metrics and health server.
     let metrics = AuthorityClaimerMetrics::new();
+    let dapp_address = config.authority_claimer_config.dapp_address.clone();
+    if let Some(dapp_address) = dapp_address {
+        info!("Creating the default broker listener");
+        let broker_listener = DefaultBrokerListener::new(
+            config.authority_claimer_config.broker_config.clone(),
+            config.authority_claimer_config.tx_manager_config.chain_id,
+            dapp_address,
+        )
+        .await?;
+        _run(metrics, config, broker_listener).await
+    } else {
+        info!("Creating the multidapp broker listener");
+        let broker_listener = MultidappBrokerListener::new(
+            config.authority_claimer_config.broker_config.clone(),
+            config.authority_claimer_config.tx_manager_config.chain_id,
+        )
+        .await?;
+        _run(metrics, config, broker_listener).await
+    }
+}
+
+async fn _run<B: BrokerListener + Send + Sync + 'static>(
+    metrics: AuthorityClaimerMetrics,
+    config: Config,
+    broker_listener: B,
+) -> Result<(), Box<dyn Error>> {
     let http_server_handle =
         http_server::start(config.http_server_config, metrics.clone().into());
-
     let config = config.authority_claimer_config;
-    let chain_id = config.tx_manager_config.chain_id;
 
-    // Creating the broker listener.
-    trace!("Creating the broker listener");
-    let broker_listener =
-        DefaultBrokerListener::new(config.broker_config.clone(), chain_id)
-            .await?;
+    let chain_id = config.tx_manager_config.chain_id;
 
-    // Creating the duplicate checker.
     trace!("Creating the duplicate checker");
     let duplicate_checker = DefaultDuplicateChecker::new(
         config.tx_manager_config.provider_http_endpoint.clone(),
@@ -46,18 +65,18 @@ pub async fn run(config: Config) -> Result<(), Box<dyn Error>> {
     )
     .await?;
 
-    // Creating the transaction sender.
     trace!("Creating the transaction sender");
     let transaction_sender =
         DefaultTransactionSender::new(config.clone(), chain_id, metrics)
             .await?;
 
-    // Creating the claimer loop.
+    // Creating the claimer.
     let claimer = DefaultClaimer::new(
         broker_listener,
         duplicate_checker,
         transaction_sender,
     );
+
     let claimer_handle = claimer.start();
 
     // Starting the HTTP server and the claimer loop.