Testing a network server
Step 9 of 10. Step 8 ended at the tag mini-ch08 with transactions working: MULTI, EXEC,
DISCARD, one queue per connection. This step starts from that tag and ends at mini-ch09, and
it changes how the server is proved, not what it does. Until now every test lived inside
src/main.rs and called the server’s own functions. This step adds a second test file that knows
none of that: the server binary runs as a child process, a client speaks RESP to it over a real
socket, and no type, function or struct crosses between them.
What the unit tests cannot see
Section titled “What the unit tests cannot see”The 88 tests from steps 2–8 all live behind #[cfg(test)] in src/main.rs. They call decode,
execute, queue_or_execute and serve directly, with a Connection built by hand. That
verifies the pieces and misses the seams between them: nothing runs main, so the snapshot load
at startup, the bind and the accept loop are untested. And a unit test cannot answer the question
a user of a server actually has — can a program that knows nothing about this code connect and
drive it?
An integration test goes in tests/. Cargo builds it as a separate crate, so it cannot import
anything from the server even by accident. That is not a technicality; it is the difference
between testing the functions and testing the server.
The port becomes an argument
Section titled “The port becomes an argument”main has hardcoded 127.0.0.1:6379 since step 4, and a test cannot own that port. The obvious
fix — grab a free port in the test, then hand it to the server — has a race in the gap between
releasing the probe and the child’s bind. So the binary takes its port as an argument, and port 0
means “any free port”: the test asks the server to pick, and reads the answer back from the ready
line. The real server refuses port 0; the clone takes it, on purpose.
// The port is the one thing the real server also takes as an // argument (`redis-server --port 6380`): an integration test needs // a port it owns, and 6379 belongs to whoever started first. Port // 0 means "any free port" — the real server refuses it, the clone // takes it, because a test that probes for a free port and then // hands it over has a race in the gap. The printed line carries // the address that was actually bound, so a test reads its port // back instead of guessing. let port: u16 = std::env::args() .nth(1) .and_then(|arg| arg.parse().ok()) .unwrap_or(6379); let listener = TcpListener::bind(("127.0.0.1", port)).await?; let port = listener.local_addr()?.port(); println!("listening on 127.0.0.1:{port}");A client that shares no code with the server
Section titled “A client that shares no code with the server”mini-redis is a binary crate, so tests/client.rs cannot import its Frame type even if it
wanted to — and it does not want to. The contract between a server and the world is the protocol,
and a client that imported the server’s types would be testing the server against itself. So the
test file builds its own small client. Writing a command is one function:
/// The replies a client can receive, decoded from the wire.#[derive(Debug, PartialEq)]enum Reply { Simple(String), Error(String), Integer(i64), Bulk(Option<Vec<u8>>), Array(Vec<Reply>),}
/// Encode one command the way redis-cli does: an array of bulk strings.fn encode(name: &str, args: &[&[u8]]) -> Vec<u8> { let mut out = format!("*{}\r\n${}\r\n{name}\r\n", args.len() + 1, name.len()) .into_bytes(); for arg in args { out.extend(format!("${}\r\n", arg.len()).into_bytes()); out.extend_from_slice(arg); out.extend_from_slice(b"\r\n"); } out}Reading is the server’s problem from step 3 pointed the other way — a buffer, a scan for a line, and enough bytes for a bulk string:
impl Client { /// Send one command, read one reply. async fn cmd(&mut self, name: &str, args: &[&[u8]]) -> std::io::Result<Reply> { self.stream.write_all(&encode(name, args)).await?; self.reply().await }
/// One reply, decoded by its first byte. Arrays recurse. async fn reply(&mut self) -> std::io::Result<Reply> { let line = self.line().await?; let (kind, rest) = (line[0], &line[1..]); let text = String::from_utf8_lossy(rest).into_owned(); match kind { b'+' => Ok(Reply::Simple(text)), b'-' => Ok(Reply::Error(text)), b':' => Ok(Reply::Integer(text.parse().expect("an integer"))), b'$' => { if text == "-1" { Ok(Reply::Bulk(None)) } else { let n: usize = text.parse().expect("a length"); Ok(Reply::Bulk(Some(self.take(n).await?))) } } b'*' => { if text == "-1" { Ok(Reply::Array(Vec::new())) } else { let n: usize = text.parse().expect("a count"); let mut items = Vec::with_capacity(n); for _ in 0..n { // Boxed so the async fn does not recurse: the // future's size would otherwise depend on itself. items.push(Box::pin(self.reply()).await?); } Ok(Reply::Array(items)) } } _ => Err(std::io::Error::new( std::io::ErrorKind::InvalidData, format!("not a RESP reply: {:?}", line), )), } }The one surprise in there is Box::pin. An async fn that calls itself compiles to a future
whose size depends on itself, which is a type error — E0733, “recursion in an async fn requires
boxing”. Boxing the inner call breaks the cycle, because a Box has a fixed size no matter what
it holds.
The miniature below runs both halves on one machine: a HashMap behind the protocol, and the
client in front of it, sharing nothing but the bytes. It prints what SET a 1 looks like on the
wire, then four decoded replies:
[dependencies]tokio = { version = "1", features = ["macros", "rt-multi-thread", "net", "io-util"] }use std::collections::HashMap;
use tokio::io::{AsyncReadExt, AsyncWriteExt};use tokio::net::{TcpListener, TcpStream};
/// The replies a server sends and a client decodes.enum Reply { Simple(String), Error(String), Bulk(Option<Vec<u8>>),}
/// One command as a RESP array of bulk strings, the way redis-cli sends/// everything.fn encode_command(name: &str, args: &[&[u8]]) -> Vec<u8> { let mut out = format!("*{}\r\n${}\r\n{name}\r\n", args.len() + 1, name.len()).into_bytes(); for arg in args { out.extend(format!("${}\r\n", arg.len()).into_bytes()); out.extend_from_slice(arg); out.extend_from_slice(b"\r\n"); } out}
fn encode_reply(reply: &Reply, out: &mut Vec<u8>) { match reply { Reply::Simple(text) => out.extend(format!("+{text}\r\n").into_bytes()), Reply::Error(text) => out.extend(format!("-{text}\r\n").into_bytes()), Reply::Bulk(None) => out.extend(b"$-1\r\n".iter().copied()), Reply::Bulk(Some(data)) => { out.extend(format!("${}\r\n", data.len()).into_bytes()); out.extend_from_slice(data); out.extend_from_slice(b"\r\n"); } }}
/// One end of the wire: the socket plus a read buffer, so a frame split/// across two packets is reassembled. The client and the server both/// run one of these — the two programs share nothing but the protocol.struct Wire { stream: TcpStream, buf: Vec<u8>, pos: usize,}
impl Wire { fn new(stream: TcpStream) -> Wire { Wire { stream, buf: Vec::new(), pos: 0 } }
/// Whatever the socket has, appended after what is buffered. async fn fill(&mut self) -> std::io::Result<()> { let mut chunk = [0u8; 1024]; let n = self.stream.read(&mut chunk).await?; if n == 0 { return Err(std::io::Error::new( std::io::ErrorKind::UnexpectedEof, "the other end hung up", )); } self.buf.extend_from_slice(&chunk[..n]); Ok(()) }
/// One CRLF-terminated line, waiting for the rest of it. async fn line(&mut self) -> std::io::Result<Vec<u8>> { loop { if let Some(i) = self.buf[self.pos..].iter().position(|&b| b == b'\n') { let mut text = self.buf[self.pos..self.pos + i].to_vec(); self.pos += i + 1; text.pop(); // the trailing \r return Ok(text); } self.fill().await?; } }
/// Exactly `n` bytes, then the CRLF that ends them. async fn take(&mut self, n: usize) -> std::io::Result<Vec<u8>> { while self.buf.len() - self.pos < n { self.fill().await?; } let data = self.buf[self.pos..self.pos + n].to_vec(); self.pos += n; assert_eq!(self.line().await?, b"", "bulk data ends with CRLF"); Ok(data) }
/// One command: `*N`, then N bulk strings. async fn command(&mut self) -> std::io::Result<Vec<Vec<u8>>> { let header = self.line().await?; let n: usize = String::from_utf8_lossy(&header[1..]) .parse() .expect("an array length"); let mut parts = Vec::with_capacity(n); for _ in 0..n { let header = self.line().await?; assert_eq!(header[0], b'$', "a bulk string"); let len: usize = String::from_utf8_lossy(&header[1..]) .parse() .expect("a length"); parts.push(self.take(len).await?); } Ok(parts) }
/// One reply, decoded by its first byte. async fn reply(&mut self) -> std::io::Result<Reply> { let line = self.line().await?; let (kind, rest) = (line[0], &line[1..]); let text = String::from_utf8_lossy(rest).into_owned(); match kind { b'+' => Ok(Reply::Simple(text)), b'-' => Ok(Reply::Error(text)), b'$' => { if text == "-1" { Ok(Reply::Bulk(None)) } else { let len: usize = text.parse().expect("a length"); Ok(Reply::Bulk(Some(self.take(len).await?))) } } _ => Err(std::io::Error::new( std::io::ErrorKind::InvalidData, format!("not a reply: {:?}", line), )), } }}
/// Send one command and print the decoded reply.async fn ask(client: &mut Wire, name: &str, args: &[&[u8]]) { client.stream.write_all(&encode_command(name, args)).await.unwrap(); let reply = client.reply().await.unwrap(); let shown = match reply { Reply::Simple(text) => format!("+{text}"), Reply::Error(text) => format!("-{text}"), Reply::Bulk(Some(data)) => { format!("\"{}\"", String::from_utf8_lossy(&data)) } Reply::Bulk(None) => "(nil)".to_string(), }; println!("{name} -> {shown}");}
#[tokio::main]async fn main() { // The server half: one connection served against a HashMap, ending // when the client hangs up. let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); tokio::spawn(async move { let (stream, _) = listener.accept().await.unwrap(); let mut wire = Wire::new(stream); let mut db = HashMap::new(); while let Ok(cmd) = wire.command().await { let name = String::from_utf8_lossy(&cmd[0]).to_uppercase(); let reply = match name.as_str() { "SET" => { db.insert(cmd[1].clone(), cmd[2].clone()); Reply::Simple("OK".to_string()) } "GET" => match db.get(&cmd[1]) { Some(value) => Reply::Bulk(Some(value.clone())), None => Reply::Bulk(None), }, _ => Reply::Error(format!("ERR unknown command '{name}'")), }; let mut out = Vec::new(); encode_reply(&reply, &mut out); if wire.stream.write_all(&out).await.is_err() { break; } } });
// The client half: a socket and the protocol, nothing else. let stream = TcpStream::connect(addr).await.unwrap(); let mut client = Wire::new(stream);
let bytes = encode_command("SET", &[b"a", b"1"]); let shown: String = bytes .iter() .map(|&b| match b { b'\r' => "\\r".to_string(), b'\n' => "\\n".to_string(), _ => (b as char).to_string(), }) .collect(); println!("on the wire: {shown}");
ask(&mut client, "SET", &[b"a", b"1"]).await; ask(&mut client, "GET", &[b"a"]).await; ask(&mut client, "GET", &[b"b"]).await; ask(&mut client, "BADCMD", &[]).await;}on the wire: *3\r\n$3\r\nSET\r\n$1\r\na\r\n$1\r\n1\r\nSET -> +OKGET -> "1"GET -> (nil)BADCMD -> -ERR unknown command 'BADCMD'Nothing in that output is about threads or tasks — the client is a socket, the protocol, and a buffer, and that is all a client ever is.
Starting the server from the test
Section titled “Starting the server from the test”env!("CARGO_BIN_EXE_mini-redis") expands at compile time to the path of the just-built server
binary, so the test spawns the real thing — the same program a reader would run:
/// The server binary, running on a port this test owns and in a/// directory this test owns — so its snapshot file cannot collide with/// another test's, and a `SAVE` in one test leaves nothing behind in/// the next.struct Server { child: Child, dir: PathBuf, port: u16, /// The read half of the server's stdout, held for the server's /// whole life. The sweeper prints to stdout when it reclaims a key, /// and if the read end were closed after the readiness line, that /// print would hit a broken pipe and kill the server mid-test. _stdout: std::io::BufReader<ChildStdout>,}
static NEXT: AtomicU64 = AtomicU64::new(0);
impl Server { async fn start() -> Server { let dir = std::env::temp_dir().join(format!( "mini-redis-it-{}-{}", std::process::id(), NEXT.fetch_add(1, Ordering::SeqCst) )); std::fs::create_dir_all(&dir).expect("makes the test directory"); let mut child = Command::new(env!("CARGO_BIN_EXE_mini-redis")) .arg("0") // any free port; the ready line says which one .current_dir(&dir) .stdout(Stdio::piped()) // The server's stderr carries per-connection error notes; // the tests provoke one on purpose and do not need it. .stderr(Stdio::null()) .spawn() .expect("spawns the server binary"); // The server prints `listening on 127.0.0.1:PORT` when it is // ready. Reading that line replaces every sleep-based readiness // hack: once it is printed, the port is bound and accepting. // Its absence at EOF means the child died starting up — bind // failure, bad argument — and the test should say so, not // connect to whatever comes next. let stdout = child.stdout.take().expect("stdout is piped"); let mut reader = std::io::BufReader::new(stdout); let mut ready = String::new(); loop { let mut line = String::new(); let n = reader.read_line(&mut line).expect("reads the server's stdout"); if n == 0 { break; // EOF: the child is gone } if line.starts_with("listening on") { ready = line.trim_end().to_string(); break; } } assert!( !ready.is_empty(), "the server exited before it was ready: {:?}", child.wait() ); let port: u16 = ready .rsplit(':') .next() .expect("the ready line has a port") .parse() .expect("the ready line ends with a port"); Server { child, dir, port, _stdout: reader, } }}
impl Drop for Server { fn drop(&mut self) { self.child.kill().ok(); self.child.wait().ok(); std::fs::remove_dir_all(&self.dir).ok(); }}Three details in that harness are load-bearing, and two of them were found the hard way. The
readiness line replaces sleep-based waiting, and its absence at EOF means the child died starting
up — a bind failure should say so, not turn into a connection error somewhere else. The read half
of stdout is held for the server’s whole life, because the sweeper prints swept N expired key(s)
when it reclaims a key, and if the test had closed the pipe after the readiness line, that print
would have hit a broken pipe and killed the server mid-test. And each server gets its own
temporary directory, so its mini.rdb cannot collide with another test’s and a SAVE in one test
leaves nothing behind in the next.
The tests
Section titled “The tests”Twelve, one server per test, all running in parallel — the port argument and the per-test
directory are what make that safe. The two-connection ones are where a unit test with a
hand-built Connection stops being able to help:
#[tokio::test]async fn publish_reaches_a_subscriber_without_being_asked() { let server = Server::start().await; let mut subscriber = server.client().await; assert_eq!( subscriber.cmd("SUBSCRIBE", &[b"news"]).await.unwrap(), Reply::Array(vec![ Reply::Bulk(Some(b"subscribe".to_vec())), Reply::Bulk(Some(b"news".to_vec())), Reply::Integer(1), ]) ); let mut publisher = server.client().await; assert_eq!( publisher.cmd("PUBLISH", &[b"news", b"hi"]).await.unwrap(), Reply::Integer(1) ); // The pushed message arrives on the subscriber's socket, unprompted. assert_eq!( subscriber.reply().await.unwrap(), Reply::Array(vec![ Reply::Bulk(Some(b"message".to_vec())), Reply::Bulk(Some(b"news".to_vec())), Reply::Bulk(Some(b"hi".to_vec())), ]) );}#[tokio::test]async fn a_value_expires_when_its_ttl_runs_out() { let server = Server::start().await; let mut client = server.client().await; client.set("session", b"secret").await.unwrap(); assert_eq!( client.cmd("EXPIRE", &[b"session", b"1"]).await.unwrap(), Reply::Integer(1) ); tokio::time::sleep(std::time::Duration::from_millis(1200)).await; assert_eq!(client.get("session").await.unwrap(), Reply::Bulk(None));}#[tokio::test]async fn a_protocol_error_ends_the_connection_without_a_reply() { let server = Server::start().await; let mut client = server.client().await; client.stream.write_all(b"?\r\n").await.unwrap(); assert!(client.reply().await.is_err(), "the server hangs up");}The rest pin the wire shapes one command at a time: PING, a SET visible to the same client
and to a second connection, a null bulk for a missing key, the exact unknown-command error, a
list through LPUSH/LRANGE, the WRONGTYPE error for reading a list as a string, a
transaction queuing into one array, two pipelined commands answered in order.
Seeing it run
Section titled “Seeing it run”$ cargo test --test client Finished `test` profile [unoptimized + debuginfo] target(s) in 0.02s Running tests/client.rs (/Volumes/RustBuild/tracks/debug/deps/client-b319679af9513e17)
running 12 teststest a_protocol_error_ends_the_connection_without_a_reply ... oktest the_server_answers_ping ... oktest an_unknown_command_is_an_error ... oktest reading_a_list_as_a_string_is_a_wrongtype_error ... oktest pipelined_commands_come_back_in_order ... oktest a_get_on_a_missing_key_is_a_null_bulk ... oktest a_set_is_visible_to_a_get_from_the_same_client ... oktest a_list_round_trips_through_lpush_and_lrange ... oktest a_set_is_visible_to_a_second_connection ... oktest publish_reaches_a_subscriber_without_being_asked ... oktest a_transaction_queues_and_execs_into_one_array ... oktest a_value_expires_when_its_ttl_runs_out ... ok
test result: ok. 12 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 1.30sThe order the tests print in varies between runs; the set does not. The Running line names this
machine’s shared build cache; elsewhere it reads target/debug/deps/client-…. cargo test with
no filter also runs the 88 unit tests in src/main.rs, which are untouched and still pass.
The port argument is a feature in its own right now — the step 8 demos that needed a rebuilt binary on a scratch port are one argument away:
$ printf 'set a 1\nget a\nexpire a 10\nttl a\n' | redis-cli -p 7783OK1110That session ran against a server started as ./mini-redis 7783, whose entire stdout was:
listening on 127.0.0.1:7783What is deliberately not here
Section titled “What is deliberately not here”- A test client from crates.io.
redis-rswould drive the server with a battle-tested client, and that is a real argument — a hand-rolled decoder can share blind spots with a hand-rolled encoder. The counter-argument: the whole protocol is under a hundred lines here, and writing it makes the contract explicit. The next step’s benchmark client uses the same shape. - Failure injection. Nothing drops packets mid-frame; the split-frame test from step 3 stays a unit test with a deliberate pause. A real fault-injection harness is its own project.
WATCHand check-and-set. Still missing, as noted in step 8 — and aWATCHtest would be the first integration test that needs retries, which is a design question, not a test detail.- Portability. The harness assumes a Unix-y localhost,
killsemantics and a temp directory. Windows would want named pipes or a port-per-worker strategy instead.
Step 10, Benchmarks, and what real Redis does, points this client at the server in a loop and measures what it costs. It is not written yet, so it gets no link — the track overview carries the full list as articles land.
Related
Section titled “Related”- Transactions with MULTI and EXEC — step 8, the tag this step starts from; its transaction queue is one of the behaviours the new tests now reach over the wire.
- Pub/sub messaging — step 7, whose fan-out the publish test drives from a second connection.
- Handling concurrent connections — step 4, the one-task-per-connection shape that lets twelve tests each own a server.