feat: add sunodo validator mode

CHANGELOG.md

@@ -14,6 +14,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Added `cartesi-rollups-cli` binary to help develop and debug the Cartesi Rollups node
 - Added read to `cartesi-rollups-cli` for inputs, notices, vouchers and reports
 - Added `Makefile` to auto-generate project files
+- Added experimental sunodo validator mode
 
 ### Changed
 

cmd/cartesi-rollups-node/main.go

@@ -20,8 +20,11 @@ func main() {
 	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
 	defer stop()
 
-	// add Redis first
-	s = append(s, newRedis())
+	sunodoValidatorEnabled := config.GetCartesiExperimentalSunodoValidatorEnabled()
+	if !sunodoValidatorEnabled {
+		// add Redis first
+		s = append(s, newRedis())
+	}
 
 	// add services without dependencies
 	s = append(s, newGraphQLServer())
@@ -35,8 +38,8 @@ func main() {
 		s = append(s, newServerManager())
 	}
 
-	// enable claimer if reader mode is disabled
-	if !config.GetCartesiFeatureReaderMode() {
+	// enable claimer if reader mode and sunodo validator mode are disabled
+	if !config.GetCartesiFeatureReaderMode() && !sunodoValidatorEnabled {
 		s = append(s, newAuthorityClaimer())
 	}
 

cmd/cartesi-rollups-node/services.go

@@ -36,6 +36,15 @@ func getPort(offset portOffset) int {
 	return config.GetCartesiHttpPort() + int(offset)
 }
 
+// Get the redis endpoint based on whether the experimental sunodo validator mode is enabled.
+func getRedisEndpoint() string {
+	if config.GetCartesiExperimentalSunodoValidatorEnabled() {
+		return config.GetCartesiExperimentalSunodoValidatorRedisEndpoint()
+	} else {
+		return fmt.Sprintf("redis://127.0.0.1:%v", getPort(portOffsetRedis))
+	}
+}
+
 const serverManagerSessionId = "default_session_id"
 
 // Create the RUST_LOG variable using the config log level.
@@ -69,7 +78,7 @@ func newAdvanceRunner() services.CommandService {
 	s.Env = append(s.Env,
 		fmt.Sprintf("SESSION_ID=%v", serverManagerSessionId))
 	s.Env = append(s.Env,
-		fmt.Sprintf("REDIS_ENDPOINT=redis://127.0.0.1:%v", getPort(portOffsetRedis)))
+		fmt.Sprintf("REDIS_ENDPOINT=%v", getRedisEndpoint()))
 	s.Env = append(s.Env,
 		fmt.Sprintf("CHAIN_ID=%v", config.GetCartesiBlockchainId()))
 	s.Env = append(s.Env,
@@ -78,6 +87,8 @@ func newAdvanceRunner() services.CommandService {
 		fmt.Sprintf("PROVIDER_HTTP_ENDPOINT=%v", config.GetCartesiBlockchainHttpEndpoint()))
 	s.Env = append(s.Env,
 		fmt.Sprintf("ADVANCE_RUNNER_HEALTHCHECK_PORT=%v", getPort(portOffsetAdvanceRunner)))
+	s.Env = append(s.Env,
+		fmt.Sprintf("READER_MODE=%v", config.GetCartesiFeatureReaderMode()))
 	if config.GetCartesiFeatureHostMode() {
 		s.Env = append(s.Env, "SNAPSHOT_ENABLED=false")
 		s.Env = append(s.Env, "SNAPSHOT_VALIDATION_ENABLED=false")
@@ -108,7 +119,7 @@ func newAuthorityClaimer() services.CommandService {
 	s.Env = append(s.Env,
 		fmt.Sprintf("TX_DEFAULT_CONFIRMATIONS=%v", config.GetCartesiBlockchainFinalityOffset()))
 	s.Env = append(s.Env,
-		fmt.Sprintf("REDIS_ENDPOINT=redis://127.0.0.1:%v", getPort(portOffsetRedis)))
+		fmt.Sprintf("REDIS_ENDPOINT=%v", getRedisEndpoint()))
 	s.Env = append(s.Env,
 		fmt.Sprintf("DAPP_ADDRESS=%v", config.GetCartesiContractsDappAddress()))
 	s.Env = append(s.Env,
@@ -155,7 +166,7 @@ func newDispatcher() services.CommandService {
 	s.Env = append(s.Env,
 		fmt.Sprintf("SC_DEFAULT_CONFIRMATIONS=%v", config.GetCartesiBlockchainFinalityOffset()))
 	s.Env = append(s.Env,
-		fmt.Sprintf("REDIS_ENDPOINT=redis://127.0.0.1:%v", getPort(portOffsetRedis)))
+		fmt.Sprintf("REDIS_ENDPOINT=%v", getRedisEndpoint()))
 	s.Env = append(s.Env,
 		fmt.Sprintf("DAPP_ADDRESS=%v", config.GetCartesiContractsDappAddress()))
 	s.Env = append(s.Env,
@@ -231,7 +242,7 @@ func newIndexer() services.CommandService {
 	s.Env = append(s.Env,
 		fmt.Sprintf("DAPP_CONTRACT_ADDRESS=%v", config.GetCartesiContractsDappAddress()))
 	s.Env = append(s.Env,
-		fmt.Sprintf("REDIS_ENDPOINT=redis://127.0.0.1:%v", getPort(portOffsetRedis)))
+		fmt.Sprintf("REDIS_ENDPOINT=%v", getRedisEndpoint()))
 	s.Env = append(s.Env,
 		fmt.Sprintf("INDEXER_HEALTHCHECK_PORT=%v", getPort(portOffsetIndexer)))
 	s.Env = append(s.Env, os.Environ()...)

docs/config.md

@@ -109,6 +109,17 @@ Address of the InputBox contract.
 
 * **Type:** `string`
 
+### CARTESI_EXPERIMENTAL_SUNODO_VALIDATOR_ENABLED
+When enabled, the node does not start the authority-claimer service and the Redis server.
+
+* **Type:** `bool`
+* **Default:** `"false"`
+
+### CARTESI_EXPERIMENTAL_SUNODO_VALIDATOR_REDIS_ENDPOINT
+External Redis endpoint for the node when running in the experimental sunodo validator mode.
+
+* **Type:** `string`
+
 ### CARTESI_FEATURE_HOST_MODE
 If set to true the node will run in host mode.
 

internal/config/generate/Config.toml

@@ -212,3 +212,18 @@ description = """
 HTTP port for the node.
 The node will also use the 20 ports after this one for internal services."""
 
+#
+# Experimental
+#
+
+[experimental.CARTESI_EXPERIMENTAL_SUNODO_VALIDATOR_ENABLED]
+default = "false" 
+go-type = "bool"
+description = """
+When enabled, the node does not start the authority-claimer service and the Redis server."""
+
+[experimental.CARTESI_EXPERIMENTAL_SUNODO_VALIDATOR_REDIS_ENDPOINT]
+go-type = "string"
+description = """
+External Redis endpoint for the node when running in the experimental sunodo validator mode."""
+

offchain/advance-runner/src/broker.rs

@@ -33,24 +33,27 @@ pub struct BrokerFacade {
     inputs_stream: RollupsInputsStream,
     outputs_stream: RollupsOutputsStream,
     claims_stream: RollupsClaimsStream,
+    reader_mode: bool,
 }
 
 impl BrokerFacade {
     #[tracing::instrument(level = "trace", skip_all)]
     pub async fn new(
         config: BrokerConfig,
         dapp_metadata: DAppMetadata,
+        reader_mode: bool,
     ) -> Result<Self> {
         tracing::trace!(?config, "connecting to broker");
+        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);
-        let client = Broker::new(config).await.context(BrokerInternalSnafu)?;
+        let claims_stream = RollupsClaimsStream::new(dapp_metadata.chain_id);
         Ok(Self {
             client,
             inputs_stream,
             outputs_stream,
             claims_stream,
+            reader_mode,
         })
     }
 
@@ -116,6 +119,10 @@ impl BrokerFacade {
         &mut self,
         rollups_claim: RollupsClaim,
     ) -> Result<()> {
+        if self.reader_mode {
+            return Ok(());
+        }
+
         tracing::trace!(rollups_claim.epoch_index,
             ?rollups_claim.epoch_hash,
             "producing rollups claim"
@@ -172,8 +179,8 @@ mod tests {
     use super::*;
     use backoff::ExponentialBackoff;
     use rollups_events::{
-        DAppMetadata, Hash, InputMetadata, Payload, RollupsAdvanceStateInput,
-        HASH_SIZE,
+        Address, DAppMetadata, Hash, InputMetadata, Payload,
+        RollupsAdvanceStateInput, ADDRESS_SIZE, HASH_SIZE,
     };
     use test_fixtures::BrokerFixture;
     use testcontainers::clients::Cli;
@@ -196,7 +203,7 @@ mod tests {
                 consume_timeout: 10,
                 backoff,
             };
-            let facade = BrokerFacade::new(config, dapp_metadata)
+            let facade = BrokerFacade::new(config, dapp_metadata, false)
                 .await
                 .expect("failed to create broker facade");
             TestState { fixture, facade }
@@ -323,8 +330,9 @@ mod tests {
         let docker = Cli::default();
         let mut state = TestState::setup(&docker).await;
         let rollups_claim = RollupsClaim {
+            dapp_address: Address::new([0xa0; ADDRESS_SIZE]),
             epoch_index: 0,
-            epoch_hash: Hash::new([0xa0; HASH_SIZE]),
+            epoch_hash: Hash::new([0xb0; HASH_SIZE]),
             first_index: 0,
             last_index: 6,
         };
@@ -348,14 +356,16 @@ mod tests {
         let docker = Cli::default();
         let mut state = TestState::setup(&docker).await;
         let rollups_claim0 = RollupsClaim {
+            dapp_address: Address::new([0xa0; ADDRESS_SIZE]),
             epoch_index: 0,
-            epoch_hash: Hash::new([0xa0; HASH_SIZE]),
+            epoch_hash: Hash::new([0xb0; HASH_SIZE]),
             first_index: 0,
             last_index: 0,
         };
         let rollups_claim1 = RollupsClaim {
+            dapp_address: Address::new([0xa1; ADDRESS_SIZE]),
             epoch_index: 1,
-            epoch_hash: Hash::new([0xa1; HASH_SIZE]),
+            epoch_hash: Hash::new([0xb1; HASH_SIZE]),
             first_index: 1,
             last_index: 1,
         };

offchain/advance-runner/src/config.rs

@@ -23,6 +23,7 @@ pub struct AdvanceRunnerConfig {
     pub log_config: LogConfig,
     pub backoff_max_elapsed_duration: Duration,
     pub healthcheck_port: u16,
+    pub reader_mode: bool,
 }
 
 impl AdvanceRunnerConfig {
@@ -38,11 +39,15 @@ impl AdvanceRunnerConfig {
             dapp_metadata.dapp_address.clone(),
         )
         .context(SnapshotConfigSnafu)?;
+
+        let log_config = LogConfig::initialize(cli_config.log_cli_config);
+
         let backoff_max_elapsed_duration =
             Duration::from_millis(cli_config.backoff_max_elapsed_duration);
+
         let healthcheck_port = cli_config.healthcheck_port;
 
-        let log_config = LogConfig::initialize(cli_config.log_cli_config);
+        let reader_mode = cli_config.reader_mode;
 
         Ok(Self {
             server_manager_config,
@@ -52,6 +57,7 @@ impl AdvanceRunnerConfig {
             log_config,
             backoff_max_elapsed_duration,
             healthcheck_port,
+            reader_mode,
         })
     }
 }
@@ -92,4 +98,7 @@ struct CLIConfig {
         default_value_t = 8080
     )]
     pub healthcheck_port: u16,
+
+    #[arg(long, env)]
+    reader_mode: bool,
 }

offchain/advance-runner/src/lib.rs

@@ -46,15 +46,22 @@ async fn start_advance_runner(
         .with_max_elapsed_time(Some(config.backoff_max_elapsed_duration))
         .build();
 
-    let server_manager =
-        ServerManagerFacade::new(config.server_manager_config, backoff)
-            .await
-            .context(error::ServerManagerSnafu)?;
+    let server_manager = ServerManagerFacade::new(
+        config.dapp_metadata.dapp_address.clone(),
+        config.server_manager_config,
+        backoff,
+    )
+    .await
+    .context(error::ServerManagerSnafu)?;
     tracing::trace!("connected to the server-manager");
 
-    let broker = BrokerFacade::new(config.broker_config, config.dapp_metadata)
-        .await
-        .context(error::BrokerSnafu)?;
+    let broker = BrokerFacade::new(
+        config.broker_config,
+        config.dapp_metadata,
+        config.reader_mode,
+    )
+    .await
+    .context(error::BrokerSnafu)?;
     tracing::trace!("connected the broker");
 
     match config.snapshot_config {

offchain/advance-runner/src/server_manager/facade.rs

@@ -84,6 +84,7 @@ macro_rules! grpc_call {
 pub type Result<T> = std::result::Result<T, ServerManagerError>;
 
 pub struct ServerManagerFacade {
+    dapp_address: rollups_events::Address,
     client: ServerManagerClient<Channel>,
     config: ServerManagerConfig,
     backoff: ExponentialBackoff,
@@ -92,6 +93,7 @@ pub struct ServerManagerFacade {
 impl ServerManagerFacade {
     #[tracing::instrument(level = "trace", skip_all)]
     pub async fn new(
+        dapp_address: rollups_events::Address,
         config: ServerManagerConfig,
         backoff: ExponentialBackoff,
     ) -> Result<Self> {
@@ -107,6 +109,7 @@ impl ServerManagerFacade {
         .max_decoding_message_size(config.max_decoding_message_size);
 
         Ok(Self {
+            dapp_address,
             client,
             config,
             backoff,
@@ -330,6 +333,7 @@ impl ServerManagerFacade {
         tracing::trace!(?proofs, "got proofs");
 
         let rollups_claim = RollupsClaim {
+            dapp_address: self.dapp_address.clone(),
             epoch_index,
             epoch_hash,
             first_index: first_input.input_index as u128,

offchain/advance-runner/tests/fixtures/mod.rs

@@ -103,6 +103,7 @@ impl AdvanceRunnerFixture {
             backoff_max_elapsed_duration,
             healthcheck_port: 0,
             log_config: LogConfig::default(),
+            reader_mode: false,
         };
         let handler = RefCell::new(Some(start_advance_runner(config.clone())));
         Self { config, handler }

offchain/authority-claimer/Cargo.toml

@@ -18,13 +18,15 @@ types = { path = "../types" }
 
 async-trait.workspace = true
 clap = { workspace = true, features = ["derive", "env"] }
+eth-tx-manager.workspace = true
 ethabi.workspace = true
-ethers.workspace = true
 ethers-signers = { workspace = true, features = ["aws"] }
-eth-tx-manager.workspace = true
+ethers.workspace = true
 rusoto_core.workspace = true
 rusoto_kms.workspace = true
 rusoto_sts.workspace = true
+serde.workspace = true
+serde_json.workspace = true
 snafu.workspace = true
 tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
 tracing.workspace = true