|
| 1 | +// SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | +// Rust guideline compliant 2026-03-30 |
| 3 | +//! NFR-16 latency budget: a `ProxyJump` chain hop should add ≤ 1.5 s of |
| 4 | +//! cold-start cost per hop on a 50 ms RTT link. |
| 5 | +//! |
| 6 | +//! Loopback is too fast to honestly check the 1.5 s budget — every hop |
| 7 | +//! costs single-digit milliseconds on `127.0.0.1`. Instead, this |
| 8 | +//! bench enforces the *relative shape*: a 2-hop chain should run in |
| 9 | +//! at most **2× the time of a 1-hop chain**, on the same link. A |
| 10 | +//! genuine 50 ms RTT validation lives at `[bench] proxy_chain |
| 11 | +//! --features=real-rtt` (NOT IMPLEMENTED in this PR; documented as |
| 12 | +//! manual procedure in the M13 plan's verification section). |
| 13 | +//! |
| 14 | +//! # Running |
| 15 | +//! |
| 16 | +//! ```sh |
| 17 | +//! GITWAY_INTEGRATION_TESTS=1 cargo bench --bench proxy_chain |
| 18 | +//! ``` |
| 19 | +//! |
| 20 | +//! Without `GITWAY_INTEGRATION_TESTS=1`, the bench body is a no-op so |
| 21 | +//! it can appear in the bench harness list without spinning up |
| 22 | +//! `russh::server` instances on every CI run. |
| 23 | +
|
| 24 | +use std::net::SocketAddr; |
| 25 | +use std::sync::Arc; |
| 26 | +use std::time::{Duration, Instant}; |
| 27 | + |
| 28 | +use anvil_ssh::proxy::JumpHost; |
| 29 | +use anvil_ssh::{AnvilConfig, AnvilSession, StrictHostKeyChecking}; |
| 30 | +use criterion::{criterion_group, criterion_main, Criterion}; |
| 31 | +use russh::keys::ssh_key::rand_core::OsRng; |
| 32 | +use russh::keys::{Algorithm, PrivateKey}; |
| 33 | +use russh::server::{Auth, Msg, Server as _, Session}; |
| 34 | +use russh::{server, ChannelId}; |
| 35 | +use tokio::net::{TcpListener, TcpStream}; |
| 36 | + |
| 37 | +fn integration_enabled() -> bool { |
| 38 | + std::env::var("GITWAY_INTEGRATION_TESTS").is_ok_and(|v| !v.is_empty()) |
| 39 | +} |
| 40 | + |
| 41 | +// ── Server fixture (mirrors tests/test_proxy_jump.rs::TestServer) ──────────── |
| 42 | + |
| 43 | +#[derive(Clone)] |
| 44 | +struct BenchServer; |
| 45 | + |
| 46 | +impl server::Server for BenchServer { |
| 47 | + type Handler = BenchSession; |
| 48 | + fn new_client(&mut self, _: Option<SocketAddr>) -> Self::Handler { |
| 49 | + BenchSession |
| 50 | + } |
| 51 | + fn handle_session_error(&mut self, _: <Self::Handler as server::Handler>::Error) {} |
| 52 | +} |
| 53 | + |
| 54 | +struct BenchSession; |
| 55 | + |
| 56 | +impl server::Handler for BenchSession { |
| 57 | + type Error = russh::Error; |
| 58 | + |
| 59 | + async fn auth_password(&mut self, _: &str, _: &str) -> Result<Auth, Self::Error> { |
| 60 | + Ok(Auth::Accept) |
| 61 | + } |
| 62 | + async fn auth_publickey( |
| 63 | + &mut self, |
| 64 | + _: &str, |
| 65 | + _: &russh::keys::ssh_key::PublicKey, |
| 66 | + ) -> Result<Auth, Self::Error> { |
| 67 | + Ok(Auth::Accept) |
| 68 | + } |
| 69 | + async fn auth_publickey_offered( |
| 70 | + &mut self, |
| 71 | + _: &str, |
| 72 | + _: &russh::keys::ssh_key::PublicKey, |
| 73 | + ) -> Result<Auth, Self::Error> { |
| 74 | + Ok(Auth::Accept) |
| 75 | + } |
| 76 | + async fn channel_open_session( |
| 77 | + &mut self, |
| 78 | + _channel: russh::Channel<Msg>, |
| 79 | + _session: &mut Session, |
| 80 | + ) -> Result<bool, Self::Error> { |
| 81 | + Ok(true) |
| 82 | + } |
| 83 | + async fn channel_open_direct_tcpip( |
| 84 | + &mut self, |
| 85 | + channel: russh::Channel<Msg>, |
| 86 | + host_to_connect: &str, |
| 87 | + port_to_connect: u32, |
| 88 | + _: &str, |
| 89 | + _: u32, |
| 90 | + session: &mut Session, |
| 91 | + ) -> Result<bool, Self::Error> { |
| 92 | + let port_u16 = u16::try_from(port_to_connect).map_err(|_truncated| { |
| 93 | + russh::Error::from(std::io::Error::other(format!( |
| 94 | + "bench fixture: direct-tcpip port {port_to_connect} out of u16 range", |
| 95 | + ))) |
| 96 | + })?; |
| 97 | + let upstream = TcpStream::connect((host_to_connect, port_u16)) |
| 98 | + .await |
| 99 | + .map_err(russh::Error::from)?; |
| 100 | + let session_handle = session.handle(); |
| 101 | + let channel_id = channel.id(); |
| 102 | + tokio::spawn(async move { |
| 103 | + relay(channel, upstream, session_handle, channel_id).await; |
| 104 | + }); |
| 105 | + Ok(true) |
| 106 | + } |
| 107 | +} |
| 108 | + |
| 109 | +async fn relay( |
| 110 | + channel: russh::Channel<Msg>, |
| 111 | + tcp: TcpStream, |
| 112 | + session: server::Handle, |
| 113 | + channel_id: ChannelId, |
| 114 | +) { |
| 115 | + use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; |
| 116 | + |
| 117 | + let (mut tcp_read, mut tcp_write) = tcp.into_split(); |
| 118 | + let mut writer = channel.make_writer(); |
| 119 | + let (mut read_half, _wh) = channel.split(); |
| 120 | + |
| 121 | + let read_to_tcp = async { |
| 122 | + loop { |
| 123 | + let Some(msg) = read_half.wait().await else { |
| 124 | + break; |
| 125 | + }; |
| 126 | + if let russh::ChannelMsg::Data { data } = msg { |
| 127 | + if tcp_write.write_all(&data).await.is_err() { |
| 128 | + break; |
| 129 | + } |
| 130 | + } else if matches!(msg, russh::ChannelMsg::Eof | russh::ChannelMsg::Close) { |
| 131 | + break; |
| 132 | + } |
| 133 | + } |
| 134 | + let _ = session.eof(channel_id).await; |
| 135 | + }; |
| 136 | + |
| 137 | + let tcp_to_channel = async { |
| 138 | + let mut buf = vec![0_u8; 32 * 1024]; |
| 139 | + loop { |
| 140 | + let Ok(n) = tcp_read.read(&mut buf).await else { |
| 141 | + break; |
| 142 | + }; |
| 143 | + if n == 0 { |
| 144 | + break; |
| 145 | + } |
| 146 | + if writer.write_all(&buf[..n]).await.is_err() { |
| 147 | + break; |
| 148 | + } |
| 149 | + } |
| 150 | + }; |
| 151 | + |
| 152 | + tokio::join!(read_to_tcp, tcp_to_channel); |
| 153 | +} |
| 154 | + |
| 155 | +async fn spawn_server() -> u16 { |
| 156 | + let host_key = PrivateKey::random(&mut OsRng, Algorithm::Ed25519).expect("generate host key"); |
| 157 | + let config = Arc::new(server::Config { |
| 158 | + inactivity_timeout: Some(Duration::from_secs(30)), |
| 159 | + auth_rejection_time: Duration::from_millis(50), |
| 160 | + auth_rejection_time_initial: Some(Duration::from_millis(50)), |
| 161 | + keys: vec![host_key], |
| 162 | + ..Default::default() |
| 163 | + }); |
| 164 | + let listener = TcpListener::bind(("127.0.0.1", 0)).await.expect("bind"); |
| 165 | + let port = listener.local_addr().expect("local_addr").port(); |
| 166 | + // The `run_on_socket` future borrows both `server` and `listener` |
| 167 | + // for `'static`, so move them into the spawned task that owns them |
| 168 | + // for the bench's duration. The handle is dropped here; the |
| 169 | + // server stays alive until the bench's tokio Runtime is shut down. |
| 170 | + tokio::spawn(async move { |
| 171 | + let mut server = BenchServer; |
| 172 | + let fut = server.run_on_socket(config, &listener); |
| 173 | + let _ = fut.await; |
| 174 | + }); |
| 175 | + port |
| 176 | +} |
| 177 | + |
| 178 | +// ── Bench bodies ───────────────────────────────────────────────────────────── |
| 179 | + |
| 180 | +fn bench_one_hop_cold(c: &mut Criterion) { |
| 181 | + if !integration_enabled() { |
| 182 | + return; |
| 183 | + } |
| 184 | + let rt = tokio::runtime::Runtime::new().expect("tokio rt"); |
| 185 | + let target_port = rt.block_on(spawn_server()); |
| 186 | + |
| 187 | + let cfg = AnvilConfig::builder("127.0.0.1") |
| 188 | + .port(target_port) |
| 189 | + .strict_host_key_checking(StrictHostKeyChecking::No) |
| 190 | + .build(); |
| 191 | + |
| 192 | + c.bench_function("proxy_chain/1_hop_direct_cold", |b| { |
| 193 | + b.iter(|| { |
| 194 | + rt.block_on(async { |
| 195 | + let session = AnvilSession::connect(&cfg).await.expect("connect"); |
| 196 | + session.close().await.expect("close"); |
| 197 | + }); |
| 198 | + }); |
| 199 | + }); |
| 200 | +} |
| 201 | + |
| 202 | +fn bench_two_hop_cold(c: &mut Criterion) { |
| 203 | + if !integration_enabled() { |
| 204 | + return; |
| 205 | + } |
| 206 | + let rt = tokio::runtime::Runtime::new().expect("tokio rt"); |
| 207 | + let bastion_port = rt.block_on(spawn_server()); |
| 208 | + let target_port = rt.block_on(spawn_server()); |
| 209 | + |
| 210 | + let cfg = AnvilConfig::builder("127.0.0.1") |
| 211 | + .port(target_port) |
| 212 | + .strict_host_key_checking(StrictHostKeyChecking::No) |
| 213 | + .build(); |
| 214 | + |
| 215 | + let jumps = vec![JumpHost { |
| 216 | + host: "127.0.0.1".to_owned(), |
| 217 | + port: bastion_port, |
| 218 | + user: Some("user".to_owned()), |
| 219 | + identity_files: Vec::new(), |
| 220 | + }]; |
| 221 | + |
| 222 | + c.bench_function("proxy_chain/2_hop_via_bastion_cold", |b| { |
| 223 | + b.iter(|| { |
| 224 | + rt.block_on(async { |
| 225 | + let session = AnvilSession::connect_via_jump_hosts(&cfg, &jumps) |
| 226 | + .await |
| 227 | + .expect("connect_via_jump_hosts"); |
| 228 | + session.close().await.expect("close"); |
| 229 | + }); |
| 230 | + }); |
| 231 | + }); |
| 232 | +} |
| 233 | + |
| 234 | +/// Hard-fail enforcement: 2-hop median ≤ 2× 1-hop median on the same |
| 235 | +/// loopback link. Runs after the Criterion measurements so the panic |
| 236 | +/// message lands after the per-bench summaries. |
| 237 | +fn enforce_nfr16_ratio(c: &mut Criterion) { |
| 238 | + const RUNS: usize = 16; |
| 239 | + |
| 240 | + if !integration_enabled() { |
| 241 | + return; |
| 242 | + } |
| 243 | + let _ = c; // Criterion arg unused; we run a separate measurement loop. |
| 244 | + |
| 245 | + let rt = tokio::runtime::Runtime::new().expect("tokio rt"); |
| 246 | + let bastion_port = rt.block_on(spawn_server()); |
| 247 | + let target_port = rt.block_on(spawn_server()); |
| 248 | + |
| 249 | + let cfg = AnvilConfig::builder("127.0.0.1") |
| 250 | + .port(target_port) |
| 251 | + .strict_host_key_checking(StrictHostKeyChecking::No) |
| 252 | + .build(); |
| 253 | + |
| 254 | + let jumps = vec![JumpHost { |
| 255 | + host: "127.0.0.1".to_owned(), |
| 256 | + port: bastion_port, |
| 257 | + user: Some("user".to_owned()), |
| 258 | + identity_files: Vec::new(), |
| 259 | + }]; |
| 260 | + |
| 261 | + let mut one_hop = Vec::with_capacity(RUNS); |
| 262 | + let mut two_hop = Vec::with_capacity(RUNS); |
| 263 | + |
| 264 | + for _ in 0..RUNS { |
| 265 | + let start = Instant::now(); |
| 266 | + rt.block_on(async { |
| 267 | + let s = AnvilSession::connect(&cfg).await.expect("1-hop connect"); |
| 268 | + s.close().await.expect("close"); |
| 269 | + }); |
| 270 | + one_hop.push(start.elapsed()); |
| 271 | + |
| 272 | + let start = Instant::now(); |
| 273 | + rt.block_on(async { |
| 274 | + let s = AnvilSession::connect_via_jump_hosts(&cfg, &jumps) |
| 275 | + .await |
| 276 | + .expect("2-hop connect"); |
| 277 | + s.close().await.expect("close"); |
| 278 | + }); |
| 279 | + two_hop.push(start.elapsed()); |
| 280 | + } |
| 281 | + one_hop.sort(); |
| 282 | + two_hop.sort(); |
| 283 | + let one_hop_median = one_hop[RUNS / 2]; |
| 284 | + let two_hop_median = two_hop[RUNS / 2]; |
| 285 | + |
| 286 | + eprintln!( |
| 287 | + "proxy_chain/nfr16: 1-hop median = {} µs, 2-hop median = {} µs (ratio {:.2}×)", |
| 288 | + one_hop_median.as_micros(), |
| 289 | + two_hop_median.as_micros(), |
| 290 | + two_hop_median.as_secs_f64() / one_hop_median.as_secs_f64(), |
| 291 | + ); |
| 292 | + |
| 293 | + assert!( |
| 294 | + two_hop_median <= one_hop_median * 2, |
| 295 | + "NFR-16 (loopback proxy): 2-hop median {} µs > 2 × 1-hop median {} µs. \ |
| 296 | + Real-RTT validation against a 50 ms link is a separate manual step.", |
| 297 | + two_hop_median.as_micros(), |
| 298 | + one_hop_median.as_micros(), |
| 299 | + ); |
| 300 | +} |
| 301 | + |
| 302 | +criterion_group!( |
| 303 | + benches, |
| 304 | + bench_one_hop_cold, |
| 305 | + bench_two_hop_cold, |
| 306 | + enforce_nfr16_ratio, |
| 307 | +); |
| 308 | +criterion_main!(benches); |
0 commit comments