feat: add dedicated websocket cli parameter (#337)

crates/topos-sequencer/src/lib.rs

@@ -20,7 +20,8 @@ mod app_context;
 pub struct SequencerConfiguration {
     pub subnet_id: Option<String>,
     pub public_key: Option<Vec<u8>>,
-    pub subnet_jsonrpc_endpoint: String,
+    pub subnet_jsonrpc_http: String,
+    pub subnet_jsonrpc_ws: Option<String>,
     pub subnet_contract_address: String,
     pub tce_grpc_endpoint: String,
     pub signing_key: SecretKey,
@@ -49,7 +50,7 @@ pub async fn launch(
     // It will retry using backoff algorithm, but if it fails (default max backoff elapsed time is 15 min) we can not proceed
     else {
         let http_endpoint =
-            topos_sequencer_subnet_runtime::derive_endpoints(&config.subnet_jsonrpc_endpoint)
+            topos_sequencer_subnet_runtime::derive_endpoints(&config.subnet_jsonrpc_http)
                 .map_err(|e| {
                     Box::new(std::io::Error::new(
                         InvalidInput,
@@ -73,8 +74,13 @@ pub async fn launch(
         }
     };
 
-    let (http_endpoint, ws_endpoint) =
-        topos_sequencer_subnet_runtime::derive_endpoints(&config.subnet_jsonrpc_endpoint)?;
+    let (http_endpoint, mut ws_endpoint) =
+        topos_sequencer_subnet_runtime::derive_endpoints(&config.subnet_jsonrpc_http)?;
+
+    if let Some(config_ws_endpoint) = config.subnet_jsonrpc_ws.as_ref() {
+        // Use explicitly provided websocket subnet endpoint
+        ws_endpoint = config_ws_endpoint.clone();
+    }
 
     // Instantiate subnet runtime proxy, handling interaction with subnet node
     let subnet_runtime_proxy_worker = match SubnetRuntimeProxyWorker::new(

crates/topos-tce-broadcast/src/task_manager_futures/mod.rs

@@ -127,7 +127,7 @@ impl TaskManager {
                                     let prev = self.validator_store.get_certificate(&cert.prev_id);
                                     if matches!(prev, Ok(Some(_))) || cert.prev_id == topos_core::uci::INITIAL_CERTIFICATE_ID  {
                                         Self::start_task(
-                                            &mut self.running_tasks,
+                                            &self.running_tasks,
                                             task,
                                             task_context.sink.clone(),
                                             self.buffered_messages.remove(&cert.id),
@@ -155,7 +155,7 @@ impl TaskManager {
 
                                 let certificate_id= task.certificate_id;
                                 Self::start_task(
-                                    &mut self.running_tasks,
+                                    &self.running_tasks,
                                     task,
                                     context.sink.clone(),
                                     self.buffered_messages.remove(&certificate_id),
@@ -182,7 +182,7 @@ impl TaskManager {
     }
 
     fn start_task(
-        running_tasks: &mut RunningTasks,
+        running_tasks: &RunningTasks,
         task: Task,
         sink: mpsc::Sender<DoubleEchoCommand>,
         messages: Option<Vec<DoubleEchoCommand>>,

crates/topos/src/components/node/services.rs

@@ -59,7 +59,8 @@ pub(crate) fn spawn_sequencer_process(
     let config = SequencerConfiguration {
         subnet_id: None,
         public_key: keys.validator_pubkey(),
-        subnet_jsonrpc_endpoint: config.subnet_jsonrpc_endpoint,
+        subnet_jsonrpc_http: config.subnet_jsonrpc_http,
+        subnet_jsonrpc_ws: config.subnet_jsonrpc_ws,
         subnet_contract_address: config.subnet_contract_address,
         tce_grpc_endpoint: config.tce_grpc_endpoint,
         signing_key: keys.validator.clone().unwrap(),

crates/topos/src/components/sequencer/commands/run.rs

@@ -9,15 +9,22 @@ pub struct Run {
     #[clap(long, env = "TOPOS_LOCAL_SUBNET_ID")]
     pub subnet_id: Option<String>,
 
-    // Subnet endpoint in the form [ip address]:[port]
-    // Topos sequencer expects both websocket and http protocol available
-    // on this subnet endpoint
+    /// Subnet endpoint in the form [ip address]:[port]
+    /// Topos sequencer expects both websocket and http protocol available
+    /// on this subnet endpoint. If optional `subnet_jsonrpc_ws` is not provided websocket endpoint
+    /// will be deduced from this parameter.
     #[clap(
         long,
         default_value = "127.0.0.1:8545",
-        env = "SUBNET_JSONRPC_ENDPOINT"
+        env = "TOPOS_SUBNET_JSONRPC_HTTP"
     )]
-    pub subnet_jsonrpc_endpoint: String,
+    pub subnet_jsonrpc_http: String,
+
+    /// Optional explicit websocket endpoint for the subnet jsonrpc api. If this parameter is not provided,
+    /// it will be derived from the `subnet_jsonrpc_http`.
+    /// Full uri value is expected, e.g. `wss://arbitrum.infura.com/v3/ws/mykey` or `ws://127.0.0.1/ws`
+    #[clap(long, env = "TOPOS_SUBNET_JSONRPC_WS")]
+    pub subnet_jsonrpc_ws: Option<String>,
 
     // Core contract address
     #[clap(long, env = "SUBNET_CONTRACT_ADDRESS")]

crates/topos/src/components/sequencer/mod.rs

@@ -21,7 +21,8 @@ pub(crate) async fn handle_command(
             let config = SequencerConfiguration {
                 subnet_id: cmd.subnet_id,
                 public_key: None,
-                subnet_jsonrpc_endpoint: cmd.subnet_jsonrpc_endpoint,
+                subnet_jsonrpc_http: cmd.subnet_jsonrpc_http,
+                subnet_jsonrpc_ws: cmd.subnet_jsonrpc_ws,
                 subnet_contract_address: cmd.subnet_contract_address,
                 tce_grpc_endpoint: cmd.base_tce_api_url,
                 signing_key: keys.validator.clone().unwrap(),

crates/topos/src/config/sequencer.rs

@@ -17,7 +17,12 @@ pub struct SequencerConfig {
     /// JSON-RPC endpoint of the Edge node, websocket and http support expected
     /// If the endpoint address starts with `https`, ssl will be used with http/websocket
     #[serde(default = "default_subnet_jsonrpc_endpoint")]
-    pub subnet_jsonrpc_endpoint: String,
+    pub subnet_jsonrpc_http: String,
+
+    // Optional explicit websocket endpoint for the subnet jsonrpc api. If this parameter is not provided,
+    // it will be derived from the `subnet_jsonrpc_http`.
+    // Full uri value is expected, e.g. `wss://arbitrum.infura.com/v3/ws/mykey` or `ws://127.0.0.1/ws`
+    pub subnet_jsonrpc_ws: Option<String>,
 
     /// Address where the Topos Core contract is deployed
     #[serde(default = "default_subnet_contract_address")]

crates/topos/tests/snapshots/sequencer__sequencer_help_display.snap

@@ -13,8 +13,10 @@ Options:
           Defines the verbosity level
       --home <HOME>
           Home directory for the configuration [env: TOPOS_HOME=] [default: /home/runner/.config/topos]
-      --subnet-jsonrpc-endpoint <SUBNET_JSONRPC_ENDPOINT>
-          [env: SUBNET_JSONRPC_ENDPOINT=] [default: 127.0.0.1:8545]
+      --subnet-jsonrpc-http <SUBNET_JSONRPC_HTTP>
+          Subnet endpoint in the form [ip address]:[port] Topos sequencer expects both websocket and http protocol available on this subnet endpoint. If optional `subnet_jsonrpc_ws` is not provided websocket endpoint will be deduced from this parameter [env: TOPOS_SUBNET_JSONRPC_HTTP=] [default: 127.0.0.1:8545]
+      --subnet-jsonrpc-ws <SUBNET_JSONRPC_WS>
+          Optional explicit websocket endpoint for the subnet jsonrpc api. If this parameter is not provided, it will be derived from the `subnet_jsonrpc_http`. Full uri value is expected, e.g. `wss://arbitrum.infura.com/v3/ws/mykey` or `ws://127.0.0.1/ws` [env: TOPOS_SUBNET_JSONRPC_WS=]
       --subnet-contract-address <SUBNET_CONTRACT_ADDRESS>
           [env: SUBNET_CONTRACT_ADDRESS=]
       --base-tce-api-url <BASE_TCE_API_URL>