changed a lot

offchain/authority-claimer/Cargo.toml

@@ -13,6 +13,8 @@ test = false
 http-server = { path = "../http-server" }
 rollups-events = { path = "../rollups-events" }
 
+thiserror = "1.0"
+
 async-trait.workspace = true
 clap = { workspace = true, features = ["derive"] }
 eth-tx-manager.workspace = true

offchain/authority-claimer/src/claimer.rs

@@ -2,105 +2,82 @@
 // SPDX-License-Identifier: Apache-2.0 (see LICENSE)
 
 use async_trait::async_trait;
-use rollups_events::DAppMetadata;
-use tracing::{info, trace, warn};
+use ethers::types::Address;
+use std::marker::PhantomData;
+use tracing::{trace, warn};
 
-use crate::{
-    config::AuthorityClaimerConfig,
-    error::AuthorityClaimerError,
-    listener::{BrokerListener, RedisBrokerListener},
-    metrics::AuthorityClaimerMetrics,
-    sender::{ClaimSender, TxManagerClaimSender},
-};
+use crate::{listener::BrokerListener, sender::ClaimSender};
 
 #[async_trait]
-pub trait AuthorityClaimerServer<S: ClaimSender, L: BrokerListener> {
+pub trait AuthorityClaimer<S: ClaimSender, L: BrokerListener> {
     async fn start(
         self,
-        dapp_address: ethers::types::Address,
+        claim_sender: S,
+        broker_listener: L,
+        dapp_address: Address,
     ) -> Result<(), AuthorityClaimerError<S, L>>;
 }
 
-pub struct DefaultAuthorityClaimerServer {
-    claim_sender: TxManagerClaimSender,
-    broker_listener: RedisBrokerListener,
-}
+#[derive(Debug, thiserror::Error)]
+pub enum AuthorityClaimerError<S: ClaimSender, L: BrokerListener> {
+    #[error("claim sender error: {0}")]
+    ClaimSenderError(S::Error),
 
-impl DefaultAuthorityClaimerServer {
-    pub async fn new(
-        config: AuthorityClaimerConfig,
-        metrics: AuthorityClaimerMetrics,
-    ) -> Result<
-        Self,
-        AuthorityClaimerError<TxManagerClaimSender, RedisBrokerListener>,
-    > {
-        info!("Setting up authority claimer with config: {:?}", config);
+    #[error("broker listener error: {0}")]
+    BrokerListenerError(L::Error),
 
-        let dapp_metadata = DAppMetadata {
-            chain_id: config.txm_config.chain_id,
-            dapp_address: rollups_events::Address::new(
-                config.dapp_address.into(),
-            ),
-        };
+    #[error("connection with the broker was closed")]
+    BrokerListenerClosed,
+}
 
-        trace!("Creating the claim sender");
-        let claim_sender =
-            TxManagerClaimSender::new(dapp_metadata.clone(), metrics)
-                .map_err(AuthorityClaimerError::ClaimSenderError)?;
+// ------------------------------------------------------------------------------------------------
+// DefaultAuthorityClaimer
+// ------------------------------------------------------------------------------------------------
 
-        trace!("Creating the broker listener");
-        let broker_listener =
-            RedisBrokerListener::new(dapp_metadata.clone())
-                .map_err(AuthorityClaimerError::BrokerListenerError)?;
+#[derive(Default)]
+pub struct DefaultAuthorityClaimer<'a> {
+    lifetime: PhantomData<&'a ()>,
+}
 
-        Ok(DefaultAuthorityClaimerServer {
-            claim_sender,
-            broker_listener,
-        })
+impl<'a> DefaultAuthorityClaimer<'a> {
+    pub fn new() -> Self {
+        Self::default()
     }
 }
 
 #[async_trait]
-impl AuthorityClaimerServer<TxManagerClaimSender, RedisBrokerListener>
-    for DefaultAuthorityClaimerServer
+impl<'a, S, L> AuthorityClaimer<S, L> for DefaultAuthorityClaimer<'a>
+where
+    S: 'a + ClaimSender,
+    L: 'a + BrokerListener + Sync,
 {
     async fn start(
         self,
-        dapp_address: ethers::types::Address,
-    ) -> Result<
-        (),
-        AuthorityClaimerError<TxManagerClaimSender, RedisBrokerListener>,
-    > {
-        start_server(self.claim_sender, self.broker_listener, dapp_address)
-            .await
-    }
-}
-
-async fn start_server<S: ClaimSender, L: BrokerListener>(
-    claim_sender: S,
-    broker_listener: L,
-    dapp_address: ethers::types::Address,
-) -> Result<(), AuthorityClaimerError<S, L>> {
-    trace!("Starting the authority claimer loop");
-    let mut claim_sender = claim_sender;
-    loop {
-        match broker_listener.listen().await {
-            Some(Ok(rollups_claim)) => {
-                trace!(
-                    "Got a claim from the broker channel: {:?}",
-                    rollups_claim
-                );
-                claim_sender = claim_sender
-                    .send_claim(dapp_address.clone(), rollups_claim)
-                    .await
-                    .map_err(AuthorityClaimerError::ClaimSenderError)?;
-            }
-            Some(Err(e)) => {
-                warn!("Broker channel error `{}`", e);
-            }
+        claim_sender: S,
+        broker_listener: L,
+        dapp_address: Address,
+    ) -> Result<(), AuthorityClaimerError<S, L>> {
+        trace!("Starting the authority claimer loop");
+        let mut claim_sender = claim_sender;
+        loop {
+            match broker_listener.listen().await {
+                Some(Ok(rollups_claim)) => {
+                    trace!(
+                        "Got a claim from the broker channel: {:?}",
+                        rollups_claim
+                    );
+                    claim_sender = claim_sender
+                        .send_claim(dapp_address.clone(), rollups_claim)
+                        .await
+                        .map_err(AuthorityClaimerError::ClaimSenderError)?;
+                }
+                Some(Err(e)) => {
+                    warn!("Broker channel error `{}`", e);
+                }
 
-            None => {
-                return Err(AuthorityClaimerError::BrokerListenerClosed);
+                None => {
+                    return Err(AuthorityClaimerError::BrokerListenerClosed);
+                }
             }
         }
     }

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

@@ -3,23 +3,23 @@
 
 use clap::{command, Parser};
 use eth_tx_manager::{
-    config::{TxEnvCLIConfig as TxManagerCLI, TxManagerConfig},
+    config::{TxEnvCLIConfig as TxManagerCLIConfig, TxManagerConfig},
     Priority,
 };
+use rollups_events::{BrokerCLIConfig, BrokerConfig};
 use rusoto_core::Region;
 use snafu::ResultExt;
 use std::{fs, path::PathBuf, str::FromStr};
 
 use crate::config::{
     error::{
-        AuthError, AuthSnafu, AuthorityClaimerConfigError, InvalidRegionSnafu,
-        MnemonicFileSnafu, TxManagerSnafu,
+        AuthConfigError, AuthSnafu, AuthorityClaimerConfigError,
+        InvalidRegionSnafu, MnemonicFileSnafu, TxManagerSnafu,
     },
+    json::{read_json_file, DappDeployment},
     AuthConfig, AuthorityClaimerConfig,
 };
 
-use super::json::{read_json_file, DappDeployment};
-
 // ------------------------------------------------------------------------------------------------
 // AuthorityClaimerCLI
 // ------------------------------------------------------------------------------------------------
@@ -28,68 +28,43 @@ use super::json::{read_json_file, DappDeployment};
 #[command(name = "rd_config")]
 #[command(about = "Configuration for rollups authority claimer")]
 pub(crate) struct AuthorityClaimerCLI {
-    // #[command(flatten)]
-    // pub sc_config: SCEnvCLIConfig,
-    //
     #[command(flatten)]
-    txm_config: TxManagerCLI,
+    txm_config: TxManagerCLIConfig,
+
+    #[command(flatten)]
+    auth_config: AuthCLIConfig,
 
-    // #[command(flatten)]
-    // pub broker_config: BrokerCLIConfig,
-    //
     #[command(flatten)]
-    auth_config: AuthCLI,
+    broker_config: BrokerCLIConfig,
 
     /// Path to file with deployment json of dapp
     #[arg(long, env, default_value = "./dapp_deployment.json")]
-    pub dapp_deployment_file: PathBuf,
-    //
-    // /// Path to file with deployment json of rollups
-    // #[arg(long, env, default_value = "./rollups_deployment.json")]
-    // pub rd_rollups_deployment_file: PathBuf,
-    //
-    // /// Duration of rollups epoch in seconds, for which dispatcher will make claims.
-    // #[arg(long, env, default_value = "604800")]
-    // pub rd_epoch_duration: u64,
+    dapp_deployment_file: PathBuf,
 }
 
 impl TryFrom<AuthorityClaimerCLI> for AuthorityClaimerConfig {
     type Error = AuthorityClaimerConfigError;
 
-    fn try_from(cli: AuthorityClaimerCLI) -> Result<Self, Self::Error> {
-        // let sc_config = SCConfig::initialize(dispatcher_config.sc_config)
-        //     .context(StateClientSnafu)?;
-
-        let txm_config = TxManagerConfig::initialize(cli.txm_config)
+    fn try_from(cli_config: AuthorityClaimerCLI) -> Result<Self, Self::Error> {
+        let txm_config = TxManagerConfig::initialize(cli_config.txm_config)
             .context(TxManagerSnafu)?;
 
-        let auth_config = cli.auth_config.try_into().context(AuthSnafu)?;
+        let auth_config =
+            AuthConfig::try_from(cli_config.auth_config).context(AuthSnafu)?;
+
+        let broker_config = BrokerConfig::from(cli_config.broker_config);
 
         let dapp_deployment =
-            read_json_file::<DappDeployment>(cli.dapp_deployment_file)?;
+            read_json_file::<DappDeployment>(cli_config.dapp_deployment_file)?;
         let dapp_address = dapp_deployment.dapp_address;
         let dapp_deploy_block_hash = dapp_deployment.dapp_deploy_block_hash;
 
-        // let path = dispatcher_config.rd_rollups_deployment_file;
-        // let rollups_deployment = read_json::<RollupsDeploymentJson>(path)
-        //     .map(RollupsDeployment::from)?;
-
-        // let broker_config = BrokerConfig::from(dispatcher_config.broker_config);
-
-        // assert!(
-        //     sc_config.default_confirmations < tx_config.default_confirmations,
-        //     "`state-client confirmations` has to be less than `tx-manager confirmations,`"
-        // );
-
         Ok(AuthorityClaimerConfig {
-            // sc_config,
             txm_config,
-            // broker_config,
             auth_config,
+            broker_config,
             dapp_address,
             dapp_deploy_block_hash,
-            // rollups_deployment,
-            // epoch_duration: dispatcher_config.rd_epoch_duration,
             txm_priority: Priority::Normal,
         })
     }
@@ -102,7 +77,7 @@ impl TryFrom<AuthorityClaimerCLI> for AuthorityClaimerConfig {
 #[derive(Debug, Clone, Parser)]
 #[command(name = "auth_config")]
 #[command(about = "Configuration for signing authentication")]
-pub(crate) struct AuthCLI {
+pub(crate) struct AuthCLIConfig {
     /// Signer mnemonic, overrides `auth_mnemonic_file` and `auth_aws_kms_*`
     #[arg(long, env)]
     auth_mnemonic: Option<String>,
@@ -124,10 +99,10 @@ pub(crate) struct AuthCLI {
     auth_aws_kms_region: Option<String>,
 }
 
-impl TryFrom<AuthCLI> for AuthConfig {
-    type Error = AuthError;
+impl TryFrom<AuthCLIConfig> for AuthConfig {
+    type Error = AuthConfigError;
 
-    fn try_from(cli: AuthCLI) -> Result<Self, Self::Error> {
+    fn try_from(cli: AuthCLIConfig) -> Result<Self, Self::Error> {
         let account_index = cli.auth_mnemonic_account_index;
         if let Some(mnemonic) = cli.auth_mnemonic {
             Ok(AuthConfig::Mnemonic {
@@ -145,8 +120,8 @@ impl TryFrom<AuthCLI> for AuthConfig {
             })
         } else {
             match (cli.auth_aws_kms_key_id, cli.auth_aws_kms_region) {
-                (None, _) => Err(AuthError::MissingConfiguration),
-                (Some(_), None) => Err(AuthError::MissingRegion),
+                (None, _) => Err(AuthConfigError::MissingConfiguration),
+                (Some(_), None) => Err(AuthConfigError::MissingRegion),
                 (Some(key_id), Some(region)) => {
                     let region = Region::from_str(&region)
                         .context(InvalidRegionSnafu)?;

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

@@ -1,22 +1,19 @@
 // (c) Cartesi and individual authors (see AUTHORS)
 // SPDX-License-Identifier: Apache-2.0 (see LICENSE)
 
-use eth_tx_manager::config::Error as TxManagerError;
+use eth_tx_manager::config::Error as TxManagerConfigError;
 use rusoto_core::region::ParseRegionError;
 use snafu::Snafu;
 use std::path::PathBuf;
 
 #[derive(Debug, Snafu)]
 #[snafu(visibility(pub(crate)))]
 pub enum AuthorityClaimerConfigError {
-    // #[snafu(display("StateClient configuration error: {}", source))]
-    // StateClientError { source: SCError },
-    //
     #[snafu(display("TxManager configuration error: {}", source))]
-    TxManagerError { source: TxManagerError },
+    TxManagerError { source: TxManagerConfigError },
 
     #[snafu(display("Auth configuration error: {}", source))]
-    AuthError { source: AuthError },
+    AuthError { source: AuthConfigError },
 
     #[snafu(display("Read file error ({})", path.display()))]
     ReadFileError {
@@ -29,17 +26,11 @@ pub enum AuthorityClaimerConfigError {
         path: PathBuf,
         source: serde_json::Error,
     },
-    //
-    // #[snafu(display("Rollups json read file error"))]
-    // RollupsJsonReadFileError { source: std::io::Error },
-
-    // #[snafu(display("Rollups json parse error"))]
-    // RollupsJsonParseError { source: serde_json::Error },
 }
 
 #[derive(Debug, Snafu)]
 #[snafu(visibility(pub(crate)))]
-pub enum AuthError {
+pub enum AuthConfigError {
     #[snafu(display("Missing auth configuration"))]
     MissingConfiguration,