fix: Resolve race conditions and improve operation atomicity (#2233)

Co-authored-by: Ian Clarke <sanity@users.noreply.github.com>
Co-authored-by: Claude <noreply@anthropic.com>

Author: 3 people

apps/freenet-ping/Cargo.lock

@@ -4,9 +4,10 @@ members = ["contracts/ping", "app", "types"]
 
 [workspace.dependencies]
 # freenet-stdlib = { path = "./../../stdlib/rust", features = ["contract"] }
-freenet-stdlib = { version = "0.1.24" } 
+freenet-stdlib = { version = "0.1.24" }
 freenet-ping-types = { path = "types", default-features = false }
 chrono = { version = "0.4", default-features = false }
+clap = "4"
 testresult = "0.4"
 
 [profile.dev.package."*"]

apps/freenet-ping/Cargo.toml

@@ -10,7 +10,7 @@ manual-tests = []
 [dependencies]
 anyhow = "1.0"
 chrono = { workspace = true, features = ["default"] }
-clap = { version = "4.5", features = ["derive"] }
+clap = { workspace = true, features = ["derive"] }
 freenet-stdlib = { version = "0.1.24", features = ["net"] }
 freenet-ping-types = { path = "../types", features = ["std", "clap"] }
 futures = "0.3.31"

apps/freenet-ping/app/Cargo.toml

@@ -1,7 +1,7 @@
 use std::collections::HashMap;
 use std::time::{Duration, Instant};
 
-use chrono::{DateTime, Utc};
+use freenet_ping_types::chrono::{DateTime, Utc};
 use freenet_ping_types::{Ping, PingContractOptions};
 use freenet_stdlib::client_api::{
     ClientRequest, ContractRequest, ContractResponse, HostResponse, WebApi,

apps/freenet-ping/app/src/ping_client.rs

@@ -19,15 +19,17 @@ use futures::{future::BoxFuture, FutureExt};
 use rand::{random, Rng, SeedableRng};
 use std::io::{Read, Write};
 use std::process::{Child, Command, Stdio};
-use std::sync::{Arc, LazyLock};
+use std::sync::{Arc, LazyLock, Mutex};
 use std::{
     collections::HashSet,
     io,
     net::{Ipv4Addr, SocketAddr, TcpListener},
     path::{Path, PathBuf},
-    sync::Mutex,
     time::Duration,
 };
+
+/// Global lock to prevent concurrent contract compilation which causes race conditions
+static COMPILE_LOCK: LazyLock<Mutex<()>> = LazyLock::new(|| Mutex::new(()));
 use tokio::{select, time::sleep};
 use tokio_tungstenite::connect_async;
 use tracing::{info, span, Instrument, Level};
@@ -36,12 +38,6 @@ use serde::{Deserialize, Serialize};
 
 const TARGET_DIR_VAR: &str = "CARGO_TARGET_DIR";
 
-pub static RNG: LazyLock<Mutex<rand::rngs::StdRng>> = LazyLock::new(|| {
-    Mutex::new(rand::rngs::StdRng::from_seed(
-        *b"0102030405060708090a0b0c0d0e0f10",
-    ))
-});
-
 #[derive(Debug)]
 pub struct PresetConfig {
     pub temp_dir: tempfile::TempDir,
@@ -58,7 +54,6 @@ pub fn get_free_socket_addr() -> Result<SocketAddr> {
 }
 
 #[allow(clippy::too_many_arguments)]
-#[allow(clippy::await_holding_lock)]
 pub async fn base_node_test_config(
     is_gateway: bool,
     gateways: Vec<String>,
@@ -68,7 +63,13 @@ pub async fn base_node_test_config(
     base_tmp_dir: Option<&Path>,
     blocked_addresses: Option<Vec<SocketAddr>>,
 ) -> Result<(ConfigArgs, PresetConfig)> {
-    let mut rng = RNG.lock().unwrap();
+    // Create RNG seeded from test name for reproducibility while maintaining isolation
+    use std::hash::{Hash, Hasher};
+    let mut hasher = std::collections::hash_map::DefaultHasher::new();
+    data_dir_suffix.hash(&mut hasher);
+    let seed = hasher.finish();
+    let mut rng = rand::rngs::StdRng::seed_from_u64(seed);
+
     base_node_test_config_with_rng(
         is_gateway,
         gateways,
@@ -83,15 +84,15 @@ pub async fn base_node_test_config(
 }
 
 #[allow(clippy::too_many_arguments)]
-pub async fn base_node_test_config_with_rng(
+pub async fn base_node_test_config_with_rng<R: Rng>(
     is_gateway: bool,
     gateways: Vec<String>,
     public_port: Option<u16>,
     ws_api_port: u16,
     data_dir_suffix: &str,
     base_tmp_dir: Option<&Path>,
     blocked_addresses: Option<Vec<SocketAddr>>,
-    rng: &mut rand::rngs::StdRng,
+    rng: &mut R,
 ) -> Result<(ConfigArgs, PresetConfig)> {
     if is_gateway {
         assert!(public_port.is_some());
@@ -146,13 +147,19 @@ pub async fn base_node_test_config_with_rng(
 }
 
 pub fn gw_config_from_path(port: u16, path: &Path) -> Result<InlineGwConfig> {
-    gw_config_from_path_with_rng(port, path, &mut RNG.lock().unwrap())
+    // Create RNG seeded from path for reproducibility while maintaining isolation
+    use std::hash::{Hash, Hasher};
+    let mut hasher = std::collections::hash_map::DefaultHasher::new();
+    path.hash(&mut hasher);
+    let seed = hasher.finish();
+    let mut rng = rand::rngs::StdRng::seed_from_u64(seed);
+    gw_config_from_path_with_rng(port, path, &mut rng)
 }
 
-pub fn gw_config_from_path_with_rng(
+pub fn gw_config_from_path_with_rng<R: Rng>(
     port: u16,
     path: &Path,
-    rng: &mut rand::rngs::StdRng,
+    rng: &mut R,
 ) -> Result<InlineGwConfig> {
     Ok(InlineGwConfig {
         address: (Ipv4Addr::LOCALHOST, port).into(),
@@ -303,6 +310,9 @@ fn find_workspace_root() -> PathBuf {
 }
 
 fn compile_contract(contract_path: &PathBuf) -> anyhow::Result<Vec<u8>> {
+    // Acquire lock to prevent concurrent compilations which cause race conditions
+    let _lock = COMPILE_LOCK.lock().unwrap();
+
     ensure_target_dir_env();
     println!("module path: {contract_path:?}");
     let target = std::env::var(TARGET_DIR_VAR)

apps/freenet-ping/app/tests/common/mod.rs

@@ -133,6 +133,7 @@ async fn run_blocked_peers_test(config: BlockedPeersConfig) -> TestResult {
         let path = preset.temp_dir.path().to_path_buf();
         (cfg, preset, gw_config_from_path(public_port, &path)?)
     };
+
     let ws_api_port_gw = config_gw.ws_api.ws_api_port.unwrap();
 
     // Configure Node1 (blocks Node2)
@@ -167,7 +168,7 @@ async fn run_blocked_peers_test(config: BlockedPeersConfig) -> TestResult {
     tracing::info!("Node 1 blocks: {:?}", node2_network_addr);
     tracing::info!("Node 2 blocks: {:?}", node1_network_addr);
 
-    // Free socket resources
+    // Free socket resources before starting nodes
     std::mem::drop(network_socket_gw);
     std::mem::drop(ws_api_port_socket_gw);
     std::mem::drop(network_socket_node1);
@@ -788,17 +789,16 @@ async fn run_blocked_peers_test(config: BlockedPeersConfig) -> TestResult {
 
 /// Standard blocked peers test (baseline)
 #[test_log::test(tokio::test(flavor = "multi_thread"))]
-#[ignore]
 async fn test_ping_blocked_peers() -> TestResult {
     run_blocked_peers_test(BlockedPeersConfig {
         test_name: "baseline",
-        initial_wait: Duration::from_secs(10),
-        operation_timeout: Duration::from_secs(20),
+        initial_wait: Duration::from_secs(25),
+        operation_timeout: Duration::from_secs(45),
         update_rounds: 3,
         update_wait: Duration::from_secs(5),
-        propagation_wait: Duration::from_secs(8),
+        propagation_wait: Duration::from_secs(15),
         verbose_logging: false,
-        check_interval: None,
+        check_interval: Some(Duration::from_secs(4)),
         send_refresh_updates: false,
         send_final_updates: true,
         subscribe_immediately: false,
@@ -808,44 +808,39 @@ async fn test_ping_blocked_peers() -> TestResult {
 
 /// Simple blocked peers test
 #[test_log::test(tokio::test(flavor = "multi_thread"))]
-#[ignore]
 async fn test_ping_blocked_peers_simple() -> TestResult {
     run_blocked_peers_test(BlockedPeersConfig {
         test_name: "simple",
-        initial_wait: Duration::from_secs(10),
-        operation_timeout: Duration::from_secs(15),
-        update_rounds: 1, // Only one round of updates
-        update_wait: Duration::from_secs(3),
-        propagation_wait: Duration::from_secs(10), // Longer wait for simpler flow
+        initial_wait: Duration::from_secs(25),
+        operation_timeout: Duration::from_secs(45),
+        update_rounds: 1,
+        update_wait: Duration::from_secs(5),
+        propagation_wait: Duration::from_secs(15),
         verbose_logging: false,
-        check_interval: None,
+        check_interval: Some(Duration::from_secs(4)),
         send_refresh_updates: false,
         send_final_updates: false,
-        subscribe_immediately: true,
+        subscribe_immediately: false,
     })
     .await
 }
 
-// Note: Redundant tests (optimized, improved, debug, reliable) were removed
-// as they only varied in non-functional aspects like timeouts and logging
-
 /// Solution/reference implementation for blocked peers
 // TODO-MUST-FIX: WebSocket connection reset during teardown - see issue #2108
 // Test passes functionally (PUT/GET/Subscribe/state propagation all work) but
 // fails with "Connection reset without closing handshake" during cleanup.
 // Likely a test teardown race rather than functional bug.
 #[test_log::test(tokio::test(flavor = "multi_thread"))]
-#[ignore]
 async fn test_ping_blocked_peers_solution() -> TestResult {
     run_blocked_peers_test(BlockedPeersConfig {
         test_name: "solution",
-        initial_wait: Duration::from_secs(12),
-        operation_timeout: Duration::from_secs(25),
+        initial_wait: Duration::from_secs(25),
+        operation_timeout: Duration::from_secs(60),
         update_rounds: 2,
-        update_wait: Duration::from_secs(4),
-        propagation_wait: Duration::from_secs(12),
+        update_wait: Duration::from_secs(6),
+        propagation_wait: Duration::from_secs(20),
         verbose_logging: false,
-        check_interval: Some(Duration::from_secs(3)), // Regular check intervals
+        check_interval: Some(Duration::from_secs(5)),
         send_refresh_updates: true,
         send_final_updates: true,
         subscribe_immediately: true,

apps/freenet-ping/app/tests/run_app_blocked_peers.rs

@@ -17,6 +17,6 @@ humantime = "2"
 humantime-serde = "1"
 serde = { version = "1", features = ["derive"] }
 chrono = { workspace = true, features = ["serde"] }
-clap = { version = "4", features = ["derive"], optional = true }
+clap = { workspace = true, features = ["derive"], optional = true }
 
 freenet-stdlib = { workspace = true }

apps/freenet-ping/types/Cargo.toml

@@ -1284,15 +1284,13 @@ mod experimental_combined {
                             });
 
                             // Sender: serialize, encrypt, send to channel
-                            let mut nonce_counter = 0u64;
-                            for _ in 0..PACKET_COUNT {
+                            for nonce_counter in 0u64..PACKET_COUNT as u64 {
                                 // Allocate buffer for packet
                                 let mut packet = vec![0u8; size + 28]; // +28 for nonce+tag
 
                                 // Create nonce
                                 let mut nonce = [0u8; 12];
                                 nonce[4..].copy_from_slice(&nonce_counter.to_le_bytes());
-                                nonce_counter += 1;
 
                                 // Copy nonce to packet
                                 packet[..12].copy_from_slice(&nonce);
@@ -1493,24 +1491,19 @@ mod experimental_syscall_batching {
                     let rt = tokio::runtime::Handle::current();
                     let mut batch = Vec::with_capacity(BATCH_SIZE);
 
-                    loop {
-                        // Try to receive up to BATCH_SIZE packets
-                        match rt.block_on(rx.recv()) {
-                            Some(packet) => {
-                                batch.push(packet);
-                                // Drain available packets up to batch size
-                                while batch.len() < BATCH_SIZE {
-                                    match rx.try_recv() {
-                                        Ok(p) => batch.push(p),
-                                        Err(_) => break,
-                                    }
-                                }
-                                // Send batch
-                                for packet in batch.drain(..) {
-                                    socket.send(&packet).unwrap();
-                                }
+                    // Try to receive up to BATCH_SIZE packets
+                    while let Some(packet) = rt.block_on(rx.recv()) {
+                        batch.push(packet);
+                        // Drain available packets up to batch size
+                        while batch.len() < BATCH_SIZE {
+                            match rx.try_recv() {
+                                Ok(p) => batch.push(p),
+                                Err(_) => break,
                             }
-                            None => break,
+                        }
+                        // Send batch
+                        for packet in batch.drain(..) {
+                            socket.send(&packet).unwrap();
                         }
                     }
                 });