wal

README

A Write-Ahead-Log for Go

go-wal is a Write-Ahead Log (WAL) implementation in Go. It has high read and write throughput, and is suitable to be used in high-performance applications.

Features

• Write entries to the log.
• Read all entries from given log segment offset.
• Read all entries from the last log segment.
• Log Rotation for efficient startup and recovery.
• Auto-Remove old log segments on reaching segment limit.
• Sync entries to disk at regular intervals.
• CRC32 checksum for data integrity.
• Auto-Repair corrupted WALs.
• Supports checkpoints

Usage

Concurrency

1. Read-write exclusivity: The WAL operates in a mutually exclusive manner. You can either write to it or read from it at a given time. Simultaneous read and write operations are not permitted.
2. Thread safety: The WAL is thread-safe, enabling concurrent writes from multiple threads without data corruption.

Log Segments

1. Immutability: Log segments are immutable once created. Existing entries cannot be modified after creation.
2. Numbering: Log segment numbers start at 0 and increment sequentially.
3. Sequence numbers: Log entries are assigned sequence numbers starting at 1 and continue sequentially across log segments.

Repair Mechanism

1. Selective repair: The WAL targets the last segment for repair. If it's corrupted, a new segment containing all repaired entries replaces the original.
2. Corruption propagation: Segments following the first corrupted segment are assumed to be corrupted and discarded due to potential data integrity issues beyond the first corruption.
3. Manual intervention: Manually delete corrupted segments following the first corrupted segment before running the repair process.

Creating a WAL

You can create a WAL using the NewWAL function. This function takes a file path.

wal, err := OpenWAL("/wal/directory", enableFsync, maxSegmentSize, maxSegments)

Writing to the WAL

You can write an entry to the WAL using the Write method. This method takes a byte slice as data. This method is thread-safe.

err := wal.WriteEntry([]byte("data"))

Checkpointing the WAL

You can checkpoint the WAL using the Checkpoint method. This method flushes the in-memory buffers and runs a sync to disk (if enabled).

Also allows the user to store application specific data in the checkpoint.

err := wal.CreateCheckpoint([]byte("checkpoint info"))

Reading from the WAL

You can read all entries from the last WAL segment using the ReadEntries method.

// Read all entries from last segment.
entries, err = wal.ReadAll(false)
if err != nil {
    log.Fatalf("Failed to read entries: %v", err)
}

// Read all entries from last segment after the checkpoint
entries, err = wal.ReadAll(true)
if err != nil {
    log.Fatalf("Failed to read entries: %v", err)
}

You can also read from a given offset (inclusive) using the ReadAllFromOffset method. This method returns all the entries from the WAL starting from given log segment offset.

// Read all entries from a given offset.
entries, err = wal.ReadAllFromOffset(offset, false)

// Read all entries from a given offset after the checkpoint.
entries, err = wal.ReadAllFromOffset(offset, true)

Restoring from the last available checkpoint.

// Starts scanning from the first available segment, and returns all entries after the last checkpoint.
entries, err = wal.ReadAllFromOffset(-1, true)

Repairing the WAL

You can repair a corrupted WAL using the Repair method. This method returns the repaired entries, and atomically replaces the corrupted WAL file with the repaired one.

The WAL is capable of recovering from corrupted entries, as well as partial damage to the WAL file. However, if the file is completely corrupted, the WAL may not be able to recover from it and would proceed with replacing the file with an empty one.

entries, err := wal.Repair()

Closing the WAL

You can close the WAL using the Close method. Closing the WAL flushes the in-memory buffers and runs a final sync to disk (if enabled).

err := wal.Close()

Testing

This project includes a set of tests. You can run these tests using the go test ./... command.

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

License

This project is licensed under the MIT License. See the LICENSE file for details.

Documentation

Index

• Variables
• func MustMarshal(entry *WAL_Entry) []byte
• func MustUnmarshal(data []byte, entry *WAL_Entry)
• type WAL
  • func OpenWAL(directory string, enableFsync bool, maxFileSize int64, maxSegments int) (*WAL, error)
  • func (wal *WAL) Close() error
  • func (wal *WAL) CreateCheckpoint(data []byte) error
  • func (wal *WAL) ReadAll(readFromCheckpoint bool) ([]*WAL_Entry, error)
  • func (wal *WAL) ReadAllFromOffset(offset int, readFromCheckpoint bool) ([]*WAL_Entry, error)
  • func (wal *WAL) Repair() ([]*WAL_Entry, error)
  • func (wal *WAL) Sync() error
  • func (wal *WAL) WriteEntry(data []byte) error
• type WAL_Entry
  • func (*WAL_Entry) Descriptor() ([]byte, []int)
  • func (x *WAL_Entry) GetCRC() uint32
  • func (x *WAL_Entry) GetData() []byte
  • func (x *WAL_Entry) GetIsCheckpoint() bool
  • func (x *WAL_Entry) GetLogSequenceNumber() uint64
  • func (*WAL_Entry) ProtoMessage()
  • func (x *WAL_Entry) ProtoReflect() protoreflect.Message
  • func (x *WAL_Entry) Reset()
  • func (x *WAL_Entry) String() string

Variables

var File_types_proto protoreflect.FileDescriptor

Functions

func MustMarshal

func MustMarshal(entry *WAL_Entry) []byte

Marshals

func MustUnmarshal

func MustUnmarshal(data []byte, entry *WAL_Entry)

Types

type WAL

type WAL struct {
	// contains filtered or unexported fields
}

WAL structure

func OpenWAL

func OpenWAL(directory string, enableFsync bool, maxFileSize int64, maxSegments int) (*WAL, error)

Initialize a new WAL. If the directory does not exist, it will be created. If the directory exists, the last log segment will be opened and the last sequence number will be read from it. enableFsync enables fsync on the log segment file every time the log flushes. maxFileSize is the maximum size of a log segment file in bytes. maxSegments is the maximum number of log segment files to keep.

func (*WAL) Close

func (wal *WAL) Close() error

Close the WAL file. It also calls Sync() on the WAL.

func (*WAL) CreateCheckpoint

func (wal *WAL) CreateCheckpoint(data []byte) error

CreateCheckpoint creates a checkpoint entry in the WAL. A checkpoint entry is a special entry that can be used to restore the state of the system to the point when the checkpoint was created.

func (*WAL) ReadAll

func (wal *WAL) ReadAll(readFromCheckpoint bool) ([]*WAL_Entry, error)

Read all entries from the WAL. If readFromCheckpoint is true, it will return all the entries from the last checkpoint (if no checkpoint is found, it will return an empty slice.)

func (*WAL) ReadAllFromOffset

func (wal *WAL) ReadAllFromOffset(offset int, readFromCheckpoint bool) ([]*WAL_Entry, error)

Starts reading from log segment files starting from the given offset (Segment Index) and returns all the entries. If readFromCheckpoint is true, it will return all the entries from the last checkpoint (if no checkpoint is found, it will return an empty slice.)

func (*WAL) Repair

func (wal *WAL) Repair() ([]*WAL_Entry, error)

Repairs a corrupted WAL by scanning the WAL from the start and reading all entries until a corrupted entry is encountered, at which point the file is truncated. The function returns the entries that were read before the corruption and overwrites the existing WAL file with the repaired entries. It checks the CRC of each entry to verify if it is corrupted, and if the CRC is invalid, the file is truncated at that point.

func (*WAL) Sync

func (wal *WAL) Sync() error

Writes out any data in the WAL's in-memory buffer to the segment file. If fsync is enabled, it also calls fsync on the segment file. It also resets the synchronization timer.

func (*WAL) WriteEntry

func (wal *WAL) WriteEntry(data []byte) error

WriteEntry writes an entry to the WAL.

type WAL_Entry

type WAL_Entry struct {
	LogSequenceNumber uint64 `protobuf:"varint,1,opt,name=logSequenceNumber,proto3" json:"logSequenceNumber,omitempty"`
	Data              []byte `protobuf:"bytes,2,opt,name=data,proto3" json:"data,omitempty"`
	CRC               uint32 `protobuf:"varint,3,opt,name=CRC,proto3" json:"CRC,omitempty"`
	// Optional field for checkpointing.
	IsCheckpoint *bool `protobuf:"varint,4,opt,name=isCheckpoint,proto3,oneof" json:"isCheckpoint,omitempty"`
	// contains filtered or unexported fields
}

func (*WAL_Entry) Descriptor

func (*WAL_Entry) Descriptor() ([]byte, []int)

Deprecated: Use WAL_Entry.ProtoReflect.Descriptor instead.

func (*WAL_Entry) GetCRC

func (x *WAL_Entry) GetCRC() uint32

func (*WAL_Entry) GetData

func (x *WAL_Entry) GetData() []byte

func (*WAL_Entry) GetIsCheckpoint

func (x *WAL_Entry) GetIsCheckpoint() bool

func (*WAL_Entry) GetLogSequenceNumber

func (x *WAL_Entry) GetLogSequenceNumber() uint64

func (*WAL_Entry) ProtoMessage

func (*WAL_Entry) ProtoMessage()

func (*WAL_Entry) ProtoReflect

func (x *WAL_Entry) ProtoReflect() protoreflect.Message

func (*WAL_Entry) Reset

func (x *WAL_Entry) Reset()

func (*WAL_Entry) String

func (x *WAL_Entry) String() string