Build a Distributed Task Queue in Rust

Introduction

30 Second Summary

Every time you order food through an app, your request doesn't go straight to the kitchen. It enters a queue where it's distributed, tracked, and confirmed. Task queues are the invisible backbone of every real-time system you use daily.

In this project, you will build a distributed task queue from scratch in Rust. You'll create a broker that manages tasks, producers that submit work, and consumers that process it. All three communicate over TCP, UDP, and HTTP.

What You'll Build

A single-node message broker with a custom wire protocol, token-based authentication, UDP heartbeat monitoring, and an HTTP status API.

By the end of this project, you'll have:

• A broker binary that simultaneously listens on three ports (TCP for task commands, UDP for heartbeats, HTTP for status) using Tokio async tasks.
• A custom wire protocol using newline-delimited JSON with commands like AUTH, PUBLISH, SUBSCRIBE, and ACK.
• A UDP heartbeat system where consumers prove liveness and the broker detects dead consumers after a timeout.
• Secret Mission: consumer groups with automatic task redelivery when a consumer's heartbeat times out.

Are there any prerequisites?

This project assumes you already know Rust fundamentals including ownership, borrowing, async/await, Arc/Mutex, channels, traits, enums with data, error handling, and serde serialization. You should have a working Rust toolchain and VS Code installed on your machine.

Set Up the Cargo Workspace

A distributed task queue needs multiple programs running at the same time. You need a broker to manage the queue, producers to publish tasks, and consumers to process them.

In Rust, a Cargo workspace lets you keep all these programs in one repository while sharing common code between them. In this step, you will scaffold a workspace with four crates and wire up the dependencies each one needs.

In this step, get ready to:

• Create a Cargo workspace with four crates (broker, producer, consumer, shared).
• Add async runtime and serialization dependencies to each crate.
• Verify the whole workspace compiles.

Scaffold the Cargo workspace

Before writing any networking code, you need a project structure that supports multiple binaries sharing a common library. A Cargo workspace gives you exactly that.

• Open your terminal.
• Move to your Desktop by running this command:

cd ~/Desktop

• Create a new project directory and move into it by running these commands:

mkdir task-queue
cd task-queue

• Confirm the folder exists by running:

ls ~/Desktop | grep task-queue

You should see task-queue in the output.

• Scaffold the four crates by running these commands:

cargo new broker
cargo new producer
cargo new consumer
cargo new --lib shared

What do these commands do?

Each cargo new command creates a subdirectory with its own Cargo.toml and src/ folder. The first three create binary crates (programs you can run).

The --lib flag on the last one creates a library crate. This is shared code that other crates import.

You should see a Created binary (application) message for each binary crate and a Created library message for the shared crate.

Seeing a permission error?

Make sure you successfully ran cd ~/Desktop and mkdir task-queue before scaffolding. You need write permission in the current directory.

Help me troubleshoot this permission error

• Open the project in VS Code by running:

code .

VS Code opens with the task-queue folder loaded in the Explorer sidebar. You should see the four subdirectories: broker, consumer, producer, and shared.

Now you need a workspace root file that tells Cargo these four crates belong together.

• Create a new file called Cargo.toml in the root of the task-queue folder (not inside any subdirectory).
• Add the workspace definition by pasting this content:

[workspace]
members = ["broker", "producer", "consumer", "shared"]
resolver = "2"

What does this file do?

The [workspace] section tells Cargo that this directory is a workspace root. The members array lists which subdirectories are part of the workspace.

The resolver = "2" line uses Cargo's newer dependency resolver, which handles feature unification more correctly for workspaces.

• Save Cargo.toml.

Next, replace the default scaffolded code in each crate's entry file with minimal stubs.

• Open broker/src/main.rs and replace its entire content with:

fn main() {
    println!("broker");
}

• Open producer/src/main.rs and replace its entire content with:

fn main() {
    println!("producer");
}

• Open consumer/src/main.rs and replace its entire content with:

fn main() {
    println!("consumer");
}

• Open shared/src/lib.rs and replace its entire content with:

// shared library

• Save all four files.

Save all files at once

Press Ctrl+K then S in VS Code to save all open files at once.

Add dependencies to each crate

Each crate needs libraries for async networking, serialization, and unique identifiers. The broker needs all of them, the producer and consumer need most of them, and the shared library only needs the serialization and ID crates.

• Open broker/Cargo.toml and replace its entire content with:

[package]
name = "broker"
version = "0.1.0"
edition = "2021"

[dependencies]
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
uuid = { version = "1", features = ["v4"] }
chrono = { version = "0.4", features = ["serde"] }
shared = { path = "../shared" }

What are these dependencies?

• tokio is the async runtime. The "full" feature enables TCP, UDP, timers, and multi-threaded scheduling.
• serde and serde_json handle serializing Rust structs to JSON and back. The "derive" feature lets you annotate structs with #[derive(Serialize, Deserialize)].
• uuid generates unique task and consumer IDs. The "v4" feature enables random UUID generation.
• chrono provides timestamps for heartbeats and logging. The "serde" feature lets timestamps serialize to JSON.
• shared = { path = "../shared" } is a path dependency pointing at the shared library crate in the workspace.

• Save broker/Cargo.toml.
• Open producer/Cargo.toml and replace its entire content with:

[package]
name = "producer"
version = "0.1.0"
edition = "2021"

[dependencies]
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
uuid = { version = "1", features = ["v4"] }
shared = { path = "../shared" }

Why no chrono here?

Producers only publish tasks. They do not track heartbeat timestamps, so they do not need chrono. Keeping dependencies minimal speeds up compile times.

• Save producer/Cargo.toml.
• Open consumer/Cargo.toml and replace its entire content with:

[package]
name = "consumer"
version = "0.1.0"
edition = "2021"

[dependencies]
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
uuid = { version = "1", features = ["v4"] }
shared = { path = "../shared" }

• Save consumer/Cargo.toml.
• Open shared/Cargo.toml and replace its entire content with:

[package]
name = "shared"
version = "0.1.0"
edition = "2021"

[dependencies]
serde = { version = "1", features = ["derive"] }
serde_json = "1"
uuid = { version = "1", features = ["v4"] }
chrono = { version = "0.4", features = ["serde"] }

Why does shared have chrono?

The shared library will define protocol types that include timestamps (like heartbeat messages). Since the struct definitions live in shared, the chrono dependency belongs here. The broker re-exports it transitively through the shared path dependency.

• Save shared/Cargo.toml.

✔️ Awesome, I've got everything!

Great. Double check all files are saved before moving on to the build step.

ⓧ I'd like to double check the full code

Here is every file you should have at this point. Compare yours against these:

[workspace]
members = ["broker", "producer", "consumer", "shared"]
resolver = "2"

[package]
name = "broker"
version = "0.1.0"
edition = "2021"

[dependencies]
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
uuid = { version = "1", features = ["v4"] }
chrono = { version = "0.4", features = ["serde"] }
shared = { path = "../shared" }

fn main() {
    println!("broker");
}

[package]
name = "producer"
version = "0.1.0"
edition = "2021"

[dependencies]
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
uuid = { version = "1", features = ["v4"] }
shared = { path = "../shared" }

fn main() {
    println!("producer");
}

[package]
name = "consumer"
version = "0.1.0"
edition = "2021"

[dependencies]
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
uuid = { version = "1", features = ["v4"] }
shared = { path = "../shared" }

fn main() {
    println!("consumer");
}

[package]
name = "shared"
version = "0.1.0"
edition = "2021"

[dependencies]
serde = { version = "1", features = ["derive"] }
serde_json = "1"
uuid = { version = "1", features = ["v4"] }
chrono = { version = "0.4", features = ["serde"] }

// shared library

Verify the project compiles

With all the files in place, Cargo should be able to resolve every dependency and compile the workspace. This confirms that the workspace structure, path dependencies, and feature flags are all correct.

• Open the integrated terminal in VS Code by pressing Ctrl+`.
• Build the entire workspace by running:

cargo build

Cargo downloads all dependencies and compiles each crate. The first build takes a minute or two while it fetches and compiles Tokio and its dependency tree.

You should see output ending with lines like Compiling broker, Compiling producer, Compiling consumer, and a final Finished line with no errors.

Seeing a compile error?

Check that your root Cargo.toml only has a [workspace] section and no [package] section. Workspace roots are not packages themselves.

Also verify that each crate's shared = { path = "../shared" } line points up one directory. A typo in the path causes an unresolved dependency error.

Help me debug my workspace compilation error

• Confirm you can run one of the binaries by executing:

cargo run --bin broker

You should see broker printed to the terminal.

Your workspace is ready. All four crates compile, share a common library, and can be built with a single command. Next up, you will design the wire protocol that the broker, producers, and consumers use to communicate over TCP.

Design the Wire Protocol

Your workspace compiles and you have four crates ready to go. Now you need a common language that the broker, producers, and consumers can all speak over TCP.

Every networked system needs a wire protocol that defines what messages look like on the network and how each side knows where one message ends and the next begins. In this step, you will define the message types and build the framing layer that all three binaries will share.

In this step, get ready to:

• Define the protocol types (Task, Command, Response) with serde serialization.
• Add async I/O dependencies to the shared crate.
• Implement newline-delimited JSON framing helper functions.

Define the protocol types

The shared crate will hold every type that crosses the network boundary. Producers send commands, the broker sends responses, and tasks flow between them.

• In VS Code, open shared/src/lib.rs from the Explorer sidebar.
• Delete the existing comment and replace the entire file contents with this code:

use serde::{Deserialize, Serialize};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};

// A task that flows through the queue
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Task {
    pub id: String,
    pub payload: String,
    pub redelivery_count: u32,
}

// Commands that clients send to the broker
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum Command {
    Auth { token: String },
    Publish { payload: String },
    Subscribe { consumer_id: String },
    Ack { task_id: String },
}

// Responses the broker sends back to clients
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum Response {
    Ok,
    Error { message: String },
    Task(Task),
}

What does this code do?

This defines the complete vocabulary of your wire protocol.

• Task holds a unique ID, a payload string, and a redelivery counter for tracking retry attempts.
• Command is everything a client (producer or consumer) can say to the broker: authenticate, publish a task, subscribe for tasks, or acknowledge completion.
• Response is everything the broker can say back: success, error with a reason, or a task delivery.
• The #[serde(tag = "type")] attribute tells serde to include a "type" field in the JSON so the receiver knows which variant it got.

Why newline-delimited JSON?

When you send data over TCP, the stream is just a continuous flow of bytes with no built-in message boundaries. You need a way to tell where one message ends and the next begins.

Newline-delimited JSON solves this by using a newline character as the separator. Each line is one complete JSON object. The receiver reads until it hits a newline, then deserializes that line as a complete message.

Add async dependencies to the shared crate

The helper functions need async I/O traits from Tokio and flexible error handling from the anyhow crate.

• Open shared/Cargo.toml in the Explorer sidebar.
• Add tokio and anyhow to the [dependencies] section so it looks like this:

[dependencies]
serde = { version = "1", features = ["derive"] }
serde_json = "1"
uuid = { version = "1", features = ["v4"] }
chrono = { version = "0.4", features = ["serde"] }
tokio = { version = "1", features = ["full"] }
anyhow = "1"

What do these dependencies do?

• tokio with the "full" feature gives you async networking and I/O utilities like BufReader and AsyncWriteExt.
• anyhow provides a flexible error type that wraps any error without requiring you to define custom error enums for every function.

• Save shared/Cargo.toml.

Implement the framing helper functions

With the types and dependencies in place, you can now write the two functions that every binary will call to read and write messages on a TCP connection.

• Switch back to shared/src/lib.rs.
• Add the following two functions below the Response enum, at the bottom of the file:

pub async fn read_command(reader: &mut BufReader<OwnedReadHalf>) -> anyhow::Result<Command> {
    let mut line = String::new();
    // Read bytes until we hit a newline (one complete JSON message)
    reader.read_line(&mut line).await?;
    // Deserialize the JSON line into a Command
    let cmd = serde_json::from_str(line.trim())?;
    Ok(cmd)
}

pub async fn write_response(writer: &mut OwnedWriteHalf, response: &Response) -> anyhow::Result<()> {
    // Serialize the Response to JSON
    let mut json = serde_json::to_string(response)?;
    // Append the newline delimiter so the receiver knows this message is complete
    json.push('\n');
    writer.write_all(json.as_bytes()).await?;
    Ok(())
}

What does this code do?

• read_command reads from the TCP stream until it encounters a newline. That single line is one JSON message, which it deserializes into a Command.
• write_response serializes a Response to JSON, appends a newline character, and writes the full byte sequence to the stream.
• Both functions use OwnedReadHalf and OwnedWriteHalf from Tokio. These are the two halves of a TCP connection after calling into_split(), which lets you read and write from separate async tasks without locking.

• Save shared/src/lib.rs.
• Verify the project compiles by running this in your terminal:

cargo build

You should see output ending with a line that says the build completed successfully, with no errors.

Seeing a compile error?

• If you see unresolved import `tokio`, make sure you added tokio to shared/Cargo.toml (not just the broker's Cargo.toml) and saved the file.
• If you see failed to resolve: use of undeclared crate or module `anyhow`, confirm anyhow = "1" is in shared/Cargo.toml's [dependencies] section.
• If you see a serde error about Task inside the Response enum, ensure the Task struct is defined above the Response enum in the file.

Help me debug a compile error in my shared crate's wire protocol types.

✔️ Awesome, I've got everything!

Great. Double check you have saved both shared/Cargo.toml and shared/src/lib.rs.

ⓧ I'd like to double check the full code

Here is the complete shared/src/lib.rs at the end of this step:

use serde::{Deserialize, Serialize};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Task {
    pub id: String,
    pub payload: String,
    pub redelivery_count: u32,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum Command {
    Auth { token: String },
    Publish { payload: String },
    Subscribe { consumer_id: String },
    Ack { task_id: String },
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum Response {
    Ok,
    Error { message: String },
    Task(Task),
}

pub async fn read_command(reader: &mut BufReader<OwnedReadHalf>) -> anyhow::Result<Command> {
    let mut line = String::new();
    reader.read_line(&mut line).await?;
    let cmd = serde_json::from_str(line.trim())?;
    Ok(cmd)
}

pub async fn write_response(writer: &mut OwnedWriteHalf, response: &Response) -> anyhow::Result<()> {
    let mut json = serde_json::to_string(response)?;
    json.push('\n');
    writer.write_all(json.as_bytes()).await?;
    Ok(())
}

And here is the complete shared/Cargo.toml:

[package]
name = "shared"
version = "0.1.0"
edition = "2021"

[dependencies]
serde = { version = "1", features = ["derive"] }
serde_json = "1"
uuid = { version = "1", features = ["v4"] }
chrono = { version = "0.4", features = ["serde"] }
tokio = { version = "1", features = ["full"] }
anyhow = "1"

Your wire protocol is locked in. Next up, you will use these types to build the broker's TCP listener and handle live connections.

Build the TCP Listener and Connection Handler

The broker is the central hub of your task queue. Every producer and consumer connects to it over TCP, so the broker needs to listen for incoming connections and handle them concurrently.

In this step, you'll set up the async runtime with Tokio, bind a TcpListener, and spawn a separate task for each client that connects. Each task reads commands from the wire and echoes back a response.

In this step, get ready to:

• Bind a TCP listener to accept incoming connections.
• Spawn an async task for each connected client.
• Parse incoming commands and echo back responses.

Add the anyhow dependency

The broker's main function will return errors using the anyhow crate, which gives you a convenient Result type that can hold any error without defining custom error types.

• Open broker/Cargo.toml and add anyhow to the [dependencies] section by adding this line:

anyhow = "1"

• Save broker/Cargo.toml.

Set up the async runtime and TCP listener

The #[tokio::main] macro transforms your main function into an async entry point. Inside it, you'll bind a TcpListener and loop forever, accepting new connections.

• Replace the entire contents of broker/src/main.rs with the following code:

use shared::{read_command, write_response, Response};
use tokio::io::BufReader;
use tokio::net::TcpListener;

// The address and port the broker listens on for TCP connections
const TCP_ADDR: &str = "127.0.0.1:7878";

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // Bind to the TCP address so clients can connect
    let tcp_listener = TcpListener::bind(TCP_ADDR).await?;
    println!("Broker TCP listening on {}", TCP_ADDR);

    loop {
        // Wait for the next client to connect
        let (stream, addr) = tcp_listener.accept().await?;
        println!("New connection from {}", addr);

        // Spawn a new async task to handle this connection independently
        tokio::spawn(async move {
            if let Err(e) = handle_connection(stream).await {
                eprintln!("Connection error: {}", e);
            }
        });
    }
}

What does this code do?

TcpListener::bind(TCP_ADDR).await tells the OS to reserve port 7878 on localhost for this process. Any client that connects to 127.0.0.1:7878 will be accepted here.

The loop calls accept() which pauses (asynchronously) until a new client arrives. When one does, tokio::spawn creates a lightweight task to handle that client. This means multiple clients can be served at the same time without blocking each other.

Handle each client connection

Each spawned task needs to read incoming commands from the client and send back responses. You'll split the TCP stream into a read half and a write half so you can read and write independently.

• Add the handle_connection function below the main function in broker/src/main.rs:

async fn handle_connection(stream: tokio::net::TcpStream) -> anyhow::Result<()> {
    // Split the stream so we can read and write independently
    let (read_half, mut write_half) = stream.into_split();
    let mut reader = BufReader::new(read_half);

    loop {
        // Read the next newline-delimited JSON command
        let cmd = read_command(&mut reader).await?;
        println!("Received: {:?}", cmd);

        // For now, respond with Ok to every command
        write_response(&mut write_half, &Response::Ok).await?;
    }
}

What does this code do?

stream.into_split() gives you an owned read half and write half. The read half goes into a BufReader so you can read line by line. The write half is used to send responses back.

The inner loop calls read_command (from your shared crate) to read one line of JSON and deserialize it into a Command. It then echoes back Response::Ok for every command. When the client disconnects, read_command returns an error, ending the loop.

• Save broker/src/main.rs.
• Verify it compiles by running:

cargo build --bin broker

You should see a successful compilation with no errors.

Seeing a compilation error?

• If you see an unresolved import error for anyhow, make sure you added anyhow = "1" to broker/Cargo.toml under [dependencies] and saved the file.
• If the error mentions handle_connection, make sure you added the function below main in the same file.
• Check that your use statements at the top import read_command, write_response, and Response from shared.

Still stuck? Help me debug a compilation error in my broker

Now test it live. Start the broker by running:

cargo run --bin broker

You should see Broker TCP listening on 127.0.0.1:7878 printed in your terminal.

• Open a second terminal tab and send a test command by running:

echo '{"type":"Auth","token":"test123"}' | nc 127.0.0.1 7878

In the broker terminal, you should see New connection from 127.0.0.1:... followed by Received: Auth { token: "test123" }.

What is nc?

nc (netcat) is a command-line tool for reading and writing data over TCP connections. Here you're using it to send a single JSON line to the broker and see the response. It's a quick way to test TCP servers without building a full client.

• Press Ctrl+C in the broker terminal to stop it.

Don't see the connection log?

• Make sure the broker is still running in the first terminal before sending the test command from the second terminal.
• If nc reports connection refused, the broker may have crashed. Check the first terminal for error messages.
• If nc is not found on your system, try ncat instead (it ships with nmap).

Still stuck? Help me test my TCP broker connection

✔️ Awesome, I've got everything!

Great. Double check you've saved broker/src/main.rs and confirmed the broker starts without errors.

ⓧ I'd like to double check the full code

use shared::{read_command, write_response, Response};
use tokio::io::BufReader;
use tokio::net::TcpListener;

const TCP_ADDR: &str = "127.0.0.1:7878";

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let tcp_listener = TcpListener::bind(TCP_ADDR).await?;
    println!("Broker TCP listening on {}", TCP_ADDR);

    loop {
        let (stream, addr) = tcp_listener.accept().await?;
        println!("New connection from {}", addr);
        tokio::spawn(async move {
            if let Err(e) = handle_connection(stream).await {
                eprintln!("Connection error: {}", e);
            }
        });
    }
}

async fn handle_connection(stream: tokio::net::TcpStream) -> anyhow::Result<()> {
    let (read_half, mut write_half) = stream.into_split();
    let mut reader = BufReader::new(read_half);

    loop {
        let cmd = read_command(&mut reader).await?;
        println!("Received: {:?}", cmd);
        write_response(&mut write_half, &Response::Ok).await?;
    }
}

Your broker can now accept TCP connections and read commands off the wire. Next up, you'll add authentication so only clients with a valid token can interact with the queue.

Implement Token-Based Authentication

Your broker accepts TCP connections and parses commands from any client that connects. But right now, there's no gate at the door. Anyone who connects can issue queue operations immediately.

In a real distributed system, every connection must prove its identity before the broker trusts it. In this step, you'll add token-based authentication that demands an AUTH command as the very first message on any connection. If the token is valid, the connection proceeds. If not, the broker drops it immediately.

In this step, get ready to:

• Define a set of valid tokens in the broker.
• Require AUTH as the first message on every new connection.
• Test authentication with valid and invalid tokens.

Define the valid token set

The broker needs to know which tokens are legitimate. For now, you'll hardcode them as a static slice of string references. This keeps things simple while still teaching the authentication handshake pattern.

• In broker/src/main.rs, find the import line at the top of the file:

use shared::{read_command, write_response, Response};

• Add Command to the import list so the line now looks like this:

use shared::{read_command, write_response, Command, Response};

• Next, add the VALID_TOKENS constant directly after the TCP_ADDR line by copying this code below it:

const VALID_TOKENS: &[&str] = &["secret-producer-token", "secret-consumer-token"];

What does this constant do?

VALID_TOKENS is a static slice of string references. When a client connects and sends an AUTH command, the broker checks whether the provided token exists in this slice.

In a production system you'd load tokens from a database or environment variables. Hardcoding them here lets you focus on the authentication flow without extra infrastructure.

Enforce authentication on new connections

Every TCP connection must authenticate before sending any other commands. The handle_connection function needs to read the first message, verify it's an AUTH with a valid token, and either proceed or drop the connection.

• Replace the entire handle_connection function in broker/src/main.rs with the following code:

async fn handle_connection(stream: tokio::net::TcpStream) -> anyhow::Result<()> {
    let (read_half, mut write_half) = stream.into_split();
    let mut reader = BufReader::new(read_half);

    // Require AUTH as first message
    let first_cmd = read_command(&mut reader).await?;
    match first_cmd {
        Command::Auth { token } if VALID_TOKENS.contains(&token.as_str()) => {
            write_response(&mut write_half, &Response::Ok).await?;
            println!("Client authenticated with token: {}", token);
        }
        _ => {
            write_response(&mut write_half, &Response::Error { message: "Unauthorized".to_string() }).await?;
            return Ok(());
        }
    }

    loop {
        let cmd = read_command(&mut reader).await?;
        println!("Received: {:?}", cmd);
        write_response(&mut write_half, &Response::Ok).await?;
    }
}

What does this code do?

The function reads the very first command from the connection and uses a match expression with a guard clause to handle two cases.

• Command::Auth { token } if VALID_TOKENS.contains(&token.as_str()) matches only when the command is Auth AND the token is in the valid set. The broker sends Response::Ok and continues to the main command loop.
• The wildcard _ arm catches everything else: a non-Auth command, or an Auth with an invalid token. The broker sends Response::Error with the message "Unauthorized" and returns immediately, which drops the TCP connection.

• Save broker/src/main.rs.
• Verify the project compiles by running:

cargo build

• You should see Compiling broker followed by Finished with no errors.

Seeing a compile error?

• If you see "cannot find value Command in this scope", make sure you added Command to the import line at the top of the file.
• If you see "cannot find value VALID_TOKENS", make sure the constant is defined outside any function, between the TCP_ADDR line and the #[tokio::main] attribute.
• Check that you replaced the entire handle_connection function (from the async fn line through the closing }), not just the body inside it.

Why won't my broker compile after adding authentication?

Test authentication with netcat

You'll use nc (netcat) to send raw JSON lines to the broker and observe the responses. This simulates what the producer and consumer binaries will do later.

• Start the broker in your first terminal by running:

cargo run --bin broker

• You should see Broker TCP listening on 127.0.0.1:7878 printed to the terminal.
• Open a second terminal tab or pane.
• Test a valid token by running:

echo '{"type":"Auth","token":"secret-producer-token"}' | nc -N 127.0.0.1 7878

• You should see {"type":"Ok"} printed back in your second terminal. In the broker terminal, you should see a log line confirming the authentication.
• Now test an invalid token by running:

echo '{"type":"Auth","token":"wrong-token"}' | nc -N 127.0.0.1 7878

• You should see {"type":"Error","message":"Unauthorized"} printed back. The broker rejected the connection.
• Finally, test sending a non-Auth command first:

echo '{"type":"Publish","payload":"hello"}' | nc -N 127.0.0.1 7878

• You should see the same {"type":"Error","message":"Unauthorized"} response. Any command that isn't a valid Auth gets rejected.

Why does the wildcard arm catch both cases?

The _ arm in the match expression catches two distinct failure scenarios: a Publish/Subscribe/Ack command sent before authenticating, and an Auth command with an invalid token.

The guard clause if VALID_TOKENS.contains(&token.as_str()) means that an Auth with a bad token fails the first arm and falls through to the wildcard. This gives you a single rejection path for all unauthorized connections.

Not seeing a response from netcat?

• Make sure the broker is still running in the first terminal. If it crashed, check for error messages and restart it.
• If nc hangs without printing anything, try pressing Ctrl+C and check that the JSON in your echo command has no typos. A malformed JSON line causes a parse error on the broker side.
• If the -N flag is not recognized, your system may have the traditional netcat variant. Try replacing -N with -q 1 instead.

Why isn't my broker responding to netcat connections?

• Stop the broker by pressing Ctrl+C in the first terminal when you're done testing.

✔️ Awesome, I've got everything!

Great. Double check you've saved broker/src/main.rs and that cargo build compiles cleanly.

ⓧ I'd like to double check the full code

Here's the complete broker/src/main.rs at this point:

use shared::{read_command, write_response, Command, Response};
use tokio::io::BufReader;
use tokio::net::TcpListener;

const TCP_ADDR: &str = "127.0.0.1:7878";
const VALID_TOKENS: &[&str] = &["secret-producer-token", "secret-consumer-token"];

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let tcp_listener = TcpListener::bind(TCP_ADDR).await?;
    println!("Broker TCP listening on {}", TCP_ADDR);

    loop {
        let (stream, addr) = tcp_listener.accept().await?;
        println!("New connection from {}", addr);
        tokio::spawn(async move {
            if let Err(e) = handle_connection(stream).await {
                eprintln!("Connection error: {}", e);
            }
        });
    }
}

async fn handle_connection(stream: tokio::net::TcpStream) -> anyhow::Result<()> {
    let (read_half, mut write_half) = stream.into_split();
    let mut reader = BufReader::new(read_half);

    // Require AUTH as first message
    let first_cmd = read_command(&mut reader).await?;
    match first_cmd {
        Command::Auth { token } if VALID_TOKENS.contains(&token.as_str()) => {
            write_response(&mut write_half, &Response::Ok).await?;
            println!("Client authenticated with token: {}", token);
        }
        _ => {
            write_response(&mut write_half, &Response::Error { message: "Unauthorized".to_string() }).await?;
            return Ok(());
        }
    }

    loop {
        let cmd = read_command(&mut reader).await?;
        println!("Received: {:?}", cmd);
        write_response(&mut write_half, &Response::Ok).await?;
    }
}

Your broker now gates every connection behind authentication. No client can publish or subscribe without first proving it holds a valid token. Next up, you'll build the in-memory queue so authenticated producers can push tasks and consumers can receive them.

Build the In-Memory Queue and Task Dispatch

Your broker now authenticates every connection, but it does not actually do anything with the commands that come in afterwards.

In this step, you will build the core of the message broker: an in-memory queue that holds tasks, dispatches them to subscribed consumers, and tracks acknowledgments. This turns your broker from a connection gatekeeper into a real task router.

In this step, get ready to:

• Create shared broker state with a task queue and consumer registry.
• Handle PUBLISH, SUBSCRIBE, and ACK commands in the broker.
• Build the producer and consumer binaries and test end-to-end dispatch.

Define the shared broker state

Multiple connections need to access the same queue and consumer list concurrently. In async Rust, the standard pattern is to wrap shared data in Arc<RwLock<T>>.

• In broker/src/main.rs, replace the entire file contents with the following code:

use shared::{read_command, write_response, Command, Response, Task};
use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
use tokio::io::BufReader;
use tokio::net::TcpListener;
use tokio::sync::{mpsc, RwLock};
use uuid::Uuid;

const TCP_ADDR: &str = "127.0.0.1:7878";
const VALID_TOKENS: &[&str] = &["secret-producer-token", "secret-consumer-token"];

// Holds the queue, registered consumers, and metrics
#[derive(Default)]
struct BrokerState {
    queue: VecDeque<Task>,
    consumers: HashMap<String, mpsc::Sender<Task>>,
    tasks_processed: u64,
}

// Type alias for the shared state handle
type SharedState = Arc<RwLock<BrokerState>>;

What does this code do?

BrokerState is a plain struct that holds three things: a VecDeque of pending tasks (a double-ended queue), a HashMap mapping consumer IDs to their mpsc::Sender<Task> channels, and a counter for processed tasks.

Arc gives thread-safe reference counting so multiple tasks can own a handle to the same data. RwLock from Tokio allows multiple readers or one exclusive writer, and its guards can be held across .await points.

Now add the updated main function below the type alias. This creates the shared state and passes a clone to each spawned connection task:

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let state: SharedState = Arc::new(RwLock::new(BrokerState::default()));
    let tcp_listener = TcpListener::bind(TCP_ADDR).await?;
    println!("Broker TCP listening on {}", TCP_ADDR);

    loop {
        let (stream, addr) = tcp_listener.accept().await?;
        println!("New connection from {}", addr);
        // Clone the Arc handle for this connection's task
        let state = Arc::clone(&state);
        tokio::spawn(async move {
            if let Err(e) = handle_connection(stream, state).await {
                eprintln!("Connection error: {}", e);
            }
        });
    }
}

What changed from before?

Two things changed. First, a SharedState is created at the top of main using BrokerState::default().

Second, Arc::clone(&state) creates a cheap reference-counted handle for each spawned task. The handle_connection function now takes this handle as a second argument so it can read and write the queue.

Handle PUBLISH, SUBSCRIBE, and ACK commands

The previous handle_connection echoed a generic Response::Ok for every command after auth. Now it needs to do real work: create tasks, register consumers, and track acknowledgments.

• Add the following handle_connection function below the main function in broker/src/main.rs:

async fn handle_connection(stream: tokio::net::TcpStream, state: SharedState) -> anyhow::Result<()> {
    let (read_half, mut write_half) = stream.into_split();
    let mut reader = BufReader::new(read_half);

    let first_cmd = read_command(&mut reader).await?;
    match first_cmd {
        Command::Auth { token } if VALID_TOKENS.contains(&token.as_str()) => {
            write_response(&mut write_half, &Response::Ok).await?;
            println!("Client authenticated");
        }
        _ => {
            write_response(&mut write_half, &Response::Error { message: "Unauthorized".to_string() }).await?;
            return Ok(());
        }
    }

    loop {
        let cmd = read_command(&mut reader).await?;
        match cmd {
            Command::Publish { payload } => {
                let task = Task { id: Uuid::new_v4().to_string(), payload, redelivery_count: 0 };
                let mut s = state.write().await;
                // Try to dispatch to a consumer
                let consumer_ids: Vec<String> = s.consumers.keys().cloned().collect();
                let mut dispatched = false;
                for cid in consumer_ids {
                    if let Some(tx) = s.consumers.get(&cid) {
                        if tx.send(task.clone()).await.is_ok() {
                            dispatched = true;
                            break;
                        } else {
                            s.consumers.remove(&cid);
                        }
                    }
                }
                if !dispatched {
                    s.queue.push_back(task);
                }
                write_response(&mut write_half, &Response::Ok).await?;
            }
            Command::Subscribe { consumer_id } => {
                let (tx, mut rx) = mpsc::channel::<Task>(32);
                {
                    let mut s = state.write().await;
                    // Drain any queued tasks to this consumer
                    while let Some(task) = s.queue.pop_front() {
                        let _ = tx.send(task).await;
                    }
                    s.consumers.insert(consumer_id.clone(), tx);
                }
                write_response(&mut write_half, &Response::Ok).await?;
                // Forward tasks from channel to TCP
                while let Some(task) = rx.recv().await {
                    write_response(&mut write_half, &Response::Task(task)).await?;
                }
            }
            Command::Ack { task_id: _ } => {
                let mut s = state.write().await;
                s.tasks_processed += 1;
                write_response(&mut write_half, &Response::Ok).await?;
            }
            Command::Auth { .. } => {
                write_response(&mut write_half, &Response::Error { message: "Already authenticated".to_string() }).await?;
            }
        }
    }
}

What does this code do?

• PUBLISH: creates a Task with a UUID v4 id, tries to send it directly to a registered consumer via their mpsc channel. If no consumer is available (or all sends fail), the task goes into the VecDeque for later delivery.
• SUBSCRIBE: creates a bounded mpsc channel (capacity 32), drains any tasks already waiting in the queue, registers the sender in the consumers map, then enters a loop forwarding tasks from the channel receiver to the TCP write half.
• ACK: increments the tasks_processed counter. This counter will power the HTTP status endpoint in a later step.
• A repeated AUTH command gets rejected with an error instead of re-authenticating.

Why RwLock instead of Mutex?

Tokio's RwLock allows its guard to be held across .await points without causing a compiler error. In the PUBLISH handler, you hold the write guard while sending on an mpsc channel (an async operation). A standard std::sync::Mutex guard is not Send and would fail to compile in this context.

• Save broker/src/main.rs.
• Verify the broker compiles by running:

cargo build -p broker

You should see a successful compilation with no errors.

Seeing a compile error?

• If you see an error about Task not being found, make sure your first use line imports Task from shared: use shared::{read_command, write_response, Command, Response, Task};.
• If you see errors about Uuid, confirm use uuid::Uuid; is at the top and that broker/Cargo.toml has uuid = { version = "1", features = ["v4"] }.

Help me debug a compile error in my broker

✔️ Awesome, I've got everything!

Great. Double check you have saved broker/src/main.rs and the build succeeded.

ⓧ I'd like to double check the full code

use shared::{read_command, write_response, Command, Response, Task};
use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
use tokio::io::BufReader;
use tokio::net::TcpListener;
use tokio::sync::{mpsc, RwLock};
use uuid::Uuid;

const TCP_ADDR: &str = "127.0.0.1:7878";
const VALID_TOKENS: &[&str] = &["secret-producer-token", "secret-consumer-token"];

#[derive(Default)]
struct BrokerState {
    queue: VecDeque<Task>,
    consumers: HashMap<String, mpsc::Sender<Task>>,
    tasks_processed: u64,
}

type SharedState = Arc<RwLock<BrokerState>>;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let state: SharedState = Arc::new(RwLock::new(BrokerState::default()));
    let tcp_listener = TcpListener::bind(TCP_ADDR).await?;
    println!("Broker TCP listening on {}", TCP_ADDR);

    loop {
        let (stream, addr) = tcp_listener.accept().await?;
        println!("New connection from {}", addr);
        let state = Arc::clone(&state);
        tokio::spawn(async move {
            if let Err(e) = handle_connection(stream, state).await {
                eprintln!("Connection error: {}", e);
            }
        });
    }
}

async fn handle_connection(stream: tokio::net::TcpStream, state: SharedState) -> anyhow::Result<()> {
    let (read_half, mut write_half) = stream.into_split();
    let mut reader = BufReader::new(read_half);

    let first_cmd = read_command(&mut reader).await?;
    match first_cmd {
        Command::Auth { token } if VALID_TOKENS.contains(&token.as_str()) => {
            write_response(&mut write_half, &Response::Ok).await?;
            println!("Client authenticated");
        }
        _ => {
            write_response(&mut write_half, &Response::Error { message: "Unauthorized".to_string() }).await?;
            return Ok(());
        }
    }

    loop {
        let cmd = read_command(&mut reader).await?;
        match cmd {
            Command::Publish { payload } => {
                let task = Task { id: Uuid::new_v4().to_string(), payload, redelivery_count: 0 };
                let mut s = state.write().await;
                // Try to dispatch to a consumer
                let consumer_ids: Vec<String> = s.consumers.keys().cloned().collect();
                let mut dispatched = false;
                for cid in consumer_ids {
                    if let Some(tx) = s.consumers.get(&cid) {
                        if tx.send(task.clone()).await.is_ok() {
                            dispatched = true;
                            break;
                        } else {
                            s.consumers.remove(&cid);
                        }
                    }
                }
                if !dispatched {
                    s.queue.push_back(task);
                }
                write_response(&mut write_half, &Response::Ok).await?;
            }
            Command::Subscribe { consumer_id } => {
                let (tx, mut rx) = mpsc::channel::<Task>(32);
                {
                    let mut s = state.write().await;
                    // Drain any queued tasks to this consumer
                    while let Some(task) = s.queue.pop_front() {
                        let _ = tx.send(task).await;
                    }
                    s.consumers.insert(consumer_id.clone(), tx);
                }
                write_response(&mut write_half, &Response::Ok).await?;
                // Forward tasks from channel to TCP
                while let Some(task) = rx.recv().await {
                    write_response(&mut write_half, &Response::Task(task)).await?;
                }
            }
            Command::Ack { task_id: _ } => {
                let mut s = state.write().await;
                s.tasks_processed += 1;
                write_response(&mut write_half, &Response::Ok).await?;
            }
            Command::Auth { .. } => {
                write_response(&mut write_half, &Response::Error { message: "Already authenticated".to_string() }).await?;
            }
        }
    }
}

Build the producer and consumer binaries

The broker is ready to route tasks. Now you need clients that actually send and receive them.

• Open producer/src/main.rs and replace its contents with the following code:

use shared::{read_command, write_response, Command, Response};
use tokio::io::{AsyncWriteExt, BufReader};
use tokio::net::TcpStream;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let stream = TcpStream::connect("127.0.0.1:7878").await?;
    let (read_half, mut write_half) = stream.into_split();
    let mut reader = BufReader::new(read_half);

    // Authenticate
    let auth = serde_json::to_string(&Command::Auth { token: "secret-producer-token".to_string() })? + "\n";
    write_half.write_all(auth.as_bytes()).await?;
    let resp = read_command_as_response(&mut reader).await?;
    println!("Auth response: {:?}", resp);

    for i in 0..10 {
        let publish = serde_json::to_string(&Command::Publish { payload: format!("task-{}", i) })? + "\n";
        write_half.write_all(publish.as_bytes()).await?;
        let resp = read_command_as_response(&mut reader).await?;
        println!("Publish {}: {:?}", i, resp);
    }
    Ok(())
}

// Helper to read a JSON response line from the broker
async fn read_command_as_response(reader: &mut tokio::io::BufReader<tokio::net::tcp::OwnedReadHalf>) -> anyhow::Result<Response> {
    use tokio::io::AsyncBufReadExt;
    let mut line = String::new();
    reader.read_line(&mut line).await?;
    Ok(serde_json::from_str(line.trim())?)
}

What does the producer do?

The producer connects to the broker on 127.0.0.1:7878, sends an AUTH command with the producer token, then publishes 10 tasks in a loop. Each task has a payload like task-0, task-1, etc.

The read_command_as_response helper reads one newline-delimited JSON response from the broker and deserializes it.

• Save producer/src/main.rs.
• Now open consumer/src/main.rs and replace its contents with the following code:

use shared::{Command, Response, Task};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::TcpStream;
use uuid::Uuid;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let consumer_id = Uuid::new_v4().to_string();
    let stream = TcpStream::connect("127.0.0.1:7878").await?;
    let (read_half, mut write_half) = stream.into_split();
    let mut reader = BufReader::new(read_half);

    // Authenticate
    let auth = serde_json::to_string(&Command::Auth { token: "secret-consumer-token".to_string() })? + "\n";
    write_half.write_all(auth.as_bytes()).await?;
    let mut line = String::new();
    reader.read_line(&mut line).await?;
    println!("Auth: {}", line.trim());

    // Subscribe
    let sub = serde_json::to_string(&Command::Subscribe { consumer_id: consumer_id.clone() })? + "\n";
    write_half.write_all(sub.as_bytes()).await?;
    let mut line = String::new();
    reader.read_line(&mut line).await?;
    println!("Subscribe: {}", line.trim());

    // Receive tasks and ACK
    loop {
        let mut line = String::new();
        reader.read_line(&mut line).await?;
        let resp: Response = serde_json::from_str(line.trim())?;
        if let Response::Task(task) = resp {
            println!("Received task: {} - {}", task.id, task.payload);
            let ack = serde_json::to_string(&Command::Ack { task_id: task.id })? + "\n";
            write_half.write_all(ack.as_bytes()).await?;
        }
    }
}

What does the consumer do?

The consumer generates a random UUID as its ID, connects to the broker, authenticates with the consumer token, and sends a SUBSCRIBE command. After subscribing, it enters a loop reading Response::Task messages and sending ACK for each received task.

• Save consumer/src/main.rs.
• Verify all three binaries compile by running:

cargo build

You should see all four crates compile without errors.

• Test the full flow. Start the broker in one terminal by running:

cargo run -p broker

• Open a second terminal and start a consumer:

cargo run -p consumer

• Open a third terminal and run the producer:

cargo run -p producer

You should see the producer print 10 "Publish" confirmations, and the consumer print 10 "Received task" lines with payloads task-0 through task-9.

Consumer not receiving tasks?

• Make sure you start the consumer before the producer. If the producer runs first and no consumer is subscribed, tasks go into the queue. Start a consumer afterwards and it will drain the queue on subscribe.
• Check that both tokens match: the producer uses secret-producer-token and the consumer uses secret-consumer-token. Both must be in the broker's VALID_TOKENS list.
• If the consumer connects but hangs without printing anything, confirm the broker is running in the first terminal and shows "Client authenticated" for both connections.

Help me debug why my consumer is not receiving tasks from the broker

Your broker is now a functional task router. It accepts tasks from producers and dispatches them to consumers in real time. Next up, you will add UDP heartbeats so the broker can detect when consumers go silent.

Add the UDP Heartbeat System

In a distributed system, connections can drop silently. A consumer might crash, lose network access, or simply hang without closing its TCP connection cleanly.

Heartbeats solve this problem. Consumers periodically send a lightweight "I'm alive" signal over UDP. If the broker stops hearing from a consumer, it knows that consumer is dead. In this step, you'll add a UDP heartbeat system to your broker and consumer.

In this step, get ready to:

• Bind a UDP socket on the broker to receive heartbeat packets.
• Send periodic UDP heartbeats from consumers every 3 seconds.
• Detect and log dead consumers when heartbeats stop arriving.

Add heartbeat tracking state to the broker

Before the broker can track heartbeats, it needs a place to store when it last heard from each consumer. You'll add a HashMap that maps consumer IDs to their last-seen timestamp.

• In broker/src/main.rs, replace the import section at the top of the file with the following:

use chrono::Utc;
use shared::{read_command, write_response, Command, Response, Task};
use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
use tokio::io::BufReader;
use tokio::net::{TcpListener, UdpSocket};
use tokio::sync::{mpsc, RwLock};
use tokio::time::{interval, Duration};
use uuid::Uuid;

What changed in the imports?

You added three new imports. chrono::Utc gives you UTC timestamps for tracking when heartbeats arrive. UdpSocket from tokio's net module lets you receive UDP packets asynchronously.

The interval and Duration types from tokio::time let you run code on a repeating timer.

• Next, find the constants section (TCP_ADDR and VALID_TOKENS) and add two new constants below TCP_ADDR:

const TCP_ADDR: &str = "127.0.0.1:7878";
const UDP_ADDR: &str = "127.0.0.1:7879";
const VALID_TOKENS: &[&str] = &["secret-producer-token", "secret-consumer-token"];
const HEARTBEAT_TIMEOUT_SECS: i64 = 10;

What do these constants do?

UDP_ADDR is the address where the broker will listen for heartbeat packets. It's a different port (7879) from the TCP listener (7878) because UDP and TCP are separate protocols.

HEARTBEAT_TIMEOUT_SECS defines how many seconds of silence before the broker declares a consumer dead. Ten seconds gives enough buffer for a few missed 3-second heartbeats before triggering.

• Now add the consumer_heartbeats field to the BrokerState struct. Find the struct and update it to:

#[derive(Default)]
struct BrokerState {
    queue: VecDeque<Task>,
    consumers: HashMap<String, mpsc::Sender<Task>>,
    tasks_processed: u64,
    consumer_heartbeats: HashMap<String, chrono::DateTime<Utc>>,
}

What does this field store?

consumer_heartbeats maps each consumer's ID to the last UTC timestamp when the broker received a heartbeat from it. The broker checks these timestamps periodically to find consumers that have gone silent.

• Save broker/src/main.rs.

Bind a UDP socket for receiving heartbeats

The broker needs to listen for incoming UDP packets on a separate port. Unlike TCP, UDP is connectionless. Each heartbeat arrives as an independent datagram with no connection setup or teardown overhead.

Why UDP for heartbeats?

Heartbeats are small, frequent, and tolerant of loss. Missing a single heartbeat packet is fine. Missing several in a row triggers a timeout. TCP's connection handshake and guaranteed delivery would be wasteful overhead for a 50-byte "I'm alive" packet sent every 3 seconds.

• In broker/src/main.rs, find the main() function. Locate this line:

    let state: SharedState = Arc::new(RwLock::new(BrokerState::default()));

• Add the UDP heartbeat listener task directly below that line, before the TCP listener:

    // UDP heartbeat listener
    let udp_state = Arc::clone(&state);
    tokio::spawn(async move {
        let socket = UdpSocket::bind(UDP_ADDR).await.expect("UDP bind failed");
        println!("Broker UDP listening on {}", UDP_ADDR);
        let mut buf = [0u8; 256];
        loop {
            if let Ok((len, _)) = socket.recv_from(&mut buf).await {
                if let Ok(consumer_id) = std::str::from_utf8(&buf[..len]) {
                    let mut s = udp_state.write().await;
                    s.consumer_heartbeats.insert(consumer_id.to_string(), Utc::now());
                }
            }
        }
    });

What does this code do?

This spawns a background task that binds a UDP socket on port 7879. It loops forever, waiting for incoming datagrams.

When a packet arrives, recv_from returns the bytes and the sender's address. The bytes are just the consumer's ID encoded as UTF-8. The task writes the current timestamp into consumer_heartbeats for that consumer.

• Save broker/src/main.rs.

Detect dead consumers with a heartbeat monitor

Receiving heartbeats is only half the picture. The broker also needs a background task that periodically checks the timestamps and logs which consumers have gone silent.

• In broker/src/main.rs, add the heartbeat monitor task directly below the UDP listener task you just added (still before the TCP listener):

    // Heartbeat monitor
    let monitor_state = Arc::clone(&state);
    tokio::spawn(async move {
        let mut ticker = interval(Duration::from_secs(5));
        loop {
            ticker.tick().await;
            let s = monitor_state.read().await;
            let now = Utc::now();
            for (cid, last_seen) in &s.consumer_heartbeats {
                let secs = (now - *last_seen).num_seconds();
                if secs > HEARTBEAT_TIMEOUT_SECS {
                    eprintln!("[HEARTBEAT] Consumer {} is DEAD (last seen {}s ago)", cid, secs);
                } else {
                    println!("[HEARTBEAT] Consumer {} alive (last seen {}s ago)", cid, secs);
                }
            }
        }
    });

What does this code do?

This task wakes up every 5 seconds using tokio::time::interval. It takes a read lock on the shared state and iterates over every consumer's last heartbeat timestamp.

If the time since a consumer's last heartbeat exceeds HEARTBEAT_TIMEOUT_SECS (10 seconds), it logs that consumer as dead. Otherwise it logs the consumer as alive. This gives you real-time visibility into your system's health.

• Save broker/src/main.rs.
• Verify the workspace compiles by running:

cargo build

• You should see a successful compilation with no errors.

Seeing a compilation error?

Make sure the chrono crate is in your broker/Cargo.toml dependencies. It should already be there from a previous step.

Check that your use statements exactly match the code above. A missing UdpSocket or interval import will cause unresolved name errors.

Help me debug a compilation error in my broker's heartbeat code.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

use chrono::Utc;
use shared::{read_command, write_response, Command, Response, Task};
use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
use tokio::io::BufReader;
use tokio::net::{TcpListener, UdpSocket};
use tokio::sync::{mpsc, RwLock};
use tokio::time::{interval, Duration};
use uuid::Uuid;

const TCP_ADDR: &str = "127.0.0.1:7878";
const UDP_ADDR: &str = "127.0.0.1:7879";
const VALID_TOKENS: &[&str] = &["secret-producer-token", "secret-consumer-token"];
const HEARTBEAT_TIMEOUT_SECS: i64 = 10;

#[derive(Default)]
struct BrokerState {
    queue: VecDeque<Task>,
    consumers: HashMap<String, mpsc::Sender<Task>>,
    tasks_processed: u64,
    consumer_heartbeats: HashMap<String, chrono::DateTime<Utc>>,
}

type SharedState = Arc<RwLock<BrokerState>>;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let state: SharedState = Arc::new(RwLock::new(BrokerState::default()));

    // UDP heartbeat listener
    let udp_state = Arc::clone(&state);
    tokio::spawn(async move {
        let socket = UdpSocket::bind(UDP_ADDR).await.expect("UDP bind failed");
        println!("Broker UDP listening on {}", UDP_ADDR);
        let mut buf = [0u8; 256];
        loop {
            if let Ok((len, _)) = socket.recv_from(&mut buf).await {
                if let Ok(consumer_id) = std::str::from_utf8(&buf[..len]) {
                    let mut s = udp_state.write().await;
                    s.consumer_heartbeats.insert(consumer_id.to_string(), Utc::now());
                }
            }
        }
    });

    // Heartbeat monitor
    let monitor_state = Arc::clone(&state);
    tokio::spawn(async move {
        let mut ticker = interval(Duration::from_secs(5));
        loop {
            ticker.tick().await;
            let s = monitor_state.read().await;
            let now = Utc::now();
            for (cid, last_seen) in &s.consumer_heartbeats {
                let secs = (now - *last_seen).num_seconds();
                if secs > HEARTBEAT_TIMEOUT_SECS {
                    eprintln!("[HEARTBEAT] Consumer {} is DEAD (last seen {}s ago)", cid, secs);
                } else {
                    println!("[HEARTBEAT] Consumer {} alive (last seen {}s ago)", cid, secs);
                }
            }
        }
    });

    let tcp_listener = TcpListener::bind(TCP_ADDR).await?;
    println!("Broker TCP listening on {}", TCP_ADDR);

    loop {
        let (stream, addr) = tcp_listener.accept().await?;
        println!("New connection from {}", addr);
        let state = Arc::clone(&state);
        tokio::spawn(async move {
            if let Err(e) = handle_connection(stream, state).await {
                eprintln!("Connection error: {}", e);
            }
        });
    }
}

async fn handle_connection(stream: tokio::net::TcpStream, state: SharedState) -> anyhow::Result<()> {
    let (read_half, mut write_half) = stream.into_split();
    let mut reader = BufReader::new(read_half);

    let first_cmd = read_command(&mut reader).await?;
    match first_cmd {
        Command::Auth { token } if VALID_TOKENS.contains(&token.as_str()) => {
            write_response(&mut write_half, &Response::Ok).await?;
            println!("Client authenticated");
        }
        _ => {
            write_response(&mut write_half, &Response::Error { message: "Unauthorized".to_string() }).await?;
            return Ok(());
        }
    }

    loop {
        let cmd = read_command(&mut reader).await?;
        match cmd {
            Command::Publish { payload } => {
                let task = Task { id: Uuid::new_v4().to_string(), payload, redelivery_count: 0 };
                let mut s = state.write().await;
                let consumer_ids: Vec<String> = s.consumers.keys().cloned().collect();
                let mut dispatched = false;
                for cid in consumer_ids {
                    if let Some(tx) = s.consumers.get(&cid) {
                        if tx.send(task.clone()).await.is_ok() {
                            dispatched = true;
                            break;
                        } else {
                            s.consumers.remove(&cid);
                        }
                    }
                }
                if !dispatched {
                    s.queue.push_back(task);
                }
                write_response(&mut write_half, &Response::Ok).await?;
            }
            Command::Subscribe { consumer_id } => {
                let (tx, mut rx) = mpsc::channel::<Task>(32);
                {
                    let mut s = state.write().await;
                    while let Some(task) = s.queue.pop_front() {
                        let _ = tx.send(task).await;
                    }
                    s.consumers.insert(consumer_id.clone(), tx);
                }
                write_response(&mut write_half, &Response::Ok).await?;
                while let Some(task) = rx.recv().await {
                    write_response(&mut write_half, &Response::Task(task)).await?;
                }
            }
            Command::Ack { task_id: _ } => {
                let mut s = state.write().await;
                s.tasks_processed += 1;
                write_response(&mut write_half, &Response::Ok).await?;
            }
            Command::Auth { .. } => {
                write_response(&mut write_half, &Response::Error { message: "Already authenticated".to_string() }).await?;
            }
        }
    }
}

Send heartbeats from the consumer

Now the broker can receive and monitor heartbeats. The consumer needs to actually send them. You'll spawn a background Tokio task that sends a UDP packet containing the consumer's ID every 3 seconds.

• Replace the entire contents of consumer/src/main.rs with the following:

use shared::Command;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{TcpStream, UdpSocket};
use tokio::time::{interval, Duration};
use uuid::Uuid;
use shared::Response;

const UDP_BROKER_ADDR: &str = "127.0.0.1:7879";

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let consumer_id = Uuid::new_v4().to_string();
    let stream = TcpStream::connect("127.0.0.1:7878").await?;
    let (read_half, mut write_half) = stream.into_split();
    let mut reader = BufReader::new(read_half);

    // Authenticate
    let auth = serde_json::to_string(&Command::Auth { token: "secret-consumer-token".to_string() })? + "\n";
    write_half.write_all(auth.as_bytes()).await?;
    let mut line = String::new();
    reader.read_line(&mut line).await?;
    println!("Auth: {}", line.trim());

    // Subscribe
    let sub = serde_json::to_string(&Command::Subscribe { consumer_id: consumer_id.clone() })? + "\n";
    write_half.write_all(sub.as_bytes()).await?;
    let mut line = String::new();
    reader.read_line(&mut line).await?;
    println!("Subscribe: {}", line.trim());

    // Heartbeat task
    let hb_id = consumer_id.clone();
    tokio::spawn(async move {
        let socket = UdpSocket::bind("0.0.0.0:0").await.expect("UDP bind failed");
        let mut ticker = interval(Duration::from_secs(3));
        loop {
            ticker.tick().await;
            let _ = socket.send_to(hb_id.as_bytes(), UDP_BROKER_ADDR).await;
        }
    });

    // Receive tasks and ACK
    loop {
        let mut line = String::new();
        reader.read_line(&mut line).await?;
        let resp: Response = serde_json::from_str(line.trim())?;
        if let Response::Task(task) = resp {
            println!("Received task: {} - {}", task.id, task.payload);
            let ack = serde_json::to_string(&Command::Ack { task_id: task.id })? + "\n";
            write_half.write_all(ack.as_bytes()).await?;
        }
    }
}

What changed in the consumer?

The new imports add UdpSocket, interval, and Duration for the heartbeat logic. The constant UDP_BROKER_ADDR points to the broker's UDP port.

The heartbeat task binds to 0.0.0.0:0, which tells the OS to pick any available port. The consumer doesn't need a fixed port because it only sends packets. Every 3 seconds it fires the consumer ID as raw bytes to the broker's UDP address.

• Save consumer/src/main.rs.
• Verify the workspace compiles by running:

cargo build

• You should see a successful compilation with no errors.

Getting an unresolved import error?

Make sure your consumer/Cargo.toml has tokio with features = ["full"]. The UdpSocket and interval types require the net and time features respectively, both included in "full".

Help me fix an import error in my consumer's heartbeat code.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

use shared::Command;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{TcpStream, UdpSocket};
use tokio::time::{interval, Duration};
use uuid::Uuid;
use shared::Response;

const UDP_BROKER_ADDR: &str = "127.0.0.1:7879";

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let consumer_id = Uuid::new_v4().to_string();
    let stream = TcpStream::connect("127.0.0.1:7878").await?;
    let (read_half, mut write_half) = stream.into_split();
    let mut reader = BufReader::new(read_half);

    // Authenticate
    let auth = serde_json::to_string(&Command::Auth { token: "secret-consumer-token".to_string() })? + "\n";
    write_half.write_all(auth.as_bytes()).await?;
    let mut line = String::new();
    reader.read_line(&mut line).await?;
    println!("Auth: {}", line.trim());

    // Subscribe
    let sub = serde_json::to_string(&Command::Subscribe { consumer_id: consumer_id.clone() })? + "\n";
    write_half.write_all(sub.as_bytes()).await?;
    let mut line = String::new();
    reader.read_line(&mut line).await?;
    println!("Subscribe: {}", line.trim());

    // Heartbeat task
    let hb_id = consumer_id.clone();
    tokio::spawn(async move {
        let socket = UdpSocket::bind("0.0.0.0:0").await.expect("UDP bind failed");
        let mut ticker = interval(Duration::from_secs(3));
        loop {
            ticker.tick().await;
            let _ = socket.send_to(hb_id.as_bytes(), UDP_BROKER_ADDR).await;
        }
    });

    // Receive tasks and ACK
    loop {
        let mut line = String::new();
        reader.read_line(&mut line).await?;
        let resp: Response = serde_json::from_str(line.trim())?;
        if let Response::Task(task) = resp {
            println!("Received task: {} - {}", task.id, task.payload);
            let ack = serde_json::to_string(&Command::Ack { task_id: task.id })? + "\n";
            write_half.write_all(ack.as_bytes()).await?;
        }
    }
}

Your broker now has eyes on every consumer. It knows who is alive, who has gone silent, and when they come back. Next up, you'll add an HTTP status API that exposes these metrics so you can query your system's health from outside.

Build the HTTP Status API

Your broker now handles TCP task connections and UDP heartbeats. But how do you check on the system's health without connecting as a producer or consumer?

In this step, you'll add a third listener to the broker: an HTTP status API. You'll parse raw HTTP requests by hand, validate an Authorization header, and return JSON metrics. This teaches a crucial insight: HTTP is just structured text over TCP.

In this step, get ready to:

• Bind a TcpListener on port 8080 for HTTP inside the broker.
• Parse incoming HTTP requests manually and route them.
• Protect the endpoint with Bearer token authentication and test it with curl.

Bind the HTTP listener on a third port

The broker already uses port 7878 for TCP and 7879 for UDP. The HTTP status API needs its own port so it can serve standard HTTP clients like curl or a browser.

• In broker/src/main.rs, find the import line:

use tokio::io::BufReader;

• Replace it with this expanded import that includes AsyncBufReadExt and AsyncWriteExt:

use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};

Why add these traits?

The HTTP handler reads and writes directly on the TCP stream using read_line and write_all. These methods come from the AsyncBufReadExt and AsyncWriteExt traits. Without importing them, the compiler won't find those methods on the stream.

• Next, find this line near the top of the file:

const UDP_ADDR: &str = "127.0.0.1:7879";

• Add the HTTP address constant directly below it:

const HTTP_ADDR: &str = "127.0.0.1:8080";

Now you need to spawn a background task that binds this HTTP port and accepts connections. This follows the same pattern as the UDP listener: clone the shared state, then spawn an async task.

• In the main function, find the line that says let tcp_listener = TcpListener::bind(TCP_ADDR).await?; and add this block directly above it:

    // HTTP status API
    let http_state = Arc::clone(&state);
    tokio::spawn(async move {
        let http_listener = TcpListener::bind(HTTP_ADDR).await.expect("HTTP bind failed");
        println!("Broker HTTP listening on {}", HTTP_ADDR);
        loop {
            if let Ok((stream, _)) = http_listener.accept().await {
                let state = Arc::clone(&http_state);
                tokio::spawn(async move {
                    if let Err(e) = handle_http_connection(stream, state).await {
                        eprintln!("HTTP error: {}", e);
                    }
                });
            }
        }
    });

What does this code do?

This spawns a third listener loop inside the broker. It binds a standard TcpListener on port 8080 and accepts connections the same way the task protocol does.

Each incoming HTTP connection is handled in its own spawned task, so one slow client doesn't block others. The handle_http_connection function (which you'll write next) parses the raw HTTP text.

• Save broker/src/main.rs.

Why does this HTTP listener use a TcpListener instead of an HTTP library?

HTTP is just a text protocol on top of TCP. By parsing the raw bytes yourself, you see exactly what happens when a browser or curl connects: a request line, headers separated by \r\n, and an empty line ending the header section.

Frameworks like Axum or Actix abstract this away. Building it by hand first means you understand what those frameworks are doing under the hood.

Implement the HTTP connection handler

The handler needs to do four things: read the request line (method + path), read headers, check the Authorization header, and return the appropriate response based on the path.

• Add the handle_http_connection function above the existing handle_connection function by pasting the following code:

async fn handle_http_connection(stream: tokio::net::TcpStream, state: SharedState) -> anyhow::Result<()> {
    let (read_half, mut write_half) = stream.into_split();
    let mut reader = BufReader::new(read_half);

    // Read the request line (e.g. "GET /status HTTP/1.1")
    let mut request_line = String::new();
    reader.read_line(&mut request_line).await?;
    let parts: Vec<&str> = request_line.trim().splitn(3, ' ').collect();
    if parts.len() < 2 { return Ok(()); }
    let method = parts[0];
    let path = parts[1];

    // Read headers until the empty line
    let mut auth_header: Option<String> = None;
    loop {
        let mut header_line = String::new();
        reader.read_line(&mut header_line).await?;
        if header_line.trim().is_empty() { break; }
        if header_line.to_lowercase().starts_with("authorization:") {
            auth_header = Some(header_line.trim().to_string());
        }
    }

    // Validate the Bearer token against VALID_TOKENS
    let authorized = auth_header.as_ref().map(|h| {
        let val = h.splitn(2, ':').nth(1).unwrap_or("").trim();
        if let Some(token) = val.strip_prefix("Bearer ") {
            VALID_TOKENS.contains(&token)
        } else { false }
    }).unwrap_or(false);

    if !authorized {
        let resp = "HTTP/1.1 401 Unauthorized\r\nContent-Length: 12\r\n\r\nUnauthorized";
        write_half.write_all(resp.as_bytes()).await?;
        return Ok(());
    }

    if method == "GET" && path == "/status" {
        // Build the JSON status payload from shared state
        let s = state.read().await;
        let now = Utc::now();
        let dead = s.consumer_heartbeats.values().filter(|t| (now - **t).num_seconds() > HEARTBEAT_TIMEOUT_SECS).count();
        let body = serde_json::json!({
            "queue_depth": s.queue.len(),
            "connected_consumers": s.consumers.len(),
            "tasks_processed": s.tasks_processed,
            "dead_consumers": dead,
        }).to_string();
        let resp = format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{}", body.len(), body);
        write_half.write_all(resp.as_bytes()).await?;
    } else {
        let resp = "HTTP/1.1 404 Not Found\r\nContent-Length: 9\r\n\r\nNot Found";
        write_half.write_all(resp.as_bytes()).await?;
    }
    Ok(())
}

What does this code do?

• The first read_line grabs the HTTP request line (like GET /status HTTP/1.1) and splits it into method and path.
• The header loop reads lines until it hits an empty line (the end-of-headers marker in HTTP). It looks for the Authorization header specifically.
• The auth check strips the Bearer prefix and validates the token against the same VALID_TOKENS list the TCP handler uses. No valid token means a 401 response.
• For GET /status, it reads the broker state and returns a JSON body with queue depth, consumer count, processed tasks, and dead consumers. Any other path returns 404.

• Save broker/src/main.rs.
• Confirm the project compiles by running:

cargo build

• You should see a successful compilation with no errors.

Seeing a compile error?

• If you see "no method named read_line found", make sure your import line includes AsyncBufReadExt.
• If you see "cannot find function handle_http_connection", make sure the function is defined outside main (not inside it).
• Check that HTTP_ADDR is defined at the top of the file alongside the other constants.

Help me fix my broker compile error

Test the status API with curl

Time to see your HTTP API in action. You'll start the broker and hit the status endpoint from a second terminal.

• Start the broker by running:

cargo run --bin broker

• You should see three "listening" messages: TCP on 7878, UDP on 7879, and HTTP on 8080.
• Open a second terminal tab or pane.
• Query the status endpoint with a valid token by running:

curl -H 'Authorization: Bearer secret-producer-token' http://127.0.0.1:8080/status

• You should see a JSON response like:

{"queue_depth":0,"connected_consumers":0,"tasks_processed":0,"dead_consumers":0}

• Now test that authentication is enforced by running a request without a token:

curl http://127.0.0.1:8080/status

• You should see Unauthorized as the response body.
• Test an invalid path by running:

curl -H 'Authorization: Bearer secret-producer-token' http://127.0.0.1:8080/other

• You should see Not Found as the response.

What do the status fields mean?

• queue_depth: number of tasks waiting to be dispatched (no consumers available).
• connected_consumers: number of consumers currently subscribed over TCP.
• tasks_processed: total tasks acknowledged by consumers since the broker started.
• dead_consumers: consumers whose heartbeat timed out (no UDP packet in 10 seconds).

Getting "Connection refused" from curl?

• Make sure the broker is still running in the first terminal. Look for the "Broker HTTP listening on 127.0.0.1:8080" message.
• If another process is using port 8080, stop it or change HTTP_ADDR to a different port.

Help me debug my HTTP status endpoint

✔️ Awesome, I've got everything!

Great. Double check you've saved broker/src/main.rs and the project compiles cleanly.

ⓧ I'd like to double check the full code

Here is the complete broker/src/main.rs at this point:

use chrono::Utc;
use shared::{read_command, write_response, Command, Response, Task};
use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{TcpListener, UdpSocket};
use tokio::sync::{mpsc, RwLock};
use tokio::time::{interval, Duration};
use uuid::Uuid;

const TCP_ADDR: &str = "127.0.0.1:7878";
const UDP_ADDR: &str = "127.0.0.1:7879";
const HTTP_ADDR: &str = "127.0.0.1:8080";
const VALID_TOKENS: &[&str] = &["secret-producer-token", "secret-consumer-token"];
const HEARTBEAT_TIMEOUT_SECS: i64 = 10;

#[derive(Default)]
struct BrokerState {
    queue: VecDeque<Task>,
    consumers: HashMap<String, mpsc::Sender<Task>>,
    tasks_processed: u64,
    consumer_heartbeats: HashMap<String, chrono::DateTime<Utc>>,
}

type SharedState = Arc<RwLock<BrokerState>>;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let state: SharedState = Arc::new(RwLock::new(BrokerState::default()));

    // UDP heartbeat listener
    let udp_state = Arc::clone(&state);
    tokio::spawn(async move {
        let socket = UdpSocket::bind(UDP_ADDR).await.expect("UDP bind failed");
        println!("Broker UDP listening on {}", UDP_ADDR);
        let mut buf = [0u8; 256];
        loop {
            if let Ok((len, _)) = socket.recv_from(&mut buf).await {
                if let Ok(consumer_id) = std::str::from_utf8(&buf[..len]) {
                    let mut s = udp_state.write().await;
                    s.consumer_heartbeats.insert(consumer_id.to_string(), Utc::now());
                }
            }
        }
    });

    // Heartbeat monitor
    let monitor_state = Arc::clone(&state);
    tokio::spawn(async move {
        let mut ticker = interval(Duration::from_secs(5));
        loop {
            ticker.tick().await;
            let s = monitor_state.read().await;
            let now = Utc::now();
            for (cid, last_seen) in &s.consumer_heartbeats {
                let secs = (now - *last_seen).num_seconds();
                if secs > HEARTBEAT_TIMEOUT_SECS {
                    eprintln!("[HEARTBEAT] Consumer {} is DEAD (last seen {}s ago)", cid, secs);
                } else {
                    println!("[HEARTBEAT] Consumer {} alive (last seen {}s ago)", cid, secs);
                }
            }
        }
    });

    // HTTP status API
    let http_state = Arc::clone(&state);
    tokio::spawn(async move {
        let http_listener = TcpListener::bind(HTTP_ADDR).await.expect("HTTP bind failed");
        println!("Broker HTTP listening on {}", HTTP_ADDR);
        loop {
            if let Ok((stream, _)) = http_listener.accept().await {
                let state = Arc::clone(&http_state);
                tokio::spawn(async move {
                    if let Err(e) = handle_http_connection(stream, state).await {
                        eprintln!("HTTP error: {}", e);
                    }
                });
            }
        }
    });

    let tcp_listener = TcpListener::bind(TCP_ADDR).await?;
    println!("Broker TCP listening on {}", TCP_ADDR);

    loop {
        let (stream, addr) = tcp_listener.accept().await?;
        println!("New connection from {}", addr);
        let state = Arc::clone(&state);
        tokio::spawn(async move {
            if let Err(e) = handle_connection(stream, state).await {
                eprintln!("Connection error: {}", e);
            }
        });
    }
}

async fn handle_http_connection(stream: tokio::net::TcpStream, state: SharedState) -> anyhow::Result<()> {
    let (read_half, mut write_half) = stream.into_split();
    let mut reader = BufReader::new(read_half);

    let mut request_line = String::new();
    reader.read_line(&mut request_line).await?;
    let parts: Vec<&str> = request_line.trim().splitn(3, ' ').collect();
    if parts.len() < 2 { return Ok(()); }
    let method = parts[0];
    let path = parts[1];

    let mut auth_header: Option<String> = None;
    loop {
        let mut header_line = String::new();
        reader.read_line(&mut header_line).await?;
        if header_line.trim().is_empty() { break; }
        if header_line.to_lowercase().starts_with("authorization:") {
            auth_header = Some(header_line.trim().to_string());
        }
    }

    let authorized = auth_header.as_ref().map(|h| {
        let val = h.splitn(2, ':').nth(1).unwrap_or("").trim();
        if let Some(token) = val.strip_prefix("Bearer ") {
            VALID_TOKENS.contains(&token)
        } else { false }
    }).unwrap_or(false);

    if !authorized {
        let resp = "HTTP/1.1 401 Unauthorized\r\nContent-Length: 12\r\n\r\nUnauthorized";
        write_half.write_all(resp.as_bytes()).await?;
        return Ok(());
    }

    if method == "GET" && path == "/status" {
        let s = state.read().await;
        let now = Utc::now();
        let dead = s.consumer_heartbeats.values().filter(|t| (now - **t).num_seconds() > HEARTBEAT_TIMEOUT_SECS).count();
        let body = serde_json::json!({
            "queue_depth": s.queue.len(),
            "connected_consumers": s.consumers.len(),
            "tasks_processed": s.tasks_processed,
            "dead_consumers": dead,
        }).to_string();
        let resp = format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{}", body.len(), body);
        write_half.write_all(resp.as_bytes()).await?;
    } else {
        let resp = "HTTP/1.1 404 Not Found\r\nContent-Length: 9\r\n\r\nNot Found";
        write_half.write_all(resp.as_bytes()).await?;
    }
    Ok(())
}

async fn handle_connection(stream: tokio::net::TcpStream, state: SharedState) -> anyhow::Result<()> {
    let (read_half, mut write_half) = stream.into_split();
    let mut reader = BufReader::new(read_half);

    let first_cmd = read_command(&mut reader).await?;
    match first_cmd {
        Command::Auth { token } if VALID_TOKENS.contains(&token.as_str()) => {
            write_response(&mut write_half, &Response::Ok).await?;
            println!("Client authenticated");
        }
        _ => {
            write_response(&mut write_half, &Response::Error { message: "Unauthorized".to_string() }).await?;
            return Ok(());
        }
    }

    loop {
        let cmd = read_command(&mut reader).await?;
        match cmd {
            Command::Publish { payload } => {
                let task = Task { id: Uuid::new_v4().to_string(), payload, redelivery_count: 0 };
                let mut s = state.write().await;
                let consumer_ids: Vec<String> = s.consumers.keys().cloned().collect();
                let mut dispatched = false;
                for cid in consumer_ids {
                    if let Some(tx) = s.consumers.get(&cid) {
                        if tx.send(task.clone()).await.is_ok() {
                            dispatched = true;
                            break;
                        } else {
                            s.consumers.remove(&cid);
                        }
                    }
                }
                if !dispatched {
                    s.queue.push_back(task);
                }
                write_response(&mut write_half, &Response::Ok).await?;
            }
            Command::Subscribe { consumer_id } => {
                let (tx, mut rx) = mpsc::channel::<Task>(32);
                {
                    let mut s = state.write().await;
                    while let Some(task) = s.queue.pop_front() {
                        let _ = tx.send(task).await;
                    }
                    s.consumers.insert(consumer_id.clone(), tx);
                }
                write_response(&mut write_half, &Response::Ok).await?;
                while let Some(task) = rx.recv().await {
                    write_response(&mut write_half, &Response::Task(task)).await?;
                }
            }
            Command::Ack { task_id: _ } => {
                let mut s = state.write().await;
                s.tasks_processed += 1;
                write_response(&mut write_half, &Response::Ok).await?;
            }
            Command::Auth { .. } => {
                write_response(&mut write_half, &Response::Error { message: "Already authenticated".to_string() }).await?;
            }
        }
    }
}

Your broker now exposes real-time metrics over HTTP, protected by the same token system as the task protocol. In the next step, you'll run the full system end-to-end with multiple producers and consumers to see everything working together.

Run the Full Integration Test

Your HTTP status API is live. Every piece of your distributed task queue is now in place: a TCP task protocol, UDP heartbeats, and HTTP observability.

In this step, you'll bring the whole system to life by running multiple processes simultaneously. You'll watch tasks flow from producers to consumers, heartbeats pulse through UDP, and the status API report real-time metrics.

In this step, get ready to:

• Start the broker and connect multiple consumers.
• Publish tasks from multiple producers and query live metrics.
• Kill a consumer and verify the broker detects its heartbeat timeout.

Start the broker and connect consumers

A distributed system needs multiple processes communicating over the network. You'll open several terminal panes in VS Code so you can watch each process independently.

• First, build the entire workspace to ensure everything compiles cleanly by running:

cargo build

Seeing compile errors?

If you get errors at this point, double check your code against the previous steps. The most common issue is a mismatched type between shared/src/lib.rs and how the broker or consumer uses the shared types.

Help me debug my cargo build errors for the task queue workspace.

Now you need multiple terminal panes open at the same time. You'll use six in total: one for the broker, three for consumers, two for producers.

• In your current terminal pane (this is Terminal 1), start the broker by running:

cargo run --bin broker

You should see three lines confirming all listeners are active:

Broker UDP listening on 127.0.0.1:7879
Broker HTTP listening on 127.0.0.1:8080
Broker TCP listening on 127.0.0.1:7878

Why three listeners?

Your broker is now listening on three ports simultaneously using Tokio tasks. Port 7878 handles TCP task protocol traffic. Port 7879 receives UDP heartbeat packets. Port 8080 serves the HTTP status API.

The broker is now running and blocking this terminal. You need new terminal panes for the consumers.

• Open a new terminal pane by clicking the + icon in the terminal panel (or press Ctrl+Shift+`).
• In this new pane (Terminal 2), start the first consumer by running:

cargo run --bin consumer

• Open another new terminal pane (Terminal 3) and start the second consumer by running:

cargo run --bin consumer

• Open another new terminal pane (Terminal 4) and start the third consumer by running:

cargo run --bin consumer

Each consumer should print its authentication and subscription confirmation. In Terminal 1 (the broker), you should see three "New connection" and "Client authenticated" messages.

What should I see in the broker terminal?

After a few seconds, the broker's heartbeat monitor kicks in and starts printing [HEARTBEAT] Consumer <uuid> alive messages every 5 seconds. This confirms your UDP heartbeat system is working. Each consumer is sending its ID over UDP every 3 seconds.

Consumer shows a connection error?

Make sure the broker is still running in Terminal 1. If the broker crashed or you accidentally stopped it, restart it with cargo run --bin broker before starting consumers.

Help me troubleshoot why my consumer can't connect to the broker.

Publish tasks and query the status API

With three consumers waiting for work, it's time to flood the system with tasks. You'll run two producer instances that each publish 10 tasks (20 total) and then query the HTTP status endpoint to see the metrics.

• Open a new terminal pane (Terminal 5) and run the first producer by running:

cargo run --bin producer

• Open another new terminal pane (Terminal 6) and run the second producer by running:

cargo run --bin producer

Each producer prints 10 "Publish" confirmations and then exits. Switch to your consumer terminals. You should see "Received task" messages distributed across the three consumers.

How are tasks distributed?

The broker dispatches each task to the first available consumer it finds in its consumers HashMap. Because HashMap iteration order is non-deterministic in Rust, you'll see tasks spread across consumers in a somewhat unpredictable pattern. This is fine. The point is that all 20 tasks get processed.

Now query the HTTP status API to see your system's metrics. The producers have exited, so you can reuse one of those terminals.

• In Terminal 5 (where the first producer finished), query the status endpoint by running:

curl -H 'Authorization: Bearer secret-producer-token' http://127.0.0.1:8080/status

You should see a JSON response similar to this:

{"queue_depth":0,"connected_consumers":3,"tasks_processed":20,"dead_consumers":0}

What do these metrics mean?

• queue_depth is 0 because all tasks were dispatched immediately to consumers.
• connected_consumers shows 3 because all three consumers are still subscribed.
• tasks_processed shows 20 because each producer published 10 tasks and every task was ACKed.
• dead_consumers is 0 because all consumers are sending heartbeats.

Getting a 401 or connection refused?

If you see "Unauthorized", check that the token in your curl command matches exactly: secret-producer-token. If you get "Connection refused", the broker may have crashed. Check Terminal 1 for errors and restart it.

Help me debug my curl request to the task queue HTTP status API.

Kill a consumer and observe heartbeat timeout

The real power of the heartbeat system shows when a consumer fails. In a real distributed system, processes crash all the time. Your broker needs to detect this automatically.

• Switch to one of your consumer terminal panes (Terminal 2, 3, or 4).
• Press Ctrl+C to kill that consumer process.
• Switch back to Terminal 1 (the broker) and watch the logs.

Within about 10 seconds, you should see the broker print a DEAD heartbeat message for the killed consumer's UUID. The other two consumers continue showing "alive" messages.

[HEARTBEAT] Consumer <killed-consumer-uuid> is DEAD (last seen 12s ago)
[HEARTBEAT] Consumer <other-uuid> alive (last seen 2s ago)
[HEARTBEAT] Consumer <other-uuid> alive (last seen 1s ago)

Why UDP for heartbeats?

This is exactly why heartbeats use UDP instead of TCP. A heartbeat is a tiny "I'm alive" packet sent every 3 seconds. TCP's connection handshake and guaranteed delivery are overkill for something this lightweight. If one heartbeat packet gets lost, no problem. The broker only cares when heartbeats stop arriving entirely.

• Query the status API again to confirm the dead consumer count updated by running:

curl -H 'Authorization: Bearer secret-producer-token' http://127.0.0.1:8080/status

The dead_consumers field should now show 1 and connected_consumers still shows 3 because the broker's consumer map tracks subscribers by their channel, not their heartbeat state.

What just happened across all three protocols?

You just witnessed each protocol doing what it does best. TCP delivered tasks reliably to consumers with guaranteed ordering and acknowledgments. UDP provided a low-overhead liveness signal that tolerated packet loss. HTTP gave you a request/response status check that any standard tool like curl can query.

Token-based authentication tied all three together. Both the TCP task protocol and the HTTP status API rejected unauthorized access.

• Stop the broker and remaining consumers by pressing Ctrl+C in each terminal pane when you're done observing.

Secret mission

Consumer Groups and Task Redelivery

Your task queue delivers tasks to any available consumer. But what happens when a consumer dies mid-task? The task vanishes into the void. Real message brokers solve this with consumer groups and automatic redelivery. Can you add both?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs or cloud resources to worry about. Decide whether to keep your project files, stop running processes, or delete everything.

Resources you used:

• The task-queue workspace folder on your Desktop (contains broker, producer, consumer, and shared crates).
• Running broker, producer, and consumer processes (if still active in your terminal tabs).

Keep everything running

No action needed. Choose this if you're still actively building or want to extend the project further.

• Your task-queue folder stays on your Desktop ready for the next session.
• Any running broker, producer, or consumer processes will continue until you close those terminal tabs.
• Rust and Cargo remain installed on your system.

Pause - I'll come back to this later

Stop all running processes to free up system resources, but keep your project files intact.

• Switch to each terminal tab running a broker, producer, or consumer process.
• Press Ctrl+C in each terminal to stop the process.
• Your task-queue folder and all source code remain on your Desktop for next time.

Delete - I don't want to use this again

Remove all project files and stop any running processes. This frees up disk space used by the compiled binaries and dependencies.

• Stop any running broker, producer, or consumer processes by pressing Ctrl+C in each terminal tab.
• Delete the project folder by running this command:

rm -rf ~/Desktop/task-queue

On Windows, run this instead:

Remove-Item -Recurse -Force ~\Desktop\task-queue

• Confirm the folder is gone by running:

ls ~/Desktop/task-queue

You should see a "No such file or directory" message.

Nice Work!

Nice Work!

You just built a distributed task queue from scratch in Rust. That is a production-grade networking project covering three protocols, authentication, and concurrent state management.

Here's what you accomplished:

• Built a multi-protocol broker that listens on TCP, UDP, and HTTP simultaneously using Tokio async tasks and a custom newline-delimited JSON wire protocol.
• Implemented token-based authentication that gates access to the system, requiring clients to prove their identity before any operation.
• Used UDP for lightweight consumer heartbeats and built an HTTP status API by hand, seeing firsthand how HTTP is just structured text over TCP.
• Completed the Secret Mission: added consumer groups with round-robin dispatch and automatic task redelivery when a consumer's heartbeat times out.

Ready to quiz yourself?