Fixed tests in connection discovery

.github/workflows/ci.yaml

@@ -206,7 +206,7 @@ jobs:
       BPF_ALIAS: /coda/0.0.1/29936104443aaf264a7f0192ac64b1c7173198c1ed404c1bcff5e562e05eb7f6-0.0.0.0
     strategy:
       matrix:
-        test: [p2p_basic_connections, p2p_basic_incoming, p2p_basic_outgoing, p2p_pubsub]
+        test: [p2p_basic_connections, p2p_basic_incoming, p2p_basic_outgoing, p2p_pubsub, p2p_kad]
       fail-fast: false
 
     services:
@@ -295,9 +295,13 @@ jobs:
         test:
           - single_node
           - multi_node
+          - connection_discovery_ocaml_to_rust_via_seed
+          - connection_discovery_ocaml_to_rust
+          - connection_discovery_rust_as_seed
+          - connection_discovery_rust_to_ocaml_via_seed
+          - connection_discovery_rust_to_ocaml
           # - webrtc_single_node
           # - webrtc_multi_node
-          # - connection_discovery
       fail-fast: false
 
     services:

node/src/snark/snark_effects.rs

@@ -1,6 +1,6 @@
 use snark::work_verify::SnarkWorkVerifyAction;
 
-use crate::{snark_pool::candidate::SnarkPoolCandidateAction, Service, SnarkPoolAction, Store};
+use crate::{snark_pool::candidate::SnarkPoolCandidateAction, Service, Store};
 
 use super::{SnarkAction, SnarkActionWithMeta};
 

node/testing/src/cluster/runner/mod.rs

@@ -9,6 +9,7 @@ use node::{event_source::Event, ledger::LedgerService, ActionKind, State};
 use rand::{rngs::StdRng, SeedableRng};
 use time::OffsetDateTime;
 
+use crate::node::OcamlStep;
 use crate::{
     cluster::{Cluster, ClusterNodeId, ClusterOcamlNodeId},
     network_debugger::Debugger,
@@ -450,4 +451,15 @@ impl<'a> ClusterRunner<'a> {
         )
         .await
     }
+
+    pub async fn wait_for_ocaml(&mut self, node_id: ClusterOcamlNodeId) {
+        self.exec_step(ScenarioStep::Ocaml {
+            node_id,
+            step: OcamlStep::WaitReady {
+                timeout: Duration::from_secs(6 * 60),
+            },
+        })
+        .await
+        .expect("Error waiting for ocaml node");
+    }
 }

node/testing/src/node/ocaml/config.rs

@@ -15,6 +15,16 @@ pub struct OcamlNodeTestingConfig {
     pub block_producer: Option<AccountSecretKey>,
 }
 
+impl Default for OcamlNodeTestingConfig {
+    fn default() -> Self {
+        Self {
+            initial_peers: vec![],
+            daemon_json: DaemonJson::Custom("/var/lib/coda/config_dc6bf78b.json".to_owned()),
+            block_producer: None,
+        }
+    }
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub enum DaemonJson {
     // TODO(binier): have presets.

node/testing/src/node/ocaml/mod.rs

@@ -25,7 +25,7 @@ pub struct OcamlNode {
     temp_dir: temp_dir::TempDir,
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone)]
+#[derive(Serialize, Deserialize, Debug, Clone, Copy)]
 pub enum OcamlStep {
     /// Wait till ocaml node is ready.
     ///
@@ -294,18 +294,9 @@ impl OcamlNode {
     }
 
     /// Queries graphql to get chain_id.
-    pub fn chain_id(&self) -> anyhow::Result<ChainId> {
-        let res = self.grapql_query("query { daemonStatus { chainId } }")?;
-        let chain_id = res["data"]["daemonStatus"]["chainId"]
-            .as_str()
-            .ok_or_else(|| anyhow::anyhow!("empty chain_id response"))?;
-        ChainId::from_hex(chain_id).map_err(|e| anyhow::anyhow!("invalid chain_id: {}", e))
-    }
-
-    /// Queries graphql to get chain_id.
-    pub async fn chain_id_async(&self) -> anyhow::Result<ChainId> {
+    pub async fn chain_id(&self) -> anyhow::Result<ChainId> {
         let res = self
-            .grapql_query_async("query { daemonStatus { chainId } }")
+            .grapql_query("query { daemonStatus { chainId } }")
             .await?;
         let chain_id = res["data"]["daemonStatus"]["chainId"]
             .as_str()
@@ -315,21 +306,9 @@ impl OcamlNode {
 
     /// Queries graphql to check if ocaml node is synced,
     /// returning it's best tip hash if yes.
-    pub fn synced_best_tip(&self) -> anyhow::Result<Option<StateHash>> {
-        let mut res = self.grapql_query("query { daemonStatus { syncStatus, stateHash } }")?;
-        let data = &mut res["data"]["daemonStatus"];
-        if data["syncStatus"].as_str() == Some("SYNCED") {
-            Ok(Some(serde_json::from_value(data["stateHash"].take())?))
-        } else {
-            Ok(None)
-        }
-    }
-
-    /// Queries graphql to check if ocaml node is synced,
-    /// returning it's best tip hash if yes.
-    pub async fn synced_best_tip_async(&self) -> anyhow::Result<Option<StateHash>> {
+    pub async fn synced_best_tip(&self) -> anyhow::Result<Option<StateHash>> {
         let mut res = self
-            .grapql_query_async("query { daemonStatus { syncStatus, stateHash } }")
+            .grapql_query("query { daemonStatus { syncStatus, stateHash } }")
             .await?;
         let data = &mut res["data"]["daemonStatus"];
         if data["syncStatus"].as_str() == Some("SYNCED") {
@@ -348,7 +327,7 @@ impl OcamlNode {
     // Only `exec` function should be exposed and instead of this, we
     // should have a step to query graphql and assert response as a part
     // of that step.
-    pub async fn grapql_query_async(&self, query: &str) -> anyhow::Result<serde_json::Value> {
+    pub async fn grapql_query(&self, query: &str) -> anyhow::Result<serde_json::Value> {
         let client = reqwest::Client::new();
         let response = client
             .post(self.graphql_addr())
@@ -362,24 +341,6 @@ impl OcamlNode {
 
         Ok(response.json().await?)
     }
-    // TODO(binier): shouldn't be publically accessible.
-    //
-    // Only `exec` function should be exposed and instead of this, we
-    // should have a step to query graphql and assert response as a part
-    // of that step.
-    pub fn grapql_query(&self, query: &str) -> anyhow::Result<serde_json::Value> {
-        let client = reqwest::blocking::Client::new();
-        let response = client
-            .post(self.graphql_addr())
-            .json(&{
-                serde_json::json!({
-                    "query": query
-                })
-            })
-            .send()?;
-
-        Ok(response.json()?)
-    }
 
     async fn wait_for_p2p(&self, timeout: Duration) -> anyhow::Result<()> {
         let port = self.libp2p_port;
@@ -408,7 +369,7 @@ impl OcamlNode {
             loop {
                 interval.tick().await;
                 if self
-                    .synced_best_tip_async()
+                    .synced_best_tip()
                     .await
                     .map_or(false, |tip| tip.is_some())
                 {