Indexer proofs

chain-signatures/node/src/cli.rs

@@ -26,6 +26,13 @@ pub enum Cli {
             default_value("https://rpc.testnet.near.org")
         )]
         near_rpc: String,
+        /// Light client address
+        #[arg(
+            long,
+            env("MPC_RECOVERY_LIGHT_CLIENT_ADDR"),
+            default_value("http://localhost:3030")
+        )]
+        light_client_addr: String,
         /// MPC contract id
         #[arg(
             long,
@@ -101,6 +108,7 @@ impl Cli {
         match self {
             Cli::Start {
                 near_rpc,
+                light_client_addr,
                 account_id,
                 mpc_contract_id,
                 account_sk,
@@ -122,6 +130,8 @@ impl Cli {
                     "start".to_string(),
                     "--near-rpc".to_string(),
                     near_rpc,
+                    "--light-client-addr".to_string(),
+                    light_client_addr,
                     "--mpc-contract-id".to_string(),
                     mpc_contract_id.to_string(),
                     "--account-id".to_string(),
@@ -178,6 +188,7 @@ fn spinup_indexer(
     options: &indexer::Options,
     mpc_contract_id: &AccountId,
     account_id: &AccountId,
+    light_client_addr: Url,
     sign_queue: &Arc<RwLock<SignQueue>>,
     gcp: &GcpService,
 ) -> std::thread::JoinHandle<()> {
@@ -203,8 +214,14 @@ fn spinup_indexer(
             // TODO/NOTE: currently indexer does not have any interrupt handlers and will never yield back
             // as successful. We can add interrupt handlers in the future but this is not important right
             // now since we managing nodes through integration tests that can kill it or through docker.
-            let Err(err) = indexer::run(options, mpc_contract_id, account_id, sign_queue, gcp)
-            else {
+            let Err(err) = indexer::run(
+                options,
+                mpc_contract_id,
+                account_id,
+                light_client_addr.clone(),
+                sign_queue,
+                gcp,
+            ) else {
                 break;
             };
             tracing::error!(%err, "indexer failed");
@@ -239,6 +256,7 @@ pub fn run(cmd: Cli) -> anyhow::Result<()> {
         Cli::Start {
             near_rpc,
             web_port,
+            light_client_addr,
             mpc_contract_id,
             account_id,
             account_sk,
@@ -262,10 +280,12 @@ pub fn run(cmd: Cli) -> anyhow::Result<()> {
                 .block_on(async {
                     let (sender, receiver) = mpsc::channel(16384);
                     let gcp_service = GcpService::init(&account_id, &storage_options).await?;
+
                     let indexer_handle = spinup_indexer(
                         &indexer_options,
                         &mpc_contract_id,
                         &account_id,
+                        Url::parse(&light_client_addr).unwrap(),
                         &sign_queue,
                         &gcp_service,
                     );

chain-signatures/node/src/indexer.rs

@@ -12,6 +12,7 @@ use serde::{Deserialize, Serialize};
 use std::sync::Arc;
 use std::time::Instant;
 use tokio::sync::RwLock;
+use url::Url;
 
 /// Configures indexer.
 #[derive(Debug, Clone, clap::Parser)]
@@ -82,15 +83,33 @@ pub struct ContractSignRequest {
 struct Context {
     mpc_contract_id: AccountId,
     node_account_id: AccountId,
+    light_client_addr: Url,
     gcp_service: GcpService,
     queue: Arc<RwLock<SignQueue>>,
     latest_block_height: Arc<RwLock<LatestBlockHeight>>,
 }
 
+// TODO:
+// 1. Spin LC docker image in tests, propagate IP address
+// 2. Get a result from it in code
+// 3. Make sure the flow is valid and actually making sure data is fine
+// 4. Add Terraform config and other staff to run it in VMs
+
 async fn handle_block(
     mut block: near_lake_primitives::block::Block,
     ctx: &Context,
 ) -> anyhow::Result<()> {
+    // log light client address
+    tracing::info!(
+        light_client_addr = ctx.light_client_addr.as_str(),
+        "lignt client address"
+    );
+    // Check block integrity/hash
+    // TODO
+    // Get proof using light client (internally - RPC): /proof TransactionOrReceiptId -> head_block_root: CryptoHash, proof: Box<BasicProof>, (proof is RpcLightClientExecutionProofResponse)
+    // TODO
+    // Verify proof: /proof/verify head_block_root: CryptoHash, proof: Box<BasicProof> -> bool
+    // TODO
     for action in block.actions().cloned().collect::<Vec<_>>() {
         if action.receiver_id() == ctx.mpc_contract_id {
             let receipt =
@@ -172,6 +191,7 @@ pub fn run(
     options: Options,
     mpc_contract_id: AccountId,
     node_account_id: AccountId,
+    light_client_addr: Url,
     queue: Arc<RwLock<SignQueue>>,
     gcp_service: crate::gcp::GcpService,
 ) -> anyhow::Result<()> {
@@ -218,6 +238,7 @@ pub fn run(
     let context = Context {
         mpc_contract_id,
         node_account_id,
+        light_client_addr,
         gcp_service,
         queue,
         latest_block_height: Arc::new(RwLock::new(latest_block_height)),

integration-tests/chain-signatures/src/containers.rs

@@ -59,6 +59,7 @@ impl<'a> Node<'a> {
         };
         let args = mpc_recovery_node::cli::Cli::Start {
             near_rpc: near_rpc.clone(),
+            light_client_addr: ctx.light_client.address.clone(),
             mpc_contract_id: mpc_contract_id.clone(),
             account_id: account_id.clone(),
             account_sk: account_sk.to_string().parse()?,
@@ -146,6 +147,7 @@ impl<'a> Node<'a> {
             near_crypto::SecretKey::from_seed(near_crypto::KeyType::ED25519, "integration-test");
         let args = mpc_recovery_node::cli::Cli::Start {
             near_rpc: near_rpc.clone(),
+            light_client_addr: ctx.light_client.address.clone(),
             mpc_contract_id: mpc_contract_id.clone(),
             account_id: account_id.clone(),
             account_sk: account_sk.to_string().parse()?,
@@ -483,6 +485,79 @@ impl<'a> LakeIndexer<'a> {
     }
 }
 
+pub struct LightClient<'a> {
+    pub container: Container<'a, GenericImage>,
+    pub bucket_name: String, // TODO: Do we need this?
+    pub region: String,      // TODO: Do we need this?
+    pub address: String,
+    pub host_address: String,
+    // TODO: any additional fields?
+}
+
+impl<'a> LightClient<'a> {
+    pub const CONTAINER_RPC_PORT: u16 = 3031; // TODO: is this port ok?
+
+    pub async fn run(
+        docker_client: &'a DockerClient,
+        network: &str,
+        bucket_name: String,
+        region: String,
+    ) -> anyhow::Result<LightClient<'a>> {
+        tracing::info!(
+            network,
+            bucket_name,
+            region,
+            "running LightClient container..."
+        );
+
+        let image = GenericImage::new(
+            "ghcr.io/near/near-light-client/light-client",
+            "f4ce326d9c2a6728e8bc39b4d8720f81be87dc43",
+        ) // TODO: deploy and update docker image version
+        // .with_env_var("AWS_ACCESS_KEY_ID", "FAKE_LOCALSTACK_KEY_ID") // Replace with LightCLient specific env vars if any
+        // .with_env_var("AWS_SECRET_ACCESS_KEY", "FAKE_LOCALSTACK_ACCESS_KEY")
+        .with_wait_for(WaitFor::message_on_stderr("Starting Streamer")) // TODO: replace with message from LightClient
+        .with_exposed_port(Self::CONTAINER_RPC_PORT);
+        let image: RunnableImage<GenericImage> = (
+            image,
+            vec![
+                "--bucket".to_string(), // TODO: check if we need these args, what args do we need?
+                bucket_name.clone(),
+                "--region".to_string(),
+                region.clone(),
+                "--stream-while-syncing".to_string(),
+                "sync-from-latest".to_string(),
+            ],
+        )
+            .into();
+        let image = image.with_network(network);
+        let container = docker_client.cli.run(image);
+        let address = docker_client
+            .get_network_ip_address(&container, network)
+            .await?;
+
+        // TODO: check address and host_port, is it what we need?
+        let address = format!("http://{}:{}", address, Self::CONTAINER_RPC_PORT);
+        let host_port = container.get_host_port_ipv4(Self::CONTAINER_RPC_PORT);
+        let host_address = format!("http://127.0.0.1:{host_port}");
+
+        tracing::info!(
+            bucket_name,
+            region,
+            address,
+            host_address,
+            "NEAR Lake Indexer container is running"
+        );
+        Ok(LightClient {
+            container,
+            bucket_name,
+            region,
+            address,
+            host_address,
+        })
+    }
+}
+
 pub struct DockerClient {
     pub docker: Docker,
     pub cli: Cli,

integration-tests/chain-signatures/src/lib.rs

@@ -11,6 +11,7 @@ use crate::containers::LocalStack;
 
 use anyhow::Context as _;
 use bollard::exec::{CreateExecOptions, StartExecResults};
+use containers::LightClient;
 use futures::StreamExt;
 use mpc_contract::primitives::CandidateInfo;
 use mpc_recovery_node::gcp::GcpService;
@@ -246,6 +247,7 @@ pub struct Context<'a> {
     pub mpc_contract: Contract,
     pub datastore: crate::containers::Datastore<'a>,
     pub storage_options: storage::Options,
+    pub light_client: LightClient<'a>,
 }
 
 pub async fn setup(docker_client: &DockerClient) -> anyhow::Result<Context<'_>> {
@@ -259,6 +261,15 @@ pub async fn setup(docker_client: &DockerClient) -> anyhow::Result<Context<'_>>
         worker,
     } = initialize_lake_indexer(docker_client, docker_network).await?;
 
+    // TODO: should it be moved to LakeIndexerCtx?
+    let light_client = LightClient::run(
+        docker_client,
+        docker_network,
+        "bucket name".to_owned(),
+        "region".to_owned(),
+    )
+    .await?;
+
     let mpc_contract = worker
         .dev_deploy(&std::fs::read(
             execute::target_dir()
@@ -290,6 +301,7 @@ pub async fn setup(docker_client: &DockerClient) -> anyhow::Result<Context<'_>>
         mpc_contract,
         datastore,
         storage_options,
+        light_client,
     })
 }
 

integration-tests/chain-signatures/src/local.rs

@@ -50,6 +50,7 @@ impl Node {
         let mpc_contract_id = ctx.mpc_contract.id().clone();
         let cli = mpc_recovery_node::cli::Cli::Start {
             near_rpc: near_rpc.clone(),
+            light_client_addr: ctx.light_client.address.clone(),
             mpc_contract_id: mpc_contract_id.clone(),
             account_id: account_id.clone(),
             account_sk: account_sk.to_string().parse()?,
@@ -108,6 +109,7 @@ impl Node {
         let mpc_contract_id = ctx.mpc_contract.id().clone();
         let cli = mpc_recovery_node::cli::Cli::Start {
             near_rpc: near_rpc.clone(),
+            light_client_addr: ctx.light_client.address.clone(),
             mpc_contract_id: mpc_contract_id.clone(),
             account_id: account_id.clone(),
             account_sk: account_sk.to_string().parse()?,

integration-tests/chain-signatures/src/utils.rs

@@ -1,5 +1,6 @@
 use anyhow::Context;
 use hyper::{Body, Client, Method, Request, StatusCode, Uri};
+use near_primitives::types::TransactionOrReceiptId;
 use near_workspaces::result::ExecutionFinalResult;
 use near_workspaces::{Account, AccountId};
 use std::fs;
@@ -119,3 +120,13 @@ pub async fn clear_local_sk_shares(sk_local_path: Option<String>) -> anyhow::Res
     }
     Ok(())
 }
+
+pub struct Proof {}
+
+pub async fn get_proof(_addr: &str, _id: TransactionOrReceiptId) -> anyhow::Result<Proof> {
+    todo!()
+}
+
+pub async fn verify_proof(_addr: &str, _proof: Proof) -> anyhow::Result<bool> {
+    todo!()
+}