wal

wal

A write-ahead log implementation for an event collector application.

Install

go get codeberg.org/kchan/wal

Usage

To create a WAL (wal.Log)

Use wal.NewLog to create a new write-ahead log:

import (
    "log/slog"
    "os"
    
    "codeberg.org/kchan/wal"
)

...

logger := slog.New(slog.NewJSONHandler(os.Stdout, nil))

config := wal.Config{
    Dir: "/path/to/wal-data",      // directory in which to store WAL files
    Segment: wal.SegmentConfig{
        MaxStoreBytes: 16 * 1024,  // maximum individual store file size 
        MaxIndexBytes: 8 * 1024,   // maximum individual index file size
    },
}
encoder := wal.NewEncoder()  // use default encoder
w, err := wal.NewLog(config, encoder, logger)
if err != nil {
    return err
}
defer w.Close()
...

To write data to the WAL

// some data to write to the log -- this should be encoded as []byte
data := []byte("some data")

// wal.Log.Write will return a LSN (log sequence number) and an error
lsn, err := w.Write(data)
if err != nil {
    return err
}

To read entries from the WAL

Create an Iterator and call Next to process each entry from the log segments:

import (
    "context"
    "time"
    
    "codeberg.org/kchan/wal"
)

...
ctx, cancel := context.WithCancel(context.Background())
defer cancel()

// each entry has a "LSN" (log sequence number)
// set starting index by calculating a "start" value
start := w.LowestIndex()       // the lowest index in the WAL
checkpoint := w.Checkpoint.C() // the last stored checkpoint
start = max(start, checkpoint) // start the sync process using the latest index

// create iterator
iter, err := w.Iterator(start)
if err != nil {
    return err
}

// iterate through each entry starting with "start"
// until the iterator returns either wal.ContextCancelled,
// wal.StopIteration, or an error
for {
    // lsn is the log sequence number
    // data is the []byte data stored in the log
    lsn, data, err := iter.Next(ctx) // supply context as parameter
    if errors.Is(err, wal.ContextCancelled) {
        return err
    }
    if errors.Is(err, wal.StopIteration) {
        return err
    }
    if err != nil {
        return err
    }

    // perform sync
    // ...

    // then update checkpoint
    err = w.Checkpoint.Write(lsn)
    if err != nil {
        return err
    }
}
...

To purge WAL segments

Run a goroutine that calls the Purge method:

import (
    "context"
    "time"
    
    "codeberg.org/kchan/wal"
)

...
ctx, cancel := context.WithCancel(context.Background())
defer cancel()

purgeInterval := 15 // purge the WAL segments every 15 seconds
ticker := time.NewTicker(time.Duration(purgeInterval) * time.Second)

for {
    select {
    case <-ctx.Done():
        return nil
    case <-ticker.C:
        // all segments prior to the latest checkpoint will be purged (deleted)
        err := w.Purge(ctx, func(segNum uint32, segment *wal.Segment) error {
            return nil
        })
        if err != nil {
            return err
        }
    }
}
...

License

MIT License. See LICENSE.