Queue Server

A lightweight distributed message queue in Rust, featuring OpenRaft consensus, human-readable disk storage, and custom client libraries.

Features

• Distributed Consensus: OpenRaft-backed 3-node clustering with automatic leader election and transparent follower-to-leader forwarding.
• Inspectable Disk Storage: WAL with fsync persistence and human-readable JSON message files under ./data (portable across containers).
• Partition Routing: Hash-based partition assignment by message key or round-robin distribution across partitions.
• Consumer Groups: At-least-once delivery with explicit async acknowledgments and real-time streaming notifications.
• Health & Monitoring: Cluster status reporting with real-time node online/offline detection and leader tracking.
• Admin CLI (qctl): Command-line tool for topic management, cluster inspection, and context persistence.

Tech Stack

• Language: Rust (Edition 2021)
• Async Runtime: Tokio
• Consensus: OpenRaft 0.9
• Protocol: Length-prefixed framing (4-byte LE length + JSON payload) over TCP
• Serialization: Serde JSON
• Containerization: Docker, Docker Compose, cargo-chef

Getting Started

Using Quickstart Scripts

# Start 3-node demo cluster
./scripts/demo-up.sh

# Stop demo cluster
./scripts/demo-down.sh

Docker Compose Example

A 3-node cluster definition (docker-compose.yml):

services:
  node1:
    image: dreamoutbox/queue-server:latest
    environment:
      NODE_ID: "1"
      LISTEN_CLIENT: "0.0.0.0:7777"
      LISTEN_RAFT: "0.0.0.0:7778"
      ADVERTISE_CLIENT: "node1:7777"
      ADVERTISE_RAFT: "node1:7778"
      PEERS: "2=node2:7778:node2:7777,3=node3:7778:node3:7777"
    ports:
      - "7777:7777"
    volumes:
      - ./data/node1:/data

  node2:
    image: dreamoutbox/queue-server:latest
    environment:
      NODE_ID: "2"
      LISTEN_CLIENT: "0.0.0.0:7777"
      LISTEN_RAFT: "0.0.0.0:7778"
      ADVERTISE_CLIENT: "node2:7777"
      ADVERTISE_RAFT: "node2:7778"
      PEERS: "1=node1:7778:node1:7777,3=node3:7778:node3:7777"
    ports:
      - "27777:7777"
    volumes:
      - ./data/node2:/data

  node3:
    image: dreamoutbox/queue-server:latest
    environment:
      NODE_ID: "3"
      LISTEN_CLIENT: "0.0.0.0:7777"
      LISTEN_RAFT: "0.0.0.0:7778"
      ADVERTISE_CLIENT: "node3:7777"
      ADVERTISE_RAFT: "node3:7778"
      PEERS: "1=node1:7778:node1:7777,2=node2:7778:node2:7777"
    ports:
      - "37777:7777"
    volumes:
      - ./data/node3:/data

Commands Usage

Note: Make sure qctl is in your PATH. You can download the qctl binary from GitHub Releases, or install it locally with cargo install --path crates/cli.

Inspect Cluster Status

qctl --addr 127.0.0.1:7777 cluster status

Manage Topics

# Create topic
qctl --addr 127.0.0.1:7777 topic create demo --partitions 3 --retention 604800

# List topics
qctl --addr 127.0.0.1:7777 topic list

Publish Messages

# Publish JSON payload
qctl publish demo '{"user": "alice", "action": "login"}'

# Publish with partition routing key (same key always routes to same partition)
qctl publish demo '{"order_id": 102}' --key user-42

# Or using the producer example app:
cargo run -p producer-example -- --addr 127.0.0.1:7777 --count 10

Consume Messages

# Stream live messages continuously (Ctrl+C to stop)
qctl consume demo

# Consume with a specific consumer group and limit
qctl consume demo --group order-workers --count 10

# Or using the consumer example app:
cargo run -p consumer-example -- --addr 127.0.0.1:27777

CLI Context Shortcut

Set a default target node so --addr is not required on subsequent CLI commands:

qctl context set 127.0.0.1:7777
qctl cluster status
qctl topic list
qctl publish demo '{"status": "ok"}'
qctl consume demo

Rust Library Code Usage

Add the client crate dependency:

[dependencies]
client = { path = "crates/client" }
tokio = { version = "1", features = ["full"] }
serde_json = "1.0"

Producer

use client::Producer;
use serde_json::json;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut producer = Producer::connect("127.0.0.1:7777").await?;

    // Publish with auto partition routing
    let (partition, offset) = producer
        .publish("orders", json!({ "order_id": 101, "item": "book" }))
        .await?;

    // Publish with key (identical keys route to the same partition)
    let (partition, offset) = producer
        .publish_with_key("orders", Some("user-42"), json!({ "order_id": 102 }))
        .await?;

    println!("Published to partition {} at offset {}", partition, offset);
    Ok(())
}

Consumer

use client::Consumer;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut consumer = Consumer::connect("127.0.0.1:27777").await?;
    consumer.subscribe("orders", "order-workers").await?;

    // Stream live messages; explicit ack persists consumer group offset
    while let Some(msg) = consumer.next().await? {
        println!(
            "Received: partition={}, offset={}, payload={}",
            msg.partition, msg.offset, msg.payload
        );
        msg.ack().await?;
    }

    Ok(())
}

Admin Client

use client::{AdminClient, CreateTopicOpts};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut admin = AdminClient::connect("127.0.0.1:7777").await?;

    admin
        .create_topic(
            "orders",
            CreateTopicOpts {
                partitions: 3,
                retention_secs: 604800,
            },
        )
        .await?;

    let status = admin.cluster_status().await?;
    println!("Status: {:?}", status);
    Ok(())
}

Development Setup

Prerequisites

• Rust 1.75+ (cargo, rustc)
• Docker & Docker Compose

Workspace Crates

• crates/proto: Wire protocol framing codec, request/response models.
• crates/storage: Raft log storage, state machine, and JSON partition files.
• crates/node: Server node binary, TCP listener, OpenRaft integration.
• crates/client: Producer, Consumer, and Admin client libraries.
• crates/cli: qctl administration CLI binary.
• crates/producer-example: Producer sample application.
• crates/consumer-example: Consumer sample application.
• tests/integration: End-to-end multi-node integration test suite.

Build Workspace

cargo build --workspace

Run Tests

# Full test suite with automated container cleanup (-j1 for serial execution)
./scripts/run-tests.sh -j1

# Single integration test
cargo test -p integration-tests --lib -- leader_failover --nocapture

Build Docker Image

./scripts/build-image.sh