Networking: A Real P2P Node

This is the chapter where your project stops being a program and becomes a network. Up to now everything happened inside one process. Now we’ll run two (or more) nodes that talk to each other over TCP, ask each other for blocks, and relay newly mined blocks so the whole network converges on the same chain. That’s the heart of what makes a blockchain decentralized: no central server, just peers gossiping.

We’ll build it in three layers, from the inside out:

1. The wire protocol — a Message enum describing what peers can say.
2. The node logic — a Node that holds the chain and mempool and decides how to respond to each message.
3. The networking — TCP listeners, threads, and dialing peers, wiring the first two layers onto an actual socket.

Why this is genuinely peer-to-peer

Every node here is both a server and a client. It listens for inbound connections, and it dials out to others. There’s no special “master” node. This is the same shape as the real thing — see The P2P Protocol for how Bitcoin nodes discover and talk to peers.

The wire protocol: one JSON object per line

Before two programs can talk, they need a shared vocabulary. Ours is an enum, where each variant is a kind of message a peer might send. This is src/message.rs:

//! The wire protocol: messages peers exchange over TCP, encoded as one JSON
//! object per line.

use serde::{Deserialize, Serialize};

use crate::block::Block;
use crate::transaction::Transaction;

#[derive(Clone, Debug, Serialize, Deserialize)]
pub enum Message {
    /// "Send me every block from this height onward."
    GetBlocks { from_height: u64 },
    /// A batch of blocks, in order.
    Blocks(Vec<Block>),
    /// A freshly mined block to relay.
    NewBlock(Block),
    /// A transaction to add to the mempool.
    NewTx(Transaction),
    /// Generic acknowledgement.
    Ack,
}

This is a great showcase of why Rust enums are so powerful: each variant can carry its own data. GetBlocks carries a named field, from_height. Blocks carries a Vec<Block>. NewBlock carries one Block, NewTx one Transaction, and Ack carries nothing at all. A single Message value is exactly one of these — the type makes it impossible to, say, have a GetBlocks with a block attached.

Because the enum derives Serialize and Deserialize, serde turns any Message into JSON and back. Our convention is one JSON object per line: each message is serialized to a single line of JSON, terminated by a newline. A newline marks where one message ends and the next begins — simple, and easy to read off a stream.

The node: state plus a handle method

A node’s brain is kept separate from its networking so we can test the logic without opening sockets. This is src/node.rs:

//! A node's in-memory state and its message-handling logic, kept separate from
//! the networking so it's easy to test.

use crate::blockchain::Blockchain;
use crate::message::Message;
use crate::transaction::Transaction;

pub struct Node {
    pub chain: Blockchain,
    pub mempool: Vec<Transaction>,
}

impl Node {
    pub fn new(chain: Blockchain) -> Self {
        Node { chain, mempool: Vec::new() }
    }

A Node is just the Blockchain plus a mempool — the list of pending transactions waiting to be mined into a block. Node::new takes a chain (loaded from disk, perhaps, using last chapter’s load()) and starts with an empty mempool.

The interesting part is handle, which takes one inbound message and optionally returns a reply:

    /// Handle one inbound message, optionally producing a reply to send back.
    pub fn handle(&mut self, msg: Message) -> Option<Message> {
        match msg {
            Message::GetBlocks { from_height } => {
                let from = from_height as usize;
                let blocks = if from < self.chain.blocks.len() {
                    self.chain.blocks[from..].to_vec()
                } else {
                    Vec::new()
                };
                Some(Message::Blocks(blocks))
            }
            Message::Blocks(blocks) => {
                for block in blocks {
                    let _ = self.chain.add_block(block);
                }
                Some(Message::Ack)
            }
            Message::NewBlock(block) => {
                match self.chain.add_block(block) {
                    Ok(()) => println!("accepted new block; height={}", self.chain.height()),
                    Err(e) => eprintln!("rejected new block: {e}"),
                }
                Some(Message::Ack)
            }
            Message::NewTx(tx) => {
                self.mempool.push(tx);
                Some(Message::Ack)
            }
            Message::Ack => None,
        }
    }
}

The whole method is one match on the message variant — and because enum variants carry data, the match arms get to unpack that data by name. Walk the arms:

• GetBlocks { from_height } — a peer wants the blocks it’s missing. We slice our blocks vector from from_height onward (self.chain.blocks[from..]) and reply with a Blocks message. If they already have everything, the slice is empty. The if from < ... len() guard prevents slicing past the end.
• Blocks(blocks) — a batch arrived (the reply to our own GetBlocks). We try to add_block each one. let _ = deliberately ignores the Result: a block we already have, or one that doesn’t yet fit, is simply skipped rather than crashing the node. We reply with Ack.
• NewBlock(block) — someone mined a block and is relaying it. We try to add it and log the outcome: accepted (with the new height) or rejected (with the reason). Then Ack.
• NewTx(tx) — a transaction to remember; push it onto the mempool, Ack.
• Ack — an acknowledgement needs no answer, so we return None.

The return type Option<Message> neatly captures “sometimes there’s a reply, sometimes there isn’t.” This is the engine of block and transaction propagation: each node, on receiving a block or tx, validates it against its own rules and folds it into its own state.

The networking: TCP, threads, and shared state

Now we connect that brain to the world. This is src/net.rs. First, the shared state type and the server loop:

//! A small but real peer-to-peer layer over TCP. Each message is one JSON line.
//! Nodes serve inbound connections on a thread each, and dial out to pull blocks
//! or push newly mined ones.

use std::io::{BufRead, BufReader, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::{Arc, Mutex};
use std::thread;

use crate::message::Message;
use crate::node::Node;

/// Shared, lock-protected node state (multiple peer threads touch it).
pub type Shared = Arc<Mutex<Node>>;

/// Listen for peers forever, handling each connection on its own thread.
pub fn serve(addr: &str, shared: Shared) -> std::io::Result<()> {
    let listener = TcpListener::bind(addr)?;
    println!("listening for peers on {addr}");
    for stream in listener.incoming() {
        let stream = stream?;
        let shared = Arc::clone(&shared);
        thread::spawn(move || {
            if let Err(e) = handle_conn(stream, shared) {
                eprintln!("connection error: {e}");
            }
        });
    }
    Ok(())
}

type Shared = Arc<Mutex<Node>> is the key idea for multi-threaded state. Multiple peer connections may want to read or modify the same node at the same time. Two wrappers solve that together: Arc (atomically reference-counted) lets several threads share ownership of the same Node, and Mutex ensures only one thread mutates it at a time. Together, Arc<Mutex<Node>> is “one node, safely shared across threads.”

serve binds a TcpListener to an address (? propagates a bind failure, e.g. the port is taken) and loops forever over listener.incoming() — each iteration is a new peer connecting. For every connection we Arc::clone(&shared) (cheap: it just bumps the reference count, it does not copy the node) and hand it to a freshly spawned thread. thread::spawn runs handle_conn concurrently, so one slow peer never blocks the others. The move keyword moves the cloned Arc and the stream into the thread’s closure, giving it ownership of what it needs.

Reading messages off a connection

fn handle_conn(stream: TcpStream, shared: Shared) -> std::io::Result<()> {
    let mut reader = BufReader::new(stream.try_clone()?);
    let mut writer = stream;
    let mut line = String::new();
    while reader.read_line(&mut line)? > 0 {
        if let Ok(msg) = serde_json::from_str::<Message>(line.trim()) {
            let reply = shared.lock().unwrap().handle(msg);
            if let Some(reply) = reply {
                send_line(&mut writer, &reply)?;
            }
        }
        line.clear();
    }
    Ok(())
}

A TcpStream is a two-way pipe. We try_clone() it so we can read and write through the same connection independently — one handle wrapped in a BufReader for reading, the other kept as the writer. BufReader buffers the stream so we can call read_line, which reads bytes up to (and including) the next newline. That newline is exactly our message delimiter.

The while reader.read_line(&mut line)? > 0 loop runs until read_line returns 0, meaning end of stream — the peer hung up. For each line we try to parse it into a Message (if let Ok(msg) = ... quietly ignores garbage lines), then shared.lock().unwrap() takes the mutex so we can call handle on the node. The lock guard is released at the end of that statement, so we hold it for the minimum time. If handle produced a reply, we write it back. Finally line.clear() empties the buffer before reading the next message.

Writing one message

fn send_line(writer: &mut impl Write, msg: &Message) -> std::io::Result<()> {
    let mut json = serde_json::to_string(msg).expect("message serializes");
    json.push('\n');
    writer.write_all(json.as_bytes())
}

This is the mirror of reading: serialize to compact JSON (note to_string, not the pretty form), append the '\n' delimiter, and write the bytes. impl Write means “anything you can write bytes to” — so the same helper serves both inbound replies and outbound dials.

Dialing a peer

/// Dial a peer, send one message, and read one reply line.
pub fn send(addr: &str, msg: &Message) -> std::io::Result<Option<Message>> {
    let stream = TcpStream::connect(addr)?;
    let mut writer = stream.try_clone()?;
    send_line(&mut writer, msg)?;
    let mut reader = BufReader::new(stream);
    let mut line = String::new();
    if reader.read_line(&mut line)? > 0 {
        let reply = serde_json::from_str::<Message>(line.trim())
            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
        Ok(Some(reply))
    } else {
        Ok(None)
    }
}

Where serve is the server side, send is the client side: it opens a connection with TcpStream::connect, sends one message, and reads exactly one reply line back. Same BufReader + read_line pattern, same JSON-per-line protocol — just initiated from our end. The reply comes back as Option<Message> (None if the peer closed without answering).

Syncing missing blocks

/// Pull any blocks we're missing from a peer into our chain.
pub fn sync_from(peer: &str, shared: &Shared) -> std::io::Result<usize> {
    let from = shared.lock().unwrap().chain.blocks.len() as u64;
    let reply = send(peer, &Message::GetBlocks { from_height: from })?;
    if let Some(Message::Blocks(blocks)) = reply {
        let mut node = shared.lock().unwrap();
        let before = node.chain.blocks.len();
        for block in blocks {
            let _ = node.chain.add_block(block);
        }
        Ok(node.chain.blocks.len() - before)
    } else {
        Ok(0)
    }
}

This is the function that brings a fresh node up to date. It locks the node just long enough to read our current height (from), then sends a GetBlocks { from_height: from } to the peer. If the reply is a Blocks batch, it locks again, adds each block, and returns how many were actually added (the height difference). Note the lock is taken, released, then taken again — we don’t hold it across the network round-trip, so other threads can keep working while we wait on the wire.

This protocol is intentionally minimal

Real nodes do far more: peer discovery, handshakes, headers-first sync, choosing the most-work chain among forks, banning misbehaving peers. We’ve stripped it to the essence — request, respond, relay — so the moving parts stay visible. The shape, though, is the real shape.

Try it

Open two terminals. In the first, start a node:

cargo run -- serve 127.0.0.1:9000

In the second, start another node and point it at the first as a peer:

cargo run -- serve 127.0.0.1:9001 127.0.0.1:9000

Watch the second node sync_from the first — pulling the blocks it’s missing — and then relay each block it mines back to its peer as a NewBlock message (the first node logs each one it accepts). Two independent processes, agreeing on one chain over a real network socket. That’s a blockchain.

Check your understanding

1. Why is the message-handling logic (Node::handle) kept in a separate module from the TCP code in net.rs?
2. What do Arc and Mutex each contribute in Arc<Mutex<Node>>, and why do we need both?
3. How does a receiver know where one message ends and the next begins on a TCP stream?
4. In handle_conn, why does the loop use while reader.read_line(&mut line)? > 0 — what does a return value of 0 signify?
5. In sync_from, why do we release the lock before sending the network request and re-acquire it afterward, instead of holding it the whole time?

Show answers

1. So the logic can be tested without opening sockets — Node::handle decides how to respond to each message as pure in-memory state, while net.rs deals only with TCP. Separating the brain from the wiring keeps each piece simple and independently testable.
2. Arc (atomically reference-counted) lets several threads share ownership of the same Node; Mutex ensures only one thread mutates it at a time (no data races). You need both: Arc alone wouldn’t prevent concurrent mutation, and Mutex alone can’t be shared across threads safely.
3. By a delimiter: the protocol is one JSON object per line, so a newline marks where one message ends and the next begins. The receiver uses read_line to read up to and including that newline.
4. read_line returns the number of bytes read, and a return of 0 means end of stream — the peer hung up. So the loop runs as long as bytes keep arriving and exits cleanly when the connection closes.
5. Because network I/O is slow and the lock gives exclusive access to the shared Node; holding it across the round-trip would block every other thread that needs the node. Releasing it before send and re-acquiring after lets other threads keep working while we wait on the wire — we only hold the lock for the quick reads/writes.