Committed from Day 1. Everything else builds on this (retention, compaction, replication, 2PC).
Append-only = sequential I/O (Kafka's core insight from the paper).
- Storage Layout (line 26)
- Message Format (line 42)
- Index Files (line 54)
- Timestamp Strategy (line 72)
- Operations (line 85)
- Normal Flow (line 87)
- Recovery Flow (line 120)
- Delete (retention) (line 168)
- Segment Immutability (line 102)
- Segment Size (line 111)
- OS Layer Optimizations (line 121)
- Write Contention (line 129)
- Batch + Single Writer (Phase 1) (line 131)
- Lock-Free (future optimization) (line 143)
- Partition Scaling (line 161)
- Partition Reassignment (line 171)
- Failure Modes (line 195)
- Dependency Flow (line 27)
- Concurrency Model
Broker --> Topic --> Partition --> Segment --> file
Dependencies flow one direction. Each layer owns a single concern:
- Broker (exported): topic management, map coordination, request routing. Entry point for external packages.
- Topic (unexported): naming, partition routing (FNV-1a key hash, round-robin nil key)
- partition (unexported): offset assignment, segment rotation, recovery orchestration
- segment (unexported): byte-level storage, CRC validation, index management
No layer depends on a higher layer. segment doesn't import partition. partition doesn't import Topic. Topic doesn't import Broker. This makes each layer independently testable and replaceable.
Directory ownership flows the same direction:
- Broker (via createTopic): creates
data/<topic>/ - partition: creates
data/<topic>/<partition>/ - segment/index/timeindex: create files only, never directories
Two-layer locking with clean separation of concerns.
Protects the topics map only. Held briefly for map lookup or insert. No disk I/O under this lock.
Broker.Produce("my-topic", key, val)
b.mu.RLock() // <-- lock
t := b.topics["my-topic"]
b.mu.RUnlock() // <-- unlock (before disk I/O)
t.Produce(key, val) // topic has its own lock
Broker.CreateTopic("new-topic", 3)
topic = createTopic(...) // disk I/O (no lock held)
b.mu.Lock() // <-- lock
b.topics["new-topic"] = topic
b.mu.Unlock() // <-- unlock
Protects the topic during operations and lifecycle.
- Produce, Consume, NextOffset, NumPartitions: acquire mu.RLock (multiple reads can proceed concurrently)
- close: acquires mu.Lock (exclusive — waits for in-flight ops)
This ensures a topic cannot be closed while Produce/Consume is in flight.
A single broker-wide lock would block all operations during disk I/O:
// BAD: disk I/O under broker lock
b.mu.Lock()
createTopic(...) // disk I/O — ALL produces/consumes blocked
b.mu.Unlock()
Two layers separate concerns:
- Broker lock: map safety only (microseconds)
- Topic lock: operation safety (held for the duration of an append or read, including its disk writes; there is no fsync on these paths)
| Concern | Mechanism |
|---|---|
| Topics map (create vs read) | Broker.mu (RWMutex) — held briefly for map ops |
| Topic lifecycle (close vs use) | Topic.mu (RWMutex) — close waits for in-flight |
| Partition writes (concurrent) | partition.mu — one lock per partition; measured contention ~12% under parallel writers, negligible |
| Round-robin counter | atomic.AddUint64 (lock-free hot path) |
-
CreateTopic holds no lock during disk I/O: If two goroutines create the same topic concurrently, one succeeds and the other cleans up its partially-created files. O_EXCL on segment files prevents duplicate segments. The map check under the lock prevents duplicate map entries.
-
Close waits for in-flight ops: Topic tracks in-flight operations with an atomic counter; Close acquires the write lock, so it waits for appends/reads in progress. There is no fsync on the append path (sync happens at segment Close/rotation), so waits are short. Close is a shutdown operation, not a hot path.
-
Per-topic lock serializes across partitions: Two goroutines producing to different partitions of the same topic hold the same Topic.mu.RLock. This is fine — the lock is a read lock, so both proceed concurrently. Only close serializes (write lock).
Two os.MkdirAll calls total. Everything below partition operates on
files within an existing directory.
Export boundary: Broker is the entry point for external packages (server, mqclient, validate). Topic, PartitionOption, and the consumer record types are also exported. partition and segment are internal implementation details. The server talks only to Broker: CreateTopic, Produce, Consume, NextOffset, NumPartitions, Topics, Close.
data/
my-topic/
0/ # partition 0
00000000000000000000.log # segment (base offset)
00000000000000000000.index # offset -> byte position
00000000000000000000.timeindex # offset -> timestamp
- Naming: base offset, zero-padded 20 digits
- Active segment: only one accepts writes
- Closed segments: immutable, can be deleted/compacted independently
- Rotation: new segment when current reaches DefaultSegmentSize (1GB), size-based only. (Time-based rotation is a future option; Kafka rotates on size or time.)
| offset (8B) | size (4B) | crc (4B) | timestamp (8B) | keylen (4B) | key (var) | value (var) |
- Offset: sequential, unique per partition, assigned by broker
- Size: total bytes of key + value (excludes header)
- CRC: CRC-32 IEEE over timestamp + key + value
- Timestamp: Unix nanosecond, assigned by broker
- Key: variable length, used for partition routing
- Value: message payload
Why CRC-32:
- Detects random bit-rot on disk (our use case)
- Hardware-accelerated on x86/ARM (CRC32 instruction, ~nanosecond)
- 1-in-4-billion undetected error rate
- Standard for storage systems (ext4, ZFS, Ethernet)
- SHA-256 is 10x slower for no benefit — nobody is tampering with segment files
- If this were a signed/audit log, we'd use SHA-256. For storage integrity, CRC-32 is correct.
Sparse in structure — entries written every Nth message (WithIndexInterval(n),
count-based since there's no limit on message size). The default interval is 1,
i.e. every append writes an index entry: with the append path measured 86%
syscall-bound, the extra pwrites are cheap relative to the .log write, and
dense indexes keep reads simple. Profiling (PROGRESS.md, 08/25) showed the
read cost is a ReadAt per binary-search probe — the Phase 2.5 optimization
target is in-memory index reads, not sparsity tuning.
.index — offset -> byte position in .log (16 bytes per entry: | offset (8B) | byte_position (8B) |)
- Used for offset-based reads: "give me message at offset X"
.timeindex — offset -> timestamp (16 bytes per entry: | offset (8B) | timestamp (8B) |)
- Used for time-based operations: retention, replay from time
- Binary search finds approximate offset, then scan to exact message
- O(log n) vs O(n) without it
Range query flow (time-based):
- Binary search timeindex for T1 -> approximate offset (sparse, not exact)
- Binary search timeindex for T2 -> approximate offset
- Scan forward in .log to find exact message with timestamp >= T1
- Read until timestamp > T2
Big-endian byte order — consistent with message format. Standard for file formats and network protocols. Little-endian would be marginally faster on x86/ARM (native byte order, no byte-swap on read), but the performance difference is negligible for sparse index lookups. Big-endian is the safer choice for a format that might be read by different tools or on different platforms.
- Broker's local time at append (
time.Now()), not producer's time - Producer clocks may be wrong; network delay makes producer timestamp unreliable
- Broker is source of truth for "when was this message received"
- Wall clock for human queries ("messages from 2pm")
Go time.Now() internals (since Go 1.9):
- Wall clock: seconds since epoch (human-readable)
- Monotonic clock: from runtime, always increases, immune to NTP
- In memory: both components, safe for duration math (
time.Since()) - On disk: monotonic stripped on serialization, wall clock only
Producer --> Partition.Append(msg)
|
+--> active.NextOffset() (offset = baseOffset + messageCount)
+--> msg.Encode() (serialize: offset | size | crc | ts | key | value)
+--> active.file.Write() (append to .log)
+--> active.idx.write() (sparse: offset -> byte position)
+--> active.tidx.write() (sparse: offset -> timestamp)
+--> active.messageCount++
+--> active.size += len(encoded)
+--> if size >= segmentSize: rotate()
Rotate:
- Close active segment (sync + nil file/idx/tidx handles)
- Append to closed list
- Create new segment at baseOffset = active.NextOffset()
- New segment becomes active
Read:
- Find segment: check active first (offset >= baseOffset), else binary search closed list
- Open .log on demand if closed (file was nil'd)
- Open .index on demand if closed (for position lookup)
- Index lookup -> byte position -> scan forward to exact offset -> decode
State derived from segment:
NextOffset()= baseOffset + messageCount (single source of truth)- No separate counter in Partition
Recover(dir)
|
+--> os.ReadDir(dir) (find all .log files, already sorted)
+--> parse base offsets from filenames
+--> for each segment:
if last (active):
recoverActiveSegment() (O_RDWR for writes)
else:
RecoverSegment() + Release() (O_RDONLY, then free fds)
recoverActiveSegment (last segment):
- Open .log as O_RDWR (needs write access for new appends)
- Open .index and .timeindex as O_RDWR (for truncation), fall back to O_CREATE if missing
- scanSegment: validate CRC, count messages, calculate size
- Truncate index files to validEntries (messageCount / indexInterval)
- Close indexes, reopen as O_RDWR (for future writes)
- Set messageCount on Segment struct
RecoverSegment + Release (closed segments):
- Open .log as O_RDONLY
- Open .index and .timeindex as O_RDONLY, fall back to O_CREATE if missing
- scanSegment: validate CRC, count messages, calculate size
- Truncate index files to validEntries
- Close indexes, reopen as O_RDONLY
- Release: close all file handles, nil fields (free fds)
- Read opens files on demand when needed
State after recovery:
- Active segment: writable, messageCount from scan, NextOffset = baseOffset + messageCount
- Closed segments: read-only, messageCount from scan, file/idx/tidx nil'd
- Partition.NextOffset() = active.NextOffset() (derived, not stored)
Corruption handling:
- scanSegment stops at first CRC failure or offset gap
- messageCount = valid messages before corruption
- Index files truncated to validEntries (removes stale entries)
- Corrupted bytes remain in .log file (dead space, harmless)
What survives a crash:
- All fsync'd data in segments (fdatasync window = recent unflushed writes)
- Index files are rebuilt from .log on recovery (scanSegment)
- No WAL needed — segment files ARE the durability mechanism
- Find segments with max timestamp < now - retention
- Delete entire segment files (.log, .index, .timeindex)
Closed segments are immutable by convention — after Close(), no Go code holds a writable file handle. Recovery (RecoverSegment) reopens files as O_RDONLY.
This is sufficient because:
- Single-process system — only our code opens these files
- No locking needed during copy (reassignment) — segments are stable
- Recovery truncates index files to remove stale entries beyond corruption
OS-level enforcement (chmod 0444) was considered but removed because it blocks recovery truncation (needs O_RDWR). The Go-level guarantee is the right tradeoff.
| Size | Pros | Cons |
|---|---|---|
| <100MB | Fast retention, quick recovery | Many files, metadata overhead |
| >10GB | Fewer files, less metadata | Slow recovery, slow retention |
| 1GB (sweet spot) | Balance | Kafka's default |
Rotation: size-based (1GB soft limit) primary, time-based (1 hour) secondary.
Soft limit, never split messages — 1GB is a target, not a hard cap. If a message doesn't fit in the remaining space, rotate to a new segment and write the whole message there. No message ever spans segments. If a single message is larger than 1GB, the segment grows beyond the target. Recovery is still fast — sequential scan is O(n) in message count, not file size.
Retention is per-segment — delete a segment only when ALL messages in it are expired. If one message in a10GB segment is still in the retention window, the whole segment stays. Storage is cheap. Per-message retention (compaction, truncation) adds complexity not worth it for most use cases.
fadvise(SEQUENTIAL)— hint to kernel, improves readahead- Filesystem block alignment — align to 4KB blocks
fdatasyncoverfsync— sync data, skip metadata, faster- Pre-allocate —
ftruncate()to full size, avoids fragmentation - Write buffer — match block size (4KB) or multiples (64KB, 256KB) for DMA
Producer A --\
Producer B ---+--> [buffer channel] --> single writer --> segment file
Producer C --/
- Producers send to channel (Go channels have internal locks, no explicit mutexes)
- Single goroutine receives and appends (no lock needed, only writer)
- Amortizes fsync cost across batches
Producer side:
linger.ms— wait this long before sending (batches accumulate). Default 0 = send immediately.batch.size— max batch size in bytes. If hit before linger expires, send early.- Producer accumulates messages in a buffer, sends them as one ProduceRequest.
Broker side:
- Receives a batch of messages in one request
- Appends all messages to the log in one sequential write
- One fsync covers the entire batch (if acks=all)
Why it matters:
- Network: one round-trip instead of N
- Disk: one sequential write instead of N
- fsync: one fsync instead of N (the expensive part)
Compression:
- Compresses the entire batch, not individual messages
- Better ratios (more data to compress)
- Decompression cost is amortized across the batch
Tradeoff: latency. linger.ms=5 adds up to 5ms latency but dramatically improves
throughput. Kafka defaults to linger.ms=0 for low latency, but most production
configs set it to 5-100ms.
Our Phase 1 (as built): fully synchronous — the server executes one request at a time per connection and calls straight down into partition.Append; there are no channels or writer goroutines (that was the pre-implementation plan; contention measured negligible, syscalls are the real cost). Batching is deferred to Phase 2.5, where the measured knee (PROGRESS.md, 08/25: 16-64 records per write) will drive buffered writes / group commit. The Kafka material below is retained as design grounding.
Kafka consumer fetches batches (fetch.min.bytes, max.partition.fetch.bytes).
A recovered consumer catching up needs batch reads — one syscall per N messages,
not one per message.
Design:
Segment.ReadBatch(offset, maxBytes) -> []ConsumeRecord, nextOffset
- Reads sequentially from offset until maxBytes reached or end of segment
- Returns slice of messages + next offset to continue from
- Amortizes index lookup (one lookup, then sequential scan)
- Reduces syscall overhead (one ReadAt per batch, not per message)
Why it's a must-have:
- Catch-up after restart: consumer needs thousands of messages, not one at a time
- Throughput: sequential I/O is the whole point — batch reads exploit it
- Kafka's
fetch.min.byteswaits until batch fills — same idea
Cross-segment batches:
- If batch reaches end of segment, return what we have
- Caller (Partition) handles switching to next segment
- No complex cross-segment seeks — just return partial batch and let caller retry
Atomic counter + pwrite() for parallel writes without locks:
pos := atomic.AddInt64(&offset, msgSize)— claim positionpwrite(buf, pos)— write at claimed position (POSIX syscall, no file pointer movement)- No mutex needed
pwrite — POSIX syscall, writes at specific offset without moving file pointer.
Go API: file.WriteAt(buf, offset) wraps pwrite, file.Write(buf) wraps write.
Gaps problem: Writer crashes between claiming and writing -> gap in segment.
Pre-write solution: pwrite(zeros, pos, size) then pwrite(message, pos, size). Crash leaves zeros (empty but valid).
pwrite vs sequential: Not a middle ground — it's random I/O with better API. But with atomic counter appending to end, it's effectively sequential.
Not worth it for Phase 1.
Don't split a partition further. Parallelism comes from multiple partitions, not splitting one.
- Each partition: one log, one leader
- Consumer groups distribute partitions across consumers
- Splitting breaks ordering guarantees within the partition
One partition = one ordered log. Keep it simple.
Moving a partition from one broker to another — for load balancing, broker decommissioning, or failure recovery.
- Controller tells old broker to stop serving reads/writes for the partition
- Old broker closes the active segment — all segments are now immutable
- Copy the segment directory to the new broker (safe — segments are immutable)
- New broker calls RecoverSegment() on each segment — recovery scan validates CRC, counts messages, truncates indexes
- Controller updates metadata to point to new broker
Messages produced but not yet acknowledged during reassignment:
- Complete before active segment closes — written to old broker, consumed from new broker
- Fail after close — producer receives NOT_LEADER error, retries on new broker
- Broker coordinates: stop accepting produces, drain in-flight, close active segment
- Partition is just a directory of files — copy them, scan them, done
- No WAL or in-flight state to transfer — segments are self-contained
- RecoverSegment() + scanSegment already handles recovery — validates everything
- Index files truncated on recovery to remove stale entries beyond corruption
Segment knows bytes, broker owns offset — the segment file is sequential bytes. Offset assignment (0, 1, 2, ...) is broker-level state. The broker tracks nextOffset per partition and assigns it at append time. The segment just appends bytes.
Recovery is sequential scan — on restart, scanSegment reads through the file
to count valid messages and total size. This is fast regardless of file size because
sequential I/O scales linearly. A10GB segment with 100 messages scans in microseconds.
Three messages with keys "user-1", "user-2", "user-1":
.log file (sequential, append-only):
Byte pos | Content
----------|----------------------------------------
0 | msg0: offset=0, size=11, crc=..., ts=..., key="user-1", value="hello" (39 bytes)
39 | msg1: offset=1, size=11, crc=..., ts=..., key="user-2", value="world" (39 bytes)
78 | msg2: offset=2, size=9, crc=..., ts=..., key="user-1", value="foo" (37 bytes)
115 | (next write goes here)
.index file (sparse, e.g. every 1 message for simplicity):
Offset | Byte Position
-------|--------------
0 | 0
1 | 39
2 | 78
.timeindex file (sparse, same interval):
Offset | Timestamp
-------|------------------
0 | 1700000000000000000
1 | 1700000000100000000
2 | 1700000000200000000
To read offset 2: binary search .index -> byte position 78 -> seek to 78 in .log -> read message.
To read "all messages after time T": binary search .timeindex -> approximate offset -> scan .log forward.
| Failure | Impact | Recovery |
|---|---|---|
| Broker crash | Segment files survive on disk | Reopen segments, check partial writes, rebuild index |
| Producer crash mid-write | Partial message in segment | CRC check catches corruption, consumer skips |
| Consumer crash | No impact on broker | Resume from last committed offset |
| Disk dies | Data gone | No recovery (no replication in Phase 1) |
| Power loss | Unwritten page cache data lost | fdatasync minimizes window |
Phase 1 is vulnerable to machine failure. Explicit tradeoff for simplicity. Phase 2 (replication) fixes this.
| Guarantee | Behavior | Tradeoff |
|---|---|---|
| At-most-once | Send and forget. No ack, no retry. | Lowest latency, can lose messages |
| At-least-once | Send, wait for ack, retry on timeout. | Duplicates on broker crash mid-ack |
| Exactly-once | Coordinator ensures all-or-nothing. | Coordinator + WAL + 2 round trips |
At-most-once: Producer sends, doesn't wait for ack. If broker crashes before writing, message is lost. No retries. Used when latency matters more than correctness (metrics, logging).
At-least-once: Producer sends, waits for ack. If timeout, producer retries. If broker crashed after write but before ack, retry creates duplicate. Consumer must deduplicate (idempotency key). Simple, works for most use cases.
Exactly-once: Coordinator orchestrates prepare/commit. Log writes to WAL first (prepare), then flushes on commit. Retries hit same transaction ID, coordinator deduplicates. Costs 2 round trips + fsync before ack.
Producer -> Coordinator: send message
Phase 1 (Prepare):
Coordinator -> Log: prepare(data)
Log: write to WAL -> vote YES
Coordinator: all YES -> proceed
Phase 2 (Commit):
Coordinator -> Log: commit
Log: flush to segment
If coordinator crashes between prepare and commit:
Log holds uncommitted entry
Coordinator recovers from WAL, resolves pending
Default: more partitions than consumers.
- 6 partitions, 3 consumers -> each gets 2 (parallel)
- 3 partitions, 6 consumers -> 3 idle (waste)
- More partitions = room to grow (add consumers without re-partitioning)
- Operator decides count, engine just supports N partitions per topic