diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index 511bb2f12f..4d7e905c01 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -7,12 +7,18 @@ jobs: runs-on: ubuntu-latest strategy: matrix: - job_args: [servers, chain, core, keychain, pool, p2p, src, api, util, store] + job_args: [servers, chain, core, keychain, pool, p2p, src, api, util, store, integration] steps: - uses: actions/checkout@v3 - name: Test ${{ matrix.job_args }} working-directory: ${{ matrix.job_args }} - run: cargo test --release + # Integration tests open real ports and multi-node clusters; run them serially. + run: | + if [ "${{ matrix.job_args }}" = "integration" ]; then + cargo test --release -- --test-threads=1 + else + cargo test --release + fi macos-tests: name: macOS Tests diff --git a/Cargo.lock b/Cargo.lock index c2ba308026..f9b3306d21 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -235,6 +235,12 @@ dependencies = [ "objc2", ] +[[package]] +name = "bufstream" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "40e38929add23cdf8a366df9b0e088953150724bcbe5fc330b0d8eb3b328eec8" + [[package]] name = "built" version = "0.8.1" @@ -1125,6 +1131,21 @@ dependencies = [ "zeroize", ] +[[package]] +name = "grin_integration" +version = "5.5.1-alpha.0" +dependencies = [ + "bufstream", + "futures 0.3.32", + "grin_api", + "grin_core", + "grin_p2p", + "grin_servers", + "grin_util", + "log", + "serde_json", +] + [[package]] name = "grin_keychain" version = "5.5.1-alpha.0" diff --git a/Cargo.toml b/Cargo.toml index 4ff9a98568..78ba46998f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -12,7 +12,7 @@ build = "src/build/build.rs" edition = "2021" [workspace] -members = ["api", "chain", "config", "core", "keychain", "p2p", "servers", "store", "util", "pool"] +members = ["api", "chain", "config", "core", "keychain", "p2p", "servers", "store", "util", "pool", "integration"] exclude = ["etc/gen_gen"] [[bin]] diff --git a/integration/Cargo.toml b/integration/Cargo.toml new file mode 100644 index 0000000000..8dff054d93 --- /dev/null +++ b/integration/Cargo.toml @@ -0,0 +1,23 @@ +[package] +name = "grin_integration" +version = "5.5.1-alpha.0" +authors = ["Grin Developers "] +description = "Multi-node integration tests for the Grin node" +license = "Apache-2.0" +repository = "https://github.com/mimblewimble/grin" +keywords = ["crypto", "grin", "mimblewimble"] +workspace = ".." +edition = "2021" +publish = false + +[dependencies] +bufstream = "0.1" +futures = "0.3" +log = "0.4" +serde_json = "1" + +grin_api = { path = "../api", version = "5.5.1-alpha.0" } +grin_core = { path = "../core", version = "5.5.1-alpha.0" } +grin_p2p = { path = "../p2p", version = "5.5.1-alpha.0" } +grin_servers = { path = "../servers", version = "5.5.1-alpha.0" } +grin_util = { path = "../util", version = "5.5.1-alpha.0" } diff --git a/integration/README.md b/integration/README.md new file mode 100644 index 0000000000..5e9decbd6e --- /dev/null +++ b/integration/README.md @@ -0,0 +1,31 @@ +# Node integration tests + +Multi-node integration coverage for the Grin **node**, re-introduced after the +wallet split ([#2957](https://github.com/mimblewimble/grin/issues/2957)). + +These tests do **not** depend on `grin-wallet`. Coinbase is burned via the +internal test miner (`start_test_miner(None, …)`). Wallet-coupled scenarios +(payments, dandelion with wallets, owner/foreign APIs) remain in the +[grin-wallet](https://github.com/mimblewimble/grin-wallet) repository. + +## Run + +From the repo root or this crate: + +```bash +cd integration +cargo test --release -- --test-threads=1 +``` + +Serial execution (`--test-threads=1`) avoids port and RocksDB races between +clusters. CI runs this crate the same way. + +## Suites + +| File | Coverage | +|------|----------| +| `tests/simulnet.rs` | Mining, seeding, block propagation, full/long body sync | +| `tests/p2p.rs` | Peer connect, ban, unban | +| `tests/stratum.rs` | Live stratum JSON-RPC + job broadcast | + +Shared helpers live in `tests/common/`. diff --git a/integration/src/lib.rs b/integration/src/lib.rs new file mode 100644 index 0000000000..921af8352d --- /dev/null +++ b/integration/src/lib.rs @@ -0,0 +1,19 @@ +// Copyright 2021 The Grin Developers +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Multi-node integration tests for the Grin node. +//! +//! These tests re-introduce node-only coverage that lived in `servers/tests` +//! before the wallet was extracted (see mimblewimble/grin#2957). Wallet-coupled +//! flows remain in the `grin-wallet` repository. diff --git a/integration/tests/common/mod.rs b/integration/tests/common/mod.rs new file mode 100644 index 0000000000..e315be79d7 --- /dev/null +++ b/integration/tests/common/mod.rs @@ -0,0 +1,332 @@ +// Copyright 2021 The Grin Developers +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Shared helpers for multi-node integration tests. +//! +//! Node-only: mining rewards are burned (no wallet dependency). + +#![allow(dead_code)] + +use futures::channel::oneshot; +use grin_core as core; +use grin_core::global::{self, ChainTypes}; +use grin_p2p as p2p; +use grin_servers as servers; +use grin_util::StopState; +use p2p::msg::PeerAddrs; +use p2p::PeerAddr; +use std::default::Default; +use std::net::{IpAddr, Ipv4Addr, SocketAddr}; +use std::path::Path; +use std::sync::Arc; +use std::{fs, thread, time}; + +/// Configure AutomatedTesting for the test thread and all server worker threads. +pub fn init_chain() { + global::set_local_chain_type(ChainTypes::AutomatedTesting); + global::set_global_chain_type(ChainTypes::AutomatedTesting); +} + +/// Remove leftover data from a previous run of this test. +pub fn clean_all_output(test_name_dir: &str) { + let target_dir = format!("target/tmp/{}", test_name_dir); + if let Err(e) = fs::remove_dir_all(&target_dir) { + // Missing dir is fine on first run. + if Path::new(&target_dir).exists() { + println!( + "can't remove output from previous test {}: {}, may be ok", + target_dir, e + ); + } + } +} + +/// Leak a oneshot channel pair required by `Server::new` for API shutdown. +pub fn leak_api_chan() -> &'static mut (oneshot::Sender<()>, oneshot::Receiver<()>) { + Box::leak(Box::new(oneshot::channel::<()>())) +} + +/// Parse `ip:port` into a `PeerAddr`. +pub fn peer_addr(addr: &str) -> PeerAddr { + PeerAddr(addr.parse().expect("valid peer address")) +} + +/// Build a `ServerConfig` for integration tests. +/// +/// - `n` offsets ports so parallel tests do not collide +/// - `seed_n` is the peer index used when `seeding_type` is `List` +pub fn config(n: u16, test_name_dir: &str, seed_n: u16) -> servers::ServerConfig { + let seed = SocketAddr::new( + IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), + 10000 + seed_n, + ); + servers::ServerConfig { + api_http_addr: format!("127.0.0.1:{}", 20000 + n), + api_secret_path: None, + foreign_api_secret_path: None, + db_root: format!("target/tmp/{}/grin-sync-{}", test_name_dir, n), + p2p_config: p2p::P2PConfig { + host: IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), + port: 10000 + n, + seeding_type: p2p::Seeding::List, + seeds: Some(PeerAddrs { + peers: vec![PeerAddr(seed)], + }), + ..p2p::P2PConfig::default() + }, + chain_type: core::global::ChainTypes::AutomatedTesting, + archive_mode: Some(true), + skip_sync_wait: Some(true), + run_tui: Some(false), + run_test_miner: Some(false), + stratum_mining_config: None, + ..Default::default() + } +} + +/// Stratum mining config suitable for tests (rewards burned). +pub fn stratum_config() -> servers::StratumServerConfig { + servers::StratumServerConfig { + enable_stratum_server: Some(true), + stratum_server_addr: Some(String::from("127.0.0.1:13416")), + attempt_time_per_block: 60, + minimum_share_difficulty: 1, + wallet_listener_url: String::from("http://127.0.0.1:13415"), + burn_reward: true, + } +} + +/// Start a server with the given config. +pub fn start_server(cfg: servers::ServerConfig) -> servers::Server { + servers::Server::new(cfg, None, None, leak_api_chan()).expect("server starts") +} + +/// Start a server and attach the internal test miner (burns coinbase). +pub fn start_mining_server(cfg: servers::ServerConfig) -> (servers::Server, Arc) { + let server = start_server(cfg); + let miner_stop = Arc::new(StopState::new()); + server.start_test_miner(None, miner_stop.clone()); + (server, miner_stop) +} + +/// Stop a list of servers cleanly (consumes them). +pub fn stop_all_servers(servers: Vec) { + for s in servers { + s.stop(); + } +} + +/// Brief pause so ports and locks release between tests. +pub fn settle() { + thread::sleep(time::Duration::from_millis(500)); +} + +/// Configuration for a single local server instance. +#[derive(Clone)] +pub struct LocalServerContainerConfig { + pub name: String, + pub base_addr: String, + pub p2p_server_port: u16, + pub api_server_port: u16, + pub start_miner: bool, + pub seed_addr: String, + pub is_seeding: bool, + pub peer_list: Vec, +} + +impl Default for LocalServerContainerConfig { + fn default() -> LocalServerContainerConfig { + LocalServerContainerConfig { + name: String::from("test_host"), + base_addr: String::from("127.0.0.1"), + api_server_port: 13413, + p2p_server_port: 13414, + seed_addr: String::from(""), + is_seeding: false, + start_miner: false, + peer_list: Vec::new(), + } + } +} + +/// Builds and starts a single node (optionally mining). +pub struct LocalServerContainer { + pub config: LocalServerContainerConfig, + working_dir: String, +} + +impl LocalServerContainer { + pub fn new(config: LocalServerContainerConfig) -> LocalServerContainer { + let working_dir = format!("target/tmp/{}", config.name); + LocalServerContainer { + config, + working_dir, + } + } + + pub fn add_peer(&mut self, addr: String) { + self.config.peer_list.push(addr); + } + + /// Start the server (and optional test miner). Returns the running server. + pub fn run_server(self) -> servers::Server { + let api_addr = format!("{}:{}", self.config.base_addr, self.config.api_server_port); + + let mut seeding_type = p2p::Seeding::None; + let mut seeds = PeerAddrs { peers: vec![] }; + + if !self.config.seed_addr.is_empty() { + seeding_type = p2p::Seeding::List; + seeds.peers = vec![peer_addr(&self.config.seed_addr)]; + } + + let cfg = servers::ServerConfig { + api_http_addr: api_addr, + api_secret_path: None, + foreign_api_secret_path: None, + db_root: format!("{}/.grin", self.working_dir), + p2p_config: p2p::P2PConfig { + host: self + .config + .base_addr + .parse() + .unwrap_or(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1))), + port: self.config.p2p_server_port, + seeds: if seeds.peers.is_empty() { + None + } else { + Some(seeds) + }, + seeding_type, + ..p2p::P2PConfig::default() + }, + chain_type: core::global::ChainTypes::AutomatedTesting, + skip_sync_wait: Some(true), + archive_mode: Some(true), + run_tui: Some(false), + run_test_miner: Some(false), + stratum_mining_config: None, + ..Default::default() + }; + + let s = start_server(cfg); + + if self.config.start_miner { + println!( + "starting test Miner on port {}", + self.config.p2p_server_port + ); + s.start_test_miner(None, s.stop_state.clone()); + } + + for p in &self.config.peer_list { + println!("{} connecting to peer: {}", self.config.p2p_server_port, p); + let _ = s.connect_peer(peer_addr(p)); + } + + s + } +} + +/// Pool configuration for multi-server tests. +pub struct LocalServerContainerPoolConfig { + pub base_name: String, + pub base_p2p_port: u16, + pub base_api_port: u16, + pub run_length_in_seconds: u64, +} + +impl Default for LocalServerContainerPoolConfig { + fn default() -> LocalServerContainerPoolConfig { + LocalServerContainerPoolConfig { + base_name: String::from("test_pool"), + base_p2p_port: 10000, + base_api_port: 11000, + run_length_in_seconds: 30, + } + } +} + +/// Convenience pool for starting several servers with consecutive ports. +pub struct LocalServerContainerPool { + pub config: LocalServerContainerPoolConfig, + server_containers: Vec, + next_p2p_port: u16, + next_api_port: u16, + is_seeding: bool, +} + +impl LocalServerContainerPool { + pub fn new(config: LocalServerContainerPoolConfig) -> LocalServerContainerPool { + LocalServerContainerPool { + next_api_port: config.base_api_port, + next_p2p_port: config.base_p2p_port, + config, + server_containers: Vec::new(), + is_seeding: false, + } + } + + /// Add a server using the next free ports. Mutates `server_config` with assigned values. + pub fn create_server(&mut self, server_config: &mut LocalServerContainerConfig) { + server_config.p2p_server_port = self.next_p2p_port; + server_config.api_server_port = self.next_api_port; + server_config.name = format!( + "{}/{}-{}", + self.config.base_name, self.config.base_name, server_config.p2p_server_port + ); + + self.next_p2p_port += 1; + self.next_api_port += 1; + + if server_config.is_seeding { + self.is_seeding = true; + } + + self.server_containers + .push(LocalServerContainer::new(server_config.clone())); + } + + /// Start all servers, returning owned `Server` instances. + pub fn run_all_servers(self) -> Vec { + let mut handles = vec![]; + let return_containers = Arc::new(std::sync::Mutex::new(Vec::new())); + let is_seeding = self.is_seeding; + + for s in self.server_containers { + let return_container_ref = return_containers.clone(); + let handle = thread::spawn(move || { + if is_seeding && !s.config.is_seeding { + // Give the seed a head start. + thread::sleep(time::Duration::from_millis(2000)); + } + let server_ref = s.run_server(); + return_container_ref.lock().unwrap().push(server_ref); + }); + // RocksDB concurrent create can fail without a short gap. + thread::sleep(time::Duration::from_millis(500)); + handles.push(handle); + } + + for handle in handles { + handle.join().expect("server thread"); + } + + // Keep run_length for API compatibility with old tests that waited this long. + let _ = self.config.run_length_in_seconds; + + let mut guard = return_containers.lock().unwrap(); + std::mem::take(&mut *guard) + } +} diff --git a/integration/tests/p2p.rs b/integration/tests/p2p.rs new file mode 100644 index 0000000000..c5d8db6638 --- /dev/null +++ b/integration/tests/p2p.rs @@ -0,0 +1,103 @@ +// Copyright 2021 The Grin Developers +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! P2P peer lifecycle: connect, ban, unban (node-side, no wallet). + +#[macro_use] +extern crate log; + +mod common; + +use crate::common::{ + clean_all_output, init_chain, peer_addr, settle, stop_all_servers, LocalServerContainer, + LocalServerContainerConfig, +}; +use grin_p2p as p2p; +use grin_util as util; +use std::{thread, time}; + +/// Two nodes handshake; ban/unban updates peer store state. +#[test] +fn test_p2p_ban_unban() { + util::init_test_logger(); + info!("starting test_p2p_ban_unban"); + init_chain(); + + let server_one_dir = "p2p_server_one"; + clean_all_output(server_one_dir); + let mut server_config_one = LocalServerContainerConfig::default(); + server_config_one.name = String::from(server_one_dir); + server_config_one.p2p_server_port = 40002; + server_config_one.api_server_port = 40003; + server_config_one.start_miner = false; + server_config_one.is_seeding = true; + let server_one = LocalServerContainer::new(server_config_one.clone()).run_server(); + + thread::sleep(time::Duration::from_millis(1000)); + + let server_two_dir = "p2p_server_two"; + clean_all_output(server_two_dir); + let mut server_config_two = LocalServerContainerConfig::default(); + server_config_two.name = String::from(server_two_dir); + server_config_two.p2p_server_port = 40004; + server_config_two.api_server_port = 40005; + server_config_two.start_miner = false; + server_config_two.is_seeding = false; + let mut container_two = LocalServerContainer::new(server_config_two.clone()); + container_two.add_peer(format!( + "{}:{}", + server_config_one.base_addr, server_config_one.p2p_server_port + )); + let server_two = container_two.run_server(); + + // Handshake + thread::sleep(time::Duration::from_millis(3000)); + + assert_eq!(server_one.peer_count(), 1); + + let peer_two = peer_addr(&format!( + "{}:{}", + server_config_two.base_addr, server_config_two.p2p_server_port + )); + + // Peer is healthy in the store. + let peer = server_one.p2p.peers.get_peer(peer_two).expect("peer known"); + assert_eq!(peer.flags, p2p::State::Healthy); + + // Ban + server_one + .p2p + .peers + .ban_peer(peer_two, p2p::ReasonForBan::ManualBan) + .expect("ban"); + thread::sleep(time::Duration::from_millis(2000)); + + let peer = server_one.p2p.peers.get_peer(peer_two).expect("peer after ban"); + assert_eq!(peer.flags, p2p::State::Banned); + + // Unban + server_one.p2p.peers.unban_peer(peer_two).expect("unban"); + let peer = server_one + .p2p + .peers + .get_peer(peer_two) + .expect("peer after unban"); + assert_eq!(peer.flags, p2p::State::Healthy); + + // Ban drops the live connection; unban does not auto-reconnect. + assert_eq!(server_one.peer_count(), 0); + + stop_all_servers(vec![server_one, server_two]); + settle(); +} diff --git a/integration/tests/simulnet.rs b/integration/tests/simulnet.rs new file mode 100644 index 0000000000..c921ff3e21 --- /dev/null +++ b/integration/tests/simulnet.rs @@ -0,0 +1,270 @@ +// Copyright 2021 The Grin Developers +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Multi-node simulation tests (mining, seeding, propagation, sync). +//! +//! Ported from the pre-wallet-split `servers/tests` suite (grin#2957). +//! Wallet-dependent scenarios stay in grin-wallet. + +#[macro_use] +extern crate log; + +mod common; + +use crate::common::{ + clean_all_output, config, init_chain, settle, start_mining_server, start_server, + stop_all_servers, LocalServerContainerConfig, LocalServerContainerPool, + LocalServerContainerPoolConfig, +}; +use grin_core::core::hash::Hashed; +use grin_util as util; +use grin_util::StopState; +use std::sync::Arc; +use std::{thread, time}; + +/// Single node mines for a short time then shuts down cleanly. +#[test] +fn basic_genesis_mine() { + util::init_test_logger(); + init_chain(); + + let test_name_dir = "genesis_mine"; + clean_all_output(test_name_dir); + + let mut pool_config = LocalServerContainerPoolConfig::default(); + pool_config.base_name = String::from(test_name_dir); + pool_config.run_length_in_seconds = 10; + pool_config.base_api_port = 30000; + pool_config.base_p2p_port = 31000; + + let mut pool = LocalServerContainerPool::new(pool_config); + + let mut server_config = LocalServerContainerConfig::default(); + server_config.start_miner = true; + server_config.is_seeding = false; + + pool.create_server(&mut server_config); + let servers = pool.run_all_servers(); + // Allow a few mining iterations. + thread::sleep(time::Duration::from_secs(8)); + assert!(servers[0].head().unwrap().height >= 1); + stop_all_servers(servers); + settle(); +} + +/// One seed plus four peers; all should end up connected via peer exchange. +#[test] +fn simulate_seeding() { + util::init_test_logger(); + init_chain(); + + let test_name_dir = "simulate_seeding"; + clean_all_output(test_name_dir); + + let mut pool_config = LocalServerContainerPoolConfig::default(); + pool_config.base_name = test_name_dir.to_string(); + pool_config.run_length_in_seconds = 30; + pool_config.base_api_port = 30020; + pool_config.base_p2p_port = 31020; + + let mut pool = LocalServerContainerPool::new(pool_config); + + let mut server_config = LocalServerContainerConfig::default(); + server_config.start_miner = false; + server_config.is_seeding = true; + + pool.create_server(&mut server_config); + + // Seed fully up before remaining servers. + thread::sleep(time::Duration::from_millis(1_000)); + + server_config.is_seeding = false; + server_config.seed_addr = format!( + "{}:{}", + server_config.base_addr, server_config.p2p_server_port + ); + + for _ in 0..4 { + pool.create_server(&mut server_config); + } + + let servers = pool.run_all_servers(); + thread::sleep(time::Duration::from_secs(8)); + + // Seed should see all four peers connected. + let seed = servers + .iter() + .find(|s| s.config.p2p_config.port == 31020) + .expect("seed server"); + assert_eq!(seed.peer_count(), 4); + + stop_all_servers(servers); + settle(); +} + +/// Five connected nodes; mine on one and verify block height propagates. +#[test] +fn simulate_block_propagation() { + util::init_test_logger(); + init_chain(); + + let test_name_dir = "grin-prop"; + clean_all_output(test_name_dir); + + let mut servers = vec![]; + for n in 0..5 { + let s = start_server(config(10 * n, test_name_dir, 0)); + servers.push(s); + thread::sleep(time::Duration::from_millis(100)); + } + + let stop = Arc::new(StopState::new()); + servers[0].start_test_miner(None, stop.clone()); + + let mut success = false; + let mut time_spent = 0; + loop { + let mut count = 0; + for n in 0..5 { + if servers[n].head().unwrap().height > 3 { + count += 1; + } + } + if count == 5 { + success = true; + break; + } + thread::sleep(time::Duration::from_millis(1_000)); + time_spent += 1; + if time_spent >= 45 { + info!("simulate_block_propagation - fail on timeout"); + break; + } + if time_spent == 12 { + servers[0].stop_test_miner(stop.clone()); + } + } + + stop_all_servers(servers); + assert!(success, "all 5 nodes should reach height > 3"); + settle(); +} + +/// Mine on s1, start s2 seeded from s1, verify s2 syncs headers/blocks. +#[test] +fn simulate_full_sync() { + util::init_test_logger(); + init_chain(); + + let test_name_dir = "grin-sync"; + clean_all_output(test_name_dir); + + let (s1, miner_stop) = start_mining_server(config(1000, test_name_dir, 1000)); + thread::sleep(time::Duration::from_secs(10)); + s1.stop_test_miner(miner_stop); + // Let the miner thread exit so the tip is stable. + thread::sleep(time::Duration::from_secs(2)); + + let s1_header = s1.chain.head_header().unwrap(); + info!( + "simulate_full_sync - s1 header head: {} at {}", + s1_header.hash(), + s1_header.height + ); + assert!( + s1_header.height >= 1, + "s1 should have mined at least one block" + ); + + let s2 = start_server(config(1001, test_name_dir, 1000)); + + let mut time_spent = 0; + let target = s1_header.height; + while s2.head().unwrap().height < target { + thread::sleep(time::Duration::from_millis(1_000)); + time_spent += 1; + if time_spent >= 45 { + info!( + "sync fail. s2.height: {}, s1.height: {}", + s2.head().unwrap().height, + s1.head().unwrap().height + ); + break; + } + } + + // Compare live tips (both archive nodes; full block body sync). + let s1_tip = s1.chain.head_header().unwrap(); + let s2_tip = s2.chain.head_header().unwrap(); + assert_eq!(s1_tip.height, s2_tip.height); + assert_eq!(s1_tip.hash(), s2_tip.hash()); + + s1.stop(); + s2.stop(); + settle(); +} + +/// Mine past the state-sync threshold on archive s1; pruned s2 should catch up. +/// +/// On current main, non-archive peers use PIBD for state sync. This test asserts +/// header + body catch-up when both peers stay in range for body sync after +/// headers are aligned (s2 starts as archive so we exercise long-chain body +/// sync without depending on live PIBD segment serving in CI). A dedicated +/// PIBD multi-node test can be added once segment serving is hardened for +/// AutomatedTesting. +#[test] +fn simulate_long_chain_sync() { + util::init_test_logger(); + init_chain(); + + let test_name_dir = "grin-long-sync"; + clean_all_output(test_name_dir); + + let (s1, miner_stop) = start_mining_server(config(2000, test_name_dir, 2000)); + + while s1.head().unwrap().height < 25 { + thread::sleep(time::Duration::from_millis(1_000)); + } + s1.stop_test_miner(miner_stop); + thread::sleep(time::Duration::from_secs(2)); + + // Second archive node seeds from s1 and body-syncs the full history. + let s2 = start_server(config(2001, test_name_dir, 2000)); + + let s1_header = s1.chain.head_header().unwrap(); + + let mut total_wait = 0; + while s2.head().unwrap().height < s1_header.height { + thread::sleep(time::Duration::from_millis(1_000)); + total_wait += 1; + if total_wait >= 90 { + error!( + "simulate_long_chain_sync timeout! s2 height: {}, s1 height: {}", + s2.head().unwrap().height, + s1_header.height, + ); + break; + } + } + + let s1_tip = s1.chain.head_header().unwrap(); + let s2_tip = s2.chain.head_header().unwrap(); + assert_eq!(s1_tip.height, s2_tip.height); + assert_eq!(s1_tip.hash(), s2_tip.hash()); + assert!(s1_tip.height >= 25); + + s1.stop(); + s2.stop(); + settle(); +} diff --git a/integration/tests/stratum.rs b/integration/tests/stratum.rs new file mode 100644 index 0000000000..8bb079b96a --- /dev/null +++ b/integration/tests/stratum.rs @@ -0,0 +1,140 @@ +// Copyright 2021 The Grin Developers +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Live stratum server integration test (JSON-RPC over TCP). + +#[macro_use] +extern crate log; + +mod common; + +use crate::common::{clean_all_output, config, init_chain, settle, start_server, stratum_config}; +use bufstream::BufStream; +use grin_util as util; +use grin_util::StopState; +use serde_json::Value; +use std::io::prelude::{BufRead, Write}; +use std::net::TcpStream; +use std::sync::Arc; +use std::{thread, time}; + +/// Stratum accepts workers, answers JSON-RPC, and broadcasts jobs when blocks are found. +#[test] +fn basic_stratum_server() { + util::init_test_logger(); + init_chain(); + + let test_name_dir = "stratum_server"; + clean_all_output(test_name_dir); + + let s = start_server(config(4000, test_name_dir, 0)); + + let mut stratum_cfg = stratum_config(); + stratum_cfg.burn_reward = true; + stratum_cfg.attempt_time_per_block = 999; + stratum_cfg.enable_stratum_server = Some(true); + stratum_cfg.stratum_server_addr = Some(String::from("127.0.0.1:11101")); + + s.start_stratum_server(stratum_cfg); + + // Wait until stratum accepts TCP connections. + loop { + if TcpStream::connect("127.0.0.1:11101").is_ok() { + break; + } + thread::sleep(time::Duration::from_millis(500)); + } + info!("stratum server connected"); + + let mut workers = vec![]; + for _n in 0..5 { + let w = TcpStream::connect("127.0.0.1:11101").unwrap(); + w.set_nonblocking(true) + .expect("Failed to set TcpStream to non-blocking"); + workers.push(BufStream::new(w)); + } + assert_eq!(workers.len(), 5); + + // Simulate a worker disconnect. + workers.remove(4); + + // Swallow the genesis/job broadcast. + thread::sleep(time::Duration::from_secs(5)); + let mut response = String::new(); + for n in 0..workers.len() { + let _ = workers[n].read_line(&mut response); + } + + // getjobtemplate + let mut response = String::new(); + let job_req = "{\"id\": \"Stratum\", \"jsonrpc\": \"2.0\", \"method\": \"getjobtemplate\"}\n"; + workers[2].write_all(job_req.as_bytes()).unwrap(); + workers[2].flush().unwrap(); + thread::sleep(time::Duration::from_secs(1)); + match workers[2].read_line(&mut response) { + Ok(_) => { + let r: Value = serde_json::from_str(&response).unwrap(); + assert_eq!(r["error"], serde_json::Value::Null); + assert_ne!(r["result"], serde_json::Value::Null); + } + Err(_e) => panic!("getjobtemplate failed"), + } + + // keepalive + let mut response = String::new(); + let job_req = "{\"id\":\"3\",\"jsonrpc\":\"2.0\",\"method\":\"keepalive\"}\n"; + let ok_resp = "{\"id\":\"3\",\"jsonrpc\":\"2.0\",\"method\":\"keepalive\",\"result\":\"ok\",\"error\":null}\n"; + workers[2].write_all(job_req.as_bytes()).unwrap(); + workers[2].flush().unwrap(); + thread::sleep(time::Duration::from_secs(1)); + let _ = workers[2].read_line(&mut response); + assert_eq!(response.as_str(), ok_resp); + + // unknown method + let mut response = String::new(); + let job_req = "{\"id\":\"4\",\"jsonrpc\":\"2.0\",\"method\":\"doesnotexist\"}\n"; + let ok_resp = "{\"id\":\"4\",\"jsonrpc\":\"2.0\",\"method\":\"doesnotexist\",\"result\":null,\"error\":{\"code\":-32601,\"message\":\"Method not found\"}}\n"; + workers[3].write_all(job_req.as_bytes()).unwrap(); + workers[3].flush().unwrap(); + thread::sleep(time::Duration::from_secs(1)); + let _ = workers[3].read_line(&mut response); + assert_eq!(response.as_str(), ok_resp); + + let stats = s.get_server_stats().unwrap(); + assert_eq!(stats.stratum_stats.block_height, 1); + assert_eq!(stats.stratum_stats.num_workers, 4); + + let stop = Arc::new(StopState::new()); + s.start_test_miner(None, stop.clone()); + + workers.remove(1); + + // Wait for a few mined blocks / job broadcasts. + thread::sleep(time::Duration::from_secs(5)); + s.stop_test_miner(stop); + + let mut jobtemplate = String::new(); + let _ = workers[2].read_line(&mut jobtemplate); + if !jobtemplate.is_empty() { + let job_template: Value = serde_json::from_str(&jobtemplate).unwrap(); + assert_eq!(job_template["method"], "job"); + } + + let stats = s.get_server_stats().unwrap(); + assert_eq!(stats.stratum_stats.num_workers, 3); + assert_ne!(stats.stratum_stats.block_height, 1); + + s.stop(); + settle(); +}