Day 7 · TCP Server

By the end of Day 6 kvlite is fast, thread-safe, and durable — but you can only talk to it from its own REPL. Today it becomes a server: a process that listens on a TCP port and answers many clients at once, each on its own connection, speaking a line protocol you can drive with nc. We do it with std::net and one OS thread per connection — no async, no tokio. That deliberate choice is the lesson: thread-per-connection is the simplest correct network server, and feeling exactly where it stops scaling is the best possible setup for async in Project 3.

Today creates two modules — src/protocol.rs (the meaning of a line) and src/server.rs (the transport) — then rewrites src/main.rs into the final serve/repl binary and adds tests/server.rs. Build order: protocol → server → main.

The line protocol: parse, then execute

The server is a thin shell; all the meaning lives in protocol.rs. One request per line, one reply per line, text not binary — so a human can speak it. Parsing turns an untrusted line into a typed Request; executing runs it against the Db and yields a Response. Keeping the two separate means we can unit-test the grammar without a socket and run the database logic without the network. Here is src/protocol.rs:

   use crate::db::Db;
   use crate::error::{KvError, Result};

   /// A parsed, validated client command.
   #[derive(Debug, Clone, PartialEq, Eq)]
   pub enum Request {
       Set { key: String, value: String },
       Get { key: String },
       Del { key: String },
       Ping,
   }

   /// A reply, kept as a small enum so `Display` owns the wire format in one place.
   #[derive(Debug, Clone, PartialEq, Eq)]
   pub enum Response {
       Ok,
       Value(String),
       Nil,
       Deleted,
       NotFound,
       Pong,
       Err(String),
   }

   impl std::fmt::Display for Response {
       fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
           match self {
               Response::Ok => write!(f, "OK"),
               Response::Value(v) => write!(f, "VALUE {v}"),
               Response::Nil => write!(f, "NIL"),
               Response::Deleted => write!(f, "DELETED"),
               Response::NotFound => write!(f, "NOT_FOUND"),
               Response::Pong => write!(f, "PONG"),
               Response::Err(msg) => write!(f, "ERR {msg}"),
           }
       }
   }

   impl Request {
       /// Parse one client line. The command keyword is case-insensitive; for
       /// `SET`, everything after the key is the value (so values may hold spaces).
       pub fn parse(line: &str) -> Result<Request> {
           let line = line.trim();
           let mut parts = line.splitn(3, ' ');
           let verb = parts
               .next()
               .ok_or_else(|| KvError::Protocol("empty line".into()))?;

           match verb.to_ascii_uppercase().as_str() {
               "SET" => {
                   let key = parts
                       .next()
                       .ok_or_else(|| KvError::Protocol("SET needs a key".into()))?;
                   let value = parts
                       .next()
                       .ok_or_else(|| KvError::Protocol("SET needs a value".into()))?;
                   crate::log::validate_key(key)?;
                   Ok(Request::Set {
                       key: key.to_string(),
                       value: value.to_string(),
                   })
               }
               "GET" => {
                   let key = parts
                       .next()
                       .ok_or_else(|| KvError::Protocol("GET needs a key".into()))?;
                   Ok(Request::Get { key: key.to_string() })
               }
               "DEL" => {
                   let key = parts
                       .next()
                       .ok_or_else(|| KvError::Protocol("DEL needs a key".into()))?;
                   Ok(Request::Del { key: key.to_string() })
               }
               "PING" => Ok(Request::Ping),
               other => Err(KvError::Protocol(format!("unknown command {other:?}"))),
           }
       }

       /// Run the request against the database.
       pub fn execute(&self, db: &Db) -> Result<Response> {
           match self {
               Request::Set { key, value } => {
                   db.set(key, value)?;
                   Ok(Response::Ok)
               }
               Request::Get { key } => Ok(match db.get(key)? {
                   Some(v) => Response::Value(v),
                   None => Response::Nil,
               }),
               Request::Del { key } => Ok(if db.delete(key)? {
                   Response::Deleted
               } else {
                   Response::NotFound
               }),
               Request::Ping => Ok(Response::Pong),
           }
       }
   }

   /// Parse-then-execute one line, converting any error into an `ERR` response so a
   /// single bad request never tears down the connection.
   pub fn handle_line(db: &Db, line: &str) -> Response {
       match Request::parse(line).and_then(|req| req.execute(db)) {
           Ok(resp) => resp,
           Err(e) => Response::Err(e.to_string()),
       }
   }

The grammar is deliberately tiny, and a bad request becomes an ERR line, not a dropped connection:

   request            response
   ───────            ────────
   SET k v rest...    OK
   GET k              VALUE <v>   | NIL
   DEL k              DELETED     | NOT_FOUND
   PING               PONG
   QUIT               BYE   (handled by the server loop; closes the connection)
   <anything else>    ERR protocol error: unknown command "..."

Two payoffs are worth naming. handle_line is the one entry point both the network server and the REPL call, so the protocol is written exactly once. And because Response’s Display owns the wire format, there is a single place that decides what bytes go on the socket.

The blocking server

A TCP server is a loop around one blocking call. TcpListener::accept waits until a client connects, hands you a connected socket, and you go again — and we hand each socket to a fresh thread. This is the whole of src/server.rs:

   use std::io::{BufRead, BufReader, BufWriter, Write};
   use std::net::{TcpListener, TcpStream, ToSocketAddrs};
   use std::thread;

   use crate::db::Db;
   use crate::error::Result;
   use crate::protocol::handle_line;

   /// Bind to `addr` and serve forever, one thread per connection.
   pub fn serve(addr: impl ToSocketAddrs, db: Db) -> Result<()> {
       let listener = TcpListener::bind(addr)?;
       serve_on(listener, db)
   }

   /// Serve on an already-bound listener. Split out from `serve` so a test can bind
   /// to port 0 (an OS-chosen free port), learn the address, then drive a client.
   pub fn serve_on(listener: TcpListener, db: Db) -> Result<()> {
       let local = listener.local_addr()?;
       println!("kvlite listening on {local}");

       for incoming in listener.incoming() {
           match incoming {
               Ok(stream) => {
                   // Clone the handle (an Arc bump) and move it into the thread.
                   let db = db.clone();
                   thread::spawn(move || {
                       let peer = stream
                           .peer_addr()
                           .map(|a| a.to_string())
                           .unwrap_or_else(|_| "?".into());
                       if let Err(e) = handle_connection(stream, db) {
                           eprintln!("connection {peer} ended: {e}");
                       }
                   });
               }
               Err(e) => eprintln!("accept failed: {e}"),
           }
       }
       Ok(())
   }

   /// Serve one client: read a line, reply with a line, repeat until EOF.
   pub fn handle_connection(stream: TcpStream, db: Db) -> Result<()> {
       // Two views of the same socket: one buffered for reading, one for writing.
       let reader = BufReader::new(stream.try_clone()?);
       let mut writer = BufWriter::new(stream);

       for line in reader.lines() {
           let line = line?;
           let trimmed = line.trim();
           if trimmed.is_empty() {
               continue;
           }
           // A friendly way out, in addition to just closing the socket.
           if trimmed.eq_ignore_ascii_case("QUIT") {
               writeln!(writer, "BYE")?;
               writer.flush()?;
               break;
           }
           let response = handle_line(&db, trimmed);
           writeln!(writer, "{response}")?;
           writer.flush()?;   // send it now, don't sit in the buffer
       }
       Ok(())
   }

The whole architecture is in serve_on:

   main thread:  accept ──▶ accept ──▶ accept ──▶ ...   (loops forever, never blocks on a client)
                    │           │          │
                    ▼           ▼          ▼
                 thread 1    thread 2    thread 3        (each: read line ▸ run ▸ reply ▸ repeat)
                    └───────────┴──────────┘
                                │
                          one shared Db   (Arc<RwLock<..>> + the log)

Each connection gets its own thread, and each thread gets a clone of the Db handle. Because that handle is Arc<RwLock<MemStore>> plus an Arc<Mutex<Wal>> (Days 5–6), every clone points at the same map and the same log — so two clients see one consistent store, and the locks serialize their writes. The move keyword transfers the handle into the closure; it’s only a handle, so the data isn’t copied.

Send is what makes thread::spawn accept the closure

thread::spawn requires its closure to be Send — everything captured must be safe to move to another thread. The closure captures stream (a TcpStream, Send) and db (a Db built from Arc<RwLock>/ Arc<Mutex>, which are Send + Sync). So it compiles. Capture something thread-unsafe — say an Rc — and you’d get the same compile error from Day 5. The network layer didn’t need any new safety machinery; the types you already built carry their thread-safety with them.

And handle_connection shows why we call stream.try_clone(): a TcpStream is bidirectional, but we want a buffered reader (to read whole lines) and a buffered writer (to batch the reply) at the same time. try_clone gives a second handle to the same underlying socket so we can wrap each direction independently. reader.lines() is the same iterator idea as log replay, now over a socket: one request per line until the client hangs up (EOF), at which point the loop ends, the thread returns, and this connection’s Db clone drops. The flush() after each reply matters — without it, the response could sit in the BufWriter and the client would hang waiting for an answer stuck in a buffer.

Under the hood — what accept actually returns

TcpListener::accept isn’t where a connection is made — by the time it returns, the kernel has already completed TCP’s three-way handshake (the client’s SYN, the server’s SYN-ACK, the client’s ACK) and placed the finished connection on the listener’s accept queue (the backlog). accept simply dequeues one completed connection and hands you its socket; if the queue is empty it blocks the calling thread until one arrives. That blocking is the defining trait of this model: the main thread is parked in accept doing nothing while it waits, and each connection thread is parked in read doing nothing between a client’s requests. Simple and correct — and, as we’re about to see, expensive at scale.

Why thread-per-connection, and where it breaks

This model is the right first answer: it’s easy to reason about (each connection is a plain top-to-bottom function), it uses real OS parallelism across cores, and it’s perfect for tens to a few hundred connections. Its ceiling is just as real, and it comes from two costs that scale with the number of connections, not the amount of work:

• Memory: every thread has its own stack. Rust’s default spawned-thread stack is 2 MiB.
• Scheduling: the OS must context-switch among all those threads; past a point, the scheduler spends more time switching than working.

By the numbers — the memory wall

Suppose you want 10,000 simultaneous connections — the classic target — with one thread each:

   stack per thread  = 2 MiB         (Rust std default)
   connections       = 10,000
   stack memory      = 2 MiB × 10,000 = 20,000 MiB ≈ 20 GiB   (just for stacks!)

Twenty gigabytes of RAM before storing a single key — and that ignores per-thread kernel bookkeeping and the context-switch tax. Mostly-idle connections (a chat server, long-lived clients) make it worse: you pay for the threads whether or not they’re doing anything. This is precisely the wall async was invented to climb.

It actually happened — the C10k problem

Around 1999, Dan Kegel named and documented “The C10K problem”: the difficulty of serving ten thousand concurrent connections on the hardware of the day, where the dominant thread-(or process-) per-connection model simply ran out of memory and scheduler. His write-up catalogued the alternative — event-driven, non-blocking I/O built on epoll (Linux) and kqueue (BSD), where one thread multiplexes thousands of sockets and never blocks on any one of them. That lineage runs straight to modern async runtimes: Rust’s tokio, which is exactly what Project 3 uses. kvlite’s thread-per-connection server is the honest “before” picture — build it, feel the wall, and async will land as a solution to a problem you’ve actually had, not a buzzword.

The binary: serve and repl

Now rewrite src/main.rs from Day 4’s in-memory REPL into the real, durable binary. It takes two subcommands, parses a tiny bit of std-only argument handling, opens the Db, and either serves it over TCP or runs a local REPL — and crucially, the REPL routes through the same handle_line the server uses:

   use std::io::{self, BufRead, Write};
   use std::process::ExitCode;

   use kvlite::protocol::handle_line;
   use kvlite::{server, Db};

   const DEFAULT_ADDR: &str = "127.0.0.1:6380";
   const DEFAULT_WAL: &str = "kvlite.wal";

   fn main() -> ExitCode {
       let args: Vec<String> = std::env::args().skip(1).collect();
       match run(&args) {
           Ok(()) => ExitCode::SUCCESS,
           Err(e) => {
               eprintln!("error: {e}");
               ExitCode::FAILURE
           }
       }
   }

   fn run(args: &[String]) -> kvlite::Result<()> {
       let mut iter = args.iter();
       let subcommand = iter.next().map(String::as_str);

       // Tiny std-only flag parsing: a positional ADDR and an optional `--wal PATH`.
       let mut addr = DEFAULT_ADDR.to_string();
       let mut wal = DEFAULT_WAL.to_string();
       let mut positional_seen = false;
       while let Some(arg) = iter.next() {
           match arg.as_str() {
               "--wal" => {
                   wal = iter
                       .next()
                       .cloned()
                       .ok_or_else(|| kvlite::KvError::Protocol("--wal needs a path".into()))?;
               }
               other if !positional_seen => {
                   addr = other.to_string();
                   positional_seen = true;
               }
               other => {
                   return Err(kvlite::KvError::Protocol(format!(
                       "unexpected argument {other:?}"
                   )))
               }
           }
       }

       match subcommand {
           Some("serve") => {
               let (db, replayed) = Db::open(&wal)?;
               println!("replayed {replayed} record(s) from {wal}");
               server::serve(addr, db)
           }
           Some("repl") => {
               let (db, replayed) = Db::open(&wal)?;
               println!("kvlite repl — replayed {replayed} record(s) from {wal}");
               println!("commands: SET k v | GET k | DEL k | PING | QUIT");
               repl(db)
           }
           Some(other) => Err(kvlite::KvError::Protocol(format!(
               "unknown subcommand {other:?} (try: serve | repl)"
           ))),
           None => {
               eprintln!("usage: kvlite <serve|repl> [ADDR] [--wal PATH]");
               Err(kvlite::KvError::Protocol("no subcommand given".into()))
           }
       }
   }

   /// A local read-eval-print loop over stdin — same protocol, no network.
   fn repl(db: Db) -> kvlite::Result<()> {
       let stdin = io::stdin();
       let mut stdout = io::stdout();
       for line in stdin.lock().lines() {
           let line = line?;
           let trimmed = line.trim();
           if trimmed.is_empty() {
               continue;
           }
           if trimmed.eq_ignore_ascii_case("QUIT") {
               break;
           }
           let response = handle_line(&db, trimmed);
           writeln!(stdout, "{response}")?;
           stdout.flush()?;
       }
       Ok(())
   }

Finally, wire the two new modules into lib.rs — the finished module list and re-exports:

   pub mod db;
   pub mod error;
   pub mod log;
   pub mod protocol;   // add
   pub mod server;     // add
   pub mod store;

   pub use db::Db;
   pub use error::{KvError, Result};
   pub use log::Command;
   pub use protocol::{handle_line, Request, Response};   // add
   pub use store::{MemStore, SharedStore, Store};

Build it: a networked kvlite, two clients sharing one store

The Day 7 deliverable is the real thing. Start the server:

   $ cargo run -- serve
   replayed 0 record(s) from kvlite.wal
   kvlite listening on 127.0.0.1:6380

In a second terminal, connect with nc and talk to it by hand:

   $ nc 127.0.0.1 6380
   SET name ada lovelace
   OK
   GET name
   VALUE ada lovelace
   PING
   PONG

Now open a third terminal and nc in again — a separate connection on a separate server thread — and GET name. You’ll get VALUE ada lovelace: both clients are hitting one shared, locked Db. That’s the payoff of the whole project visible in one command. Automate exactly this with tests/server.rs, which binds to port 0, spawns the server, and drives real client sockets over the wire:

   use std::io::{BufRead, BufReader, Write};
   use std::net::{TcpListener, TcpStream};
   use std::sync::atomic::{AtomicU64, Ordering};
   use std::thread;

   use kvlite::{server, Db};

   fn temp_wal() -> std::path::PathBuf {
       static N: AtomicU64 = AtomicU64::new(0);
       std::env::temp_dir().join(format!(
           "kvlite-srv-{}-{}.wal",
           std::process::id(),
           N.fetch_add(1, Ordering::Relaxed)
       ))
   }

   /// Send one line, return the one-line reply.
   fn round_trip(reader: &mut impl BufRead, writer: &mut impl Write, line: &str) -> String {
       writeln!(writer, "{line}").unwrap();
       writer.flush().unwrap();
       let mut resp = String::new();
       reader.read_line(&mut resp).unwrap();
       resp.trim_end().to_string()
   }

   #[test]
   fn tcp_set_get_del_roundtrip() {
       let path = temp_wal();
       let _ = std::fs::remove_file(&path);
       let (db, _) = Db::open(&path).unwrap();

       // Bind first so we know the port before the client connects.
       let listener = TcpListener::bind("127.0.0.1:0").unwrap();
       let addr = listener.local_addr().unwrap();
       thread::spawn(move || {
           let _ = server::serve_on(listener, db);
       });

       let stream = TcpStream::connect(addr).unwrap();
       let mut writer = stream.try_clone().unwrap();
       let mut reader = BufReader::new(stream);

       assert_eq!(round_trip(&mut reader, &mut writer, "PING"), "PONG");
       assert_eq!(round_trip(&mut reader, &mut writer, "GET name"), "NIL");
       assert_eq!(round_trip(&mut reader, &mut writer, "SET name ada lovelace"), "OK");
       assert_eq!(
           round_trip(&mut reader, &mut writer, "GET name"),
           "VALUE ada lovelace" // value with spaces survives the wire
       );
       assert_eq!(round_trip(&mut reader, &mut writer, "DEL name"), "DELETED");
       assert_eq!(round_trip(&mut reader, &mut writer, "DEL name"), "NOT_FOUND");
       assert!(round_trip(&mut reader, &mut writer, "bogus").starts_with("ERR"));

       let _ = std::fs::remove_file(&path);
   }

(The crate’s tests/server.rs adds a second test, two_clients_share_one_store, that connects two separate clients and proves one’s SET is visible to the other’s GET.) Run the whole suite with cargo test.

The thread, tied off

Step back and look at what four days forced you to understand — and what the compiler protected you from at each layer. Day 4: a trait and generics, so the engine is reusable and the borrow rules are written into its signatures. Day 5: Arc<RwLock> and the Send/Sync traits, so sharing the store across threads is type-checked free of data races — fearless concurrency, made concrete. Day 6: Result/? and a thiserror enum, so every disk failure is a value you must handle, and durability is a dial you set with open eyes. Day 7: std::net and one thread per connection, so the network is just more of the same ownership and Send rules you already knew — until the C10k wall hands you the reason async exists.

You built a real key-value store, and at no point could you have shipped the most dangerous bug in systems programming, because the compiler wouldn’t let the program that contained it exist. That’s the whole playbook in one project.

→ Next: Project 3 · apilite — An Async Web API · Prev: Day 6 · Persistence & Errors · Back to the Project 2 overview

Check your understanding

1. In the serve_on loop, what exactly is cloned and moved into each connection thread, and why does that give every client a consistent view of one store rather than separate copies?
2. Why does handle_connection call stream.try_clone(), and what would go wrong if you forgot the flush() after writing a reply?
3. By the time accept returns a socket, what has already happened at the TCP level, and what does accept do when no connection is waiting?
4. Thread-per-connection has a hard ceiling. Name the two costs that scale with the number of connections, and use the 2 MiB default stack to estimate the stack memory for 10,000 connections.
5. handle_line is called by both the server and the REPL. Why is routing both through one function a better design than letting each parse commands itself?

Show answers

1. A clone of the Db handle is moved in — and Db is Arc<RwLock<MemStore>> plus Arc<Mutex<Wal>>, so cloning just bumps Arc counts. Every clone points at the same map and log, and the locks serialize access, so all clients share one consistent store rather than independent copies.
2. try_clone gives a second handle to the same socket, so we can wrap one in a BufReader (reading lines) and the other in a BufWriter (writing replies) simultaneously. Without the flush(), a reply could sit in the BufWriter’s buffer and never reach the client, so the client would hang waiting for an answer that was never sent.
3. The kernel has already completed TCP’s three-way handshake (SYN, SYN-ACK, ACK) and put the finished connection on the listener’s accept/backlog queue. accept dequeues one completed connection; if the queue is empty it blocks the calling thread until a connection arrives.
4. The two costs are memory (one stack per thread) and scheduling (context-switching among all the threads). At 2 MiB per thread, 10,000 connections need 2 MiB × 10,000 ≈ 20 GiB of stack memory alone — before storing any data.
5. Both paths need the identical grammar and semantics; writing it once in handle_line means the wire protocol has a single definition, is tested in one place, and can never drift between the REPL and the server. Duplicating the parse/execute logic would risk the two speaking subtly different dialects.