chore: start cell-emitter in CI

.github/workflows/ibc-test.yaml

@@ -28,9 +28,10 @@ jobs:
     timeout-minutes: 75
     env:
       SRC_DIR: ${{ github.workspace }}/ibc-test-src
-      # https://github.com/axonweb3/axon/commits/forcerelay-dev
-      AXON_COMMIT: 922fc3858b4c470a39b3ae98a479980e774896b5
-      IBC_CONTRACT_COMMIT: c5417573ec15c8aaab048caa1ec5f3bd50c2170e
+      # Branch forcerelay-dev
+      AXON_COMMIT: 82e07910f687f59cf1f97e45120223f9ee8e948f
+      IBC_CONTRACT_COMMIT: d33d0ee32b0b89aafac885fd3f68946ea1f2dc68
+      CELL_EMITTER_COMMIT: 0a897111b389472a078512815d293703910c25d5
     strategy:
       fail-fast: false
       matrix:
@@ -73,6 +74,9 @@ jobs:
       - name: Prepare Axon source
         run: git clone --recursive https://github.com/axonweb3/axon.git $SRC_DIR/axon && cd $SRC_DIR/axon && git checkout $AXON_COMMIT
 
+      - name: Prepare Cell Emitter
+        run: git clone --recursive https://github.com/axonweb3/emitter.git $SRC_DIR/emitter && cd $SRC_DIR/emitter && git checkout $CELL_EMITTER_COMMIT
+
       - name: Rust cache
         uses: Swatinem/rust-cache@v2
         with:
@@ -110,6 +114,15 @@ jobs:
           command: build
           args: -p ibc-test --tests --jobs=4
 
+      - name: Build Cell Emitter
+        uses: actions-rs/cargo@v1
+        with:
+          command: build
+          args: --release --manifest-path ${{env.SRC_DIR}}/emitter/Cargo.toml
+
+      - name: Add cell-emitter bin to path
+        run: echo "${{env.SRC_DIR}}/emitter/target/release/" >> $GITHUB_PATH
+
       - name: Run IBC tests
         uses: actions-rs/cargo@v1
         env:

crates/relayer/src/chain/axon.rs

@@ -1383,6 +1383,22 @@ macro_rules! convert {
 
 impl AxonChain {
     fn send_message(&mut self, message: Any) -> Result<IbcEventWithHeight, Error> {
+        loop {
+            match self.send_message_inner(message.clone()) {
+                Ok(e) => return Ok(e),
+                Err(e) => {
+                    if e.to_string().contains("reverted: getHeader") {
+                        std::thread::sleep(Duration::from_secs(5));
+                        tracing::info!("getHeader failed, waiting for header sync");
+                        continue;
+                    }
+                    return Err(e);
+                }
+            }
+        }
+    }
+
+    fn send_message_inner(&mut self, message: Any) -> Result<IbcEventWithHeight, Error> {
         use contract::*;
         let msg = message.clone();
         let tx_receipt: eyre::Result<_> = match msg.type_url.as_str() {

tools/ibc-test/src/consts.rs

@@ -20,4 +20,4 @@ pub const CLIENT_TYPE_ARGS: H256 =
 
 // Don't know how to pass this from axon to ckb. It doesn't change often, so
 // let's hardcode it for now.
-pub const AXON_IBC_HANDLER_ADDRESS: H160 = h160!("0xdc64a140aa3e981100a9beca4e685f962f0cf6c9");
+pub const AXON_IBC_HANDLER_ADDRESS: H160 = h160!("0x0165878a594ca255338adfa4d48449f69242eb8f");

tools/ibc-test/src/framework/binary/node.rs

@@ -1,13 +1,15 @@
 use ibc_test_framework::{
-    chain::builder::ChainBuilder,
+    chain::{builder::ChainBuilder, chain_type::ChainType},
     framework::{
         base::BasicTest,
         binary::node::{NodeConfigOverride, NodeGenesisOverride},
     },
     prelude::*,
 };
 
-use crate::framework::bootstrap::node::bootstrap_single_node;
+use crate::framework::{
+    bootstrap::node::bootstrap_single_node, utils::common::prepare_cell_emitter,
+};
 
 /**
    A wrapper type that lifts a test case that implements [`BinaryNodeTest`]
@@ -46,6 +48,25 @@ where
         let _node_process_a = node_a.process.clone();
         let _node_process_b = node_b.process.clone();
 
+        // start cell-emitter if only one part of connected chains is Axon
+        let chain_types = (
+            &node_a.chain_driver.chain_type,
+            &node_b.chain_driver.chain_type,
+        );
+        if matches!(chain_types, (&ChainType::Axon, &ChainType::Ckb)) {
+            let axon_port = node_a.chain_driver.rpc_port;
+            let ckb_port = node_b.chain_driver.rpc_port;
+            println!("start cell-emiter for Axon:{axon_port} and CKB:{ckb_port}");
+            let emitter = prepare_cell_emitter(axon_port, ckb_port)?;
+            config.extra_process.replace(Some(emitter));
+        } else if matches!(chain_types, (&ChainType::Ckb, &ChainType::Axon)) {
+            let axon_port = node_b.chain_driver.rpc_port;
+            let ckb_port = node_a.chain_driver.rpc_port;
+            println!("start cell-emiter for Axon:{axon_port} and CKB:{ckb_port}");
+            let emitter = prepare_cell_emitter(axon_port, ckb_port)?;
+            config.extra_process.replace(Some(emitter));
+        }
+
         eprintln!("Node is initialized, Starting running inner test..........");
 
         self.test.run(config, node_a, node_b)

tools/ibc-test/src/framework/utils/axon.rs

@@ -258,7 +258,7 @@ pub(crate) fn add_axon_devnet_relayer_wallet(
         prefix.to_string()
     };
     let private_key = {
-        let data = hex::decode("37aa0f893d05914a4def0460c0a984d3611546cfb26924d7a7ca6e0db9950a2d")
+        let data = hex::decode("ac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80")
             .unwrap();
         SecretKey::from_slice(&data).unwrap()
     };

tools/ibc-test/src/framework/utils/common.rs

@@ -1,15 +1,18 @@
 use ckb_sdk::AddressPayload;
 use ckb_types::packed::Script;
 use ckb_types::prelude::Entity;
+use ethers::core::k256::pkcs8::der::Writer;
 use futures::Future;
 use ibc_test_framework::prelude::*;
 use relayer::chain::ChainType;
+use secp256k1::rand::rngs;
 use secp256k1::Secp256k1;
 use secp256k1::{
     rand::{self, Rng},
     SecretKey,
 };
 use std::path::PathBuf;
+use std::process::{Child, Command};
 use std::str::FromStr;
 use tokio::runtime::Runtime;
 
@@ -62,3 +65,61 @@ pub fn transfer_port_id(chain_type: ChainType) -> PortId {
         }
     }
 }
+
+pub fn prepare_cell_emitter(
+    axon_port: u16,
+    ckb_port: u16,
+) -> Result<(Child, std::sync::mpsc::Sender<()>), Error> {
+    let listen_port = rngs::OsRng.gen_range(9000..10000);
+    let private_path = std::env::current_dir()
+        .unwrap()
+        .join(format!("emitter-privkey-{listen_port}"));
+    std::fs::File::create(&private_path)
+        .map_err(|err| eyre!("failed to create emitter private file: {err}"))?
+        .write(
+            &hex::decode("37aa0f893d05914a4def0460c0a984d3611546cfb26924d7a7ca6e0db9950a2d")
+                .unwrap(),
+        )
+        .unwrap();
+    let store_path = std::env::current_dir()
+        .unwrap()
+        .join(format!("emitter-store-{listen_port}"));
+    std::fs::create_dir_all(&store_path)
+        .map_err(|err| eyre!("failed to create emitter store path: {err}"))?;
+    let emitter_thread = Command::new("emitter")
+        .arg("-c")
+        .arg(format!("http://127.0.0.1:{ckb_port}"))
+        .arg("--i")
+        .arg(format!("http://127.0.0.1:{axon_port}"))
+        .arg("-l")
+        .arg(format!("127.0.0.1:{listen_port}"))
+        .arg("-s")
+        .arg(store_path)
+        .arg("-p")
+        .arg(private_path)
+        .spawn()
+        .map_err(|err| eyre!("failed to start emitter: {err}"))?;
+    // check header sync progress
+    let (tx, rx) = std::sync::mpsc::channel();
+    std::thread::spawn(move || loop {
+        std::thread::sleep(Duration::from_secs(10));
+        let output = Command::new("curl")
+            .arg("-H")
+            .arg("content-type: application/json")
+            .arg("-d")
+            .arg("{\"id\": 2, \"jsonrpc\": \"2.0\", \"method\": \"info\", \"params\": [] }")
+            .arg(format!("http://127.0.0.1:{listen_port}"))
+            .output()
+            .unwrap();
+        let log = if output.status.success() {
+            output.stdout
+        } else {
+            output.stderr
+        };
+        println!("\n[CellEmitter] {}", String::from_utf8(log).unwrap());
+        if rx.try_recv().is_ok() {
+            return;
+        }
+    });
+    Ok((emitter_thread, tx))
+}

tools/test-framework/src/bootstrap/init.rs

@@ -5,8 +5,10 @@
 
 use eyre::Report as Error;
 use ibc_relayer_cli::components::enable_ansi;
+use std::cell::RefCell;
 use std::env;
 use std::fs;
+use std::rc::Rc;
 use std::sync::Once;
 use tracing_subscriber::{
     self as ts,
@@ -65,6 +67,7 @@ pub fn init_test() -> Result<TestConfig, Error> {
         account_prefixes,
         hang_on_fail,
         bootstrap_with_random_ids: false,
+        extra_process: Rc::new(RefCell::new(None)),
     })
 }
 

tools/test-framework/src/types/config.rs

@@ -3,7 +3,7 @@
 */
 
 use core::fmt::Debug;
-use std::path::PathBuf;
+use std::{cell::RefCell, path::PathBuf, process::Child, rc::Rc, sync::mpsc::Sender};
 
 /**
    The test config to be passed to each test case. Currently this is loaded
@@ -58,4 +58,18 @@ pub struct TestConfig {
     pub hang_on_fail: bool,
 
     pub bootstrap_with_random_ids: bool,
+
+    pub extra_process: Rc<RefCell<Option<(Child, Sender<()>)>>>,
+}
+
+impl Drop for TestConfig {
+    fn drop(&mut self) {
+        println!("release cell-emitter child process");
+        let mut process = self.extra_process.borrow_mut();
+        if process.is_some() {
+            let (ref mut child, handler) = process.as_mut().unwrap();
+            child.kill().unwrap();
+            handler.send(()).unwrap();
+        }
+    }
 }