walrq: Distributed High-Throughput Queue Engine

walrq is a high-throughput, low-latency distributed message queue engine built in pure Rust. It delivers AWS SQS-style semantics (visibility timeouts, dead-letter queues, delayed messages, batching) powered by Raft quorum consensus, Postcard binary framing over persistent TCP, and a zero-BTree visibility-based disk tiering architecture.

   W . A . L                  R . A . F . T                  Q . U . E . U . E
 (Write-Ahead Log)         (Consensus Engine)             (Message Semantics)
  - Sequential I/O          - Majority Quorum              - FIFO per Queue
  - Flip-Flop Compaction    - Fast-Path Commit             - Visibility Leases
  - Cold Disk Spilling      - Snapshot Log Catchup         - 30-Day Horizons

---

📑 Table of Contents

1. Core Philosophy & Architecture
2. Protocol Choice: Why Postcard over TCP?
3. Consensus Engine: Raft vs Gossip
4. In-Memory Sharding & Kernel Profiling
5. Storage Engine & Flip-Flop Compaction
6. Visibility-Based Disk Spilling (0-BTree Design)
7. Production Readiness & Operability
8. Battle Testing & Soak Verification
9. Quick Start & Usage

---

💡 Core Philosophy & Architecture

The name walrq reflects its three foundational pillars:

• WAL (Write-Ahead Log): Every push and acknowledgement is streamed to an append-only binary log before confirmation. Cold data is tiered to disk, preventing out-of-memory crashes on unbounded surges.
• R (Raft Quorum): Strong, synchronous consistency across $N/2 + 1$ cluster nodes. No split-brain, no double-acks, and no state divergence under partition or rolling reboot.
• Q (Queue Semantics): Independent multi-tenant queues with strictly isolated head-of-line execution, sub-second visibility leases, and delayed delivery up to 30 days.

                      +-----------------------------+
                      |        walrq-client         |
                      | (32 Worker Shards + Window) |
                      +--------------+--------------+
                                     | (Postcard over TCP)
                                     v
                           +--------------------+
                           |   Node 1 (Leader)  |
                           +---------+----------+
                                     | (Raft AppendEntries over TCP)
                      +--------------+--------------+
                      |                             |
                      v                             v
            +--------------------+        +--------------------+
            |  Node 2 (Follower) |        |  Node 3 (Follower) |
            +--------------------+        +--------------------+

---

⚡ Protocol Choice: Why Postcard over TCP?

walrq explicitly replaces general-purpose RPC stacks (gRPC, Tonic, Protobuf) with Postcard over 4-byte length-prefixed TCP:

Metric / Dimension	Protobuf + gRPC (HTTP/2)	Postcard over TCP (walrq)	Engineering Impact
Throughput	~180,000 ops/sec	670,000 ops/sec	+272% throughput improvement
Serialization Overhead	High (Field tags, proto v-tables)	Zero-copy / LEB128 varints	Tighter wire footprint
Framing Overhead	HPACK headers, HTTP/2 multiplexing frames	4-byte Big-Endian length header	Microsecond socket dispatch
Compilation & Codegen	External .proto files, protoc, build stubs	Pure Rust enum Request / Response	Zero build friction, instant builds

Why Postcard?

• Zero Schema Generation: Request and Response are standard Rust enums annotated with #[derive(Serialize, Deserialize)].
• Compact Wire Encoding: Postcard leverages LEB128 varints. Identifiers, terms, timestamps, and payload lengths occupy 1–2 bytes rather than fixed 8-byte ints.
• Persistent Socket Pooling: walrq-client uses persistent TCP connection pools with TCP keepalive and TCP_NODELAY, bypassing connection handshake penalties.

---

🛡️ Consensus Engine: Raft vs Gossip

Earlier designs explored gossip-based synchronization (e.g., SWIM/Scuttlebutt). For strict queue semantics, gossip fundamentally breaks:

• Double Acknowledgements: Under network jitter, two consumers can poll the same message on partition boundaries.
• Tombstone Bloat: Deleted message IDs accumulate endlessly across nodes.

The Pure Raft Implementation:

1. Synchronous Quorum Commitment: Every push proposal is replicated across a majority ($N/2 + 1$) before returning success to the producer.
2. Fast-Path Early Quorum Exit: As soon as a quorum of peers responds positively, the leader completes the caller's request without waiting for lagging nodes.
3. Compacted Snapshot Log Catchup: When followers fall behind due to network partition, the leader sends in-memory snapshots (compact_log_snapshot), truncating Raft logs and preventing memory leaks.
4. Follower Redirection: Followers automatically redirect client requests to the active leader via Response::Redirect { leader }.

---

🚀 In-Memory Sharding & Kernel Profiling

32-Shard Concurrent Architecture

The in-memory engine splits queue state across 32 independent partitions (NUM_SHARDS = 32):

• Each shard contains its own Mutex<ShardState>, TimerWheel, and queue hashmap.
• Queues hash to shards via 64-bit FNV-1a.
• Zero Head-of-Line Blocking: Two queues mapped to the same or different shards operate with distinct VecDeque<Ulid> ready buffers.

The getentropy Kernel Trap Discovery

During early profiling at ~533k ops/sec, profilers identified massive kernel lock contention inside the OS kernel:

c_thread_start -> ... -> sys_getentropy -> spinlock contention

• The Culprit: Uuid::new_v4() called on every poll receipt was trapping into the kernel CSPRNG (/dev/urandom / getentropy).
• The Solution: Switched receipt handles and message IDs to user-space monotonic Ulid::new().
• The Result: Bypassed kernel lock traps entirely, increasing throughput from 533k ops/sec to 670k ops/sec (+25.6%).

---

💾 Storage Engine & Flip-Flop Compaction

walrq uses an ultra-compact append-only binary Write-Ahead Log:

+------------------+-----------------------------------------------------------+
| Record Type      | Binary Disk Format                                        |
+------------------+-----------------------------------------------------------+
| Manifest         | [2B queue_id (u16)][2B name_len (u16)][UTF-8 queue_name]  |
| Push Entry (27B) | [1B magic 0x01][16B ULID][2B queue_id][8B visible_at][payload] |
| Ack Entry (19B)  | [1B magic 0x02][16B ULID][2B queue_id]                    |
+------------------+-----------------------------------------------------------+

Zero-Leak Flip-Flop Compaction:

• When the active log segment reaches max_wal_segment_size (e.g., 64MB):
  1. The server opens an alternate segment (Segment B).
  2. Active unacknowledged messages are copied forward.
  3. The exhausted segment (Segment A) is truncated to 0 bytes in $O(1)$ time via set_len(0).
• Guaranteed Bounded Footprint: Disk usage is strictly capped at $\le 2 \times \text{MaxSegmentSize}$.

---

🌊 Visibility-Based Disk Spilling (0-BTree Design)

To handle unbounded push spikes without crashing or allocating unbounded RAM, walrq implements a visibility-based cold disk tiering system:

The Architectural Rejection of BTrees

Instead of maintaining in-memory pointer structures like BTreeMap<u64, Vec<Pointer>>, walrq adopts a sequential design:

1. 5-Minute Hot Horizon Window:
   • Messages maturing within the next 300 seconds (visible_at <= now + 300s) are stored in RAM across 1-minute resolution buckets (hot_delay_buckets).
2. Far-Future Messages Spill to Disk (0 RAM):
   • Messages scheduled beyond 5 minutes (visible_at > now + 300s) are written to the durable disk WAL and omitted from RAM (0 bytes heap allocated).
3. Sequential Streaming Hydration:
   • When the ready deque needs messages, the engine streams sequentially from spill_read_offset in the WAL file, hydrating records maturing into the current 5-minute horizon.
4. In-Place Updates (Zero Runtime Heap Churn):
   • On ack, message payloads are deallocated immediately in $O(1)$.
   • Queue offsets and metrics update via atomic integers in-place.

---

🏭 Production Readiness & Operability

walrq includes production-grade reliability primitives:

1. Graceful Shutdown & Clean Drain:
   • Intercepts Unix SIGINT (Ctrl+C) and SIGTERM.
   • Halts the socket accept loop, drains in-flight requests, flushes pending disk WAL buffers, and commits Raft log snapshots before exiting.
2. File Descriptor Guard (max_connections):
   • A tokio::sync::Semaphore caps concurrent socket connections (default: 10,000), rejecting excessive connections gracefully rather than crashing with OS EMFILE (Too many open files).
3. Optional Prometheus Telemetry (ENABLE_METRICS=true):
   • Standard RFC 0029 compliant Prometheus text exposition format exposing counters (pushed, acked, polled, wal_compactions) and gauges (in_flight, hot_messages_in_ram, is_leader).
   • Disabled by default to protect high-throughput workloads from telemetry scraping overhead.
4. Dynamic Raft Cluster Membership:
   • Add (JoinCluster) or remove (LeaveCluster) cluster nodes on the fly without restarting existing nodes.

---

🧪 Battle Testing & Soak Verification

The walrq test harness includes 48+ test suites validating edge cases under extreme conditions:

• Continuous 57.7 Million Operation Soak Test (soak_2min_test.rs):
  • 480,000 ops/sec maintained under steady client load.
  • RSS memory plateaued stably at ~1.05 GB with disk usage bounded under 16 MB.
• Cyclone Strobe Chaos (cyclone_strobe_chaos_tests.rs):
  • Raft cluster subjected to 80ms node kills and revivals under 12 concurrent workers with zero uncommitted state leaks.
• WAL Hole-Punching & Catastrophic Bit-Flip Repair (ultimate_catastrophic_repair_tests.rs):
  • Injected random bitflips into binary WAL logs and tested autonomous checksum verification and clean recovery.
• Real-Life 20-Tenant Producer Surge (real_life_multi_tenant_chaos_tests.rs):
  • 10,000 messages burst across 20 distinct queues with a strict 500-message RAM cap (95%+ spilled to cold WAL).
  • 30% of worker visibility leases abandoned/dropped.
  • Exactly 10,000 / 10,000 messages drained and verified with zero duplicates and zero lost data.

---

🚀 Quick Start & Usage

Running the Server

# Build release binaries
cargo build --release

# Run a standalone node
DATA_DIR="./data/node1" NODE_ID="127.0.0.1:50051" ./target/release/walrq

# Run with optional Prometheus metrics enabled
ENABLE_METRICS=true DATA_DIR="./data/node1" NODE_ID="127.0.0.1:50051" ./target/release/walrq

Client SDK Usage (walrq-client)

Add walrq-client to your Cargo.toml:

[dependencies]
walrq-client = { path = "walrq-client" }
bytes = "1.10"
tokio = { version = "1.43", features = ["full"] }

use bytes::Bytes;
use walrq_client::{ClientConfig, WalrClient};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = WalrClient::with_config(
        vec!["127.0.0.1:50051".to_string()],
        ClientConfig::default(),
    );

    // Push message (with 0s delay)
    let msg_id = client.push("orders", Bytes::from("order #1001"), 0).await?;
    println!("Pushed message: {}", msg_id);

    // Poll messages (visibility lease = 30 seconds)
    let messages = client.poll("orders", 30, 10).await?;
    for msg in messages {
        println!("Received: {:?}", String::from_utf8_lossy(&msg.payload));

        // Acknowledge receipt
        client.ack("orders", &msg.message_id, &msg.receipt_handle).await?;
    }

    Ok(())
}

---

👥 What This Feels Like as a Consumer (Worker Guide)

If you are writing worker code that consumes messages from walrq, here is what you need to know:

1. The Good

• Super Fast Polls: When messages are ready in memory, poll() returns in microseconds. Workers never sit idle waiting on the queue.
• Server Never Crashes on Spikes: If producers suddenly push millions of messages while your workers are slow, the server spills the overflow safely to disk. It will not run out of memory (OOM) or drop your jobs.
• Protection Against Slow/Stuck Workers (receipt_handle):
  • When you poll a message, you get a temporary lease token (receipt_handle).
  • If your worker freezes or takes longer than the timeout, the lease expires and another worker picks up the job.
  • If the old worker wakes up later and tries to ACK, the server rejects it. Your database state stays safe.
• No Duplicate Deliveries: Powered by strict Raft consensus. Network glitches will not deliver ghost duplicates.

2. Things You Must Keep in Mind

• Always Poll in Batches: Do not poll 1 message at a time in tight loops. Network round-trips will slow you down. Poll in batches of 10 to 50 items for maximum speed: client.poll("orders", 30, 50).
• 1-Second Timer Minimum: Delay timers and visibility timeouts tick in 1-second steps. You cannot set a sub-second timeout (like 200ms). Use whole seconds: 5s, 30s, 60s.
• No History Replay: Once you ACK a message, it is deleted forever. You cannot rewind the queue to re-read yesterday's messages like Kafka.
• Slight Delay on Deep Backlogs: Polling messages from memory takes $<0.1\text{ms}$. If you are draining a huge backlog that spilled to disk, reading the disk batch takes 1–3ms.

---

📄 License

This project is licensed under the MIT License — free to use, copy, modify, and distribute, with attribution required to the original repository.