moonpool_sim/network/sim/
provider.rs

1use super::stream::{SimTcpListener, SimTcpStream};
2use crate::NetworkProvider;
3use crate::buggify;
4use crate::network::ConnectFailureMode;
5use crate::sim::rng::sim_random;
6use crate::{Event, WeakSimWorld};
7use std::io;
8use tracing::instrument;
9
10/// Simulated networking implementation
11#[derive(Debug, Clone)]
12pub struct SimNetworkProvider {
13    sim: WeakSimWorld,
14}
15
16impl SimNetworkProvider {
17    /// Create a new simulated network provider
18    #[must_use]
19    pub fn new(sim: WeakSimWorld) -> Self {
20        Self { sim }
21    }
22
23    /// Sleep in simulation time.
24    ///
25    /// This allows workloads to introduce delays for coordination purposes.
26    /// The sleep completes when the simulation processes the corresponding Wake event.
27    ///
28    /// # Errors
29    ///
30    /// Returns an error if the simulation cannot schedule the sleep event.
31    pub fn sleep(
32        &self,
33        duration: std::time::Duration,
34    ) -> crate::SimulationResult<crate::SleepFuture> {
35        self.sim.sleep(duration)
36    }
37}
38
39impl NetworkProvider for SimNetworkProvider {
40    type TcpStream = SimTcpStream;
41    type TcpListener = SimTcpListener;
42
43    #[instrument(skip(self))]
44    async fn bind(&self, addr: &str) -> io::Result<Self::TcpListener> {
45        let sim = self
46            .sim
47            .upgrade()
48            .map_err(|_| io::Error::other("simulation shutdown"))?;
49
50        // Get bind delay from network configuration and schedule bind completion event
51        let delay =
52            sim.with_network_config(|config| crate::network::sample_latency(&config.bind_latency));
53
54        // Schedule bind completion event to advance simulation time
55        let listener_id = sim.create_listener();
56
57        // Schedule an event to simulate the bind delay - this advances simulation time
58        sim.schedule_event(
59            Event::Connection {
60                id: listener_id.0,
61                state: crate::ConnectionStateChange::BindComplete,
62            },
63            delay,
64        );
65
66        let listener = SimTcpListener::new(self.sim.clone(), addr.to_string());
67        Ok(listener)
68    }
69
70    /// Connect to a remote address.
71    ///
72    /// When chaos is enabled, connection establishment can fail or hang forever
73    /// based on the `connect_failure_mode` setting (FDB ref: sim2.actor.cpp:1243-1250):
74    /// - Disabled: Normal operation (no failure injection)
75    /// - `AlwaysFail`: Always fail with `ConnectionRefused` when buggified
76    /// - Probabilistic: 50% fail with error, 50% hang forever (tests timeout handling)
77    #[instrument(skip(self))]
78    async fn connect(&self, addr: &str) -> io::Result<Self::TcpStream> {
79        let sim = self
80            .sim
81            .upgrade()
82            .map_err(|_| io::Error::other("simulation shutdown"))?;
83
84        // Check connect failure mode (FDB SIM_CONNECT_ERROR_MODE pattern)
85        // FDB ref: sim2.actor.cpp:1243-1250
86        let (failure_mode, failure_probability) = sim.with_network_config(|config| {
87            (
88                config.chaos.connect_failure_mode,
89                config.chaos.connect_failure_probability,
90            )
91        });
92
93        match failure_mode {
94            ConnectFailureMode::Disabled => {} // Normal operation
95            ConnectFailureMode::AlwaysFail => {
96                // Always fail with connection_failed when buggified
97                if buggify!() {
98                    tracing::debug!(addr = %addr, "Connection establishment failed (AlwaysFail mode)");
99                    return Err(io::Error::new(
100                        io::ErrorKind::ConnectionRefused,
101                        "Connection establishment failed (AlwaysFail mode)",
102                    ));
103                }
104            }
105            ConnectFailureMode::Probabilistic => {
106                // Probabilistic - fail or hang forever
107                if buggify!() {
108                    if sim_random::<f64>() > failure_probability {
109                        // Throw connection_failed error
110                        tracing::debug!(addr = %addr, "Connection establishment failed (Probabilistic mode - error)");
111                        return Err(io::Error::new(
112                            io::ErrorKind::ConnectionRefused,
113                            "Connection establishment failed (Probabilistic mode)",
114                        ));
115                    }
116                    // Hang forever - create a future that never completes
117                    // This tests timeout handling in connection retry logic
118                    tracing::debug!(addr = %addr, "Connection hanging forever (Probabilistic mode - hang)");
119                    std::future::pending::<()>().await;
120                    unreachable!("pending() never resolves");
121                }
122            }
123        }
124
125        // Get connect delay from network configuration and schedule connection event
126        let delay = sim
127            .with_network_config(|config| crate::network::sample_latency(&config.connect_latency));
128
129        // Create a connection pair for bidirectional communication
130        let (client_id, server_id) = sim.create_connection_pair("client-addr", addr);
131
132        // FDB SimClogging: fix a permanent per-pair latency at first contact, for
133        // both directions. No-op (returns ZERO, no RNG) when max_pair_latency is off.
134        let _ = sim.connection_base_latency(client_id);
135        let _ = sim.connection_base_latency(server_id);
136
137        // Store the server side for accept() to pick up later
138        sim.store_pending_connection(addr, server_id);
139
140        // Schedule connection ready event to advance simulation time
141        sim.schedule_event(
142            Event::Connection {
143                id: client_id.0,
144                state: crate::ConnectionStateChange::ConnectionReady,
145            },
146            delay,
147        );
148
149        let stream = SimTcpStream::new(self.sim.clone(), client_id);
150        Ok(stream)
151    }
152}
moonpool_sim/network/sim/provider.rs

moonpool_sim/network/sim/
provider.rs
