Segmented write-ahead log
Append-only WAL with CRC32-validated records, automatic segment rotation at 128 MiB, background periodic fsync, and crash recovery by truncation at the first corrupt or incomplete record. Record format: type(1) | len(4) | payload | crc32c(4). Two payload types: series (ref + labels) and samples (batch of ref/t/v). Fixed-width encoding througout. Torn-write harness truncates at every byte offset and asserts recovery produces a valid prefix of the original sequnce, both single-segment and multi-segment. DESIGN.md, NOTICE.md, and README.md.
This commit is contained in:
@@ -0,0 +1,142 @@
|
||||
package wal
|
||||
|
||||
import (
|
||||
"encoding/binary"
|
||||
"errors"
|
||||
"math"
|
||||
|
||||
"git.dvdt.dev/david/ingot/labels"
|
||||
)
|
||||
|
||||
var ErrShortPayload = errors.New("wal: payload too short")
|
||||
|
||||
// SeriesRecord is a WAL record that registers a new series.
|
||||
type SeriesRecord struct {
|
||||
Ref uint64
|
||||
Labels []labels.Label
|
||||
}
|
||||
|
||||
// RefSample is a single sample keyed by series ref.
|
||||
type RefSample struct {
|
||||
Ref uint64
|
||||
T int64
|
||||
V float64
|
||||
}
|
||||
|
||||
// EncodeSeriesRecord appends the encoded series record to dst.
|
||||
//
|
||||
// ref(8) | nlabels(4) | for each: namelen(2) name valuelen(2) value
|
||||
func EncodeSeriesRecord(dst []byte, rec SeriesRecord) []byte {
|
||||
n := 8 + 4
|
||||
for _, l := range rec.Labels {
|
||||
n += 2 + len(l.Name) + 2 + len(l.Value)
|
||||
}
|
||||
dst = grow(dst, n)
|
||||
off := len(dst) - n
|
||||
|
||||
binary.BigEndian.PutUint64(dst[off:], rec.Ref)
|
||||
off += 8
|
||||
binary.BigEndian.PutUint32(dst[off:], uint32(len(rec.Labels)))
|
||||
off += 4
|
||||
|
||||
for _, l := range rec.Labels {
|
||||
binary.BigEndian.PutUint16(dst[off:], uint16(len(l.Name)))
|
||||
off += 2
|
||||
off += copy(dst[off:], l.Name)
|
||||
binary.BigEndian.PutUint16(dst[off:], uint16(len(l.Value)))
|
||||
off += 2
|
||||
off += copy(dst[off:], l.Value)
|
||||
}
|
||||
|
||||
return dst
|
||||
}
|
||||
|
||||
// DecodeSeriesRecord decodes a series payload. The returned Labels
|
||||
// hold copies of the strings (not sub-slices of data).
|
||||
func DecodeSeriesRecord(data []byte) (SeriesRecord, error) {
|
||||
if len(data) < 12 {
|
||||
return SeriesRecord{}, ErrShortPayload
|
||||
}
|
||||
|
||||
ref := binary.BigEndian.Uint64(data)
|
||||
nLabels := int(binary.BigEndian.Uint32(data[8:]))
|
||||
off := 12
|
||||
|
||||
ls := make([]labels.Label, nLabels)
|
||||
for i := range ls {
|
||||
if off+2 > len(data) {
|
||||
return SeriesRecord{}, ErrShortPayload
|
||||
}
|
||||
nameLen := int(binary.BigEndian.Uint16(data[off:]))
|
||||
off += 2
|
||||
if off+nameLen > len(data) {
|
||||
return SeriesRecord{}, ErrShortPayload
|
||||
}
|
||||
name := string(data[off : off+nameLen])
|
||||
off += nameLen
|
||||
|
||||
if off+2 > len(data) {
|
||||
return SeriesRecord{}, ErrShortPayload
|
||||
}
|
||||
valueLen := int(binary.BigEndian.Uint16(data[off:]))
|
||||
off += 2
|
||||
if off+valueLen > len(data) {
|
||||
return SeriesRecord{}, ErrShortPayload
|
||||
}
|
||||
value := string(data[off : off+valueLen])
|
||||
off += valueLen
|
||||
|
||||
ls[i] = labels.Label{Name: name, Value: value}
|
||||
}
|
||||
|
||||
return SeriesRecord{Ref: ref, Labels: ls}, nil
|
||||
}
|
||||
|
||||
// EncodeSamplesRecord appends the encoded samples record to dst.
|
||||
//
|
||||
// nsamples(4) | for each: ref(8) t(8) v(8)
|
||||
func EncodeSamplesRecord(dst []byte, samples []RefSample) []byte {
|
||||
n := 4 + len(samples)*24
|
||||
dst = grow(dst, n)
|
||||
off := len(dst) - n
|
||||
|
||||
binary.BigEndian.PutUint32(dst[off:], uint32(len(samples)))
|
||||
off += 4
|
||||
|
||||
for _, s := range samples {
|
||||
binary.BigEndian.PutUint64(dst[off:], s.Ref)
|
||||
off += 8
|
||||
binary.BigEndian.PutUint64(dst[off:], uint64(s.T))
|
||||
off += 8
|
||||
binary.BigEndian.PutUint64(dst[off:], math.Float64bits(s.V))
|
||||
off += 8
|
||||
}
|
||||
|
||||
return dst
|
||||
}
|
||||
|
||||
// DecodeSamplesRecord decodes a samples payload.
|
||||
func DecodeSamplesRecord(data []byte) ([]RefSample, error) {
|
||||
if len(data) < 4 {
|
||||
return nil, ErrShortPayload
|
||||
}
|
||||
|
||||
n := int(binary.BigEndian.Uint32(data))
|
||||
off := 4
|
||||
|
||||
if len(data) < 4+n*24 {
|
||||
return nil, ErrShortPayload
|
||||
}
|
||||
|
||||
samples := make([]RefSample, n)
|
||||
for i := range samples {
|
||||
samples[i].Ref = binary.BigEndian.Uint64(data[off:])
|
||||
off += 8
|
||||
samples[i].T = int64(binary.BigEndian.Uint64(data[off:]))
|
||||
off += 8
|
||||
samples[i].V = math.Float64frombits(binary.BigEndian.Uint64(data[off:]))
|
||||
off += 8
|
||||
}
|
||||
|
||||
return samples, nil
|
||||
}
|
||||
@@ -0,0 +1,270 @@
|
||||
package wal
|
||||
|
||||
import (
|
||||
"encoding/binary"
|
||||
"math"
|
||||
"testing"
|
||||
|
||||
"git.dvdt.dev/david/ingot/labels"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestSeriesRecord(t *testing.T) {
|
||||
type result struct {
|
||||
rec SeriesRecord
|
||||
err error
|
||||
}
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
data []byte
|
||||
want result
|
||||
}{
|
||||
{
|
||||
name: "single_label",
|
||||
data: EncodeSeriesRecord(nil, SeriesRecord{
|
||||
Ref: 42,
|
||||
Labels: []labels.Label{{Name: "__name__", Value: "temp"}},
|
||||
}),
|
||||
want: result{SeriesRecord{
|
||||
Ref: 42,
|
||||
Labels: []labels.Label{{Name: "__name__", Value: "temp"}},
|
||||
}, nil},
|
||||
},
|
||||
{
|
||||
name: "multiple_labels",
|
||||
data: EncodeSeriesRecord(nil, SeriesRecord{
|
||||
Ref: 1,
|
||||
Labels: []labels.Label{
|
||||
{Name: "__name__", Value: "cpu_usage"},
|
||||
{Name: "host", Value: "web-01"},
|
||||
{Name: "region", Value: "us-east"},
|
||||
},
|
||||
}),
|
||||
want: result{SeriesRecord{
|
||||
Ref: 1,
|
||||
Labels: []labels.Label{
|
||||
{Name: "__name__", Value: "cpu_usage"},
|
||||
{Name: "host", Value: "web-01"},
|
||||
{Name: "region", Value: "us-east"},
|
||||
},
|
||||
}, nil},
|
||||
},
|
||||
{
|
||||
name: "zero_labels",
|
||||
data: EncodeSeriesRecord(nil, SeriesRecord{Ref: 99, Labels: nil}),
|
||||
want: result{SeriesRecord{Ref: 99, Labels: []labels.Label{}}, nil},
|
||||
},
|
||||
{
|
||||
name: "unicode_labels",
|
||||
data: EncodeSeriesRecord(nil, SeriesRecord{
|
||||
Ref: 7,
|
||||
Labels: []labels.Label{{Name: "名前", Value: "温度"}},
|
||||
}),
|
||||
want: result{SeriesRecord{
|
||||
Ref: 7,
|
||||
Labels: []labels.Label{{Name: "名前", Value: "温度"}},
|
||||
}, nil},
|
||||
},
|
||||
{
|
||||
name: "empty_label_strings",
|
||||
data: EncodeSeriesRecord(nil, SeriesRecord{
|
||||
Ref: 1,
|
||||
Labels: []labels.Label{{Name: "", Value: ""}},
|
||||
}),
|
||||
want: result{SeriesRecord{
|
||||
Ref: 1,
|
||||
Labels: []labels.Label{{Name: "", Value: ""}},
|
||||
}, nil},
|
||||
},
|
||||
|
||||
// Error cases.
|
||||
{
|
||||
name: "nil",
|
||||
data: nil,
|
||||
want: result{SeriesRecord{}, ErrShortPayload},
|
||||
},
|
||||
{
|
||||
name: "truncated_ref",
|
||||
data: make([]byte, 6),
|
||||
want: result{SeriesRecord{}, ErrShortPayload},
|
||||
},
|
||||
{
|
||||
name: "truncated_nlabels",
|
||||
data: make([]byte, 10),
|
||||
want: result{SeriesRecord{}, ErrShortPayload},
|
||||
},
|
||||
{
|
||||
name: "truncated_name_len",
|
||||
data: func() []byte {
|
||||
d := make([]byte, 13) // ref(8) + nlabels=1(4) + 1 byte (short)
|
||||
binary.BigEndian.PutUint32(d[8:], 1)
|
||||
return d
|
||||
}(),
|
||||
want: result{SeriesRecord{}, ErrShortPayload},
|
||||
},
|
||||
{
|
||||
name: "truncated_name_data",
|
||||
data: func() []byte {
|
||||
d := make([]byte, 16) // ref(8) + nlabels=1(4) + namelen=10(2) + 2 bytes
|
||||
binary.BigEndian.PutUint32(d[8:], 1)
|
||||
binary.BigEndian.PutUint16(d[12:], 10) // claims 10 bytes, only 2 available
|
||||
return d
|
||||
}(),
|
||||
want: result{SeriesRecord{}, ErrShortPayload},
|
||||
},
|
||||
{
|
||||
name: "truncated_value_len",
|
||||
data: func() []byte {
|
||||
d := make([]byte, 15) // ref(8) + nlabels=1(4) + namelen=0(2) + 1 byte
|
||||
binary.BigEndian.PutUint32(d[8:], 1)
|
||||
binary.BigEndian.PutUint16(d[12:], 0) // 0-length name
|
||||
return d
|
||||
}(),
|
||||
want: result{SeriesRecord{}, ErrShortPayload},
|
||||
},
|
||||
{
|
||||
name: "truncated_value_data",
|
||||
data: func() []byte {
|
||||
d := make([]byte, 18) // ref(8) + nlabels=1(4) + namelen=0(2) + vallen=5(2) + 2 bytes
|
||||
binary.BigEndian.PutUint32(d[8:], 1)
|
||||
binary.BigEndian.PutUint16(d[12:], 0)
|
||||
binary.BigEndian.PutUint16(d[14:], 5) // claims 5 bytes, only 2 available
|
||||
return d
|
||||
}(),
|
||||
want: result{SeriesRecord{}, ErrShortPayload},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
rec, err := DecodeSeriesRecord(tc.data)
|
||||
assert.Equal(t, tc.want.rec, rec, "record")
|
||||
assert.Equal(t, tc.want.err, err, "error")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// refSampleBits holds a RefSample with the value stored as raw bits
|
||||
// so NaN and negative zero compare correctly.
|
||||
type refSampleBits struct {
|
||||
Ref uint64
|
||||
T int64
|
||||
VBits uint64
|
||||
}
|
||||
|
||||
func toBits(s RefSample) refSampleBits {
|
||||
return refSampleBits{s.Ref, s.T, math.Float64bits(s.V)}
|
||||
}
|
||||
|
||||
func samplesToBits(ss []RefSample) []refSampleBits {
|
||||
out := make([]refSampleBits, len(ss))
|
||||
for i, s := range ss {
|
||||
out[i] = toBits(s)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func TestSamplesRecord(t *testing.T) {
|
||||
type result struct {
|
||||
samples []refSampleBits
|
||||
err error
|
||||
}
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
data []byte
|
||||
want result
|
||||
}{
|
||||
{
|
||||
name: "single_sample",
|
||||
data: EncodeSamplesRecord(nil, []RefSample{
|
||||
{Ref: 1, T: 1000, V: 71.3},
|
||||
}),
|
||||
want: result{samplesToBits([]RefSample{{Ref: 1, T: 1000, V: 71.3}}), nil},
|
||||
},
|
||||
{
|
||||
name: "multiple_samples",
|
||||
data: EncodeSamplesRecord(nil, []RefSample{
|
||||
{Ref: 1, T: 1000, V: 71.3},
|
||||
{Ref: 1, T: 1015, V: 71.4},
|
||||
{Ref: 2, T: 1000, V: 0},
|
||||
}),
|
||||
want: result{samplesToBits([]RefSample{
|
||||
{Ref: 1, T: 1000, V: 71.3},
|
||||
{Ref: 1, T: 1015, V: 71.4},
|
||||
{Ref: 2, T: 1000, V: 0},
|
||||
}), nil},
|
||||
},
|
||||
{
|
||||
name: "zero_samples",
|
||||
data: EncodeSamplesRecord(nil, nil),
|
||||
want: result{samplesToBits([]RefSample{}), nil},
|
||||
},
|
||||
{
|
||||
name: "special_float_values",
|
||||
data: EncodeSamplesRecord(nil, []RefSample{
|
||||
{Ref: 1, T: 0, V: math.NaN()},
|
||||
{Ref: 2, T: 0, V: math.Inf(1)},
|
||||
{Ref: 3, T: 0, V: math.Inf(-1)},
|
||||
{Ref: 4, T: 0, V: math.Copysign(0, -1)},
|
||||
}),
|
||||
want: result{samplesToBits([]RefSample{
|
||||
{Ref: 1, T: 0, V: math.NaN()},
|
||||
{Ref: 2, T: 0, V: math.Inf(1)},
|
||||
{Ref: 3, T: 0, V: math.Inf(-1)},
|
||||
{Ref: 4, T: 0, V: math.Copysign(0, -1)},
|
||||
}), nil},
|
||||
},
|
||||
{
|
||||
name: "negative_timestamp",
|
||||
data: EncodeSamplesRecord(nil, []RefSample{
|
||||
{Ref: 1, T: -5000, V: 1.5},
|
||||
}),
|
||||
want: result{samplesToBits([]RefSample{{Ref: 1, T: -5000, V: 1.5}}), nil},
|
||||
},
|
||||
|
||||
// Error cases.
|
||||
{
|
||||
name: "nil",
|
||||
data: nil,
|
||||
want: result{nil, ErrShortPayload},
|
||||
},
|
||||
{
|
||||
name: "truncated_count",
|
||||
data: []byte{0, 0},
|
||||
want: result{nil, ErrShortPayload},
|
||||
},
|
||||
{
|
||||
name: "truncated_mid_sample",
|
||||
data: func() []byte {
|
||||
d := make([]byte, 20) // nsamples=1(4) + 16 bytes (need 24)
|
||||
binary.BigEndian.PutUint32(d, 1)
|
||||
return d
|
||||
}(),
|
||||
want: result{nil, ErrShortPayload},
|
||||
},
|
||||
{
|
||||
name: "count_exceeds_data",
|
||||
data: func() []byte {
|
||||
d := make([]byte, 28) // nsamples=2(4) + 24 bytes (only 1 sample)
|
||||
binary.BigEndian.PutUint32(d, 2)
|
||||
return d
|
||||
}(),
|
||||
want: result{nil, ErrShortPayload},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
samples, err := DecodeSamplesRecord(tc.data)
|
||||
got := samplesToBits(samples)
|
||||
assert.Equal(t, tc.want.err, err, "error")
|
||||
require.Equal(t, len(tc.want.samples), len(got), "sample count")
|
||||
for i := range tc.want.samples {
|
||||
assert.Equal(t, tc.want.samples[i], got[i], "sample %d", i)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,237 @@
|
||||
package wal
|
||||
|
||||
import (
|
||||
"io"
|
||||
"os"
|
||||
)
|
||||
|
||||
// Record is a decoded WAL record returned by the Reader.
|
||||
type Record struct {
|
||||
Type RecordType
|
||||
Data []byte // raw payload; decode with DecodeSeriesRecord / DecodeSamplesRecord
|
||||
}
|
||||
|
||||
// Reader scans WAL segments sequentially, validating CRC on each record.
|
||||
// It follows the iterator pattern: Next() advances, Record() returns the
|
||||
// current record, Err() returns any error after Next() returns false.
|
||||
type Reader struct {
|
||||
dir string
|
||||
segments []int // sorted segment indices
|
||||
segIdx int // position in segments slice
|
||||
f *os.File
|
||||
buf []byte // read buffer, grown as needed
|
||||
rec Record
|
||||
err error
|
||||
}
|
||||
|
||||
// NewReader creates a Reader over all segments in dir.
|
||||
func NewReader(dir string) (*Reader, error) {
|
||||
segs, err := listSegments(dir)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &Reader{dir: dir, segments: segs}, nil
|
||||
}
|
||||
|
||||
// Next advances to the next record. Returns false when no more records
|
||||
// are available or an error is encountered. After Next returns false,
|
||||
// call Err() to distinguish clean EOF from corruption.
|
||||
func (r *Reader) Next() bool {
|
||||
for {
|
||||
if r.err != nil {
|
||||
return false
|
||||
}
|
||||
|
||||
// Open the next segment file if needed.
|
||||
if r.f == nil {
|
||||
if !r.openNextSegment() {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// Read the record header (type + len).
|
||||
header, ok := r.readExact(recordHeaderSize)
|
||||
if !ok {
|
||||
// EOF or short read at record boundary.
|
||||
if r.err == nil {
|
||||
// Clean EOF on this segment — try next.
|
||||
r.closeFile()
|
||||
continue
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
typ := RecordType(header[0])
|
||||
payloadLen := int(header[1])<<24 | int(header[2])<<16 | int(header[3])<<8 | int(header[4])
|
||||
|
||||
// Read payload + CRC trailer.
|
||||
body, ok := r.readExact(payloadLen + recordTrailerSize)
|
||||
if !ok {
|
||||
// Torn write: header was read but payload/CRC is truncated.
|
||||
if r.err == nil {
|
||||
r.err = ErrInvalidRecord
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// Validate CRC over header + payload.
|
||||
full := append(header, body[:payloadLen]...)
|
||||
_, _, _, decErr := DecodeRecord(r.reassemble(header, body, payloadLen))
|
||||
if decErr != nil {
|
||||
r.err = decErr
|
||||
return false
|
||||
}
|
||||
_ = full // replaced by reassemble
|
||||
|
||||
r.rec = Record{Type: typ, Data: cloneBytes(body[:payloadLen])}
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
// reassemble reconstructs the full framed record from the separately read
|
||||
// header and body (payload + CRC) for CRC validation via DecodeRecord.
|
||||
func (r *Reader) reassemble(header []byte, body []byte, payloadLen int) []byte {
|
||||
total := recordHeaderSize + payloadLen + recordTrailerSize
|
||||
if cap(r.buf) < total {
|
||||
r.buf = make([]byte, total)
|
||||
}
|
||||
r.buf = r.buf[:total]
|
||||
copy(r.buf, header)
|
||||
copy(r.buf[recordHeaderSize:], body)
|
||||
return r.buf
|
||||
}
|
||||
|
||||
// Record returns the most recently read record.
|
||||
func (r *Reader) Record() Record {
|
||||
return r.rec
|
||||
}
|
||||
|
||||
// Err returns the error encountered during reading, if any.
|
||||
// A nil error after Next() returns false means all records were read cleanly.
|
||||
func (r *Reader) Err() error {
|
||||
return r.err
|
||||
}
|
||||
|
||||
// Close releases any open file handle.
|
||||
func (r *Reader) Close() error {
|
||||
return r.closeFile()
|
||||
}
|
||||
|
||||
func (r *Reader) openNextSegment() bool {
|
||||
if r.segIdx >= len(r.segments) {
|
||||
return false
|
||||
}
|
||||
f, err := os.Open(segmentPath(r.dir, r.segments[r.segIdx]))
|
||||
if err != nil {
|
||||
r.err = err
|
||||
return false
|
||||
}
|
||||
r.f = f
|
||||
r.segIdx++
|
||||
return true
|
||||
}
|
||||
|
||||
func (r *Reader) closeFile() error {
|
||||
if r.f == nil {
|
||||
return nil
|
||||
}
|
||||
err := r.f.Close()
|
||||
r.f = nil
|
||||
return err
|
||||
}
|
||||
|
||||
// readExact reads exactly n bytes from the current file. On short read
|
||||
// at EOF, it sets r.err to ErrInvalidRecord (torn write) and returns false.
|
||||
// On clean EOF (zero bytes read), it returns false with r.err == nil.
|
||||
func (r *Reader) readExact(n int) ([]byte, bool) {
|
||||
if n == 0 {
|
||||
return nil, true
|
||||
}
|
||||
buf := make([]byte, n)
|
||||
_, err := io.ReadFull(r.f, buf)
|
||||
if err == io.EOF {
|
||||
// Clean EOF — no bytes at all.
|
||||
return nil, false
|
||||
}
|
||||
if err == io.ErrUnexpectedEOF {
|
||||
// Partial read — torn write.
|
||||
r.err = ErrInvalidRecord
|
||||
return nil, false
|
||||
}
|
||||
if err != nil {
|
||||
r.err = err
|
||||
return nil, false
|
||||
}
|
||||
return buf, true
|
||||
}
|
||||
|
||||
func cloneBytes(b []byte) []byte {
|
||||
c := make([]byte, len(b))
|
||||
copy(c, b)
|
||||
return c
|
||||
}
|
||||
|
||||
// recover scans all segments, validating records. On the first corrupt or
|
||||
// truncated record, it truncates the segment file at the start of that record
|
||||
// and deletes all subsequent segments. Returns the records that survived.
|
||||
func recover(dir string) error {
|
||||
segs, err := listSegments(dir)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if len(segs) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
for i, idx := range segs {
|
||||
truncated, err := recoverSegment(dir, idx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if truncated {
|
||||
// Delete all segments after this one.
|
||||
for _, laterIdx := range segs[i+1:] {
|
||||
if err := os.Remove(segmentPath(dir, laterIdx)); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// recoverSegment validates all records in a single segment. If it encounters
|
||||
// corruption, it truncates the file at the last valid record boundary.
|
||||
// Returns true if truncation occurred.
|
||||
func recoverSegment(dir string, index int) (bool, error) {
|
||||
path := segmentPath(dir, index)
|
||||
data, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
|
||||
// Walk records, tracking the offset of the last valid boundary.
|
||||
validEnd := 0
|
||||
off := 0
|
||||
for off < len(data) {
|
||||
_, _, consumed, err := DecodeRecord(data[off:])
|
||||
if err != nil {
|
||||
// Corruption or truncation at this offset.
|
||||
break
|
||||
}
|
||||
off += consumed
|
||||
validEnd = off
|
||||
}
|
||||
|
||||
if validEnd == len(data) {
|
||||
// Entire segment is valid.
|
||||
return false, nil
|
||||
}
|
||||
|
||||
// Truncate the file at the last valid boundary.
|
||||
if err := os.Truncate(path, int64(validEnd)); err != nil {
|
||||
return false, err
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
@@ -0,0 +1,88 @@
|
||||
// Package wal implements a segmented write-ahead log for crash-safe
|
||||
// persistence of time-series data.
|
||||
//
|
||||
// WAL design informed by Prometheus tsdb/wal. See /NOTICE.md.
|
||||
package wal
|
||||
|
||||
import (
|
||||
"encoding/binary"
|
||||
"errors"
|
||||
"hash/crc32"
|
||||
)
|
||||
|
||||
// RecordType identifies the kind of record stored in the WAL.
|
||||
type RecordType byte
|
||||
|
||||
const (
|
||||
RecordSeries RecordType = 1
|
||||
RecordSamples RecordType = 2
|
||||
)
|
||||
|
||||
const (
|
||||
recordHeaderSize = 5 // type(1) + len(4)
|
||||
recordTrailerSize = 4 // crc32(4)
|
||||
)
|
||||
|
||||
var (
|
||||
ErrInvalidRecord = errors.New("wal: invalid record")
|
||||
ErrCorruptRecord = errors.New("wal: corrupt record (CRC mismatch)")
|
||||
)
|
||||
|
||||
var castagnoliTable = crc32.MakeTable(crc32.Castagnoli)
|
||||
|
||||
// RecordSize returns the total on-disk size of a record with the given payload length.
|
||||
func RecordSize(payloadLen int) int {
|
||||
return recordHeaderSize + payloadLen + recordTrailerSize
|
||||
}
|
||||
|
||||
// EncodeRecord appends a framed record (type + len + payload + crc32c) to dst
|
||||
// and returns the extended slice.
|
||||
func EncodeRecord(dst []byte, typ RecordType, payload []byte) []byte {
|
||||
n := RecordSize(len(payload))
|
||||
dst = grow(dst, n)
|
||||
off := len(dst) - n
|
||||
|
||||
dst[off] = byte(typ)
|
||||
binary.BigEndian.PutUint32(dst[off+1:], uint32(len(payload)))
|
||||
copy(dst[off+recordHeaderSize:], payload)
|
||||
|
||||
checksum := crc32.Checksum(dst[off:off+recordHeaderSize+len(payload)], castagnoliTable)
|
||||
binary.BigEndian.PutUint32(dst[off+recordHeaderSize+len(payload):], checksum)
|
||||
|
||||
return dst
|
||||
}
|
||||
|
||||
// DecodeRecord parses a framed record from b. It returns the record type,
|
||||
// the payload slice (a sub-slice of b), the total number of bytes consumed,
|
||||
// and any error. On success, consumed == RecordSize(len(payload)).
|
||||
func DecodeRecord(b []byte) (typ RecordType, payload []byte, consumed int, err error) {
|
||||
if len(b) < recordHeaderSize {
|
||||
return 0, nil, 0, ErrInvalidRecord
|
||||
}
|
||||
|
||||
typ = RecordType(b[0])
|
||||
payloadLen := int(binary.BigEndian.Uint32(b[1:]))
|
||||
total := RecordSize(payloadLen)
|
||||
|
||||
if len(b) < total {
|
||||
return 0, nil, 0, ErrInvalidRecord
|
||||
}
|
||||
|
||||
want := crc32.Checksum(b[:recordHeaderSize+payloadLen], castagnoliTable)
|
||||
got := binary.BigEndian.Uint32(b[recordHeaderSize+payloadLen:])
|
||||
if want != got {
|
||||
return 0, nil, 0, ErrCorruptRecord
|
||||
}
|
||||
|
||||
return typ, b[recordHeaderSize : recordHeaderSize+payloadLen], total, nil
|
||||
}
|
||||
|
||||
// grow appends n zero bytes to dst and returns the extended slice.
|
||||
func grow(dst []byte, n int) []byte {
|
||||
if cap(dst)-len(dst) >= n {
|
||||
return dst[:len(dst)+n]
|
||||
}
|
||||
buf := make([]byte, len(dst)+n, 2*(len(dst)+n))
|
||||
copy(buf, dst)
|
||||
return buf
|
||||
}
|
||||
@@ -0,0 +1,183 @@
|
||||
package wal
|
||||
|
||||
import (
|
||||
"encoding/binary"
|
||||
"hash/crc32"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestRecord(t *testing.T) {
|
||||
type decodeResult struct {
|
||||
typ RecordType
|
||||
payload []byte
|
||||
consumed int
|
||||
err error
|
||||
}
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
data []byte // raw bytes to decode
|
||||
want decodeResult
|
||||
encode bool // if true, data was produced by EncodeRecord (round-trip test)
|
||||
}{
|
||||
// --- Round-trip cases ---
|
||||
{
|
||||
name: "series_payload",
|
||||
data: EncodeRecord(nil, RecordSeries, []byte{0xDE, 0xAD}),
|
||||
want: decodeResult{RecordSeries, []byte{0xDE, 0xAD}, RecordSize(2), nil},
|
||||
encode: true,
|
||||
},
|
||||
{
|
||||
name: "samples_payload",
|
||||
data: EncodeRecord(nil, RecordSamples, []byte{1, 2, 3, 4, 5}),
|
||||
want: decodeResult{RecordSamples, []byte{1, 2, 3, 4, 5}, RecordSize(5), nil},
|
||||
encode: true,
|
||||
},
|
||||
{
|
||||
name: "empty_payload",
|
||||
data: EncodeRecord(nil, RecordSeries, nil),
|
||||
want: decodeResult{RecordSeries, []byte{}, RecordSize(0), nil},
|
||||
encode: true,
|
||||
},
|
||||
{
|
||||
name: "large_payload",
|
||||
data: EncodeRecord(nil, RecordSamples, make([]byte, 8192)),
|
||||
want: decodeResult{RecordSamples, make([]byte, 8192), RecordSize(8192), nil},
|
||||
encode: true,
|
||||
},
|
||||
{
|
||||
name: "unknown_record_type",
|
||||
data: EncodeRecord(nil, RecordType(255), []byte{0xFF}),
|
||||
want: decodeResult{RecordType(255), []byte{0xFF}, RecordSize(1), nil},
|
||||
encode: true,
|
||||
},
|
||||
|
||||
// --- Error cases ---
|
||||
{
|
||||
name: "nil_input",
|
||||
data: nil,
|
||||
want: decodeResult{0, nil, 0, ErrInvalidRecord},
|
||||
},
|
||||
{
|
||||
name: "empty_input",
|
||||
data: []byte{},
|
||||
want: decodeResult{0, nil, 0, ErrInvalidRecord},
|
||||
},
|
||||
{
|
||||
name: "truncated_at_type",
|
||||
data: []byte{byte(RecordSeries)},
|
||||
want: decodeResult{0, nil, 0, ErrInvalidRecord},
|
||||
},
|
||||
{
|
||||
name: "truncated_at_len",
|
||||
data: []byte{byte(RecordSeries), 0, 0},
|
||||
want: decodeResult{0, nil, 0, ErrInvalidRecord},
|
||||
},
|
||||
{
|
||||
name: "truncated_at_payload",
|
||||
data: func() []byte {
|
||||
// Header says 10 bytes of payload, but only 4 present.
|
||||
d := make([]byte, recordHeaderSize+4)
|
||||
d[0] = byte(RecordSeries)
|
||||
binary.BigEndian.PutUint32(d[1:], 10)
|
||||
return d
|
||||
}(),
|
||||
want: decodeResult{0, nil, 0, ErrInvalidRecord},
|
||||
},
|
||||
{
|
||||
name: "truncated_at_crc",
|
||||
data: func() []byte {
|
||||
// Full header + full payload, but missing CRC.
|
||||
d := make([]byte, recordHeaderSize+2) // 2-byte payload, no CRC
|
||||
d[0] = byte(RecordSeries)
|
||||
binary.BigEndian.PutUint32(d[1:], 2)
|
||||
return d
|
||||
}(),
|
||||
want: decodeResult{0, nil, 0, ErrInvalidRecord},
|
||||
},
|
||||
{
|
||||
name: "corrupted_crc",
|
||||
data: func() []byte {
|
||||
d := EncodeRecord(nil, RecordSeries, []byte{0xAB, 0xCD})
|
||||
d[len(d)-1] ^= 0xFF // flip last CRC byte
|
||||
return d
|
||||
}(),
|
||||
want: decodeResult{0, nil, 0, ErrCorruptRecord},
|
||||
},
|
||||
{
|
||||
name: "corrupted_payload",
|
||||
data: func() []byte {
|
||||
d := EncodeRecord(nil, RecordSeries, []byte{0xAB, 0xCD})
|
||||
d[recordHeaderSize] ^= 0xFF // flip first payload byte
|
||||
return d
|
||||
}(),
|
||||
want: decodeResult{0, nil, 0, ErrCorruptRecord},
|
||||
},
|
||||
{
|
||||
name: "corrupted_type_byte",
|
||||
data: func() []byte {
|
||||
d := EncodeRecord(nil, RecordSeries, []byte{0xAB})
|
||||
d[0] ^= 0xFF // flip type byte
|
||||
return d
|
||||
}(),
|
||||
want: decodeResult{0, nil, 0, ErrCorruptRecord},
|
||||
},
|
||||
{
|
||||
name: "corrupted_len_field",
|
||||
data: func() []byte {
|
||||
d := EncodeRecord(nil, RecordSeries, []byte{0xAB})
|
||||
d[1] ^= 0x01 // flip len byte — now claims different length
|
||||
return d
|
||||
}(),
|
||||
want: decodeResult{0, nil, 0, ErrInvalidRecord},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
typ, payload, consumed, err := DecodeRecord(tc.data)
|
||||
assert.Equal(t, tc.want.typ, typ, "type")
|
||||
assert.Equal(t, tc.want.payload, payload, "payload")
|
||||
assert.Equal(t, tc.want.consumed, consumed, "consumed")
|
||||
assert.Equal(t, tc.want.err, err, "error")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestEncodeRecordAppendsToExisting(t *testing.T) {
|
||||
prefix := []byte("existing")
|
||||
result := EncodeRecord(prefix, RecordSeries, []byte{0x01})
|
||||
assert.Equal(t, []byte("existing"), result[:8])
|
||||
|
||||
_, payload, _, err := DecodeRecord(result[8:])
|
||||
assert.Equal(t, []byte{0x01}, payload)
|
||||
assert.Equal(t, nil, err)
|
||||
}
|
||||
|
||||
func TestRecordSize(t *testing.T) {
|
||||
tests := []struct {
|
||||
payloadLen int
|
||||
want int
|
||||
}{
|
||||
{0, 9},
|
||||
{1, 10},
|
||||
{100, 109},
|
||||
}
|
||||
|
||||
for _, tc := range tests {
|
||||
assert.Equal(t, tc.want, RecordSize(tc.payloadLen))
|
||||
}
|
||||
}
|
||||
|
||||
func TestEncodeCRCCoversHeaderAndPayload(t *testing.T) {
|
||||
payload := []byte{0x01, 0x02, 0x03}
|
||||
rec := EncodeRecord(nil, RecordSamples, payload)
|
||||
|
||||
// Manually compute expected CRC over type+len+payload.
|
||||
headerAndPayload := rec[:recordHeaderSize+len(payload)]
|
||||
want := crc32.Checksum(headerAndPayload, castagnoliTable)
|
||||
got := binary.BigEndian.Uint32(rec[recordHeaderSize+len(payload):])
|
||||
assert.Equal(t, want, got)
|
||||
}
|
||||
@@ -0,0 +1,63 @@
|
||||
package wal
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
const (
|
||||
defaultSegmentMaxSize = 128 * 1024 * 1024 // 128 MiB
|
||||
segmentNameLen = 8 // "00000001"
|
||||
)
|
||||
|
||||
// segmentFileName returns the zero-padded filename for a segment index.
|
||||
func segmentFileName(index int) string {
|
||||
return fmt.Sprintf("%0*d", segmentNameLen, index)
|
||||
}
|
||||
|
||||
// parseSegmentIndex parses a segment filename back to its index.
|
||||
// Returns -1 if the name is not a valid segment file.
|
||||
func parseSegmentIndex(name string) int {
|
||||
if len(name) != segmentNameLen {
|
||||
return -1
|
||||
}
|
||||
n, err := strconv.Atoi(name)
|
||||
if err != nil {
|
||||
return -1
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// listSegments returns the sorted indices of all segment files in dir.
|
||||
func listSegments(dir string) ([]int, error) {
|
||||
entries, err := os.ReadDir(dir)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var indices []int
|
||||
for _, e := range entries {
|
||||
if e.IsDir() {
|
||||
continue
|
||||
}
|
||||
idx := parseSegmentIndex(e.Name())
|
||||
if idx >= 0 {
|
||||
indices = append(indices, idx)
|
||||
}
|
||||
}
|
||||
sort.Ints(indices)
|
||||
return indices, nil
|
||||
}
|
||||
|
||||
// segmentPath returns the full path for a segment index within dir.
|
||||
func segmentPath(dir string, index int) string {
|
||||
return filepath.Join(dir, segmentFileName(index))
|
||||
}
|
||||
|
||||
// createSegment creates a new segment file and returns it open for writing.
|
||||
func createSegment(dir string, index int) (*os.File, error) {
|
||||
return os.OpenFile(segmentPath(dir, index), os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0644)
|
||||
}
|
||||
@@ -0,0 +1,251 @@
|
||||
package wal
|
||||
|
||||
import (
|
||||
"os"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Options configures WAL behavior.
|
||||
type Options struct {
|
||||
// SegmentMaxSize is the maximum size of a single segment file in bytes.
|
||||
// A new segment is created when the current one would exceed this.
|
||||
// Default: 128 MiB.
|
||||
SegmentMaxSize int
|
||||
|
||||
// SyncInterval controls background fsync frequency.
|
||||
// Default (zero): 1s. Negative: sync on every Log call.
|
||||
SyncInterval time.Duration
|
||||
}
|
||||
|
||||
func (o *Options) segmentMaxSize() int {
|
||||
if o.SegmentMaxSize > 0 {
|
||||
return o.SegmentMaxSize
|
||||
}
|
||||
return defaultSegmentMaxSize
|
||||
}
|
||||
|
||||
func (o *Options) syncInterval() time.Duration {
|
||||
if o.SyncInterval < 0 {
|
||||
return -1 // sync-per-write sentinel
|
||||
}
|
||||
if o.SyncInterval == 0 {
|
||||
return time.Second
|
||||
}
|
||||
return o.SyncInterval
|
||||
}
|
||||
|
||||
// WAL is a segmented write-ahead log.
|
||||
type WAL struct {
|
||||
dir string
|
||||
opts Options
|
||||
|
||||
mu sync.Mutex
|
||||
segment *os.File
|
||||
segmentIdx int
|
||||
segmentOff int64
|
||||
buf []byte
|
||||
|
||||
done chan struct{}
|
||||
wg sync.WaitGroup
|
||||
}
|
||||
|
||||
// Open opens or creates a WAL in dir. If segments already exist, it runs
|
||||
// recovery (truncating at the first corrupt record) before returning.
|
||||
func Open(dir string, opts Options) (*WAL, error) {
|
||||
if err := os.MkdirAll(dir, 0755); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Recover existing segments.
|
||||
if err := recover(dir); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
w := &WAL{
|
||||
dir: dir,
|
||||
opts: opts,
|
||||
done: make(chan struct{}),
|
||||
}
|
||||
|
||||
// Open or create the active segment.
|
||||
segs, err := listSegments(dir)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if len(segs) == 0 {
|
||||
// Fresh WAL.
|
||||
w.segmentIdx = 1
|
||||
f, err := createSegment(dir, 1)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
w.segment = f
|
||||
} else {
|
||||
// Append to the last segment.
|
||||
idx := segs[len(segs)-1]
|
||||
w.segmentIdx = idx
|
||||
f, err := os.OpenFile(segmentPath(dir, idx), os.O_WRONLY|os.O_APPEND, 0644)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
info, err := f.Stat()
|
||||
if err != nil {
|
||||
f.Close()
|
||||
return nil, err
|
||||
}
|
||||
w.segment = f
|
||||
w.segmentOff = info.Size()
|
||||
}
|
||||
|
||||
// Start background syncer.
|
||||
if interval := opts.syncInterval(); interval > 0 {
|
||||
w.wg.Add(1)
|
||||
go w.syncLoop(interval)
|
||||
}
|
||||
|
||||
return w, nil
|
||||
}
|
||||
|
||||
// Log writes a framed record to the WAL. The payload is wrapped with
|
||||
// the record envelope (type + length + CRC).
|
||||
func (w *WAL) Log(typ RecordType, payload []byte) error {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
|
||||
w.buf = EncodeRecord(w.buf[:0], typ, payload)
|
||||
|
||||
// Rotate if this write would exceed the segment size limit.
|
||||
if w.segmentOff+int64(len(w.buf)) > int64(w.opts.segmentMaxSize()) {
|
||||
if err := w.rotate(); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
n, err := w.segment.Write(w.buf)
|
||||
w.segmentOff += int64(n)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Sync-per-write mode.
|
||||
if w.opts.syncInterval() < 0 {
|
||||
return w.segment.Sync()
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// LogSeries encodes and writes a series record.
|
||||
func (w *WAL) LogSeries(recs []SeriesRecord) error {
|
||||
for _, rec := range recs {
|
||||
payload := EncodeSeriesRecord(nil, rec)
|
||||
if err := w.Log(RecordSeries, payload); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// LogSamples encodes and writes a samples record.
|
||||
func (w *WAL) LogSamples(samples []RefSample) error {
|
||||
payload := EncodeSamplesRecord(nil, samples)
|
||||
return w.Log(RecordSamples, payload)
|
||||
}
|
||||
|
||||
// Replay returns a Reader over all WAL segments. The caller must Close
|
||||
// the reader when done.
|
||||
func (w *WAL) Replay() (*Reader, error) {
|
||||
return NewReader(w.dir)
|
||||
}
|
||||
|
||||
// Sync forces an fsync of the current segment.
|
||||
func (w *WAL) Sync() error {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
return w.segment.Sync()
|
||||
}
|
||||
|
||||
// Truncate deletes all segments with index less than below.
|
||||
func (w *WAL) Truncate(below int) error {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
|
||||
segs, err := listSegments(w.dir)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, idx := range segs {
|
||||
if idx >= below {
|
||||
break
|
||||
}
|
||||
if err := os.Remove(segmentPath(w.dir, idx)); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// LastSegment returns the index of the current active segment.
|
||||
func (w *WAL) LastSegment() int {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
return w.segmentIdx
|
||||
}
|
||||
|
||||
// Close stops the background syncer, fsyncs, and closes the active segment.
|
||||
func (w *WAL) Close() error {
|
||||
close(w.done)
|
||||
w.wg.Wait()
|
||||
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
|
||||
if w.segment == nil {
|
||||
return nil
|
||||
}
|
||||
if err := w.segment.Sync(); err != nil {
|
||||
w.segment.Close()
|
||||
return err
|
||||
}
|
||||
return w.segment.Close()
|
||||
}
|
||||
|
||||
// rotate fsyncs the current segment, closes it, and creates a new one.
|
||||
// Caller must hold w.mu.
|
||||
func (w *WAL) rotate() error {
|
||||
if err := w.segment.Sync(); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := w.segment.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
w.segmentIdx++
|
||||
f, err := createSegment(w.dir, w.segmentIdx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
w.segment = f
|
||||
w.segmentOff = 0
|
||||
return nil
|
||||
}
|
||||
|
||||
func (w *WAL) syncLoop(interval time.Duration) {
|
||||
defer w.wg.Done()
|
||||
ticker := time.NewTicker(interval)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-w.done:
|
||||
return
|
||||
case <-ticker.C:
|
||||
w.mu.Lock()
|
||||
w.segment.Sync()
|
||||
w.mu.Unlock()
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,266 @@
|
||||
package wal
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"git.dvdt.dev/david/ingot/labels"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// collectRecords replays all records from a WAL directory.
|
||||
func collectRecords(t *testing.T, dir string) []Record {
|
||||
t.Helper()
|
||||
r, err := NewReader(dir)
|
||||
require.NoError(t, err)
|
||||
defer r.Close()
|
||||
|
||||
var recs []Record
|
||||
for r.Next() {
|
||||
recs = append(recs, r.Record())
|
||||
}
|
||||
require.NoError(t, r.Err())
|
||||
return recs
|
||||
}
|
||||
|
||||
func TestWAL(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
setup func(t *testing.T, dir string) // write to WAL, close it, optionally corrupt
|
||||
wantRecords int
|
||||
wantMinSegs int // assert segment count >= this
|
||||
}{
|
||||
{
|
||||
name: "write_and_replay",
|
||||
setup: func(t *testing.T, dir string) {
|
||||
w, err := Open(dir, Options{})
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, w.LogSeries([]SeriesRecord{
|
||||
{Ref: 1, Labels: []labels.Label{{Name: "__name__", Value: "temp"}, {Name: "room", Value: "office"}}},
|
||||
}))
|
||||
require.NoError(t, w.LogSamples([]RefSample{
|
||||
{Ref: 1, T: 1000, V: 71.3},
|
||||
{Ref: 1, T: 1015, V: 71.4},
|
||||
}))
|
||||
require.NoError(t, w.Close())
|
||||
},
|
||||
wantRecords: 2,
|
||||
wantMinSegs: 1,
|
||||
},
|
||||
{
|
||||
name: "empty",
|
||||
setup: func(t *testing.T, dir string) {
|
||||
w, err := Open(dir, Options{})
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, w.Close())
|
||||
},
|
||||
wantRecords: 0,
|
||||
wantMinSegs: 1,
|
||||
},
|
||||
{
|
||||
name: "reopen_and_append",
|
||||
setup: func(t *testing.T, dir string) {
|
||||
w, err := Open(dir, Options{})
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, w.LogSamples([]RefSample{{Ref: 1, T: 1000, V: 1.0}}))
|
||||
require.NoError(t, w.Close())
|
||||
|
||||
w, err = Open(dir, Options{})
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, w.LogSamples([]RefSample{{Ref: 2, T: 2000, V: 2.0}}))
|
||||
require.NoError(t, w.Close())
|
||||
},
|
||||
wantRecords: 2,
|
||||
wantMinSegs: 1,
|
||||
},
|
||||
{
|
||||
name: "segment_rotation",
|
||||
setup: func(t *testing.T, dir string) {
|
||||
w, err := Open(dir, Options{SegmentMaxSize: 50})
|
||||
require.NoError(t, err)
|
||||
for i := 0; i < 10; i++ {
|
||||
require.NoError(t, w.LogSamples([]RefSample{{Ref: uint64(i), T: int64(i * 1000), V: float64(i)}}))
|
||||
}
|
||||
require.NoError(t, w.Close())
|
||||
},
|
||||
wantRecords: 10,
|
||||
wantMinSegs: 2,
|
||||
},
|
||||
{
|
||||
name: "truncate_old_segments",
|
||||
setup: func(t *testing.T, dir string) {
|
||||
w, err := Open(dir, Options{SegmentMaxSize: 50})
|
||||
require.NoError(t, err)
|
||||
for i := 0; i < 10; i++ {
|
||||
require.NoError(t, w.LogSamples([]RefSample{{Ref: uint64(i), T: int64(i), V: float64(i)}}))
|
||||
}
|
||||
lastSeg := w.LastSegment()
|
||||
require.NoError(t, w.Truncate(lastSeg))
|
||||
require.NoError(t, w.Close())
|
||||
},
|
||||
wantRecords: 1, // only the last segment's record(s) survive
|
||||
wantMinSegs: 1,
|
||||
},
|
||||
{
|
||||
name: "recovery_truncates_trailing_garbage",
|
||||
setup: func(t *testing.T, dir string) {
|
||||
w, err := Open(dir, Options{})
|
||||
require.NoError(t, err)
|
||||
for i := 0; i < 3; i++ {
|
||||
require.NoError(t, w.LogSamples([]RefSample{{Ref: uint64(i), T: int64(i), V: float64(i)}}))
|
||||
}
|
||||
require.NoError(t, w.Close())
|
||||
|
||||
// Append garbage after valid records.
|
||||
segs, err := listSegments(dir)
|
||||
require.NoError(t, err)
|
||||
f, err := os.OpenFile(segmentPath(dir, segs[0]), os.O_WRONLY|os.O_APPEND, 0644)
|
||||
require.NoError(t, err)
|
||||
_, err = f.Write([]byte{0xFF, 0xFF, 0xFF, 0xFF, 0xFF})
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, f.Close())
|
||||
|
||||
// Reopen triggers recovery.
|
||||
w, err = Open(dir, Options{})
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, w.Close())
|
||||
},
|
||||
wantRecords: 3,
|
||||
wantMinSegs: 1,
|
||||
},
|
||||
{
|
||||
name: "recovery_truncates_corrupt_mid_record",
|
||||
setup: func(t *testing.T, dir string) {
|
||||
w, err := Open(dir, Options{})
|
||||
require.NoError(t, err)
|
||||
for i := 0; i < 3; i++ {
|
||||
require.NoError(t, w.LogSamples([]RefSample{{Ref: uint64(i), T: int64(i), V: float64(i)}}))
|
||||
}
|
||||
require.NoError(t, w.Close())
|
||||
|
||||
// Write a valid header but truncated payload (looks like a torn write).
|
||||
segs, err := listSegments(dir)
|
||||
require.NoError(t, err)
|
||||
f, err := os.OpenFile(segmentPath(dir, segs[0]), os.O_WRONLY|os.O_APPEND, 0644)
|
||||
require.NoError(t, err)
|
||||
// type=1, len=100 (big), but no payload follows.
|
||||
_, err = f.Write([]byte{0x01, 0x00, 0x00, 0x00, 0x64})
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, f.Close())
|
||||
|
||||
w, err = Open(dir, Options{})
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, w.Close())
|
||||
},
|
||||
wantRecords: 3,
|
||||
wantMinSegs: 1,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
dir := filepath.Join(t.TempDir(), "wal")
|
||||
tc.setup(t, dir)
|
||||
|
||||
recs := collectRecords(t, dir)
|
||||
assert.Equal(t, tc.wantRecords, len(recs), "record count")
|
||||
|
||||
segs, err := listSegments(dir)
|
||||
require.NoError(t, err)
|
||||
assert.GreaterOrEqual(t, len(segs), tc.wantMinSegs, "segment count")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestTornWriteRecovery is the headline test from DESIGN.md: for every possible
|
||||
// byte offset, truncate the WAL there and verify recovery produces a valid
|
||||
// prefix of the original record sequence.
|
||||
func TestTornWriteRecovery(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
opts Options
|
||||
recs func(t *testing.T, w *WAL) // write records to the WAL
|
||||
}{
|
||||
{
|
||||
name: "single_segment_mixed_records",
|
||||
opts: Options{},
|
||||
recs: func(t *testing.T, w *WAL) {
|
||||
require.NoError(t, w.LogSeries([]SeriesRecord{
|
||||
{Ref: 1, Labels: []labels.Label{{Name: "__name__", Value: "temp"}}},
|
||||
}))
|
||||
require.NoError(t, w.LogSamples([]RefSample{{Ref: 1, T: 1000, V: 71.3}}))
|
||||
require.NoError(t, w.LogSeries([]SeriesRecord{
|
||||
{Ref: 2, Labels: []labels.Label{{Name: "__name__", Value: "humidity"}, {Name: "room", Value: "lab"}}},
|
||||
}))
|
||||
require.NoError(t, w.LogSamples([]RefSample{
|
||||
{Ref: 1, T: 1015, V: 71.4},
|
||||
{Ref: 2, T: 1000, V: 55.0},
|
||||
}))
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "multi_segment",
|
||||
opts: Options{SegmentMaxSize: 50},
|
||||
recs: func(t *testing.T, w *WAL) {
|
||||
for i := 0; i < 10; i++ {
|
||||
require.NoError(t, w.LogSamples([]RefSample{{Ref: uint64(i), T: int64(i * 1000), V: float64(i)}}))
|
||||
}
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
// Write the reference WAL.
|
||||
srcDir := filepath.Join(t.TempDir(), "src")
|
||||
w, err := Open(srcDir, tc.opts)
|
||||
require.NoError(t, err)
|
||||
tc.recs(t, w)
|
||||
require.NoError(t, w.Close())
|
||||
|
||||
origRecs := collectRecords(t, srcDir)
|
||||
require.Greater(t, len(origRecs), 0)
|
||||
|
||||
segs, err := listSegments(srcDir)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Read all segment data.
|
||||
segData := make(map[int][]byte)
|
||||
for _, idx := range segs {
|
||||
data, err := os.ReadFile(segmentPath(srcDir, idx))
|
||||
require.NoError(t, err)
|
||||
segData[idx] = data
|
||||
}
|
||||
|
||||
// Truncate the last segment at every byte offset.
|
||||
lastSeg := segs[len(segs)-1]
|
||||
lastData := segData[lastSeg]
|
||||
|
||||
for cutoff := 0; cutoff <= len(lastData); cutoff++ {
|
||||
walDir := filepath.Join(t.TempDir(), "wal")
|
||||
require.NoError(t, os.MkdirAll(walDir, 0755))
|
||||
|
||||
// Copy earlier segments intact.
|
||||
for _, idx := range segs[:len(segs)-1] {
|
||||
require.NoError(t, os.WriteFile(segmentPath(walDir, idx), segData[idx], 0644))
|
||||
}
|
||||
// Write truncated last segment.
|
||||
require.NoError(t, os.WriteFile(segmentPath(walDir, lastSeg), lastData[:cutoff], 0644))
|
||||
|
||||
w2, err := Open(walDir, tc.opts)
|
||||
require.NoError(t, err, "cutoff=%d", cutoff)
|
||||
recovered := collectRecords(t, walDir)
|
||||
require.NoError(t, w2.Close(), "cutoff=%d", cutoff)
|
||||
|
||||
// Must be a valid prefix.
|
||||
assert.LessOrEqual(t, len(recovered), len(origRecs), "cutoff=%d count", cutoff)
|
||||
for i, rec := range recovered {
|
||||
assert.Equal(t, origRecs[i].Type, rec.Type, "cutoff=%d rec=%d type", cutoff, i)
|
||||
assert.Equal(t, origRecs[i].Data, rec.Data, "cutoff=%d rec=%d data", cutoff, i)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user