Replace toxiproxy_rust with an in-repo Toxiproxy HTTP client

CHANGELOG.md

@@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Changed
 
 - **Breaking:** Replaced the unmaintained `fxhash` crate ([RUSTSEC-2025-0057](https://rustsec.org/advisories/RUSTSEC-2025-0057)) with `rustc-hash`. `set_multi`, `add_multi` and `stats` now return `rustc_hash::FxHashMap` instead of `fxhash::FxHashMap`. Both are `std::collections::HashMap` with a different hasher type, so code that only calls map methods compiles unchanged. Code that names the `fxhash` type must switch to `rustc_hash::FxHashMap` from `rustc-hash` 2.x or use type inference.
+- Replaced the `toxiproxy_rust` dev-dependency with a small Toxiproxy HTTP client in the test support code. `toxiproxy_rust` pulled in `reqwest` 0.11, which depends on the vulnerable `h2` 0.3 ([RUSTSEC-2026-0258](https://rustsec.org/advisories/RUSTSEC-2026-0258)) and the unmaintained `rustls-pemfile` ([RUSTSEC-2025-0134](https://rustsec.org/advisories/RUSTSEC-2025-0134)). `toxiproxy_rust` has been a dev-dependency since 0.8.1, so this change does not affect users of 0.8.1 or later. Versions 0.5.0 to 0.8.0 list it as a normal dependency; their users drop it by upgrading.
 
 ## [0.9.0] - 2026-10-19
 

Cargo.toml

@@ -29,7 +29,7 @@ tokio = { version = "1.26", features = ["full"] }
 rand = "0.9"
 criterion = { version = "0.8", features = ["async_tokio"] }
 serial_test = "3.1"
-toxiproxy_rust = "0.1"
+serde_json = "1.0"
 
 [features]
 default = []

tests/resiliency_tests.rs

@@ -95,7 +95,7 @@ fn test_set_multi_errors_on_upstream_with_toxic_client_via_limit_data() {
     let byte_limit = command.len() - 10;
     fixture
         .proxy
-        .with_limit_data("upstream".into(), byte_limit as u32, 1.0)
+        .with_limit_data("upstream", byte_limit as u32, 1.0)
         .apply(|| {
             rt.block_on(within(async {
                 assert_disconnect(client.set_multi(&pairs, None, None).await);
@@ -127,7 +127,7 @@ fn test_set_multi_errors_on_downstream_with_toxic_client_via_limit_data() {
     ];
     fixture
         .proxy
-        .with_limit_data("downstream".into(), (b"STORED\r\n".len() + 1) as u32, 1.0)
+        .with_limit_data("downstream", (b"STORED\r\n".len() + 1) as u32, 1.0)
         .apply(|| {
             rt.block_on(within(async {
                 assert_disconnect(client.set_multi(&pairs, None, None).await);
@@ -235,7 +235,7 @@ fn truncated_read_bodies_never_return_partial_values() {
         let mut client = rt.block_on(within(Client::new(fixture.dsn()))).unwrap();
         fixture
             .proxy
-            .with_limit_data("downstream".into(), 128, 1.0)
+            .with_limit_data("downstream", 128, 1.0)
             .apply(|| {
                 rt.block_on(within(async {
                     assert_disconnect(operation.read(&mut client).await);
@@ -256,10 +256,10 @@ fn fragmented_requests_and_responses_preserve_payloads() {
     let fixture = ToxicMemcached::new();
     let rt = runtime();
     let value: Vec<u8> = (0..=255).cycle().take(1024).collect();
-    fixture.proxy.with_slicer("upstream".into(), 7, 0, 100, 1.0);
+    fixture.proxy.with_slicer("upstream", 7, 0, 100, 1.0);
     fixture
         .proxy
-        .with_slicer("downstream".into(), 7, 0, 100, 1.0)
+        .with_slicer("downstream", 7, 0, 100, 1.0)
         .apply(|| {
             rt.block_on(within(async {
                 let mut client = Client::new(fixture.dsn()).await.unwrap();
@@ -283,6 +283,14 @@ fn fragmented_requests_and_responses_preserve_payloads() {
         .unwrap();
 }
 
+#[test]
+#[ignore = "Requires the memcached and toxiproxy-server executables"]
+#[should_panic(expected = "stream was invalid")]
+fn toxiproxy_api_errors_fail_the_test() {
+    let fixture = ToxicMemcached::new();
+    fixture.proxy.with_slicer("sideways", 7, 0, 100, 1.0);
+}
+
 #[test]
 #[ignore = "Requires the memcached and toxiproxy-server executables"]
 fn caller_timeouts_allow_recovery_with_a_new_connection() {
@@ -296,7 +304,7 @@ fn caller_timeouts_allow_recovery_with_a_new_connection() {
         }));
         fixture
             .proxy
-            .with_timeout("downstream".into(), 0, 1.0)
+            .with_timeout("downstream", 0, 1.0)
             .apply(|| {
                 rt.block_on(within(async {
                     let result = tokio::time::timeout(

tests/support/toxiproxy.rs

@@ -1,12 +1,12 @@
 use super::{Memcached, TestProcess};
+use serde_json::{json, Value};
 use std::io::{Read, Write};
-use std::net::{TcpListener, TcpStream};
+use std::net::{SocketAddr, TcpListener, TcpStream};
 use std::process::Command;
 use std::time::Duration;
-use toxiproxy_rust::{
-    client::Client,
-    proxy::{Proxy, ProxyPack},
-};
+
+const PROXY_NAME: &str = "memcached";
+const API_TIMEOUT: Duration = Duration::from_secs(5);
 
 pub struct ToxicMemcached {
     pub proxy: Proxy,
@@ -34,31 +34,10 @@ impl ToxicMemcached {
         drop(reservation);
         process.spawn(&mut command);
         process.wait_for("the Toxiproxy HTTP API", || {
-            let timeout = Duration::from_millis(200);
-            let mut stream = TcpStream::connect_timeout(&address, timeout).ok()?;
-            stream.set_read_timeout(Some(timeout)).ok()?;
-            stream.set_write_timeout(Some(timeout)).ok()?;
-            stream
-                .write_all(b"GET /version HTTP/1.0\r\nHost: localhost\r\n\r\n")
-                .ok()?;
-            let mut response = String::new();
-            stream.read_to_string(&mut response).ok()?;
-            response
-                .starts_with("HTTP/1.0 200")
-                .then_some(())
-                .or_else(|| response.starts_with("HTTP/1.1 200").then_some(()))
+            request(address, "GET", "/version", None, Duration::from_millis(200)).ok()
         });
-        let client = Client::new(address);
-        let proxy = client
-            .populate(vec![ProxyPack::new(
-                "memcached".into(),
-                "127.0.0.1:0".into(),
-                memcached.address(),
-            )])
-            .expect("Cannot create the test proxy")
-            .pop()
-            .unwrap();
-        assert_ne!(proxy.proxy_pack.listen, "127.0.0.1:0");
+        let proxy = Proxy::create(address, &memcached.address());
+        assert_ne!(proxy.listen, "127.0.0.1:0");
         Self {
             proxy,
             memcached,
@@ -67,6 +46,126 @@ impl ToxicMemcached {
     }
 
     pub fn dsn(&self) -> String {
-        format!("tcp://{}", self.proxy.proxy_pack.listen)
+        format!("tcp://{}", self.proxy.listen)
+    }
+}
+
+/// A minimal client for one proxy on the Toxiproxy HTTP API.
+///
+/// See <https://github.com/Shopify/toxiproxy#http-api>.
+pub struct Proxy {
+    api: SocketAddr,
+    listen: String,
+}
+
+impl Proxy {
+    fn create(api: SocketAddr, upstream: &str) -> Self {
+        let body = json!({
+            "name": PROXY_NAME,
+            "listen": "127.0.0.1:0",
+            "upstream": upstream,
+            "enabled": true,
+        });
+        let response = request(api, "POST", "/proxies", Some(&body), API_TIMEOUT)
+            .expect("Cannot create the test proxy");
+        let proxy: Value =
+            serde_json::from_str(&response).expect("Toxiproxy returned an invalid proxy");
+        let listen = proxy["listen"]
+            .as_str()
+            .expect("Toxiproxy did not return a listen address")
+            .to_owned();
+        Self { api, listen }
+    }
+
+    /// Runs `closure` with the proxy disabled, then enables it again.
+    pub fn with_down(&self, closure: impl FnOnce()) -> Result<(), String> {
+        self.set_enabled(false)?;
+        closure();
+        self.set_enabled(true)
+    }
+
+    /// Runs `closure` with the registered toxics, then resets the proxy.
+    /// The reset removes all toxics and enables the proxy.
+    pub fn apply(&self, closure: impl FnOnce()) -> Result<(), String> {
+        closure();
+        // The fixture owns this Toxiproxy process, so a reset affects only this proxy.
+        request(self.api, "POST", "/reset", None, API_TIMEOUT).map(drop)
+    }
+
+    pub fn with_limit_data(&self, stream: &str, bytes: u32, toxicity: f32) -> &Self {
+        self.add_toxic("limit_data", stream, toxicity, json!({ "bytes": bytes }))
+    }
+
+    pub fn with_slicer(
+        &self,
+        stream: &str,
+        average_size: u32,
+        size_variation: u32,
+        delay: u32,
+        toxicity: f32,
+    ) -> &Self {
+        let attributes = json!({
+            "average_size": average_size,
+            "size_variation": size_variation,
+            "delay": delay,
+        });
+        self.add_toxic("slicer", stream, toxicity, attributes)
+    }
+
+    pub fn with_timeout(&self, stream: &str, timeout: u32, toxicity: f32) -> &Self {
+        self.add_toxic("timeout", stream, toxicity, json!({ "timeout": timeout }))
+    }
+
+    fn add_toxic(&self, kind: &str, stream: &str, toxicity: f32, attributes: Value) -> &Self {
+        let body = json!({
+            "name": format!("{kind}_{stream}"),
+            "type": kind,
+            "stream": stream,
+            "toxicity": toxicity,
+            "attributes": attributes,
+        });
+        let path = format!("/proxies/{PROXY_NAME}/toxics");
+        if let Err(error) = request(self.api, "POST", &path, Some(&body), API_TIMEOUT) {
+            panic!("Cannot create the {} toxic: {}", kind, error);
+        }
+        self
+    }
+
+    fn set_enabled(&self, enabled: bool) -> Result<(), String> {
+        let path = format!("/proxies/{PROXY_NAME}");
+        let body = json!({ "enabled": enabled });
+        request(self.api, "PATCH", &path, Some(&body), API_TIMEOUT).map(drop)
+    }
+}
+
+/// Sends one HTTP/1.0 request and returns the body of a 2xx response.
+fn request(
+    api: SocketAddr,
+    method: &str,
+    path: &str,
+    body: Option<&Value>,
+    timeout: Duration,
+) -> Result<String, String> {
+    let body = body.map(Value::to_string).unwrap_or_default();
+    let context = |error: std::io::Error| format!("{method} {path} failed: {error}");
+    let mut stream = TcpStream::connect_timeout(&api, timeout).map_err(context)?;
+    stream.set_read_timeout(Some(timeout)).map_err(context)?;
+    stream.set_write_timeout(Some(timeout)).map_err(context)?;
+    // HTTP/1.0 makes the server close the connection after the response, so no chunked bodies.
+    write!(
+        stream,
+        "{method} {path} HTTP/1.0\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{body}",
+        body.len()
+    )
+    .map_err(context)?;
+    let mut response = String::new();
+    stream.read_to_string(&mut response).map_err(context)?;
+    let (head, body) = response
+        .split_once("\r\n\r\n")
+        .ok_or_else(|| format!("{method} {path} returned an incomplete response"))?;
+    let status = head.lines().next().unwrap_or_default();
+    match status.split(' ').nth(1) {
+        Some(code) if code.starts_with('2') => Ok(body.to_owned()),
+        _ => Err(format!("{method} {path} returned {status:?}: {body}")),
     }
 }