Project Overview

• trixter — a high‑performance, runtime‑tunable TCP chaos proxy — a minimal, blazing‑fast written in Rust with Tokio. It lets you inject latency, throttle bandwidth, slice writes (to simulate small MTUs/Nagle‑like behavior), corrupt bytes in flight by injecting random bytes, randomly terminate connections, and hard‑timeout sessions – all controllable per connection via a simple REST API.

• tokio-netem — a collection of Tokio AsyncRead/AsyncWrite adapters (delay, throttle, slice, terminate, shutdown, corrupt data, inject data) that power the Trixter proxy and can be used independently in tests and harnesses.

The remainder of this document dives into the proxy. For the adapter crate’s detailed guide, follow the tokio-netem link above.

Trixter – Chaos Monkey TCP Proxy

A high‑performance, runtime‑tunable TCP chaos proxy — a minimal, blazing‑fast alternative to Toxiproxy written in Rust with Tokio. It lets you inject latency, throttle bandwidth, slice writes (to simulate small MTUs/Nagle‑like behavior), corrupt bytes in flight by injecting random bytes, randomly terminate connections, and hard‑timeout sessions – all controllable per connection via a simple REST API.

Why Trixter?

• Zero-friction: one static binary, no external deps.
• Runtime knobs: flip chaos on/off without restarting.
• Per-conn control: target just the flows you want.
• Minimal overhead: adapters are lightweight and composable.

Features

• Fast path: tokio::io::copy_bidirectional on a multi‑thread runtime;
• Runtime control (per active connection):
• Latency: add/remove delay in ms.
• Throttle: cap bytes/sec.
• Slice: split writes into fixed‑size chunks.
• Corrupt: inject random bytes with a tunable probability.
• Chaos termination: probability [0.0..=1.0] to abort on each read/write.
• Hard timeout: stop a session after N milliseconds.
• REST API to list connections and change settings on the fly.
• Targeted kill: shut down a single connection with a reason.
• RST on chaos: resets (best‑effort) when a timeout/termination triggers.

Quick start

1) Build

cd trixter/trixter
cargo build --release

2) Run an upstream echo server (demo)

Use any TCP server. Examples:

# Example: simple echo with netcat (macOS/Linux)
nc -lk 127.0.0.1 8181
# or Python
python3 -m socketserver -b 127.0.0.1 8181

3) Start the trixter proxy

RUST_LOG=info \
./target/release/trixter \
  --listen 0.0.0.0:8080 \
  --upstream 127.0.0.1:8181 \
  --api 127.0.0.1:8888 \
  --delay-ms 0 \
  --throttle-rate-bytes 0 \
  --slice-size-bytes 0 \
  --corrupt-probability-rate 0.0 \
  --terminate-probability-rate 0.0 \
  --connection-duration-ms 0

Now connect your app/CLI to localhost:8080. The proxy forwards to 127.0.0.1:8181.

REST API

Base URL is the --api address, e.g. http://127.0.0.1:8888.

Data model

{
  "conn_info": {
    "id": "pN7e3y...",
    "downstream": "127.0.0.1:59024",
    "upstream": "127.0.0.1:8181"
  },
  "delay": { "secs": 2, "nanos": 500000000 },
  "throttle_rate": 10240,
  "slice_size": 512,
  "terminate_probability_rate": 0.05,
  "corrupt_probability_rate": 0.02
}

Notes:

• delay serializes as a std::time::Duration object with secs/nanos fields (zeroed when the delay is disabled).
• id is unique per connection; use it to target a single connection.
• corrupt_probability_rate reports the current per-operation flip probability (0.0 when corruption is off).

Health check

curl -s http://127.0.0.1:8888/health

List connections

curl -s http://127.0.0.1:8888/connections | jq

Kill a connection

ID=$(curl -s http://127.0.0.1:8888/connections | jq -r '.[0].conn_info.id')

curl -i -X POST \
  http://127.0.0.1:8888/connections/$ID/shutdown \
  -H 'Content-Type: application/json' \
  -d '{"reason":"test teardown"}'

Kill all connections

curl -i -X POST \
  http://127.0.0.1:8888/connections/_all/shutdown \
  -H 'Content-Type: application/json' \
  -d '{"reason":"test teardown"}'

Set latency (ms)

curl -i -X PATCH \
  http://127.0.0.1:8888/connections/$ID/delay \
  -H 'Content-Type: application/json' \
  -d '{"delay_ms":250}'

# Remove latency
curl -i -X PATCH \
  http://127.0.0.1:8888/connections/$ID/delay \
  -H 'Content-Type: application/json' \
  -d '{"delay_ms":0}'

Throttle bytes/sec

curl -i -X PATCH \
  http://127.0.0.1:8888/connections/$ID/throttle \
  -H 'Content-Type: application/json' \
  -d '{"rate_bytes":10240}'  # 10 KiB/s

Slice writes (bytes)

curl -i -X PATCH \
  http://127.0.0.1:8888/connections/$ID/slice \
  -H 'Content-Type: application/json' \
  -d '{"size_bytes":512}'

Randomly terminate reads/writes

# Set 5% probability per read/write operation
curl -i -X PATCH \
  http://127.0.0.1:8888/connections/$ID/termination \
  -H 'Content-Type: application/json' \
  -d '{"probability_rate":0.05}'

Inject random bytes

# Corrupt ~1% of operations
curl -i -X PATCH \
  http://127.0.0.1:8888/connections/$ID/corruption \
  -H 'Content-Type: application/json' \
  -d '{"probability_rate":0.01}'

# Remove corruption
curl -i -X PATCH \
  http://127.0.0.1:8888/connections/$ID/corruption \
  -H 'Content-Type: application/json' \
  -d '{"probability_rate":0.0}'

Error responses

• 404 Not Found — bad connection ID
• 400 Bad Request — invalid probability (outside 0.0..=1.0) for termination/corruption
• 500 Internal Server Error — internal channel/handler error

CLI flags

--listen <ip:port>                  # e.g. 0.0.0.0:8080
--upstream <ip:port>                # e.g. 127.0.0.1:8181
--api <ip:port>                     # e.g. 127.0.0.1:8888
--delay-ms <ms>                     # 0 = off (default)
--throttle-rate-bytes <bytes/s>     # 0 = unlimited (default)
--slice-size-bytes <bytes>          # 0 = off (default)
--terminate-probability-rate <0..1> # 0.0 = off (default)
--corrupt-probability-rate <0..1>   # 0.0 = off (default)
--connection-duration-ms <ms>       # 0 = unlimited (default)

All of the above can be changed per connection at runtime via the REST API, except --connection-duration-ms which is a process‑wide default applied to new connections.

How it works (architecture)

Each accepted downstream connection spawns a task that:

1. Connects to the upstream target.

2. Wraps both sides with tunable adapters with tokio-netem:

3. DelayedWriter → optional latency

4. ThrottledWriter → bandwidth cap
5. SlicedWriter → fixed‑size write chunks
6. Terminator → probabilistic aborts
7. Corrupter → probabilistic random byte injector
8. Shutdowner (downstream only) → out‑of‑band shutdown via oneshot channel
9. Runs tokio::io::copy_bidirectional until EOF/error/timeout.
10. Tracks the live connection in a DashMap so the API can query/mutate it.

Use cases

• Flaky networks: simulate 3G/EDGE/satellite latency and low bandwidth.
• MTU/segmentation bugs: force small write slices to uncover packetization assumptions.
• Resilience drills: randomly kill connections during critical paths.
• Data validation: corrupt bytes to exercise checksums and retry logic.
• Timeout tuning: enforce hard upper‑bounds to validate client retry/backoff logic.
• Canary/E2E tests: target only specific connections and tweak dynamically.
• Load/soak: run for hours with varying chaos settings from CI/scripts.

Recipes

Simulate a shaky mobile link

# Add ~250ms latency and 64 KiB/s cap to the first active connection
ID=$(curl -s localhost:8888/connections | jq -r '.[0].conn_info.id')

curl -s -X PATCH localhost:8888/connections/$ID/delay    \
  -H 'Content-Type: application/json' -d '{"delay_ms":250}'

curl -s -X PATCH localhost:8888/connections/$ID/throttle \
  -H 'Content-Type: application/json' -d '{"rate_bytes":65536}'

Force tiny packets (find buffering bugs)

curl -s -X PATCH localhost:8888/connections/$ID/slice \
  -H 'Content-Type: application/json' -d '{"size_bytes":256}'

Introduce flakiness (5% ops abort)

curl -s -X PATCH localhost:8888/connections/$ID/termination \
  -H 'Content-Type: application/json' -d '{"probability_rate":0.05}'

Add data corruption

curl -s -X PATCH localhost:8888/connections/$ID/corruption \
  -H 'Content-Type: application/json' -d '{"probability_rate":0.01}'

Timebox a connection to 5s at startup

./trixter \
  --listen 0.0.0.0:8080 \
  --upstream 127.0.0.1:8181 \
  --api 127.0.0.1:8888 \
  --connection-duration-ms 5000

Kill the slowpoke

curl -s -X POST localhost:8888/connections/$ID/shutdown \
  -H 'Content-Type: application/json' -d '{"reason":"too slow"}'

Integration: CI & E2E tests

• Spin up the proxy as a sidecar/container.
• Discover the right connection (by downstream/upstream pair) via GET /connections.
• Apply chaos during specific test phases with PATCH calls.
• Always clean up with POST /connections/{id}/shutdown to free ports quickly.

Performance notes

• Built on Tokio multi‑thread runtime; avoid heavy CPU work on the I/O threads.
• Throttling and slicing affect throughput by design; set them to 0 to disable.
• Use loopback or a fast NIC for local tests; network stack/OS settings still apply.
• Logging: RUST_LOG=info (or debug) for visibility; turn off for max throughput.

Security

• The API performs no auth; bind it to a trusted interface (e.g., 127.0.0.1).
• The proxy is transparent TCP; apply your own TLS/ACLs at the edges if needed.

Error handling

• Invalid probability returns 400 with { "error": "invalid probability; must be between 0.0 and 1.0" }.
• Unknown connection IDs return 404.
• Internal channel/handler errors return 500.

tokio-netem

tokio-netem provides a toolbox of Tokio AsyncRead/AsyncWrite adapters that let you emulate latency, throttling, slicing, terminations, forced shutdowns, data injections and data corruption without touching your application code. Compose them around TcpStream (or any Tokio I/O type) to run realistic integration tests and chaos experiments.

Table of Contents

• Why tokio-netem?
• Installation
• Adapters
• DelayedReader & DelayedWriter
• SlicedWriter
• ThrottledReader & ThrottledWriter
• Corrupter
• ReadInjector & WriteInjector
• Terminator
• Shutdowner

Why tokio-netem?

• Reproduce hard-to-trigger network edge cases in unit and integration tests.
• Model network delays, fragmented writes, corruption, or partial packet delivery without external proxies.
• Flip failures on and off at runtime to probe resilience and backoff strategies.
• Share dynamic knobs across tasks so admin endpoints or test harnesses can adjust conditions without rebuilding pipelines.

Installation

Add the crate to your Cargo.toml:

[dependencies]
tokio-netem = "0.1"

Adapters

Each adapter accepts either a static configuration (plain Duration, usize, or f64) or a dynamic handle (Arc<...>). Static setups are ideal for fixed scenarios in documentation or smoke tests. Dynamic handles shine when you want to tweak behavior while the pipeline is running. All examples below use a TCP client stream to emphasise end-to-end usage.

In practice the writer half is the most common place to inject chaos: outbound adapters such as DelayedWriter, SlicedWriter, ThrottledWriter, and Terminator add near-zero overhead because they operate on the data already in-flight and do not require extra buffering in memory the way a reader-side shim might. Start perturbing on the write path first, then layer reader adapters if you need full-duplex scenarios.

NetEm extension traits

There are two extension traits – NetEmReadExt and NetEmWriteExt which significantly simplify usage of some adapters (not all supported, see the full documentation):

```rust,no_run use std::io; use tokio::net::TcpStream; use tokio_netem::io::NetEmWriteExt;

[tokio::main]

async fn main() -> io::Result<()> { let mut stream = TcpStream::connect("localhost:80") .await? .throttle_writes(32 * 1024) // 32 KB/s .slice_writes(16); // fragment writes Ok(()) } ```

DelayedReader & DelayedWriter

Adds latency before reads or writes progress. DelayedReader delays the first byte of each buffered burst; DelayedWriter delays every outbound write.

Static config example

```rust,no_run use std::time::Duration; use tokio::io::{self, AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::TcpStream; use tokio_netem::io::{NetEmReadExt, NetEmWriteExt};

[tokio::main]

async fn main() -> io::Result<()> { let stream = TcpStream::connect("127.0.0.1:5555").await?;

let mut stream = stream.delay_writes(Duration::from_millis(25));
let mut stream = BufReader::new(stream).delay_reads(Duration::from_millis(10));

stream.write_all(b"ping").await?; // ~25 ms pause before bytes depart
stream.flush().await?;

let mut line = String::new();