chore: initial sanitized public snapshot

crates/aria2-rust-pro-tests/benches/rpc_pressure/bt_visibility.rs

@@ -0,0 +1,202 @@
+#![expect(
+    clippy::redundant_pub_crate,
+    reason = "criterion bench entry points are re-exported only to the private bench root module"
+)]
+
+use super::support::{
+    BT_TORRENT_FIXTURE, BenchmarkId, Criterion, InProcessRpcDispatcher, RpcMethod, RpcValue,
+    RuntimeConfig, Throughput, TorrentPeerModel, TrackerPeerListModel, TrackerResponseModel,
+    rpc_request,
+};
+
+#[derive(Clone, Copy, Debug)]
+struct BtVisibilityPressureScenario {
+    task_count: usize,
+    rounds: usize,
+}
+
+fn seed_bt_visibility_dispatcher(
+    scenario: BtVisibilityPressureScenario,
+) -> (InProcessRpcDispatcher, Vec<String>) {
+    let runtime = RuntimeConfig {
+        allow_jsonrpc: true,
+        allow_xmlrpc: true,
+        split: 4,
+        max_connections_per_server: 4,
+        max_connection_per_server: 4,
+        min_split_size: 1024,
+        piece_length: 1024,
+        ..RuntimeConfig::default()
+    };
+    let mut dispatcher = InProcessRpcDispatcher::with_runtime(runtime);
+    let mut gids = Vec::with_capacity(scenario.task_count);
+    for index in 0..scenario.task_count {
+        let add_response = dispatcher.dispatch_json(rpc_request(
+            RpcMethod::Aria2AddTorrent,
+            vec![RpcValue::String(BT_TORRENT_FIXTURE.to_owned())],
+        ));
+        let gid = match add_response.result {
+            Some(RpcValue::String(gid)) => gid,
+            other => panic!("unexpected addTorrent result in visibility seed: {other:?}"),
+        };
+        dispatcher
+            .apply_tracker_announce_result(
+                &gid,
+                &TrackerResponseModel {
+                    peers: TrackerPeerListModel {
+                        interval_sec: 900,
+                        peers: vec![TorrentPeerModel {
+                            ip: format!("198.51.100.{}", (index % 200) + 1),
+                            port: 51413 + (index as u16 % 32),
+                            peer_id: None,
+                            client_name: None,
+                            interested: false,
+                            choked: false,
+                        }],
+                        min_interval_sec: None,
+                        tracker_id: Some(format!("bench-{index}")),
+                    },
+                    scrape: None,
+                },
+            )
+            .expect("tracker seed should populate visible bt peers");
+        dispatcher
+            .apply_bt_runtime_tick(
+                &gid,
+                8_192 + (index as u64 % 4) * 2_048,
+                if index % 3 == 0 { 4_096 } else { 0 },
+                512 + index as u64,
+                128 + index as u64,
+                if index % 3 == 0 { 5 } else { 0 },
+                if index % 3 == 0 { 5 } else { 0 },
+                index % 3 == 0,
+                Some(2 + (index % 6) as u32),
+            )
+            .expect("runtime tick should seed visible bt progress");
+        if index % 3 == 0 {
+            dispatcher
+                .set_bt_seeding_state(&gid, true, Some(1_000 + index as u64))
+                .expect("seeded torrents should enter seeding");
+            dispatcher
+                .tick_bt_runtime_clock(&gid, 1_010 + index as u64, true)
+                .expect("seeded torrents should advance share clocks");
+        }
+        gids.push(gid);
+    }
+    (dispatcher, gids)
+}
+
+fn run_bt_visibility_pressure(
+    dispatcher: &mut InProcessRpcDispatcher,
+    gids: &[String],
+    scenario: BtVisibilityPressureScenario,
+) -> usize {
+    let mut calls = 0_usize;
+    for round in 0..scenario.rounds {
+        for (index, gid) in gids.iter().enumerate() {
+            dispatcher
+                .apply_bt_runtime_tick(
+                    gid,
+                    if round % 2 == 0 { 512 } else { 0 },
+                    if round % 2 == 1 { 256 } else { 0 },
+                    1_024 + round as u64 + index as u64,
+                    256 + round as u64 + index as u64,
+                    u64::from(index % 3 == 0),
+                    u64::from(index % 3 == 0),
+                    index % 3 == 0,
+                    Some(2 + ((round + index) % 6) as u32),
+                )
+                .expect("pressure tick should keep bt runtime visible");
+
+            let tell_status = dispatcher.dispatch_json(rpc_request(
+                RpcMethod::Aria2TellStatus,
+                vec![RpcValue::String(gid.clone())],
+            ));
+            match tell_status.result {
+                Some(RpcValue::Object(payload)) => {
+                    assert_eq!(payload.get("isBt"), Some(&RpcValue::Bool(true)));
+                    assert!(payload.contains_key("announceList"));
+                    assert!(payload.contains_key("bitfield"));
+                    assert!(payload.contains_key("shareRatio"));
+                    assert!(payload.contains_key("shareTime"));
+                    assert!(payload.contains_key("files"));
+                    assert!(payload.contains_key("seeder"));
+                }
+                other => panic!("unexpected tellStatus payload in bt visibility bench: {other:?}"),
+            }
+            calls += 1;
+
+            let get_files = dispatcher.dispatch_json(rpc_request(
+                RpcMethod::Aria2GetFiles,
+                vec![RpcValue::String(gid.clone())],
+            ));
+            match get_files.result {
+                Some(RpcValue::Array(entries)) => match entries.first() {
+                    Some(RpcValue::Object(file)) => {
+                        assert!(file.contains_key("selected"));
+                        assert!(file.contains_key("completedLength"));
+                        assert!(file.contains_key("bitfield"));
+                    }
+                    other => {
+                        panic!("unexpected getFiles payload in bt visibility bench: {other:?}")
+                    }
+                },
+                other => panic!("unexpected getFiles result in bt visibility bench: {other:?}"),
+            }
+            calls += 1;
+
+            let get_peers = dispatcher.dispatch_json(rpc_request(
+                RpcMethod::Aria2GetPeers,
+                vec![RpcValue::String(gid.clone())],
+            ));
+            match get_peers.result {
+                Some(RpcValue::Array(entries)) => match entries.first() {
+                    Some(RpcValue::Object(peer)) => {
+                        assert!(peer.contains_key("ip"));
+                        assert!(peer.contains_key("port"));
+                        assert!(peer.contains_key("peerChoking"));
+                    }
+                    other => {
+                        panic!("unexpected getPeers payload in bt visibility bench: {other:?}")
+                    }
+                },
+                other => panic!("unexpected getPeers result in bt visibility bench: {other:?}"),
+            }
+            calls += 1;
+        }
+    }
+    calls
+}
+
+pub(super) fn bench_bt_visibility_pressure(c: &mut Criterion) {
+    let mut group = c.benchmark_group("bt_visibility_pressure");
+    for scenario in [
+        BtVisibilityPressureScenario {
+            task_count: 32,
+            rounds: 4,
+        },
+        BtVisibilityPressureScenario {
+            task_count: 64,
+            rounds: 4,
+        },
+    ] {
+        group.throughput(Throughput::Elements(
+            (scenario.task_count * scenario.rounds * 3) as u64,
+        ));
+        group.bench_with_input(
+            BenchmarkId::new("bt_visibility", scenario.task_count),
+            &scenario,
+            |b, &scenario| {
+                b.iter_batched(
+                    || seed_bt_visibility_dispatcher(scenario),
+                    |(mut dispatcher, gids)| {
+                        let calls = run_bt_visibility_pressure(&mut dispatcher, &gids, scenario);
+                        assert_eq!(calls, scenario.task_count * scenario.rounds * 3);
+                    },
+                    criterion::BatchSize::SmallInput,
+                );
+            },
+        );
+    }
+    group.finish();
+}

crates/aria2-rust-pro-tests/benches/rpc_pressure/live_http_transfer.rs

@@ -0,0 +1,602 @@
+#![expect(
+    clippy::redundant_pub_crate,
+    reason = "criterion bench entry points are re-exported only to the private bench root module"
+)]
+
+use super::support::{
+    Arc, AtomicBool, BTreeMap, BenchmarkId, ConnectorBackedDownloader, Criterion, Duration,
+    Instant, Invocation, Mutex, Ordering, PathBuf, Read, ReqwestHttpConnector, SocketAddr,
+    SystemTime, TcpListener, TcpStream, Throughput, UNIX_EPOCH, Write,
+    execute_runtime_with_downloader, fs, thread,
+};
+
+const LIVE_TRANSFER_SPLIT: usize = 4;
+const LIVE_TRANSFER_MAX_CONNECTIONS_PER_SERVER: usize = 4;
+
+#[derive(Clone, Copy, Debug)]
+struct LiveTransferScenario {
+    label: &'static str,
+    total_length: usize,
+    piece_length: usize,
+    overall_download_limit: Option<usize>,
+    disk_cache_bytes: Option<usize>,
+    response_delay_ms: u64,
+}
+
+#[derive(Clone, Copy, Debug)]
+struct LiveTransferContentionScenario {
+    label: &'static str,
+    download_count: usize,
+    total_length: usize,
+    piece_length: usize,
+    overall_download_limit: Option<usize>,
+    disk_cache_bytes: Option<usize>,
+    response_delay_ms: u64,
+}
+
+#[derive(Clone, Debug, Default)]
+struct SegmentRequestMetrics {
+    total_requests: usize,
+    requests_by_path: BTreeMap<String, usize>,
+}
+
+struct TempConfigFile {
+    path: PathBuf,
+}
+
+impl TempConfigFile {
+    fn new(contents: &str) -> Self {
+        let unique = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .expect("system time should be after unix epoch")
+            .as_nanos();
+        let path = std::env::temp_dir().join(format!("aria2-rust-pro-{unique}.conf"));
+        fs::write(&path, contents).expect("benchmark config should write");
+        Self { path }
+    }
+
+    const fn path(&self) -> &PathBuf {
+        &self.path
+    }
+}
+
+impl Drop for TempConfigFile {
+    fn drop(&mut self) {
+        let _ = fs::remove_file(&self.path);
+    }
+}
+
+struct LocalHttpSegmentServer {
+    base_url: String,
+    metrics: Arc<Mutex<SegmentRequestMetrics>>,
+    stop: Arc<AtomicBool>,
+    handle: Option<thread::JoinHandle<()>>,
+    listen_address: SocketAddr,
+}
+
+impl LocalHttpSegmentServer {
+    fn spawn(scenario: LiveTransferScenario) -> Self {
+        let expected_requests_per_download = scenario.total_length.div_ceil(scenario.piece_length);
+        Self::spawn_many(scenario, 1, expected_requests_per_download)
+    }
+
+    fn spawn_many(
+        scenario: LiveTransferScenario,
+        _download_count: usize,
+        _expected_requests_per_download: usize,
+    ) -> Self {
+        let listener = TcpListener::bind("127.0.0.1:0").expect("loopback listener should bind");
+        let addr = listener
+            .local_addr()
+            .expect("loopback listener should report local addr");
+        let piece_length = scenario.piece_length;
+        let total_length = scenario.total_length;
+        let response_delay = Duration::from_millis(scenario.response_delay_ms);
+        let payload_bytes = Arc::<[u8]>::from(vec![b'x'; total_length]);
+        let metrics = Arc::new(Mutex::new(SegmentRequestMetrics::default()));
+        let metrics_for_thread = Arc::clone(&metrics);
+        let stop = Arc::new(AtomicBool::new(false));
+        let stop_for_thread = Arc::clone(&stop);
+        let handle = thread::spawn(move || {
+            loop {
+                let (stream, _) = listener
+                    .accept()
+                    .unwrap_or_else(|error| panic!("bench client should connect: {error}"));
+                if stop_for_thread.load(Ordering::Relaxed) {
+                    break;
+                }
+                let payload_bytes = Arc::clone(&payload_bytes);
+                let metrics = Arc::clone(&metrics_for_thread);
+                thread::spawn(move || {
+                    handle_loopback_segment_request(
+                        stream,
+                        &payload_bytes,
+                        &metrics,
+                        piece_length,
+                        total_length,
+                        response_delay,
+                    );
+                });
+            }
+        });
+        Self {
+            base_url: format!("http://{addr}"),
+            metrics,
+            stop,
+            handle: Some(handle),
+            listen_address: addr,
+        }
+    }
+
+    fn url_for(&self, path: &str) -> String {
+        format!("{}/{}", self.base_url, path.trim_start_matches('/'))
+    }
+
+    fn snapshot_metrics(&self) -> SegmentRequestMetrics {
+        self.metrics
+            .lock()
+            .expect("segment request metrics mutex should not poison")
+            .clone()
+    }
+}
+
+impl Drop for LocalHttpSegmentServer {
+    fn drop(&mut self) {
+        self.stop.store(true, Ordering::Relaxed);
+        let _ = TcpStream::connect(self.listen_address);
+        if let Some(handle) = self.handle.take() {
+            handle.join().expect("loopback server thread should join");
+        }
+    }
+}
+
+fn handle_loopback_segment_request(
+    mut stream: TcpStream,
+    payload_bytes: &Arc<[u8]>,
+    metrics: &Arc<Mutex<SegmentRequestMetrics>>,
+    piece_length: usize,
+    total_length: usize,
+    response_delay: Duration,
+) {
+    stream
+        .set_nonblocking(false)
+        .expect("accepted bench socket should switch back to blocking mode");
+    let mut request = [0_u8; 4096];
+    let read = stream.read(&mut request).expect("request should read");
+    let request_text = String::from_utf8_lossy(&request[..read]);
+    let request_path = parse_request_path(&request_text).to_owned();
+    let (start, end_inclusive) = parse_requested_range(&request_text, piece_length, total_length);
+    let len = end_inclusive.saturating_sub(start).saturating_add(1);
+    let body = payload_bytes.get(start..=end_inclusive).unwrap_or(&[]);
+    {
+        let mut metrics = metrics
+            .lock()
+            .expect("segment request metrics mutex should not poison");
+        metrics.total_requests += 1;
+        *metrics.requests_by_path.entry(request_path).or_default() += 1;
+    }
+    thread::sleep(response_delay);
+    let response_head = format!(
+        "HTTP/1.1 206 Partial Content\r\nContent-Length: {len}\r\nContent-Range: bytes {start}-{end_inclusive}/{total_length}\r\nConnection: close\r\n\r\n"
+    );
+    stream
+        .write_all(response_head.as_bytes())
+        .expect("response head should write");
+    stream.write_all(body).expect("response body should write");
+}
+
+fn parse_request_path(request_text: &str) -> &str {
+    request_text
+        .lines()
+        .next()
+        .and_then(|line| line.split_whitespace().nth(1))
+        .unwrap_or("/payload.bin")
+}
+
+fn parse_requested_range(
+    request_text: &str,
+    piece_length: usize,
+    total_length: usize,
+) -> (usize, usize) {
+    let requested = request_text
+        .lines()
+        .find_map(|line| {
+            let (name, value) = line.split_once(':')?;
+            if !name.trim().eq_ignore_ascii_case("range") {
+                return None;
+            }
+            value.trim().strip_prefix("bytes=")
+        })
+        .and_then(|value| value.split_once('-'))
+        .map(|(start, end)| {
+            let start = start.parse::<usize>().expect("range start should parse");
+            let end = end.parse::<usize>().expect("range end should parse");
+            (start, end.min(total_length.saturating_sub(1)))
+        });
+
+    requested.unwrap_or_else(|| {
+        let start = 0usize;
+        let end = piece_length
+            .saturating_sub(1)
+            .min(total_length.saturating_sub(1));
+        (start, end)
+    })
+}
+
+fn build_live_transfer_config(scenario: LiveTransferScenario) -> TempConfigFile {
+    let mut lines = vec![
+        format!("split={LIVE_TRANSFER_SPLIT}"),
+        format!("max-connection-per-server={LIVE_TRANSFER_MAX_CONNECTIONS_PER_SERVER}"),
+        format!("min-split-size={}", scenario.piece_length),
+        format!("piece-length={}", scenario.piece_length),
+    ];
+    if let Some(limit) = scenario.overall_download_limit {
+        lines.push(format!("max-overall-download-limit={limit}"));
+    }
+    if let Some(disk_cache_bytes) = scenario.disk_cache_bytes {
+        lines.push(format!("disk-cache={disk_cache_bytes}"));
+    }
+    TempConfigFile::new(&lines.join("\n"))
+}
+
+fn min_expected_live_requests_per_download(scenario: LiveTransferContentionScenario) -> usize {
+    let total_segments = scenario.total_length.div_ceil(scenario.piece_length);
+    if LIVE_TRANSFER_MAX_CONNECTIONS_PER_SERVER >= 4 {
+        let probe_segments = if scenario.piece_length <= 64 * 1024 {
+            LIVE_TRANSFER_MAX_CONNECTIONS_PER_SERVER.saturating_mul(2)
+        } else {
+            2usize
+        };
+        if total_segments <= probe_segments {
+            return 1;
+        }
+
+        let remaining_segments = total_segments.saturating_sub(probe_segments);
+        let followup_floor = if remaining_segments <= 2 {
+            1
+        } else if remaining_segments <= 3 {
+            2
+        } else {
+            remaining_segments.min(LIVE_TRANSFER_SPLIT.saturating_sub(1))
+        };
+        return 1usize.saturating_add(followup_floor);
+    }
+
+    let mut segmented_floor = 1usize.saturating_add(
+        total_segments
+            .saturating_sub(1)
+            .min(LIVE_TRANSFER_SPLIT.saturating_sub(1)),
+    );
+    if segmented_floor > 3 && total_segments <= LIVE_TRANSFER_SPLIT {
+        segmented_floor = segmented_floor.saturating_sub(1);
+    }
+    total_segments.min(segmented_floor).max(1)
+}
+
+fn request_budget_per_download(scenario: LiveTransferContentionScenario) -> usize {
+    min_expected_live_requests_per_download(scenario).saturating_mul(4)
+}
+
+fn run_live_transfer_contention_scenario(scenario: LiveTransferScenario) -> usize {
+    let config = build_live_transfer_config(scenario);
+    let server = LocalHttpSegmentServer::spawn(scenario);
+    let connector = ReqwestHttpConnector::new().expect("reqwest connector should build");
+    let downloader = ConnectorBackedDownloader::new(connector.clone(), connector);
+    let report = execute_runtime_with_downloader(
+        Invocation::Run {
+            config_path: Some(config.path().clone()),
+            uris: vec![server.url_for("payload.bin")],
+        },
+        &downloader,
+    )
+    .expect("live transfer benchmark should complete");
+    assert_eq!(report.completed_download_count, 1);
+    assert_eq!(report.first_status.as_deref(), Some("complete"));
+    assert_eq!(
+        report.first_completed_length,
+        Some(scenario.total_length as u64)
+    );
+    scenario.total_length
+}
+
+fn run_live_transfer_fairness_contention_scenario(
+    scenario: LiveTransferContentionScenario,
+) -> usize {
+    let single_transfer = LiveTransferScenario {
+        label: scenario.label,
+        total_length: scenario.total_length,
+        piece_length: scenario.piece_length,
+        overall_download_limit: scenario.overall_download_limit,
+        disk_cache_bytes: scenario.disk_cache_bytes,
+        response_delay_ms: scenario.response_delay_ms,
+    };
+    let min_requests_per_download = min_expected_live_requests_per_download(scenario);
+    let request_budget_per_download = request_budget_per_download(scenario);
+    let server = LocalHttpSegmentServer::spawn_many(
+        single_transfer,
+        scenario.download_count,
+        request_budget_per_download,
+    );
+    let worker_handles = (0..scenario.download_count)
+        .map(|_| build_live_transfer_config(single_transfer))
+        .enumerate()
+        .map(|(index, config)| {
+            let config_path = config.path().clone();
+            let uri = server.url_for(&format!("payload-{index}.bin"));
+            thread::spawn(move || {
+                let _config_guard = config;
+                let connector =
+                    ReqwestHttpConnector::new().expect("reqwest connector should build");
+                let downloader = ConnectorBackedDownloader::new(connector.clone(), connector);
+                let started = Instant::now();
+                let report = execute_runtime_with_downloader(
+                    Invocation::Run {
+                        config_path: Some(config_path),
+                        uris: vec![uri],
+                    },
+                    &downloader,
+                )
+                .expect("parallel live transfer benchmark should complete");
+                (started.elapsed(), report)
+            })
+        })
+        .collect::<Vec<_>>();
+
+    let mut elapsed = Vec::with_capacity(scenario.download_count);
+    for worker in worker_handles {
+        let (duration, report) = worker
+            .join()
+            .expect("parallel live transfer worker thread should join");
+        assert_eq!(report.completed_download_count, 1);
+        assert_eq!(report.first_status.as_deref(), Some("complete"));
+        assert_eq!(
+            report.first_completed_length,
+            Some(scenario.total_length as u64)
+        );
+        elapsed.push(duration);
+    }
+
+    let metrics = server.snapshot_metrics();
+    assert_eq!(metrics.requests_by_path.len(), scenario.download_count);
+    assert!(
+        metrics.total_requests >= min_requests_per_download * scenario.download_count,
+        "parallel live transfer should keep every download progressing through segmented requests: observed total_requests={} floor_per_download={} download_count={}",
+        metrics.total_requests,
+        min_requests_per_download,
+        scenario.download_count
+    );
+    for index in 0..scenario.download_count {
+        let path = format!("/payload-{index}.bin");
+        let observed = metrics
+            .requests_by_path
+            .get(&path)
+            .copied()
+            .unwrap_or_default();
+        assert!(
+            observed >= min_requests_per_download,
+            "each download should exercise the segmented path under contention: path={path} observed={observed} floor={min_requests_per_download}"
+        );
+    }
+
+    let min_elapsed = elapsed
+        .iter()
+        .min()
+        .copied()
+        .expect("at least one live contention run should exist");
+    let max_elapsed = elapsed
+        .iter()
+        .max()
+        .copied()
+        .expect("at least one live contention run should exist");
+    assert!(max_elapsed >= min_elapsed);
+    scenario.total_length * scenario.download_count
+}
+
+fn run_live_shared_runtime_multi_download_scenario(
+    scenario: LiveTransferContentionScenario,
+) -> usize {
+    let single_transfer = LiveTransferScenario {
+        label: scenario.label,
+        total_length: scenario.total_length,
+        piece_length: scenario.piece_length,
+        overall_download_limit: scenario.overall_download_limit,
+        disk_cache_bytes: scenario.disk_cache_bytes,
+        response_delay_ms: scenario.response_delay_ms,
+    };
+    let config = build_live_transfer_config(single_transfer);
+    let min_requests_per_download = min_expected_live_requests_per_download(scenario);
+    let request_budget_per_download = request_budget_per_download(scenario);
+    let server = LocalHttpSegmentServer::spawn_many(
+        single_transfer,
+        scenario.download_count,
+        request_budget_per_download,
+    );
+    let connector = ReqwestHttpConnector::new().expect("reqwest connector should build");
+    let downloader = ConnectorBackedDownloader::new(connector.clone(), connector);
+    let report = execute_runtime_with_downloader(
+        Invocation::Run {
+            config_path: Some(config.path().clone()),
+            uris: (0..scenario.download_count)
+                .map(|index| server.url_for(&format!("shared-runtime-{index}.bin")))
+                .collect(),
+        },
+        &downloader,
+    )
+    .expect("shared-runtime live transfer benchmark should complete");
+
+    assert_eq!(report.accepted_uri_count, scenario.download_count);
+    assert_eq!(report.tracked_download_count, scenario.download_count);
+    assert_eq!(report.completed_download_count, scenario.download_count);
+    assert_eq!(report.first_status.as_deref(), Some("complete"));
+    assert_eq!(
+        report.first_completed_length,
+        Some(scenario.total_length as u64)
+    );
+
+    let metrics = server.snapshot_metrics();
+    assert_eq!(metrics.requests_by_path.len(), scenario.download_count);
+    assert!(
+        metrics.total_requests >= min_requests_per_download * scenario.download_count,
+        "shared-runtime live transfer should keep every registered download progressing: observed total_requests={} floor_per_download={} download_count={}",
+        metrics.total_requests,
+        min_requests_per_download,
+        scenario.download_count
+    );
+    for index in 0..scenario.download_count {
+        let path = format!("/shared-runtime-{index}.bin");
+        let observed = metrics
+            .requests_by_path
+            .get(&path)
+            .copied()
+            .unwrap_or_default();
+        assert!(
+            observed >= min_requests_per_download,
+            "shared-runtime live transfer should fully progress each registered download: path={path} observed={observed} floor={min_requests_per_download}"
+        );
+    }
+
+    scenario.total_length * scenario.download_count
+}
+
+pub(super) fn bench_live_http_transfer_contention_pressure(c: &mut Criterion) {
+    let mut group = c.benchmark_group("live_http_transfer_contention_pressure");
+    for scenario in [
+        LiveTransferScenario {
+            label: "loose_cap",
+            total_length: 16 * 1024,
+            piece_length: 4 * 1024,
+            overall_download_limit: None,
+            disk_cache_bytes: None,
+            response_delay_ms: 8,
+        },
+        LiveTransferScenario {
+            label: "tight_cap",
+            total_length: 16 * 1024,
+            piece_length: 4 * 1024,
+            overall_download_limit: Some(4 * 1024),
+            disk_cache_bytes: None,
+            response_delay_ms: 8,
+        },
+    ] {
+        group.throughput(Throughput::Bytes(scenario.total_length as u64));
+        group.bench_with_input(
+            BenchmarkId::new("live_http_transfer", scenario.label),
+            &scenario,
+            |b, &scenario| {
+                b.iter_batched(
+                    || scenario,
+                    |scenario| {
+                        let bytes = run_live_transfer_contention_scenario(scenario);
+                        assert_eq!(bytes, scenario.total_length);
+                    },
+                    criterion::BatchSize::SmallInput,
+                );
+            },
+        );
+    }
+    group.finish();
+}
+
+pub(super) fn bench_live_http_multi_download_contention_pressure(c: &mut Criterion) {
+    let mut group = c.benchmark_group("live_http_multi_download_contention_pressure");
+    for scenario in [
+        LiveTransferContentionScenario {
+            label: "loose_cap",
+            download_count: 3,
+            total_length: 16 * 1024,
+            piece_length: 4 * 1024,
+            overall_download_limit: None,
+            disk_cache_bytes: None,
+            response_delay_ms: 8,
+        },
+        LiveTransferContentionScenario {
+            label: "tight_cap",
+            download_count: 3,
+            total_length: 16 * 1024,
+            piece_length: 4 * 1024,
+            overall_download_limit: Some(4 * 1024),
+            disk_cache_bytes: None,
+            response_delay_ms: 8,
+        },
+    ] {
+        group.throughput(Throughput::Bytes(
+            (scenario.total_length * scenario.download_count) as u64,
+        ));
+        group.bench_with_input(
+            BenchmarkId::new("multi_live_http_transfer", scenario.label),
+            &scenario,
+            |b, &scenario| {
+                b.iter_batched(
+                    || scenario,
+                    |scenario| {
+                        let bytes = run_live_transfer_fairness_contention_scenario(scenario);
+                        assert_eq!(bytes, scenario.total_length * scenario.download_count);
+                    },
+                    criterion::BatchSize::SmallInput,
+                );
+            },
+        );
+    }
+    group.finish();
+}
+
+pub(super) fn bench_live_http_shared_runtime_multi_download_pressure(c: &mut Criterion) {
+    let mut group = c.benchmark_group("live_http_shared_runtime_multi_download_pressure");
+    for scenario in [
+        LiveTransferContentionScenario {
+            label: "loose_cap",
+            download_count: 3,
+            total_length: 16 * 1024,
+            piece_length: 4 * 1024,
+            overall_download_limit: None,
+            disk_cache_bytes: None,
+            response_delay_ms: 8,
+        },
+        LiveTransferContentionScenario {
+            label: "tight_cap",
+            download_count: 3,
+            total_length: 16 * 1024,
+            piece_length: 4 * 1024,
+            overall_download_limit: Some(4 * 1024),
+            disk_cache_bytes: None,
+            response_delay_ms: 8,
+        },
+        LiveTransferContentionScenario {
+            label: "tight_cap_6way",
+            download_count: 6,
+            total_length: 16 * 1024,
+            piece_length: 4 * 1024,
+            overall_download_limit: Some(4 * 1024),
+            disk_cache_bytes: None,
+            response_delay_ms: 8,
+        },
+        LiveTransferContentionScenario {
+            label: "cache_pressure_6way_256k",
+            download_count: 6,
+            total_length: 256 * 1024,
+            piece_length: 32 * 1024,
+            overall_download_limit: Some(64 * 1024),
+            disk_cache_bytes: Some(64 * 1024),
+            response_delay_ms: 2,
+        },
+    ] {
+        group.throughput(Throughput::Bytes(
+            (scenario.total_length * scenario.download_count) as u64,
+        ));
+        group.bench_with_input(
+            BenchmarkId::new("shared_runtime_live_http_transfer", scenario.label),
+            &scenario,
+            |b, &scenario| {
+                b.iter_batched(
+                    || scenario,
+                    |scenario| {
+                        let bytes = run_live_shared_runtime_multi_download_scenario(scenario);
+                        assert_eq!(bytes, scenario.total_length * scenario.download_count);
+                    },
+                    criterion::BatchSize::SmallInput,
+                );
+            },
+        );
+    }
+    group.finish();
+}

crates/aria2-rust-pro-tests/benches/rpc_pressure/rpc_runtime_pressure.rs

@@ -0,0 +1,667 @@
+#![expect(
+    clippy::redundant_pub_crate,
+    reason = "criterion bench entry points are re-exported only to the private bench root module"
+)]
+
+use super::support::{
+    BTreeMap, BenchmarkId, Criterion, InProcessRpcDispatcher, RpcMethod, RpcValue, RuntimeConfig,
+    Throughput, rpc_request,
+};
+
+#[derive(Clone, Copy, Debug)]
+struct PressureScenario {
+    task_count: usize,
+    rounds: usize,
+    split: usize,
+    max_connections_per_server: usize,
+}
+
+impl PressureScenario {
+    fn runtime(self) -> RuntimeConfig {
+        RuntimeConfig {
+            split: self.split,
+            max_connections_per_server: self.max_connections_per_server,
+            max_connection_per_server: self.max_connections_per_server,
+            min_split_size: 1024,
+            piece_length: 1024,
+            ..RuntimeConfig::default()
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug)]
+struct ResourceLimitScenario {
+    task_count: usize,
+    rounds: usize,
+    global_download_limit: u64,
+    per_download_limit: u64,
+    global_upload_limit: u64,
+    per_upload_limit: u64,
+    disk_cache_bytes: u64,
+}
+
+#[derive(Clone, Copy, Debug)]
+struct SharedRuntimeFairnessScenario {
+    label: &'static str,
+    task_count: usize,
+    rounds: usize,
+    global_download_limit: u64,
+    default_per_download_limit: u64,
+    constrained_per_download_limit: u64,
+    global_upload_limit: u64,
+    default_per_upload_limit: u64,
+    constrained_per_upload_limit: u64,
+}
+
+fn seed_pressure_dispatcher(scenario: PressureScenario) -> (InProcessRpcDispatcher, Vec<String>) {
+    let mut dispatcher = InProcessRpcDispatcher::with_runtime(scenario.runtime());
+    let mut gids = Vec::with_capacity(scenario.task_count);
+    for i in 0..scenario.task_count {
+        let magnet = format!(
+            "magnet:?xt=urn:btih:{:040x}&dn=bench-{i}&tr=http%3A%2F%2Ftracker.example.org%2Fannounce",
+            i + 40_001
+        );
+        let response = dispatcher.dispatch_json(rpc_request(
+            RpcMethod::Aria2AddUri,
+            vec![RpcValue::String(magnet)],
+        ));
+        match response.result {
+            Some(RpcValue::String(gid)) => gids.push(gid),
+            other => panic!("unexpected addUri result in bench setup: {other:?}"),
+        }
+    }
+
+    for (index, gid) in gids.iter().enumerate() {
+        dispatcher
+            .apply_bt_runtime_tick(
+                gid,
+                1024 + (index % 4) as u64 * 256,
+                256 + (index % 3) as u64 * 64,
+                300 + index as u64 % 80,
+                120 + index as u64 % 40,
+                1,
+                1,
+                false,
+                Some(4 + (index % 4) as u32),
+            )
+            .expect("bench setup runtime tick should succeed");
+    }
+
+    (dispatcher, gids)
+}
+
+fn seed_limited_dispatcher(
+    scenario: ResourceLimitScenario,
+) -> (InProcessRpcDispatcher, Vec<String>) {
+    let base = PressureScenario {
+        task_count: scenario.task_count,
+        rounds: scenario.rounds,
+        split: 6,
+        max_connections_per_server: 6,
+    };
+    let (mut dispatcher, gids) = seed_pressure_dispatcher(base);
+    let change_global = dispatcher.dispatch_json(rpc_request(
+        RpcMethod::Aria2ChangeGlobalOption,
+        vec![RpcValue::Object(BTreeMap::from([
+            (
+                "max-overall-download-limit".to_owned(),
+                RpcValue::String(scenario.global_download_limit.to_string()),
+            ),
+            (
+                "max-overall-upload-limit".to_owned(),
+                RpcValue::String(scenario.global_upload_limit.to_string()),
+            ),
+            (
+                "disk-cache".to_owned(),
+                RpcValue::String(scenario.disk_cache_bytes.to_string()),
+            ),
+        ]))],
+    ));
+    assert!(
+        change_global.error.is_none(),
+        "changeGlobalOption should succeed in benchmark setup"
+    );
+
+    for gid in &gids {
+        let change_option = dispatcher.dispatch_json(rpc_request(
+            RpcMethod::Aria2ChangeOption,
+            vec![
+                RpcValue::String(gid.clone()),
+                RpcValue::Object(BTreeMap::from([
+                    (
+                        "max-download-limit".to_owned(),
+                        RpcValue::String(scenario.per_download_limit.to_string()),
+                    ),
+                    (
+                        "max-upload-limit".to_owned(),
+                        RpcValue::String(scenario.per_upload_limit.to_string()),
+                    ),
+                ])),
+            ],
+        ));
+        assert!(
+            change_option.error.is_none(),
+            "changeOption should succeed in benchmark setup"
+        );
+    }
+
+    (dispatcher, gids)
+}
+
+fn seed_shared_runtime_fairness_dispatcher(
+    scenario: SharedRuntimeFairnessScenario,
+) -> (InProcessRpcDispatcher, Vec<String>) {
+    let base = PressureScenario {
+        task_count: scenario.task_count,
+        rounds: scenario.rounds,
+        split: 6,
+        max_connections_per_server: 6,
+    };
+    let (mut dispatcher, gids) = seed_pressure_dispatcher(base);
+    let change_global = dispatcher.dispatch_json(rpc_request(
+        RpcMethod::Aria2ChangeGlobalOption,
+        vec![RpcValue::Object(BTreeMap::from([
+            (
+                "max-overall-download-limit".to_owned(),
+                RpcValue::String(scenario.global_download_limit.to_string()),
+            ),
+            (
+                "max-overall-upload-limit".to_owned(),
+                RpcValue::String(scenario.global_upload_limit.to_string()),
+            ),
+        ]))],
+    ));
+    assert!(
+        change_global.error.is_none(),
+        "changeGlobalOption should succeed in shared-runtime fairness setup"
+    );
+
+    for (index, gid) in gids.iter().enumerate() {
+        let (download_limit, upload_limit) = if index == 0 {
+            (
+                scenario.constrained_per_download_limit,
+                scenario.constrained_per_upload_limit,
+            )
+        } else {
+            (
+                scenario.default_per_download_limit,
+                scenario.default_per_upload_limit,
+            )
+        };
+        let change_option = dispatcher.dispatch_json(rpc_request(
+            RpcMethod::Aria2ChangeOption,
+            vec![
+                RpcValue::String(gid.clone()),
+                RpcValue::Object(BTreeMap::from([
+                    (
+                        "max-download-limit".to_owned(),
+                        RpcValue::String(download_limit.to_string()),
+                    ),
+                    (
+                        "max-upload-limit".to_owned(),
+                        RpcValue::String(upload_limit.to_string()),
+                    ),
+                ])),
+            ],
+        ));
+        assert!(
+            change_option.error.is_none(),
+            "changeOption should succeed in shared-runtime fairness setup"
+        );
+    }
+
+    (dispatcher, gids)
+}
+
+fn run_tell_status_pressure(
+    dispatcher: &mut InProcessRpcDispatcher,
+    gids: &[String],
+    rounds: usize,
+) -> usize {
+    let mut calls = 0usize;
+    for round in 0..rounds {
+        let gid = &gids[round % gids.len()];
+        dispatcher
+            .apply_bt_runtime_tick(
+                gid,
+                0,
+                0,
+                400 + round as u64 * 10,
+                160 + round as u64 * 5,
+                0,
+                0,
+                false,
+                Some(8),
+            )
+            .expect("bench tellStatus churn should succeed");
+
+        for gid in gids {
+            let response = dispatcher.dispatch_json(rpc_request(
+                RpcMethod::Aria2TellStatus,
+                vec![RpcValue::String(gid.clone())],
+            ));
+            match response.result {
+                Some(RpcValue::Object(payload)) => {
+                    assert_eq!(payload.get("isBt"), Some(&RpcValue::Bool(true)));
+                    calls += 1;
+                }
+                other => panic!("unexpected tellStatus payload in bench: {other:?}"),
+            }
+        }
+    }
+    calls
+}
+
+fn run_resource_limit_pressure(
+    dispatcher: &mut InProcessRpcDispatcher,
+    gids: &[String],
+    scenario: ResourceLimitScenario,
+) -> usize {
+    let mut calls = 0usize;
+    let expected_download_speed = (scenario.global_download_limit / scenario.task_count as u64)
+        .min(scenario.per_download_limit)
+        .max(1);
+    let expected_upload_speed = (scenario.global_upload_limit / scenario.task_count as u64)
+        .min(scenario.per_upload_limit)
+        .max(1);
+
+    for round in 0..scenario.rounds {
+        for gid in gids {
+            dispatcher
+                .apply_bt_runtime_tick(gid, 64, 32, 5_000, 2_000, 1, 1, false, Some(6))
+                .expect("bench limited runtime tick should succeed");
+        }
+
+        let gid = &gids[round % gids.len()];
+        let tell_status = dispatcher.dispatch_json(rpc_request(
+            RpcMethod::Aria2TellStatus,
+            vec![RpcValue::String(gid.clone())],
+        ));
+        match tell_status.result {
+            Some(RpcValue::Object(payload)) => {
+                assert_eq!(
+                    payload.get("downloadSpeed"),
+                    Some(&RpcValue::String(expected_download_speed.to_string()))
+                );
+                assert_eq!(
+                    payload.get("uploadSpeed"),
+                    Some(&RpcValue::String(expected_upload_speed.to_string()))
+                );
+                calls += 1;
+            }
+            other => panic!("unexpected tellStatus payload under resource caps: {other:?}"),
+        }
+
+        let tell_global =
+            dispatcher.dispatch_json(rpc_request(RpcMethod::Aria2TellGlobalStat, Vec::new()));
+        match tell_global.result {
+            Some(RpcValue::Object(payload)) => {
+                assert_eq!(
+                    payload.get("downloadSpeed"),
+                    Some(&RpcValue::String(
+                        expected_download_speed
+                            .saturating_mul(scenario.task_count as u64)
+                            .to_string(),
+                    ))
+                );
+                assert_eq!(
+                    payload.get("uploadSpeed"),
+                    Some(&RpcValue::String(
+                        expected_upload_speed
+                            .saturating_mul(scenario.task_count as u64)
+                            .to_string(),
+                    ))
+                );
+                calls += 1;
+            }
+            other => panic!("unexpected tellGlobalStat payload under resource caps: {other:?}"),
+        }
+    }
+
+    calls
+}
+
+fn run_mixed_rpc_pressure(
+    dispatcher: &mut InProcessRpcDispatcher,
+    gids: &[String],
+    rounds: usize,
+) -> usize {
+    let mut calls = 0usize;
+    for round in 0..rounds {
+        let gid = &gids[round % gids.len()];
+        dispatcher
+            .apply_bt_runtime_tick(
+                gid,
+                128,
+                64,
+                500 + round as u64 * 15,
+                180 + round as u64 * 6,
+                1,
+                1,
+                round + 1 >= rounds,
+                Some(6 + (round % 4) as u32),
+            )
+            .expect("bench mixed churn should succeed");
+
+        let tell_status = dispatcher.dispatch_json(rpc_request(
+            RpcMethod::Aria2TellStatus,
+            vec![RpcValue::String(gid.clone())],
+        ));
+        assert!(matches!(tell_status.result, Some(RpcValue::Object(_))));
+        calls += 1;
+
+        let tell_active =
+            dispatcher.dispatch_json(rpc_request(RpcMethod::Aria2TellActive, Vec::new()));
+        assert!(matches!(tell_active.result, Some(RpcValue::Array(_))));
+        calls += 1;
+
+        let tell_global =
+            dispatcher.dispatch_json(rpc_request(RpcMethod::Aria2TellGlobalStat, Vec::new()));
+        assert!(matches!(tell_global.result, Some(RpcValue::Object(_))));
+        calls += 1;
+    }
+    calls
+}
+
+fn run_shared_runtime_fairness_pressure(
+    dispatcher: &mut InProcessRpcDispatcher,
+    gids: &[String],
+    scenario: SharedRuntimeFairnessScenario,
+) -> usize {
+    let mut calls = 0usize;
+    let initial_shared_download_speed =
+        (scenario.global_download_limit / scenario.task_count as u64).max(1);
+    let initial_shared_upload_speed =
+        (scenario.global_upload_limit / scenario.task_count as u64).max(1);
+    let constrained_initial_download_speed = initial_shared_download_speed
+        .min(scenario.constrained_per_download_limit)
+        .max(1);
+    let constrained_initial_upload_speed = initial_shared_upload_speed
+        .min(scenario.constrained_per_upload_limit)
+        .max(1);
+    let rebalanced_active_count = scenario.task_count.saturating_sub(1).max(1);
+    let rebalanced_download_speed = (scenario.global_download_limit
+        / rebalanced_active_count as u64)
+        .min(scenario.default_per_download_limit)
+        .max(1);
+    let rebalanced_upload_speed = (scenario.global_upload_limit / rebalanced_active_count as u64)
+        .min(scenario.default_per_upload_limit)
+        .max(1);
+
+    for round in 0..scenario.rounds {
+        let active_slice = if round == 0 { gids } else { &gids[1..] };
+        for gid in active_slice {
+            dispatcher
+                .apply_bt_runtime_tick(gid, 128, 64, 5_000, 2_000, 0, 0, false, Some(6))
+                .expect("shared-runtime fairness tick should succeed");
+        }
+
+        for (index, gid) in gids.iter().enumerate() {
+            let tell_status = dispatcher.dispatch_json(rpc_request(
+                RpcMethod::Aria2TellStatus,
+                vec![RpcValue::String(gid.clone())],
+            ));
+            match tell_status.result {
+                Some(RpcValue::Object(payload)) => {
+                    if round > 0 && index == 0 {
+                        assert_eq!(
+                            payload.get("status"),
+                            Some(&RpcValue::String("complete".to_owned()))
+                        );
+                    } else {
+                        let expected_download_speed = if index == 0 {
+                            constrained_initial_download_speed
+                        } else if round == 0 {
+                            initial_shared_download_speed
+                                .min(scenario.default_per_download_limit)
+                                .max(1)
+                        } else {
+                            rebalanced_download_speed
+                        };
+                        let expected_upload_speed = if index == 0 {
+                            constrained_initial_upload_speed
+                        } else if round == 0 {
+                            initial_shared_upload_speed
+                                .min(scenario.default_per_upload_limit)
+                                .max(1)
+                        } else {
+                            rebalanced_upload_speed
+                        };
+                        assert_eq!(
+                            payload.get("downloadSpeed"),
+                            Some(&RpcValue::String(expected_download_speed.to_string()))
+                        );
+                        assert_eq!(
+                            payload.get("uploadSpeed"),
+                            Some(&RpcValue::String(expected_upload_speed.to_string()))
+                        );
+                    }
+                    calls += 1;
+                }
+                other => panic!(
+                    "unexpected tellStatus payload in shared-runtime fairness bench: {other:?}"
+                ),
+            }
+        }
+
+        let tell_global =
+            dispatcher.dispatch_json(rpc_request(RpcMethod::Aria2TellGlobalStat, Vec::new()));
+        match tell_global.result {
+            Some(RpcValue::Object(payload)) => {
+                let (expected_download_speed, expected_upload_speed) = if round == 0 {
+                    (
+                        constrained_initial_download_speed.saturating_add(
+                            initial_shared_download_speed
+                                .min(scenario.default_per_download_limit)
+                                .max(1)
+                                .saturating_mul(rebalanced_active_count as u64),
+                        ),
+                        constrained_initial_upload_speed.saturating_add(
+                            initial_shared_upload_speed
+                                .min(scenario.default_per_upload_limit)
+                                .max(1)
+                                .saturating_mul(rebalanced_active_count as u64),
+                        ),
+                    )
+                } else {
+                    (
+                        constrained_initial_download_speed
+                            .saturating_add(
+                                rebalanced_download_speed
+                                    .saturating_mul(rebalanced_active_count as u64),
+                            )
+                            .min(scenario.global_download_limit),
+                        constrained_initial_upload_speed
+                            .saturating_add(
+                                rebalanced_upload_speed
+                                    .saturating_mul(rebalanced_active_count as u64),
+                            )
+                            .min(scenario.global_upload_limit),
+                    )
+                };
+                assert_eq!(
+                    payload.get("downloadSpeed"),
+                    Some(&RpcValue::String(expected_download_speed.to_string()))
+                );
+                assert_eq!(
+                    payload.get("uploadSpeed"),
+                    Some(&RpcValue::String(expected_upload_speed.to_string()))
+                );
+                calls += 1;
+            }
+            other => panic!(
+                "unexpected tellGlobalStat payload in shared-runtime fairness bench: {other:?}"
+            ),
+        }
+
+        if round == 0 {
+            dispatcher
+                .mark_complete(&gids[0])
+                .expect("shared-runtime fairness benchmark should complete one gid");
+        }
+    }
+
+    calls
+}
+
+pub(super) fn bench_tell_status_pressure(c: &mut Criterion) {
+    let mut group = c.benchmark_group("rpc_tell_status_pressure");
+    for scenario in [
+        PressureScenario {
+            task_count: 64,
+            rounds: 4,
+            split: 4,
+            max_connections_per_server: 4,
+        },
+        PressureScenario {
+            task_count: 128,
+            rounds: 4,
+            split: 8,
+            max_connections_per_server: 8,
+        },
+    ] {
+        group.throughput(Throughput::Elements(
+            (scenario.task_count * scenario.rounds) as u64,
+        ));
+        group.bench_with_input(
+            BenchmarkId::new("tell_status", scenario.task_count),
+            &scenario,
+            |b, &scenario| {
+                b.iter_batched(
+                    || seed_pressure_dispatcher(scenario),
+                    |(mut dispatcher, gids)| {
+                        let calls =
+                            run_tell_status_pressure(&mut dispatcher, &gids, scenario.rounds);
+                        assert_eq!(calls, scenario.task_count * scenario.rounds);
+                    },
+                    criterion::BatchSize::SmallInput,
+                );
+            },
+        );
+    }
+    group.finish();
+}
+
+pub(super) fn bench_mixed_rpc_pressure(c: &mut Criterion) {
+    let mut group = c.benchmark_group("rpc_mixed_pressure");
+    for scenario in [
+        PressureScenario {
+            task_count: 96,
+            rounds: 6,
+            split: 6,
+            max_connections_per_server: 6,
+        },
+        PressureScenario {
+            task_count: 192,
+            rounds: 6,
+            split: 8,
+            max_connections_per_server: 8,
+        },
+    ] {
+        group.throughput(Throughput::Elements((scenario.rounds * 3) as u64));
+        group.bench_with_input(
+            BenchmarkId::new("mixed_rpc", scenario.task_count),
+            &scenario,
+            |b, &scenario| {
+                b.iter_batched(
+                    || seed_pressure_dispatcher(scenario),
+                    |(mut dispatcher, gids)| {
+                        let calls = run_mixed_rpc_pressure(&mut dispatcher, &gids, scenario.rounds);
+                        assert_eq!(calls, scenario.rounds * 3);
+                    },
+                    criterion::BatchSize::SmallInput,
+                );
+            },
+        );
+    }
+    group.finish();
+}
+
+pub(super) fn bench_resource_limit_pressure(c: &mut Criterion) {
+    let mut group = c.benchmark_group("rpc_speed_limit_pressure");
+    for scenario in [
+        ResourceLimitScenario {
+            task_count: 32,
+            rounds: 6,
+            global_download_limit: 2_560,
+            per_download_limit: 160,
+            global_upload_limit: 1_280,
+            per_upload_limit: 80,
+            disk_cache_bytes: 8 * 1024 * 1024,
+        },
+        ResourceLimitScenario {
+            task_count: 64,
+            rounds: 6,
+            global_download_limit: 5_120,
+            per_download_limit: 120,
+            global_upload_limit: 2_560,
+            per_upload_limit: 60,
+            disk_cache_bytes: 16 * 1024 * 1024,
+        },
+    ] {
+        group.throughput(Throughput::Elements((scenario.rounds * 2) as u64));
+        group.bench_with_input(
+            BenchmarkId::new("speed_limit", scenario.task_count),
+            &scenario,
+            |b, &scenario| {
+                b.iter_batched(
+                    || seed_limited_dispatcher(scenario),
+                    |(mut dispatcher, gids)| {
+                        let calls = run_resource_limit_pressure(&mut dispatcher, &gids, scenario);
+                        assert_eq!(calls, scenario.rounds * 2);
+                    },
+                    criterion::BatchSize::SmallInput,
+                );
+            },
+        );
+    }
+    group.finish();
+}
+
+pub(super) fn bench_shared_runtime_fairness_pressure(c: &mut Criterion) {
+    let mut group = c.benchmark_group("rpc_shared_runtime_fairness_pressure");
+    for scenario in [
+        SharedRuntimeFairnessScenario {
+            label: "three_way_rebalance",
+            task_count: 3,
+            rounds: 4,
+            global_download_limit: 1_200,
+            default_per_download_limit: 900,
+            constrained_per_download_limit: 250,
+            global_upload_limit: 600,
+            default_per_upload_limit: 500,
+            constrained_per_upload_limit: 120,
+        },
+        SharedRuntimeFairnessScenario {
+            label: "four_way_rebalance",
+            task_count: 4,
+            rounds: 4,
+            global_download_limit: 1_600,
+            default_per_download_limit: 900,
+            constrained_per_download_limit: 220,
+            global_upload_limit: 800,
+            default_per_upload_limit: 500,
+            constrained_per_upload_limit: 100,
+        },
+    ] {
+        group.throughput(Throughput::Elements(
+            (scenario.task_count * scenario.rounds) as u64,
+        ));
+        group.bench_with_input(
+            BenchmarkId::new("shared_runtime_fairness", scenario.label),
+            &scenario,
+            |b, &scenario| {
+                b.iter_batched(
+                    || seed_shared_runtime_fairness_dispatcher(scenario),
+                    |(mut dispatcher, gids)| {
+                        let calls =
+                            run_shared_runtime_fairness_pressure(&mut dispatcher, &gids, scenario);
+                        assert_eq!(calls, scenario.rounds * (scenario.task_count + 1));
+                    },
+                    criterion::BatchSize::SmallInput,
+                );
+            },
+        );
+    }
+    group.finish();
+}

crates/aria2-rust-pro-tests/benches/rpc_pressure/runtime_engine_pressure.rs

@@ -0,0 +1,206 @@
+#![expect(
+    clippy::redundant_pub_crate,
+    reason = "criterion bench entry points are re-exported only to the private bench root module"
+)]
+
+use super::support::{
+    BenchmarkId, Criterion, DownloadEngine, DownloadStatus, PieceId, PieceState, RuntimeConfig,
+    Throughput,
+};
+
+#[derive(Clone, Copy, Debug)]
+struct BackpressureScenario {
+    task_count: usize,
+    rounds: usize,
+    disk_cache_bytes: u64,
+    split: usize,
+    max_connections_per_server: usize,
+}
+
+fn seed_instrumented_engine(task_count: usize) -> DownloadEngine {
+    let runtime = RuntimeConfig {
+        split: 4,
+        max_connections_per_server: 4,
+        max_connection_per_server: 4,
+        min_split_size: 1024,
+        piece_length: 1024,
+        ..RuntimeConfig::default()
+    };
+    let mut engine = DownloadEngine::with_runtime(runtime);
+    for i in 0..task_count {
+        let gid = engine
+            .add_uri(format!("magnet:?xt=urn:btih:{:040x}", 90_001 + i))
+            .gid();
+        let group = engine
+            .handle_mut(gid)
+            .expect("newly inserted benchmark group should exist");
+        group.set_status(if i % 3 == 0 {
+            DownloadStatus::Waiting
+        } else {
+            DownloadStatus::Active
+        });
+        group.set_total_length(8 * 1024);
+        group.set_completed_length((i % 4) as u64 * 1024);
+        group.set_piece_length(1024);
+        group.set_piece_state(PieceId(0), PieceState::Verified);
+        group.set_piece_state(PieceId(1), PieceState::Pending);
+        group.set_piece_state(PieceId(2), PieceState::Downloading);
+        group.set_piece_state(PieceId(3), PieceState::Queued);
+        group.set_piece_state(PieceId(4), PieceState::Missing);
+    }
+
+    for _ in 0..3 {
+        let _ = engine.schedule_once();
+    }
+    engine
+}
+
+fn seed_backpressure_engine(scenario: BackpressureScenario) -> DownloadEngine {
+    let runtime = RuntimeConfig {
+        split: scenario.split,
+        max_connections_per_server: scenario.max_connections_per_server,
+        max_connection_per_server: scenario.max_connections_per_server,
+        min_split_size: 1024,
+        piece_length: 1024,
+        disk_cache_bytes: scenario.disk_cache_bytes,
+        ..RuntimeConfig::default()
+    };
+    let mut engine = DownloadEngine::with_runtime(runtime);
+    for i in 0..scenario.task_count {
+        let gid = engine
+            .add_uri(format!("https://example.org/backpressure-{i}.bin"))
+            .gid();
+        let group = engine
+            .handle_mut(gid)
+            .expect("newly inserted backpressure group should exist");
+        group.set_status(match i % 4 {
+            0 => DownloadStatus::Waiting,
+            2 => DownloadStatus::Error,
+            _ => DownloadStatus::Active,
+        });
+        group.set_total_length(16 * 1024);
+        group.set_completed_length((i % 8) as u64 * 1024);
+        group.set_piece_length(1024);
+        group.set_download_speed(2_500 + i as u64 * 11);
+        group.set_upload_speed(900 + i as u64 * 5);
+        group.set_num_connections((scenario.max_connections_per_server.min(8)) as u32);
+        group.set_retry_count((i % 3) as u32);
+        group.set_piece_state(PieceId(0), PieceState::Verified);
+        group.set_piece_state(PieceId(1), PieceState::Pending);
+        group.set_piece_state(PieceId(2), PieceState::Downloading);
+        group.set_piece_state(PieceId(3), PieceState::Queued);
+        group.set_piece_state(PieceId(4), PieceState::Missing);
+    }
+
+    for _ in 0..4 {
+        let _ = engine.schedule_once();
+    }
+    engine
+}
+
+fn run_runtime_snapshot_pressure(engine: &mut DownloadEngine, rounds: usize) -> usize {
+    let gids = engine.registry().handles().collect::<Vec<_>>();
+    for round in 0..rounds {
+        let gid = gids[round % gids.len()].gid();
+        let group = engine
+            .handle_mut(gid)
+            .expect("benchmark group should still exist");
+        group.set_download_speed(600 + round as u64 * 10);
+        group.set_upload_speed(200 + round as u64 * 5);
+        let _ = engine.schedule_once();
+        let runtime = engine.runtime_instrumentation_snapshot();
+        assert!(runtime.download_count >= gids.len());
+        assert!(runtime.scheduler_counters.schedule_run_count >= 1);
+    }
+    rounds
+}
+
+fn run_backpressure_runtime_pressure(
+    engine: &mut DownloadEngine,
+    scenario: BackpressureScenario,
+) -> usize {
+    let gids = engine.registry().handles().collect::<Vec<_>>();
+    for round in 0..scenario.rounds {
+        let gid = gids[round % gids.len()].gid();
+        let group = engine
+            .handle_mut(gid)
+            .expect("backpressure benchmark group should still exist");
+        group.set_retry_count((round % 5) as u32);
+        group.set_status(if round % 3 == 0 {
+            DownloadStatus::Waiting
+        } else {
+            DownloadStatus::Active
+        });
+        group.set_download_speed(4_000 + round as u64 * 40);
+        group.set_upload_speed(1_500 + round as u64 * 25);
+
+        let _ = engine.schedule_once();
+        let runtime = engine.runtime_instrumentation_snapshot();
+        assert_eq!(
+            runtime.configured_disk_cache_bytes,
+            scenario.disk_cache_bytes
+        );
+        assert!(runtime.total_active_segments >= 1);
+        assert!(runtime.scheduler_counters.schedule_run_count >= 1);
+    }
+    scenario.rounds
+}
+
+pub(super) fn bench_runtime_snapshot_pressure(c: &mut Criterion) {
+    let mut group = c.benchmark_group("runtime_snapshot_pressure");
+    for task_count in [64usize, 128, 256] {
+        group.throughput(Throughput::Elements(task_count as u64));
+        group.bench_with_input(
+            BenchmarkId::new("runtime_snapshot", task_count),
+            &task_count,
+            |b, &task_count| {
+                b.iter_batched(
+                    || seed_instrumented_engine(task_count),
+                    |mut engine| {
+                        let observations = run_runtime_snapshot_pressure(&mut engine, 8);
+                        assert_eq!(observations, 8);
+                    },
+                    criterion::BatchSize::SmallInput,
+                );
+            },
+        );
+    }
+    group.finish();
+}
+
+pub(super) fn bench_scheduler_backpressure_pressure(c: &mut Criterion) {
+    let mut group = c.benchmark_group("scheduler_backpressure_pressure");
+    for scenario in [
+        BackpressureScenario {
+            task_count: 64,
+            rounds: 10,
+            disk_cache_bytes: 4 * 1024 * 1024,
+            split: 4,
+            max_connections_per_server: 4,
+        },
+        BackpressureScenario {
+            task_count: 128,
+            rounds: 10,
+            disk_cache_bytes: 32 * 1024 * 1024,
+            split: 8,
+            max_connections_per_server: 8,
+        },
+    ] {
+        group.throughput(Throughput::Elements(scenario.task_count as u64));
+        group.bench_with_input(
+            BenchmarkId::new("backpressure", scenario.task_count),
+            &scenario,
+            |b, &scenario| {
+                b.iter_batched(
+                    || seed_backpressure_engine(scenario),
+                    |mut engine| {
+                        let observations = run_backpressure_runtime_pressure(&mut engine, scenario);
+                        assert_eq!(observations, scenario.rounds);
+                    },
+                    criterion::BatchSize::SmallInput,
+                );
+            },
+        );
+    }
+    group.finish();
+}

crates/aria2-rust-pro-tests/benches/rpc_pressure/support.rs

@@ -0,0 +1,41 @@
+#![expect(
+    clippy::redundant_pub_crate,
+    reason = "private criterion bench modules share fixtures through pub(super) support exports"
+)]
+
+pub(super) use std::{
+    collections::BTreeMap,
+    fs,
+    io::{Read, Write},
+    net::{SocketAddr, TcpListener, TcpStream},
+    path::PathBuf,
+    sync::{
+        Arc, Mutex,
+        atomic::{AtomicBool, Ordering},
+    },
+    thread,
+    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
+};
+
+pub(super) use aria2_rust_pro_cli::{Invocation, execute_runtime_with_downloader};
+pub(super) use aria2_rust_pro_core::{
+    DownloadEngine, DownloadStatus, PieceId, PieceState, RuntimeConfig,
+};
+pub(super) use aria2_rust_pro_protocol::{
+    ReqwestHttpConnector, TorrentPeerModel, TrackerPeerListModel, TrackerResponseModel,
+    downloader::ConnectorBackedDownloader,
+};
+pub(super) use aria2_rust_pro_rpc::{InProcessRpcDispatcher, JsonRpcRequest, RpcMethod, RpcValue};
+pub(super) use criterion::{BenchmarkId, Criterion, Throughput};
+
+pub(super) const BT_TORRENT_FIXTURE: &str = "ZDg6YW5ub3VuY2UzNTpodHRwOi8vdHJhY2tlci5leGFtcGxlLm9yZy9hbm5vdW5jZTQ6aW5mb2Q0Om5hbWUxMDp1YnVudHUuaXNvMTI6cGllY2UgbGVuZ3RoaTE2Mzg0ZTY6bGVuZ3RoaTMyNzY4ZTY6cGllY2VzNDA6YWFhYWFhYWFhYWFhYWFhYWFhYWFiYmJiYmJiYmJiYmJiYmJiYmJiYmVl";
+
+pub(super) fn rpc_request(method: RpcMethod, params: Vec<RpcValue>) -> JsonRpcRequest {
+    JsonRpcRequest {
+        jsonrpc: Some("2.0".to_owned()),
+        id: None,
+        method: method.as_str().to_owned(),
+        params,
+        meta: Default::default(),
+    }
+}