Merge pull request #1 from iosis-tech/init-gossip

basic gossip peer with node impl

Cargo.toml

@@ -3,7 +3,13 @@ overflow-checks = true
 
 [workspace]
 resolver = "2"
-members = [ "crates/common", "crates/delegator", "crates/executor", "crates/peer" , "crates/prover"]
+members = [
+    "crates/common",
+    "crates/delegator",
+    "crates/executor",
+    "crates/peer",
+    "crates/prover",
+]
 exclude = []
 
 [workspace.package]
@@ -15,14 +21,25 @@ license-file = "LICENSE"
 [workspace.dependencies]
 async-process = "2.2.0"
 cairo-felt = "0.9.1"
-cairo-proof-parser = { git = "https://github.com/Okm165/cairo-proof-parser", rev = "97a04bbee07330311b38d6f4cecfed3acb237626"}
+cairo-proof-parser = { git = "https://github.com/Okm165/cairo-proof-parser", rev = "97a04bbee07330311b38d6f4cecfed3acb237626" }
 futures = "0.3.30"
 hex = "0.4.3"
 itertools = "0.12.1"
-libp2p = { version = "0.53.2", features = [ "tokio", "gossipsub", "kad", "mdns", "noise", "macros", "tcp", "yamux", "quic"] }
+libp2p = { version = "0.53.2", features = [
+    "tokio",
+    "gossipsub",
+    "kad",
+    "mdns",
+    "noise",
+    "macros",
+    "tcp",
+    "yamux",
+    "quic",
+] }
 num-bigint = "0.4.4"
 serde = "1.0.197"
 serde_json = "1.0.115"
+starknet = "0.9.0"
 tempfile = "3.10.1"
 thiserror = "1.0.58"
 tokio = { version = "1.36", features = ["full"] }
@@ -32,4 +49,5 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 sharp-p2p-common = { path = "crates/common" }
 sharp-p2p-delegator = { path = "crates/delegator" }
 sharp-p2p-executor = { path = "crates/executor" }
-sharp-p2p-prover = { path = "crates/prover" }
+sharp-p2p-prover = { path = "crates/prover" }
+sharp-p2p-peer = { path = "crates/peer" }

crates/delegator/Cargo.toml

@@ -8,6 +8,8 @@ license-file.workspace = true
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+libp2p.workspace = true
 tokio.workspace = true
 tracing-subscriber.workspace = true
-tracing.workspace = true
+tracing.workspace = true
+sharp-p2p-peer.workspace = true

crates/delegator/src/main.rs

@@ -1,3 +1,31 @@
-fn main() {
-    println!("Hello, world!");
+use sharp_p2p_peer::network::Network;
+use sharp_p2p_peer::node::{Node, NodeConfig, NodeType};
+use sharp_p2p_peer::store::Store;
+use std::error::Error;
+use std::time::Duration;
+use tokio::time::sleep;
+use tracing_subscriber::EnvFilter;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    let _ = tracing_subscriber::fmt().with_env_filter(EnvFilter::from_default_env()).try_init();
+
+    // 1. Generate keypair for the node
+    let p2p_local_keypair = libp2p::identity::Keypair::generate_ed25519();
+
+    // 2. Initiate a new node to sync with other peers
+    let store = Store::new();
+    let node_config = NodeConfig::new(
+        NodeType::Delegator,
+        Network::Sepolia,
+        p2p_local_keypair,
+        Vec::new(),
+        store,
+    );
+    let node = Node::new(node_config).await.unwrap();
+    println!("node: {:?}", node);
+
+    loop {
+        sleep(Duration::from_secs(1)).await;
+    }
 }

crates/executor/Cargo.toml

@@ -8,6 +8,8 @@ license-file.workspace = true
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+libp2p.workspace = true
 tokio.workspace = true
 tracing-subscriber.workspace = true
-tracing.workspace = true
+tracing.workspace = true
+sharp-p2p-peer.workspace = true

crates/executor/readme.md

@@ -0,0 +1 @@
+## Executor

crates/executor/src/main.rs

@@ -1,3 +1,28 @@
-fn main() {
-    println!("Hello, world!");
+use sharp_p2p_peer::network::Network;
+use sharp_p2p_peer::node::{Node, NodeConfig, NodeType};
+use sharp_p2p_peer::store::Store;
+use std::error::Error;
+
+use std::time::Duration;
+use tokio::time::sleep;
+
+use tracing_subscriber::EnvFilter;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    let _ = tracing_subscriber::fmt().with_env_filter(EnvFilter::from_default_env()).try_init();
+
+    // 1. Generate keypair for the node
+    let p2p_local_keypair = libp2p::identity::Keypair::generate_ed25519();
+
+    // 2. Initiate a new node to sync with other peers
+    let store = Store::new();
+    let node_config =
+        NodeConfig::new(NodeType::Executor, Network::Sepolia, p2p_local_keypair, Vec::new(), store);
+    let node = Node::new(node_config).await.unwrap();
+    println!("node: {:?}", node);
+
+    loop {
+        sleep(Duration::from_secs(1)).await;
+    }
 }

crates/peer/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "peer"
+name = "sharp-p2p-peer"
 version.workspace = true
 edition.workspace = true
 repository.workspace = true
@@ -11,4 +11,6 @@ license-file.workspace = true
 libp2p.workspace = true
 tokio.workspace = true
 tracing-subscriber.workspace = true
-tracing.workspace = true
+tracing.workspace = true
+sharp-p2p-common.workspace = true
+starknet.workspace = true

crates/peer/src/lib.rs

@@ -0,0 +1,5 @@
+pub mod network;
+pub mod node;
+pub mod registry;
+pub mod store;
+pub mod swarm;

crates/peer/src/network.rs

@@ -0,0 +1,16 @@
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Network {
+    /// Starknet mainnet.
+    Mainnet,
+    /// Sepolia testnet.
+    Sepolia,
+}
+
+impl Network {
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            Network::Mainnet => "mainnet",
+            Network::Sepolia => "sepolia",
+        }
+    }
+}

crates/peer/src/node.rs

@@ -0,0 +1,64 @@
+use libp2p::identity::Keypair;
+use libp2p::Multiaddr;
+use std::error::Error;
+use std::sync::Arc;
+
+use crate::network::Network;
+use crate::registry::RegistryHandler;
+use crate::store::Store;
+use crate::swarm::SwarmRunner;
+
+pub enum NodeType {
+    Delegator,
+    Executor,
+}
+
+pub struct NodeConfig {
+    pub node_type: NodeType,
+    /// An id of the network to connect to.
+    pub network: Network,
+    /// The keypair to be used as [`Node`]s identity.
+    pub p2p_local_keypair: Keypair,
+    /// List of the addresses where [`Node`] will listen for incoming connections.
+    pub p2p_listen_on: Vec<Multiaddr>,
+    /// The store for job record.
+    pub store: Store,
+}
+
+impl NodeConfig {
+    pub fn new(
+        node_type: NodeType,
+        network: Network,
+        p2p_local_keypair: Keypair,
+        p2p_listen_on: Vec<Multiaddr>,
+        store: Store,
+    ) -> Self {
+        Self { node_type, network, p2p_local_keypair, p2p_listen_on, store }
+    }
+}
+
+#[derive(Debug)]
+pub struct Node {
+    pub store: Arc<Store>,
+}
+
+impl Node {
+    pub async fn new(node_config: NodeConfig) -> Result<Self, Box<dyn Error>> {
+        let mut swarm_runner = SwarmRunner::new(&node_config)?;
+        let registry_handler = RegistryHandler::new(
+            "https://starknet-sepolia.public.blastapi.io",
+            "0xcdd51fbc4e008f4ef807eaf26f5043521ef5931bbb1e04032a25bd845d286b",
+        );
+        // Node should run swarm runner and registry handler concurrently.
+        tokio::spawn(async move {
+            swarm_runner.run(node_config.node_type).await;
+        });
+        tokio::spawn(async move {
+            registry_handler.run().await;
+        });
+
+        let store = Arc::new(node_config.store);
+
+        Ok(Self { store })
+    }
+}

crates/peer/src/registry.rs

@@ -0,0 +1,72 @@
+use std::error::Error;
+
+use starknet::{
+    core::types::{BlockId, EmittedEvent, EventFilter, FieldElement},
+    providers::{jsonrpc::HttpTransport, JsonRpcClient, Provider, Url},
+};
+
+pub struct RegistryHandler {
+    pub provider: JsonRpcClient<HttpTransport>,
+    address: FieldElement,
+}
+
+impl RegistryHandler {
+    pub fn new(url: &str, address: &str) -> Self {
+        let provider = JsonRpcClient::new(HttpTransport::new(Url::parse(url).unwrap()));
+        let address = FieldElement::from_hex_be(address).unwrap();
+        Self { provider, address }
+    }
+
+    async fn scrape_event(
+        &self,
+        event_keys: Vec<String>,
+        from_block: u64,
+    ) -> Result<Vec<EmittedEvent>, Box<dyn Error>> {
+        let keys = event_keys
+            .iter()
+            .map(|key| FieldElement::from_hex_be(key))
+            .collect::<Result<Vec<FieldElement>, _>>()?;
+
+        let latest_block_number = self.provider.block_number().await?;
+
+        let filter = EventFilter {
+            from_block: Some(BlockId::Number(from_block)),
+            to_block: Some(BlockId::Number(latest_block_number)),
+            address: Some(self.address),
+            keys: Some(vec![keys.clone()]),
+        };
+
+        let events = self.provider.get_events(filter, None, 1000).await?.events;
+        Ok(events)
+    }
+
+    pub async fn run(&self) {
+        // Create an interval of every 5 seconds
+        let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(5));
+
+        loop {
+            interval.tick().await;
+
+            println!("Scraping events...");
+
+            // Scrape the event
+            let result = self
+                .scrape_event(
+                    vec!["0x17ef19eae2188756c1689ef60586c692a3aee6fecc18ee1b21f3028f75b9988"
+                        .to_string()],
+                    0,
+                )
+                .await;
+
+            // Handle the result
+            match result {
+                Ok(events) => {
+                    println!("{} Events Found", events.len());
+                }
+                Err(e) => {
+                    eprintln!("Error scraping events: {:?}", e);
+                }
+            }
+        }
+    }
+}

crates/peer/src/store.rs

@@ -0,0 +1,22 @@
+use sharp_p2p_common::job::Job;
+
+use std::collections::VecDeque;
+
+#[derive(Debug)]
+pub struct Store {
+    /// For delegator, FIFO queue to publish message
+    /// For executor, FIFO queue to prove job
+    pub job_queue: VecDeque<Job>,
+}
+
+impl Store {
+    pub fn new() -> Self {
+        Self { job_queue: VecDeque::new() }
+    }
+}
+
+impl Default for Store {
+    fn default() -> Self {
+        Self::new()
+    }
+}