Implementing File-Based Append-Only Logs for Partition Storage

Introduction to the Lesson

Today, we're going to build our first concrete implementation of that interface. This is where the theory of interfaces meets the practical reality of file I/O and data structures on disk.

Your Goal for Today: Your learning outcome is to implement the storage interface using a file-based append-only log for each partition.

We will construct a durable, append-only log that writes messages to files. To do this efficiently, we'll build a system inspired by the storage engines of high-performance message brokers like Kafka. This involves not just one file, but a set of coordinated files that allow for fast writes and quick lookups.

By the end of this lesson, you will have a fully functional, file-based CommitLog implementation that can be plugged directly into the Partition struct we defined previously.

1. The Anatomy of a File-Based Log: Segments

Writing everything to a single, massive log file is inefficient. It becomes difficult to manage, clean up, and recover. A much better approach is to break the log into smaller chunks called segments.

Each segment will be responsible for a contiguous range of offsets. A complete topic partition log will then be a sequence of these segments.

A key design choice is how to structure each segment. To enable fast reads, we will use two separate files for each segment:

• A store file (.store): This is where we append the raw message data.
• An index file (.index): This file maps a message's logical offset to its physical byte position in the store file.

This two-file structure is a classic pattern. The store file allows for fast, sequential writes, while the index file provides a way to quickly find a message without scanning the entire store file.

The following resource provides an excellent overview of this segmented log architecture.

The key idea is that each segment is identified by the offset of the first record it contains, known as the baseOffset. For example, a segment starting at offset 20 would have files named 20.store and 20.index.

The image below shows how an index maps logical keys (our offsets) to physical positions in a data file.

2. Implementing the Log Segment

Let's start by creating the building block: the segment. We'll create three new files in our internal/log directory:

• store.go: Manages the raw data file.
• index.go: Manages the offset index file.
• segment.go: A wrapper that coordinates the store and index.

The Store

The store is responsible for appending and reading raw byte data from a file.

Create internal/log/store.go:

// internal/log/store.go
package log

import (
	"bufio"
	"encoding/binary"
	"os"
)

var (
	enc = binary.BigEndian // Use Big Endian for network byte order consistency
)

const (
	lenWidth = 8 // Number of bytes used to store the record's length
)

type store struct {
	*os.File
	buf *bufio.Writer
	size uint64
}

// newStore creates a new store for the given file.
func newStore(f *os.File) (*store, error) {
	fi, err := os.Stat(f.Name())
	if err != nil {
		return nil, err
	}
	size := uint64(fi.Size())
	return &store{
		File: f,
		size: size,
		buf:  bufio.NewWriter(f),
	}, nil
}

// Append persists the given bytes to the store. We write the length of
// the record so that when we read the record, we know how many bytes to read.
// Returns the number of bytes written and the position where the store
// holds the record.
func (s *store) Append(p []byte) (n uint64, pos uint64, err error) {
	pos = s.size
	if err := binary.Write(s.buf, enc, uint64(len(p))); err != nil {
		return 0, 0, err
	}
	w, err := s.buf.Write(p)
	if err != nil {
		return 0, 0, err
	}
	w += lenWidth
	s.size += uint64(w)
	return uint64(w), pos, nil
}

// Read returns the record stored at the given position.
func (s *store) Read(pos uint64) ([]byte, error) {
	s.Flush() // Ensure all buffered data is written to disk
	size := make([]byte, lenWidth)
	if _, err := s.File.ReadAt(size, int64(pos)); err != nil {
		return nil, err
	}
	b := make([]byte, enc.Uint64(size))
	if _, err := s.File.ReadAt(b, int64(pos+lenWidth)); err != nil {
		return nil, err
	}
	return b, nil
}

func (s *store) ReadAt(p []byte, off int64) (int, error) {
	s.Flush()
	return s.File.ReadAt(p, off)
}

func (s *store) Flush() error {
	return s.buf.Flush()
}

func (s *store) Close() error {
	s.Flush()
	return s.File.Close()
}

Notice the length prefixing in Append. This is a crucial technique for framing messages in a stream, ensuring we know how many bytes to read for each record.

The Index

The index maps offsets to file positions. For performance, we'll use a memory-mapped file, which allows us to treat the file like an in-memory byte slice while the OS handles the actual file I/O.

Create internal/log/index.go:

// internal/log/index.go
package log

import (
	"os"
	"io"
	"github.com/tysontate/gommap"
)

var (
	offWidth uint64 = 4 // Bytes for offset
	posWidth uint64 = 8 // Bytes for position
	entWidth = offWidth + posWidth
)

type index struct {
	file *os.File
	mmap gommap.MMap
	size uint64
}

// newIndex creates an index for the given file. We memory-map the file
// and create the file if it doesn't exist yet.
func newIndex(f *os.File, c Config) (*index, error) {
	fi, err := os.Stat(f.Name())
	if err != nil {
		return nil, err
	}
	size := uint64(fi.Size())
	if err = os.Truncate(f.Name(), int64(c.Segment.MaxIndexBytes)); err != nil {
		return nil, err
	}
	mmap, err := gommap.Map(f.Fd(), gommap.PROT_READ|gommap.PROT_WRITE, gommap.MAP_SHARED)
	if err != nil {
		return nil, err
	}
	return &index{
		file: f,
		mmap: mmap,
		size: size,
	}, nil
}

// Read takes in a relative offset and returns the record's absolute offset
// and its position in the store file.
func (i *index) Read(in int64) (out uint32, pos uint64, err error) {
	if i.size == 0 {
		return 0, 0, io.EOF
	}
	if in == -1 {
		out = uint32((i.size / entWidth) - 1)
	} else {
		out = uint32(in)
	}
	pos = uint64(out) * entWidth
	if i.size < pos+entWidth {
		return 0, 0, io.EOF
	}
	out = enc.Uint32(i.mmap[pos : pos+offWidth])
	pos = enc.Uint64(i.mmap[pos+offWidth : pos+entWidth])
	return out, pos, nil
}

func (i *index) Append(off uint32, pos uint64) error {
	if uint64(len(i.mmap)) < i.size+entWidth {
		return io.EOF
	}
	enc.PutUint32(i.mmap[i.size:i.size+offWidth], off)
	enc.PutUint64(i.mmap[i.size+offWidth:i.size+entWidth], pos)
	i.size += entWidth
	return nil
}

func (i *index) Close() error {
	// Sync memory-mapped file to stable storage.
	if err := i.mmap.Sync(gommap.MS_SYNC); err != nil {
		return err
	}
	// Flush file system buffers.
	if err := i.file.Sync(); err != nil {
		return err
	}
	// Truncate the file to its actual size.
	if err := i.file.Truncate(int64(i.size)); err != nil {
		return err
	}
	return i.file.Close()
}

func (i *index) Name() string {
	return i.file.Name()
}

Note: We're using github.com/tysontate/gommap for memory mapping. You'll need to add this dependency: go get github.com/tysontate/gommap.

The Segment

The segment ties the store and index together.

Create internal/log/segment.go:

// internal/log/segment.go
package log

import (
	"fmt"
	"os"
	"path"

	api "github.com/your-username/your-project/api/v1" // Assuming protobufs are defined here
)

type segment struct {
	store      *store
	index      *index
	baseOffset, nextOffset uint64
	config     Config
}

// newSegment creates a new segment, creating the store and index files
// if they don't exist.
func newSegment(dir string, baseOffset uint64, c Config) (*segment, error) {
	s := &segment{
		baseOffset: baseOffset,
		config:     c,
	}
	storeFile, err := os.OpenFile(
		path.Join(dir, fmt.Sprintf("%d.store", baseOffset)),
		os.O_RDWR|os.O_CREATE|os.O_APPEND,
		0644,
	)
	if err != nil {
		return nil, err
	}
	if s.store, err = newStore(storeFile); err != nil {
		return nil, err
	}
	indexFile, err := os.OpenFile(
		path.Join(dir, fmt.Sprintf("%d.index", baseOffset)),
		os.O_RDWR|os.O_CREATE,
		0644,
	)
	if err != nil {
		return nil, err
	}
	if s.index, err = newIndex(indexFile, c); err != nil {
		return nil, err
	}
	if off, _, err := s.index.Read(-1); err != nil {
		s.nextOffset = baseOffset
	} else {
		s.nextOffset = baseOffset + uint64(off) + 1
	}
	return s, nil
}

// Append writes the record to the segment and returns the newly appended record's
// absolute offset.
func (s *segment) Append(record *api.Record) (offset uint64, err error) {
	cur := s.nextOffset
	record.Offset = cur
	p, err := proto.Marshal(record)
	if err != nil {
		return 0, err
	}
	_, pos, err := s.store.Append(p)
	if err != nil {
		return 0, err
	}
	if err = s.index.Append(uint32(s.nextOffset-uint64(s.baseOffset)), pos); err != nil {
		return 0, err
	}
	s.nextOffset++
	return cur, nil
}

// Read returns the record for the given offset.
func (s *segment) Read(off uint64) (*api.Record, error) {
	_, pos, err := s.index.Read(int64(off - s.baseOffset))
	if err != nil {
		return nil, err
	}
	p, err := s.store.Read(pos)
	if err != nil {
		return nil, err
	}
	record := &api.Record{}
	err = proto.Unmarshal(p, record)
	return record, err
}

// IsMaxed returns whether the segment has reached its max size.
func (s *segment) IsMaxed() bool {
	return s.store.size >= s.config.Segment.MaxStoreBytes ||
		s.index.size >= s.config.Segment.MaxIndexBytes
}

func (s *segment) Close() error {
	if err := s.index.Close(); err != nil {
		return err
	}
	if err := s.store.Close(); err != nil {
		return err
	}
	return nil
}

func (s *segment) Remove() error {
    if err := s.Close(); err != nil {
        return err
    }
    if err := os.Remove(s.index.Name()); err != nil {
        return err
    }
    if err := os.Remove(s.store.Name()); err != nil {
        return err
    }
    return nil
}

Note: For the record data, this implementation uses Protocol Buffers (api.Record), similar to many high-performance systems. You would define Record in a .proto file and generate the Go code. For simplicity, you could adapt this to use the []byte directly from the CommitLog interface and handle serialization with encoding/binary as we did in the store, avoiding a protobuf dependency for now.

3. The log Struct: Managing Segments

Now we have the segment building block. The final step is to create a log struct that manages a list of segments and implements our CommitLog interface.

Create internal/log/log.go:

// internal/log/log.go
package log

import (
	"fmt"
	"io/ioutil"
	"path"
	"sort"
	"strconv"
	"strings"
	"sync"
)

// Add a Config struct for configurability
type Config struct {
    Segment struct {
        MaxStoreBytes uint64
        MaxIndexBytes uint64
        InitialOffset uint64
    }
}

type Log struct {
	mu sync.RWMutex
	Dir    string
	Config Config

	activeSegment *segment
	segments      []*segment
}

func NewLog(dir string, c Config) (*Log, error) {
	if c.Segment.MaxStoreBytes == 0 {
		c.Segment.MaxStoreBytes = 1024
	}
	if c.Segment.MaxIndexBytes == 0 {
		c.Segment.MaxIndexBytes = 1024
	}
	l := &Log{
		Dir:    dir,
		Config: c,
	}
	return l, l.setup()
}

func (l *Log) setup() error {
	files, err := ioutil.ReadDir(l.Dir)
	if err != nil {
		return err
	}
	var baseOffsets []uint64
	for _, file := range files {
		offStr := strings.TrimSuffix(file.Name(), path.Ext(file.Name()))
		off, _ := strconv.ParseUint(offStr, 10, 0)
		baseOffsets = append(baseOffsets, off)
	}
	sort.Slice(baseOffsets, func(i, j int) bool {
		return baseOffsets[i] < baseOffsets[j]
	})
	for i := 0; i < len(baseOffsets); i++ {
		if err = l.newSegment(baseOffsets[i]); err != nil {
			return err
		}
		// baseOffset contains dup for index and store so we skip the dup
		i++
	}
	if l.segments == nil {
		if err = l.newSegment(l.Config.Segment.InitialOffset); err != nil {
			return err
		}
	}
	return nil
}

// Append appends a record to the active segment. If the segment is full,
// it creates a new one.
func (l *Log) Append(record *api.Record) (uint64, error) {
	l.mu.Lock()
	defer l.mu.Unlock()
	off, err := l.activeSegment.Append(record)
	if err != nil {
		return 0, err
	}
	if l.activeSegment.IsMaxed() {
		err = l.newSegment(off + 1)
	}
	return off, err
}

// Read reads the record at the given offset by finding the segment
// that contains the offset.
func (l *Log) Read(off uint64) (*api.Record, error) {
	l.mu.RLock()
	defer l.mu.RUnlock()
	var s *segment
	for _, segment := range l.segments {
		if segment.baseOffset <= off && off < segment.nextOffset {
			s = segment
			break
		}
	}
	if s == nil || s.nextOffset <= off {
		return nil, fmt.Errorf("offset out of range: %d", off)
	}
	return s.Read(off)
}

func (l *Log) newSegment(off uint64) error {
	s, err := newSegment(l.Dir, off, l.Config)
	if err != nil {
		return err
	}
	l.segments = append(l.segments, s)
	l.activeSegment = s
	return nil
}

func (l *Log) Close() error {
    l.mu.Lock()
    defer l.mu.Unlock()
    for _, segment := range l.segments {
        if err := segment.Close(); err != nil {
            return err
        }
    }
    return nil
}

This Log struct is our final implementation. It manages the lifecycle of segments, handles concurrent access, and correctly routes Append and Read calls to the appropriate segment. It discovers existing log files on startup, making it persistent across restarts.

4. Durability and Data Integrity

A critical aspect of a commit log is ensuring data is not lost in a crash. Simply writing to a file is not enough; modern operating systems heavily cache writes in memory.

In our index.go, the Close method calls i.mmap.Sync() and i.file.Sync(), which are wrappers that ultimately issue fsync-like calls to ensure data is persisted. For the highest level of durability, you could call store.Flush() and store.File.Sync() after every append, but this comes at a significant performance cost. A common strategy is to make this configurable: sync on every write, sync periodically, or let the OS handle it.

Another concern is data corruption. The same video introduces using a CRC (Cyclic Redundancy Check) checksum to verify data integrity on reads. While we haven't implemented it here, a production-grade system would embed a CRC checksum with every record written to the store file and verify it upon reading.

Conclusion

Congratulations! You have just built a sophisticated, persistent, file-based commit log from scratch. This is a significant piece of systems engineering and forms the core of our message broker's storage engine.

Key Takeaways:

• A production-ready log is built from a series of segments to be manageable.
• Each segment uses a .store file for data and a .index file for fast lookups.
• Memory-mapping the index file provides a major performance boost.
• The top-level Log struct orchestrates the segments, handles log rotation, and provides the CommitLog interface implementation.
• True durability requires forcing the OS to sync its buffers to disk, typically via fsync.

Next Up

In the next lesson, we will implement the same CommitLog interface again, but this time using PostgreSQL as the backend. This will powerfully demonstrate the flexibility of our interface-driven design and give you experience building a completely different type of storage engine.