Segmented log

Deep dive

How it works

In: records to append, offsets to read, a retention limit · Out: record offsets, payloads, deleted segments · State: segment files <base offset>.log sorted by base offset, only the newest (active) one written; per segment a sparse index of (offset - base, file position) every index_every bytes, in memory or mmap'd, plus an optional time index (max timestamp, offset) · Cost: append O(1); read O(log segments + log entries + a scan of at most index_every bytes); retention O(files deleted)

Records are [len][crc][offset][payload]. The segment list is copy-on-write so readers keep a stable view.

1. Append: if the active segment is full, flush it and create a file named by the next offset. Write the record; add an index entry if index_every bytes passed since the last.
2. Read(offset): binary search the list for the last base ≤ offset, then its index for the last entry ≤ offset; scan forward.
3. Retention: delete sealed segments wholly beyond the age or size limit, never the active one.
4. Recovery: scan the active segment, verify each CRC, truncate at the first bad record. Damage in a sealed segment is corruption: alert.
5. Durability: fsync is a policy: per batch, per interval, or never when replicas provide it.

00000000000000000000.log  sealed   index [0→0][35→4096][70→8192]...
00000000000000524288.log  sealed   read(524301): base 524288 → index → scan
00000000000001048576.log  active   retention unlinks files from the top

Code

Compiled and run with asserts for roll, cross-segment lookup, retention, reopen, torn tail, bit flip, and crc32("123456789") = 0xCBF43926.

use std::fs::{self, File, OpenOptions};
use std::io;
use std::os::unix::fs::FileExt;
use std::path::{Path, PathBuf};

pub fn crc32(data: &[u8]) -> u32 { // bitwise IEEE; real code uses a table or hardware CRC
    let mut crc = !0u32;
    for &b in data {
        crc ^= b as u32;
        for _ in 0..8 { crc = (crc >> 1) ^ (0xEDB8_8320 & (crc & 1).wrapping_neg()); }
    }
    !crc
}

const HDR: usize = 16; // len u32 | crc u32 over offset+payload | offset u64
struct Segment {
    base: u64,
    next: u64,                // next offset to assign
    size: u64,
    index: Vec<(u32, u64)>,   // sparse: (offset - base, file position)
    since: u64,               // bytes since the last index entry
    file: File,
}
pub struct Log { dir: PathBuf, max_bytes: u64, index_every: u64, segs: Vec<Segment> }
impl Segment {
    fn note(&mut self, pos: u64, bytes: u64, every: u64) {
        if self.index.is_empty() || self.since >= every {
            self.index.push(((self.next - self.base) as u32, pos));
            self.since = 0;
        }
        self.since += bytes;
    }
}
fn seg_path(dir: &Path, base: u64) -> PathBuf { dir.join(format!("{base:020}.log")) }
impl Log {
    /// Rebuilds each sparse index by scanning; cuts a torn tail off the last segment.
    pub fn open(dir: &Path, max_bytes: u64, index_every: u64) -> io::Result<Log> {
        fs::create_dir_all(dir)?;
        let mut bases: Vec<u64> = fs::read_dir(dir)?
            .filter_map(|e| e.ok()?.file_name().to_str()?.strip_suffix(".log")?.parse().ok()).collect();
        bases.sort();
        let mut log = Log { dir: dir.into(), max_bytes, index_every, segs: vec![] };
        let last = bases.len().saturating_sub(1);
        for (n, base) in bases.into_iter().enumerate() {
            let file = OpenOptions::new().read(true).write(true).open(seg_path(dir, base))?;
            let mut s = Segment { base, next: base, size: 0, index: vec![], since: 0, file };
            let data = fs::read(seg_path(dir, base))?;
            let mut pos = 0usize;
            while pos + HDR <= data.len() {
                let len = u32::from_le_bytes(data[pos..pos + 4].try_into().unwrap()) as usize;
                let crc = u32::from_le_bytes(data[pos + 4..pos + 8].try_into().unwrap());
                if pos + HDR + len > data.len() || crc32(&data[pos + 8..pos + HDR + len]) != crc { break; }
                s.note(pos as u64, (HDR + len) as u64, index_every);
                (s.next, pos) = (s.next + 1, pos + HDR + len);
            }
            if pos != data.len() && n != last { // damage in a sealed segment is corruption
                return Err(io::Error::new(io::ErrorKind::InvalidData, "corrupt sealed segment"));
            }
            s.file.set_len(pos as u64)?; // torn tail of the active segment: cut it off
            s.size = pos as u64;
            log.segs.push(s);
        }
        if log.segs.is_empty() { log.roll(0)?; }
        Ok(log)
    }
    fn roll(&mut self, base: u64) -> io::Result<()> {
        let file = OpenOptions::new().read(true).write(true).create_new(true).open(seg_path(&self.dir, base))?;
        self.segs.push(Segment { base, next: base, size: 0, index: vec![], since: 0, file });
        Ok(())
    }
    pub fn append(&mut self, payload: &[u8]) -> io::Result<u64> {
        let a = self.segs.last().unwrap();
        if a.size >= self.max_bytes { let next = a.next; self.flush()?; self.roll(next)?; } // seal durably, open next
        let (every, a) = (self.index_every, self.segs.last_mut().unwrap());
        let mut rec = (payload.len() as u32).to_le_bytes().to_vec();
        rec.extend_from_slice(&[0; 4]); // crc placeholder
        rec.extend_from_slice(&a.next.to_le_bytes());
        rec.extend_from_slice(payload);
        let crc = crc32(&rec[8..]);
        rec[4..8].copy_from_slice(&crc.to_le_bytes());
        a.file.write_all_at(&rec, a.size)?; // only the active segment is ever written
        a.note(a.size, rec.len() as u64, every);
        (a.size, a.next) = (a.size + rec.len() as u64, a.next + 1);
        Ok(a.next - 1)
    }
    pub fn flush(&mut self) -> io::Result<()> { self.segs.last().unwrap().file.sync_data() }
    /// Binary search on base offset, then the sparse index, then a short scan.
    pub fn read(&self, offset: u64) -> io::Result<Option<Vec<u8>>> {
        let i = self.segs.partition_point(|s| s.base <= offset);
        if i == 0 { return Ok(None); }
        let s = &self.segs[i - 1];
        if offset >= s.next { return Ok(None); }
        let j = s.index.partition_point(|&(rel, _)| rel as u64 <= offset - s.base);
        let mut pos = s.index[j - 1].1;
        loop {
            let mut h = [0u8; HDR];
            s.file.read_exact_at(&mut h, pos)?;
            let len = u32::from_le_bytes(h[0..4].try_into().unwrap()) as u64;
            if u64::from_le_bytes(h[8..16].try_into().unwrap()) == offset {
                let mut p = vec![0u8; len as usize];
                s.file.read_exact_at(&mut p, pos + HDR as u64)?;
                return Ok(Some(p));
            }
            pos += HDR as u64 + len;
        }
    }
    /// Retention: unlink sealed segments wholly below `offset`; nothing is rewritten.
    pub fn delete_before(&mut self, offset: u64) -> io::Result<usize> {
        let n = self.segs.iter().take(self.segs.len() - 1).take_while(|s| s.next <= offset).count();
        for s in self.segs.drain(..n) { fs::remove_file(seg_path(&self.dir, s.base))?; }
        Ok(n)
    }
    pub fn segments(&self) -> usize { self.segs.len() }
    pub fn start_offset(&self) -> u64 { self.segs[0].base }
}

Complexity and constants

• Append: one pwrite. Index: one entry per index_every bytes (16 B here, about 0.4 % of the log at 4 KiB). Read: the scan is at most index_every bytes.
• Cost drivers: syscalls per record, fsync per commit, open files (KIP-33: the time index adds about 1/3 more). Retention granularity is one segment.

Efficiency

• Cost: measured on an Apple-silicon Mac, SSD, 116 B records, warm cache, code above: append without flush 3.4-4.0 µs (one pwrite); sync_data (full flush on macOS) about 4.2 ms; lookup 1.9 µs with an entry per 512 B, 6 µs per 4 KiB, 78 µs per 64 KiB, 1.3 ms per 1 MiB. Bound: flush latency when writing, syscalls when reading.
• Levers:
  • One flush per batch — 100 records per flush gave 17.7k records/s against 238/s per record, 74× (measured) — when acks can wait for the batch (group commit).
  • Replicate instead of flushing each write — fsync per write can cost two to three orders of magnitude (Kafka design docs) — when replicas fail independently.
  • Index gap — 4 KiB was 13× faster than 64 KiB, 3× slower than 512 B (measured) — largest gap that meets the read latency target.
  • sendfile on sealed segments — no user-space copy (Kafka design docs) — not with user-space TLS.
  • Delete files, not records — no read, write or copy — when age or size decides what dies.
• Naive → standard → state of the art: one file, rewrite to trim · O(log size) per trim. Size-rolled segments, sparse index, unlink retention · O(1) trim, bounded recovery. Plus preallocation, mmap'd index, group fsync, sendfile · fewer syscalls.

Implementation notes

• Fixed-width base offsets in names (Kafka: 20 digits) keep directory order equal to log order.
• Preallocate the active file (etcd: 64 MB; Kafka log.preallocate is off by default) so fdatasync skips size updates; truncate on sealing.
• fsync the directory after create, rename or delete.
• Kafka preallocates a 10 MiB index file and shrinks it on roll.

Pitfalls

• Deleting a segment a reader has open: copy-on-write list and delayed unlink (Kafka file.delete.delay.ms, 1 minute).
• Truncating at mid-log corruption drops valid later data; cut only the last segment.
• File mtime breaks after copy or restore; retain by record timestamps (KIP-33).
• Rolling before flushing the old segment can leave a gap after a crash.

Variants and alternatives

In production

• Messaging: Kafka, Redpanda, NATS JetStream (.blk files), Pulsar (ledgers dropped once all cursors consumed them).
• Databases and consensus: PostgreSQL WAL, etcd WAL, TiKV's raft-engine; observability: Prometheus write-ahead log (128 MB segments).
• Ledgers: Bitcoin Core blk?????.dat files (at most 128 MiB), pruned whole.

Implementations

Sources

• J. Kreps, N. Narkhede, J. Rao, Kafka: a Distributed Messaging System for Log Processing, NetDB 2011.
• Apache Kafka documentation: log, design, topic configs.
• KIP-33, Add a time based log index. KIP
• Docs: Redpanda disk utilization, BookKeeper concepts, PostgreSQL WAL internals, Prometheus WAL format.
• etcd server/storage/wal code; NATS server/filestore.go code; Bitcoin Core src/node/blockstorage.h code