Distributed Message Queue

A Kafka-style message queue built in Go. Phase 1 (single-node log engine + broker + TCP server) is complete. Phase 2 (exactly-once via 2PC) is in progress — design complete, implementation next.

The full lifecycle: keyed produce (watch FNV-1a partition assignment), consume merged across partitions, and byte-level on-disk validation. Run it yourself: ./scripts/demo.sh — recorded 2026-08-26.

Detailed ADRs with commit links: docs/decisions/.

Append-only segment files — Kafka's core insight: sequential I/O is 10-100x faster than random. Every other optimization (zero-copy, batching) is secondary to this one choice.

OS page cache over application cache — Kafka doesn't implement its own cache. It relies on the OS page cache and sendfile() for zero-copy transfers. The broker never touches message bytes on the consume path. This works because the broker is a dumb file server — no filtering, no transformation.

Pull over push — Consumers pull at their own pace. The broker stays stateless (no per-consumer delivery tracking). Consumers store their own offsets in the log.

Exactly-once via 2PC — Kafka chose at-least-once + idempotent producers as a practical compromise (20-30% throughput cost). This project implements full 2PC to show what the real version costs: 10-100x throughput reduction, 2-5x latency increase, blocking on coordinator failure.

Per-partition locking — Each partition has its own mutex; parallelism comes from multiple partitions, not splitting one. Measured contention: ~12% under 4-goroutine parallel writes (PROGRESS.md) — negligible compared to the 86% syscall cost on the append path.

Producer --> Server (goroutine per conn, synchronous)
                  |
                  v
            Broker --> Topic --> Partition x N
                                    |
                                    v
                          append-only segment files
                          (.log + .index + .timeindex)
                                    |
                                    v
Consumer <-------- [read from offset] <----- OS page cache

Each layer has its own lock (see Concurrency Model). Parallelism comes from multiple partitions, not from pipelining within one. See architecture overview for the full diagram with layer responsibilities and invariants.

# Run all tests (with race detector)
cd message-queue
go test ./... -race

# Start the server
go run ./cmd/server/ -addr :9092 -data ./data

# Run tests with coverage
go test ./... -coverprofile=coverage/coverage.out
go tool cover -html=coverage/coverage.out

# Fuzz the serialization layer (catches panics on malformed input)
go test ./pkg/log/ -fuzz=FuzzDecodeMessage -fuzztime=30s
go test ./pkg/server/ -fuzz=FuzzReadFrame -fuzztime=30s

Data stays inside the container volume — nothing written to the host disk.

docker compose up --build -d   # server on :9092, data in mq-data volume
docker compose down            # stop (volume persists)

scripts/demo.sh runs the full lifecycle against the container — build the CLI client, create a topic with 3 partitions, produce keyed messages (watch the broker's FNV-1a partition assignment), consume everything merged across partitions, and verify the bytes at rest with mqvalidate:

./scripts/demo.sh

Sample output:

=== produce keyed messages (note partition assignments) ===
produced to demo-events-1787715727 partition=1 offset=0
produced to demo-events-1787715727 partition=1 offset=1
...
-- 4 message(s) across 3 partition(s), merged by timestamp

4 topics, 12 segments, 6 messages, 246 bytes, 0 problems
on-disk data: OK

The same client works by hand:

go run ./cmd/mqclient produce -addr localhost:9092 -topic t -key user-1 -value hello
go run ./cmd/mqclient consume -addr localhost:9092 -topic t -all

Ordering guarantee (matching Kafka): messages are ordered per partition — and per key when producing with keys, since a key always routes to the same partition. There is no global ordering across partitions; consume --all merges by timestamp on a best-effort basis. The raw asciicast of the demo is available (docs/demo.cast) for terminal players (asciinema play docs/demo.cast).

TestExternalServer runs a create/produce/consume round-trip against a live server. It is skipped unless MQ_ADDR is set:

docker compose up --build -d
MQ_ADDR=localhost:9092 go test -count=1 -run TestExternalServer -v ./pkg/server/

-count=1 disables Go's test cache — required when testing against external infrastructure, since the cache cannot see that the container (or its data) changed.

• Kafka Paper (2011)
• Jay Kreps - I Love Logs (Stanford slides)
• Kafka Internals (Confluent)