wal.go

// Copyright 2017 The Prometheus Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package tsdb

import (
	"bufio"
	"encoding/binary"
	"fmt"
	"hash"
	"hash/crc32"
	"io"
	"math"
	"os"
	"path/filepath"
	"sort"
	"sync"
	"time"

	"github.com/go-kit/kit/log"
	"github.com/go-kit/kit/log/level"
	"github.com/pkg/errors"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/tsdb/fileutil"
	"github.com/prometheus/tsdb/labels"
)

// WALEntryType indicates what data a WAL entry contains.
type WALEntryType uint8

const (
	// WALMagic is a 4 byte number every WAL segment file starts with.
	WALMagic = uint32(0x43AF00EF)

	// WALFormatDefault is the version flag for the default outer segment file format.
	WALFormatDefault = byte(1)
)

// Entry types in a segment file.
const (
	WALEntrySymbols WALEntryType = 1
	WALEntrySeries  WALEntryType = 2
	WALEntrySamples WALEntryType = 3
	WALEntryDeletes WALEntryType = 4
)

type walMetrics struct {
	fsyncDuration prometheus.Summary
	corruptions   prometheus.Counter
}

func newWalMetrics(wal *SegmentWAL, r prometheus.Registerer) *walMetrics {
	m := &walMetrics{}

	m.fsyncDuration = prometheus.NewSummary(prometheus.SummaryOpts{
		Name: "prometheus_tsdb_wal_fsync_duration_seconds",
		Help: "Duration of WAL fsync.",
	})
	m.corruptions = prometheus.NewCounter(prometheus.CounterOpts{
		Name: "prometheus_tsdb_wal_corruptions_total",
		Help: "Total number of WAL corruptions.",
	})

	if r != nil {
		r.MustRegister(
			m.fsyncDuration,
			m.corruptions,
		)
	}
	return m
}

// WAL is a write ahead log that can log new series labels and samples.
// It must be completely read before new entries are logged.
type WAL interface {
	Reader() WALReader
	LogSeries([]RefSeries) error
	LogSamples([]RefSample) error
	LogDeletes([]Stone) error
	Truncate(mint int64, keep func(uint64) bool) error
	Close() error
}

// NopWAL is a WAL that does nothing.
func NopWAL() WAL {
	return nopWAL{}
}

type nopWAL struct{}

func (nopWAL) Read(
	seriesf func([]RefSeries),
	samplesf func([]RefSample),
	deletesf func([]Stone),
) error {
	return nil
}
func (w nopWAL) Reader() WALReader                     { return w }
func (nopWAL) LogSeries([]RefSeries) error             { return nil }
func (nopWAL) LogSamples([]RefSample) error            { return nil }
func (nopWAL) LogDeletes([]Stone) error                { return nil }
func (nopWAL) Truncate(int64, func(uint64) bool) error { return nil }
func (nopWAL) Close() error                            { return nil }

// WALReader reads entries from a WAL.
type WALReader interface {
	Read(
		seriesf func([]RefSeries),
		samplesf func([]RefSample),
		deletesf func([]Stone),
	) error
}

// RefSeries is the series labels with the series ID.
type RefSeries struct {
	Ref    uint64
	Labels labels.Labels
}

// RefSample is a timestamp/value pair associated with a reference to a series.
type RefSample struct {
	Ref uint64
	T   int64
	V   float64

	series *memSeries
}

// segmentFile wraps a file object of a segment and tracks the highest timestamp
// it contains. During WAL truncating, all segments with no higher timestamp than
// the truncation threshold can be compacted.
type segmentFile struct {
	*os.File
	maxTime   int64  // highest tombstone or sample timpstamp in segment
	minSeries uint64 // lowerst series ID in segment
}

func newSegmentFile(f *os.File) *segmentFile {
	return &segmentFile{
		File:      f,
		maxTime:   math.MinInt64,
		minSeries: math.MaxUint64,
	}
}

const (
	walSegmentSizeBytes = 256 * 1024 * 1024 // 256 MB
)

// The table gets initialized with sync.Once but may still cause a race
// with any other use of the crc32 package anywhere. Thus we initialize it
// before.
var castagnoliTable *crc32.Table

func init() {
	castagnoliTable = crc32.MakeTable(crc32.Castagnoli)
}

// newCRC32 initializes a CRC32 hash with a preconfigured polynomial, so the
// polynomial may be easily changed in one location at a later time, if necessary.
func newCRC32() hash.Hash32 {
	return crc32.New(castagnoliTable)
}

// SegmentWAL is a write ahead log for series data.
type SegmentWAL struct {
	mtx     sync.Mutex
	metrics *walMetrics

	dirFile *os.File
	files   []*segmentFile

	logger        log.Logger
	flushInterval time.Duration
	segmentSize   int64

	crc32 hash.Hash32
	cur   *bufio.Writer
	curN  int64

	stopc   chan struct{}
	donec   chan struct{}
	actorc  chan func() error // sequentialized background operations
	buffers sync.Pool
}

// OpenSegmentWAL opens or creates a write ahead log in the given directory.
// The WAL must be read completely before new data is written.
func OpenSegmentWAL(dir string, logger log.Logger, flushInterval time.Duration, r prometheus.Registerer) (*SegmentWAL, error) {
	if err := os.MkdirAll(dir, 0777); err != nil {
		return nil, err
	}
	df, err := fileutil.OpenDir(dir)
	if err != nil {
		return nil, err
	}
	if logger == nil {
		logger = log.NewNopLogger()
	}

	w := &SegmentWAL{
		dirFile:       df,
		logger:        logger,
		flushInterval: flushInterval,
		donec:         make(chan struct{}),
		stopc:         make(chan struct{}),
		actorc:        make(chan func() error, 1),
		segmentSize:   walSegmentSizeBytes,
		crc32:         newCRC32(),
	}
	w.metrics = newWalMetrics(w, r)

	fns, err := sequenceFiles(w.dirFile.Name())
	if err != nil {
		return nil, err
	}

	for i, fn := range fns {
		f, err := w.openSegmentFile(fn)
		if err == nil {
			w.files = append(w.files, newSegmentFile(f))
			continue
		}
		level.Warn(logger).Log("msg", "invalid segment file detected, truncating WAL", "err", err, "file", fn)

		for _, fn := range fns[i:] {
			if err := os.Remove(fn); err != nil {
				return w, errors.Wrap(err, "removing segment failed")
			}
		}
		break
	}

	go w.run(flushInterval)

	return w, nil
}

// repairingWALReader wraps a WAL reader and truncates its underlying SegmentWAL after the last
// valid entry if it encounters corruption.
type repairingWALReader struct {
	wal *SegmentWAL
	r   WALReader
}

func (r *repairingWALReader) Read(
	seriesf func([]RefSeries),
	samplesf func([]RefSample),
	deletesf func([]Stone),
) error {
	err := r.r.Read(seriesf, samplesf, deletesf)
	if err == nil {
		return nil
	}
	cerr, ok := errors.Cause(err).(walCorruptionErr)
	if !ok {
		return err
	}
	r.wal.metrics.corruptions.Inc()
	return r.wal.truncate(cerr.err, cerr.file, cerr.lastOffset)
}

// truncate the WAL after the last valid entry.
func (w *SegmentWAL) truncate(err error, file int, lastOffset int64) error {
	level.Error(w.logger).Log("msg", "WAL corruption detected; truncating",
		"err", err, "file", w.files[file].Name(), "pos", lastOffset)

	// Close and delete all files after the current one.
	for _, f := range w.files[file+1:] {
		if err := f.Close(); err != nil {
			return err
		}
		if err := os.Remove(f.Name()); err != nil {
			return err
		}
	}
	w.mtx.Lock()
	defer w.mtx.Unlock()

	w.files = w.files[:file+1]

	// Seek the current file to the last valid offset where we continue writing from.
	_, err = w.files[file].Seek(lastOffset, os.SEEK_SET)
	return err
}

// Reader returns a new reader over the the write ahead log data.
// It must be completely consumed before writing to the WAL.
func (w *SegmentWAL) Reader() WALReader {
	return &repairingWALReader{
		wal: w,
		r:   newWALReader(w.files, w.logger),
	}
}

func (w *SegmentWAL) getBuffer() *encbuf {
	b := w.buffers.Get()
	if b == nil {
		return &encbuf{b: make([]byte, 0, 64*1024)}
	}
	return b.(*encbuf)
}

func (w *SegmentWAL) putBuffer(b *encbuf) {
	b.reset()
	w.buffers.Put(b)
}

// Truncate deletes the values prior to mint and the series which the keep function
// does not indiciate to preserve.
func (w *SegmentWAL) Truncate(mint int64, keep func(uint64) bool) error {
	// The last segment is always active.
	if len(w.files) < 2 {
		return nil
	}
	var candidates []*segmentFile

	// All files have to be traversed as there could be two segments for a block
	// with first block having times (10000, 20000) and SECOND one having (0, 10000).
	for _, sf := range w.files[:len(w.files)-1] {
		if sf.maxTime >= mint {
			break
		}
		// Past WAL files are closed. We have to reopen them for another read.
		f, err := w.openSegmentFile(sf.Name())
		if err != nil {
			return errors.Wrap(err, "open old WAL segment for read")
		}
		candidates = append(candidates, &segmentFile{
			File:      f,
			minSeries: sf.minSeries,
			maxTime:   sf.maxTime,
		})
	}
	if len(candidates) == 0 {
		return nil
	}

	r := newWALReader(candidates, w.logger)

	// Create a new tmp file.
	f, err := w.createSegmentFile(filepath.Join(w.dirFile.Name(), "compact.tmp"))
	if err != nil {
		return errors.Wrap(err, "create compaction segment")
	}
	var (
		csf          = newSegmentFile(f)
		crc32        = newCRC32()
		decSeries    = []RefSeries{}
		activeSeries = []RefSeries{}
	)

	for r.next() {
		rt, flag, byt := r.at()

		if rt != WALEntrySeries {
			continue
		}
		decSeries = decSeries[:0]
		activeSeries = activeSeries[:0]

		err := r.decodeSeries(flag, byt, &decSeries)
		if err != nil {
			return errors.Wrap(err, "decode samples while truncating")
		}
		for _, s := range decSeries {
			if keep(s.Ref) {
				activeSeries = append(activeSeries, s)
			}
		}

		buf := w.getBuffer()
		flag = w.encodeSeries(buf, activeSeries)

		_, err = w.writeTo(csf, crc32, WALEntrySeries, flag, buf.get())
		w.putBuffer(buf)

		if err != nil {
			return errors.Wrap(err, "write to compaction segment")
		}
	}
	if r.Err() != nil {
		return errors.Wrap(r.Err(), "read candidate WAL files")
	}

	off, err := csf.Seek(0, os.SEEK_CUR)
	if err != nil {
		return err
	}
	if err := csf.Truncate(off); err != nil {
		return err
	}
	csf.Sync()
	csf.Close()

	candidates[0].Close() // need close before remove on platform windows
	if err := renameFile(csf.Name(), candidates[0].Name()); err != nil {
		return errors.Wrap(err, "rename compaction segment")
	}
	for _, f := range candidates[1:] {
		f.Close() // need close before remove on platform windows
		if err := os.RemoveAll(f.Name()); err != nil {
			return errors.Wrap(err, "delete WAL segment file")
		}
	}
	if err := w.dirFile.Sync(); err != nil {
		return err
	}

	// The file object of csf still holds the name before rename. Recreate it so
	// subsequent truncations do not look at a non-existant file name.
	csf.File, err = w.openSegmentFile(candidates[0].Name())
	if err != nil {
		return err
	}
	// We don't need it to be open.
	csf.Close()

	w.mtx.Lock()
	w.files = append([]*segmentFile{csf}, w.files[len(candidates):]...)
	w.mtx.Unlock()

	return nil
}

// LogSeries writes a batch of new series labels to the log.
// The series have to be ordered.
func (w *SegmentWAL) LogSeries(series []RefSeries) error {
	buf := w.getBuffer()

	flag := w.encodeSeries(buf, series)

	w.mtx.Lock()
	defer w.mtx.Unlock()

	err := w.write(WALEntrySeries, flag, buf.get())

	w.putBuffer(buf)

	if err != nil {
		return errors.Wrap(err, "log series")
	}

	tf := w.head()

	for _, s := range series {
		if tf.minSeries > s.Ref {
			tf.minSeries = s.Ref
		}
	}
	return nil
}

// LogSamples writes a batch of new samples to the log.
func (w *SegmentWAL) LogSamples(samples []RefSample) error {
	buf := w.getBuffer()

	flag := w.encodeSamples(buf, samples)

	w.mtx.Lock()
	defer w.mtx.Unlock()

	err := w.write(WALEntrySamples, flag, buf.get())

	w.putBuffer(buf)

	if err != nil {
		return errors.Wrap(err, "log series")
	}
	tf := w.head()

	for _, s := range samples {
		if tf.maxTime < s.T {
			tf.maxTime = s.T
		}
	}
	return nil
}

// LogDeletes write a batch of new deletes to the log.
func (w *SegmentWAL) LogDeletes(stones []Stone) error {
	buf := w.getBuffer()

	flag := w.encodeDeletes(buf, stones)

	w.mtx.Lock()
	defer w.mtx.Unlock()

	err := w.write(WALEntryDeletes, flag, buf.get())

	w.putBuffer(buf)

	if err != nil {
		return errors.Wrap(err, "log series")
	}
	tf := w.head()

	for _, s := range stones {
		for _, iv := range s.intervals {
			if tf.maxTime < iv.Maxt {
				tf.maxTime = iv.Maxt
			}
		}
	}
	return nil
}

// openSegmentFile opens the given segment file and consumes and validates header.
func (w *SegmentWAL) openSegmentFile(name string) (*os.File, error) {
	// We must open all files in read/write mode as we may have to truncate along
	// the way and any file may become the head.
	f, err := os.OpenFile(name, os.O_RDWR, 0666)
	if err != nil {
		return nil, err
	}
	metab := make([]byte, 8)

	// If there is an error, we need close f for platform windows before gc.
	// Otherwise, file op may fail.
	hasError := true
	defer func() {
		if hasError {
			f.Close()
		}
	}()

	if n, err := f.Read(metab); err != nil {
		return nil, errors.Wrapf(err, "validate meta %q", f.Name())
	} else if n != 8 {
		return nil, errors.Errorf("invalid header size %d in %q", n, f.Name())
	}

	if m := binary.BigEndian.Uint32(metab[:4]); m != WALMagic {
		return nil, errors.Errorf("invalid magic header %x in %q", m, f.Name())
	}
	if metab[4] != WALFormatDefault {
		return nil, errors.Errorf("unknown WAL segment format %d in %q", metab[4], f.Name())
	}
	hasError = false
	return f, nil
}

// createSegmentFile creates a new segment file with the given name. It preallocates
// the standard segment size if possible and writes the header.
func (w *SegmentWAL) createSegmentFile(name string) (*os.File, error) {
	f, err := os.Create(name)
	if err != nil {
		return nil, err
	}
	if err = fileutil.Preallocate(f, w.segmentSize, true); err != nil {
		return nil, err
	}
	// Write header metadata for new file.
	metab := make([]byte, 8)
	binary.BigEndian.PutUint32(metab[:4], WALMagic)
	metab[4] = WALFormatDefault

	if _, err := f.Write(metab); err != nil {
		return nil, err
	}
	return f, err
}

// cut finishes the currently active segments and opens the next one.
// The encoder is reset to point to the new segment.
func (w *SegmentWAL) cut() error {
	// Sync current head to disk and close.
	if hf := w.head(); hf != nil {
		if err := w.flush(); err != nil {
			return err
		}
		// Finish last segment asynchronously to not block the WAL moving along
		// in the new segment.
		go func() {
			w.actorc <- func() error {
				off, err := hf.Seek(0, os.SEEK_CUR)
				if err != nil {
					return errors.Wrapf(err, "finish old segment %s", hf.Name())
				}
				if err := hf.Truncate(off); err != nil {
					return errors.Wrapf(err, "finish old segment %s", hf.Name())
				}
				if err := hf.Sync(); err != nil {
					return errors.Wrapf(err, "finish old segment %s", hf.Name())
				}
				if err := hf.Close(); err != nil {
					return errors.Wrapf(err, "finish old segment %s", hf.Name())
				}
				return nil
			}
		}()
	}

	p, _, err := nextSequenceFile(w.dirFile.Name())
	if err != nil {
		return err
	}
	f, err := w.createSegmentFile(p)
	if err != nil {
		return err
	}

	go func() {
		w.actorc <- func() error {
			return errors.Wrap(w.dirFile.Sync(), "sync WAL directory")
		}
	}()

	w.files = append(w.files, newSegmentFile(f))

	// TODO(gouthamve): make the buffer size a constant.
	w.cur = bufio.NewWriterSize(f, 8*1024*1024)
	w.curN = 8

	return nil
}

func (w *SegmentWAL) head() *segmentFile {
	if len(w.files) == 0 {
		return nil
	}
	return w.files[len(w.files)-1]
}

// Sync flushes the changes to disk.
func (w *SegmentWAL) Sync() error {
	var head *segmentFile
	var err error

	// Flush the writer and retrieve the reference to the head segment under mutex lock.
	func() {
		w.mtx.Lock()
		defer w.mtx.Unlock()
		if err = w.flush(); err != nil {
			return
		}
		head = w.head()
	}()
	if err != nil {
		return errors.Wrap(err, "flush buffer")
	}
	if head != nil {
		// But only fsync the head segment after releasing the mutex as it will block on disk I/O.
		start := time.Now()
		err := fileutil.Fdatasync(head.File)
		w.metrics.fsyncDuration.Observe(time.Since(start).Seconds())
		return err
	}
	return nil
}

func (w *SegmentWAL) sync() error {
	if err := w.flush(); err != nil {
		return err
	}
	if w.head() == nil {
		return nil
	}

	start := time.Now()
	err := fileutil.Fdatasync(w.head().File)
	w.metrics.fsyncDuration.Observe(time.Since(start).Seconds())
	return err
}

func (w *SegmentWAL) flush() error {
	if w.cur == nil {
		return nil
	}
	return w.cur.Flush()
}

func (w *SegmentWAL) run(interval time.Duration) {
	var tick <-chan time.Time

	if interval > 0 {
		ticker := time.NewTicker(interval)
		defer ticker.Stop()
		tick = ticker.C
	}
	defer close(w.donec)

	for {
		// Processing all enqueued operations has precedence over shutdown and
		// background syncs.
		select {
		case f := <-w.actorc:
			if err := f(); err != nil {
				level.Error(w.logger).Log("msg", "operation failed", "err", err)
			}
			continue
		default:
		}
		select {
		case <-w.stopc:
			return
		case f := <-w.actorc:
			if err := f(); err != nil {
				level.Error(w.logger).Log("msg", "operation failed", "err", err)
			}
		case <-tick:
			if err := w.Sync(); err != nil {
				level.Error(w.logger).Log("msg", "sync failed", "err", err)
			}
		}
	}
}

// Close syncs all data and closes the underlying resources.
func (w *SegmentWAL) Close() error {
	close(w.stopc)
	<-w.donec

	w.mtx.Lock()
	defer w.mtx.Unlock()

	if err := w.sync(); err != nil {
		return err
	}
	// On opening, a WAL must be fully consumed once. Afterwards
	// only the current segment will still be open.
	if hf := w.head(); hf != nil {
		return errors.Wrapf(hf.Close(), "closing WAL head %s", hf.Name())
	}

	return w.dirFile.Close()
}

const (
	minSectorSize = 512

	// walPageBytes is the alignment for flushing records to the backing Writer.
	// It should be a multiple of the minimum sector size so that WAL can safely
	// distinguish between torn writes and ordinary data corruption.
	walPageBytes = 16 * minSectorSize
)

func (w *SegmentWAL) write(t WALEntryType, flag uint8, buf []byte) error {
	// Cut to the next segment if the entry exceeds the file size unless it would also
	// exceed the size of a new segment.
	// TODO(gouthamve): Add a test for this case where the commit is greater than segmentSize.
	var (
		sz    = int64(len(buf)) + 6
		newsz = w.curN + sz
	)
	// XXX(fabxc): this currently cuts a new file whenever the WAL was newly opened.
	// Probably fine in general but may yield a lot of short files in some cases.
	if w.cur == nil || w.curN > w.segmentSize || newsz > w.segmentSize && sz <= w.segmentSize {
		if err := w.cut(); err != nil {
			return err
		}
	}
	n, err := w.writeTo(w.cur, w.crc32, t, flag, buf)

	w.curN += int64(n)

	return err
}

func (w *SegmentWAL) writeTo(wr io.Writer, crc32 hash.Hash, t WALEntryType, flag uint8, buf []byte) (int, error) {
	if len(buf) == 0 {
		return 0, nil
	}
	crc32.Reset()
	wr = io.MultiWriter(crc32, wr)

	var b [6]byte
	b[0] = byte(t)
	b[1] = flag

	binary.BigEndian.PutUint32(b[2:], uint32(len(buf)))

	n1, err := wr.Write(b[:])
	if err != nil {
		return n1, err
	}
	n2, err := wr.Write(buf)
	if err != nil {
		return n1 + n2, err
	}
	n3, err := wr.Write(crc32.Sum(b[:0]))

	return n1 + n2 + n3, err
}

const (
	walSeriesSimple  = 1
	walSamplesSimple = 1
	walDeletesSimple = 1
)

func (w *SegmentWAL) encodeSeries(buf *encbuf, series []RefSeries) uint8 {
	for _, s := range series {
		buf.putBE64(s.Ref)
		buf.putUvarint(len(s.Labels))

		for _, l := range s.Labels {
			buf.putUvarintStr(l.Name)
			buf.putUvarintStr(l.Value)
		}
	}
	return walSeriesSimple
}

func (w *SegmentWAL) encodeSamples(buf *encbuf, samples []RefSample) uint8 {
	if len(samples) == 0 {
		return walSamplesSimple
	}
	// Store base timestamp and base reference number of first sample.
	// All samples encode their timestamp and ref as delta to those.
	//
	// TODO(fabxc): optimize for all samples having the same timestamp.
	first := samples[0]

	buf.putBE64(first.Ref)
	buf.putBE64int64(first.T)

	for _, s := range samples {
		buf.putVarint64(int64(s.Ref) - int64(first.Ref))
		buf.putVarint64(s.T - first.T)
		buf.putBE64(math.Float64bits(s.V))
	}
	return walSamplesSimple
}

func (w *SegmentWAL) encodeDeletes(buf *encbuf, stones []Stone) uint8 {
	for _, s := range stones {
		for _, iv := range s.intervals {
			buf.putBE64(s.ref)
			buf.putVarint64(iv.Mint)
			buf.putVarint64(iv.Maxt)
		}
	}
	return walDeletesSimple
}

// walReader decodes and emits write ahead log entries.
type walReader struct {
	logger log.Logger

	files []*segmentFile
	cur   int
	buf   []byte
	crc32 hash.Hash32

	curType    WALEntryType
	curFlag    byte
	curBuf     []byte
	lastOffset int64 // offset after last successfully read entry

	err error
}

func newWALReader(files []*segmentFile, l log.Logger) *walReader {
	if l == nil {
		l = log.NewNopLogger()
	}
	return &walReader{
		logger: l,
		files:  files,
		buf:    make([]byte, 0, 128*4096),
		crc32:  newCRC32(),
	}
}

// Err returns the last error the reader encountered.
func (r *walReader) Err() error {
	return r.err
}

func (r *walReader) Read(
	seriesf func([]RefSeries),
	samplesf func([]RefSample),
	deletesf func([]Stone),
) error {
	// Concurrency for replaying the WAL is very limited. We at least split out decoding and
	// processing into separate threads.
	// Historically, the processing is the bottleneck with reading and decoding using only
	// 15% of the CPU.
	var (
		seriesPool sync.Pool
		samplePool sync.Pool
		deletePool sync.Pool
	)
	donec := make(chan struct{})
	datac := make(chan interface{}, 100)

	go func() {
		defer close(donec)

		for x := range datac {
			switch v := x.(type) {
			case []RefSeries:
				if seriesf != nil {
					seriesf(v)
				}
				seriesPool.Put(v[:0])
			case []RefSample:
				if samplesf != nil {
					samplesf(v)
				}
				samplePool.Put(v[:0])
			case []Stone:
				if deletesf != nil {
					deletesf(v)
				}
				deletePool.Put(v[:0])
			default:
				level.Error(r.logger).Log("msg", "unexpected data type")
			}
		}
	}()

	var err error

	for r.next() {
		et, flag, b := r.at()

		// In decoding below we never return a walCorruptionErr for now.
		// Those should generally be catched by entry decoding before.
		switch et {
		case WALEntrySeries:
			var series []RefSeries
			if v := seriesPool.Get(); v == nil {
				series = make([]RefSeries, 0, 512)
			} else {
				series = v.([]RefSeries)
			}

			err := r.decodeSeries(flag, b, &series)
			if err != nil {
				err = errors.Wrap(err, "decode series entry")
				break
			}
			datac <- series

			cf := r.current()
			for _, s := range series {
				if cf.minSeries > s.Ref {
					cf.minSeries = s.Ref
				}
			}
		case WALEntrySamples:
			var samples []RefSample
			if v := samplePool.Get(); v == nil {
				samples = make([]RefSample, 0, 512)
			} else {
				samples = v.([]RefSample)
			}

			err := r.decodeSamples(flag, b, &samples)
			if err != nil {
				err = errors.Wrap(err, "decode samples entry")
				break
			}
			datac <- samples

			// Update the times for the WAL segment file.
			cf := r.current()
			for _, s := range samples {
				if cf.maxTime < s.T {
					cf.maxTime = s.T
				}
			}
		case WALEntryDeletes:
			var deletes []Stone
			if v := deletePool.Get(); v == nil {
				deletes = make([]Stone, 0, 512)
			} else {
				deletes = v.([]Stone)
			}

			err := r.decodeDeletes(flag, b, &deletes)
			if err != nil {
				err = errors.Wrap(err, "decode delete entry")
				break
			}
			datac <- deletes

			// Update the times for the WAL segment file.
			cf := r.current()
			for _, s := range deletes {
				for _, iv := range s.intervals {
					if cf.maxTime < iv.Maxt {
						cf.maxTime = iv.Maxt
					}
				}
			}
		}
	}
	close(datac)
	<-donec

	if err != nil {
		return err
	}
	if r.Err() != nil {
		return errors.Wrap(r.Err(), "read entry")
	}
	return nil
}

func (r *walReader) at() (WALEntryType, byte, []byte) {
	return r.curType, r.curFlag, r.curBuf
}

// next returns decodes the next entry pair and returns true
// if it was succesful.
func (r *walReader) next() bool {
	if r.cur >= len(r.files) {
		return false
	}
	cf := r.files[r.cur]

	// Remember the offset after the last correctly read entry. If the next one
	// is corrupted, this is where we can safely truncate.
	r.lastOffset, r.err = cf.Seek(0, os.SEEK_CUR)
	if r.err != nil {
		return false
	}

	et, flag, b, err := r.entry(cf)
	// If we reached the end of the reader, advance to the next one
	// and close.
	// Do not close on the last one as it will still be appended to.
	if err == io.EOF {
		if r.cur == len(r.files)-1 {
			return false
		}
		// Current reader completed, close and move to the next one.
		if err := cf.Close(); err != nil {
			r.err = err
			return false
		}
		r.cur++
		return r.next()
	}
	if err != nil {
		r.err = err
		return false
	}

	r.curType = et
	r.curFlag = flag
	r.curBuf = b
	return r.err == nil
}

func (r *walReader) current() *segmentFile {
	return r.files[r.cur]
}

// walCorruptionErr is a type wrapper for errors that indicate WAL corruption
// and trigger a truncation.
type walCorruptionErr struct {
	err        error
	file       int
	lastOffset int64
}

func (e walCorruptionErr) Error() string {
	return fmt.Sprintf("%s <file: %d, lastOffset: %d>", e.err, e.file, e.lastOffset)
}

func (r *walReader) corruptionErr(s string, args ...interface{}) error {
	return walCorruptionErr{
		err:        errors.Errorf(s, args...),
		file:       r.cur,
		lastOffset: r.lastOffset,
	}
}

func (r *walReader) entry(cr io.Reader) (WALEntryType, byte, []byte, error) {
	r.crc32.Reset()
	tr := io.TeeReader(cr, r.crc32)

	b := make([]byte, 6)
	if n, err := tr.Read(b); err != nil {
		return 0, 0, nil, err
	} else if n != 6 {
		return 0, 0, nil, r.corruptionErr("invalid entry header size %d", n)
	}

	var (
		etype  = WALEntryType(b[0])
		flag   = b[1]
		length = int(binary.BigEndian.Uint32(b[2:]))
	)
	// Exit if we reached pre-allocated space.
	if etype == 0 {
		return 0, 0, nil, io.EOF
	}
	if etype != WALEntrySeries && etype != WALEntrySamples && etype != WALEntryDeletes {
		return 0, 0, nil, r.corruptionErr("invalid entry type %d", etype)
	}

	if length > len(r.buf) {
		r.buf = make([]byte, length)
	}
	buf := r.buf[:length]

	if n, err := tr.Read(buf); err != nil {
		return 0, 0, nil, err
	} else if n != length {
		return 0, 0, nil, r.corruptionErr("invalid entry body size %d", n)
	}

	if n, err := cr.Read(b[:4]); err != nil {
		return 0, 0, nil, err
	} else if n != 4 {
		return 0, 0, nil, r.corruptionErr("invalid checksum length %d", n)
	}
	if exp, has := binary.BigEndian.Uint32(b[:4]), r.crc32.Sum32(); has != exp {
		return 0, 0, nil, r.corruptionErr("unexpected CRC32 checksum %x, want %x", has, exp)
	}

	return etype, flag, buf, nil
}

func (r *walReader) decodeSeries(flag byte, b []byte, res *[]RefSeries) error {
	dec := decbuf{b: b}

	for len(dec.b) > 0 && dec.err() == nil {
		ref := dec.be64()

		lset := make(labels.Labels, dec.uvarint())

		for i := range lset {
			lset[i].Name = dec.uvarintStr()
			lset[i].Value = dec.uvarintStr()
		}
		sort.Sort(lset)

		*res = append(*res, RefSeries{
			Ref:    ref,
			Labels: lset,
		})
	}
	if dec.err() != nil {
		return dec.err()
	}
	if len(dec.b) > 0 {
		return errors.Errorf("unexpected %d bytes left in entry", len(dec.b))
	}
	return nil
}

func (r *walReader) decodeSamples(flag byte, b []byte, res *[]RefSample) error {
	if len(b) == 0 {
		return nil
	}
	dec := decbuf{b: b}

	var (
		baseRef  = dec.be64()
		baseTime = dec.be64int64()
	)

	for len(dec.b) > 0 && dec.err() == nil {
		dref := dec.varint64()
		dtime := dec.varint64()
		val := dec.be64()

		*res = append(*res, RefSample{
			Ref: uint64(int64(baseRef) + dref),
			T:   baseTime + dtime,
			V:   math.Float64frombits(val),
		})
	}

	if dec.err() != nil {
		return errors.Wrapf(dec.err(), "decode error after %d samples", len(*res))
	}
	if len(dec.b) > 0 {
		return errors.Errorf("unexpected %d bytes left in entry", len(dec.b))
	}
	return nil
}

func (r *walReader) decodeDeletes(flag byte, b []byte, res *[]Stone) error {
	dec := &decbuf{b: b}

	for dec.len() > 0 && dec.err() == nil {
		*res = append(*res, Stone{
			ref: dec.be64(),
			intervals: Intervals{
				{Mint: dec.varint64(), Maxt: dec.varint64()},
			},
		})
	}
	if dec.err() != nil {
		return dec.err()
	}
	if len(dec.b) > 0 {
		return errors.Errorf("unexpected %d bytes left in entry", len(dec.b))
	}
	return nil
}